/** * 同步操作账本:保存逻辑意图、冻结请求、阶段、结果与恢复证据,是 scheduler 的事实源。 * 本模块只生成内存快照,不执行 HTTP/落盘;engine 将其纳入单 Root,再由持久化屏障允许发送。 * 逻辑 operation 与 wire request 分开:多个意图可共享请求,历史和资源保留不因请求合并而消失。 */ import { assertElisionLineage, proveElision, type ElisionProof } from './elision.ts'; import { applyChanges } from '../canvas-core/document.ts'; import type { Change, Id, Json, OperationIntent } from '../canvas-core/types.ts'; import { clone, equal, fault, freeze, readMap, validJson } from '../canvas-core/values.ts'; /** 操作所依据的元素起始状态;unknown 不能当作 absent 来抵消 create/delete。 */ export type Presence = 'absent' | 'exists' | 'unknown'; /** 请求生命周期:冻结→保守发送标记→等待协调→已解决;发送标记必须可靠保存后才调用外部端口。 */ export type RequestPhase = 'frozen' | 'dispatch-marked' | 'waiting' | 'resolved'; /** accepted 仅表示受理;applied 是执行证明,仍需本地协调;not-applied 要求协议的确定零写入终态。 */ export type Outcome = 'unresolved' | 'accepted' | 'applied' | 'not-applied'; /** 冻结后的原 body/key、协议、逻辑 ID、参与 canvas 与预期版本,重试不得重新生成它们。 * phase/attempt/retry 是可推进的控制元数据;generation 记录冻结时的账本代次。 */ export type FrozenRequest = { readonly kind?: OperationIntent['operationType']; readonly ownerCanvasId?: Id; readonly runId?: Id; readonly id: Id; readonly protocolVersion: number; readonly generation: number; readonly body: string; readonly operationIds: readonly Id[]; readonly canvasIds: readonly Id[]; readonly expectedVersions: Readonly>; readonly phase: RequestPhase; readonly outcome: Outcome; readonly error?: string; readonly attempt: number; readonly retryAt?: number; readonly retryAction?: 'submit' | 'query'; }; /** 意图及其本地解析状态;blocked 不等于丢弃,elided/retired 不等于收到服务器 applied 回执。 */ export type OperationRecord = { readonly intent: OperationIntent; readonly phase: 'pending' | 'frozen' | 'resolved' | 'blocked'; readonly requestId?: Id; readonly blockers: readonly string[]; readonly resolution?: 'applied' | 'not-applied' | 'superseded' | 'discarded' | 'satisfied' | 'elided'; readonly elisionId?: Id; readonly elision?: ElisionProof; readonly recoveryId?: Id; readonly replacementId?: Id; readonly basePresence: Readonly> }; /** generation 随账本事实改变递增,durableGeneration 仅确认已实际存储的前缀。 */ export type LedgerSnapshot = { readonly generation: number; readonly durableGeneration: number; readonly operations: ReadonlyMap; readonly requests: ReadonlyMap }; /** A field-safe compressor for a wholly unsent segment. Frozen records never enter here. */ /** 仅压缩完全未冻结的变化段:create 后字段修改并入新对象,同字段连续变化合并。 * preserveLifecycle 保留创建/删除,preserveChecks 保留条件 no-op;不能用最后整节点覆盖替代字段链。 */ export function compressChanges(changes: readonly Change[], options: { preserveLifecycle?: boolean; preserveChecks?: boolean } = {}): readonly Change[] { // positions 按元素/canvas ID 保存结果索引栈,只尝试合并同对象最近变化,保留原条件顺序。 const result: (Change | undefined)[] = [], positions = new Map(); for (const change of changes) { const id = change.kind === 'field' ? change.target.id : change.id; const key = `${change.kind === 'canvas' ? 'canvas' : 'element'}:${id}`, stack = positions.get(key) ?? []; const previousIndex = stack.at(-1), previous = previousIndex === undefined ? undefined : result[previousIndex]; /** 标空而不移动全数组,弹掉该对象上一索引;最后统一过滤空槽。 */ const removePrevious = () => { result[previousIndex!] = undefined; stack.pop(); }; if (previous?.kind === 'element' && previous.after && !previous.before && change.kind === 'field') { // Apply the same conditional field semantics as the document; never discard a failed before check. // 用与正常文档相同的条件字段执行器合并创建内容,before 不匹配仍应抛错,不能被压缩吞掉。 const document = { projectId: '', elements: new Map([[id, previous.after]]), canvases: new Map(), tombstones: new Map() }; const after = applyChanges(document, [change]).elements.get(id)!; result[previousIndex!] = freeze({ ...previous, after }); } else if (previous?.kind === 'element' && previous.after && !previous.before && change.kind === 'element' && !change.after && equal(previous.after, change.before) && !options.preserveLifecycle) { removePrevious(); } else if (previous?.kind === 'field' && change.kind === 'field' && previous.field === change.field && previous.target.incarnation === change.target.incarnation && equal(previous.after, change.before) && (previous.afterPresent ?? true) === (change.beforePresent ?? true)) { const merged = freeze({ ...previous, after: clone(change.after), afterPresent: change.afterPresent ?? true }); if (!options.preserveChecks && equal(merged.before, merged.after) && (merged.beforePresent ?? true) === (merged.afterPresent ?? true)) removePrevious(); else result[previousIndex!] = merged; } else { stack.push(result.length); positions.set(key, stack); result.push(clone(change)); } } return freeze(result.filter((change): change is Change => change !== undefined)); } /** 从已有快照复制可变表,所有有效修改后重新捕获只读快照;旧 Snapshot 的成员不会改变。 */ export function createLedger(initial?: LedgerSnapshot) { let generation = initial?.generation ?? 0, durableGeneration = initial?.durableGeneration ?? 0; const operations = new Map(initial?.operations), requests = new Map(initial?.requests); let cached: LedgerSnapshot; /** 捕获当前 generation 和表成员,供外部一致读取;不在此处通知 UI 或执行持久化。 */ const refresh = () => { cached = Object.freeze({ generation, durableGeneration, operations: readMap(operations), requests: readMap(requests) }); }; refresh(); /** 账本事实变化推进内存 generation,durable 水位保持不动。 */ function changed() { generation++; refresh(); } /** 按原 request ID 定位,缺失抛领域错误,不自动创建一条替代请求。 */ function requireRequest(id: Id) { const request = requests.get(id); if (!request) throw fault('request_missing', [id]); return request; } /** 验证新增意图依赖存在且无环;visiting 仅表示当前 DFS 路径,退出节点后移除。 */ function assertAcyclic(intent: OperationIntent) { const visiting = new Set(); /** 依赖不能回到待添加意图或已在当前路径上的节点,避免 scheduler 永久等待闭环。 */ const visit = (id: Id): void => { if (id === intent.id || visiting.has(id)) throw fault('dependency_cycle', [id]); const operation = operations.get(id); if (!operation) throw fault('dependency_missing', [id]); visiting.add(id); for (const dependency of operation.intent.dependencies) visit(dependency); visiting.delete(id); }; for (const id of intent.dependencies) visit(id); } return Object.freeze({ /** 读取最新捕获的只读事实,不暴露内部可变表。 */ getSnapshot: () => cached, /** 校验身份/序列/基线与依赖,再添加 pending;同 ID 同意图幂等,改意图则拒绝。 */ add(intent: OperationIntent, basePresence: Readonly>) { if (!intent.id || !Number.isSafeInteger(intent.sequence) || intent.sequence < 1 || new Set(intent.dependencies).size !== intent.dependencies.length || Object.values(basePresence).some(value => !['absent', 'exists', 'unknown'].includes(value))) throw fault('invalid_operation'); const existing = operations.get(intent.id); if (existing) { if (!equal(existing.intent, intent)) throw fault('operation_identity_conflict', [intent.id]); return; } assertAcyclic(intent); operations.set(intent.id, freeze({ intent: clone(intent), phase: 'pending', blockers: [], basePresence: { ...basePresence } })); changed(); }, /** 初步可执行性:pending、无 blocker、依赖均 resolved;结果的业务语义由更上层协调。 */ eligible(id: Id) { const operation = operations.get(id); return !!operation && operation.phase === 'pending' && !operation.blockers.length && operation.intent.dependencies.every(dependency => operations.get(dependency)?.phase === 'resolved'); }, /** 给可变意图设置/清除 blocker;有原因变 blocked,无原因回 pending,不改冻结请求。 */ block(id: Id, reasons: readonly string[]) { const operation = operations.get(id); if (!operation || !['pending', 'blocked'].includes(operation.phase)) return; operations.set(id, freeze({ ...operation, phase: reasons.length ? 'blocked' : 'pending', blockers: [...reasons] })); changed(); }, /** 原子绑定多个 ready 意图到同一请求;参与者版本精确覆盖、类型一致、依赖已解决或在同批。 * 只在这里序列化 payload 一次,后续 submit/query 都使用保存的 body 和请求 ID。 */ freeze(input: { id: Id; kind?: FrozenRequest['kind']; ownerCanvasId?: Id; runId?: Id; protocolVersion?: number; operationIds: readonly Id[]; canvasIds: readonly Id[]; expectedVersions: Readonly>; payload: Json }) { if (!input.id || requests.has(input.id) || !Number.isSafeInteger(input.protocolVersion ?? 1) || (input.protocolVersion ?? 1) < 1 || !input.operationIds.length || (!input.canvasIds.length && input.kind!=='asset.upload' && input.kind!=='run.cancel') || !validJson(input.payload) || new Set(input.operationIds).size !== input.operationIds.length || new Set(input.canvasIds).size !== input.canvasIds.length || Object.keys(input.expectedVersions).length !== input.canvasIds.length || input.canvasIds.some(id => !id || !Number.isSafeInteger(input.expectedVersions[id]) || input.expectedVersions[id] < 0)) throw fault('invalid_frozen_request'); if (input.kind === 'run.start' && (!input.ownerCanvasId || !input.canvasIds.includes(input.ownerCanvasId))) throw fault('invalid_run_request'); if ((input.kind === 'run.cancel' || input.kind === 'run.apply') && (!input.ownerCanvasId || !input.runId)) throw fault('invalid_run_request'); for (const id of input.operationIds) { const operation = operations.get(id); if ((input.kind ?? 'document.change') !== operation?.intent.operationType) throw fault('request_kind_mismatch',[id]); if (!operation || operation.phase !== 'pending' || operation.blockers.length || operation.intent.dependencies.some(dependency => operations.get(dependency)?.phase !== 'resolved' && !input.operationIds.includes(dependency))) throw fault('operation_not_ready', [id]); } // canvasIds 排序稳定 claim 顺序;请求和所有 operation 的 requestId/phase 同次更新。 const request: FrozenRequest = freeze({ id: input.id, kind: input.kind ?? 'document.change', ...(input.ownerCanvasId ? { ownerCanvasId: input.ownerCanvasId } : {}), ...(input.runId ? { runId: input.runId } : {}), protocolVersion: input.protocolVersion ?? 1, generation: generation + 1, body: JSON.stringify(input.payload), operationIds: [...input.operationIds], canvasIds: [...input.canvasIds].sort(), expectedVersions: { ...input.expectedVersions }, phase: 'frozen', outcome: 'unresolved', attempt: 0 }); requests.set(request.id, request); for (const id of input.operationIds) operations.set(id, freeze({ ...operations.get(id)!, phase: 'frozen', requestId: request.id })); changed(); return request; }, /** 保守标记请求可能即将发出并增加 attempt;自身不发送,持久化后才允许外部 I/O。 */ markDispatch(id: Id) { const request = requireRequest(id); if (request.phase === 'resolved') throw fault('request_resolved', [id]); const next: FrozenRequest = freeze({ ...request, phase: 'dispatch-marked', attempt: request.attempt + 1, retryAt: undefined, retryAction: undefined }); requests.set(id, next); changed(); return next; }, /** 记录执行结果;applied 且 reconciled 才解除逻辑操作,受理/未协调成功仍 waiting。 * not-applied 结束旧请求但阻塞原意图,不能自动换 key 当作用户已重新授权执行。 */ result(id: Id, outcome: Outcome, error?: string, reconciled = true) { const request = requireRequest(id); if ((request.phase === 'resolved' || request.outcome === 'applied') && request.outcome !== outcome) throw fault('receipt_conflict', [id]); const phase = (outcome === 'applied' && reconciled) || outcome === 'not-applied' ? 'resolved' : 'waiting'; requests.set(id, freeze({ ...request, phase, outcome, error, retryAt: undefined, retryAction: undefined })); if (outcome === 'applied' && reconciled) for (const operationId of request.operationIds) operations.set(operationId, freeze({ ...operations.get(operationId)!, phase: 'resolved', resolution: 'applied' })); if (outcome === 'not-applied') for (const operationId of request.operationIds) operations.set(operationId, freeze({ ...operations.get(operationId)!, phase: 'blocked', blockers: [error || 'not_applied'] })); changed(); }, /** 仅凭完整闭合生命周期证明抵消连续未发意图;复验依赖、absent 基线、序列与墓碑。 * 标记 resolved/elided 并保留意图/本地生命周期,history 和资源 pin 不在这里删除。 */ elide(proof: ElisionProof) { const ids = new Set(proof.operationIds), seen = new Set(), intents = proof.operationIds.map(id => { const record = operations.get(id); if (!record || record.phase !== 'pending' || record.requestId || record.blockers.length || record.intent.dependencies.some(dependency => ids.has(dependency) ? (operations.get(dependency)?.intent.sequence ?? Infinity) >= record.intent.sequence : operations.get(dependency)?.phase !== 'resolved')) throw fault('elision_not_safe', [id]); for (const change of record.intent.changes) if (change.kind !== 'canvas') { const target = change.kind === 'field' ? change.target.id : change.id; if (!seen.has(target) && (!Object.hasOwn(record.basePresence, target) || record.basePresence[target] !== 'absent')) throw fault('elision_not_safe', [target]); seen.add(target); } return record.intent; }); if (!equal(proof, proveElision(intents, proof.baseTombstones))) throw fault('invalid_elision'); // 之前已抵消的生命周期也占用 incarnation 高水位,不能因服务端从未见过就复用旧代次。 const previous = new Map(); for (const record of operations.values()) if (record.elision) for (const [id, incarnation] of Object.entries(record.elision.tombstones)) previous.set(id, Math.max(previous.get(id) ?? 0, incarnation)); assertElisionLineage(proof, intents, previous); const first = intents[0].sequence, last = intents.at(-1)!.sequence; if ([...operations.values()].some(record => record.intent.sequence >= first && record.intent.sequence <= last && !ids.has(record.intent.id))) throw fault('elision_not_contiguous'); for (const id of proof.operationIds) operations.set(id, freeze({ ...operations.get(id)!, phase: 'resolved', resolution: 'elided', elisionId: proof.operationIds[0], ...(id === proof.operationIds[0] ? { elision: clone(proof) } : {}) })); changed(); }, /** 对已确定 not-applied 的 run.start/cancel 解除控制依赖,不把它标作 applied。 */ resolveNoEffect(id: Id) { const request=requireRequest(id); if(request.outcome!=='not-applied'||!['run.start','run.cancel'].includes(request.kind??''))throw fault('invalid_no_effect_resolution'); for(const operationId of request.operationIds)operations.set(operationId,freeze({...operations.get(operationId)!,phase:'resolved',resolution:'not-applied'})); changed(); }, /** Local retirement is not an applied receipt. Validate the complete closure before mutating. */ /** 用户恢复决策的本地退休:先验证整组与依赖闭包,再标记 superseded/discarded/satisfied。 * 上传和执行结果未知的请求不能在这里被静默丢弃;replacement ID 必须是全新的。 */ retire(recoveryId:Id,decisions:readonly {operationId:Id;resolution:'superseded'|'discarded'|'satisfied';replacementId?:Id}[]) { const ids=new Set(decisions.map(decision=>decision.operationId)); if(!recoveryId||ids.size!==decisions.length||new Set(decisions.flatMap(decision=>decision.replacementId?[decision.replacementId]:[])).size!==decisions.filter(decision=>decision.replacementId).length)throw fault('invalid_recovery_decisions'); for(const decision of decisions){ const record=operations.get(decision.operationId),request=record?.requestId?requests.get(record.requestId):undefined; if(!record||record.phase==='resolved'||record.intent.operationType==='asset.upload'||record.requestId&&(!request||request.phase!=='resolved'||request.outcome!=='not-applied'))throw fault('recovery_operation_not_safe',[decision.operationId]); if(!['superseded','discarded','satisfied'].includes(decision.resolution)||(decision.resolution==='superseded')!==!!decision.replacementId||decision.replacementId&&(operations.has(decision.replacementId)||ids.has(decision.replacementId)))throw fault('invalid_recovery_decisions'); } // 外部未解决操作若仍依赖这组意图,必须一并审阅,不能留下指向被退休操作的悬空工作。 for(const record of operations.values())if(record.phase!=='resolved'&&!ids.has(record.intent.id)&&record.intent.dependencies.some(id=>ids.has(id)))throw fault('recovery_dependency_closure',[record.intent.id]); for(const decision of decisions)operations.set(decision.operationId,freeze({...operations.get(decision.operationId)!,phase:'resolved',resolution:decision.resolution,recoveryId,...(decision.replacementId?{replacementId:decision.replacementId}:{})})); if(decisions.length)changed(); }, /** 保存原请求的 dueAt/action,重试计划也是可恢复事实;实际 heap 是 scheduler 的派生索引。 */ scheduleRetry(id: Id, dueAt: number, action: 'submit' | 'query') { const request = requireRequest(id); if (request.phase === 'resolved' || !Number.isFinite(dueAt)) throw fault('invalid_retry', [id]); requests.set(id, freeze({ ...request, retryAt: dueAt, retryAction: action })); changed(); }, /** 为未解决请求记录解释器/权限等阻碍,并清重试计划;保留 body/key 和 claims 的恢复依据。 */ blockRequest(id: Id, error: string) { const request = requireRequest(id); if (request.phase === 'resolved' || request.error === error) return; requests.set(id, freeze({ ...request, error, retryAt: undefined, retryAction: undefined })); changed(); }, /** 仅允许 durable 水位单调前进到当前 generation 内;不新增业务 generation。 */ markDurable(through: number) { if (!Number.isSafeInteger(through) || through < durableGeneration || through > generation) throw fault('invalid_durable_generation'); durableGeneration = through; refresh(); }, /** 发送标记后且未确定 not-applied 均按可能已执行处理;这是保守风险依据,不是成功证明。 */ possiblyApplied(id: Id) { const request = requireRequest(id); return request.phase !== 'frozen' && request.outcome !== 'not-applied'; }, }); }