Zayden-Jung/Endoleap
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

docs(runtime): explain engine file checks and response reconciliation

Browse Source
This commit is contained in:
Codex Bot
2026-11-12 14:55:15 +08:00
parent 56b83e2009
commit b3a634ce6b
1 changed files with 55 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/lib/canvas-runtime/engine.ts
+55 -25
View File
@@ -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<Id,OrphanFile>();
/** 区分引用覆盖完整、扫描到末页与字节预算耗尽；这些状态不能只压成一个 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<Id,OrphanFile>();
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<typeof createLedger>) => 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<loadingCompletionRevision)return false;return (root.runtime.loading?.version===6&&intent.operationType==='asset.upload'||loadingCanSend(root.runtime,operationCanvases(intent,root.document),root.document))&&(!root.runtime.loading||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<Id,number>,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();
// 新受理任务在后续微任务启动追尾，让本次根发布和监听通知先完成。