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

Recover verified applied receipts against a fresh canonical base

Browse Source
This commit is contained in:
Codex Bot
2026-11-08 02:09:36 +08:00
parent 89b6f4a253
commit f9cb1ae120
13 changed files with 293 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/components/canvas/headless-recovery-dialog.tsx
+5 -1
View File
@@ -10,9 +10,12 @@ import { Dialog, DialogContent, DialogDescription, DialogFooter, DialogTitle } f
function message(error:unknown){
const code=(error as {code?:string;message?:string})?.code??(error as Error)?.message;
if(code==='recovery_waiting_for_receipts')return '还有请求正在确认。请关闭此窗口，重试同步后再处理。';
if(code==='recovery_read_behind_receipt')return '服务端读取尚未包含已确认的修改，请稍后重新检查。';
if(code?.endsWith('receipt_interpreter_unavailable')||code==='recovery_receipt_missing')return '暂时无法验证原成功回执，修改和原请求仍已保留。请稍后重试。';
if(code==='recovery_in_progress')return '上一次处理尚未结束。可先恢复编辑，再重新检查。';
if(code==='recovery_choices_incomplete')return '请为每一项修改选择处理方式。';
if(code==='stale_incarnation')return '服务端已删除或替换了这个元素。可以放弃这项本地修改。';
if(code==='operation_context_changed')return '元素所属分区已改变，请重新预览这项修改。';
if(code==='ownership_conflict')return '元素已移入其它分区，当前坐标需要重新确认，暂时无法直接保留。';
if(code?.startsWith('recovery_history')||code==='recovery_structure_not_supported')return '这批修改含结构或撤销操作，暂时无法自动处理。原内容和历史均已保留。';
if(code==='recovery_action_requires_new_command')return '生成任务需要重新手动发起。可以在这里放弃尚未执行的请求。';
@@ -31,7 +34,7 @@ export function HeadlessRecoveryDialog({engine,close}:{engine:ProjectEngine;clos
const [choices,setChoices]=useState(new Map<Id,RecoveryChoice>()),[plan,setPlan]=useState<ConflictPlan>(),[error,setError]=useState(''),[busy,setBusy]=useState(true),[frozen,setFrozen]=useState(false);
useEffect(()=>{
let live=true;
Promise.resolve().then(()=>live?engine.prepareRecovery():undefined).then(async session=>{if(!session)return;if(!live){await session.cancel();return;}owner.current=session;setSession(session);setRecords([...engine.getSnapshot().runtime.ledger.operations.values()].filter(record=>record.phase!=='resolved'&&record.intent.operationType!=='asset.upload').sort((a,b)=>a.intent.sequence-b.intent.sequence).map(record=>record.intent));}).catch(error=>{if(live)setError(message(error));}).finally(()=>{if(live)setBusy(false);});
Promise.resolve().then(()=>live?engine.prepareRecovery():undefined).then(async session=>{if(!session)return;if(!live){await session.cancel();return;}owner.current=session;setSession(session);setRecords([...engine.getSnapshot().runtime.ledger.operations.values()].filter(record=>session.pendingOperationIds.includes(record.intent.id)&&record.intent.operationType!=='asset.upload').sort((a,b)=>a.intent.sequence-b.intent.sequence).map(record=>record.intent));}).catch(error=>{if(live)setError(message(error));}).finally(()=>{if(live)setBusy(false);});
return()=>{live=false;void owner.current?.cancel().catch(()=>{/* Engine retains the failed durable barrier; retry/cancelRecovery can finish it. */});};
},[engine]);
async function cancel(){if(busy&&session)return;setBusy(true);try{await engine.cancelRecovery();close();}catch(error){setError(message(error));}finally{setBusy(false);}}
@@ -41,6 +44,7 @@ export function HeadlessRecoveryDialog({engine,close}:{engine:ProjectEngine;clos
<DialogTitle>处理同步修改</DialogTitle>
<DialogDescription>检查期间暂停编辑。逐项选择要保留的修改，预览后再应用；原内容和撤销记录会保留备份。</DialogDescription>
{busy&&!session&&<p role="status">正在等待同步并读取最新内容…</p>}
{!!session?.acknowledgedRequestIds.length&&<p>已找回 {session.acknowledgedRequestIds.length} 个成功回执，仅处理后续尚未同步的修改。</p>}
{session&&records.length===0&&<p>没有待处理的本地修改，可以采用服务端最新内容。</p>}
{session&&records.map((intent,index)=><article key={intent.id} style={{borderBottom:'1px solid #ddd5ca',paddingBottom:12}}>
<strong>修改 {index+1}{intent.operationType==='run.start'?' · 尚未受理的生成':intent.operationType==='run.cancel'?' · 尚未确认的取消':''}</strong>
src/lib/canvas-runtime/applied-recovery.ts
+56
View File
@@ -0,0 +1,56 @@
import type { Id } from '../canvas-core/types.ts';
import { equal, fault, freeze, readMap } from '../canvas-core/values.ts';
import type { AppliedModelProof, CanonicalProposal, EngineSnapshot, PersistencePort } from './contracts.ts';
import { createLedger, type FrozenRequest } from './ledger.ts';
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. */
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');
const unknown=waiting.filter(request=>request.outcome!=='applied');
if(unknown.length)throw fault('recovery_waiting_for_receipts',unknown.map(request=>request.id));
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));
const ledger=createLedger(snapshot.runtime.ledger),runs=new Map(snapshot.runtime.runs),resources=new Map(snapshot.runtime.resources),conflicts=new Map(snapshot.runtime.conflicts);
const proofs:AppliedModelProof[]=[],runIds=new Set<Id>();
for(const request of waiting){
const receipt=snapshot.runtime.receipts.get(request.id);if(receipt===undefined)throw fault('recovery_receipt_missing',[request.id]);
let proof:AppliedModelProof|undefined;
if(request.kind==='asset.upload'){
const intent=snapshot.runtime.ledger.operations.get(request.operationIds[0])?.intent,resource=intent?.operationType==='asset.upload'?resources.get(intent.resourceId):undefined;
if(!ports.assets||!resource)throw fault('asset_interpreter_unavailable',[request.id]);
const stored=freeze(ports.assets.accept(request,{outcome:'applied',canonical:receipt},resource));
if(stored.id!==resource.id||stored.bytes!==resource.byteLength)throw fault('asset_receipt_mismatch');
resources.set(resource.id,freeze({...resource,stored}));
}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;
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');
const existing=runs.get(run.id);
if(existing&&(existing.requestId!==run.requestId||existing.operationId!==run.operationId||!equal(existing.plan,run.plan)))throw fault('run_identity_mismatch');
if(!existing)runs.set(run.id,freeze(run));runIds.add(run.id);
}else{
const action=verified.action,run=action?runs.get(action.runId):undefined;
if(!action||verified.admission||action.runId!==request.runId||!run)throw fault('run_identity_mismatch');
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');
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);
}
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);}
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]),
assertCanonical(canonical:CanonicalProposal){for(const proof of proofs){for(const [canvas,version]of proof.versions)if((canonical.versions.get(canvas)??-1)<version)throw fault('recovery_read_behind_receipt',[canvas]);proof.assertCanonical(canonical);}},
});
}
src/lib/canvas-runtime/contracts.ts
+3
View File
@@ -18,9 +18,12 @@ export type RuntimeState = {
};
export type EngineSnapshot = Snapshot<RuntimeState>;
export type CanonicalProposal = { readonly document: Document; readonly versions: ReadonlyMap<Id, number> };
/** A validated historical execution receipt; its old model is never published over a fresh read. */
export type AppliedModelProof = { readonly versions:ReadonlyMap<Id,number>; assertCanonical(proposal:CanonicalProposal):void };
export type PersistencePort = {
/** Complete canonical read for a quiescent recovery session; never a legacy fallback. */
read?: (signal: AbortSignal) => Promise<CanonicalProposal>;
verifyApplied?: (request:FrozenRequest,receipt:Json) => AppliedModelProof;
supportsRequest?: (request: FrozenRequest) => boolean;
compile(intent: OperationIntent, requestId: Id, snapshot: EngineSnapshot): CompiledRequest;
submit(request: FrozenRequest, signal: AbortSignal): Promise<OperationStatus>;
src/lib/canvas-runtime/engine.ts
+7 -6
View File
@@ -11,6 +11,7 @@ import { operationCanvases, replayPending, validateCanonicalRead } from './recon
import { decodeRecovery, encodeRecovery } from './recovery.ts';
import { createScheduler, type Clock, type OperationStatus } from './scheduler.ts';
import { planConflictRecovery, type ConflictPlan, type RecoveryChoice } from './conflicts.ts';
import { stageAppliedRecovery } from './applied-recovery.ts';
import { conflictsWith, maxResourceBytes, operationFootprint, referencedResources, type AssetPort, type ResourceRecord, type ResourceHandle, type StagedFile } from './resources.ts';
import type { StoragePort, StorageScope } from './storage.ts';
@@ -24,6 +25,7 @@ export type RecoverySession = {
readonly baseRootRevision: number;
readonly canonical: CanonicalProposal;
readonly pendingOperationIds: readonly Id[];
readonly acknowledgedRequestIds: readonly Id[];
/** A conflict is diagnostic only: the existing document/history/requests remain intact. */
readonly conflict?: { readonly code: string; readonly targets: readonly Id[] };
plan(choices:ReadonlyMap<Id,RecoveryChoice>):ConflictPlan;
@@ -311,21 +313,20 @@ export async function openProjectEngine(options: EngineOptions) {
await Promise.all([scheduler.settle(),runTracker.settle()]);assertCurrent();
while(digestTasks.size){await Promise.all([...digestTasks.values()]);assertCurrent();}
await persist();assertCurrent();
const root=editor.getSnapshot(),previous=root.runtime;
const unresolved=[...previous.ledger.requests.values()].filter(request=>request.phase!=='resolved');
if(unresolved.length)throw fault('recovery_waiting_for_receipts',unresolved.map(request=>request.id));
const staged=stageAppliedRecovery(editor.getSnapshot(),{persistence:options.persistence,runs:options.runs,assets:options.assets,now:runClock.now()});
const root=staged.snapshot,previous=root.runtime;
const loaded=await network.run(controller.signal,()=>options.persistence.read!(controller.signal));assertCurrent();
if(editor.getSnapshot().rootRevision!==root.rootRevision)throw fault('stale_plan');
validateCanonicalRead(previous.acknowledged,previous.versions,loaded.document,loaded.versions,options.registry);
const acknowledged=createDocument(loaded.document.projectId,[...loaded.document.canvases.values()],[...loaded.document.elements.values()],loaded.document.tombstones);
const canonical=Object.freeze({document:acknowledged,versions:readMap(loaded.versions)});
const canonical=Object.freeze({document:acknowledged,versions:readMap(loaded.versions)});staged.assertCanonical(canonical);
let working:Document|undefined,conflict:RecoverySession['conflict'];
try{working=replayPending(acknowledged,previous.ledger,options.registry);}catch(error){const rejected=reject(error);conflict=freeze({code:rejected.code,targets:rejected.targets});}
let committed:ConflictPlan|'refresh'|undefined,committedIds:readonly Id[]=[];
const plans=new WeakSet<ConflictPlan>();
const finish=async()=>{await persist();assertCurrent();scheduler.notify(committedIds);active.release();return{status:'applied' as const,revision:editor.getSnapshot().revision,operationIds:committedIds};};
const finish=async()=>{await persist();assertCurrent();await scheduler.resume();assertCurrent();scheduler.notify(committedIds);runTracker.notify(staged.runIds);active.release();return{status:'applied' as const,revision:editor.getSnapshot().revision,operationIds:committedIds};};
active.cancel=async()=>{if(recovery!==active)return;if(committed){await finish();return;}active.release();};
return Object.freeze({baseRootRevision:root.rootRevision,canonical,
return Object.freeze({baseRootRevision:root.rootRevision,canonical,acknowledgedRequestIds:staged.requestIds,
pendingOperationIds:Object.freeze([...previous.ledger.operations.values()].filter(record=>record.phase!=='resolved').map(record=>record.intent.id)),
...(conflict?{conflict}:{}),
plan(choices:ReadonlyMap<Id,RecoveryChoice>){
src/lib/canvas-runtime/reconcile.ts
+1
View File
@@ -52,6 +52,7 @@ export function replayPending(base: Document, ledger: LedgerSnapshot, registry:
for (const record of [...ledger.operations.values()].sort((a, b) => a.intent.sequence - b.intent.sequence)) {
if (record.phase === 'resolved' || !['document.change','run.apply'].includes(record.intent.operationType)) continue;
for (const change of record.intent.changes) {
if(change.kind==='field'){const owner=document.elements.get(change.target.id)?.ownerCanvasId;if(owner&&record.intent.canvasIds?.length&&!record.intent.canvasIds.includes(owner))throw fault('operation_context_changed',[change.target.id]);}
if (satisfied(document, change)) continue;
document = applyChanges(document, [change]);
}
src/lib/canvas-runtime/runs.ts
+2 -1
View File
@@ -1,5 +1,5 @@
import type { Id, Json, OperationIntent, Target, RunInput } from '../canvas-core/types.ts';
import type { CanonicalProposal, EngineSnapshot } from './contracts.ts';
import type { AppliedModelProof, CanonicalProposal, EngineSnapshot } from './contracts.ts';
import type { FrozenRequest } from './ledger.ts';
import type { CompiledRequest, OperationStatus, Clock } from './scheduler.ts';
import { fault, validJson } from '../canvas-core/values.ts';
@@ -21,6 +21,7 @@ export type RunRecord = {
readonly controlEpoch?: number; readonly appliedOutputIds?: readonly Id[]; readonly nextPollAt?: number; readonly error?: string; readonly blocked?: boolean; readonly pendingPage?: RunPage;
};
export type RunPort = {
verifyApplied?: (request:FrozenRequest,receipt:Json)=>{readonly model:AppliedModelProof;readonly admission?:RunRecord;readonly action?:{readonly runId:Id;readonly runStatus?:RunStatus;readonly outputId?:Id}};
catalog(boardId: Id, signal: AbortSignal): Promise<RunCatalog>;
compile(intent: OperationIntent, requestId: Id, snapshot: EngineSnapshot): CompiledRequest;
submit(request: FrozenRequest, signal: AbortSignal): Promise<OperationStatus>;
src/lib/canvas-service/foundry.ts
+28 -1
View File
@@ -1,7 +1,7 @@
import type { CanvasRecord, Document, ElementRecord, Id, Json, OperationIntent, Registry } from '../canvas-core/types.ts';
import { createDocument, validateDocument } from '../canvas-core/document.ts';
import { equal, fault, freeze, readMap, validJson } from '../canvas-core/values.ts';
import type { EngineSnapshot, PersistencePort } from '../canvas-runtime/contracts.ts';
import type { AppliedModelProof, EngineSnapshot, PersistencePort } from '../canvas-runtime/contracts.ts';
import type { FrozenRequest } from '../canvas-runtime/ledger.ts';
import type { OperationStatus } from '../canvas-runtime/scheduler.ts';
@@ -114,6 +114,13 @@ export function createFoundryPersistence(options: { workspaceId: Id; rootId: Id;
}
return Object.freeze({
supportsRequest: (request: FrozenRequest) => request.protocolVersion === 2,
verifyApplied(request:FrozenRequest,receipt:Json){
validateFrozen(request);
if(statusFor(request,{status:200,body:receipt},false).outcome!=='applied')throw fault('receipt_identity_mismatch');
const proof=verifyFoundryModelProof(record(receipt).model,request.canvasIds,options.rootId);
for(const [canvas,version]of proof.versions)if(version!==request.expectedVersions[canvas]+1)throw fault('receipt_version_mismatch',[canvas]);
return proof;
},
async read(signal: AbortSignal) {
const loaded=await loadFoundryModel(options.transport,options.workspaceId,options.rootId,options.registry,signal);
if(loaded.status!=='ready')throw fault('canonical_migration_required');
@@ -142,6 +149,26 @@ export function createFoundryPersistence(options: { workspaceId: Id; rootId: Id;
});
}
/** Verify coverage independently of the current ACK graph. Only a complete fresh model is adopted. */
export function verifyFoundryModelProof(value:unknown,participants:readonly Id[],rootId:Id):AppliedModelProof {
const chunk=modelChunk(value);
if(chunk.projectId!==rootId||chunk.complete||chunk.versions.size!==participants.length||participants.some(id=>!chunk.versions.has(id)))throw fault('receipt_coverage_mismatch');
return Object.freeze({versions:chunk.versions,assertCanonical(proposal){
if(proposal.document.projectId!==rootId)throw fault('project_mismatch');
const unchanged=new Set<Id>(),canvases=new Map(chunk.canvases.map(canvas=>[canvas.id,canvas]));
for(const [canvas,version]of chunk.versions){
const fresh=proposal.versions.get(canvas);if(fresh===undefined||fresh<version)throw fault('recovery_read_behind_receipt',[canvas]);
if(fresh!==version)continue;
if(!equal(canvases.get(canvas),proposal.document.canvases.get(canvas)))throw fault('canonical_version_mismatch',[canvas]);
unchanged.add(canvas);
}
const expected=new Map(chunk.elements.filter(element=>unchanged.has(element.ownerCanvasId)).map(element=>[element.id,element]));
let count=0;for(const element of proposal.document.elements.values())if(unchanged.has(element.ownerCanvasId)){count++;if(!equal(expected.get(element.id),element))throw fault('canonical_version_mismatch',[element.ownerCanvasId]);}
if(count!==expected.size)throw fault('canonical_version_mismatch');
for(const [id,incarnation]of chunk.tombstones)if((proposal.document.tombstones.get(id)??0)<incarnation)throw fault('stale_canonical_lifecycle',[id]);
}});
}
/** Shared participant replacement for document receipts and canonical run events. */
export function mergeFoundryModel(value: unknown, participants: readonly Id[], snapshot: EngineSnapshot, registry: Registry, rootId: Id) {
const chunk = modelChunk(value);
src/lib/canvas-service/runs.ts
+34 -22
View File
@@ -1,10 +1,10 @@
import type { Id, Json, Registry, RunInput } from '../canvas-core/types.ts';
import { equal, fault, freeze, validJson } from '../canvas-core/values.ts';
import { equal, fault, freeze, readMap, validJson } from '../canvas-core/values.ts';
import { validateRunInput } from '../canvas-runtime/runs.ts';
import type { RunEvent, RunPage, RunPort, RunRecord, RunStatus, RunModel } from '../canvas-runtime/runs.ts';
import type { FrozenRequest } from '../canvas-runtime/ledger.ts';
import type { OperationStatus } from '../canvas-runtime/scheduler.ts';
import { mergeFoundryModel, mergeFoundryRunContext, type ServiceTransport } from './foundry.ts';
import { mergeFoundryModel, mergeFoundryRunContext, verifyFoundryModelProof, type ServiceTransport } from './foundry.ts';
const uuid = /^[a-f\d]{8}-[a-f\d]{4}-[a-f\d]{4}-[a-f\d]{4}-[a-f\d]{12}$/i;
const statuses = new Set(['queued','running','succeeded','failed','conflict','cancelled','partially_succeeded']);
@@ -41,6 +41,31 @@ function planSteps(plan: unknown, rootId: Id) {
export function createFoundryRuns(options: {workspaceId: Id; rootId: Id; registry: Registry; transport: ServiceTransport}): RunPort {
if(!uuid.test(options.workspaceId)||!uuid.test(options.rootId))throw fault('invalid_service_scope');
const path=(board: Id)=>`workspaces/${encodeURIComponent(options.workspaceId)}/materials/${encodeURIComponent(board)}/canvas/ai`;
function admissionReceipt(request:FrozenRequest,receipt:Json){
const raw=payload(request),body=object(receipt),runId=string(body.run_id),board=request.ownerCanvasId!;
if(request.kind!=='run.start'||body.protocol_version!==2||body.request_id!==request.id||body.operation_id!==raw.operation_id||body.admission!=='accepted'||body.action!==undefined&&body.action!=='run.start'||body.status!=='queued'||!uuid.test(runId)||body.target_incarnation!==raw.target_incarnation||body.document_format_version!==1||body.canvas_version!==request.expectedVersions[board]+1)throw fault('run_identity_mismatch');
const steps=planSteps(body.plan,options.rootId),frozen=object(object(body.plan).frozen_versions),expected={...request.expectedVersions,[board]:request.expectedVersions[board]+1};
// The admission vector covers all active canvases; the execution plan lists only snapshots actually read.
if(frozen[board]!==expected[board]||Object.entries(frozen).some(([id,version])=>!Object.hasOwn(expected,id)||version!==expected[id])||!steps.some(step=>step.board_id===board&&step.node_id===raw.node_id&&step.target_incarnation===raw.target_incarnation)||steps.some(step=>!Object.hasOwn(frozen,step.board_id as Id)))throw fault('run_identity_mismatch');
const model=verifyFoundryModelProof(body.model,[board],options.rootId),target=array(object(body.model).elements).map(object).find(element=>element.id===raw.node_id);
if(model.versions.get(board)!==body.canvas_version||!target||target.ownerCanvasId!==board||target.incarnation!==raw.target_incarnation)throw fault('run_identity_mismatch');
const admission:RunRecord=freeze({id:runId,requestId:request.id,operationId:request.operationIds[0],boardId:board,target:{id:raw.node_id as Id,incarnation:raw.target_incarnation as number},plan:body.plan as Json,status:'queued',cursor:0,hasMore:true,steps:[],outputs:[],outputEvents:[],attempt:0});
return{model,admission,rawModel:body.model};
}
function actionReceipt(request:FrozenRequest,receipt:Json){
const raw=payload(request),body=object(receipt);
if(!request.runId||body.protocol_version!==2||body.request_id!==request.id||body.operation_id!==request.operationIds[0]||body.run_id!==request.runId||body.action!==request.kind||body.status!=='applied')throw fault('run_identity_mismatch');
if(request.kind==='run.cancel'){
if(!statuses.has(string(body.run_status)))throw fault('invalid_run_response');
return{action:{runId:request.runId,runStatus:body.run_status as RunStatus},model:{versions:readMap(new Map<Id,number>()),assertCanonical(){}},rawModel:undefined};
}
if(request.kind!=='run.apply'||body.node_id!==raw.node_id||body.output_id!==raw.output_id||body.board_id!==raw.target_canvas_id||body.target_incarnation!==raw.target_incarnation||body.document_format_version!==1)throw fault('run_identity_mismatch');
const source=object(body.model),board=raw.target_canvas_id as Id;
const scoped=source.complete===true?{...source,complete:false,canvases:array(source.canvases).filter(value=>object(value).id===board),elements:array(source.elements).filter(value=>object(value).ownerCanvasId===board),versions:{[board]:object(source.versions)[board]},tombstones:{},removedCanvasIds:[]}:source;
const model=verifyFoundryModelProof(scoped,[board],options.rootId),version=integer(body.canvas_version,1),target=array(scoped.elements).map(object).find(element=>element.id===raw.node_id);
if(model.versions.get(board)!==version||version!==request.expectedVersions[board]+1||target?.incarnation!==raw.target_incarnation||target?.ownerCanvasId!==board)throw fault('run_identity_mismatch');
return{action:{runId:request.runId,outputId:raw.output_id as Id},model,rawModel:scoped};
}
async function execute(request: FrozenRequest,query: boolean,signal: AbortSignal): Promise<OperationStatus> {
try{
const raw=payload(request),suffix=request.kind==='run.start'?'runs':`runs/${encodeURIComponent(request.runId!)}/${request.kind==='run.cancel'?'cancel':'apply'}`;
@@ -83,29 +108,16 @@ export function createFoundryRuns(options: {workspaceId: Id; rootId: Id; registr
return freeze({kind:'run.start',ownerCanvasId:run.boardId,protocolVersion:2,canvasIds,expectedVersions,payload:{protocol_version:2,request_id:requestId,operation_id:intent.id,target_incarnation:run.target.incarnation,expected_versions:expectedVersions,node_id:run.target.id,model_id:run.modelId,prompt:run.prompt,parameters:run.parameters,propagate:run.propagate}});
},
submit:(request,signal)=>execute(request,false,signal),query:(request,signal)=>execute(request,true,signal),
verifyApplied(request,receipt){return request.kind==='run.start'?admissionReceipt(request,receipt):actionReceipt(request,receipt);},
accept(request,status,snapshot){
const raw=payload(request),body=object(status.canonical),runId=string(body.run_id),board=request.ownerCanvasId!;
if(status.outcome!=='applied'||body.protocol_version!==2||body.request_id!==request.id||body.operation_id!==raw.operation_id||body.admission!=='accepted'||body.action!==undefined&&body.action!=='run.start'||body.status!=='queued'||!uuid.test(runId)||body.target_incarnation!==raw.target_incarnation||body.document_format_version!==1||body.canvas_version!==request.expectedVersions[board]+1)throw fault('run_identity_mismatch');
const steps=planSteps(body.plan,options.rootId);
if(!steps.some(step=>step.board_id===board&&step.node_id===raw.node_id&&step.target_incarnation===raw.target_incarnation)||steps.some(step=>!request.canvasIds.includes(step.board_id as Id)))throw fault('run_identity_mismatch');
const proposal=mergeFoundryModel(body.model,[board],snapshot,options.registry,options.rootId),target=proposal.document.elements.get(raw.node_id as string);
if(proposal.versions.get(board)!==body.canvas_version||!target||target.ownerCanvasId!==board||target.incarnation!==raw.target_incarnation)throw fault('run_identity_mismatch');
const run:RunRecord=freeze({id:runId,requestId:request.id,operationId:request.operationIds[0],boardId:board,target:{id:target.id,incarnation:target.incarnation},plan:body.plan as Json,status:'queued',cursor:0,hasMore:true,steps:[],outputs:[],outputEvents:[],attempt:0});
return{run,proposal};
if(status.outcome!=='applied'||status.canonical===undefined)throw fault('run_identity_mismatch');
const parsed=admissionReceipt(request,status.canonical),proposal=mergeFoundryModel(parsed.rawModel,[request.ownerCanvasId!],snapshot,options.registry,options.rootId);
return{run:parsed.admission,proposal};
},
acceptAction(request,status,snapshot){
const raw=payload(request),body=object(status.canonical);
if(!request.runId||status.outcome!=='applied'||body.protocol_version!==2||body.request_id!==request.id||body.operation_id!==request.operationIds[0]||body.run_id!==request.runId||body.action!==request.kind||body.status!=='applied')throw fault('run_identity_mismatch');
if(request.kind==='run.cancel'){
if(!statuses.has(string(body.run_status)))throw fault('invalid_run_response');
return{runId:request.runId,runStatus:body.run_status as RunStatus};
}
if(request.kind!=='run.apply'||body.node_id!==raw.node_id||body.output_id!==raw.output_id||body.board_id!==raw.target_canvas_id||body.target_incarnation!==raw.target_incarnation||body.document_format_version!==1)throw fault('run_identity_mismatch');
const model=object(body.model),board=raw.target_canvas_id as Id;
const scoped=model.complete===true?{...model,complete:false,canvases:array(model.canvases).filter(value=>object(value).id===board),elements:array(model.elements).filter(value=>object(value).ownerCanvasId===board),versions:{[board]:object(model.versions)[board]},tombstones:{},removedCanvasIds:[]}:model;
const proposal=mergeFoundryModel(scoped,[board],snapshot,options.registry,options.rootId),target=proposal.document.elements.get(raw.node_id as Id),version=integer(body.canvas_version,1);
if(proposal.versions.get(raw.target_canvas_id as Id)!==version||version!==request.expectedVersions[raw.target_canvas_id as Id]+1||target?.incarnation!==raw.target_incarnation||target?.ownerCanvasId!==raw.target_canvas_id)throw fault('run_identity_mismatch');
return{runId:request.runId,outputId:raw.output_id as Id,proposal};
if(status.outcome!=='applied'||status.canonical===undefined)throw fault('run_identity_mismatch');
const parsed=actionReceipt(request,status.canonical);
return{...parsed.action,...(parsed.rawModel?{proposal:mergeFoundryModel(parsed.rawModel,request.canvasIds,snapshot,options.registry,options.rootId)}:{})};
},
async poll(run,signal){
const response=await options.transport({method:'GET',path:`${path(run.boardId)}/runs/${encodeURIComponent(run.id)}?after_seq=${run.cursor}&limit=100`,signal}),body=object(response.body);
tests/headless/applied-recovery.test.mjs
+61
View File
@@ -0,0 +1,61 @@
import { test } from 'node:test';
import assert from 'node:assert/strict';
import { createRegistry,builtinComponents,createDocument,applyChanges } from '../../src/lib/canvas-core/index.ts';
import { openProjectEngine } from '../../src/lib/canvas-runtime/engine.ts';
import { createFoundryPersistence } from '../../src/lib/canvas-service/foundry.ts';
import { decodeRecovery } from '../../src/lib/canvas-runtime/recovery.ts';
const root='10000000-0000-4000-8000-000000000001',workspace='10000000-0000-4000-8000-000000000002',account='10000000-0000-4000-8000-000000000003';
const target={id:'a',incarnation:1},element={...target,structure:'node',componentId:'text',dataVersion:1,ownerCanvasId:root,data:{title:'a',text:''},geometry:{x:0,y:0,width:100,height:100,rotation:0}};
const registry=createRegistry(builtinComponents()),original=createDocument(root,[{id:root,parent:null}],[element]);
function fixture(){
let server=original,version=1,image,leader;const calls=[],receipts=new Map();
const model=(document=server,currentVersion=version,complete=true)=>({documentFormatVersion:1,projectId:root,complete,canvases:[...document.canvases.values()],elements:[...document.elements.values()],tombstones:Object.fromEntries(document.tombstones),versions:{[root]:currentVersion},removedCanvasIds:[]});
const f={failCodec:true,broken:false,behind:false,calls,receipts,storage:{async claim(scope){assert.equal(leader,undefined);return leader={scope,fence:1,active:true,async release(){leader=undefined;}};},async recover(){return image&&structuredClone(image);},async commit(owner,batch){if(f.broken)throw Error('quota');assert.equal(owner,leader);image=structuredClone(batch.image);return{durableRootRevision:image.rootRevision};},async readFile(){},close(){}},image:()=>structuredClone(image),remoteGeometry(patch){const before=server.elements.get('a').geometry;server=applyChanges(server,[{kind:'field',target,field:'geometry',before,after:{...before,...patch}}]);version++;}};
const transport=async request=>{
calls.push(request);if(request.method==='GET'){if(request.path.endsWith('/model'))return{status:200,body:{model:f.behind?model(original,1):model()}};const requestId=request.path.split('/').at(-1),receipt=receipts.get(requestId);return receipt?{status:200,body:receipt}:{status:404,body:{protocol_version:2,request_id:requestId,status:'unknown'}};}
const input=JSON.parse(request.body);assert.equal(input.expected_versions[root],version);for(const operation of input.operations)server=applyChanges(server,operation.changes);version++;
const receipt={protocol_version:2,request_id:request.requestId,status:'applied',operation_ids:input.operations.map(operation=>operation.id),model:model(server,version,false)};receipts.set(request.requestId,receipt);return{status:200,body:receipt};
};
const base=createFoundryPersistence({workspaceId:workspace,rootId:root,registry,transport});
f.port={...base,canonical(...args){if(f.failCodec)throw Error('codec_temporarily_unavailable');return base.canonical(...args);}};
f.open=()=>openProjectEngine({scope:{accountId:account,workspaceId:workspace,projectId:root},registry,document:original,versions:new Map([[root,1]]),storage:f.storage,persistence:f.port});
return f;
}
async function applied(f){const engine=await f.open();engine.editor.dispatch({type:'element.data',target,patch:{title:'first'}});await engine.settle();const request=[...engine.getSnapshot().runtime.ledger.requests.values()][0];assert.equal(request.outcome,'applied');assert.equal(request.phase,'waiting');return{engine,request};}
test('a verified applied request adopts a newer full read without reapplying its old receipt or generating a new write',async()=>{
const f=fixture(),{engine,request}=await applied(f);f.remoteGeometry({x:500,width:300});const history=engine.getSnapshot().history,session=await engine.prepareRecovery();
assert.deepEqual(session.acknowledgedRequestIds,[request.id]);assert.deepEqual(session.pendingOperationIds,[]);assert.equal(engine.getSnapshot().runtime.ledger.requests.get(request.id).phase,'waiting');
await session.apply();await engine.settle();const current=engine.getSnapshot();assert.equal(current.history,history);assert.equal(current.document.elements.get('a').data.title,'first');assert.equal(current.document.elements.get('a').geometry.x,500);assert.equal(current.document.elements.get('a').geometry.width,300);
assert.equal(current.runtime.versions.get(root),3);assert.equal(current.runtime.ledger.requests.get(request.id).phase,'resolved');assert.equal(current.runtime.ledger.requests.get(request.id).body,request.body);assert.equal(f.calls.filter(call=>call.method==='POST').length,1);
await engine.close();const recovered=await f.open();await recovered.settle();assert.equal(recovered.getSnapshot().runtime.ledger.requests.get(request.id).phase,'resolved');assert.equal(f.calls.filter(call=>call.method==='POST').length,1);await recovered.close();
});
test('acknowledgement and explicit replan commit together; original writer releases only after the new cut is durable',async()=>{
const f=fixture(),{engine,request}=await applied(f);const edit=engine.editor.dispatch({type:'element.geometry',target,patch:{x:200}});await engine.settle();f.remoteGeometry({x:500,width:300});
const session=await engine.prepareRecovery();assert.ok(session.conflict);assert.deepEqual(session.pendingOperationIds,edit.operationIds);const plan=session.plan(new Map([[edit.operationIds[0],'keep-local']]));
assert.equal(plan.audit.decisions.length,1);assert.equal(plan.ledger.operations.get(request.operationIds[0]).resolution,'applied');f.failCodec=false;f.broken=true;await assert.rejects(session.resolve(plan),/quota/);assert.equal(f.calls.filter(call=>call.method==='POST').length,1);
f.broken=false;await session.resolve(plan);await engine.settle();assert.equal(f.calls.filter(call=>call.method==='POST').length,2);assert.equal(engine.getSnapshot().document.elements.get('a').geometry.width,300);
assert.equal(engine.editor.history.undo().status,'applied');await engine.settle();assert.equal(engine.getSnapshot().document.elements.get('a').geometry.x,500);assert.equal(engine.getSnapshot().document.elements.get('a').geometry.width,300);
assert.equal(engine.editor.history.undo().status,'applied');await engine.settle();assert.equal(engine.getSnapshot().document.elements.get('a').data.title,'a');
assert.equal(decodeRecovery(f.image(),registry).runtime.ledger.requests.get(request.id).body,request.body);await engine.close();
});
test('a cancelled acknowledgement preview leaves the applied request pending and its original history untouched',async()=>{
const f=fixture(),{engine,request}=await applied(f),before=engine.getSnapshot(),session=await engine.prepareRecovery();await session.cancel();
assert.equal(engine.getSnapshot(),before);assert.equal(engine.getSnapshot().runtime.ledger.requests.get(request.id).phase,'waiting');assert.equal(f.calls.filter(call=>call.method==='POST').length,1);await engine.close();
});
test('a fresh read behind a proven receipt cannot release its claim; malformed and missing proofs remain preserved',async()=>{
const f=fixture(),{engine,request}=await applied(f);f.behind=true;await assert.rejects(engine.prepareRecovery(),/recovery_read_behind_receipt/);assert.equal(engine.getSnapshot().runtime.ledger.requests.get(request.id).phase,'waiting');
f.behind=false;const beforeReads=f.calls.filter(call=>call.path.endsWith('/model')).length;
const real=engine.getSnapshot().runtime.receipts.get(request.id);engine.editor.updateRuntime(runtime=>({...runtime,receipts:new Map([[request.id,{...real,operation_ids:['wrong']} ]])}));await assert.rejects(engine.prepareRecovery(),/receipt_coverage_mismatch/);
assert.equal(f.calls.filter(call=>call.path.endsWith('/model')).length,beforeReads);engine.editor.updateRuntime(runtime=>({...runtime,receipts:new Map()}));await assert.rejects(engine.prepareRecovery(),/recovery_receipt_missing/);assert.equal(f.calls.filter(call=>call.method==='POST').length,1);await engine.close();
});
test('a locally durable success proof survives server receipt expiry without resending the original operation',async()=>{
const f=fixture(),{engine,request}=await applied(f);await engine.close();f.receipts.clear();const recovered=await f.open();await recovered.settle();
const session=await recovered.prepareRecovery();assert.deepEqual(session.acknowledgedRequestIds,[request.id]);await session.apply();await recovered.settle();
assert.equal(recovered.getSnapshot().runtime.ledger.requests.get(request.id).phase,'resolved');assert.equal(f.calls.filter(call=>call.method==='POST').length,1);assert.equal(recovered.getSnapshot().document.elements.get('a').data.title,'first');await recovered.close();
});
tests/headless/browser-storage-cases.mjs
+27
View File
@@ -5,6 +5,31 @@ import { createFoundryRuns } from '../../src/lib/canvas-service/runs.ts';
import { createFoundryAssets } from '../../src/lib/canvas-service/assets.ts';
import { createFoundryPersistence } from '../../src/lib/canvas-service/foundry.ts';
async function appliedReceiptRecoveryCase(name,assert){
const root=crypto.randomUUID(),workspace=crypto.randomUUID(),account=crypto.randomUUID(),registry=createRegistry(builtinComponents()),target={id:'a',incarnation:1};
const element={...target,structure:'node',componentId:'text',dataVersion:1,ownerCanvasId:root,geometry:{x:0,y:0,width:300,height:200,rotation:0},data:{title:'a',text:'original'}};
const original=createDocument(root,[{id:root,parent:null}],[element]),storage=createBrowserStorage({name:`${name}-applied-receipt`}),calls=[],receipts=new Map();let server=original,version=1,engine,broken=true;
const model=complete=>({documentFormatVersion:1,projectId:root,complete,removedCanvasIds:[],canvases:[...server.canvases.values()],elements:[...server.elements.values()],tombstones:{},versions:{[root]:version}});
const transport=async request=>{
calls.push(request);if(request.method==='GET')return{status:200,body:request.path.endsWith('/model')?{model:model(true)}:receipts.get(request.path.split('/').at(-1))};
const body=JSON.parse(request.body);assert(body.expected_versions[root]===version,'write did not use reviewed version');for(const operation of body.operations)server=applyChanges(server,operation.changes);version++;
const receipt={protocol_version:2,request_id:request.requestId,status:'applied',operation_ids:body.operations.map(operation=>operation.id),model:model(false)};receipts.set(request.requestId,receipt);return{status:200,body:receipt};
};
const base=createFoundryPersistence({workspaceId:workspace,rootId:root,registry,transport}),persistence={...base,canonical(...args){if(broken)throw Error('codec_temporarily_unavailable');return base.canonical(...args);}};
const options={scope:{accountId:account,workspaceId:workspace,projectId:root},registry,document:original,versions:new Map([[root,1]]),storage,persistence};
try{
engine=await openProjectEngine(options);engine.editor.dispatch({type:'element.data',target,patch:{title:'first'}});await engine.settle();
const request=[...engine.getSnapshot().runtime.ledger.requests.values()][0];assert(request.outcome==='applied'&&request.phase==='waiting','failed coordination lost the successful receipt');
const edit=engine.editor.dispatch({type:'element.geometry',target,patch:{x:200}});await engine.settle();await engine.close();
server=applyChanges(server,[{kind:'field',target,field:'geometry',before:element.geometry,after:{...element.geometry,x:500,width:400}}]);version++;
engine=await openProjectEngine(options);const session=await engine.prepareRecovery();assert(session.acknowledgedRequestIds.includes(request.id),'saved applied receipt was not recovered');assert(session.pendingOperationIds.length===1&&session.pendingOperationIds[0]===edit.operationIds[0],'already executed operation was offered for local re-execution');
const plan=session.plan(new Map([[edit.operationIds[0],'keep-local']]));broken=false;await session.resolve(plan);await engine.settle();assert(calls.filter(call=>call.method==='POST').length===2,'successful original operation was sent again');
assert(engine.getSnapshot().runtime.ledger.requests.get(request.id).body===request.body,'original frozen receipt identity changed');await engine.close();
engine=await openProjectEngine(options);assert(engine.getSnapshot().runtime.ledger.requests.get(request.id).phase==='resolved','acknowledgement and rebased document did not recover together');assert(engine.getSnapshot().document.elements.get('a').geometry.width===400,'old receipt rewound newer geometry');
assert(engine.editor.history.undo().status==='applied','rebased geometry history did not recover');await engine.settle();assert(engine.getSnapshot().document.elements.get('a').geometry.x===500,'undo did not return to the reviewed geometry');
}finally{await engine?.close();storage.close();indexedDB.deleteDatabase(`${name}-applied-receipt`);}
}
async function conflictReplanCase(name,assert){
const root=crypto.randomUUID(),workspace=crypto.randomUUID(),account=crypto.randomUUID(),registry=createRegistry(builtinComponents()),target={id:'a',incarnation:1};
const element={...target,structure:'node',componentId:'text',dataVersion:1,ownerCanvasId:root,geometry:{x:0,y:0,width:300,height:200,rotation:0},data:{title:'a',text:'original'}};
@@ -256,6 +281,8 @@ export async function run() {
results.push('quiescent canonical refresh preserves original requests and history; the new base/document/version cut survives IndexedDB reopening');
await conflictReplanCase(name,assert);
results.push('conflict replan persists its audit/history/new identity; a lost replacement ACK recovers by the exact key and undo restores the reviewed server value');
await appliedReceiptRecoveryCase(name,assert);
results.push('a persisted successful receipt with failed coordination adopts a fresh base and replan in one cut, releases its writer and never repeats the original write');
return results;
} finally { await leader?.release(); other?.close(); storage.close(); indexedDB.deleteDatabase(name); }
}
tests/headless/resources.test.mjs
+10
View File
@@ -35,6 +35,7 @@ function fixture() {
accept: (_request, status) => status.canonical,
};
const persistence = {
async read(){return{document:server,versions:new Map([['canvas',version]])};},
compile(intent, id, snapshot) { return { payload: { intent, id }, canvasIds: intent.canvasIds, expectedVersions: Object.fromEntries(intent.canvasIds.map(id => [id, snapshot.runtime.versions.get(id)])) }; },
async submit(request) { const { intent } = JSON.parse(request.body); writes.push(request); server = applyChanges(server, intent.changes); receipts.set(request.id, { document: server, versions: new Map([['canvas', ++version]]) }); return { outcome: 'applied' }; },
async query(request) { return receipts.has(request.id) ? { outcome: 'applied' } : { outcome: 'unresolved' }; },
@@ -154,3 +155,12 @@ test('a failed image durable cut keeps the visible element and sends neither byt
assert.equal(decodeRecovery(f.image(), registry).history.undo.length, 1);
assert.ok(errors.length);
});
test('known successful upload coordination recovers stored identity and releases the dependent image without uploading again',async()=>{
const f=fixture(),accept=f.assets.accept;let fail=true;f.assets.accept=(...args)=>{if(fail)throw Error('codec_temporarily_unavailable');return accept(...args);};
const engine=await f.open(),resource=addImage(engine);await until(()=>f.uploads.length===1);await engine.settle();const request=f.uploads[0],history=engine.getSnapshot().history;
assert.equal(engine.getSnapshot().runtime.ledger.requests.get(request.id).phase,'waiting');assert.equal(f.writes.length,0);fail=false;
const session=await engine.prepareRecovery();assert.deepEqual(session.acknowledgedRequestIds,[request.id]);await session.apply();await engine.settle();
assert.equal(f.uploads.length,1);assert.equal(f.writes.length,1);assert.equal(engine.getSnapshot().history,history);assert.equal(engine.getSnapshot().runtime.resources.get(resource.id).stored.id,resource.id);assert.equal(engine.getSnapshot().runtime.ledger.requests.get(request.id).body,request.body);
assert.equal(await f.files.get(resource.id).text(),'image bytes');assert.equal(decodeRecovery(f.image(),registry).runtime.resources.get(resource.id).stored.id,resource.id);await engine.close();
});
tests/headless/runs.test.mjs
+45 -1
View File
@@ -22,6 +22,7 @@ function fixture(){
const context=()=>({...partial([...document.canvases.keys()]),complete:true});
const transport=async request=>{
calls.push(request);
if(request.method==='GET'&&request.path.endsWith('/model'))return{status:200,body:{model:context()}};
if(request.path.endsWith('/model/operations')){
if(docGate)await docGate.promise;
const input=JSON.parse(request.body);let body=receipts.get(request.requestId);
@@ -75,7 +76,7 @@ function fixture(){
};
const f={storage,clock,calls,runs,errors,receipts,transport,loseAdmission:false,
holdAdmission(){postGate=defer();return postGate;},holdDocument(){docGate=defer();return docGate;},
async open(){return openProjectEngine({scope:{accountId:account,workspaceId:workspace,projectId:root},registry,document:original,versions:new Map([[root,1],[child,7]]),storage,clock,createId:()=>id(++next),persistence:createFoundryPersistence({workspaceId:workspace,rootId:root,registry,transport}),runs:createFoundryRuns({workspaceId:workspace,rootId:root,registry,transport}),onError:error=>errors.push(error)});},
async open(){return openProjectEngine({scope:{accountId:account,workspaceId:workspace,projectId:root},registry,document:original,versions:new Map([[root,1],[child,7]]),storage,clock,createId:()=>id(++next),persistence:createFoundryPersistence({workspaceId:workspace,rootId:root,registry,transport}),runs:(()=>{const port=createFoundryRuns({workspaceId:workspace,rootId:root,registry,transport});return{...port,accept(...args){if(f.failAdmissionCodec)throw Error('codec_temporarily_unavailable');return port.accept(...args);},acceptAction(...args){if(f.failActionCodec)throw Error('codec_temporarily_unavailable');return port.acceptAction(...args);}};})(),onError:error=>errors.push(error)});},
move(owner){const element=document.elements.get('a');document=applyChanges(document,[{kind:'field',target,field:'ownerCanvasId',before:element.ownerCanvasId,after:owner}]);versions.set(element.ownerCanvasId,versions.get(element.ownerCanvasId)+1);versions.set(owner,versions.get(owner)+1);},
finish(text='generated',conflict=false){
const [runId,run]=[...runs].at(-1),element=document.elements.get('a'),owner=element.ownerCanvasId;
@@ -264,3 +265,46 @@ test('complete run context retains later unrelated canvas ACKs and rejects chang
const proposal=mergeFoundryRunContext(model,snapshot,registry,root);assert.equal(proposal.document.elements.get('a').data.text,'generated');assert.equal(proposal.document.elements.get('sibling').data.title,'newer title');assert.equal(proposal.versions.get(third),10);
const invalid=structuredClone(model);invalid.elements.find(element=>element.id==='f').data.title='same-version change';assert.throws(()=>mergeFoundryRunContext(invalid,snapshot,registry,root),/same_version_context_conflict/);
});
test('a saved successful AI admission recovers its RunRecord and later local configuration without another invocation',async()=>{
const f=fixture();f.failAdmissionCodec=true;const engine=await f.open();engine.startRun(runInput);await engine.settle();
const request=[...engine.getSnapshot().runtime.ledger.requests.values()][0];assert.equal(request.outcome,'applied');assert.equal(request.phase,'waiting');assert.equal(engine.getSnapshot().runtime.runs.size,0);
const config={enabled:true,modelId:'text-model',prompt:'Local next prompt',parameters:{},propagate:false};const edit=engine.editor.dispatch({type:'element.data',target,patch:{ai:config}});await engine.settle();f.finish('already generated');
const session=await engine.prepareRecovery();assert.deepEqual(session.acknowledgedRequestIds,[request.id]);assert.deepEqual(session.pendingOperationIds,edit.operationIds);assert.ok(session.conflict);
const plan=session.plan(new Map([[edit.operationIds[0],'keep-local']]));f.failAdmissionCodec=false;await session.resolve(plan);await engine.settle();await f.clock.advance(0);await engine.settle();
const current=engine.getSnapshot();assert.equal(current.runtime.runs.size,1);assert.equal([...current.runtime.runs.values()][0].cursor,4);assert.equal([...current.runtime.runs.values()][0].status,'succeeded');assert.equal(current.document.elements.get('a').data.text,'already generated');assert.deepEqual(current.document.elements.get('a').data.ai,config);
assert.equal(f.calls.filter(call=>call.method==='POST'&&call.path.endsWith('/ai/runs')).length,1);assert.equal(current.runtime.ledger.requests.get(request.id).body,request.body);assert.equal(current.runtime.ledger.requests.get(request.id).phase,'resolved');
await engine.close();const restored=await f.open();assert.equal(restored.getSnapshot().runtime.runs.size,1);await restored.close();
});
test('saved successful cancel and apply receipts recover controls without sending the action again',async()=>{
for(const action of ['cancel','apply']){
const f=fixture(),engine=await f.open();engine.startRun(runInput);await engine.settle();const runId=[...engine.getSnapshot().runtime.runs.keys()][0];
if(action==='apply'){f.finish('retained output',true);await f.clock.advance(0);await engine.settle();}
f.failActionCodec=true;if(action==='cancel')engine.cancelRun(runId);else engine.applyRunOutput(runId,target,id(999));await engine.settle();
const request=[...engine.getSnapshot().runtime.ledger.requests.values()].find(request=>request.kind==='run.'+action);assert.equal(request.outcome,'applied');assert.equal(request.phase,'waiting');
let edit;if(action==='apply'){edit=engine.editor.dispatch({type:'element.data',target,patch:{text:'later local'}});await engine.settle();f.move(child);}
const session=await engine.prepareRecovery();assert.deepEqual(session.acknowledgedRequestIds,[request.id]);
f.failActionCodec=false;if(edit){assert.ok(session.conflict);await session.resolve(session.plan(new Map([[edit.operationIds[0],'keep-local']])));}else await session.apply();
await engine.settle();await f.clock.advance(0);await engine.settle();const run=engine.getSnapshot().runtime.runs.get(runId);
assert.equal(f.calls.filter(call=>call.method==='POST'&&call.path.endsWith('/'+action)).length,1);assert.equal(engine.getSnapshot().runtime.ledger.requests.get(request.id).body,request.body);assert.equal(run.controlEpoch,1);
if(action==='cancel')assert.equal(run.status,'cancelled');else{assert.ok(run.appliedOutputIds.includes(id(999)));assert.equal(engine.getSnapshot().document.elements.get('a').data.text,'later local');assert.equal(engine.getSnapshot().document.elements.get('a').ownerCanvasId,child);const documentCall=f.calls.filter(call=>call.method==='POST'&&call.path.endsWith('/model/operations')).at(-1);assert.deepEqual(Object.keys(JSON.parse(documentCall.body).expected_versions),[child]);}
await engine.close();
}
});
test('historical AI success proofs bind immutable request, operation, run, target, version and frozen plan',async()=>{
for(const action of ['start','cancel','apply']){
const f=fixture(),engine=await f.open();engine.startRun(runInput);await engine.settle();const runId=[...engine.getSnapshot().runtime.runs.keys()][0];
if(action==='cancel'){engine.cancelRun(runId);await engine.settle();}
if(action==='apply'){f.finish('retained',true);await f.clock.advance(0);await engine.settle();engine.applyRunOutput(runId,target,id(999));await engine.settle();}
const request=[...engine.getSnapshot().runtime.ledger.requests.values()].find(request=>request.kind==='run.'+action),receipt=f.receipts.get(request.id),port=createFoundryRuns({workspaceId:workspace,rootId:root,registry,transport:f.transport});
const proof=port.verifyApplied(request,receipt);proof.model.assertCanonical(f.server());
if(action==='start'){const subset=structuredClone(receipt);delete subset.plan.frozen_versions[child];assert.ok(port.verifyApplied(request,subset).admission);const foreign=structuredClone(subset);foreign.plan.frozen_versions[id(9900)]=1;assert.throws(()=>port.verifyApplied(request,foreign),/run_identity_mismatch/);}
const mutations=[body=>body.request_id=id(9991),body=>body.operation_id=id(9992),body=>body.action='different',body=>body.run_id=action==='start'?'invalid-run-id':id(9993)];
if(action==='start')mutations.push(body=>body.target_incarnation=2,body=>body.canvas_version++,body=>body.plan.frozen_versions[root]++);
if(action==='apply')mutations.push(body=>body.target_incarnation=2,body=>body.canvas_version++,body=>body.output_id=id(9994),body=>body.board_id=child);
for(const mutate of mutations){const invalid=structuredClone(receipt);mutate(invalid);assert.throws(()=>port.verifyApplied(request,invalid),/run_identity_mismatch|receipt/);}
await engine.close();
}
});
tests/headless/service.test.mjs
+14
View File
@@ -111,3 +111,17 @@ test('recovery reader requires a complete canonical model for its bound project
[{status:403,body:{error:'forbidden'}},/forbidden/],
]){const{port:invalid}=setup(async()=>response);await assert.rejects(invalid.read(controller.signal),error);}
});
test('historical applied proof validates frozen identity and coverage, retains newer reads and rejects same-version divergence',async()=>{
const{port,request}=setup(),status=await port.submit(request,new AbortController().signal),proof=port.verifyApplied(request,status.canonical),proposal=port.canonical(request,status,snapshot);
proof.assertCanonical(proposal);assert.equal(proof.versions.get(root),4);assert.equal(proof.versions.set,undefined);
const changed=createDocument(root,canvases,[{...a,data:{...a.data,title:'later remote'}},f,future]);
proof.assertCanonical({document:changed,versions:new Map([[root,5],[child,7]])});
assert.throws(()=>proof.assertCanonical({document:changed,versions:new Map([[root,4],[child,7]])}),/canonical_version_mismatch/);
assert.throws(()=>proof.assertCanonical({document,versions:new Map([[root,3],[child,7]])}),/recovery_read_behind_receipt/);
for(const mutate of[
body=>body.request_id=account,body=>body.operation_ids=[account],body=>body.protocol_version=99,body=>body.status='accepted',body=>body.model.complete=true,body=>body.model.versions[root]=5,
]){const body=structuredClone(status.canonical);mutate(body);assert.throws(()=>port.verifyApplied(request,body),/receipt|canonical/);}
const tombstone=structuredClone(status.canonical);tombstone.model.tombstones={deleted:2};const lifecycle=port.verifyApplied(request,tombstone);
assert.throws(()=>lifecycle.assertCanonical({document:changed,versions:new Map([[root,5],[child,7]])}),/stale_canonical_lifecycle/);
});