From 28f1a74d40fee52be3152b60925ff5ab2f24874e Mon Sep 17 00:00:00 2001 From: Codex Bot Date: Thu, 12 Nov 2026 00:57:17 +0800 Subject: [PATCH] docs(canvas): explain resumable import assembly and batch ownership --- src/lib/canvas-runtime/engine.ts | 90 ++++++++++++++++++++++++++++++-- 1 file changed, 87 insertions(+), 3 deletions(-) diff --git a/src/lib/canvas-runtime/engine.ts b/src/lib/canvas-runtime/engine.ts index 620b98c..52635ab 100644 --- a/src/lib/canvas-runtime/engine.ts +++ b/src/lib/canvas-runtime/engine.ts @@ -871,20 +871,26 @@ export async function openProjectEngine(options: EngineOptions) { const result=editor.updateRuntime(previous=>{const ledger=createLedger(previous.ledger);ledger.add(freeze(intent),{});return Object.freeze({...previous,ledger:ledger.getSnapshot()});},root.rootRevision); if(result.status==='rejected')return result;queueMicrotask(()=>{if(!closing&&!closed)scheduler.notify([id]);});return{status:'applied',revision:root.revision,operationIds:[id]}; } + /** 导入允许渐进加载 v5+ 的已认证范围；仍复用一般写入的身份、租约和恢复会话护栏。 */ function assertImportWritable(){assertWritable(false);const loading=editor.getSnapshot().runtime.loading;if(loading&&loading.version<5)throw fault('model_loading');} + /** 从唯一根读取当前作业，不以 runner 捕获的旧对象代替可恢复事实。 */ function getImport(id:Id){const job=editor.getSnapshot().runtime.imports?.get(id);if(!job)throw fault('import_missing',[id]);return job;} + /** 仅发布作业元数据；generation 乐观校验防止旧异步结果覆盖暂停/取消/替换后的作业。 + * integrityOwner 只在本次同步更新内允许该 job 推进校验信息，finally 撤销权限。 */ function publishImport(job:ImportJob,expectedGeneration:number){ assertImportWritable();const permission=options.policy?.({type:'changes.apply',changes:[]},editor.getSnapshot().document);if(permission&&!permission.enabled)throw fault(permission.code,permission.targets); integrityOwner=job.id;try{const result=editor.updateRuntime(previous=>{const imports=new Map(previous.imports??[]);if(imports.get(job.id)?.generation!==expectedGeneration)throw fault('stale_import_job');imports.set(job.id,job.version===3?job:freeze({...job,version:3 as const}));return Object.freeze({...previous,imports:readMap(imports)});});if(result.status==='rejected')throw fault(result.code,result.targets);}finally{integrityOwner=undefined;} } - // These records are newly constructed by the engine, including source and - // preparation envelopes. Shape ownership never asserts ownership of children. + /** 为新建检查点声明各层记录容器的所有权，便于增量恢复编码复用。 + * dataRecord/dataArray 不代表所有子数据也归这里所有，不跳过后续结构与内容校验。 */ function importCheckpoint(job:ImportJob):ImportJob { return dataRecord(freeze({...job, ...(job.source?{source:dataRecord(freeze({...job.source}))}:{}), ...(job.preparation?{preparation:dataRecord(freeze({...job.preparation,allocationIds:dataArray(freeze(job.preparation.allocationIds))}))}:{}), })); } + /** 将当前分块读取证明附入检查点；SHA/校验错误以根中最新值为准， + * 防止前台解析的旧 job 覆盖后台全文件验证刚发布的结果。 */ function checkpointImportSource(job:ImportJob):ImportJob{ if(job.source?.integrityVersion!==2)return job; const context=proofReaders.get(job.id),current=editor.getSnapshot().runtime.imports?.get(job.id)?.source; @@ -892,47 +898,68 @@ export async function openProjectEngine(options: EngineOptions) { const {sha256:_,integrityError:__,...source}=job.source; return importCheckpoint({...job,source:{...source,readProof:context.reader.getProof(),...(current?.sha256?{sha256:current.sha256}:{}),...(current?.integrityError?{integrityError:current.integrityError}:{})}}); } + /** 文件校验失败增加 generation，保留已取消/部分完成状态，其余转为暂停。 + * 错误写入源证据，既有已提交内容保留；短暂 hold 触发 runner 的暂停边界，不持有永久暂停租约。 */ function failImportVerification(job:ImportJob,error:unknown){ const code=(error as {code?:string}).code??'import_source_verification_failed'; publishImport(importCheckpoint({...job,generation:job.generation+1,status:['cancelled','partial'].includes(job.status)?job.status:'paused',source:{...job.source!,integrityError:code},error:code}),job.generation); const release=importRunners.get(job.id)?.hold();release?.();report(error); } + /** 每 job 单飞的后台完整性校验：先核对已有分块证明，再计算整文件 SHA。 + * 中止/关闭/全局 hold 时不把取消当损坏；真正失败记录证据，成功保存 seal 后才允许相应同步。 */ function verifyImportSource(id:Id,context:{fileId:Id;bytes:Blob;reader:ReturnType}):Promise{ + // 已中止的旧实例要先结束，再确认上下文仍有效才重建；不能并行验证两个同名来源。 const active=verificationTasks.get(id);if(active)return active.controller.signal.aborted?active.promise.then(()=>{if(!closing&&!closed&&!importHolds.size&&proofReaders.get(id)===context&&sourceNeedsSeal(getImport(id)))return verifyImportSource(id,context);}):active.promise; const admitted=getImport(id),controller=new AbortController(),signal=AbortSignal.any([controller.signal,digestLifecycle.signal]); + /** 每次跨 await 后按 fingerprint、fileId、reader 身份复查，不把旧文件结果写给新作业。 */ const current=()=>{signal.throwIfAborted();if(closing||closed||importHolds.size)throw fault('import_aborted');const job=getImport(id);if(job.sourceFingerprint!==admitted.sourceFingerprint||job.source?.fileId!==context.fileId||proofReaders.get(id)!==context)throw fault('import_identity_changed');return job;}; const promise=(async()=>{ let sha256:string; try{await context.reader.verify(signal);sha256=await abortableFileRead(sourceDigest(context.bytes,signal),signal);const job=current();if(job.source?.sha256&&job.source.sha256!==sha256)throw fault('import_source_changed');} catch(error){if(signal.aborted||closing||closed||importHolds.size)return;failImportVerification(current(),error);try{await persist();}catch{}return;} + // 已取消且从未提交内容就无需发布 seal；有已提交前缀时仍需完成其来源证据。 const job=current();if(job.status==='cancelled'&&!job.committedSourceKeys.length)return;const{integrityError:oldError,...source}=job.source!,next={...job,source:{...source,sha256,readProof:context.reader.getProof()}};if(oldError&&next.error===oldError)delete next.error; publishImport(importCheckpoint(next),job.generation);await persist(); })().catch(error=>{if(!signal.aborted&&!closing&&!closed)report(error);}).finally(()=>{if(verificationTasks.get(id)?.controller===controller)verificationTasks.delete(id);}); verificationTasks.set(id,{controller,promise});return promise; } + /** 重建读取器前先中止并等验证结束、关闭 runner，最后释放分块证明缓存；持久作业不删除。 */ async function resetImportReader(id:Id){ const active=verificationTasks.get(id);active?.controller.abort();if(active)await active.promise; const runner=importRunners.get(id);if(runner){await runner.close();importRunners.delete(id);} proofReaders.get(id)?.reader.close();proofReaders.delete(id); } + /** 将是否仍欠来源校验的协议判断交给 imports 模块，终态也可能仍需 seal。 */ function sourceNeedsSeal(job:ImportJob){return importNeedsSeal(job);} + // 暂停期间被中断、尚未建立 runner 的任务要记下，释放 hold 后才能重新装配。 const deferredImportLoads=new Set(); + /** 可嵌套全局导入暂停：中止准备/验证，给已装配 runner 各加一个 hold。 + * 释放只撤销本 token 的租约，再按根中状态启动需要恢复的任务；不把业务 status 改成 cancelled。 */ function holdImports(){for(const id of importLoads.keys())if(!importRunners.has(id))deferredImportLoads.add(id);const token=Symbol(),releases=new Mapvoid>();importHolds.set(token,releases);for(const controller of preparationControllers.values())controller.abort();for(const {controller}of verificationTasks.values())controller.abort();for(const[id,runner]of importRunners)releases.set(id,runner.hold());return()=>{if(!importHolds.delete(token))return;for(const release of releases.values())release();if(!closing&&!closed)for(const job of editor.getSnapshot().runtime.imports?.values()??[])if(sourceNeedsSeal(job)||['running','preparing'].includes(job.status)&&!importRunners.has(job.id))void loadImport(job.id);};} + /** 从固定复制快照恢复规划。规划版本决定生成可流式读取的计划，或旧版 NDJSON 文件。 + * 期间不断保存 ID 分配/步骤检查点，保证恢复后新元素身份稳定；此阶段还未向画布加入元素。 */ async function materializeCopy(id:Id,bytes:Blob,controller:AbortController){ const admitted=getImport(id),preparation=admitted.preparation; if(!preparation||admitted.source?.kind!=='duplicate-snapshot')throw fault('invalid_copy_preparation'); + /** 复制规划只接受原 fingerprint 且仍 preparing 的作业；关闭、hold、取消均使旧结果失效。 */ const check=()=>{const job=getImport(id);if(controller.signal.aborted||closing||closed||importHolds.size||job.status!=='preparing'||job.sourceFingerprint!==admitted.sourceFingerprint)throw fault('copy_aborted');return job;}; + // 先保存快照作业再规划；v2 索引快照可按需读取，旧快照走完整解码。 await persist();check();const indexed=admitted.source.snapshotVersion===2?await openIndexedCopySnapshot(bytes,options.scope.projectId,options.registry,controller.signal):undefined,document=indexed?.document??await decodeCopySnapshot(bytes,options.scope.projectId,options.registry,controller.signal); + // 渐进模型还要证明复制快照对应合法来源范围；类型/复制声明变化则拒绝续用旧规划器。 validateLoadingCopySource(editor.getSnapshot(),admitted,document); if(copyRegistrySignature(document,options.registry)!==preparation.registrySignature)throw fault('copy_interpreter_changed'); + // 新规划器持久化拓扑与稳定 ID，不先展开所有复制数据为另一份 NDJSON。 if(preparation.version===3||preparation.version===4){ const state=await prepareDuplicateStream(document,options.registry,preparation.selection,{worldOffset:preparation.worldOffset,createId},{sourceToken:admitted.source.sha256!,schemaToken:preparation.registrySignature,checkpoint:preparation.planner,topologyOnly:!!indexed},{signal:controller.signal,checkpointIntervalMs:50,maxStepsPerCheckpoint:8192,checkpoint:async state=>{ + // 只采纳向前推进的 planner 检查点，保存完成后再继续规划，避免恢复丢失已分配身份。 const job=check(),previous=job.preparation!;if(previous.planner&&state.steps<=previous.steps)return; publishImport(importCheckpoint({...job,generation:job.generation+1,preparation:{...previous,steps:state.steps,allocationIds:[...state.elementIds,...state.canvasIds].map(pair=>pair[1]),planner:state}}),job.generation);await persist();check(); }}),job=check();assertNativeImportTarget(job,editor.getSnapshot().document); + // 将旧 ID → 新 ID 和计划一起发布为 running；移除 preparation 后由普通导入 runner 逐批落图。 const stableIds=dataRecord(Object.fromEntries([...state.elementIds.map(([old,id])=>[`element:${old}`,id]),...state.canvasIds.map(([old,id])=>[`canvas:${old}`,id])])),next={...job,generation:job.generation+1,parserVersion:2,status:'running' as const,stableIds,source:{...job.source!,kind:'duplicate-plan' as const,plan:state,copy:dataRecord({sourceRevision:job.source!.copy!.sourceRevision,roots:dataArray(state.roots.map(old=>stableIds[`element:${old}`]))})}};delete next.preparation;delete next.error; publishImport(importCheckpoint(next),job.generation);return{job:getImport(id),bytes,document,indexed}; } + // 兼容旧规划协议：v2 恢复 planner；v1 按原分配顺序重放 IDs，再追加新的分配。 const allocationIds=[...preparation.allocationIds];let cursor=0; const plan=preparation.version===2?await planDuplicateResumable(document,options.registry,preparation.selection,{worldOffset:preparation.worldOffset,createId},{sourceToken:admitted.source.sha256!,schemaToken:preparation.registrySignature,checkpoint:preparation.planner},{signal:controller.signal,checkpoint:async state=>{ const job=check(),previous=job.preparation!;if(previous.planner&&state.steps<=previous.steps)return; @@ -942,121 +969,178 @@ export async function openProjectEngine(options: EngineOptions) { if(steps<=previous.steps&&allocationIds.length===previous.allocationIds.length)return; publishImport(importCheckpoint({...job,generation:job.generation+1,preparation:{...previous,steps:Math.max(previous.steps,steps),allocationIds:[...allocationIds]}}),job.generation);await persist();check(); }}); + // 旧解释器若消耗的 ID 数量变了，原分配顺序已不可信，不能静默生成另一份复制结果。 check();if(preparation.version===1&&cursor!==allocationIds.length)throw fault('copy_interpreter_changed'); + // 旧版计划编码成固定 NDJSON 并算 SHA；异步完成后再次校验目标入口及当前业务策略。 const encoded=await encodeDuplicateSource(document,options.registry,plan,controller.signal),sha256=await sourceDigest(encoded,controller.signal),job=check();assertImportWritable();assertNativeImportTarget(job,editor.getSnapshot().document); const permission=options.policy?.({type:'changes.apply',changes:plan.changes},editor.getSnapshot().document);if(permission&&!permission.enabled)throw fault(permission.code,permission.targets); const fileId=createId(),runtime=editor.getSnapshot().runtime;if(!fileId||fileWrites.has(fileId)||runtime.resources.has(fileId)||[...(runtime.imports?.values()??[])].some(job=>job.source?.fileId===fileId))throw fault('import_identity_conflict'); const stableIds=Object.fromEntries([...Object.entries(plan.elementIds).map(([old,id])=>[`element:${old}`,id]),...Object.entries(plan.canvasIds).map(([old,id])=>[`canvas:${old}`,id])]); + // 新源替换旧快照时把原文件保留在 previousFiles，后续清理仍需保留期与引用证明。 const next={...job,generation:job.generation+1,status:'running' as const,stableIds,sourceFingerprint:`local-file:${fileId}:${encoded.size}`,source:{...job.source!,previousFiles:[...(job.source!.previousFiles??[]),{fileId:job.source!.fileId,byteLength:job.source!.byteLength,...(job.source!.sha256?{sha256:job.source!.sha256}:{})}],kind:'duplicate-ndjson' as const,fileId,byteLength:encoded.size,sha256,integrityVersion:1 as const,copy:{sourceRevision:job.source!.copy!.sourceRevision,roots:plan.roots}}};delete next.preparation;delete next.error; + // 先把字节放入待保存集再发布引用它的根，发布失败则撤回本次字节暂存。 fileWrites.set(fileId,encoded);try{publishImport(importCheckpoint(next),job.generation);}catch(error){fileWrites.delete(fileId);throw error;} return{job:getImport(id),bytes:encoded,document:undefined,indexed:undefined}; } + /** 按 job 装配或复用导入执行器。验证来源、选择协议解析器、准备依赖与资源，再逐批提交。 + * importLoads 是装配任务的单飞表，importRunners 是执行器表，两者都从持久 job 派生。 */ function loadImport(id:Id):Promise{ + // 重复调用共享正在装配的任务；已有 runner 时只补 seal/继续运行，不重复解析和提交。 if(importLoads.has(id))return importLoads.get(id)!; if(importRunners.has(id)){const context=proofReaders.get(id);if(context&&sourceNeedsSeal(getImport(id)))void verifyImportSource(id,context);if(getImport(id).status==='running')void importRunners.get(id)!.start();return Promise.resolve();} deferredImportLoads.delete(id); + // 复制物化可能合法替换源指纹；失败收尾仅可修改这个装配流程最终对应的来源。 let expectedFingerprint=getImport(id).sourceFingerprint; const preparationController=new AbortController();if(getImport(id).source?.integrityVersion===2||['native-document','duplicate-snapshot','duplicate-plan'].includes(getImport(id).source?.kind??''))preparationControllers.set(id,preparationController); const task=(async()=>{ + // kind、parserVersion 与 integrityVersion 是独立协议轴；未知组合必须保留任务并停止解释。 let job=getImport(id),source=job.source;if(!source||!['native-document','native-json-array','native-ndjson','duplicate-ndjson','duplicate-snapshot','duplicate-plan'].includes(source.kind)||(source.kind==='native-document'?![4,5].includes(job.parserVersion):job.parserVersion!==(source.kind==='native-json-array'?3:source.kind==='duplicate-plan'?2:1)))throw fault('import_interpreter_unavailable'); let bytes=fileWrites.get(source.fileId)??await abortableFileRead(options.storage.readFile(leader,source.fileId),preparationController.signal); if(closing||closed)return; + // 读文件期间可能已取消/完成/换源；终态仅在仍欠分块协议 seal 时继续校验，不再加入内容。 const latest=getImport(id);if(latest.sourceFingerprint!==job.sourceFingerprint||latest.source?.fileId!==source.fileId||['completed','partial','cancelled'].includes(latest.status)&&(source.integrityVersion!==2||!sourceNeedsSeal(latest)))return; if(!bytes)throw fault('import_source_unavailable');if(bytes.size!==source.byteLength)throw fault('import_source_changed'); let readBytes:ReturnType['port']|undefined; + // 分块协议可在按需验证读取的同时做全文件校验；已有 SHA 则先确认整文件匹配再继续。 if(source.integrityVersion===2){ let context=proofReaders.get(id);if(!context||context.fileId!==source.fileId){context={fileId:source.fileId,bytes,reader:createImportReadProof({bytes,fileId:source.fileId,proof:source.readProof})};proofReaders.set(id,context);} + // 恢复先重验已保存分块证明；后续解析读口同样经过 reader，不能直接绕过它读取任意字节。 await context.reader.verify(preparationController.signal);const checking=verifyImportSource(id,context);if(source.sha256)await checking; job=getImport(id);source=job.source!;if(preparationController.signal.aborted||['completed','partial','cancelled','paused'].includes(job.status)||importHolds.size)return;readBytes=context.reader.port; }else{ + // 旧完整性协议在开始解析前计算全文件摘要，不用新的分块语义猜测旧格式。 if(source.integrityVersion!==1)throw fault('import_source_integrity_unavailable'); const sha256=await sourceDigest(bytes,preparationController.signal); if(closing||closed)return; + // 整文件散列可能较久，严格对照开始读取时的 generation；用户控制操作优先于迟到结果。 const verified=getImport(id);if(verified.generation!==latest.generation||verified.sourceFingerprint!==job.sourceFingerprint||verified.source?.fileId!==source.fileId||['completed','partial','cancelled','paused'].includes(verified.status))return; if(importHolds.size){deferredImportLoads.add(id);return;} if(source.sha256&&source.sha256!==sha256)throw fault('import_source_changed'); if(!source.sha256){publishImport(importCheckpoint({...verified,generation:verified.generation+1,source:{...source,sha256}}),verified.generation);job=getImport(id);source=job.source!;}else job=verified; } + // 复制源先完成规划或打开索引；普通原生源直接选择对应文档/JSON 数组/逐行解析器。 let copyDocument:Document|undefined,indexed:IndexedCopySnapshot|undefined; if(source.kind==='duplicate-snapshot'){if(preparationController.signal.aborted||importHolds.size||latest.status!=='preparing')return;const materialized=await materializeCopy(id,bytes,preparationController);job=materialized.job;source=job.source!;bytes=materialized.bytes;copyDocument=materialized.document;indexed=materialized.indexed;expectedFingerprint=job.sourceFingerprint;} if(source.kind==='duplicate-plan'&&source.snapshotVersion===2){indexed??=await openIndexedCopySnapshot(bytes,options.scope.projectId,options.registry,preparationController.signal);copyDocument=indexed.document;} if(source.kind==='duplicate-plan'&©Document)validateLoadingCopySource(editor.getSnapshot(),job,copyDocument); + // parser v5 使用文档流读取器，v4 兼容索引读取器；格式选择来自保存的协议，不靠扩展名猜测。 const importSource=source.kind==='duplicate-plan'?createDuplicatePlanSource(copyDocument??await decodeCopySnapshot(bytes,options.scope.projectId,options.registry,preparationController.signal),options.registry,source.plan!,job.sourceFingerprint,indexed?.readElement):source.kind==='native-document'?await (job.parserVersion===5?openStreamingDocumentImportSource:openDocumentImportSource)({bytes,readBytes,fingerprint:job.sourceFingerprint,signal:preparationController.signal}):source.kind==='native-json-array'?createJsonArrayImportSource({bytes,readBytes,fingerprint:job.sourceFingerprint}):createNdjsonImportSource({bytes,readBytes,fingerprint:job.sourceFingerprint}); if(closing||closed||preparationController.signal.aborted||getImport(id).generation!==job.generation)return; if(importHolds.size){deferredImportLoads.add(id);return;} + // beforeCommit 与同步 commit 之间的临时资源授权；不是第二份持久资源表。 const authorizedImportAssets=new Map(),assetPreparationSignals=new WeakSet(); - // Ordinary-batch ceiling only; commit timing and indivisible SCC limits still apply. + // 乐观本地模式允许边解析边落图；普通批次最多128项，积压32根或存储失败时施加保存背压。 + // SCC 等不可分依赖单元另有限制；128 并不承诺每128项一定让出或一定耗时很短。 const runner=createImportRunner({maxBatch:128,optimisticLocal:true,checkpointSource:source.integrityVersion===2?checkpointImportSource:undefined,awaitDurabilityBudget:stage=>pending.size>=32||stage!=='parsed'&&persistenceFailed?persist():undefined,source:importSource,getJob:()=>getImport(id),createId, + // prepare 只生成候选 unit/稳定 ID，实际文档修改由后面的同步事务执行。 publish:publishImport,prepare:(line,job,reserve)=>['duplicate-ndjson','duplicate-plan'].includes(job.source?.kind??'')?prepareDuplicateImport(line,job,reserve,options.registry):prepareNativeImport(line,job,reserve,editor.getSnapshot().document,options.registry,depot&&options.assets?.authority?options.assets.sourceURL:undefined),persist,onError:report, + /** 原生文档格式引用的共享资源先逐项授权、校验 URL、读取字节；批次累计不超过32MiB。 + * 异步阶段不改业务根，失败清空准备材料；同步 commit 仍须重新核对 job/目标/依赖。 */ async beforeCommit(units,job,signal){ + // 其它源格式不走该外部资源领用路径；其引用规则由各自 unit 准备/守卫处理。 authorizedImportAssets.clear();if(job.source?.kind!=='native-document')return{enabled:true}; const ids=referencedResources(units.flatMap(unit=>unit.changes),options.registry);if(!ids.size)return{enabled:true};if(editor.getSnapshot().runtime.loading&&editor.getSnapshot().runtime.loading?.version!==6)return{enabled:false,code:'model_loading',targets:[]}; if(!depot||!options.assets?.authority||!options.storage.assetDepot)return{enabled:false,code:'asset_depot_unavailable',targets:[]}; if(!assetPreparationSignals.has(signal)){assetPreparationSignals.add(signal);signal.addEventListener('abort',()=>authorizedImportAssets.clear(),{once:true});}let total=0; try{for(const id of ids){const authorized=await depot.authorize(leader,{scope:options.assets.authority.scope,resolve:(id,abort)=>network.run(abort,()=>options.assets!.authority!.resolve(id,abort)),read:(id,abort)=>network.run(abort,()=>options.assets!.authority!.read(id,abort))},id,signal);signal.throwIfAborted();assertImportWritable();if(authorized.ref.asset.url!==options.assets.sourceURL(id))throw fault('asset_authorization_mismatch');total+=authorized.bytes.size;if(total>32*1024*1024)throw fault('import_asset_capacity');authorizedImportAssets.set(id,authorized);}return{enabled:true};}catch(error){authorizedImportAssets.clear();return{enabled:false,code:(error as {code?:string}).code??'import_asset_authorization_failed',targets:[]};} }, + /** 将一批元素/canvas、授权资源绑定及 next job 检查点放入一次 editor 发布。 + * 失败回滚的只是本批临时领用材料；此前成功加入并显示的批次保持不变。 */ commit(units,next,expected){ + // 消费一次 beforeCommit 材料，防止后续批次或重入误用上次授权。 const preparedAssets=new Map(authorizedImportAssets);authorizedImportAssets.clear(); assertImportWritable();if(importing)throw fault('reentrant_dispatch');const current=getImport(id);if(current.generation!==expected)throw fault('stale_import_job'); + // 等待资源授权期间目标或外部连线端点可能变化，提交前重新检查入口代次和依赖。 try{assertNativeImportTarget(current,editor.getSnapshot().document);for(const unit of units){assertNativeImportDependencies(current,editor.getSnapshot().document,unit.dependencies);if(['duplicate-ndjson','duplicate-plan'].includes(current.source?.kind??''))assertDuplicateExternal({cursor:0,nextCursor:1,key:unit.key,value:unit.guard},current,editor.getSnapshot().document,options.registry);}}catch(error){return reject(error);} + // 临时登记待保存的领用字节，同时保存旧值，保证事务拒绝后可准确恢复原暂存集合。 const previousFiles=new Map();let accepted=false; try{for(const [assetId,authorized]of preparedAssets){const runtime=editor.getSnapshot().runtime,own=runtime.resources.get(assetId);if(own){if(!own.stored||own.digest!==authorized.ref.digest||!equal(own.stored,authorized.ref.asset))throw fault('shared_asset_identity_conflict',[assetId]);continue;}if(runtime.orphanFiles?.has(assetId)||importFiles(runtime.imports).has(assetId)||fileWrites.has(assetId)||collectionReservations.has(assetId))throw fault('resource_identity_conflict',[assetId]);previousFiles.set(assetId,adoptedFiles.get(assetId));adoptedFiles.set(assetId,{bytes:authorized.bytes,rootRevision:Number.MAX_SAFE_INTEGER});}importAdoptions=[...preparedAssets].filter(([assetId])=>!editor.getSnapshot().runtime.resources.has(assetId)).map(([,value])=>value.ref);}catch(error){for(const [assetId,previous]of previousFiles)if(previous)adoptedFiles.set(assetId,previous);else adoptedFiles.delete(assetId);authorizedImportAssets.clear();return reject(error);} + // reducer 在此同步窗口读 importing/importAdoptions，使 job 游标和新增元素同根更新。 importing={job:next,expectedGeneration:expected}; + // 只允许创建元素/画布，所有 unit 共用逻辑导入组；成功后将领用字节绑定到实际发布 revision。 try{const result=editor.transaction(tx=>{for(const unit of units)for(const change of unit.changes){if(change.kind==='element'&&change.before===null&&change.after)tx.dispatch({type:'element.create',element:change.after});else if(change.kind==='canvas'&&change.before===null&&change.after)tx.dispatch({type:'canvas.create',canvas:change.after});else throw fault('invalid_native_import_unit');}},{label:['duplicate-ndjson','duplicate-plan'].includes(current.source?.kind??'')?'Duplicate selection':'Import canvas content',groupId:`import:${id}`});accepted=result.status==='applied';if(accepted)for(const assetId of previousFiles.keys())adoptedFiles.set(assetId,{bytes:preparedAssets.get(assetId)!.bytes,rootRevision:editor.getSnapshot().rootRevision});return result;}finally{importing=undefined;importAdoptions=[];if(!accepted)for(const [assetId,previous]of previousFiles)if(previous)adoptedFiles.set(assetId,previous);else adoptedFiles.delete(assetId);authorizedImportAssets.clear();} }, }); + // 装配完成后继承仍有效的全局 holds；业务 status=running 才启动加入画布的工作。 importRunners.set(id,runner);for(const releases of importHolds.values())releases.set(id,runner.hold()); if(getImport(id).status==='running')void runner.start(); + // 中止不改业务终态；真实读取损坏留下完整性错误，其它装配失败转 paused。 + // finally 清除执行索引，仅在仍有延迟恢复义务且 job 可运行时重新装配，避免覆盖后来换源的状态。 })().catch(error=>{if(preparationController.signal.aborted)return;if(!closing&&!closed){try{const job=getImport(id);if(job.source?.integrityVersion===2&&!importRunners.has(id)&&!verificationTasks.has(id)){proofReaders.get(id)?.reader.close();proofReaders.delete(id);}if(job.sourceFingerprint!==expectedFingerprint)return;if(job.source?.integrityVersion===2&&(isImportByteFailure(error)||(error as {code?:string}).code==='import_source_unavailable')){failImportVerification(job,error);void persist().catch(report);return;}if(['completed','partial','cancelled','paused'].includes(job.status))return;publishImport(importCheckpoint({...job,generation:job.generation+1,status:'paused',error:(error as {code?:string}).code??'import_failed'}),job.generation);}catch{/* Recovery/identity fences retain the original job. */}report(error);}}).finally(()=>{preparationControllers.delete(id);importLoads.delete(id);if(!['running','preparing'].includes(getImport(id).status))deferredImportLoads.delete(id);if(!closing&&!closed&&!importHolds.size&&deferredImportLoads.delete(id)&&!importRunners.has(id)&&['running','preparing'].includes(getImport(id).status))queueMicrotask(()=>{if(!closing&&!closed)void loadImport(id);});}); importLoads.set(id,task);return task; } + /** 接收本地 Blob 并同步登记一个可恢复作业（源文件上限64MiB），随后后台读取解析。 + * 文件 ID、源证明和目标入口代次从一开始就写进根，页面可立即观察 running，不等待全文件解析。 */ function startNativeImport(bytes:Blob,targetCanvasId:Id,format:'ndjson'|'json-array'|'document'):Id{ if(!['ndjson','json-array','document'].includes(format))throw fault('import_interpreter_unavailable'); assertImportWritable();const loadingAdmission=loadingImportAdmission(editor.getSnapshot(),targetCanvasId);const permission=options.policy?.({type:'changes.apply',changes:[]},editor.getSnapshot().document);if(permission&&!permission.enabled)throw fault(permission.code,permission.targets);const canvas=editor.getSnapshot().document.canvases.get(targetCanvasId);if(!canvas)throw fault('canvas_not_loaded',[targetCanvasId]);if(!(bytes instanceof Blob)||bytes.size>64*1024*1024)throw fault('invalid_import_source'); const id=createId(),fileId=createId(),current=editor.getSnapshot().runtime;if(!id||!fileId||id===fileId||current.imports?.has(id)||current.resources.has(fileId)||current.orphanFiles?.has(fileId)||fileWrites.has(fileId)||importFiles(current.imports).has(fileId))throw fault('import_identity_conflict'); + // 子分区以入口元素 ID+代次绑定目标；同名/同 ID 重建入口不能悄悄接收旧导入。 const entry=canvas.parent?editor.getSnapshot().document.elements.get(canvas.parent.entryElementId)!:undefined; + // 新导入采用分块完整性 v2；解析器版本按内容格式分别选择，不与 job/schema 版本混用。 const job=importCheckpoint({...createImportJob({id,projectId:options.scope.projectId,sourceFingerprint:`local-file:${fileId}:${bytes.size}`,parserVersion:format==='document'?5:format==='json-array'?3:1,source:{...(loadingAdmission?{loadingAdmission}:{}),integrityVersion:2,readProof:{version:1,fileId,byteLength:bytes.size,blockBytes:IMPORT_PROOF_BLOCK_BYTES,blocks:[]},kind:format==='document'?'native-document':format==='json-array'?'native-json-array':'native-ndjson',name:((bytes as Blob&{name?:string}).name||'画布数据').slice(0,255),fileId,byteLength:bytes.size,targetCanvasId,...(entry?{targetEntry:{id:entry.id,incarnation:entry.incarnation}}:{})}}),status:'running' as const}); fileWrites.set(fileId,bytes); integrityOwner=id;try{const result=editor.updateRuntime(previous=>{const jobs=new Map(previous.imports??[]);jobs.set(id,job);return Object.freeze({...previous,imports:readMap(jobs)});}); if(result.status==='rejected'){fileWrites.delete(fileId);throw fault(result.code,result.targets);}void loadImport(id);return id;}finally{integrityOwner=undefined;} } + /** 项目公开的导入控制面：查询/能力判断无副作用，start/duplicate 登记作业，pause/resume/cancel 控制后续批次。 + * 暂停、取消保留已发布前缀，原字节与完整性证据按各自生命周期保存和清理。 */ const imports=Object.freeze({ + /** 返回根中当前作业的只读 Map，不暴露 runner 内部状态。 */ getJobs:()=>readMap(editor.getSnapshot().runtime.imports??new Map()), + /** 检查目标存在、策略与加载范围；只给出当前可用性，真正开始时仍重新校验。 */ canStart(targetCanvasId:Id){try{assertImportWritable();const snapshot=editor.getSnapshot(),permission=options.policy?.({type:'changes.apply',changes:[]},snapshot.document);if(permission&&!permission.enabled)throw fault(permission.code,permission.targets);if(!snapshot.document.canvases.has(targetCanvasId))throw fault('canvas_not_loaded',[targetCanvasId]);loadingImportAdmission(snapshot,targetCanvasId);return{enabled:true as const};}catch(error){const result=reject(error);return{enabled:false as const,code:result.code,targets:result.targets};}}, + /** 检查选择代次、目标范围和清理占用；完整复制闭包/外部引用验证留给后续规划和提交。 */ canDuplicate(selection:readonly Target[],settings:{viewCanvasId?:Id}={}){try{assertImportWritable();if(collectionReservations.size)throw fault('resource_collection_in_progress',[...collectionReservations]);const snapshot=editor.getSnapshot(),permission=options.policy?.({type:'changes.apply',changes:[]},snapshot.document);if(permission&&!permission.enabled)throw fault(permission.code,permission.targets);if(!selection.length)throw fault('empty_copy_selection');for(const selected of selection){const element=snapshot.document.elements.get(selected.id);if(!element||element.incarnation!==selected.incarnation)throw fault('stale_incarnation',[selected.id]);}const target=settings.viewCanvasId??snapshot.document.elements.get(selection[0].id)!.ownerCanvasId;if(!snapshot.document.canvases.has(target))throw fault('canvas_not_loaded',[target]);loadingImportAdmission(snapshot,target,selection);return{enabled:true as const};}catch(error){const result=reject(error);return{enabled:false as const,code:result.code,targets:result.targets};}}, + // 原文件借阅返回带 release 的句柄；三种原生入口各自固定解析协议。 openSource:openImportSource, startNdjson:(bytes:Blob,targetCanvasId:Id)=>startNativeImport(bytes,targetCanvasId,'ndjson'), startDocument:(bytes:Blob,targetCanvasId:Id)=>startNativeImport(bytes,targetCanvasId,'document'), startJsonArray:(bytes:Blob,targetCanvasId:Id)=>startNativeImport(bytes,targetCanvasId,'json-array'), + /** 捕获当前不可变文档和独立选择/偏移，异步编码索引快照，再登记 preparing 复制任务。 + * 后续原画布编辑不会改变复制来源；完成编码不代表已创建全部副本。 */ async duplicate(selection:readonly Target[],settings:{readonly worldOffset?:Point;readonly signal?:AbortSignal;readonly viewCanvasId?:Id}={}):Promise{ assertImportWritable();if(collectionReservations.size)throw fault('resource_collection_in_progress',[...collectionReservations]); const permission=options.policy?.({type:'changes.apply',changes:[]},editor.getSnapshot().document);if(permission&&!permission.enabled)throw fault(permission.code,permission.targets); + // 快照编码会跨 await，先撤销未实际清除的退休标记，配合 copyPreparations pin 防止来源字节中途清理。 if([...editor.getSnapshot().runtime.resources.values(),...(editor.getSnapshot().runtime.orphanFiles?.values()??[])].some(file=>file.retirement&&file.retirement.purgedAt===undefined)){ const reset=editor.updateRuntime(previous=>{const resources=new Map(previous.resources);for(const [id,resource]of resources)if(resource.retirement&&resource.retirement.purgedAt===undefined){const {retirement:_,...kept}=resource;resources.set(id,freeze(kept));}const files=new Map(previous.orphanFiles);for(const [id,file]of files)if(file.retirement&&file.retirement.purgedAt===undefined){const {retirement:_,...kept}=file;files.set(id,freeze(kept));}return Object.freeze({...previous,resources:readMap(resources),...(previous.orphanFiles?{orphanFiles:readMap(files)}:{})});});if(reset.status==='rejected')throw fault(reset.code,reset.targets); } + // 来源固定在 captured；复制目标入口也取自该快照，后续物化/提交再检验它是否仍有效。 const captured=editor.getSnapshot(),target=settings.viewCanvasId??(selection[0]?captured.document.elements.get(selection[0].id)?.ownerCanvasId:undefined),signal=settings.signal,controller=new AbortController(),abort=()=>controller.abort(); const loadingAdmission=target?loadingImportAdmission(captured,target,selection):undefined; + // 与调用方数组/对象断开引用；不能让外部在编码期间改选择或偏移，影响已捕获的复制意图。 const targets=detachJsonData(selection),offset=detachJsonData(settings.worldOffset??{x:32,y:32}); for(const selected of targets){const element=captured.document.elements.get(selected.id);if(!element||element.incarnation!==selected.incarnation)throw fault('stale_incarnation',[selected.id]);}if(![offset.x,offset.y].every(Number.isFinite))throw fault('invalid_copy_offset'); signal?.addEventListener('abort',abort,{once:true});if(signal?.aborted)controller.abort(); const task=(async()=>{ if(!target||!targets.length)throw fault('empty_copy_selection'); + // 快照编码合作让出线程且支持取消；回到这里再检查项目写入资格。 const bytes=await encodeIndexedCopySnapshot(captured.document,controller.signal);assertImportWritable();if(controller.signal.aborted)throw fault('copy_aborted'); const canvas=captured.document.canvases.get(target);if(!canvas)throw fault('canvas_not_loaded',[target]); const id=createId(),fileId=createId(),current=editor.getSnapshot().runtime; if(!id||!fileId||id===fileId||current.imports?.has(id)||current.resources.has(fileId)||current.orphanFiles?.has(fileId)||fileWrites.has(fileId)||importFiles(current.imports).has(fileId))throw fault('import_identity_conflict'); const entry=canvas.parent?captured.document.elements.get(canvas.parent.entryElementId)!:undefined; + // 复制快照 v2 + 规划 v4 是当前组合；初始没有副本根 IDs，规划器会稳定分配并保存检查点。 const job=importCheckpoint({...createImportJob({id,projectId:options.scope.projectId,sourceFingerprint:`local-file:${fileId}:${bytes.size}`,parserVersion:1,source:{...(loadingAdmission?{loadingAdmission}:{}),kind:'duplicate-snapshot',snapshotVersion:2,name:'复制选中内容',fileId,byteLength:bytes.size,targetCanvasId:target,copy:{sourceRevision:captured.revision,roots:[]},...(entry?{targetEntry:{id:entry.id,incarnation:entry.incarnation}}:{})}}),status:'preparing' as const,preparation:{version:4 as const,registrySignature:copyRegistrySignature(captured.document,options.registry),selection:targets,worldOffset:offset,allocationIds:[],steps:0}}); fileWrites.set(fileId,bytes);const result=editor.updateRuntime(previous=>{const jobs=new Map(previous.imports??[]);jobs.set(id,job);return Object.freeze({...previous,imports:readMap(jobs)});}); if(result.status==='rejected'){fileWrites.delete(fileId);throw fault(result.code,result.targets);}void loadImport(id);return id; })(); + // 从编码开始到作业登记结束持有准备任务/pin；所有结束路径清除任务及外部 abort 监听。 copyPreparations.set(controller,task);void task.finally(()=>{copyPreparations.delete(controller);signal?.removeEventListener('abort',abort);}).catch(()=>{});return task; }, + /** 有 runner 就停在其批次边界；未装配时直接推进 job generation 并保存 paused。 + * 分块来源的验证仍可能继续补齐已提交前缀的证据，不能把暂停当作删除来源。 */ async pause(id:Id){assertImportWritable();const runner=importRunners.get(id);if(runner)await runner.pause();else{const job=getImport(id);if(['running','preparing'].includes(job.status)){if(job.source?.integrityVersion!==2)preparationControllers.get(id)?.abort();publishImport(importCheckpoint({...job,generation:job.generation+1,status:'paused'}),job.generation);await persist();}}}, + /** 终态不重启；paused 先等旧装配/runner 结束，必要时重建证明读取器，再保存并恢复调度。 + * 复制快照回 preparing，其它源回 running，避免跳过尚未完成的复制规划。 */ async resume(id:Id){assertImportWritable();const job=getImport(id);if(['completed','partial','cancelled'].includes(job.status))return;if(['running','preparing'].includes(job.status)){await loadImport(id);return;}await importLoads.get(id);await importRunners.get(id)?.settle();if(getImport(id).source?.integrityVersion===2)await resetImportReader(id);await persist();await scheduler.resume();assertImportWritable();const current=getImport(id);if(current.status!=='paused')return;const next={...current,generation:current.generation+1,status:current.source?.kind==='duplicate-snapshot'?'preparing' as const:'running' as const};delete next.error;publishImport(importCheckpoint(next),current.generation);await loadImport(id);}, + /** 取消后续加入，保留已有内容；尚欠 seal 时仍可继续校验，证据完整才中止额外验证。 */ async cancel(id:Id){assertImportWritable();const runner=importRunners.get(id);if(runner)await runner.cancel();else{const job=getImport(id);if(['running','preparing','paused'].includes(job.status)){if(job.source?.integrityVersion!==2)preparationControllers.get(id)?.abort();const next={...job,generation:job.generation+1,status:'cancelled' as const};delete next.error;publishImport(importCheckpoint(next),job.generation);await persist();}}if(!sourceNeedsSeal(getImport(id)))verificationTasks.get(id)?.controller.abort();}, }); + // 全部协调回调已接线后启动调度，恢复既有生成轮询和资源任务。 scheduler.start(); prepareDigests();prepareImages(); runTracker.notify([...initial.runs.keys()]); + // 终态作业若仍欠 seal 也装配验证；running/preparing 则恢复尚未完成的内容处理。 for(const job of initial.imports?.values()??[])if(sourceNeedsSeal(job)||['running','preparing'].includes(job.status))void loadImport(job.id); return Object.freeze({ editor, sessions, imports,