From 9c6f17f2365e583249e0f769f1c16651f357a1ea Mon Sep 17 00:00:00 2001 From: Codex Bot Date: Thu, 12 Nov 2026 14:01:52 +0800 Subject: [PATCH] docs(runtime): clarify applied recovery staging and port contracts --- src/lib/canvas-runtime/applied-recovery.ts | 24 ++++++++++++++------ src/lib/canvas-runtime/contracts.ts | 26 ++++++++++++++++------ 2 files changed, 36 insertions(+), 14 deletions(-) diff --git a/src/lib/canvas-runtime/applied-recovery.ts b/src/lib/canvas-runtime/applied-recovery.ts index ca68f7c..1a047bd 100644 --- a/src/lib/canvas-runtime/applied-recovery.ts +++ b/src/lib/canvas-runtime/applied-recovery.ts @@ -11,13 +11,17 @@ import { runTerminal, type RunPort } from './runs.ts'; import type { AssetPort } from './resources.ts'; /** Stage acknowledgements only. The caller must validate a complete fresh read and durably publish it. */ -/** 在私有表上解释已保存回执，返回 staged 快照与 fresh read 断言；缺证据/解释器即整体拒绝。 */ +/** 取全部非 resolved 请求，要求已有 outcome=applied 且有保存回执，再按最早逻辑 sequence 解释。 + * 只构造 ledger/runs/resources/conflicts 候选，document/history/acknowledged/versions 仍沿用原 snapshot。 + * 没有待处理请求时也可返回空 stage；不会自己读取 canonical、重放 pending、发布 Root 或等待持久化。 + * 依赖输入账本及 ports 的验证/纯解释合同，表复制不隔离恶意 adapter 对传入记录的修改；now 合法性也由宿主保证。 */ export function stageAppliedRecovery(snapshot:EngineSnapshot,ports:{persistence:PersistencePort;runs?:RunPort;assets?:AssetPort;now:number}) { const waiting=[...snapshot.runtime.ledger.requests.values()].filter(request=>request.phase!=='resolved'); // 只要还有 accepted/unresolved，就不能把整组当成确定已执行恢复，也不能新建替代请求重发。 const unknown=waiting.filter(request=>request.outcome!=='applied'); if(unknown.length)throw fault('recovery_waiting_for_receipts',unknown.map(request=>request.id)); - /** 同一请求可能包含多个意图，以最早逻辑 sequence 排序，保证 run 受理先于其控制动作解释。 */ + /** 以请求成员中最小 sequence 排序；成员缺失会取 Infinity，不在此独立证明账本成员/依赖合法。 + * 合法顺序让 run 受理先建立候选记录，后续控制动作才能定位它；不是按回执到达时间解释。 */ const sequence=(request:FrozenRequest)=>request.operationIds.reduce((result,id)=>Math.min(result,snapshot.runtime.ledger.operations.get(id)?.intent.sequence??Infinity),Infinity); waiting.sort((a,b)=>sequence(a)-sequence(b)); // 新建候选账本和资源/run 表，后面的任何校验失败不会污染原 snapshot。 @@ -37,7 +41,8 @@ export function stageAppliedRecovery(snapshot:EngineSnapshot,ports:{persistence: }else if(['run.start','run.cancel','run.apply'].includes(request.kind??'')){ if(!ports.runs?.verifyApplied)throw fault('run_receipt_interpreter_unavailable',[request.id]); const verified=ports.runs.verifyApplied(request,receipt);proof=verified.model; - // 受理必须绑定原 intent/request/board/target；已有 run 也必须具有相同身份和计划。 + // 新受理核对原 intent/request/board/target，已有同 runId 时再比较 request/operation/plan。 + // 已有记录保留其当前进度，不用旧 admission 覆盖；完整 run 形状/服务身份还依赖 verifyApplied 与外围校验。 if(request.kind==='run.start'){ const run=verified.admission,intent=snapshot.runtime.ledger.operations.get(request.operationIds[0])?.intent; if(!run||verified.action||intent?.operationType!=='run.start'||run.operationId!==intent.id||run.requestId!==request.id||run.boardId!==intent.run.boardId||!equal(run.target,intent.run.target))throw fault('run_identity_mismatch'); @@ -50,23 +55,28 @@ export function stageAppliedRecovery(snapshot:EngineSnapshot,ports:{persistence: if(request.kind==='run.cancel'){ if(!action.runStatus||action.outputId||runTerminal(run.status)&&runTerminal(action.runStatus)&&run.status!==action.runStatus)throw fault('run_status_conflict'); }else if(!action.outputId||action.runStatus)throw fault('run_identity_mismatch'); - // 控制 epoch 递增阻止旧 poll 盖回新状态；标 hasMore/nextPollAt 让任务按原 ID 继续追尾。 + // 控制动作推进 epoch 并去重登记已采用 output ID，清旧 pendingPage/错误，让原 run ID 从 now 继续追尾。 + // 这里只暂存状态，不开始 poll；clock/epoch 范围与当前身份仍需宿主最终采纳时验证。 runs.set(run.id,freeze({...run,controlEpoch:(run.controlEpoch??0)+1,...(action.runStatus?{status:action.runStatus}:{}),...(action.outputId?{appliedOutputIds:[...new Set([...(run.appliedOutputIds??[]),action.outputId])]}:{}),hasMore:true,blocked:false,error:undefined,pendingPage:undefined,nextPollAt:ports.now}));runIds.add(run.id); } }else{ if(!ports.persistence.verifyApplied)throw fault('document_receipt_interpreter_unavailable',[request.id]); proof=ports.persistence.verifyApplied(request,receipt); } - // 各请求有不同写版本范围：start 只 owner，cancel/upload 无文档写，普通/apply 要完整参与者。 + // 证明版本键必须恰好覆盖写范围：start 只 owner，cancel/upload 空，普通/apply 为 request.canvasIds。 + // 这里只要求 proof 版本是安全整数且高于原 expected，不独自要求恰好 +1；精确协议规则由解释器保证。 const writes=request.kind==='run.start'?[request.ownerCanvasId!]:request.kind==='run.cancel'||request.kind==='asset.upload'?[]:request.canvasIds; if(writes.length!==(proof?.versions.size??0)||writes.some(id=>!proof?.versions.has(id)))throw fault('receipt_coverage_mismatch',[request.id]); if(proof){for(const [canvas,version]of proof.versions)if(!Number.isSafeInteger(version)||version<=request.expectedVersions[canvas])throw fault('receipt_version_mismatch',[canvas]);proofs.push(proof);} - // 仅在候选账本标已协调 applied；真正采用仍需外层 fresh read 校验和持久化。 + // result 默认按 reconciled applied 归约候选账本并移除此 request 的冲突，原账本/原冻结 body 仍不变。 + // staged 未通过 fresh read 断言前不可直接发布，也不因这一步就释放真实 writer 或开始下一请求。 ledger.result(request.id,'applied');conflicts.delete(request.id); } const staged:EngineSnapshot=Object.freeze({...snapshot,runtime:Object.freeze({...snapshot.runtime,ledger:ledger.getSnapshot(),runs:readMap(runs),resources:readMap(resources),conflicts:readMap(conflicts)})}); return Object.freeze({snapshot:staged,requestIds:Object.freeze(waiting.map(request=>request.id)),runIds:Object.freeze([...runIds]), - /** fresh read 版本不得落后于任何执行证明，并逐个执行协议专用内容断言。 */ + /** 核对 fresh 版本至少覆盖各 proof 水位，再逐个调用协议内容断言；不拿旧回执模型覆盖 fresh。 + * 不在此调用 validateCanonicalRead/validateDocument 或验证读取来源，engine 须先完成这些检查。 + * proof Map/回调按端口不可变纯证据合同保留；断言成功也不是持久化完成回执。 */ assertCanonical(canonical:CanonicalProposal){for(const proof of proofs){for(const [canvas,version]of proof.versions)if((canonical.versions.get(canvas)??-1); @@ -46,17 +48,25 @@ export type RuntimeState = { }; /** core Snapshot 的 runtime 具象化，单次订阅可同时看到模型和对应运行事实。 */ export type EngineSnapshot = Snapshot; -/** 经过服务 codec 校验的文档与版本向量候选；只有协调完成后才能成为新基线。 */ +/** 服务 codec 提供的文档/版本候选，按合同应已校验；类型没有品牌或运行时校验，任意外部对象仍需边界验证。 + * 不包含 history/ledger/Root revision，也不自动证明完整捕获或当前授权；协调完成后才可作为 ACK 基线。 */ export type CanonicalProposal = { readonly document: Document; readonly versions: ReadonlyMap }; /** Receiving records are read deltas, not a validated Document or an ACK base. */ -/** receiving 是尚未完成的读取增量，数量/owner 证据不表示整份 payload 已通过校验。 */ +/** receiving 为尚未完成的 delta：头部一次声明 canvas/墓碑/可选范围证据，后续按读取协议增加元素。 + * received 可包括依赖未闭合、尚不能发布的元素；不是完整 Document、ACK 或可直接提交给 editor 的 working。 + * CanonicalReadUpdate 不携带网络 query/token/sequence 的完整围栏，那些由外层 LoadChunk 绑定。 */ export type CanonicalReadUpdate = CanonicalReadResult | { readonly status:'receiving'; readonly canvasElementCounts?:import('./scope-counts.ts').CanvasElementCounts;readonly elementOwners?:import('./scope-counts.ts').ElementOwners; readonly projectId:Id; readonly canvases:readonly CanvasRecord[]; readonly elements:readonly ElementRecord[]; readonly tombstones:ReadonlyMap; readonly versions:ReadonlyMap; readonly receivedElements:number; readonly totalElements:number }; /** ready 提供新模型；migration-required 描述显式迁移资格，不在读取时偷偷执行迁移。 */ export type CanonicalReadResult = ({ readonly status: 'ready';readonly sourceToken?:string } & CanonicalProposal) | { readonly status: 'migration-required'; readonly rootId: Id; readonly sourceVersions: ReadonlyMap }; /** A validated historical execution receipt; its old model is never published over a fresh read. */ -/** 验证历史 applied 回执的执行证据；assertCanonical 比对 fresh read，不发布旧回执里的旧模型。 */ +/** 由请求解释器验证历史 applied 回执后构造：versions 表示该次写入下界，assertCanonical 检查 fresh 内容兼容。 + * proof 不包含待发布旧模型，也不持久化函数；恢复时由原请求/回执重新解释，回调应纯且基于不可变证据。 + * 证明服务执行不等于本地已协调/已保存，更不授权再提交相同变化。 */ export type AppliedModelProof = { readonly versions:ReadonlyMap; assertCanonical(proposal:CanonicalProposal):void }; -/** 增量服务端口：compile/canonical 是纯编解码，submit/query/read 是可取消外部 I/O。 */ +/** 服务适配合同：compile/canonical/verifyApplied 等同步方法负责纯编解码/验证，异步方法负责外部 I/O。 + * 引擎拥有本地发布/重放/持久化顺序；adapter 不应直接修改 snapshot、history 或 ledger。 + * AbortSignal 只表达取消要求，不证明远端操作没生效；任何失败/超时都须按原 request 身份处理。 + * 接口声明不会自动执行这些规则，具体实现和协调入口必须分别校验。 */ export type PersistencePort = { /** Complete canonical read for a quiescent recovery session; never a legacy fallback. */ /** 恢复协调所需的完整 canonical read；不能拿 legacy fallback 当目标协议基线。 */ @@ -65,7 +75,8 @@ export type PersistencePort = { verifyApplied?: (request:FrozenRequest,receipt:Json) => AppliedModelProof; /** 判断是否有原冻结协议解释器，不支持时保留原 body/key 并阻塞，不能重编码换身份。 */ supportsRequest?: (request: FrozenRequest) => boolean; - /** 将稳定意图与当前基线编成待冻结请求；网络发送前仍需满足持久化屏障。 */ + /** 从稳定 intent/requestId/当前快照计算参与范围、expected versions 与待冻结 payload，不执行网络写入。 + * scheduler/ledger 随后冻结字节并过持久屏障；重试已经冻结的请求应复用原 body，不以最新快照重编。 */ compile(intent: OperationIntent, requestId: Id, snapshot: EngineSnapshot): CompiledRequest; /** Compile an ordered mutable batch; logical IDs and original intents remain independent. */ /** 可把有序未冻结意图合并为一份 wire 请求，原 logical IDs 与意图事实仍独立保留。 */ @@ -78,6 +89,7 @@ export type PersistencePort = { /** 可选精简回执 codec，仅在联合已协调发布时采用；失败应保留原恢复输入。 */ compactResolvedReceipt?: (request:FrozenRequest,receipt:Json,applied:CanonicalProposal)=>Json; /** Pure codec, merging only participants into the latest base, never an old working snapshot. */ - /** 将回执参与者合并进最新已确认基线，不能拿旧 working 快照替换新本地编辑。 */ + /** 解释原请求/结果，对最新 acknowledged 合并参与者并返回提案，不能拿旧 working 覆盖本地新编辑。 + * 不自行发布 ACK 或重放 pending；引擎还需校验候选与版本，重建 working 后统一发布。 */ canonical(request: FrozenRequest, status: OperationStatus, snapshot: EngineSnapshot): CanonicalProposal; };