Add proof-backed parser byte reads and preserve retryable source failures

headless/README.md

@@ -617,3 +617,30 @@ excluded. Target component/version changes cancel with `target_changed`; removal
 of the view canvas publishes fallback navigation and `canvas_changed` together.
 Late updates/commit cannot use the canceled preview. Failed admission does not
 replace the accepted baseline, and no chain, preview or gesture enters recovery.
+
+Internal parser byte ports and read evidence (`import-byte-reader.ts`,
+`import-read-proof.ts`) provide the next integration boundary for parsing before
+whole-file hashing. NDJSON, JSON array, streaming document and its indexed
+fallback all accept the optional `readBytes` port; the default Blob path keeps
+its original slice request bounds and verifies the returned byte count.
+
+The proof reader binds a file ID and byte length to SHA-256 hashes of aligned
+16 KiB blocks, recording evidence only after an entire range is delivered.
+A restored manifest can verify its recorded blocks before parsing resumes.
+Unrecorded blocks are explicitly outside this proof: the manifest is neither a
+whole-file digest nor authority to dispatch mutations. The current engine still
+uses integrity version 1 and waits for its complete SHA; activating the new path
+requires atomic job/proof checkpoints plus a durable verification/send barrier.
+No recovery or source interpreter version changed in this unit.
+
+The reader owns a hash Map (at most 4096 blocks for the 64 MiB source limit), a
+bounded LRU byte Map (four blocks by default), and up to four shared in-flight
+block Promises/four range reads. Each range is capped at 4 MiB. Aborted waiters
+release their own read but do not free occupied native reads early or cancel
+another waiter. Close fences late results and clears byte buffers; checkpoints
+contain no byte buffers. These bounds do not assert native Blob memory release.
+
+Byte I/O/capacity/integrity failures throw instead of becoming a terminal malformed
+record. In particular a document reader preserves its ordinal for retry; the
+engine pauses the import, retains its editable prefix, and resumes after the I/O
+recovers. Ordinary malformed JSON/UTF-8 still uses the existing record diagnostics.

src/lib/canvas-runtime/import-byte-reader.ts

@@ -0,0 +1,17 @@
+import {fault} from '../canvas-core/values.ts';
+
+const failures=new Set(['import_source_changed','invalid_import_byte_range','import_read_capacity','import_source_closed','import_source_read_failed','invalid_import_read_proof','import_read_proof_interpreter_unavailable']);
+export const isImportByteFailure=(error:unknown)=>failures.has((error as {code?:string}|undefined)?.code??'');
+
+/** Optional immutable-source port. Implementations return exactly the requested bytes. */
+export type ImportByteReader={readonly byteLength:number;read(start:number,end:number,signal:AbortSignal):Promise<Uint8Array>};
+export async function readImportRange(bytes:Blob,reader:ImportByteReader|undefined,start:number,end:number,signal:AbortSignal):Promise<Uint8Array<ArrayBuffer>>{
+ if(signal.aborted)throw fault('import_aborted');
+ if(!Number.isSafeInteger(start)||!Number.isSafeInteger(end)||start<0||end<start||start>bytes.size||reader&&reader.byteLength!==bytes.size)throw fault('invalid_import_byte_range');
+ const limit=Math.min(end,bytes.size);let value:Uint8Array;
+ try{value=reader?await reader.read(start,limit,signal):new Uint8Array(await bytes.slice(start,end).arrayBuffer());}
+ catch(error){if(signal.aborted)throw fault('import_aborted');if(isImportByteFailure(error))throw error;throw fault('import_source_read_failed');}
+ if(signal.aborted)throw fault('import_aborted');
+ if(!(value instanceof Uint8Array)||value.byteLength!==limit-start)throw fault('import_source_changed');
+ return reader?new Uint8Array(value):value as Uint8Array<ArrayBuffer>;
+}

src/lib/canvas-runtime/import-document-stream.ts

@@ -1,3 +1,4 @@
+import {readImportRange,isImportByteFailure,type ImportByteReader} from './import-byte-reader.ts';
 import type {ImportLine,ImportSource} from './imports.ts';
 import {createImportBudget,type ImportScheduling} from './import-budget.ts';
 import {openDocumentImportSource} from './import-document.ts';
@@ -6,7 +7,7 @@ import {fault,validJson} from '../canvas-core/values.ts';
 type Range={start:number;end:number};
 type ScanCache={start:number;bytes:Uint8Array};
 type Canvas={id:string;parent:null|{canvasId:string;entryElementId:string}};
-type Options=ImportScheduling&{bytes:Blob;fingerprint:string;signal:AbortSignal;maxRecordBytes?:number};
+type Options=ImportScheduling&{bytes:Blob;fingerprint:string;readBytes?:ImportByteReader;signal:AbortSignal;maxRecordBytes?:number};
 const allowed=new Set(['documentFormatVersion','projectId','canvases','elements','tombstones']);
 const whitespace=(byte:number)=>byte===9||byte===10||byte===13||byte===32;
 
@@ -17,7 +18,7 @@ function scanner(options:Options,signal:AbortSignal,start=0,cache:ScanCache={sta
  const check=()=>{if(signal.aborted)throw fault('import_aborted');};
  async function peek(){
   check();if(cursor>=bytes.size)return -1;
-  if(cursor<cache.start||cursor>=cache.start+cache.bytes.length){const loaded=new Uint8Array(await bytes.slice(cursor,cursor+16384).arrayBuffer());cache.start=cursor;cache.bytes=loaded;check();}
+  if(cursor<cache.start||cursor>=cache.start+cache.bytes.length){const loaded=await readImportRange(bytes,options.readBytes,cursor,cursor+16384,signal);cache.start=cursor;cache.bytes=loaded;check();}
   return cache.bytes[cursor-cache.start];
  }
  function step(){cursor++;return budget.step();}
@@ -40,8 +41,8 @@ function scanner(options:Options,signal:AbortSignal,start=0,cache:ScanCache={sta
  }
  async function decode(range:Range,limit=options.maxRecordBytes??256*1024):Promise<any>{
   check();if(range.end-range.start>limit)throw fault('record_too_large');
-  let text:string;
-  try{text=new TextDecoder('utf-8',{fatal:true}).decode(await bytes.slice(range.start,range.end).arrayBuffer());}catch(error){check();throw fault('invalid_utf8');}
+  const data=await readImportRange(bytes,options.readBytes,range.start,range.end,signal);let text:string;
+  try{text=new TextDecoder('utf-8',{fatal:true}).decode(data);}catch(error){check();throw fault('invalid_utf8');}
   check();try{const value=JSON.parse(text);if(!validJson(value))throw Error();return value;}catch{throw fault('invalid_json');}
  }
  return{peek,step,spaces,take,value,decode,position:()=>cursor};
@@ -124,7 +125,7 @@ export async function openStreamingDocumentImportSource(input:Options):Promise<I
   if(!Number.isSafeInteger(position)||position<0)throw fault('invalid_import_cursor');
   const elementIndex=position-children.length;
   while(elementIndex>=ranges.length&&phase!=='done'){
-   try{await scanNext(readSignal);}catch(error){if(readSignal.aborted)throw fault('import_aborted');failure=(error as {code?:string}).code??'invalid_document_import';phase='done';}
+   try{await scanNext(readSignal);}catch(error){if(readSignal.aborted)throw fault('import_aborted');if(isImportByteFailure(error))throw error;failure=(error as {code?:string}).code??'invalid_document_import';phase='done';}
   }
   const count=children.length+ranges.length;
   if(position>=count){
@@ -137,7 +138,7 @@ export async function openStreamingDocumentImportSource(input:Options):Promise<I
    const kind=position<children.length?'canvas':'element',record=kind==='canvas'?children[position]:await scanner(options,readSignal).decode(ranges[elementIndex]);
    if(!record||typeof record!=='object'||Array.isArray(record)||typeof record.id!=='string'||!record.id)throw fault('invalid_native_import_record');
    return{...base,value:{kind,key:record.id,record,sourceRootId:rootId}};
-  }catch(error){if(readSignal.aborted)throw fault('import_aborted');return{...base,error:(error as {code?:string}).code??'invalid_document_import'};}
+  }catch(error){if(readSignal.aborted)throw fault('import_aborted');if(isImportByteFailure(error))throw error;return{...base,error:(error as {code?:string}).code??'invalid_document_import'};}
  }
  return Object.freeze({fingerprint:options.fingerprint,parserVersion:5,read(position:number,readSignal:AbortSignal){
   // A slow future scan must not block re-reading a pending unit for commit/recovery.

src/lib/canvas-runtime/import-document.ts

@@ -1,15 +1,16 @@
+import {readImportRange,isImportByteFailure,type ImportByteReader} from './import-byte-reader.ts';
 import {createImportBudget,type ImportScheduling} from './import-budget.ts';
 import type {ImportSource,ImportLine} from './imports.ts';
 import {fault,validJson} from '../canvas-core/values.ts';
 type Range={start:number;end:number};
 /** Index the envelope without decoding element payloads. Original bytes remain the source.
  * Preparation scans O(bytes) and retains O(records) ranges; this is not constant memory. */
-export async function openDocumentImportSource(options:{bytes:Blob;fingerprint:string;signal:AbortSignal;maxRecordBytes?:number}&ImportScheduling):Promise<ImportSource>{
+export async function openDocumentImportSource(options:{bytes:Blob;fingerprint:string;readBytes?:ImportByteReader;signal:AbortSignal;maxRecordBytes?:number}&ImportScheduling):Promise<ImportSource>{
  const {bytes,signal}=options,max=options.maxRecordBytes??256*1024;
  if(!(bytes instanceof Blob)||bytes.size>64*1024*1024||!options.fingerprint||!Number.isSafeInteger(max)||max<1)throw fault('invalid_import_source');
  let cursor=0,cache={start:-1,bytes:new Uint8Array()};const budget=createImportBudget(signal,options);
  const space=(b:number)=>[9,10,13,32].includes(b);
- async function peek(){if(signal.aborted)throw fault('import_aborted');if(cursor>=bytes.size)return -1;if(cursor<cache.start||cursor>=cache.start+cache.bytes.length){cache={start:cursor,bytes:new Uint8Array(await bytes.slice(cursor,cursor+16384).arrayBuffer())};if(signal.aborted)throw fault('import_aborted');}return cache.bytes[cursor-cache.start];}
+ async function peek(){if(signal.aborted)throw fault('import_aborted');if(cursor>=bytes.size)return -1;if(cursor<cache.start||cursor>=cache.start+cache.bytes.length){cache={start:cursor,bytes:await readImportRange(bytes,options.readBytes,cursor,cursor+16384,signal)};if(signal.aborted)throw fault('import_aborted');}return cache.bytes[cursor-cache.start];}
  function step(){cursor++;return budget.step();}
  async function spaces(){while(space(await peek()))await step();}
  async function take(byte:number){await spaces();if(await peek()!==byte)throw fault('invalid_document_import');await step();}
@@ -19,7 +20,7 @@ export async function openDocumentImportSource(options:{bytes:Blob;fingerprint:s
    const ended=!quoted&&stack.length===0&&(b===34||b===93||b===125);await step();if(ended)break;
   }if(cursor===start)throw fault('invalid_document_import');return{start,end:cursor};
  }
- async function decode(range:Range,limit=max){if(range.end-range.start>limit)throw fault('record_too_large');let text:string;try{text=new TextDecoder('utf-8',{fatal:true}).decode(await bytes.slice(range.start,range.end).arrayBuffer());}catch{throw fault('invalid_utf8');}try{const result=JSON.parse(text);if(!validJson(result))throw Error();return result;}catch{throw fault('invalid_json');}}
+ async function decode(range:Range,limit=max,readSignal=signal){if(range.end-range.start>limit)throw fault('record_too_large');const data=await readImportRange(bytes,options.readBytes,range.start,range.end,readSignal);let text:string;try{text=new TextDecoder('utf-8',{fatal:true}).decode(data);}catch{throw fault('invalid_utf8');}try{const result=JSON.parse(text);if(!validJson(result))throw Error();return result;}catch{throw fault('invalid_json');}}
  const fields=new Map<string,unknown>(),elements:Range[]=[];
  await take(123);await spaces();if(await peek()===125)throw fault('invalid_document_import');
  while(true){const key=await decode(await value(),1024);if(typeof key!=='string'||fields.has(key))throw fault('invalid_document_import');await take(58);
@@ -38,6 +39,6 @@ export async function openDocumentImportSource(options:{bytes:Blob;fingerprint:s
  return Object.freeze({fingerprint:options.fingerprint,parserVersion:4,async read(position:number,readSignal:AbortSignal):Promise<ImportLine|null>{
   if(readSignal.aborted)throw fault('import_aborted');if(!Number.isSafeInteger(position)||position<0||position>count)throw fault('invalid_import_cursor');if(position===count)return null;
   const base={cursor:position,nextCursor:position+1,key:`record:${position}`};let record:any,kind:string;
-  try{if(position<children.length){kind='canvas';record=children[position];}else{kind='element';record=await decode(elements[position-children.length]);}if(readSignal.aborted)throw fault('import_aborted');if(!record||typeof record!=='object'||Array.isArray(record)||typeof record.id!=='string'||!record.id)throw fault('invalid_native_import_record');return{...base,value:{kind,key:record.id,record,sourceRootId:rootId}};}catch(error){if(readSignal.aborted)throw fault('import_aborted');return{...base,error:(error as {code?:string}).code??'invalid_document_import'};}
+  try{if(position<children.length){kind='canvas';record=children[position];}else{kind='element';record=await decode(elements[position-children.length],max,readSignal);}if(readSignal.aborted)throw fault('import_aborted');if(!record||typeof record!=='object'||Array.isArray(record)||typeof record.id!=='string'||!record.id)throw fault('invalid_native_import_record');return{...base,value:{kind,key:record.id,record,sourceRootId:rootId}};}catch(error){if(readSignal.aborted)throw fault('import_aborted');if(isImportByteFailure(error))throw error;return{...base,error:(error as {code?:string}).code??'invalid_document_import'};}
  }});
 }

src/lib/canvas-runtime/import-json-array.ts

@@ -1,9 +1,10 @@
+import {readImportRange,type ImportByteReader} from './import-byte-reader.ts';
 import {createImportBudget,type ImportScheduling} from './import-budget.ts';
 import type {ImportSource,ImportLine} from './imports.ts';
 import {fault,validJson} from '../canvas-core/values.ts';
 /** Seekable JSON array reader. Cursors are byte offsets after '[' or a record comma.
  * Structure errors end the source; previously published records remain explicit. */
-export function createJsonArrayImportSource(options:{bytes:Blob;fingerprint:string;maxRecordBytes?:number;chunkBytes?:number}&ImportScheduling):ImportSource{
+export function createJsonArrayImportSource(options:{bytes:Blob;fingerprint:string;readBytes?:ImportByteReader;maxRecordBytes?:number;chunkBytes?:number}&ImportScheduling):ImportSource{
  const bytes=options.bytes,max=options.maxRecordBytes??256*1024,chunk=options.chunkBytes??16384;
  if(!(bytes instanceof Blob)||!options.fingerprint||!Number.isSafeInteger(max)||max<1||!Number.isSafeInteger(chunk)||chunk<1||chunk>max)throw fault('invalid_import_source');
  let cache={offset:-1,value:new Uint8Array()};const whitespace=(b:number)=>[9,10,13,32].includes(b);
@@ -12,7 +13,7 @@ export function createJsonArrayImportSource(options:{bytes:Blob;fingerprint:stri
   if(start>0&&start===bytes.size){if(signal.aborted)throw fault('import_aborted');return null;}
   if(bytes.size===0)throw fault('invalid_json_array');
   let cursor=start;const budget=createImportBudget(signal,options);
-  async function peek(){if(signal.aborted)throw fault('import_aborted');if(cursor>=bytes.size)return -1;if(cursor<cache.offset||cursor>=cache.offset+cache.value.length){cache={offset:cursor,value:new Uint8Array(await bytes.slice(cursor,cursor+chunk).arrayBuffer())};if(signal.aborted)throw fault('import_aborted');}return cache.value[cursor-cache.offset];}
+  async function peek(){if(signal.aborted)throw fault('import_aborted');if(cursor>=bytes.size)return -1;if(cursor<cache.offset||cursor>=cache.offset+cache.value.length){cache={offset:cursor,value:await readImportRange(bytes,options.readBytes,cursor,cursor+chunk,signal)};if(signal.aborted)throw fault('import_aborted');}return cache.value[cursor-cache.offset];}
   function advance(){cursor++;return budget.step();}
   async function spaces(){while(whitespace(await peek()))await advance();}
   const failure=(error:string):ImportLine=>({cursor:start,nextCursor:bytes.size,key:`byte:${start}`,error});

src/lib/canvas-runtime/import-read-proof.ts

@@ -0,0 +1,71 @@
+import {fault,freeze,validJson} from '../canvas-core/values.ts';
+import {createSha256} from './sha256.ts';
+import {createImportBudget,type ImportScheduling} from './import-budget.ts';
+import type {ImportByteReader} from './import-byte-reader.ts';
+
+export const IMPORT_PROOF_BLOCK_BYTES=16384;
+const MAX_SOURCE_BYTES=64*1024*1024;
+export type ImportReadProof={readonly version:1;readonly fileId:string;readonly byteLength:number;readonly blockBytes:16384;readonly blocks:readonly {readonly index:number;readonly sha256:string}[]};
+/** This binds only read blocks. It is not a whole-source digest or permission to send mutations. */
+export function validateImportReadProof(value:unknown,fileId:string,byteLength:number):ImportReadProof{
+ if(typeof fileId!=='string'||!fileId||!Number.isSafeInteger(byteLength)||byteLength<0||byteLength>MAX_SOURCE_BYTES)throw fault('invalid_import_read_proof');
+ const proof=value as ImportReadProof;
+ if(!proof||!validJson(proof)||proof.version!==1)throw fault('import_read_proof_interpreter_unavailable');
+ if(proof.fileId!==fileId||proof.byteLength!==byteLength||proof.blockBytes!==IMPORT_PROOF_BLOCK_BYTES||!Array.isArray(proof.blocks)||Object.keys(proof).some(key=>!['version','fileId','byteLength','blockBytes','blocks'].includes(key)))throw fault('invalid_import_read_proof');
+ let last=-1;const count=Math.ceil(byteLength/IMPORT_PROOF_BLOCK_BYTES);
+ if(proof.blocks.length>count)throw fault('invalid_import_read_proof');
+ for(const block of proof.blocks){if(!block||!Number.isSafeInteger(block.index)||block.index<=last||block.index>=count||typeof block.sha256!=='string'||!/^[a-f0-9]{64}$/.test(block.sha256)||Object.keys(block).some(key=>!['index','sha256'].includes(key)))throw fault('invalid_import_read_proof');last=block.index;}
+ return freeze({version:1 as const,fileId,byteLength,blockBytes:IMPORT_PROOF_BLOCK_BYTES,blocks:proof.blocks.map(block=>({...block}))});
+}
+
+/** Bounded, seekable verified reads for parsers; checkpoints contain hashes, never byte buffers.
+ * Restored evidence must be checked with verify() before a runner resumes business changes. */
+export function createImportReadProof(input:{bytes:Blob;fileId:string;proof?:ImportReadProof;maxResidentBlocks?:number;maxReadBytes?:number}&ImportScheduling){
+ const options=Object.freeze({...input});
+ const {bytes,fileId}=options,residentLimit=options.maxResidentBlocks??4,readLimit=options.maxReadBytes??4*1024*1024;
+ if(!(bytes instanceof Blob)||!fileId||typeof fileId!=='string'||bytes.size>MAX_SOURCE_BYTES||!Number.isSafeInteger(residentLimit)||residentLimit<1||residentLimit>64||!Number.isSafeInteger(readLimit)||readLimit<1||readLimit>4*1024*1024)throw fault('invalid_import_read_source');
+ const initial:ImportReadProof=options.proof?validateImportReadProof(options.proof,fileId,bytes.size):freeze({version:1 as const,fileId,byteLength:bytes.size,blockBytes:IMPORT_PROOF_BLOCK_BYTES,blocks:[]});
+ const hashes=new Map(initial.blocks.map(block=>[block.index,block.sha256]));
+ type Block={index:number;sha256:string;data:Uint8Array};
+ const resident=new Map<number,Block>(),pending=new Map<number,Promise<Block>>(),lifecycle=new AbortController();
+ let closed=false,activeReads=0,proof:ImportReadProof=initial;
+ const check=(signal:AbortSignal)=>{if(closed)throw fault('import_source_closed');if(signal.aborted)throw fault('import_aborted');};
+ function wait<T>(task:Promise<T>,signal:AbortSignal):Promise<T>{
+  check(signal);return new Promise((resolve,reject)=>{const abort=()=>{try{check(signal);}catch(error){reject(error);}};signal.addEventListener('abort',abort,{once:true});task.then(value=>{signal.removeEventListener('abort',abort);try{check(signal);resolve(value);}catch(error){reject(error);}},error=>{signal.removeEventListener('abort',abort);reject(error);});});
+ }
+ function block(index:number):Promise<Block>{
+  const cached=resident.get(index);if(cached){resident.delete(index);resident.set(index,cached);return Promise.resolve(cached);}
+  const active=pending.get(index);if(active)return active;
+  if(pending.size>=4)throw fault('import_read_capacity');
+  const task=(async()=>{
+   const start=index*IMPORT_PROOF_BLOCK_BYTES,end=Math.min(start+IMPORT_PROOF_BLOCK_BYTES,bytes.size),data=new Uint8Array(await bytes.slice(start,end).arrayBuffer());
+   if(closed)throw fault('import_source_closed');if(data.byteLength!==end-start)throw fault('import_source_changed');
+   const hash=createSha256();hash.update(data);const sha256=hash.hex(),expected=hashes.get(index);
+   if(expected!==undefined&&expected!==sha256)throw fault('import_source_changed');
+   const value={index,sha256,data};resident.set(index,value);while(resident.size>residentLimit)resident.delete(resident.keys().next().value!);return value;
+  })();
+  pending.set(index,task);void task.finally(()=>{if(pending.get(index)===task)pending.delete(index);}).catch(()=>{});return task;
+ }
+ const port:ImportByteReader=Object.freeze({byteLength:bytes.size,async read(start:number,end:number,signal:AbortSignal){
+  const combined=AbortSignal.any([signal,lifecycle.signal]);check(combined);
+  if(!Number.isSafeInteger(start)||!Number.isSafeInteger(end)||start<0||end<start||end>bytes.size||end-start>readLimit)throw fault('invalid_import_byte_range');
+  if(start===end)return new Uint8Array();
+  if(activeReads>=4||pending.size>=4&&!resident.has(Math.floor(start/IMPORT_PROOF_BLOCK_BYTES))&&!pending.has(Math.floor(start/IMPORT_PROOF_BLOCK_BYTES)))throw fault('import_read_capacity');
+  activeReads++;try{
+  const output=new Uint8Array(end-start),used:Pick<Block,'index'|'sha256'>[]=[],budget=createImportBudget(combined,options);
+  for(let index=Math.floor(start/IMPORT_PROOF_BLOCK_BYTES);index*IMPORT_PROOF_BLOCK_BYTES<end;index++){
+   const value=await wait(block(index),combined),offset=index*IMPORT_PROOF_BLOCK_BYTES,from=Math.max(start,offset),to=Math.min(end,offset+value.data.byteLength);
+   output.set(value.data.subarray(from-offset,to-offset),from-start);used.push({index:value.index,sha256:value.sha256});await budget.step(128);
+  }
+  check(combined);for(const value of used)if(hashes.has(value.index)&&hashes.get(value.index)!==value.sha256)throw fault('import_source_changed');
+  let changed=false;for(const value of used)if(!hashes.has(value.index)){hashes.set(value.index,value.sha256);changed=true;}
+  if(changed)proof=freeze({version:1 as const,fileId,byteLength:bytes.size,blockBytes:IMPORT_PROOF_BLOCK_BYTES,blocks:[...hashes].sort(([a],[b])=>a-b).map(([index,sha256])=>({index,sha256}))});
+  return output;
+  }finally{activeReads--;}
+ }});
+ return Object.freeze({port,getProof:()=>proof,
+  async verify(signal:AbortSignal){const combined=AbortSignal.any([signal,lifecycle.signal]),budget=createImportBudget(combined,options);check(combined);for(const index of [...hashes.keys()]){await wait(block(index),combined);await budget.step(128);}check(combined);},
+  close(){if(closed)return;closed=true;resident.clear();lifecycle.abort();},
+  stats:()=>Object.freeze({residentBlocks:resident.size,activeReads,pendingBlocks:pending.size,proofBlocks:hashes.size,closed}),
+ });
+}

src/lib/canvas-runtime/imports.ts

@@ -1,3 +1,4 @@
+import {readImportRange,type ImportByteReader} from './import-byte-reader.ts';
 import {createImportBudget,type ImportScheduling} from './import-budget.ts';
 import {validateImportFileRetention,type FileRetirement,type PreviousImportFile} from './import-file-retention.ts';
 import { validateDuplicateCheckpoint, type DuplicateCheckpoint } from '../canvas-core/duplicate-checkpoint.ts';
@@ -67,7 +68,7 @@ export function validateImportJob(job: ImportJob): void {
 const yieldTask=()=>new Promise<void>(resolve=>setTimeout(resolve,0));
 
 /** Seekable bounded NDJSON parser. Fingerprint must identify the host's immutable, durable byte source. */
-export function createNdjsonImportSource(options:{bytes:Blob;fingerprint:string;maxRecordBytes?:number;chunkBytes?:number}&ImportScheduling):ImportSource {
+export function createNdjsonImportSource(options:{bytes:Blob;fingerprint:string;readBytes?:ImportByteReader;maxRecordBytes?:number;chunkBytes?:number}&ImportScheduling):ImportSource {
   const max=options.maxRecordBytes??256*1024,chunk=options.chunkBytes??16*1024;
   if(!(options.bytes instanceof Blob)||!options.fingerprint||!Number.isSafeInteger(max)||max<1||!Number.isSafeInteger(chunk)||chunk<1||chunk>max)throw fault('invalid_import_source');
   const sourceBytes=options.bytes;
@@ -79,7 +80,7 @@ export function createNdjsonImportSource(options:{bytes:Blob;fingerprint:string;
       const parts:Uint8Array[]=[],start=cursor,budget=createImportBudget(signal,options);let total=0,oversized=false;
       while(cursor<sourceBytes.size){
         if(signal.aborted)throw fault('import_aborted');
-        if(cursor<cache.offset||cursor>=cache.offset+cache.bytes.length)cache={offset:cursor,bytes:new Uint8Array(await sourceBytes.slice(cursor,cursor+chunk).arrayBuffer())};
+        if(cursor<cache.offset||cursor>=cache.offset+cache.bytes.length)cache={offset:cursor,bytes:await readImportRange(sourceBytes,options.readBytes,cursor,cursor+chunk,signal)};
         if(signal.aborted)throw fault('import_aborted');
         const begin=cursor-cache.offset,newline=cache.bytes.indexOf(10,begin),end=newline<0?cache.bytes.length:newline,part=cache.bytes.subarray(begin,end);total+=part.length;
         if(total>max){oversized=true;parts.length=0;}else if(!oversized)parts.push(part);

tests/headless/import-read-proof.test.mjs

@@ -0,0 +1,77 @@
+import {test} from 'node:test';
+import assert from 'node:assert/strict';
+import {createImportReadProof,validateImportReadProof,IMPORT_PROOF_BLOCK_BYTES as B} from '../../src/lib/canvas-runtime/import-read-proof.ts';
+import {readImportRange} from '../../src/lib/canvas-runtime/import-byte-reader.ts';
+import {openStreamingDocumentImportSource} from '../../src/lib/canvas-runtime/import-document-stream.ts';
+import {createNdjsonImportSource} from '../../src/lib/canvas-runtime/imports.ts';
+import {createJsonArrayImportSource} from '../../src/lib/canvas-runtime/import-json-array.ts';
+import {fault} from '../../src/lib/canvas-core/values.ts';
+const signal=()=>new AbortController().signal,deferred=()=>{let resolve;const promise=new Promise(done=>resolve=done);return{promise,resolve};};
+const sha=async bytes=>Buffer.from(await crypto.subtle.digest('SHA-256',bytes)).toString('hex');
+const tick=()=>new Promise(resolve=>setTimeout(resolve,0));
+
+test('range reads return exact owned bytes and checkpoint only the aligned blocks they consumed',async()=>{
+ const data=Uint8Array.from({length:B*3+19},(_,i)=>i%251),bytes=new Blob([data]),reader=createImportReadProof({bytes,fileId:'f',maxResidentBlocks:1});
+ const value=await reader.port.read(B-5,B+7,signal());assert.deepEqual(value,data.slice(B-5,B+7));value.fill(0);assert.deepEqual(await reader.port.read(B-5,B+7,signal()),data.slice(B-5,B+7));
+ const proof=reader.getProof();assert.deepEqual(proof.blocks,[{index:0,sha256:await sha(data.slice(0,B))},{index:1,sha256:await sha(data.slice(B,B*2))}]);assert.strictEqual(reader.getProof(),proof);assert.ok(Object.isFrozen(proof)&&Object.isFrozen(proof.blocks));assert.equal(reader.stats().residentBlocks,1);
+ assert.deepEqual(validateImportReadProof(JSON.parse(JSON.stringify(proof)),'f',bytes.size),proof);reader.close();assert.equal(reader.stats().residentBlocks,0);
+});
+
+test('parser 5 emits its first proof-backed record while the unread tail is unavailable',async()=>{
+ const reads=[];class Gated extends Blob{slice(start=0,end=this.size){reads.push([start,end]);if(start>=B)throw Error('tail unavailable');return super.slice(start,end);}}
+ const bytes=new Gated([JSON.stringify({documentFormatVersion:1,projectId:'p',canvases:[{id:'root',parent:null}],elements:[{id:'first'},{id:'later',text:'x'.repeat(B*3)}]})]),reader=createImportReadProof({bytes,fileId:'f'});
+ const source=await openStreamingDocumentImportSource({bytes,readBytes:reader.port,fingerprint:'f',signal:signal()});assert.equal((await source.read(0,signal())).value.record.id,'first');assert.deepEqual(reader.getProof().blocks.map(block=>block.index),[0]);assert.equal(reads.some(([start])=>start>=B),false);reader.close();
+});
+
+test('NDJSON, array and out-of-order document fallback all use the byte reader',async()=>{
+ const cases=[['{"id":"a"}\n',o=>createNdjsonImportSource(o)],['[{"id":"a"}]',o=>createJsonArrayImportSource(o)],[JSON.stringify({elements:[{id:'a'}],documentFormatVersion:1,projectId:'p',canvases:[{id:'root',parent:null}]}),o=>openStreamingDocumentImportSource({...o,signal:signal()})]];
+ for(const[text,open]of cases){const bytes=new Blob([text]),reader=createImportReadProof({bytes,fileId:'f'}),source=await open({bytes,readBytes:reader.port,fingerprint:'f'}),row=await source.read(0,signal());assert.equal((row.value.record??row.value).id,'a');assert.equal(reader.getProof().blocks.length,1);reader.close();}
+});
+
+test('restored proof rejects changed consumed bytes and does not claim to authenticate unread blocks',async()=>{
+ const data=new Uint8Array(B*3).fill(7),first=createImportReadProof({bytes:new Blob([data]),fileId:'f'});await first.port.read(0,1,signal());await first.port.read(B*2,B*2+1,signal());const proof=first.getProof();first.close();
+ const changed=data.slice();changed[B*2]=8;const bad=createImportReadProof({bytes:new Blob([changed]),fileId:'f',proof});await assert.rejects(bad.verify(signal()),/import_source_changed/);assert.strictEqual(bad.getProof().blocks.length,2);bad.close();
+ changed[B*2]=7;changed[B]=8;const unseen=createImportReadProof({bytes:new Blob([changed]),fileId:'f',proof});await unseen.verify(signal());assert.deepEqual(unseen.getProof(),proof);assert.equal((await unseen.port.read(B,B+1,signal()))[0],8);assert.equal(unseen.getProof().blocks.length,3);unseen.close();
+});
+
+test('eviction does not discard the expected hash; changed rereads fail without poisoning a retry',async()=>{
+ const data=new Uint8Array(B*2).fill(4);let altered=false;class Mutable extends Blob{slice(start,end){const part=data.slice(start,end);if(altered&&start===0)part[0]=9;return new Blob([part]);}}
+ const reader=createImportReadProof({bytes:new Mutable([data]),fileId:'f',maxResidentBlocks:1});await reader.port.read(0,1,signal());await reader.port.read(B,B+1,signal());const proof=reader.getProof();altered=true;await assert.rejects(reader.port.read(0,1,signal()),/import_source_changed/);assert.strictEqual(reader.getProof(),proof);altered=false;assert.equal((await reader.port.read(0,1,signal()))[0],4);reader.close();
+});
+
+test('one canceled waiter cannot cancel a shared block or advance its checkpoint',async()=>{
+ const gate=deferred();let reads=0;class Slow extends Blob{slice(start,end){const part=super.slice(start,end);return{async arrayBuffer(){reads++;await gate.promise;return part.arrayBuffer();}};}}
+ const reader=createImportReadProof({bytes:new Slow([new Uint8Array(B)]),fileId:'f'}),controller=new AbortController(),a=reader.port.read(0,1,controller.signal),b=reader.port.read(1,2,signal());controller.abort();await assert.rejects(a,/import_aborted/);assert.equal(reader.getProof().blocks.length,0);assert.equal(reader.stats().pendingBlocks,1);gate.resolve();await b;assert.equal(reads,1);assert.equal(reader.getProof().blocks.length,1);reader.close();
+});
+
+test('canceled multi-block reads and close never publish partial read evidence',async()=>{
+ const gate=deferred(),entered=deferred();class Slow extends Blob{slice(start,end){const part=super.slice(start,end);if(start===0)return part;return{async arrayBuffer(){entered.resolve();await gate.promise;return part.arrayBuffer();}};}}
+ const reader=createImportReadProof({bytes:new Slow([new Uint8Array(B*2)]),fileId:'f'}),controller=new AbortController(),read=reader.port.read(0,B+1,controller.signal);await entered.promise;controller.abort();await assert.rejects(read,/import_aborted/);assert.equal(reader.getProof().blocks.length,0);
+ const late=reader.port.read(B,B+1,signal());reader.close();reader.close();await assert.rejects(late,/import_source_closed/);assert.equal(reader.stats().pendingBlocks,1);gate.resolve();await tick();assert.equal(reader.stats().pendingBlocks,0);assert.equal(reader.stats().residentBlocks,0);assert.equal(reader.getProof().blocks.length,0);await assert.rejects(reader.port.read(0,0,signal()),/import_source_closed/);
+});
+
+test('bounded admission retains occupied native reads until settlement instead of opening more slots on abort',async()=>{
+ const gate=deferred();let reads=0;class Slow extends Blob{slice(start,end){const part=super.slice(start,end);return{async arrayBuffer(){reads++;await gate.promise;return part.arrayBuffer();}};}}
+ const reader=createImportReadProof({bytes:new Slow([new Uint8Array(B*5)]),fileId:'f',maxResidentBlocks:2}),controllers=Array.from({length:4},()=>new AbortController()),pending=controllers.map((controller,i)=>reader.port.read(B*i,B*i+1,controller.signal));
+ await assert.rejects(reader.port.read(B*4,B*4+1,signal()),/import_read_capacity/);assert.equal((await reader.port.read(B*4,B*4,signal())).byteLength,0);controllers.forEach(controller=>controller.abort());await Promise.all(pending.map(task=>assert.rejects(task,/import_aborted/)));assert.equal(reader.stats().activeReads,0);assert.equal(reader.stats().pendingBlocks,4);await assert.rejects(reader.port.read(B*4,B*4+1,signal()),/import_read_capacity/);assert.equal(reads,4);gate.resolve();await tick();assert.ok(reader.stats().residentBlocks<=2);assert.equal(reader.getProof().blocks.length,0);reader.close();
+});
+
+test('zero-length reads do no I/O and invalid proof, identity and ranges fail closed',async()=>{
+ let reads=0;class Count extends Blob{slice(...args){reads++;return super.slice(...args);}}
+ const bytes=new Count(['abc']),reader=createImportReadProof({bytes,fileId:'f'});await reader.port.read(1,1,signal());await reader.port.read(3,3,signal());assert.equal(reads,0);assert.equal(reader.getProof().blocks.length,0);const proof=reader.getProof();
+ for(const bad of [{...proof,version:2},{...proof,fileId:'other'},{...proof,byteLength:4},{...proof,blockBytes:1},{...proof,extra:true},{...proof,blocks:[{index:1,sha256:'a'.repeat(64)}]},{...proof,blocks:[{index:0,sha256:'A'.repeat(64)}]},{...proof,blocks:[{index:0,sha256:'a'.repeat(64)},{index:0,sha256:'a'.repeat(64)}]}])assert.throws(()=>createImportReadProof({bytes,fileId:'f',proof:bad}),/import_read_proof/);
+ for(const[start,end]of [[-1,1],[2,1],[0,4],[.5,1]])await assert.rejects(reader.port.read(start,end,signal()),/invalid_import_byte_range/);reader.close();
+});
+
+test('byte ports cannot return short bytes, wrong source length, or disguise integrity errors as UTF-8 failures',async()=>{
+ const bytes=new Blob(['abcdef']);await assert.rejects(readImportRange(bytes,{byteLength:6,async read(){return new Uint8Array(1);}},0,2,signal()),/import_source_changed/);await assert.rejects(readImportRange(bytes,{byteLength:7,async read(){throw Error('must not run');}},0,2,signal()),/invalid_import_byte_range/);
+ const document=new Blob([JSON.stringify({documentFormatVersion:1,projectId:'p',canvases:[{id:'root',parent:null}],elements:[{id:'a'}]})]);let rejectDecode=false;const port={byteLength:document.size,async read(start,end){if(rejectDecode&&end-start<100)throw fault('import_source_changed');return new Uint8Array(await document.slice(start,end).arrayBuffer());}};
+ const source=await openStreamingDocumentImportSource({bytes:document,readBytes:port,fingerprint:'f',signal:signal()});rejectDecode=true;await assert.rejects(source.read(0,signal()),/import_source_changed/);rejectDecode=false;assert.equal((await source.read(0,signal())).value.record.id,'a');
+});
+
+
+test('transient byte I/O failure does not advance a streaming ordinal into a terminal malformed-tail error',async()=>{
+ const bytes=new Blob([JSON.stringify({documentFormatVersion:1,projectId:'p',canvases:[{id:'root',parent:null}],elements:[{id:'a'},{id:'b',text:'x'.repeat(B*2)}]})]);let fail=true;
+ const readBytes={byteLength:bytes.size,async read(start,end){if(fail&&start>=B)throw Error('offline');return new Uint8Array(await bytes.slice(start,end).arrayBuffer());}};
+ const source=await openStreamingDocumentImportSource({bytes,readBytes,fingerprint:'f',signal:signal()});assert.equal((await source.read(0,signal())).value.record.id,'a');await assert.rejects(source.read(1,signal()),/import_source_read_failed/);fail=false;assert.equal((await source.read(1,signal())).value.record.id,'b');assert.equal(await source.read(2,signal()),null);
+});

tests/headless/imports-engine.test.mjs

@@ -324,3 +324,13 @@ test('legacy format 15 parser 4 document jobs keep their original interpreter an
   engine=await f.open();assert.equal(engine.imports.getJobs().get(id).parserVersion,4);await engine.imports.resume(id);await engine.settle();assert.equal(engine.imports.getJobs().get(id).status,'completed');assert.equal(engine.imports.getJobs().get(id).stableIds['element:legacy-one'],first);assert.equal(engine.getSnapshot().document.elements.size,2);assert.equal(f.image().formatVersion,15);assert.equal(f.errors.length,0);
  }finally{await engine.close();}
 });
+
+test('temporary document byte I/O pauses without consuming a bad-tail ordinal and resumes with edited prefix intact',async()=>{
+ const f=fixture(),engine=await f.open();let fail=true;
+ class Flaky extends Blob{slice(start=0,end=this.size,...rest){if(fail&&end-start===16384&&start>=16384)throw Error('temporary source I/O');return super.slice(start,end,...rest);}}
+ try{
+  const input={documentFormatVersion:1,projectId:'source',canvases:[{id:'source',parent:null}],elements:[{...node('io-first').record,ownerCanvasId:'source'},{...node('io-tail').record,ownerCanvasId:'source',data:{title:'tail',text:'x'.repeat(100000)}}]},id=engine.imports.startDocument(new Flaky([JSON.stringify(input)]),'canvas');
+  await until(engine,snapshot=>snapshot.runtime.imports.get(id).status==='paused');const job=engine.imports.getJobs().get(id),first=job.stableIds['element:io-first'];assert.equal(job.error,'import_source_read_failed');assert.equal(job.eof,false);assert.equal(job.cursor,1);assert.equal(job.failures.length,0);assert.equal(job.committedSourceKeys.length,1);
+  assert.equal(engine.editor.dispatch({type:'element.data',target:{id:first,incarnation:1},patch:{title:'edited prefix'}}).status,'applied');fail=false;await engine.imports.resume(id);await engine.settle();const done=engine.imports.getJobs().get(id);assert.equal(done.status,'completed');assert.equal(done.failures.length,0);assert.equal(engine.getSnapshot().document.elements.size,2);assert.equal(engine.getSnapshot().document.elements.get(first).data.title,'edited prefix');assert.equal(f.server().elements.get(first).data.title,'edited prefix');assert.ok(f.errors.some(error=>error.code==='import_source_read_failed'));decodeRecovery(f.image(),registry);
+ }finally{fail=false;await engine.close();}
+});