docs(runtime): explain progressive import publication and recovery

2026-11-12 14:40:09 +08:00
parent 9c1dd28f3a
commit e425b28064
1 changed files with 61 additions and 25 deletions
@@ -1,7 +1,8 @@
 /**
 * 渐进导入的持久job合同、NDJSON解释器和合作式runner。解析得到稳定ID/locator，依赖闭合后原子加入文档，
 * job进度与文档同Root发布；暂停/取消保留已提交内容，失败逐记录或按不可分依赖组报告partial。
- * runner不拥有第二份持久job store：ports.getJob是事实源，所有await后的业务采纳受generation/身份检查。
+ * runner不拥有第二份持久job store：ports.getJob是事实源，关键异步结果采纳前重查generation/身份。
+ * 本地加入、可靠保存与后端发送分属不同边界；此模块不调用网络写入，也不把 EOF 当成整源已封存。
 */
 import {dataArray,dataRecord} from '../canvas-core/data-containers.ts';
 import {createImportBatches} from './import-batches.ts';
@@ -31,7 +32,9 @@ export type LoadingImportAdmission={readonly version:1;readonly sourceToken:stri
 export type NativeImportSource = { readonly loadingAdmission?:LoadingImportAdmission; readonly kind: 'native-document' | 'native-json-array' | 'native-ndjson' | 'duplicate-ndjson' | 'duplicate-snapshot' | 'duplicate-plan'; readonly retirement?:FileRetirement;readonly previousFiles?:readonly PreviousImportFile[];readonly snapshotVersion?:2;readonly plan?:DuplicateCheckpoint; readonly copy?: {readonly sourceRevision:number; readonly roots:readonly Id[]}; readonly integrityVersion?:1|2;readonly readProof?:ImportReadProof;readonly integrityError?:string;readonly sha256?:string; readonly name?:string; readonly fileId: Id; readonly byteLength: number; readonly targetCanvasId: Id; readonly targetEntry?: {readonly id:Id;readonly incarnation:number} };
 /** 已加入新ID的生命周期绑定；canvas还绑定入口，防用户删除/恢复后被后续尾批误认。 */
 export type ImportBinding = {readonly kind:'element';readonly incarnation:number}|{readonly kind:'canvas';readonly entryElementId?:Id};
-/** 唯一持久任务记录；preparation是准备阶段游标，cursor/eof是解析进度，committed才是已加入前缀。 */
+/** 根状态中的唯一任务事实；可靠性仍取决于包含它的 Root 已 persist。
+ * preparation 是复制准备进度，cursor/eof 是解析进度，committedSourceKeys 是成功源项集合。
+ * 依赖调度可跳过尚未闭合的项，因此成功集合不必是文件顺序的连续前缀；pending 留待重读/协调。 */
 export type ImportJob = {
  readonly version: 1 | 2 | 3; readonly id: Id; readonly projectId: Id; readonly sourceFingerprint: string; readonly parserVersion: number;
  readonly source?: NativeImportSource;
@@ -42,14 +45,19 @@ export type ImportJob = {
  /** 稳定分配跨重开复用；pending可重读，committed不会因后续失败/取消被重新创建。 */
  readonly stableIds: Readonly<Record<string, Id>>; readonly pending: readonly ImportLocator[]; readonly committedSourceKeys: readonly string[]; readonly failures: readonly ImportFailure[]; readonly error?: string;
 };
-/** integrity-v2除零前缀取消外，缺完整SHA或有完整性错误都继续阻止发送；解析EOF不等于源已封存。 */
+/** integrity-v2 除无成功源项的取消任务外，缺完整 SHA 或有完整性错误都仍需封存。
+ * 这里只按记录字段判定，不验证 SHA/readProof、不调度验证任务；网络发送门禁由调用方采用此结论。 */
 export const importNeedsSeal=(job:ImportJob)=>job.source?.integrityVersion===2&&!(job.status==='cancelled'&&!job.committedSourceKeys.length)&&(!job.source.sha256||!!job.source.integrityError);
-/** 验证基本身份并创建v3 paused空任务；source旧调用缺省integrityVersion=1，不自动伪造摘要证明。 */
+/** 验证基本身份并创建v3 paused空任务；source旧调用缺省integrityVersion=1，不自动伪造摘要证明。
+ * source 经 JSON 往返捕获并冻结，但这里不做完整源/任务校验；接纳/恢复仍需 validateImportJob。
+ * 工厂不预留 ID、不读取文件，也不写根状态或磁盘。 */
 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),integrityVersion:input.source.integrityVersion??1}}:{}), version: 3, generation: 0, status: 'paused', cursor: 0, eof: false, bindings: {}, stableIds: {}, pending: [], committedSourceKeys: [], failures: [] });
 }
-/** 校验job及独立解释器、准备/加入状态、源证明、映射/绑定/pending/failure间的一致性。 */
+/** 校验job及独立解释器、准备/加入状态、源证明、映射/绑定/pending/failure间的一致性。
+ * 只检查记录内声明，不读取源文件/当前文档，不重算单元 digest 或校验绑定对象仍存活。
+ * 本函数不迁移/发布 job；旧版本合同允许部分字段缺省，不能仅凭校验通过就跳过端口及完整性门禁。 */
 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 && job.version !== 2 && job.version !== 3) throw fault('import_interpreter_unavailable');
@@ -90,6 +98,7 @@ export function validateImportJob(job: ImportJob): void {
    const ids=Object.fromEntries([...plan.elementIds.map(([id,mapped])=>[`element:${id}`,mapped]),...plan.canvasIds.map(([id,mapped])=>[`canvas:${id}`,mapped])]);
    if(job.parserVersion!==2||plan.projectId!==job.projectId||plan.sourceToken!==job.source.sha256||plan.phase!=='emit'||plan.emitted||plan.emittedEdges||!equal(ids,job.stableIds)||!equal(plan.roots.map(id=>ids[`element:${id}`]),job.source.copy!.roots)||job.cursor>plan.order.length+plan.edges.length)throw fault('invalid_copy_stream_checkpoint');
  }else if(job.source?.plan!==undefined)throw fault('invalid_copy_stream_checkpoint');
+  // bindings 的键是目标 ID，不是 source key；有 source 的已提交项还须能从 stableIds 找到对应绑定。
  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');}
@@ -103,14 +112,18 @@ 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. */
-/** parser1按换行边界返回一条记录，最大记录与读取chunk分开约束，空白行skip但仍推进cursor。 */
+/** parser1按 LF/EOF 返回一条记录，cursor 为字节偏移；key 使用起始字节位置，空白行 skip 也推进游标。
+ * 捕获 Blob 与 max/chunk 标量，但 readBytes/调度配置仍从 options 读取，宿主须保持源与配置稳定。
+ * 不校验 cursor 是否真是行边界，不限制整个 Blob 总长，也不在这里证明 fingerprint 对应持久字节。 */
 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;
  let cache={offset:-1,bytes:new Uint8Array()};
  return Object.freeze({fingerprint:options.fingerprint,parserVersion:1,
-    /** 从已保存byte游标扫描到LF/EOF，超大行丢缓存但继续定位下一行；坏UTF8/JSON返回单项error。 */
+    /** 从游标扫描到 LF/EOF；单行上限不计 LF（CRLF 的 CR 仍计入），无最后换行的记录也接受。
+     * 超大行丢累积 parts 后继续扫描至下一边界，保留读块缓存；坏 UTF8/JSON 返回单项 error。
+     * readBytes/取消/调度失败会抛出，不伪装坏记录；同步解码/JSON.parse 只能在整行读取后执行。 */
    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;
@@ -125,6 +138,7 @@ export function createNdjsonImportSource(options:{bytes:Blob;fingerprint:string;
        cursor=cache.offset+end+(newline<0?0:1);await budget.step(end-begin);
        if(newline>=0||cursor===sourceBytes.size)break;
      }
+      // 只有定位到完整行尾才形成可推进游标；一次 read 不直接修改 ImportJob。
      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;}
@@ -157,7 +171,9 @@ export type ImportRunnerPorts={
 };

 /** Cooperative producer/consumer. Pending work stores seekable locators, not unbounded decoded payloads. */
-/** 创建单job执行器；先校验源解释器与完整性能力，duplicate-snapshot准备任务走另一专用路径。 */
+/** 创建单job执行器；先校验源解释器与完整性能力，duplicate-snapshot准备任务走另一专用路径。
+ * 构造只读初始事实，不自动 start。ports 的同步比较发布/原子提交/错误形状是宿主合同，runner 不实现数据库事务。
+ * 预算限制 pending 定位表、热解码表及一次提交，不保证源图、稳定映射、累计成功/失败记录的总内存恒定。 */
 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);
@@ -171,7 +187,8 @@ export function createImportRunner(ports:ImportRunnerPorts){
  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. */}};
-  /** 每次读事实都检查job/project/fingerprint/parser身份，旧runner不能采纳新任务的同名状态。 */
+  /** 每次读事实检查固定的 job/project/fingerprint/parser 身份；generation 由各采纳点另行检查。
+   * 不深比较 source 的 fileId/目标/全部字段，也不重新完整验证 job；宿主须维持同身份记录的生命周期约束。 */
  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;}
  /** 只允许checkpointSource补readProof/SHA/integrityError；其它source身份和整个job主体都不可偷偷改变。 */
  function checkpointEvidence(job:ImportJob):ImportJob{
@@ -184,11 +201,14 @@ export function createImportRunner(ports:ImportRunnerPorts){
    else if(!equal(job.source,value.source))throw fault('invalid_import_checkpoint');
    return value;
  }
-  /** 合成下一代候选job并捕获源证明；不在这里写私有任务store。 */
+  /** 合成 generation+1 的 v3 候选并捕获源证明；error=undefined 时删键以保持 JSON 合同。
+   * 仅规划，不调用 publish/persist；校验钩子也须纯同步，字段比对不能回滚外部副作用。 */
  function next(job:ImportJob,patch:Partial<ImportJob>):ImportJob{const value={...job,...patch,version:3 as const,generation:job.generation+1};if(value.error===undefined)delete value.error;const checked=checkpointEvidence(value);return checked===value?dataRecord(freeze(value)):freeze(checked);}
  /** 以expectedGeneration compare-and-publish，再核对getJob确实返回该候选，防宿主半接受/丢字段。 */
  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;}
-  /** 用stableIds解码单元，校验纯JSON/非空changes/依赖/成本，新创建ID必须来自reserve分配集合。 */
+  /** 用stableIds解码单元，校验纯JSON/非空changes/依赖/成本，新创建ID必须来自reserve分配集合。
+   * 成功后捕获整个 unit 和映射；这里不校验每条 Change 的领域语义/最终文档，guard 交给宿主提交准备。
+   * prepare 失败时不发布本次映射，但 createId 已发生的分配不会回退；端口不得在解码中产生外部业务效果。 */
  function prepare(line:ImportLine,job:ImportJob){
    const ids=identities(job.stableIds,allocateIdentity);
    const unit=ports.prepare(line,job,ids.reserve);
@@ -197,17 +217,21 @@ export function createImportRunner(ports:ImportRunnerPorts){
    if(new TextEncoder().encode(JSON.stringify(unit)).byteLength>maxBytes)throw fault('import_unit_capacity');
    return{unit:freeze(clone(unit)),ids:ids.finish()};
  }
-  /** optimisticLocal可按宿主背压延后磁盘，但网络仍须过durable门；默认每次直接等待persist。 */
+  /** optimisticLocal 且提供 awaitDurabilityBudget 时按宿主背压等待，否则直接 persist。
+   * 返回不一定代表刚才每个 Root 已落盘；网络仍须由宿主单独确保不可越过相同 Root 的 durable 屏障。 */
  const checkpoint=async(stage:'parsed'|'committed'='committed')=>{if(ports.optimisticLocal&&ports.awaitDurabilityBudget)await ports.awaitDurabilityBudget(stage);else await ports.persist();};
  /** 对含guard的完整ImportUnit取SHA，恢复重读时校验转换结果而不只比较源key。 */
  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('');
-  /** 为成功创建记录保存代次/入口绑定，并把source key追加committed，同时移除对应pending。 */
+  /** 规划成功后的 job：新建元素记 incarnation，新建 canvas 记入口，追加成功 source key 并采用过滤后 pending。
+   * 这里尚未执行提交；失败不应发布此候选。字段修改/删除不新建绑定，旧绑定留给宿主生命周期校验。 */
  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:dataArray(freeze(pending)),committedSourceKeys:dataArray(freeze([...job.committedSourceKeys,...units.map(unit=>unit.key)])),bindings:dataRecord(freeze({...bindings}))});
  }
-  /** 同步提交前再查generation，提交后核对job与Result一致；只有applied样本用于batch成本反馈。 */
+  /** 同步提交前再查 generation，提交后核对 applied 已发布候选、rejected 保持原 job。
+   * 只比 job，不在此审计全部文档变化；原子发布是 ports.commit 合同，后置核对失败不能自动回滚宿主副作用。
+   * 成本反馈只计同步 ports.commit 耗时，不含 beforeCommit、解析、SHA 或磁盘保存。 */
  function commit(units:readonly ImportUnit[],job:ImportJob,value:ImportJob){const base=current();if(base.generation!==job.generation)throw fault('stale_import_job');value=freeze(checkpointEvidence(value));const start=now(),result=ports.commit(units,value,job.generation),elapsed=Math.max(0,now()-start);if(!['applied','rejected'].includes(result?.status)||!equal(current(),result.status==='applied'?value:base))throw fault('invalid_import_commit');if(result.status==='applied')batches.record(units.length,elapsed);return result;}
  /** 合作式生产/消费循环：先消化闭合pending，再按有界slice解析新记录，EOF最后确认partial/completed。 */
  async function pump(signal:AbortSignal){
@@ -223,13 +247,16 @@ export function createImportRunner(ports:ImportRunnerPorts){
    /** 预读将异常转成结果以便接住迟到拒绝，不让并行准备出现未处理Promise。 */
    type ReadResult={kind:'line';line:ImportLine|null}|{kind:'error';error:unknown};
    let reading:{cursor:number;result:Promise<ReadResult>}|undefined;
-    // 仅内存有界热缓存：按maxBatch和maxBytes淘汰，pending持久状态始终是locators而非这些payload。
+    // 每次 pump 独立的热缓存按插入/替换顺序淘汰，读取命中不刷新顺序；不是另一份持久 job。
+    // maxBytes 计 unit 序列化成本，不等于 JS 堆占用，且不包括当前 units、源缓存和暂存解析切片。
    const decoded=new Map<string,{unit:ImportUnit;digest:string;bytes:number}>();let decodedBytes=0;
    /** 更新热单元/字节水位，超任一预算淘汰最旧项；淘汰后可按locator重读，不丢业务事实。 */
    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;}}
    /** 提交/失败处理后及时丢热payload，避免已加入内容继续占runner缓存。 */
    function forget(key:string){const value=decoded.get(key);if(value){decodedBytes-=value.bytes;decoded.delete(key);}}
-    /** 同一游标共享一次预读；已有普通ready项时与0ms timer竞速，可先flush显示而不等慢尾部。 */
+    /** 同一游标共享一次预读；已有普通 ready 项时与0ms timer竞速，慢读取保留以便先 flush。
+     * flush 不取消该读取，下一轮仍复用同一 Promise；不同时对源发第二条 read。
+     * 这里只识别全部依赖已 committed 的单项，不替代后面的 SCC 闭合选择。 */
    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}))};
@@ -245,7 +272,8 @@ export function createImportRunner(ports:ImportRunnerPorts){
      didWork=true;
      const committed=new Set(job.committedSourceKeys);
      const batchLimit=batches.limit;
-      // 组选择把循环引用SCC视为不可拆单元；chosen源顺序与依赖可用性共同约束原子提交。
+      // 先消费已闭合 pending：SCC 将循环引用视为不可拆单元，单个闭合组可一起创建互相依赖的记录。
+      // 组选择受单位数/字节/变化数限制；没有可提交组才继续解析或检查 EOF/容量。
      const groups=selectImportGroups(job.pending,committed,{units:batchLimit,bytes:maxBytes,changes:10000}),chosen=groups.flat();
      if(chosen.length){
        const units:ImportUnit[]=[];
@@ -253,7 +281,7 @@ export function createImportRunner(ports:ImportRunnerPorts){
          const cached=decoded.get(locator.key);
          // 热缓存命中须与持久digest相同；否则seek重读并核对nextCursor/依赖/映射/成本及完整unit摘要。
          if(cached&&cached.digest===locator.digest){units.push(cached.unit);continue;}
-          // A source need not support concurrent reads. Finish any speculative read before seeking.
+          // 源不要求支持并发 read；先等旧预读结束再 seek，不丢 reading 句柄，返回顺序读取时仍可复用。
          if(reading)await reading.result;
          const line=await ports.source.read(locator.cursor,signal);
          if(closed||signal.aborted||current().generation!==job.generation)return;
@@ -289,11 +317,13 @@ export function createImportRunner(ports:ImportRunnerPorts){
        const unresolved=job.pending.filter(unit=>!job.failures.some(failure=>failure.key===unit.key&&failure.code==='unresolved_dependency')).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;
      }
-      // 等待依赖超容量是可恢复paused，不继续无限解析并在内存里堆积整个文件。
+      // 没有可提交组且 pending 达阈值就暂停；统计定位表的 JSON 成本，不把 unit 热缓存大小混用。
+      // 检查发生在加入下一项之前，单次追加可跨过阈值；它不是全任务严格字节硬上限。
      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;}
      // Stage a bounded parser slice, then publish its cursor/IDs/locators once. This is
      // unpublished transaction preparation, never another durable task state store.
-      // staged是一次未发布的解析slice，不是第二份任务事实；最终cursor/ID/locator一起发布。
+      // staged 是基于 base.generation 的私有解析 slice，最终 cursor/稳定映射/locator/失败一次发布。
+      // 当前 doc 不随纯解析改变；真正可见内容仍由上面的 commit 发布，保存可按 optimistic 背压稍后跟进。
      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)));
@@ -322,14 +352,17 @@ export function createImportRunner(ports:ImportRunnerPorts){
        }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.
+        // 首次观察到可提交项便尽快结束解析 slice，随后 commit 才能显示；不宣称此处已渲染或落盘。
+        // 之后依 batch 反馈和帧耗时摊销发布；单条 prepare/JSON 计算仍同步，SHA 则 await 其异步结果。
        if(!batches.observed&&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);const ready=staged.pending.some(unit=>unit.dependencies.every(key=>committed.has(key)));if(ports.awaitDurabilityBudget||!ports.optimisticLocal||!ready)await checkpoint(ready?'parsed':'committed');}

    }
  }
-  /** 单执行句柄：异常把仍属于自己的运行任务暂停并尽量保存；finally清索引并按restart条件继续。 */
+  /** active 共享同一 pump Promise；重复 launch 只记 restart，当前轮收尾后按状态决定是否再起一轮。
+   * 非取消异常尽量把同固定身份的 running/completed/partial 改 paused 并保存，已加入内容不回滚。
+   * 该异常兜底不另比原 generation；宿主仍须防同身份的非法替换。finally 清稳定 ID 临时索引及执行句柄。 */
  function launch(){
    if(closed||holds.size)return Promise.resolve();if(active){restart=true;return active;}
    restart=false;controller=new AbortController();
@@ -343,11 +376,14 @@ export function createImportRunner(ports:ImportRunnerPorts){
  return Object.freeze({
    /** paused可显式启动，completed/partial/cancelled不会自动续跑；新一轮清job.error但不清历史内容。 */
    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();},
-    /** 临时叠加冻结并abort旧pump，释放最后hold后仅在job仍running时续跑，不制造新的任务身份。 */
+    /** 临时叠加 hold 并 abort 当前 pump，释放最后 token 后仅 job 仍 running 才续跑；不改持久 status。
+     * 与 run tracker 的 hold 不同，这里请求中止在途源读取（实际响应由端口决定）；已提交内容不会回滚。 */
    hold(){const token=Symbol();holds.add(token);controller?.abort();return()=>{if(!holds.delete(token)||holds.size||closed)return;if(current().status==='running')void launch();};},
-    /** running→paused并持久化，保留cursor/映射/pending/已加入前缀；迟到结果由代次检查挡住。 */
+    /** 仅 running→paused 并 persist，保留 cursor/映射/pending/成功项；不等待 active 完全退出。
+     * publish 推进 generation，加上 abort 挡住迟到采纳；如需排空另外 await settle。 */
    async pause(){const job=current();if(job.status==='running'){controller?.abort();publish(job,{status:'paused'});await ports.persist();}},
-    /** running/paused→cancelled，保留成功前缀与失败报告，不执行默认整批回滚。 */
+    /** 仅 running/paused→cancelled 并 persist，保留成功内容与失败报告，不撤销已提交操作。
+     * 不负责取消 preparing 复制计划、删除源文件或退回后台请求；这些生命周期由 engine 协调。 */
    async cancel(){const job=current();if(['running','paused'].includes(job.status)){controller?.abort();publish(job,{status:'cancelled',error:undefined});await ports.persist();}},
    /** 等当前pump（及其已触发重启）收尾，不代表未来任务/网络同步都已完成。 */
    async settle(){while(active)await active;},