@@ -6,7 +6,8 @@ import path from "node:path";
66import type { KnowledgeReader , KnowledgeDocumentRef , KnowledgeReadContext } from "@devcodex/capability-graph" ;
77
88type ReadResult = Awaited < ReturnType < KnowledgeReader [ "read" ] > > ;
9- interface Snapshot { file : string ; etag : string ; sha256 : string ; contentType : string ; bytes : number ; expires : number ; leases : number ; retired : boolean }
9+ type FileIdentity = { dev : bigint ; ino : bigint ; size : bigint ; mtimeNs : bigint ; ctimeNs : bigint } ;
10+ interface Snapshot { file : string ; identity : FileIdentity ; etag : string ; sha256 : string ; contentType : string ; bytes : number ; expires : number ; leases : number ; retired : boolean }
1011interface SnapshotOptions { directory : string ; maxBytes ?: number ; maxEntries ?: number ; ttlMs ?: number }
1112interface Options { allowedOrigins : readonly string [ ] ; timeoutMs ?: number ; maxConcurrent ?: number ;
1213 /** Host-owned Agent (e.g. proxy-agent on Node 20/22/24); Reader never closes it. */
@@ -24,6 +25,7 @@ export class HttpKnowledgeReader implements KnowledgeReader {
2425 private readonly snapshotLimits ?: { maxBytes : number ; maxEntries : number ; ttlMs : number } ;
2526 private directory ?: Promise < string > ;
2627 private closed = false ;
28+ readonly readRange ?: NonNullable < KnowledgeReader [ "readRange" ] > ;
2729 constructor ( private readonly options : Options ) {
2830 this . timeoutMs = options . timeoutMs ?? 2000 ; this . maxConcurrent = options . maxConcurrent ?? 4 ;
2931 if ( ! [ this . timeoutMs , this . maxConcurrent ] . every ( ( value ) => Number . isSafeInteger ( value ) && value > 0 ) || ! options . allowedOrigins . length ) throw new Error ( "Explicit origins and positive HTTP limits are required" ) ;
@@ -36,6 +38,7 @@ export class HttpKnowledgeReader implements KnowledgeReader {
3638 if ( ! path . isAbsolute ( options . snapshot . directory ) ) throw new Error ( "Snapshot directory must be an absolute host-owned path outside the source checkout" ) ;
3739 this . snapshotLimits = { maxBytes : options . snapshot . maxBytes ?? 16_777_216 , maxEntries : options . snapshot . maxEntries ?? 8 , ttlMs : options . snapshot . ttlMs ?? 300_000 } ;
3840 if ( ! Object . values ( this . snapshotLimits ) . every ( ( value ) => Number . isSafeInteger ( value ) && value > 0 ) ) throw new Error ( "Invalid snapshot budgets" ) ;
41+ this . readRange = this . readSnapshotRange . bind ( this ) ;
3942 }
4043 }
4144 get activeRequests ( ) : number { return this . requests . size ; }
@@ -85,6 +88,50 @@ export class HttpKnowledgeReader implements KnowledgeReader {
8588 finish ( ) ; this . streams . delete ( done ) ;
8689 }
8790 }
91+ /** Read only a leased, fully downloaded representation after Core's strong revalidation.
92+ * Core independently verifies aligned block hashes; file identity detects local replacement
93+ * or edits early, and is never used as proof that the remote source is current. */
94+ private async readSnapshotRange ( ref : KnowledgeDocumentRef , _context : KnowledgeReadContext ,
95+ options : Parameters < NonNullable < KnowledgeReader [ "readRange" ] > > [ 2 ] ) : ReturnType < NonNullable < KnowledgeReader [ "readRange" ] > > {
96+ if ( this . closed || ref . locator . type !== "http" || this . streams . size >= this . maxConcurrent ) throw new Error ( "Reader unavailable" ) ;
97+ if ( ! [ options . startOffset , options . endOffset ] . every ( Number . isSafeInteger ) || options . startOffset < 0 || options . endOffset < options . startOffset ) throw new Error ( "Invalid range" ) ;
98+ options . signal ?. throwIfAborted ( ) ;
99+ const source = ref . locator . url ; const snapshot = this . snapshots . get ( source ) ;
100+ if ( ! snapshot || snapshot . expires <= Date . now ( ) || options . contentId !== `k:${ snapshot . sha256 . slice ( 0 , 16 ) } ) return undefined ;
101+ if ( options . endOffset > snapshot . bytes ) throw new Error ( "Invalid range" ) ;
102+ let finish ! : ( ) => void ; const done = new Promise < void > ( ( resolve ) => { finish = resolve ; } ) ; this . streams . add ( done ) ;
103+ snapshot . leases ++ ; let handle : FileHandle | undefined ;
104+ const unchanged = ( actual : FileIdentity ) => Object . entries ( snapshot . identity ) . every ( ( [ key , value ] ) => actual [ key as keyof FileIdentity ] === value ) ;
105+ try {
106+ try { handle = await open ( snapshot . file , "r" ) ; }
107+ catch ( error ) {
108+ if ( ( error as NodeJS . ErrnoException ) . code !== "ENOENT" ) throw error ;
109+ if ( this . snapshots . get ( source ) === snapshot ) await this . retire ( source ) ;
110+ return undefined ;
111+ }
112+ if ( ! unchanged ( await handle . stat ( { bigint : true } ) ) ) throw new Error ( "Snapshot changed" ) ;
113+ const bytes = Buffer . alloc ( options . endOffset - options . startOffset ) ; let offset = 0 ;
114+ while ( offset < bytes . length ) {
115+ options . signal ?. throwIfAborted ( ) ; if ( this . closed ) throw new Error ( "Reader unavailable" ) ;
116+ const { bytesRead } = await handle . read ( bytes , offset , Math . min ( 32768 , bytes . length - offset ) , options . startOffset + offset ) ;
117+ if ( ! bytesRead ) throw new Error ( "Snapshot incomplete" ) ; offset += bytesRead ;
118+ }
119+ options . signal ?. throwIfAborted ( ) ; if ( this . closed || ! unchanged ( await handle . stat ( { bigint : true } ) ) ) throw new Error ( "Snapshot changed" ) ;
120+ if ( this . snapshots . get ( source ) === snapshot ) { this . snapshots . delete ( source ) ; this . snapshots . set ( source , snapshot ) ; }
121+ return { bytes, contentId : options . contentId , source, contentType : snapshot . contentType , totalBytes : snapshot . bytes } ;
122+ } catch ( error ) {
123+ if ( this . snapshots . get ( source ) === snapshot ) await this . retire ( source ) ;
124+ throw error ;
125+ } finally {
126+ try {
127+ try { await handle ?. close ( ) ; }
128+ finally {
129+ snapshot . leases -- ;
130+ if ( snapshot . retired && ! snapshot . leases ) await rm ( snapshot . file , { force : true } ) ;
131+ }
132+ } finally { finish ( ) ; this . streams . delete ( done ) ; }
133+ }
134+ }
88135 /** Strong ETag validates a previously complete disk snapshot; weak/no validator falls back to full GET.
89136 * Normal streams still verify complete bytes; Core may reuse its own proved copy for body pages. No 206 is accepted. */
90137 async * stream ( ref : KnowledgeDocumentRef , _context : KnowledgeReadContext , options : { chunkBytes : number ; signal ?: AbortSignal } ) {
@@ -161,9 +208,14 @@ export class HttpKnowledgeReader implements KnowledgeReader {
161208 if ( ! emitted ) yield { bytes : new Uint8Array ( ) , source, contentType } ;
162209 if ( writer && temporary && ! this . closed ) {
163210 await writer . close ( ) ; writer = undefined ;
211+ const inspection = await open ( temporary , "r" ) ; let identity : FileIdentity ;
212+ try {
213+ const stat = await inspection . stat ( { bigint : true } ) ;
214+ identity = { dev : stat . dev , ino : stat . ino , size : stat . size , mtimeNs : stat . mtimeNs , ctimeNs : stat . ctimeNs } ;
215+ } finally { await inspection . close ( ) ; }
164216 while ( this . snapshots . size && ( this . snapshots . size >= this . snapshotLimits ! . maxEntries || this . cachedSnapshotBytes + total > this . snapshotLimits ! . maxBytes ) ) await this . retire ( this . snapshots . keys ( ) . next ( ) . value ! ) ;
165217 await this . retire ( source ) ;
166- this . snapshots . set ( source , { file : temporary , etag : etag ! , sha256 : sourceHash . digest ( "hex" ) , bytes : total , contentType, expires : Date . now ( ) + this . snapshotLimits ! . ttlMs , leases : 0 , retired : false } ) ; published = true ;
218+ this . snapshots . set ( source , { file : temporary , identity , etag : etag ! , sha256 : sourceHash . digest ( "hex" ) , bytes : total , contentType, expires : Date . now ( ) + this . snapshotLimits ! . ttlMs , leases : 0 , retired : false } ) ; published = true ;
167219 }
168220 } finally {
169221 clearTimeout ( timer ) ; options . signal ?. removeEventListener ( "abort" , abort ) ;
0 commit comments