feat(canvas): add bounded resumable import coordination

2026-11-08 15:40:10 +08:00
parent be94446f12
commit 160c16afb7
4 changed files with 314 additions and 3 deletions
@@ -0,0 +1,51 @@
+import type { Document, ElementRecord, Id, Json, Registry } from '../canvas-core/types.ts';
+import { remapCopyData } from '../canvas-core/duplicate.ts';
+import { clone, fault } from '../canvas-core/values.ts';
+import type { ImportJob, ImportLine, ImportUnit } from './imports.ts';
+
+export function assertNativeImportTarget(job:ImportJob,document:Document){
+  const source=job.source,canvas=source?document.canvases.get(source.targetCanvasId):undefined;
+  if(!source||!canvas)throw fault('import_target_missing');
+  if(source.targetEntry?(canvas.parent?.entryElementId!==source.targetEntry.id||document.elements.get(source.targetEntry.id)?.incarnation!==source.targetEntry.incarnation):canvas.parent!==null)throw fault('import_target_changed');
+  return source;
+}
+export function assertNativeImportDependencies(job:ImportJob,document:Document,dependencies:readonly string[]){
+  for(const key of dependencies)if(job.committedSourceKeys.includes(key)){
+    const id=job.stableIds[key],binding=job.bindings?.[id];
+    if(!binding)throw fault('import_dependency_changed');
+    if(binding.kind==='element'){if(document.elements.get(id)?.incarnation!==binding.incarnation)throw fault('import_dependency_changed',[id]);}
+    else{const canvas=document.canvases.get(id),entry=binding.entryElementId,entryBinding=entry?job.bindings?.[entry]:undefined;if(!canvas||canvas.parent?.entryElementId!==entry||entry&&(entryBinding?.kind!=='element'||document.elements.get(entry)?.incarnation!==entryBinding.incarnation))throw fault('import_dependency_changed',[id]);}
+  }
+}
+
+/** Native NDJSON records use source IDs; $target names the explicitly chosen existing canvas. */
+export function prepareNativeImport(line:ImportLine,job:ImportJob,reserve:(key:string)=>Id,document:Document,registry:Registry):ImportUnit {
+  const source=assertNativeImportTarget(job,document);
+  const raw=line.value as Record<string,Json>|undefined;
+  if(!raw||Array.isArray(raw)||!['element','canvas'].includes(raw.kind as string)||typeof raw.key!=='string'||!raw.key||raw.key==='$target'||!raw.record||typeof raw.record!=='object'||Array.isArray(raw.record))throw fault('invalid_native_import_record');
+  const record=clone(raw.record) as Record<string,Json>;if(record.id!==raw.key)throw fault('import_source_identity_mismatch');
+  const dependencies=new Set<string>(),elementIds=new Map<Id,Id>(),canvasIds=new Map<Id,Id>();
+  const element=(id:Json):Id=>{if(typeof id!=='string'||!id)throw fault('invalid_import_reference');const key=`element:${id}`;dependencies.add(key);const mapped=reserve(key);elementIds.set(id,mapped);return mapped;};
+  const owner=(id:Json):Id=>{if(id==='$target'){canvasIds.set('$target',source.targetCanvasId);return source.targetCanvasId;}if(typeof id!=='string'||!id)throw fault('invalid_import_reference');const key=`canvas:${id}`;dependencies.add(key);const mapped=reserve(key);canvasIds.set(id,mapped);return mapped;};
+  const key=`${raw.kind}:${raw.key}`,id=reserve(key);
+  if(raw.kind==='canvas'){
+    const parent=record.parent as Record<string,Json>|null;
+    if(!parent||typeof parent!=='object'||Array.isArray(parent))throw fault('invalid_import_canvas');
+    const after={...record,id,parent:{...parent,canvasId:owner(parent.canvasId),entryElementId:element(parent.entryElementId)}};
+    dependencies.delete(key);return{key,dependencies:[...dependencies],changes:[{kind:'canvas',id,before:null,after:after as never}]};
+  }
+  const original=record as unknown as ElementRecord,definition=registry.components.get(original.componentId);
+  if(registry.resolve(original)!=='known'||!definition?.copy||!original.data||typeof original.data!=='object'||Array.isArray(original.data))throw fault('import_component_unavailable');
+  for(const ref of definition.copy.references){
+    let value:Json|undefined=original.data;for(const part of ref.path){if(!value||typeof value!=='object'||!Object.hasOwn(value,part)){value=undefined;break;}value=(value as Record<string,Json>)[part];}
+    if(value===undefined||value===null)continue;
+    if(ref.many&&!Array.isArray(value))throw fault('invalid_import_reference');
+    for(const item of ref.many?value as Json[]:[value]){if(typeof item!=='string'||!item)throw fault('invalid_import_reference');if(ref.target==='element')element(item);else owner(item);}
+  }
+  const after={...record,id,incarnation:1,ownerCanvasId:owner(original.ownerCanvasId),data:remapCopyData(original,registry,elementIds,canvasIds,document)};
+  if(original.structure==='connection'){
+    if(!original.source||!original.target)throw fault('invalid_import_endpoint');
+    Object.assign(after,{source:{...original.source,elementId:element(original.source.elementId)},target:{...original.target,elementId:element(original.target.elementId)}});
+  }else if(original.structure!=='node')throw fault('invalid_import_structure');
+  dependencies.delete(key);return{key,dependencies:[...dependencies],changes:[{kind:'element',id,before:null,after:after as ElementRecord}]};
+}
@@ -0,0 +1,192 @@
+import type { Change, Id, Json, Result } from '../canvas-core/types.ts';
+import { clone, equal, fault, freeze, validJson } from '../canvas-core/values.ts';
+
+export type ImportLine = { readonly cursor: number; readonly nextCursor: number; readonly key: string; readonly value?: Json; readonly error?: string; readonly skip?: boolean };
+export type ImportSource = { readonly fingerprint: string; readonly parserVersion: number; read(cursor: number, signal: AbortSignal): Promise<ImportLine | null> };
+export type ImportUnit = { readonly key: string; readonly dependencies: readonly string[]; readonly changes: readonly Change[] };
+export type ImportLocator = { readonly key: string; readonly cursor: number; readonly nextCursor: number; readonly dependencies: readonly string[]; readonly digest: string; readonly byteLength: number; readonly changeCount: number };
+export type ImportFailure = { readonly key: string; readonly cursor: number; readonly code: string; readonly dependencies?: readonly string[] };
+export type NativeImportSource = { readonly kind: 'native-ndjson'; readonly name?:string; readonly fileId: Id; readonly byteLength: number; readonly targetCanvasId: Id; readonly targetEntry?: {readonly id:Id;readonly incarnation:number} };
+export type ImportBinding = {readonly kind:'element';readonly incarnation:number}|{readonly kind:'canvas';readonly entryElementId?:Id};
+export type ImportJob = {
+  readonly version: 1; readonly id: Id; readonly projectId: Id; readonly sourceFingerprint: string; readonly parserVersion: number;
+  readonly source?: NativeImportSource;
+  readonly generation: number; readonly status: 'paused' | 'running' | 'completed' | 'partial' | 'cancelled'; readonly cursor: number; readonly eof: boolean;
+  readonly bindings?: Readonly<Record<Id,ImportBinding>>;
+  readonly stableIds: Readonly<Record<string, Id>>; readonly pending: readonly ImportLocator[]; readonly committedSourceKeys: readonly string[]; readonly failures: readonly ImportFailure[]; readonly error?: string;
+};
+export function createImportJob(input: Pick<ImportJob, 'id' | 'projectId' | 'sourceFingerprint' | 'parserVersion' | 'source'>): ImportJob {
+  if (![input.id,input.projectId,input.sourceFingerprint].every(value=>typeof value==='string'&&!!value) || !Number.isSafeInteger(input.parserVersion) || input.parserVersion<1) throw fault('invalid_import_job');
+  return freeze({ id: input.id, projectId: input.projectId, sourceFingerprint: input.sourceFingerprint, parserVersion: input.parserVersion, ...(input.source?{source:clone(input.source)}:{}), version: 1, generation: 0, status: 'paused', cursor: 0, eof: false, bindings: {}, stableIds: {}, pending: [], committedSourceKeys: [], failures: [] });
+}
+export function validateImportJob(job: ImportJob): void {
+  if(!job||typeof job!=='object'||!validJson(job)||!job.stableIds||typeof job.stableIds!=='object'||Array.isArray(job.stableIds)||![job.pending,job.committedSourceKeys,job.failures].every(Array.isArray)||job.error!==undefined&&typeof job.error!=='string')throw fault('invalid_import_job');
+  if (job.version !== 1) throw fault('import_interpreter_unavailable');
+  createImportJob(job);
+  if (!Number.isSafeInteger(job.generation) || job.generation<0 || !Number.isSafeInteger(job.cursor) || job.cursor<0 || typeof job.eof!=='boolean' || !['paused','running','completed','partial','cancelled'].includes(job.status)) throw fault('invalid_import_job');
+  if(job.source&&(job.source.kind!=='native-ndjson'||job.source.name!==undefined&&(typeof job.source.name!=='string'||job.source.name.length>255)||!job.source.fileId||!job.source.targetCanvasId||!Number.isSafeInteger(job.source.byteLength)||job.source.byteLength<0||job.source.targetEntry&&(!job.source.targetEntry.id||!Number.isSafeInteger(job.source.targetEntry.incarnation)||job.source.targetEntry.incarnation<1)))throw fault('invalid_import_source');
+  const keys=new Set<string>(),ids=Object.values(job.stableIds),stableIdSet=new Set(ids);
+  if (stableIdSet.size!==ids.length || ids.some(id=>typeof id!=='string'||!id) || Object.keys(job.stableIds).some(key=>!key)) throw fault('invalid_import_job');
+  for(const[id,binding]of Object.entries(job.bindings??{}))if(!id||!binding||!stableIdSet.has(id)||!['element','canvas'].includes(binding.kind)||binding.kind==='element'&&(!Number.isSafeInteger(binding.incarnation)||binding.incarnation<1)||binding.kind==='canvas'&&binding.entryElementId!==undefined&&typeof binding.entryElementId!=='string')throw fault('invalid_import_job');
+  for(const key of job.committedSourceKeys){if(!key||keys.has(key))throw fault('invalid_import_job');keys.add(key);}
+  if(job.source)for(const key of job.committedSourceKeys){const id=job.stableIds[key],binding=job.bindings?.[id];if(!Object.hasOwn(job.stableIds,key)||!binding||!key.startsWith(`${binding.kind}:`)||binding.kind==='canvas'&&(!binding.entryElementId||job.bindings?.[binding.entryElementId]?.kind!=='element'))throw fault('invalid_import_job');}
+  for(const unit of job.pending){if(!unit||typeof unit.key!=='string'||!unit.key||!Array.isArray(unit.dependencies)||keys.has(unit.key)||!Number.isSafeInteger(unit.cursor)||unit.cursor<0||!Number.isSafeInteger(unit.nextCursor)||unit.nextCursor<=unit.cursor||unit.nextCursor>job.cursor||!(/^[a-f0-9]{64}$/.test(unit.digest))||!Number.isSafeInteger(unit.byteLength)||unit.byteLength<1||!Number.isSafeInteger(unit.changeCount)||unit.changeCount<1||unit.changeCount>10000||new Set(unit.dependencies).size!==unit.dependencies.length||unit.dependencies.some(key=>typeof key!=='string'||!key))throw fault('invalid_import_job');keys.add(unit.key);}
+  if(job.status==='partial'&&!job.eof)throw fault('invalid_import_job');
+  if(job.status==='completed'&&(!job.eof||job.pending.length||job.failures.length))throw fault('invalid_import_job');
+  for(const failure of job.failures)if(!failure||typeof failure.key!=='string'||!failure.key||typeof failure.code!=='string'||!failure.code||!Number.isSafeInteger(failure.cursor)||failure.cursor<0||failure.cursor>job.cursor)throw fault('invalid_import_job');
+}
+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;yieldTask?:()=>Promise<void>}):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,sourceYield=options.yieldTask??yieldTask;
+  let cache={offset:-1,bytes:new Uint8Array()};
+  return Object.freeze({fingerprint:options.fingerprint,parserVersion:1,
+    async read(cursor:number,signal:AbortSignal):Promise<ImportLine|null>{
+      if(!Number.isSafeInteger(cursor)||cursor<0||cursor>sourceBytes.size)throw fault('invalid_import_cursor');
+      if(signal.aborted)throw fault('import_aborted');if(cursor===sourceBytes.size)return null;
+      const parts:Uint8Array[]=[],start=cursor;let total=0,oversized=false,scanned=0;
+      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(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);
+        cursor=cache.offset+end+(newline<0?0:1);scanned+=end-begin;
+        if(newline>=0||cursor===sourceBytes.size)break;
+        if(scanned>=64*1024){scanned=0;await sourceYield();}
+      }
+      const base={cursor:start,nextCursor:cursor,key:`byte:${start}`};
+      if(oversized)return{...base,error:'record_too_large'};
+      const bytes=new Uint8Array(total);let offset=0;for(const part of parts){bytes.set(part,offset);offset+=part.length;}
+      let text:string;try{text=new TextDecoder('utf-8',{fatal:true}).decode(bytes);}catch{return{...base,error:'invalid_utf8'};}
+      if(!text.trim())return{...base,skip:true};
+      try{const value=JSON.parse(text);return validJson(value)?{...base,value}:{...base,error:'invalid_json_value'};}catch{return{...base,error:'invalid_json'};}
+    },
+  });
+}
+
+export type ImportRunnerPorts={
+  source:ImportSource;getJob():ImportJob;createId():Id;
+  /** Synchronous compare-and-publish; never write a second private job store. */
+  publish(next:ImportJob,expectedGeneration:number):void;
+  /** Pure decoding; reserve() returns stable source-key identities. No upload/run effects here. */
+  prepare(line:ImportLine,job:ImportJob,reserve:(sourceKey:string)=>Id):ImportUnit;
+  /** Publish all document changes AND next job in the same StateRoot, or publish nothing. */
+  commit(units:readonly ImportUnit[],next:ImportJob,expectedGeneration:number):Result;
+  persist():Promise<void>;onError(error:unknown):void;
+  /** Host must durably gate every network write on the same document/job cut. */
+  optimisticLocal?:boolean;
+  now?:()=>number;yieldTask?:()=>Promise<void>;frameBudgetMs?:number;maxPending?:number;maxPendingBytes?:number;maxBatch?:number;
+};
+
+/** Cooperative producer/consumer. Pending work stores seekable locators, not unbounded decoded payloads. */
+export function createImportRunner(ports:ImportRunnerPorts){
+  const initial=ports.getJob(),identity={id:initial.id,projectId:initial.projectId,sourceFingerprint:initial.sourceFingerprint,parserVersion:initial.parserVersion},budget=ports.frameBudgetMs??6,maxPending=ports.maxPending??4096,maxBytes=ports.maxPendingBytes??1024*1024,maxBatch=ports.maxBatch??32,now=ports.now??(()=>performance.now());
+  validateImportJob(initial);
+  if(!Number.isFinite(budget)||budget<=0||![maxPending,maxBytes,maxBatch].every(value=>Number.isSafeInteger(value)&&value>0))throw fault('invalid_import_budget');
+  let costPerUnit:number|undefined;
+  let active:Promise<void>|undefined,controller:AbortController|undefined,closed=false,restart=false;const holds=new Set<symbol>();
+  const report=(error:unknown)=>{try{ports.onError(error);}catch{/* Reporting cannot change imported content. */}};
+  function current(){const job=ports.getJob();if(job.id!==identity.id||job.projectId!==identity.projectId||job.sourceFingerprint!==identity.sourceFingerprint||job.parserVersion!==identity.parserVersion)throw fault('import_identity_changed');return job;}
+  function next(job:ImportJob,patch:Partial<ImportJob>):ImportJob{const value={...job,...patch,generation:job.generation+1};if(value.error===undefined)delete value.error;return freeze(value);}
+  function publish(job:ImportJob,patch:Partial<ImportJob>){const value=next(job,patch);ports.publish(value,job.generation);if(!equal(current(),value))throw fault('invalid_import_publish');return value;}
+  function prepare(line:ImportLine,job:ImportJob){
+    const ids:Record<string,Id>=Object.assign(Object.create(null),job.stableIds),used=new Set(Object.values(ids));
+    const reserve=(key:string)=>{if(typeof key!=='string'||!key)throw fault('invalid_import_key');if(Object.hasOwn(ids,key))return ids[key];const id=ports.createId();if(typeof id!=='string'||!id||used.has(id))throw fault('duplicate_import_identity');ids[key]=id;used.add(id);return id;};
+    const unit=ports.prepare(line,job,reserve);
+    if(!validJson(unit)||typeof unit.key!=='string'||!unit.key||!Array.isArray(unit.changes)||!Array.isArray(unit.dependencies)||!unit.changes.length||unit.changes.length>10000||new Set(unit.dependencies).size!==unit.dependencies.length||unit.dependencies.some(key=>typeof key!=='string'||!key))throw fault('invalid_import_unit');
+    for(const change of unit.changes)if(change.kind!=='field'&&change.before===null&&change.after&&!used.has(change.id))throw fault('unstable_import_identity');
+    if(new TextEncoder().encode(JSON.stringify(unit)).byteLength>maxBytes)throw fault('import_unit_capacity');
+    return{unit:freeze(clone(unit)),ids:{...ids}};
+  }
+  const digest=async(unit:ImportUnit)=>[...new Uint8Array(await crypto.subtle.digest('SHA-256',new TextEncoder().encode(JSON.stringify(unit))))].map(value=>value.toString(16).padStart(2,'0')).join('');
+  function completed(job:ImportJob,units:readonly ImportUnit[],pending:readonly ImportLocator[]){
+    const bindings:Record<Id,ImportBinding>=Object.assign(Object.create(null),job.bindings??{});
+    for(const unit of units)for(const change of unit.changes)if(change.kind==='element'&&change.before===null&&change.after)bindings[change.id]={kind:'element',incarnation:change.after.incarnation};else if(change.kind==='canvas'&&change.before===null&&change.after)bindings[change.id]={kind:'canvas',...(change.after.parent?{entryElementId:change.after.parent.entryElementId}:{})};
+    return next(job,{pending,committedSourceKeys:[...job.committedSourceKeys,...units.map(unit=>unit.key)],bindings:{...bindings}});
+  }
+  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;
+    if(!ports.optimisticLocal)await ports.persist();
+    while(!closed&&!signal.aborted&&!holds.size){
+      let job=current();if(job.status!=='running')return;
+      if(ports.source.fingerprint!==job.sourceFingerprint||ports.source.parserVersion!==job.parserVersion)throw fault('import_source_mismatch');
+      if(didWork&&now()-frame>=budget){await(ports.yieldTask??yieldTask)();frame=now();didWork=false;continue;}
+      didWork=true;
+      const committed=new Set(job.committedSourceKeys),chosen:ImportLocator[]=[];let chosenBytes=0,chosenChanges=0;
+      const batchLimit=Math.max(1,Math.min(maxBatch,costPerUnit===undefined?1:Math.floor(budget/(costPerUnit*1.5))));
+      for(const locator of job.pending)if(chosen.length<batchLimit&&chosenBytes+locator.byteLength<=maxBytes&&chosenChanges+locator.changeCount<=10000&&locator.dependencies.every(key=>committed.has(key)||chosen.some(unit=>unit.key===key))){chosen.push(locator);chosenBytes+=locator.byteLength;chosenChanges+=locator.changeCount;}
+      if(chosen.length){
+        const units:ImportUnit[]=[];let failed:ImportFailure|undefined;
+        for(const locator of chosen){
+          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');
+          const value=prepare(line,job);
+          if(value.unit.key!==locator.key||!equal(value.unit.dependencies,locator.dependencies)||!equal(value.ids,job.stableIds)||value.unit.changes.length!==locator.changeCount||new TextEncoder().encode(JSON.stringify(value.unit)).byteLength!==locator.byteLength||await digest(value.unit)!==locator.digest)throw fault('import_source_changed');
+          if(closed||signal.aborted||current().generation!==job.generation)return;
+          units.push(value.unit);
+        }
+        let proposed=completed(job,units,job.pending.filter(unit=>!chosen.includes(unit)));
+        let result=commit(units,job,proposed);
+        if(result.status==='rejected'){
+          // Isolate a failed record without discarding valid independent records in the same candidate batch.
+          for(let index=0;index<units.length;index++){
+            job=current();const unit=units[index],locator=chosen[index];
+            if(!unit.dependencies.every(key=>job.committedSourceKeys.includes(key)))continue;
+            proposed=completed(job,[unit],job.pending.filter(value=>value.key!==unit.key));
+            result=commit([unit],job,proposed);
+            if(result.status==='rejected'){failed={key:unit.key,cursor:locator.cursor,code:result.code};publish(job,{pending:job.pending.filter(value=>value.key!==unit.key),failures:[...job.failures,failed]});}
+            await ports.persist();if(closed||signal.aborted||current().status!=='running')return;
+            if(now()-frame>=budget){await(ports.yieldTask??yieldTask)();frame=now();}
+          }
+        }else await ports.persist();
+        continue;
+      }
+      if(job.eof){
+        const unresolved=job.pending.map(unit=>({key:unit.key,cursor:unit.cursor,code:'unresolved_dependency',dependencies:unit.dependencies.filter(key=>!committed.has(key))}));
+        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();
+    }
+  }
+  function launch(){
+    if(closed||holds.size)return Promise.resolve();if(active){restart=true;return active;}
+    restart=false;controller=new AbortController();
+    active=pump(controller.signal).catch(async error=>{
+      if(closed||controller?.signal.aborted)return;
+      try{const job=current();if(job.status==='running'){publish(job,{status:'paused',error:(error as {code?:string}).code??'import_failed'});try{await ports.persist();}catch{/* Storage may still be unavailable. */}}}catch{/* A replaced identity cannot receive an old error. */}
+      report(error);
+    }).finally(()=>{active=undefined;controller=undefined;if(restart&&!closed&&!holds.size){try{if(current().status==='running')void launch();}catch{/* Replaced scope stays untouched. */}}});
+    return active;
+  }
+  return Object.freeze({
+    start(){if(closed)throw fault('import_runner_closed');const job=current();if(['completed','partial','cancelled'].includes(job.status))return Promise.resolve();if(job.status!=='running')publish(job,{status:'running',error:undefined});return launch();},
+    hold(){const token=Symbol();holds.add(token);controller?.abort();return()=>{if(!holds.delete(token)||holds.size||closed)return;if(current().status==='running')void launch();};},
+    async pause(){const job=current();if(job.status==='running'){controller?.abort();publish(job,{status:'paused'});await ports.persist();}},
+    async cancel(){const job=current();if(['running','paused'].includes(job.status)){controller?.abort();publish(job,{status:'cancelled',error:undefined});await ports.persist();}},
+    async settle(){while(active)await active;},
+    async close(){closed=true;controller?.abort();while(active)await active;},
+  });
+}