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

Replan reviewed field conflicts with durable audit and conditional history

Browse Source
This commit is contained in:
Codex Bot
2026-11-08 01:56:21 +08:00
parent a0417b4811
commit 89b6f4a253
11 changed files with 453 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/lib/canvas-runtime/engine.ts
+35 -11
View File
@@ -10,6 +10,7 @@ import { createLedger, type FrozenRequest } from './ledger.ts';
import { operationCanvases, replayPending, validateCanonicalRead } from './reconcile.ts';
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 { conflictsWith, maxResourceBytes, operationFootprint, referencedResources, type AssetPort, type ResourceRecord, type ResourceHandle, type StagedFile } from './resources.ts';
import type { StoragePort, StorageScope } from './storage.ts';
@@ -25,6 +26,8 @@ export type RecoverySession = {
readonly pendingOperationIds: 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;
resolve(plan:ConflictPlan):Promise<Result>;
apply(): Promise<Result>;
cancel(): Promise<void>;
};
@@ -46,9 +49,9 @@ export async function openProjectEngine(options: EngineOptions) {
const fileWrites = new Map<Id, Blob>(), digestTasks = new Map<Id, Promise<void>>();
let staged: readonly ResourceRecord[] = [];
let applying: RunApplyInput | undefined;
let recovery: { controller: AbortController; release(): void } | undefined;
let recovery: { controller: AbortController; release(): void; cancel():Promise<void> } | undefined;
const assertWritable = () => { if(closing||closed)throw fault('engine_closing');if(recovery)throw fault('recovery_in_progress'); };
const initial: RuntimeState = recovered?.runtime ?? Object.freeze({ runs: readMap(new Map()), acknowledged: createDocument(options.document.projectId, [...options.document.canvases.values()], [...options.document.elements.values()], options.document.tombstones), versions: readMap(options.versions), ledger: createLedger().getSnapshot(), receipts: readMap(new Map()), conflicts: readMap(new Map()), resources: readMap(new Map()), resourcePins: readMap(new Map()) });
const initial: RuntimeState = recovered?.runtime ?? Object.freeze({ recoveries:readMap(new Map()), runs: readMap(new Map()), acknowledged: createDocument(options.document.projectId, [...options.document.canvases.values()], [...options.document.elements.values()], options.document.tombstones), versions: readMap(options.versions), ledger: createLedger().getSnapshot(), receipts: readMap(new Map()), conflicts: readMap(new Map()), resources: readMap(new Map()), resourcePins: readMap(new Map()) });
const editor = createEditor({ registry: options.registry, document: options.document, recovery: recovered, runtime: initial, createId: options.createId, historyLimit: options.historyLimit,
policy(command, document) { if (closing || closed || recovery) return { enabled: false, code: closing||closed?'engine_closing':'recovery_in_progress', targets: [] }; return options.policy?.(command, document) ?? { enabled: true }; },
traversalPolicy(entry, direction) { return closing || closed || recovery ? { enabled: false, code: closing||closed?'engine_closing':'recovery_in_progress', targets: [] } : options.traversalPolicy?.(entry, direction) ?? { enabled: true }; },
@@ -299,14 +302,14 @@ export async function openProjectEngine(options: EngineOptions) {
assertWritable();
if(!options.persistence.read)throw fault('canonical_reader_unavailable');
const releaseScheduler=scheduler.hold(),releaseRuns=runTracker.hold(),controller=new AbortController();
const active={controller,release(){if(recovery!==active)return;recovery=undefined;releaseRuns();releaseScheduler();}};
const active={controller,release(){if(recovery!==active)return;recovery=undefined;releaseRuns();releaseScheduler();},async cancel(){controller.abort();active.release();}};
recovery=active;
sessions.cancel('recovery');
const assertCurrent=()=>{if(closing||closed)throw fault('engine_closing');if(recovery!==active||controller.signal.aborted)throw fault('recovery_session_closed');};
try {
// Both holds are acquired before awaiting either lane; already admitted callbacks finish normally.
await Promise.all([scheduler.settle(),runTracker.settle()]);
while(digestTasks.size)await Promise.all([...digestTasks.values()]);
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');
@@ -318,24 +321,44 @@ export async function openProjectEngine(options: EngineOptions) {
const canonical=Object.freeze({document:acknowledged,versions:readMap(loaded.versions)});
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=false;
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};};
active.cancel=async()=>{if(recovery!==active)return;if(committed){await finish();return;}active.release();};
return Object.freeze({baseRootRevision:root.rootRevision,canonical,
pendingOperationIds:Object.freeze([...previous.ledger.operations.values()].filter(record=>record.phase!=='resolved').map(record=>record.intent.id)),
...(conflict?{conflict}:{}),
plan(choices:ReadonlyMap<Id,RecoveryChoice>){
assertCurrent();if(committed)throw fault('recovery_already_committed');if(editor.getSnapshot().rootRevision!==root.rootRevision)throw fault('stale_plan');
const plan=planConflictRecovery(root,canonical,choices,options.registry,createId);plans.add(plan);return plan;
},
async resolve(plan:ConflictPlan):Promise<Result>{
assertCurrent();if(!plans.has(plan))throw fault('invalid_recovery_plan');if(committed&&committed!==plan)throw fault('recovery_already_committed');
if(!committed){
const permission=options.policy?.({type:'changes.apply',changes:[]},plan.document);if(permission&&!permission.enabled)return{status:'rejected',code:permission.code,targets:permission.targets};
for(const id of plan.operationIds){const intent=plan.operations.get(id)!;const allowed=options.policy?.({type:'changes.apply',changes:intent.changes},plan.document);if(allowed&&!allowed.enabled)return{status:'rejected',code:allowed.code,targets:allowed.targets};
if(intent.operationType==='run.apply'){const run=previous.runs.get(intent.action.runId);if(!run||run.blocked||run.status!=='conflict'||!runContainsTarget(run,intent.action.target))return{status:'rejected',code:'output_not_applicable',targets:[intent.action.target.id]};}}
const recoveries=new Map(previous.recoveries),resourcePins=new Map(previous.resourcePins),conflicts=new Map(previous.conflicts);recoveries.set(plan.id,plan.audit);resourcePins.set(`recovery:${plan.id}`,plan.resourceIds);conflicts.delete('refresh');
for(const decision of plan.audit.decisions){const requestId=previous.ledger.operations.get(decision.operationId)?.requestId;if(requestId)conflicts.delete(requestId);}
const result=editor.reconcile(plan.document,Object.freeze({...previous,ledger:plan.ledger,acknowledged,versions:canonical.versions,recoveries:readMap(recoveries),resourcePins:readMap(resourcePins),conflicts:readMap(conflicts)}),root.rootRevision,plan);
if(result.status==='rejected')return result;committed=plan;committedIds=plan.operationIds;
}
return finish();
},
async apply():Promise<Result>{
assertCurrent();
if(committed&&committed!=='refresh')throw fault('recovery_already_committed');
if(conflict)return{status:'rejected',...conflict};
if(!committed){
const conflicts=new Map(previous.conflicts);conflicts.delete('refresh');
const result=editor.reconcile(working!,Object.freeze({...previous,acknowledged,versions:canonical.versions,conflicts:readMap(conflicts)}),root.rootRevision);
if(result.status==='rejected')return result;
committed=true;
committed='refresh';
}
// Persistence failure retains the hold; apply/cancel can retry this barrier without republishing.
await persist();assertCurrent();active.release();
return{status:'applied',revision:editor.getSnapshot().revision,operationIds:[]};
return finish();
},
async cancel(){if(recovery!==active)return;if(committed)await persist();active.release();},
cancel:()=>active.cancel(),
});
}catch(error){active.release();throw error;}
},
@@ -433,8 +456,9 @@ export async function openProjectEngine(options: EngineOptions) {
displays.add(release); return Object.freeze({ url, release });
},
getSnapshot: editor.getSnapshot, persist,
cancelRecovery:async()=>{await recovery?.cancel();},
getDurability: () => Object.freeze({ rootRevision: editor.getSnapshot().rootRevision, durableRootRevision }),
retry: async () => { assertWritable();await scheduler.resume();await runTracker.retry(); },
retry: async () => { if(recovery)await recovery.cancel();assertWritable();await scheduler.resume();await runTracker.retry(); },
async settle() { await Promise.resolve(); while (digestTasks.size) await Promise.all([...digestTasks.values()]); await scheduler.settle(); await runTracker.settle(); await persist(); },
async close() {
if (closed) return;