diff --git a/src/lib/canvas-runtime/imports.ts b/src/lib/canvas-runtime/imports.ts
index 759bb43..f93ec83 100644
--- a/src/lib/canvas-runtime/imports.ts
+++ b/src/lib/canvas-runtime/imports.ts
@@ -118,6 +118,18 @@ export function createImportRunner(ports:ImportRunnerPorts){
   function commit(units:readonly ImportUnit[],job:ImportJob,value:ImportJob){const start=now(),result=ports.commit(units,value,job.generation),cost=Math.max(.01,(now()-start)/units.length);costPerUnit=costPerUnit===undefined?cost:Math.max(cost,costPerUnit*.8+cost*.2);if(!['applied','rejected'].includes(result?.status)||!equal(current(),result.status==='applied'?value:job))throw fault('invalid_import_commit');return result;}
   async function pump(signal:AbortSignal){
     let frame=now(),didWork=false;
+    type ReadResult={kind:'line';line:ImportLine|null}|{kind:'error';error:unknown};
+    let reading:{cursor:number;result:Promise<ReadResult>}|undefined;
+    const decoded=new Map<string,{unit:ImportUnit;digest:string;bytes:number}>();let decodedBytes=0;
+    function cache(key:string,value:{unit:ImportUnit;digest:string;bytes:number}){const previous=decoded.get(key);if(previous)decodedBytes-=previous.bytes;decoded.delete(key);decoded.set(key,value);decodedBytes+=value.bytes;while(decoded.size>maxBatch||decodedBytes>maxBytes){const[key,value]=decoded.entries().next().value!;decoded.delete(key);decodedBytes-=value.bytes;}}
+    function forget(key:string){const value=decoded.get(key);if(value){decodedBytes-=value.bytes;decoded.delete(key);}}
+    async function read(cursor:number,flushReady:boolean):Promise<ReadResult|{kind:'flush'}>{
+      if(reading&&reading.cursor!==cursor)throw fault('invalid_import_cursor');
+      reading??={cursor,result:Promise.resolve().then(()=>ports.source.read(cursor,signal)).then(line=>({kind:'line' as const,line}),error=>({kind:'error' as const,error}))};
+      let timer:ReturnType<typeof setTimeout>|undefined;
+      const result=flushReady?await Promise.race([reading.result,new Promise<{kind:'flush'}>(resolve=>{timer=setTimeout(()=>resolve({kind:'flush'}),0);})]):await reading.result;
+      if(timer!==undefined)clearTimeout(timer);if(result.kind!=='flush')reading=undefined;return result;
+    }
     if(!ports.optimisticLocal)await ports.persist();
     while(!closed&&!signal.aborted&&!holds.size){
       let job=current();if(job.status!=='running')return;
@@ -130,6 +142,10 @@ export function createImportRunner(ports:ImportRunnerPorts){
       if(chosen.length){
         const units:ImportUnit[]=[];let failed:ImportFailure|undefined;
         for(const locator of chosen){
+          const cached=decoded.get(locator.key);
+          if(cached&&cached.digest===locator.digest){units.push(cached.unit);continue;}
+          // A source need not support concurrent reads. Finish any speculative read before seeking.
+          if(reading)await reading.result;
           const line=await ports.source.read(locator.cursor,signal);
           if(closed||signal.aborted||current().generation!==job.generation)return;
           if(!line||line.error||line.skip||line.nextCursor!==locator.nextCursor)throw fault('import_source_changed');
@@ -152,6 +168,7 @@ export function createImportRunner(ports:ImportRunnerPorts){
             if(now()-frame>=budget){await(ports.yieldTask??yieldTask)();frame=now();}
           }
         }else await ports.persist();
+        for(const unit of units)forget(unit.key);
         continue;
       }
       if(job.eof){
@@ -159,23 +176,38 @@ export function createImportRunner(ports:ImportRunnerPorts){
         publish(job,{status:job.failures.length||unresolved.length?'partial':'completed',failures:[...job.failures,...unresolved]});await ports.persist();return;
       }
       if(job.pending.length>=maxPending||new TextEncoder().encode(JSON.stringify(job.pending)).byteLength>=maxBytes){publish(job,{status:'paused',error:'dependency_capacity'});await ports.persist();return;}
-      const line=await ports.source.read(job.cursor,signal);
-      if(closed||signal.aborted||current().generation!==job.generation)return;
-      if(!line){publish(job,{eof:true});await ports.persist();continue;}
-      if(line.cursor!==job.cursor||!Number.isSafeInteger(line.nextCursor)||line.nextCursor<=job.cursor)throw fault('invalid_import_cursor');
-      if(line.skip){publish(job,{cursor:line.nextCursor});await ports.persist();continue;}
-      if(line.error){publish(job,{cursor:line.nextCursor,failures:[...job.failures,{key:line.key,cursor:line.cursor,code:line.error}]});await ports.persist();continue;}
-      let patch:Partial<ImportJob>;
-      try{
-        const prepared=prepare(line,job),key=prepared.unit.key;
-        if(job.committedSourceKeys.includes(key)||job.pending.some(unit=>unit.key===key)||job.failures.some(failure=>failure.key===key))throw fault('duplicate_import_source_key');
-        const locator={key,cursor:line.cursor,nextCursor:line.nextCursor,dependencies:prepared.unit.dependencies,digest:await digest(prepared.unit),byteLength:new TextEncoder().encode(JSON.stringify(prepared.unit)).byteLength,changeCount:prepared.unit.changes.length};
-        patch={cursor:line.nextCursor,stableIds:prepared.ids,pending:[...job.pending,locator]};
-      }catch(error){patch={cursor:line.nextCursor,failures:[...job.failures,{key:line.key,cursor:line.cursor,code:(error as {code?:string}).code??'import_decode_failed'}]};}
-      if(closed||signal.aborted||current().generation!==job.generation)return;
-      publish(job,patch);
-      const staged=patch.pending?.at(-1),immediatelyReady=staged?.dependencies.every(key=>job.committedSourceKeys.includes(key));
-      if(!ports.optimisticLocal||!immediatelyReady)await ports.persist();
+      // Stage a bounded parser slice, then publish its cursor/IDs/locators once. This is
+      // unpublished transaction preparation, never another durable task state store.
+      const base=job;let staged=job,records=0;
+      while(!staged.eof&&records<maxBatch&&staged.pending.length<maxPending&&new TextEncoder().encode(JSON.stringify(staged.pending)).byteLength<maxBytes){
+        const ready=staged.pending.some(unit=>unit.dependencies.every(key=>committed.has(key)));
+        const result=await read(staged.cursor,ready);
+        if(closed||signal.aborted||current().generation!==base.generation)return;
+        if(result.kind==='flush')break;
+        if(result.kind==='error'){
+          if(ready){reading={cursor:staged.cursor,result:Promise.resolve(result)};break;}
+          if(staged!==base){const{generation,...patch}=staged;publish(base,patch);}throw result.error;
+        }
+        const line=result.line;
+        if(!line){staged={...staged,eof:true};break;}
+        if(line.cursor!==staged.cursor||!Number.isSafeInteger(line.nextCursor)||line.nextCursor<=staged.cursor)throw fault('invalid_import_cursor');
+        let patch:Partial<ImportJob>;
+        if(line.skip)patch={cursor:line.nextCursor};
+        else if(line.error)patch={cursor:line.nextCursor,failures:[...staged.failures,{key:line.key,cursor:line.cursor,code:line.error}]};
+        else try{
+          const prepared=prepare(line,staged),key=prepared.unit.key;
+          if(staged.committedSourceKeys.includes(key)||staged.pending.some(unit=>unit.key===key)||staged.failures.some(failure=>failure.key===key))throw fault('duplicate_import_source_key');
+          const locator={key,cursor:line.cursor,nextCursor:line.nextCursor,dependencies:prepared.unit.dependencies,digest:await digest(prepared.unit),byteLength:new TextEncoder().encode(JSON.stringify(prepared.unit)).byteLength,changeCount:prepared.unit.changes.length};
+          cache(key,{unit:prepared.unit,digest:locator.digest,bytes:locator.byteLength});
+          patch={cursor:line.nextCursor,stableIds:prepared.ids,pending:[...staged.pending,locator]};
+        }catch(error){patch={cursor:line.nextCursor,failures:[...staged.failures,{key:line.key,cursor:line.cursor,code:(error as {code?:string}).code??'import_decode_failed'}]};}
+        if(closed||signal.aborted||current().generation!==base.generation)return;
+        staged={...staged,...patch};records++;
+        // First ready record displays immediately; subsequent slices amortize publication.
+        if(costPerUnit===undefined&&staged.pending.some(unit=>unit.dependencies.every(key=>committed.has(key)))||now()-frame>=budget)break;
+      }
+      if(staged!==base){const{generation,...patch}=staged;publish(base,patch);if(!ports.optimisticLocal||!staged.pending.some(unit=>unit.dependencies.every(key=>committed.has(key))))await ports.persist();}
+
     }
   }
   function launch(){
diff --git a/tests/headless/imports-engine.test.mjs b/tests/headless/imports-engine.test.mjs
index 1e177e4..a2aa8aa 100644
--- a/tests/headless/imports-engine.test.mjs
+++ b/tests/headless/imports-engine.test.mjs
@@ -127,12 +127,12 @@ test('read-only policy rejects copy/import admission without storing a source or
 });
 
 
-test('preparing copy checkpoints source and allocated identities, reopens paused and reconstructs the same mapping',async()=>{
- const f=fixture();let engine=await f.open();try{
+test('preparing copy checkpoints source and allocated identities, reopens paused and reconstructs the same mapping',async(t)=>{
+ const f=fixture(),save=f.storage.commit;let durableCuts=0;f.storage.commit=async(...args)=>{durableCuts++;return save(...args);};let engine=await f.open();try{
   await seedCopy(engine);assert.equal(engine.editor.transaction(tx=>{for(let i=0;i<1000;i++)tx.dispatch({type:'element.create',element:copyNode(`extra-${i}`,'child')});}).status,'applied');await engine.settle();
   const id=await engine.imports.duplicate([{id:'F',incarnation:1}]);await until(engine,snapshot=>(snapshot.runtime.imports.get(id).preparation?.allocationIds.length??0)>0);await engine.imports.pause(id);await engine.settle();const checkpoint=engine.imports.getJobs().get(id);assert.equal(checkpoint.status,'paused');assert.equal(checkpoint.source.kind,'duplicate-snapshot');assert.ok(checkpoint.preparation.allocationIds.length>0);assert.ok(checkpoint.preparation.steps>0);assert.equal(checkpoint.committedSourceKeys.length,0);decodeRecovery(f.image(),registry);
   engine.editor.dispatch({type:'element.data',target:{id:'F',incarnation:1},patch:{title:'new source title'}});await engine.settle();await engine.close();engine=await f.open();assert.equal(engine.imports.getJobs().get(id).status,'paused');assert.deepEqual(engine.imports.getJobs().get(id).preparation.allocationIds,checkpoint.preparation.allocationIds);await engine.imports.resume(id);await engine.settle();
-  const job=engine.imports.getJobs().get(id);assert.equal(job.status,'completed');assert.deepEqual(Object.values(job.stableIds).slice(0,checkpoint.preparation.allocationIds.length),checkpoint.preparation.allocationIds);assert.equal(engine.getSnapshot().document.elements.get(job.stableIds['element:F']).data.title,'F');assert.equal(engine.getSnapshot().document.elements.get('F').data.title,'new source title');assert.equal(job.preparation,undefined);assert.equal(f.errors.length,0);
+  const job=engine.imports.getJobs().get(id);assert.equal(job.status,'completed');assert.deepEqual(Object.values(job.stableIds).slice(0,checkpoint.preparation.allocationIds.length),checkpoint.preparation.allocationIds);assert.equal(engine.getSnapshot().document.elements.get(job.stableIds['element:F']).data.title,'F');assert.equal(engine.getSnapshot().document.elements.get('F').data.title,'new source title');assert.equal(job.preparation,undefined);assert.equal(f.errors.length,0);t.diagnostic(JSON.stringify({imported:job.committedSourceKeys.length,durableCuts,requests:f.requests.length}));
  }finally{await engine.close();}
 });
 
diff --git a/tests/headless/imports.test.mjs b/tests/headless/imports.test.mjs
index a2f01e7..7c6c60c 100644
--- a/tests/headless/imports.test.mjs
+++ b/tests/headless/imports.test.mjs
@@ -66,3 +66,26 @@ test('dependency backpressure pauses with locators intact and a larger budget re
 test('future interpreter versions and malformed job cuts fail before any source read',()=>{
  const job=createImportJob({id:'i',projectId:'p',sourceFingerprint:'s',parserVersion:1});assert.throws(()=>validateImportJob({...job,version:2}),/import_interpreter_unavailable/);assert.throws(()=>validateImportJob({...job,stableIds:{a:'same',b:'same'}}),/invalid_import_job/);assert.throws(()=>validateImportJob({...job,status:'completed'}),/invalid_import_job/);
 });
+
+test('fast independent records stage bounded publications and commits while the first record remains immediate',async()=>{
+ const f=fixture(Array.from({length:80},(_,index)=>({id:`batch-${index}`})),{ports:{now:()=>0,maxBatch:16,optimisticLocal:true}}),publish=f.ports.publish;let publications=0;f.ports.publish=(...args)=>{publications++;return publish(...args);};
+ await f.runner.start();assert.equal(f.root.job.status,'completed');assert.equal(f.root.document.elements.size,80);assert.deepEqual(f.calls[0],['batch-0']);assert.ok(f.calls.some(group=>group.length>1));assert.ok(f.calls.every(group=>group.length<=16));assert.ok(publications<20,`${publications} metadata publications for 80 records`);assert.equal(f.errors.length,0);validateImportJob(f.root.job);
+});
+
+test('a stalled speculative read cannot hide the ready staged prefix or require concurrent source seeks',async()=>{
+ const gate=deferred(),entered=deferred(),source=createNdjsonImportSource({bytes:bytes([{id:'first'},{id:'second'},{id:'third'}]),fingerprint:'immutable-source'}),signal=new AbortController().signal,first=await source.read(0,signal),second=await source.read(first.nextCursor,signal);let active=0;
+ const f=fixture([],{source:{...source,async read(cursor,signal){assert.equal(active,0,'concurrent source seeks');active++;try{if(cursor===second.nextCursor){entered.resolve();await gate.promise;}return await source.read(cursor,signal);}finally{active--;}}},ports:{now:()=>0,optimisticLocal:true}});
+ const running=f.runner.start();await entered.promise;
+ for(let i=0;i<100&&f.root.document.elements.size<2;i++)await new Promise(resolve=>setTimeout(resolve,1));assert.equal(f.root.document.elements.size,2);assert.deepEqual(f.root.job.committedSourceKeys,['first','second']);assert.equal(f.root.job.eof,false);await f.runner.cancel();gate.resolve();await running;assert.equal(f.root.document.elements.size,2);assert.equal(f.root.job.status,'cancelled');assert.equal(f.errors.length,0);
+});
+
+test('cancelling during a staged decode discards unpublished cursor and identity changes',async()=>{
+ const f=fixture([{id:'first'},{id:'second'},{id:'third'}],{ports:{now:()=>0,optimisticLocal:true}}),prepare=f.ports.prepare;let cancelled;
+ f.ports.prepare=(line,job,reserve)=>{const unit=prepare(line,job,reserve);if(line.value.id==='second')queueMicrotask(()=>{cancelled=f.runner.cancel();});return unit;};
+ await f.runner.start();await cancelled;assert.equal(f.root.job.status,'cancelled');assert.deepEqual(f.root.job.committedSourceKeys,['first']);assert.equal(f.root.document.elements.size,1);assert.equal(Object.hasOwn(f.root.job.stableIds,'second'),false);assert.equal(f.root.job.pending.length,0);assert.equal(f.errors.length,0);validateImportJob(f.root.job);
+});
+
+test('a source I/O failure is reported after publishing all already prepared ready records',async()=>{
+ const raw=createNdjsonImportSource({bytes:bytes([{id:'first'},{id:'second'}]),fingerprint:'immutable-source'}),first=await raw.read(0,new AbortController().signal),second=await raw.read(first.nextCursor,new AbortController().signal),f=fixture([],{source:{...raw,async read(cursor,signal){if(cursor===second.nextCursor)throw fault('source_offline');return raw.read(cursor,signal);}},ports:{now:()=>0,optimisticLocal:true}});
+ await f.runner.start();assert.equal(f.root.job.status,'paused');assert.equal(f.root.job.error,'source_offline');assert.deepEqual(f.root.job.committedSourceKeys,['first','second']);assert.equal(f.root.document.elements.size,2);assert.equal(f.root.job.pending.length,0);assert.equal(f.errors.length,1);
+});