From b3a634ce6be3bec4328cf4f2afbb0d0b88a9397a Mon Sep 17 00:00:00 2001 From: Codex Bot Date: Thu, 12 Nov 2026 14:55:15 +0800 Subject: [PATCH] docs(runtime): explain engine file checks and response reconciliation --- src/lib/canvas-runtime/engine.ts | 80 ++++++++++++++++++++++---------- 1 file changed, 55 insertions(+), 25 deletions(-) diff --git a/src/lib/canvas-runtime/engine.ts b/src/lib/canvas-runtime/engine.ts index 6e8b992..3018169 100644 --- a/src/lib/canvas-runtime/engine.ts +++ b/src/lib/canvas-runtime/engine.ts @@ -444,8 +444,10 @@ export async function openProjectEngine(options: EngineOptions) { })(); sourceCollection=task.finally(()=>{sourceCollection=undefined;});return sourceCollection; } - /** 为备份/查看借出导入原文件。先加引用 pin 再异步读字节，校验长度及已有 SHA， - * 成功返回必须释放的句柄；失败自动释放，关闭期间的迟到结果不再交给调用方。 */ + /** 为备份/查看借出当前或指定历史源文件。同步选定文件描述并 pin，再从暂存/可靠存储读 Blob。 + * 校验长度；仅已有 sha256 时重算比对，不因返回 Blob 就推断 integrity-v2 已完成整源封存。 + * 成功返回 {bytes,release}，调用方必须释放；失败自动释放，关闭期间的迟到结果不交付。 + * 不调用完整 assertWritable：恢复会话/加载期也可借出文件，远端权限不由此入口重新授权。 */ async function openImportSource(jobId:Id,fileId?:Id) { if(closing||closed)throw fault('engine_closing');const job=getImport(jobId),file=fileId!==undefined?importFiles(new Map([[jobId,job]])).get(fileId)?.file:job.source;if(!file)throw fault('import_source_unavailable'); if(collectionReservations.has(file.fileId))throw fault('resource_collection_in_progress',[file.fileId]); @@ -454,7 +456,8 @@ export async function openProjectEngine(options: EngineOptions) { /** 幂等归还本次 pin；从 displays 释放表移除，项目关闭不会重复扣除引用计数。 */ const release=()=>{if(released)return;released=true;displays.delete(release);const count=displayPins.get(file.fileId)??1;if(count<=1)displayPins.delete(file.fileId);else displayPins.set(file.fileId,count-1);};displays.add(release); try{ - // pin 已建立后发布一次 runtime 更新，让保留规则撤销已过时的退休标记。 + // pin 先建立，空 runtime 更新触发 prepareRuntime 的保留规则，撤掉因无人引用产生的软退休标记。 + // 只是重新派生引用状态，不更换本次已捕获的文件身份；本函数不等待该更新的持久化回执。 if(file.retirement){const result=editor.updateRuntime(previous=>Object.freeze({...previous}));if(result.status==='rejected')throw fault(result.code,result.targets);} const bytes=fileWrites.get(file.fileId)??await options.storage.readFile(leader,file.fileId); if(closing||closed)throw fault('engine_closing');if(!bytes||bytes.size!==file.byteLength)throw fault('import_source_unavailable',[file.fileId]); @@ -463,7 +466,8 @@ export async function openProjectEngine(options: EngineOptions) { }catch(error){release();if(closing||closed)throw fault('engine_closing');throw error;} } /** 校验存储提供的文件分页：绑定根 revision、ID 严格递增、大小合法、尾游标匹配。 - * 读取期间根变更则拒绝旧清单，不能把两个时点的所有权证据拼成同一份库存。 */ + * 不只比较 revision：await 后要求 editor 仍是同一个 root 对象，避免把变更前清单用于当前归属判断。 + * 这里只验证页结构/顺序，不读 Blob、不哈希，也不证明跨多页调用持有同一个存储快照。 */ async function scanLocalFiles(root:EngineSnapshot,after:Id|undefined,limit:number){ if(!options.storage.scanFiles)throw fault('file_inventory_unavailable'); const page=await options.storage.scanFiles(leader,{expectedRootRevision:root.rootRevision,after,limit}); @@ -474,7 +478,9 @@ export async function openProjectEngine(options: EngineOptions) { if(page.nextAfter!==undefined&&(!page.files.length||page.nextAfter!==previous))throw fault('invalid_file_inventory');return page; } /** 先保存根再读取有界库存，给每个文件附上资源/导入源/孤立文件归属与 pin 依据。 - * 这是诊断读取；unregistered 只表示未登记，不直接证明该文件可安全删除。 */ + * 这是诊断读取；unregistered 只表示未登记，不直接证明该文件可安全删除。 + * ownership 来自当前捕获根，retired 表示 purge 声明，complete 表示引用覆盖；三者不是同一个判定。 + * 本次只返回一页，页尾 nextAfter 供调用者后续继续，没有自动遍历整仓库。 */ async function inventoryLocalFiles(settings:{after?:Id;limit?:number}={}){ assertWritable();const limit=settings.limit??128;if(!Number.isSafeInteger(limit)||limit<1||limit>128||settings.after!==undefined&&(typeof settings.after!=='string'||!settings.after))throw fault('invalid_file_inventory'); await persist();assertWritable();const root=editor.getSnapshot(),page=await scanLocalFiles(root,settings.after,limit),sources=importFiles(root.runtime.imports),pins=orphanCoverage(root,options.registry,displayPins,admissionPins(root)); @@ -482,7 +488,9 @@ export async function openProjectEngine(options: EngineOptions) { } /** 分页盘点并清理没有资源/导入源归属的本地文件。 * 仅对本批核对过长度、MIME 和 SHA 的候选字节安排清理；先记无人引用时间，再经保留期删除。 - * 文件数与散列字节预算分别限流，nextAfter 指向已处理边界，调用方可继续下一页。 */ + * 一次只取 pageSize 一页；maxFiles 限实际验证候选数，maxHashBytes 限本页实际散列字节。 + * 不够预算的候选不推进 nextAfter，调用者需提高预算或换下一次策略，不会悄悄跳过这个文件。 + * orphanCollection 单飞共用首轮 settings；失败/取消只清执行索引，已保存的退休事实保留供后续继续。 */ function collectOrphanFiles(settings:{retentionMs:number;maxFiles?:number;after?:Id;pageSize?:number;maxHashBytes?:number;signal?:AbortSignal}){ assertWritable();if(options.storage.fileDeletion!=='atomic-v1'||!options.storage.scanFiles)throw fault('file_inventory_unavailable'); const retentionMs=settings.retentionMs,maxFiles=settings.maxFiles??64,pageSize=settings.pageSize??128,maxHashBytes=settings.maxHashBytes??64*1024*1024; @@ -495,6 +503,7 @@ export async function openProjectEngine(options: EngineOptions) { const task=(async()=>{ await persist();check();const now=runClock.now();if(!Number.isSafeInteger(now)||now<0)throw fault('invalid_resource_collection_clock'); let root=editor.getSnapshot(),pins=orphanCoverage(root,options.registry,displayPins,admissionPins(root)),scanned=0,nextAfter=settings.after,hashBudgetExhausted=false,scanFinished=false; + // verified 只存本轮已实际核对的内容证据；不能仅用此前持久 orphan 元信息授权本轮物理删除。 const markedIds:Id[]=[],purgedIds:Id[]=[],changedIds:Id[]=[],verified=new Map(); /** 区分引用覆盖完整、扫描到末页与字节预算耗尽；这些状态不能只压成一个 complete。 */ const result=()=>Object.freeze({markedIds:Object.freeze(markedIds),purgedIds:Object.freeze(purgedIds),changedIds:Object.freeze(changedIds),complete:pins.complete,blockers:pins.blockers,checkedRootRevision:root.rootRevision,scanned,nextAfter,hashBudgetExhausted,scanFinished,durableRootRevision}); @@ -518,7 +527,8 @@ export async function openProjectEngine(options: EngineOptions) { // 整页走完才使用存储尾游标；部分处理时保留最后实际检查过的文件 ID。 if(finished)nextAfter=page.nextAfter;scanFinished=finished&&page.nextAfter===undefined; pins=orphanCoverage(root,options.registry,displayPins,admissionPins(root));if(!pins.complete)return result(); - // 同 ID 字节发生变化时重新起算保留期；旧退休证据不能授权删除一份新内容。 + // 同 ID 的 digest/长度/MIME 任一变化都重新起算保留期；changedIds 不是自动恢复成原有资源的许可。 + // now 在本轮开头捕获，重查引用不顺带延长本轮时钟；retentionMs=0 时本轮新标记也可进入删除阶段。 const updates=new Map(); for(const [id,file]of verified){if(pins.pins.has(id))continue;const old=root.runtime.orphanFiles?.get(id),changed=!!old&&(old.digest!==file.digest||old.byteLength!==file.byteLength||old.mimeType!==file.mimeType); if(!old?.retirement||changed){updates.set(id,freeze({...file,retirement:{version:1 as const,unreferencedAt:now}}));markedIds.push(id);if(changed)changedIds.push(id);} @@ -538,23 +548,29 @@ export async function openProjectEngine(options: EngineOptions) { // 成功、失败或取消都撤销单飞索引；不会删除根中的退休记录或伪造 purged 回执。 orphanCollection=task.finally(()=>{orphanCollection=undefined;if(orphanController===controller)orphanController=undefined;});return orphanCollection; } - /** 将可变 ledger helper 的一次归约转换为 editor 的同根 runtime 发布；拒绝结果转成领域异常。 */ + /** 以当前 ledger 快照建立私有 helper，执行同步 reduce，再把其新快照交 editor 一次发布。 + * 不允许 helper 绕过根发布单独成为事实源；reduce 须保持纯同步，外部副作用不由这里回滚。 + * Result.rejected 转异常供 scheduler/持久化调用者处理，成功不等于该根已经 durable。 */ function mutate(reduce: (ledger: ReturnType) => void) { const result = editor.updateRuntime(previous => { const ledger = createLedger(previous.ledger); reduce(ledger); return Object.freeze({ ...previous, ledger: ledger.getSnapshot() }); }); if (result.status === 'rejected') throw fault(result.code, result.targets); } /** 网络结果的统一协调入口。先记录请求是否生效，再按请求种类解释回执。 * applied 只证明服务端生效；仍需验证身份/版本并重放本地未决操作，才能发布新的 ACK 和 working。 - * 协调失败保留原回执与原工作文档，标为“已生效但未协调”，不当作可重新提交的新请求。 */ + * 协调失败保留原回执与原工作文档，标为“已生效但未协调”，不当作可重新提交的新请求。 + * 全部协调在同一同步调用内规划/发布；磁盘屏障及请求 generation/身份采纳由外层 scheduler 负责。 + * 此处 resolved 快速返回检查传入 request 快照，不把任意迟到响应视为可直接采纳的许可。 */ function complete(request: FrozenRequest, status: OperationStatus): { readonly blocked: true } | void { const root = editor.getSnapshot(), previous = root.runtime, ledger = createLedger(previous.ledger); if (request.phase === 'resolved') return; - // 未生效/未知结果只更新账本和回执；明确未生效的启动、取消动作可终结为无效果。 + // 非 applied 结果不改 ACK/working，只更新 ledger 与可选回执；not-applied 仍是“确定未生效”，不是传输失败。 + // run.start/run.cancel 的确定无效果可解决控制依赖；其它重试/阻塞与 writer 占用由 ledger/scheduler 管。 if (status.outcome !== 'applied') { const result=editor.updateRuntime(runtime=>{const ledger=createLedger(runtime.ledger),receipts=new Map(runtime.receipts);ledger.result(request.id,status.outcome,status.error);if(status.outcome==='not-applied'&&['run.start','run.cancel'].includes(request.kind??''))ledger.resolveNoEffect(request.id);if(status.canonical!==undefined)receipts.set(request.id,freeze(status.canonical));return Object.freeze({...runtime,ledger:ledger.getSnapshot(),receipts:readMap(receipts)});},root.rootRevision); if(result.status==='rejected')throw fault(result.code,result.targets);return; } - // 取消是控制结果，不直接用响应替换文档；提高 controlEpoch 使在途旧轮询失效，并继续读尾事件。 + // run.cancel 没有文档提案：只更新任务事实/回执/ledger，再异步通知 tracker 追尾。 + // controlEpoch++ 让旧轮询页失效；hasMore=true 继续取取消后的事件，不把“取消受理”当作事件已读完。 if(request.kind==='run.cancel'){ const receipts=new Map(previous.receipts);if(status.canonical!==undefined)receipts.set(request.id,freeze(status.canonical)); try{ @@ -565,7 +581,8 @@ export async function openProjectEngine(options: EngineOptions) { queueMicrotask(()=>runTracker.notify([run.id]));return; }catch(error){ledger.result(request.id,'applied',(error as Error).message,false);const result=editor.updateRuntime(()=>Object.freeze({...previous,receipts:readMap(receipts),ledger:ledger.getSnapshot()}),root.rootRevision);if(result.status==='rejected')throw fault(result.code,result.targets);report(error);return{blocked:true};} } - // 上传回执只更新资源已存储事实；资源 ID 和原字节长度必须与冻结请求关联的记录一致。 + // asset.upload 成功只把 stored 写回资源及解决 ledger，不替换元素数据，不在此删除本地原文件。 + // 先由 assets.accept 校验协议证据，再核 stored ID/长度；显示 URL、共享仓库提升与回收走各自链路。 if (request.kind === 'asset.upload') { const receipts = new Map(previous.receipts); if (status.canonical !== undefined) receipts.set(request.id, freeze(status.canonical)); try { @@ -606,12 +623,14 @@ export async function openProjectEngine(options: EngineOptions) { const version = proposal.versions.get(canvas); if (!version || version <= request.expectedVersions[canvas] || version < (previous.versions.get(canvas) ?? 0)) throw fault('stale_canonical_version', [canvas]); } - // A codec cannot roll back an unrelated canvas while reconciling this request. + // 不在本次写入集合中的已知 canvas，版本、canvas 记录及直接 owner 成员必须保持 ACK 原值。 + // 这是参与范围检查，不是以局部响应替换全图的许可；最终重放/发布还会校验候选文档。 for (const [canvas, version] of previous.versions) if (!writtenCanvases.includes(canvas)) { if (proposal.versions.get(canvas) !== version || !equal(proposal.document.canvases.get(canvas), previous.acknowledged.canvases.get(canvas)) || !equal([...proposal.document.elements.values()].filter(element => element.ownerCanvasId === canvas), [...previous.acknowledged.elements.values()].filter(element => element.ownerCanvasId === canvas))) throw fault('canonical_scope_mismatch', [canvas]); } ledger.result(request.id, 'applied'); - // 先把回执模型复制为独立 ACK，再只重放仍未解决的本地意图，保留用户随后做出的编辑。 + // 在私有 ledger 中先解决本次请求，再从回执模型捕获 ACK，重放其它未决意图得到 working。 + // 本地后续编辑不会因为服务端返回了较早模型而丢失；ACK、working、ledger 直到 reconcile 才一起公开。 const acknowledged = createDocument(proposal.document.projectId, [...proposal.document.canvases.values()], [...proposal.document.elements.values()], proposal.document.tombstones); const document = replayPending(acknowledged, ledger.getSnapshot(), options.registry); conflicts.delete(request.id); @@ -634,14 +653,14 @@ export async function openProjectEngine(options: EngineOptions) { if(compact!==status.canonical){const owned=freeze(clone(compact));if(!validJson(owned))throw fault('invalid_receipt_compaction');resolvedReceipts=new Map(receipts);resolvedReceipts.set(request.id,owned);} }catch(error){report(error);} } - // Keep the original map intact for a rejected final publication below. + // receipts 原表保留原回执；resolvedReceipts 只供成功候选使用，最终 reconcile 拒绝时仍回到原证据。 const runtime: RuntimeState = Object.freeze({ ...previous, runs: readMap(runs), ledger: ledger.getSnapshot(), acknowledged, versions: appliedVersions, receipts: readMap(resolvedReceipts), conflicts: readMap(conflicts) }); // loading v3+ 允许已就绪范围的 ACK 更新；只在这次同步 reconcile 内打开受控发布窗口。 const oldAccepting=acceptingLoading;acceptingLoading=oldAccepting||(previous.loading?.version??0)>=3;let result:Result;try{result=editor.reconcile(document, runtime, root.rootRevision);}finally{acceptingLoading=oldAccepting;} if (result.status === 'rejected') throw fault(result.code, result.targets); if(action)queueMicrotask(()=>runTracker.notify([action.runId])); } catch (error) { - // Preserve both the known receipt and the user's working document. No rollback or lane release. + // 失败候选的 ACK/working/runs/压缩回执均不发布，另用旧 runtime 保存 applied-but-unreconciled 及原回执。 // 从旧账本另起失败分支，避免候选 ledger 中的成功推进泄漏；保留参与画布占用等待恢复。 const waiting = createLedger(previous.ledger); waiting.result(request.id, 'applied', (error as Error).message, false); conflicts.set(request.id, (error as Error).message || 'reconcile_conflict'); @@ -651,10 +670,12 @@ export async function openProjectEngine(options: EngineOptions) { } } // scheduler 只挑选和执行账本事实；注入的 persistThrough 保证请求身份及前置根先可靠保存。 - // admit 另检查导入 seal、加载覆盖和加载完成保存屏障；v1/2 加载的整体暂停由下方 hold 实现。 + // admit 另检查导入 seal、加载覆盖和加载完成保存屏障；v1/2 分支返回 true 不代表能发送，整体 hold 在下方。 + // admit 是额外范围门禁，依赖、冻结/不确定结果、writer 占用及 durableGeneration 仍由 scheduler 自己校验。 const scheduler = createScheduler({ admit:intent=>{const root=editor.getSnapshot();if(!importsMaySend(root.runtime.imports,durableImportSeals))return false;if(root.runtime.loading&&root.runtime.loading.version<3)return true;if(closing||closed||loadingCompletionRevision!==undefined&&durableRootRevision=root.rootRevision);},getLedger: () => editor.getSnapshot().runtime.ledger, mutate, persist:persistThrough, createId: options.createId ?? (() => crypto.randomUUID()), - /** 计算实际写入画布，自动尺寸补写还需验证 historyEffect 的版本与结构。 */ + /** 沿用非空 intent.canvasIds，否则按变化和当前 working 推导参与画布；尺寸补写先验证 historyEffect。 + * 这里只计算写入参与者，不分配锁；writer claim 与冻结 expectedVersions 在 scheduler/ledger 中建立。 */ participants: intent => { if(intent.historyEffect)validateCreationSizeEffect(intent.historyEffect);return operationCanvases(intent, editor.getSnapshot().document); }, /** 按操作种类生成请求描述。调度器负责随后冻结身份/请求体；这里不发网络请求。 */ compile(intent, id) { @@ -677,7 +698,8 @@ export async function openProjectEngine(options: EngineOptions) { // 批量编译是端口的显式能力，缺少时不假设协议支持多操作请求。 compileBatch: options.persistence.compileBatch ? (intents, id) => options.persistence.compileBatch!(intents, id, editor.getSnapshot()) : undefined, /** 所有服务共用网络容量；上传先从可靠本地存储取字节并核对冻结长度/摘要。 - * 不能用 fileWrites 里的尚未保存文件绕过持久化屏障，也不为重试重新生成请求身份。 */ + * 不能用 fileWrites 里的尚未保存文件绕过持久化屏障，也不为重试重新生成请求身份。 + * 上传的本地读取/摘要发生在 network.run 之前，不占远端网络槽；只把原字节交给 assets.upload。 */ async submit(request, signal) { if (['run.start','run.cancel','run.apply'].includes(request.kind??'')) return options.runs ? network.run(signal, () => options.runs!.submit(request,signal)) : { outcome: 'unresolved', blocked: true, error: 'run_interpreter_unavailable' }; if (request.kind !== 'asset.upload') return network.run(signal, () => options.persistence.submit(request, signal)); @@ -695,7 +717,9 @@ export async function openProjectEngine(options: EngineOptions) { if(initial.loading&&initial.loading.version<3)loadingRelease=scheduler.hold(); /** 把完整读取合入渐进模型。fresh 表示新的权威读取；否则是当前来源流的最终补全。 * 对照来源令牌、版本和所有权证据校验，再合成 ACK 并重放本地意图，一次发布退出 loading。 - * 这里只记录完成 revision，真正解除发送屏障要等 writePrefix 确认该根可靠保存。 */ + * 这里只记录完成 revision，真正解除发送屏障要等 writePrefix 确认该根可靠保存。 + * 完成流读取与 fresh 刷新走不同证据：前者可合并加载期间 ACK，后者须先排除未确定请求。 + * 函数返回领域 Result；异常可登记 loading 冲突，但不把失败的候选部分合入 working。 */ function completeLoading(document:Document,versions:ReadonlyMap,settings:{fresh?:boolean;sourceToken?:string}={}):Result{ try{assertWritable(false);const root=editor.getSnapshot(),previous=root.runtime;if(!previous.loading)throw fault('loading_not_active'); // 新权威读取不能越过结果未确认的请求，否则无法区分读取包含了哪些本地写入。 @@ -720,7 +744,9 @@ export async function openProjectEngine(options: EngineOptions) { if(result.status==='rejected')throw fault(result.code,result.targets); } /** 协调一页生成事件：校验任务/游标/控制代次，依次采用新的模型事件，最后与游标一起发布。 - * 有未确认写入或模型冲突时保留 pendingPage 和输出预览，working、ACK 与游标保持原值。 */ + * 有未确认写入或模型冲突时保留 pendingPage 和输出预览，working、ACK 与游标保持原值。 + * 依赖 RunPort.poll 保证页协议与事件顺序；此协调器不重复完整解析所有 RunPage 字段。 + * 页内模型在私有 acknowledged/versions 上依次累积，最后一次 reconcile；不逐事件发布中间文档。 */ function acceptRunPage(run: RunRecord, page: RunPage) { const root=editor.getSnapshot(),previous=root.runtime,current=previous.runs.get(run.id); if(!current || current.requestId!==run.requestId)throw fault('run_identity_mismatch'); @@ -738,7 +764,7 @@ export async function openProjectEngine(options: EngineOptions) { if(event.seq<=current.cursor || event.type!=='step.applied'&&!(event.type==='step.output'&&event.model&&typeof event.model==='object'&&!Array.isArray(event.model)&&event.model.complete===true))continue; const canvas=event.board_id as Id,version=event.canvas_version as number; if(!Number.isSafeInteger(version)||version<1)throw fault('invalid_run_response'); - // A later document receipt already includes this event; never rewind its participant. + // 本地 ACK 版本已包含该事件就跳过，避免旧模型倒退；其余事件仍要按本页顺序继续协调。 if(version<=(versions.get(canvas)??0))continue; // 完整上下文可能涉及多画布，所有参与者的未决请求都需先确认，不只检查事件主画布。 const model=event.model as {[key:string]:Json}|undefined,context=model?.complete===true,participants=context?Object.keys(model.versions as {[key:string]:Json}):[canvas]; @@ -747,7 +773,8 @@ export async function openProjectEngine(options: EngineOptions) { // 同页前一个事件已更新 ACK/版本，后一个事件的解释应看到该临时累计结果。 const snapshot={...root,runtime:{...previous,acknowledged,versions}}; const proposal=options.runs.canonical(event,snapshot); - // 本地尚有几何/路径编辑时，服务端变更元素归属会改变其坐标语义，拒绝盲目重放。 + // 额外检查未决 field geometry/path 目标的 owner 变化，避免将旧坐标语义重放到新分区。 + // 此局部检查不宣称涵盖任意复合结构冲突；其它 before/owner/schema 条件仍由 replayPending 校验。 for(const record of previous.ledger.operations.values())if(record.phase!=='resolved')for(const change of record.intent.changes)if(change.kind==='field'&&(change.field==='geometry'||change.field==='path')){const before=acknowledged.elements.get(change.target.id),after=proposal.document.elements.get(change.target.id);if(before&&after&&before.ownerCanvasId!==after.ownerCanvasId)throw fault('ownership_conflict',[change.target.id]);} acknowledged=proposal.document;versions=proposal.versions; } @@ -762,7 +789,8 @@ export async function openProjectEngine(options: EngineOptions) { } catch(error) { // 等待写入回执属于可自动再试的协调条件；其它错误阻塞任务并登记冲突。 const code=(error as Error).message; - // Save the page as evidence without overwriting local content or advancing the cursor. + // 整页保存为 pendingPage，同时保留可预览输出；不采用 try 中的临时 ACK/版本，也不推进 cursor。 + // 仅 run_waiting_for_receipts 可自动轮询重试，协议/合并等其它错误登记 run 冲突并 blocked。 const result=editor.updateRuntime(runtime=>{const runs=new Map(runtime.runs);runs.set(run.id,freeze({...current,outputs:page.outputs,outputEvents:[...outputEvents.values()],steps:page.steps,pendingPage:page,error:code,blocked:code!=='run_waiting_for_receipts'}));if(code!=='run_waiting_for_receipts')conflicts.set(`run:${run.id}`,code);return Object.freeze({...runtime,runs:readMap(runs),conflicts:readMap(conflicts)});},root.rootRevision); if(result.status==='rejected')throw fault(result.code,result.targets); if(code!=='run_waiting_for_receipts')report(error); @@ -777,7 +805,9 @@ export async function openProjectEngine(options: EngineOptions) { const sessions = createSessions(editor); // 手势结束后重试默认尺寸补写，避免解码结果在用户拖动/缩放中途抢改几何。 const unsubscribeSizing = sessions.subscribe(() => { if (!sessions.getSnapshot().gesture) queueDefaultSizes(); }); - /** 每次成功发布后登记保存义务，再安排后台任务；通知只传现有操作 ID，不另造一份操作状态。 */ + /** 每次成功发布先把完整 snapshot 按 rootRevision 加入 pending，再安排保存/后台发现微任务。 + * scheduled 只合并保存触发，不丢中间根；任务通知在监听回调返回后执行，避免在发布栈里发 I/O。 + * 订阅不创建第二份业务任务：operationIds/runs 均已在本次根中，后续只建立执行索引。 */ const unsubscribe = editor.subscribe((commit, snapshot) => { pending.set(snapshot.rootRevision, snapshot);notifyDurability(); // 新受理任务在后续微任务启动追尾，让本次根发布和监听通知先完成。