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

Add durable AI cancellation and undoable output application

Browse Source
This commit is contained in:
Codex Bot
2026-11-08 01:03:05 +08:00
parent d07bc2ef91
commit 6243602402
13 changed files with 255 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/lib/canvas-runtime/engine.ts
+64 -10
View File
@@ -1,4 +1,4 @@
import type { Document, Id, Registry, Result, RunInput } from '../canvas-core/types.ts';
import type { Document, Id, Registry, Result, RunInput, RunApplyInput, RunCancelInput, Json, OperationIntent, Data } from '../canvas-core/types.ts';
import { createEditor, type EditorOptions, type Transaction } from '../canvas-core/editor.ts';
import { createDocument } from '../canvas-core/document.ts';
import { createSessions } from '../canvas-core/gestures.ts';
@@ -36,11 +36,13 @@ export async function openProjectEngine(options: EngineOptions) {
const displays = new Set<() => void>();
const fileWrites = new Map<Id, Blob>(), digestTasks = new Map<Id, Promise<void>>();
let staged: readonly ResourceRecord[] = [];
let applying: RunApplyInput | undefined;
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 editor = createEditor({ registry: options.registry, document: options.document, recovery: recovered, runtime: initial, createId: options.createId, historyLimit: options.historyLimit,
policy(command, document) { if (closing || closed) return { enabled: false, code: 'engine_closing', targets: [] }; return options.policy?.(command, document) ?? { enabled: true }; },
traversalPolicy(entry, direction) { return closing || closed ? { enabled: false, code: 'engine_closing', targets: [] } : options.traversalPolicy?.(entry, direction) ?? { enabled: true }; },
onListenerError: report,
prepareIntent: intent => applying && intent.origin==='user' ? {...intent,operationType:'run.apply',action:applying} : intent,
nextOperationSequence(runtime, fallback) { let next = fallback; for (const record of runtime.ledger.operations.values()) next = Math.max(next, record.intent.sequence + 1); return next + staged.length; },
reduceRuntime(previous, next, commit) {
if (!commit.operationIds.length) return previous;
@@ -55,7 +57,7 @@ export async function openProjectEngine(options: EngineOptions) {
for (const id of commit.operationIds) {
const intent = next.operations.get(id)!, footprint = operationFootprint(intent, next.document);
const dependencies = new Set(intent.dependencies);
for (const record of ledger.getSnapshot().operations.values()) if (record.phase !== 'resolved' && (record.intent.operationType === 'run.start' || record.intent.operationType === 'document.change' && conflictsWith(operationFootprint(record.intent, next.document), footprint))) dependencies.add(record.intent.id);
for (const record of ledger.getSnapshot().operations.values()) if (record.phase !== 'resolved' && (record.intent.operationType === 'run.start' || ['document.change','run.apply'].includes(record.intent.operationType) && conflictsWith(operationFootprint(record.intent, next.document), footprint))) dependencies.add(record.intent.id);
for (const id of referencedResources(intent.changes, options.registry)) {
const resource = resources.get(id);
if (resource && !resource.stored) dependencies.add(resource.operationId);
@@ -93,7 +95,20 @@ export async function openProjectEngine(options: EngineOptions) {
function complete(request: FrozenRequest, status: OperationStatus): { readonly blocked: true } | void {
const root = editor.getSnapshot(), previous = root.runtime, ledger = createLedger(previous.ledger);
if (request.phase === 'resolved') return;
if (status.outcome !== 'applied') { mutate(draft => draft.result(request.id, status.outcome, status.error)); return; }
if (status.outcome !== 'applied') {
const result=editor.updateRuntime(runtime=>{const ledger=createLedger(runtime.ledger),receipts=new Map(runtime.receipts);ledger.result(request.id,status.outcome,status.error);if(status.outcome==='not-applied'&&['run.start','run.cancel'].includes(request.kind??''))ledger.resolveNoEffect(request.id);if(status.canonical!==undefined)receipts.set(request.id,freeze(status.canonical));return Object.freeze({...runtime,ledger:ledger.getSnapshot(),receipts:readMap(receipts)});},root.rootRevision);
if(result.status==='rejected')throw fault(result.code,result.targets);return;
}
if(request.kind==='run.cancel'){
const receipts=new Map(previous.receipts);if(status.canonical!==undefined)receipts.set(request.id,freeze(status.canonical));
try{
if(!options.runs)throw fault('run_interpreter_unavailable');
const accepted=options.runs.acceptAction(request,status,root),run=previous.runs.get(accepted.runId);if(!run||!accepted.runStatus)throw fault('run_identity_mismatch');
ledger.result(request.id,'applied');const runs=new Map(previous.runs);runs.set(run.id,freeze({...run,controlEpoch:(run.controlEpoch??0)+1,status:accepted.runStatus,hasMore:true,blocked:false,error:undefined,nextPollAt:runClock.now()}));
const result=editor.updateRuntime(()=>Object.freeze({...previous,runs:readMap(runs),receipts:readMap(receipts),ledger:ledger.getSnapshot()}),root.rootRevision);if(result.status==='rejected')throw fault(result.code,result.targets);
queueMicrotask(()=>runTracker.notify([run.id]));return;
}catch(error){ledger.result(request.id,'applied',(error as Error).message,false);const result=editor.updateRuntime(()=>Object.freeze({...previous,receipts:readMap(receipts),ledger:ledger.getSnapshot()}),root.rootRevision);if(result.status==='rejected')throw fault(result.code,result.targets);report(error);return{blocked:true};}
}
if (request.kind === 'asset.upload') {
const receipts = new Map(previous.receipts); if (status.canonical !== undefined) receipts.set(request.id, freeze(status.canonical));
try {
@@ -118,9 +133,11 @@ export async function openProjectEngine(options: EngineOptions) {
const receipts = new Map(previous.receipts), conflicts = new Map(previous.conflicts);
if (status.canonical !== undefined) receipts.set(request.id, freeze(status.canonical));
try {
const action = request.kind==='run.apply' ? options.runs?.acceptAction(request,status,root) : undefined;
if(request.kind==='run.apply'&&!action?.proposal)throw fault('run_interpreter_unavailable');
const admission = request.kind === 'run.start' ? options.runs?.accept(request, status, root) : undefined;
if (request.kind === 'run.start' && !admission) throw fault('run_interpreter_unavailable');
const proposal = admission?.proposal ?? options.persistence.canonical(request, status, root);
const proposal = action?.proposal ?? admission?.proposal ?? options.persistence.canonical(request, status, root);
const writtenCanvases = admission ? [request.ownerCanvasId!] : request.canvasIds;
if (proposal.document.projectId !== root.document.projectId) throw fault('project_mismatch');
for (const canvas of writtenCanvases) {
@@ -141,9 +158,11 @@ export async function openProjectEngine(options: EngineOptions) {
if (existing && (existing.requestId !== request.id || existing.operationId !== admission.run.operationId)) throw fault('run_identity_mismatch');
if (!existing) runs.set(admission.run.id, admission.run);
}
if(action){const run=runs.get(action.runId);if(!run)throw fault('run_identity_mismatch');runs.set(run.id,freeze({...run,controlEpoch:(run.controlEpoch??0)+1,appliedOutputIds:[...new Set([...(run.appliedOutputIds??[]),action.outputId!])],hasMore:true,blocked:false,error:undefined,pendingPage:undefined,nextPollAt:runClock.now()}));}
const runtime: RuntimeState = Object.freeze({ ...previous, runs: readMap(runs), ledger: ledger.getSnapshot(), acknowledged, versions: readMap(proposal.versions), receipts: readMap(receipts), conflicts: readMap(conflicts) });
const result = editor.reconcile(document, runtime, root.rootRevision);
if (result.status === 'rejected') throw fault(result.code, result.targets);
if(action)queueMicrotask(()=>runTracker.notify([action.runId]));
} catch (error) {
// Preserve both the known receipt and the user's working document. No rollback or lane release.
const waiting = createLedger(previous.ledger); waiting.result(request.id, 'applied', (error as Error).message, false);
@@ -158,21 +177,22 @@ export async function openProjectEngine(options: EngineOptions) {
participants: intent => operationCanvases(intent, editor.getSnapshot().document),
compile(intent, id) {
if (intent.operationType === 'document.change') return options.persistence.compile(intent, id, editor.getSnapshot());
if (intent.operationType === 'run.start') { if (!options.runs) throw fault('run_interpreter_unavailable');return options.runs.compile(intent,id,editor.getSnapshot()); }
if (['run.start','run.cancel','run.apply'].includes(intent.operationType)) { if (!options.runs) throw fault('run_interpreter_unavailable');return options.runs.compile(intent,id,editor.getSnapshot()); }
if(intent.operationType!=='asset.upload')throw fault('operation_interpreter_unavailable');
const resource = editor.getSnapshot().runtime.resources.get(intent.resourceId);
if (!options.assets || !resource?.digest) throw fault('asset_not_ready');
return { kind: 'asset.upload', protocolVersion: 2, canvasIds: [], expectedVersions: {}, payload: { protocol_version: 2, request_id: id, resource_id: resource.id, operation_id: resource.operationId, sha256: resource.digest, byte_length: resource.byteLength, name: resource.name, mime_type: resource.mimeType } };
},
async submit(request, signal) {
if (request.kind === 'run.start') return options.runs ? network.run(signal, () => options.runs!.submit(request,signal)) : { outcome: 'unresolved', blocked: true, error: 'run_interpreter_unavailable' };
if (['run.start','run.cancel','run.apply'].includes(request.kind??'')) return options.runs ? network.run(signal, () => options.runs!.submit(request,signal)) : { outcome: 'unresolved', blocked: true, error: 'run_interpreter_unavailable' };
if (request.kind !== 'asset.upload') return network.run(signal, () => options.persistence.submit(request, signal));
if (!options.assets) return { outcome: 'unresolved', blocked: true, error: 'asset_interpreter_unavailable' };
const payload = JSON.parse(request.body), bytes = await options.storage.readFile(leader, payload.resource_id);
if (!bytes || bytes.size !== payload.byte_length || await digest(bytes) !== payload.sha256) return { outcome: 'unresolved', blocked: true, error: 'resource_bytes_unavailable' };
return network.run(signal, () => options.assets!.upload(request, bytes, signal));
},
query: (request, signal) => network.run(signal, () => request.kind === 'run.start' ? options.runs?.query(request,signal) ?? Promise.resolve({ outcome: 'unresolved', blocked: true, error: 'run_interpreter_unavailable' }) : request.kind === 'asset.upload' ? options.assets?.query(request, signal) ?? Promise.resolve({ outcome: 'unresolved', blocked: true, error: 'asset_interpreter_unavailable' }) : options.persistence.query(request, signal)),
supportsRequest: request => request.kind === 'run.start' ? !!options.runs && request.protocolVersion === 2 : request.kind === 'asset.upload' ? !!options.assets && request.protocolVersion === 2 : options.persistence.supportsRequest?.(request) ?? request.protocolVersion === 1, complete, onError: report, clock: options.clock,
query: (request, signal) => network.run(signal, () => ['run.start','run.cancel','run.apply'].includes(request.kind??'') ? options.runs?.query(request,signal) ?? Promise.resolve({ outcome: 'unresolved', blocked: true, error: 'run_interpreter_unavailable' }) : request.kind === 'asset.upload' ? options.assets?.query(request, signal) ?? Promise.resolve({ outcome: 'unresolved', blocked: true, error: 'asset_interpreter_unavailable' }) : options.persistence.query(request, signal)),
supportsRequest: request => ['run.start','run.cancel','run.apply'].includes(request.kind??'') ? !!options.runs && request.protocolVersion === 2 : request.kind === 'asset.upload' ? !!options.assets && request.protocolVersion === 2 : options.persistence.supportsRequest?.(request) ?? request.protocolVersion === 1, complete, onError: report, clock: options.clock,
});
function updateRun(id: Id, reduce: (run: RunRecord) => RunRecord) {
const result = editor.updateRuntime(previous => { const current = previous.runs.get(id);if(!current)throw fault('run_missing');const runs = new Map(previous.runs);runs.set(id,freeze(reduce(current)));return Object.freeze({...previous,runs:readMap(runs)}); });
@@ -180,7 +200,9 @@ export async function openProjectEngine(options: EngineOptions) {
}
function acceptRunPage(run: RunRecord, page: RunPage) {
const root=editor.getSnapshot(),previous=root.runtime,current=previous.runs.get(run.id);
if(!current || current.cursor!==run.cursor || current.requestId!==run.requestId)throw fault('run_identity_mismatch');
if(!current || current.requestId!==run.requestId)throw fault('run_identity_mismatch');
if((current.controlEpoch??0)!==(run.controlEpoch??0))return;
if(current.cursor!==run.cursor)throw fault('run_identity_mismatch');
const outputEvents=new Map(current.outputEvents.map(event=>[event.step_id,event]));
for(const event of page.events)if(event.type==='step.output')outputEvents.set(event.step_id,event);
let acknowledged=previous.acknowledged,versions=previous.versions;
@@ -198,7 +220,8 @@ export async function openProjectEngine(options: EngineOptions) {
const proposal=options.runs.canonical(event,snapshot);acknowledged=proposal.document;versions=proposal.versions;
}
const document=replayPending(acknowledged,previous.ledger,options.registry),runs=new Map(previous.runs);
runs.set(run.id,freeze({...current,cursor:page.next,status:page.status,hasMore:page.hasMore,outputs:page.outputs,outputEvents:[...outputEvents.values()],steps:page.steps,error:page.error,blocked:false,pendingPage:undefined}));
const appliedOutputIds=[...new Set([...(current.appliedOutputIds??[]),...page.events.filter(event=>event.type==='step.applied'&&typeof event.output_id==='string').map(event=>event.output_id as Id)])];
runs.set(run.id,freeze({...current,appliedOutputIds,cursor:page.next,status:page.status,hasMore:page.hasMore,outputs:page.outputs,outputEvents:[...outputEvents.values()],steps:page.steps,error:page.error,blocked:false,pendingPage:undefined}));
conflicts.delete(`run:${run.id}`);
const result=editor.reconcile(document,Object.freeze({...previous,acknowledged,versions:readMap(versions),runs:readMap(runs),conflicts:readMap(conflicts)}),root.rootRevision);
if(result.status==='rejected')throw fault(result.code,result.targets);
@@ -248,6 +271,12 @@ export async function openProjectEngine(options: EngineOptions) {
digestTasks.set(resource.id, task);
}
}
function queueRunAction(operationType:'run.cancel'|'run.apply',action:RunCancelInput|RunApplyInput):Result {
const root=editor.getSnapshot();let sequence=1;for(const record of root.runtime.ledger.operations.values())sequence=Math.max(sequence,record.intent.sequence+1);
const id=createId(),intent={id,operationType,action,operationSchemaVersion:1,sequence,origin:'user',canvasIds:operationType==='run.apply'?[(action as RunApplyInput).targetCanvasId]:[],changes:[],dependencies:operationType==='run.cancel'?[]:[...root.runtime.ledger.operations.values()].filter(record=>record.phase!=='resolved').map(record=>record.intent.id)} as OperationIntent;
const result=editor.updateRuntime(previous=>{const ledger=createLedger(previous.ledger);ledger.add(freeze(intent),{});return Object.freeze({...previous,ledger:ledger.getSnapshot()});},root.rootRevision);
if(result.status==='rejected')return result;queueMicrotask(()=>{if(!closing&&!closed)scheduler.notify([id]);});return{status:'applied',revision:root.revision,operationIds:[id]};
}
scheduler.start(); prepareDigests(); runTracker.notify([...initial.runs.keys()]);
return Object.freeze({
editor, sessions,
@@ -256,6 +285,31 @@ export async function openProjectEngine(options: EngineOptions) {
if(!editor.getSnapshot().document.canvases.has(boardId))return Promise.reject(fault('canvas_missing'));
return network.run(signal,()=>options.runs!.catalog(boardId,signal));
},
cancelRun(runId:Id):Result {
try{
if(closing||closed)throw fault('engine_closing');if(!options.runs)throw fault('run_interpreter_unavailable');
const root=editor.getSnapshot(),run=root.runtime.runs.get(runId);if(!run)throw fault('run_missing');
const allowed=options.policy?.({type:'element.data',target:run.target,patch:{}},root.document);if(allowed&&!allowed.enabled)throw fault(allowed.code,allowed.targets);
if([...root.runtime.ledger.operations.values()].some(record=>record.phase!=='resolved'&&record.intent.operationType==='run.cancel'&&record.intent.action.runId===runId))throw fault('run_action_pending');
return queueRunAction('run.cancel',{runId,boardId:run.boardId});
}catch(error){return reject(error);}
},
applyRunOutput(runId:Id,target:RunApplyInput['target'],outputId:Id):Result {
try{
if(closing||closed)throw fault('engine_closing');if(!options.runs)throw fault('run_interpreter_unavailable');if(applying)throw fault('reentrant_dispatch');
const root=editor.getSnapshot(),run=root.runtime.runs.get(runId),element=root.document.elements.get(target.id);
if(!run||run.blocked||!runContainsTarget(run,target)||!element||element.incarnation!==target.incarnation)throw fault('target_changed');
if((run.appliedOutputIds??[]).includes(outputId))throw fault('output_already_applied');
if([...root.runtime.ledger.operations.values()].some(record=>record.phase!=='resolved'&&(record.intent.operationType==='run.apply'||record.intent.operationType==='run.cancel')&&record.intent.action.runId===runId))throw fault('run_action_pending');
const event=run.outputEvents.find(event=>event.node_id===target.id&&event.output&&typeof event.output==='object'&&!Array.isArray(event.output)&&event.output.output_id===outputId),output=event?.output as {[key:string]:Json}|undefined;
if(!output||run.status!=='conflict')throw fault('output_not_applicable');
const patch:Data|undefined=output.kind==='text'&&typeof output.text==='string'?{text:output.text}:output.kind==='image'&&typeof output.src==='string'&&typeof output.assetId==='string'?{src:output.src,assetId:output.assetId}:undefined;if(!patch)throw fault('invalid_run_output');
const action:RunApplyInput={runId,boardId:run.boardId,targetCanvasId:element.ownerCanvasId,target,outputId};applying=action;
const result=editor.dispatch({type:'element.data',target,patch},{label:'Apply generated result'});
if(result.status==='rejected'||result.operationIds.length)return result;
return queueRunAction('run.apply',action);
}catch(error){return reject(error);}finally{applying=undefined;}
},
startRun(input: RunInput): Result {
try {
if(closing||closed)throw fault('engine_closing');