Persist AI admissions and reconcile canvas runs across navigation
This commit is contained in:
@@ -0,0 +1,54 @@
|
||||
"use client";
|
||||
|
||||
import { useEffect, useState } from "react";
|
||||
import { ArrowUp, Sparkles, X } from "lucide-react";
|
||||
import type { ElementRecord, Json, Result } from "@/lib/canvas-core/types";
|
||||
import { runContainsTarget } from "@/lib/canvas-runtime/runs";
|
||||
import type { RunModel } from "@/lib/canvas-runtime/runs";
|
||||
import type { ProjectEngine } from "@/lib/canvas-runtime/workspace";
|
||||
import { errorText } from "@/lib/api";
|
||||
import { managedImageSource } from "@/lib/canvas-v2";
|
||||
import styles from "./canvas.module.css";
|
||||
|
||||
const object = (value: Json | undefined) => value && typeof value === 'object' && !Array.isArray(value) ? value : {};
|
||||
export function HeadlessAIPanel({ engine, element, close }: { engine: ProjectEngine; element: ElementRecord; close(): void }) {
|
||||
const saved = object(element.data.ai), root = engine.getSnapshot();
|
||||
const [models, setModels] = useState<readonly RunModel[]>([]), [loading, setLoading] = useState(true), [error, setError] = useState('');
|
||||
const [model, setModel] = useState(typeof saved.modelId === 'string' ? saved.modelId : '');
|
||||
const [prompt, setPrompt] = useState(typeof saved.prompt === 'string' ? saved.prompt : '');
|
||||
const [propagate, setPropagate] = useState(saved.propagate === true);
|
||||
const image = ['image','plain-image'].includes(element.componentId);
|
||||
const relevant = [...root.runtime.runs.values()].filter(run => runContainsTarget(run,element));
|
||||
const run = relevant.at(-1);
|
||||
const pending = [...root.runtime.ledger.operations.values()].filter(record => record.intent.operationType === 'run.start' && record.intent.run.target.id === element.id && record.intent.run.target.incarnation === element.incarnation && record.phase !== 'resolved');
|
||||
const busy = pending.length > 0 || !!run && (run.hasMore || !!run.pendingPage || ['queued','running'].includes(run.status));
|
||||
const outputEvent = run?.outputEvents.filter(event => event.node_id === element.id).at(-1), output = object(outputEvent?.output);
|
||||
useEffect(() => {
|
||||
const controller = new AbortController();setLoading(true);setError('');
|
||||
engine.models(element.ownerCanvasId,controller.signal).then(catalog => {
|
||||
if(controller.signal.aborted)return;
|
||||
const compatible = catalog.models.filter(model => model.output_modalities.includes(image ? 'image' : 'text'));
|
||||
setModels(compatible);setModel(current => compatible.some(model => model.id === current) ? current : compatible.find(model => model.available)?.id ?? '');
|
||||
if(catalog.unavailableReason)setError(errorText(new Error(catalog.unavailableReason)));
|
||||
}).catch(error => { if(!controller.signal.aborted)setError(errorText(error)); }).finally(() => { if(!controller.signal.aborted)setLoading(false); });
|
||||
return () => controller.abort();
|
||||
},[engine,element.ownerCanvasId,image]);
|
||||
function check(result: Result) { if(result.status === 'rejected') { setError(errorText(new Error(result.code)));return false; }return true; }
|
||||
function save() {
|
||||
if(!model || busy)return true;
|
||||
return check(engine.editor.dispatch({type:'element.data',target:{id:element.id,incarnation:element.incarnation},patch:{ai:{enabled:true,modelId:model,prompt,propagate,parameters:{}}}},{label:'Configure generation'}));
|
||||
}
|
||||
const message = pending.length ? pending.some(record => record.phase === 'blocked') ? '生成受理需要处理' : '正在等待同步并受理生成…' : run?.blocked ? '生成结果已保留，当前编辑需要协调' : run?.error ? '任务跟踪需要处理，可重试' : run?.hasMore ? '正在接收生成结果…' : run ? ({queued:'已受理，等待生成',running:'正在生成…',succeeded:'已更新当前元素',failed:'生成未完成，原内容已保留',conflict:'内容已改变，生成结果已保留',cancelled:'已取消生成',partially_succeeded:'部分步骤已完成'}[run.status]) : '结果将显示在当前元素';
|
||||
return <section className={`${styles.aiPanel} nodrag nopan nowheel`} aria-label="元素 AI 工具" style={{position:'absolute',right:24,bottom:90,zIndex:110,maxWidth:'calc(100% - 48px)',maxHeight:'70%',overflow:'auto'}} onPointerDown={event=>event.stopPropagation()} onKeyDown={event=>event.stopPropagation()}>
|
||||
<header><Sparkles size={14}/><strong>{image ? '图像创作' : '文本创作'}</strong><span>{element.data.title as string}</span><button aria-label="关闭 AI 工具" onClick={close}><X size={14}/></button></header>
|
||||
<textarea aria-label="AI 指令" placeholder="希望如何处理当前内容？" maxLength={16000} disabled={busy} value={prompt} onChange={event=>setPrompt(event.target.value)} onBlur={save}/>
|
||||
<footer><select aria-label="AI 模型" value={model} disabled={loading || busy} onChange={event=>setModel(event.target.value)}>{!model && <option value="">{loading ? '正在读取模型…' : '暂无可用模型'}</option>}{models.map(model=><option key={model.id} value={model.id} disabled={!model.available}>{model.label || model.display_name || model.id}</option>)}</select>
|
||||
<label><input type="checkbox" aria-label="继续向下传播" checked={propagate} disabled={busy} onChange={event=>setPropagate(event.target.checked)}/>继续向下传播</label>
|
||||
<button aria-label="执行 AI" title="在当前元素生成结果" disabled={busy || loading || !prompt.trim() || !models.some(item=>item.id===model && item.available)} onClick={()=>{setError('');if(save())check(engine.startRun({target:{id:element.id,incarnation:element.incarnation},boardId:element.ownerCanvasId,modelId:model,prompt,parameters:{},propagate}));}}><ArrowUp size={18}/></button>
|
||||
</footer>
|
||||
<p role="status">{message}</p>
|
||||
{(run?.blocked || run?.status==='conflict') && <div className={styles.aiConflict}>{typeof output.text==='string' && <pre>{output.text}</pre>}{typeof output.src==='string' && <img src={managedImageSource(output.src)} alt="保留的生成结果"/>}</div>}
|
||||
{(run?.blocked || run?.error || pending.some(record=>record.phase==='blocked')) && <button onClick={()=>void engine.retry().catch(error=>setError(errorText(error)))}>重试任务跟踪</button>}
|
||||
{error && <p role="alert">{error}</p>}
|
||||
</section>;
|
||||
}
|
||||
@@ -2,7 +2,7 @@
|
||||
|
||||
import { useCallback, useEffect, useMemo, useRef, useState, useSyncExternalStore } from "react";
|
||||
import { Background, BackgroundVariant, ConnectionMode, ReactFlow, ReactFlowProvider, useReactFlow, type Connection, type NodeProps } from "@xyflow/react";
|
||||
import { Download, Frame, Hand, ImagePlus, Maximize, Minus, MousePointer2, Plus, Redo2, StickyNote, Type, Undo2 } from "lucide-react";
|
||||
import { Download, Frame, Hand, ImagePlus, Maximize, Minus, MousePointer2, Plus, Redo2, Sparkles, StickyNote, Type, Undo2 } from "lucide-react";
|
||||
import type { Board } from "@/lib/workspace";
|
||||
import type { Data, ElementRecord, Geometry, Id, Json, Result } from "@/lib/canvas-core/types";
|
||||
import { equal } from "@/lib/canvas-core/values";
|
||||
@@ -15,6 +15,7 @@ import toolbar from "@/components/ui/toolbar.module.css";
|
||||
import { guardEditingLease } from "./editing-lease";
|
||||
import { NodeEditing } from "./transformable-node";
|
||||
import { EdgeEditing } from "./edge-editing";
|
||||
import { HeadlessAIPanel } from "./headless-ai-panel";
|
||||
import { CanvasEdge } from "./canvas-edge";
|
||||
import { MaterialNode } from "./material-node";
|
||||
import { PlainTextNode } from "./plain-text-node";
|
||||
@@ -36,6 +37,7 @@ function CanvasView({ engine, board, viewId }: { engine: ProjectEngine; board: B
|
||||
const liveView = sessions.views.get(viewId), lastView = useRef(liveView);
|
||||
if (liveView) lastView.current = liveView;
|
||||
const view = liveView ?? lastView.current!, flow = useReactFlow<ProjectionNode>();
|
||||
const [showAI, setShowAI] = useState(false);
|
||||
const [error, setError] = useState(""), [sources, setSources] = useState(new Map<string, string>());
|
||||
const pointerStart = useRef<{ x: number; y: number } | null>(null), drag = useRef<{ token: number; start: { x: number; y: number }; positions: Map<Id, { x: number; y: number }> } | null>(null);
|
||||
const measurements = useRef(new Map<Id, { width: number; height: number; measured: { width: number; height: number } }>());
|
||||
@@ -112,6 +114,8 @@ function CanvasView({ engine, board, viewId }: { engine: ProjectEngine; board: B
|
||||
return () => { window.removeEventListener('blur', cancel); window.removeEventListener('keydown', escape, true); };
|
||||
}, [engine, viewId]);
|
||||
const modes = view.modes;
|
||||
const selected = view.selection.size === 1 ? root.document.elements.get([...view.selection][0]) : undefined;
|
||||
const aiElement = selected?.structure === 'node' && engine.editor.registry.resolve(selected) === 'known' && ['text','plain-text','script','image','plain-image'].includes(selected.componentId) ? selected : undefined;
|
||||
if (!liveView) return <p role="status">画布已关闭</p>;
|
||||
return <div className={`${styles.editor} ${view.tool === 'pan' ? styles.handMode : ''}`} data-headless-canvas tabIndex={-1}
|
||||
onMouseDownCapture={event => { if (event.button === 0 && (event.target as HTMLElement).closest('.react-flow__node')) pointerStart.current = flow.screenToFlowPosition({ x: event.clientX, y: event.clientY }); }}
|
||||
@@ -193,9 +197,11 @@ function CanvasView({ engine, board, viewId }: { engine: ProjectEngine; board: B
|
||||
</div><p role="status" style={{ fontSize: 11 }}>{blocked ? '有修改需要处理' : pending.length ? '正在同步…' : '已同步'}{blocked && <button onClick={() => void engine.retry().catch(e => setError(errorText(e)))}>重试</button>}</p></div>
|
||||
<div className={`${toolbar.surface} ${toolbar.relaxed} ${styles.toolbar} ${styles.bottomToolbar}`} role="toolbar" aria-label="画布创作工具">
|
||||
<IconButton icon={MousePointer2} label="选择工具（V）" aria-pressed={view.tool === 'select'} onClick={() => engine.sessions.tool(viewId, 'select')} /><IconButton icon={Hand} label="抓手工具（H）" aria-pressed={view.tool === 'pan'} onClick={() => engine.sessions.tool(viewId, 'pan')} />
|
||||
<IconButton icon={Sparkles} label="AI 创作" disabled={!aiElement} aria-pressed={showAI} onClick={() => setShowAI(value => !value)} />
|
||||
<IconButton icon={Type} label="添加纯文本" onClick={() => add('plain-text')} /><IconButton icon={StickyNote} label="添加文本卡片" onClick={() => add('text')} /><IconButton icon={ImagePlus} label="添加图片" onClick={() => imageInput.current?.click()} /><IconButton icon={Frame} label="添加框选区" onClick={() => add('frame')} />
|
||||
</div>
|
||||
<input ref={imageInput} type="file" accept="image/png,image/jpeg,image/gif,image/webp" multiple hidden onChange={event => { const files = [...event.target.files ?? []]; event.target.value = ''; try { report(engine.stageFiles(files.map(file => ({ name: file.name, bytes: file })), (refs, tx) => { for (const ref of refs) add('image', ref, tx); })); } catch (e) { setError(errorText(e)); } }} />
|
||||
{showAI && aiElement && <HeadlessAIPanel key={`${aiElement.id}:${aiElement.incarnation}:${aiElement.ownerCanvasId}`} engine={engine} element={aiElement} close={() => setShowAI(false)} />}
|
||||
{error && <div className={styles.error} role="alert">{error} <button onClick={() => setError('')}>关闭</button></div>}
|
||||
</div>;
|
||||
}
|
||||
|
||||
@@ -7,7 +7,7 @@ import { createRegistry, builtinComponents } from "@/lib/canvas-core/index";
|
||||
import { createRuntimeHost, type RuntimeHost } from "@/lib/canvas-runtime/host";
|
||||
import { createWorkspaceRuntime } from "@/lib/canvas-runtime/workspace";
|
||||
import { createBrowserStorage } from "@/lib/canvas-runtime/storage";
|
||||
import { createFoundryAssets, createFoundryPersistence, createFoundryTransport } from "@/lib/canvas-service/index";
|
||||
import { createFoundryAssets, createFoundryPersistence, createFoundryTransport, createFoundryRuns } from "@/lib/canvas-service/index";
|
||||
|
||||
const Context = createContext<RuntimeHost | null>(null);
|
||||
export function CanvasRuntimeHost({ children }: { children: ReactNode }) {
|
||||
@@ -18,7 +18,7 @@ export function CanvasRuntimeHost({ children }: { children: ReactNode }) {
|
||||
return createWorkspaceRuntime({ accountId, workspaceId, authEpoch, isCurrentIdentity, registry, storage: createBrowserStorage(),
|
||||
adapters(rootId, current) {
|
||||
const transport = createFoundryTransport({ accountId, isCurrentIdentity: current });
|
||||
return { persistence: createFoundryPersistence({ workspaceId, rootId, registry, transport }), assets: createFoundryAssets({ workspaceId, transport, project: process.env.NEXT_PUBLIC_FOUNDRY_PROJECT || "endoleap", origin: process.env.NEXT_PUBLIC_FOUNDRY_API_ORIGIN }) };
|
||||
return { runs: createFoundryRuns({ workspaceId, rootId, registry, transport }), persistence: createFoundryPersistence({ workspaceId, rootId, registry, transport }), assets: createFoundryAssets({ workspaceId, transport, project: process.env.NEXT_PUBLIC_FOUNDRY_PROJECT || "endoleap", origin: process.env.NEXT_PUBLIC_FOUNDRY_API_ORIGIN }) };
|
||||
},
|
||||
onError: error => notify({ id: `canvas-runtime:${accountId}:${workspaceId}`, title: "画布同步需要处理", description: errorText(error), level: "error" }),
|
||||
});
|
||||
|
||||
@@ -35,7 +35,8 @@ export type Result = Applied | Rejected;
|
||||
export type HistoryEntry = { readonly id: Id; readonly label: string; readonly changes: readonly Change[]; readonly operationIds: readonly Id[]; readonly groupId?: Id };
|
||||
export type HistoryState = { readonly undo: readonly Id[]; readonly redo: readonly Id[]; readonly entries: ReadonlyMap<Id, HistoryEntry> };
|
||||
type IntentBase = { readonly id: Id; readonly operationSchemaVersion: 1; readonly sequence: number; readonly origin: Origin; readonly canvasIds: readonly Id[]; readonly changes: readonly Change[]; readonly dependencies: readonly Id[] };
|
||||
export type OperationIntent = IntentBase & ({ readonly operationType: 'document.change' } | { readonly operationType: 'asset.upload'; readonly resourceId: Id });
|
||||
export type RunInput = { readonly target: Target; readonly boardId: Id; readonly modelId: string; readonly prompt: string; readonly parameters: Data; readonly propagate: boolean };
|
||||
export type OperationIntent = IntentBase & ({ readonly operationType: 'document.change' } | { readonly operationType: 'asset.upload'; readonly resourceId: Id } | { readonly operationType: 'run.start'; readonly run: RunInput });
|
||||
export type Commit = { readonly revision: number; readonly origin: Origin; readonly changes: readonly Change[]; readonly operationIds: readonly Id[] };
|
||||
export type CoreSnapshot = { readonly revision: number; readonly document: Document; readonly history: HistoryState; readonly operations: ReadonlyMap<Id, OperationIntent> };
|
||||
/** One publication boundary. Runtime metadata changes do not dirty the document revision. */
|
||||
|
||||
@@ -1,10 +1,12 @@
|
||||
import type { Document, Id, Json, OperationIntent, Snapshot } from '../canvas-core/types.ts';
|
||||
import type { LedgerSnapshot, FrozenRequest } from './ledger.ts';
|
||||
import type { CompiledRequest, OperationStatus } from './scheduler.ts';
|
||||
import type { RunRecord } from './runs.ts';
|
||||
import type { ResourceRecord } from './resources.ts';
|
||||
|
||||
export type RuntimeState = {
|
||||
readonly ledger: LedgerSnapshot;
|
||||
readonly runs: ReadonlyMap<Id, RunRecord>;
|
||||
readonly acknowledged: Document;
|
||||
readonly versions: ReadonlyMap<Id, number>;
|
||||
readonly receipts: ReadonlyMap<Id, Json>;
|
||||
|
||||
@@ -1,9 +1,11 @@
|
||||
import type { Document, Id, Registry, Result } from '../canvas-core/types.ts';
|
||||
import type { Document, Id, Registry, Result, RunInput } 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';
|
||||
import { equal, fault, freeze, readMap, reject } from '../canvas-core/values.ts';
|
||||
import type { EngineSnapshot, PersistencePort, RuntimeState } from './contracts.ts';
|
||||
import { createRunTracker, validateRunInput, runContainsTarget, type RunPort, type RunRecord, type RunPage } from './runs.ts';
|
||||
import { createNetworkGate } from './network.ts';
|
||||
import { createLedger, type FrozenRequest } from './ledger.ts';
|
||||
import { operationCanvases, replayPending } from './reconcile.ts';
|
||||
import { decodeRecovery, encodeRecovery } from './recovery.ts';
|
||||
@@ -13,7 +15,7 @@ import type { StoragePort, StorageScope } from './storage.ts';
|
||||
|
||||
export type EngineOptions = {
|
||||
scope: StorageScope; registry: Registry; document: Document; versions: ReadonlyMap<Id, number>;
|
||||
storage: StoragePort; persistence: PersistencePort; assets?: AssetPort; createId?: () => Id;
|
||||
storage: StoragePort; persistence: PersistencePort; assets?: AssetPort; runs?: RunPort; createId?: () => Id;
|
||||
historyLimit?: number; policy?: EditorOptions['policy']; traversalPolicy?: EditorOptions['traversalPolicy']; clock?: Clock;
|
||||
onError?: (error: unknown) => void;
|
||||
};
|
||||
@@ -29,10 +31,12 @@ export async function openProjectEngine(options: EngineOptions) {
|
||||
let closing = false, closed = false, durableRootRevision = image?.rootRevision ?? -1;
|
||||
const report = (error: unknown) => { try { options.onError?.(error); } catch { /* Error reporting cannot alter state. */ } };
|
||||
const createId = options.createId ?? (() => crypto.randomUUID());
|
||||
const network = createNetworkGate(2);
|
||||
const runClock: Clock = options.clock ?? { now: () => Date.now(), setTimer: (fn, delay) => setTimeout(fn, delay), clearTimer: handle => clearTimeout(handle as ReturnType<typeof setTimeout>) };
|
||||
const displays = new Set<() => void>();
|
||||
const fileWrites = new Map<Id, Blob>(), digestTasks = new Map<Id, Promise<void>>();
|
||||
let staged: readonly ResourceRecord[] = [];
|
||||
const initial: RuntimeState = recovered?.runtime ?? Object.freeze({ 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({ 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 }; },
|
||||
@@ -51,7 +55,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 === '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' || record.intent.operationType === 'document.change' && 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);
|
||||
@@ -114,21 +118,30 @@ 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 proposal = options.persistence.canonical(request, status, root);
|
||||
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 writtenCanvases = admission ? [request.ownerCanvasId!] : request.canvasIds;
|
||||
if (proposal.document.projectId !== root.document.projectId) throw fault('project_mismatch');
|
||||
for (const canvas of request.canvasIds) {
|
||||
for (const canvas of writtenCanvases) {
|
||||
const version = proposal.versions.get(canvas);
|
||||
if (!version || version <= request.expectedVersions[canvas] || version < (previous.versions.get(canvas) ?? 0)) throw fault('stale_canonical_version', [canvas]);
|
||||
}
|
||||
// A codec cannot roll back an unrelated canvas while reconciling this request.
|
||||
for (const [canvas, version] of previous.versions) if (!request.canvasIds.includes(canvas)) {
|
||||
for (const [canvas, version] of previous.versions) if (!writtenCanvases.includes(canvas)) {
|
||||
if (proposal.versions.get(canvas) !== version || !equal(proposal.document.canvases.get(canvas), previous.acknowledged.canvases.get(canvas)) || !equal([...proposal.document.elements.values()].filter(element => element.ownerCanvasId === canvas), [...previous.acknowledged.elements.values()].filter(element => element.ownerCanvasId === canvas))) throw fault('canonical_scope_mismatch', [canvas]);
|
||||
}
|
||||
ledger.result(request.id, 'applied');
|
||||
const acknowledged = createDocument(proposal.document.projectId, [...proposal.document.canvases.values()], [...proposal.document.elements.values()], proposal.document.tombstones);
|
||||
const document = replayPending(acknowledged, ledger.getSnapshot(), options.registry);
|
||||
conflicts.delete(request.id);
|
||||
const runtime: RuntimeState = Object.freeze({ ...previous, ledger: ledger.getSnapshot(), acknowledged, versions: readMap(proposal.versions), receipts: readMap(receipts), conflicts: readMap(conflicts) });
|
||||
const runs = new Map(previous.runs);
|
||||
if (admission) {
|
||||
const existing = runs.get(admission.run.id);
|
||||
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);
|
||||
}
|
||||
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);
|
||||
} catch (error) {
|
||||
@@ -145,23 +158,66 @@ 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()); }
|
||||
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 !== 'asset.upload') return options.persistence.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 (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 options.assets.upload(request, bytes, signal);
|
||||
return network.run(signal, () => options.assets!.upload(request, bytes, signal));
|
||||
},
|
||||
query: (request, signal) => 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 === '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, () => 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,
|
||||
});
|
||||
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)}); });
|
||||
if(result.status==='rejected')throw fault(result.code,result.targets);
|
||||
}
|
||||
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');
|
||||
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;
|
||||
const conflicts=new Map(previous.conflicts);
|
||||
try {
|
||||
for(const event of page.events) {
|
||||
if(event.seq<=current.cursor || event.type!=='step.applied')continue;
|
||||
const canvas=event.board_id as Id,version=event.canvas_version as number;
|
||||
if(!Number.isSafeInteger(version)||version<1)throw fault('invalid_run_response');
|
||||
// A later document receipt already includes this event; never rewind its participant.
|
||||
if(version<=(versions.get(canvas)??0))continue;
|
||||
if([...previous.ledger.requests.values()].some(request=>request.phase!=='resolved'&&request.canvasIds.includes(canvas)))throw fault('run_waiting_for_receipts');
|
||||
if(!options.runs)throw fault('run_interpreter_unavailable');
|
||||
const snapshot={...root,runtime:{...previous,acknowledged,versions}};
|
||||
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}));
|
||||
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);
|
||||
} catch(error) {
|
||||
const code=(error as Error).message;
|
||||
// Save the page as evidence without overwriting local content or advancing the cursor.
|
||||
const result=editor.updateRuntime(runtime=>{const runs=new Map(runtime.runs);runs.set(run.id,freeze({...current,outputs:page.outputs,outputEvents:[...outputEvents.values()],steps:page.steps,pendingPage:page,error:code,blocked:code!=='run_waiting_for_receipts'}));if(code!=='run_waiting_for_receipts')conflicts.set(`run:${run.id}`,code);return Object.freeze({...runtime,runs:readMap(runs),conflicts:readMap(conflicts)});},root.rootRevision);
|
||||
if(result.status==='rejected')throw fault(result.code,result.targets);
|
||||
if(code!=='run_waiting_for_receipts')report(error);
|
||||
}
|
||||
}
|
||||
const runTracker=createRunTracker({clock:runClock,getRuns:()=>editor.getSnapshot().runtime.runs,persist,update:updateRun,accept:acceptRunPage,
|
||||
poll:(run,signal)=>options.runs ? network.run(signal,()=>options.runs!.poll(run,signal)) : Promise.reject(fault('run_interpreter_unavailable')),onError:report});
|
||||
const trackedRuns=new Set(initial.runs.keys());
|
||||
const sessions = createSessions(editor);
|
||||
const unsubscribe = editor.subscribe((commit, snapshot) => {
|
||||
pending.set(snapshot.rootRevision, snapshot);
|
||||
const addedRuns=[...snapshot.runtime.runs.keys()].filter(id=>!trackedRuns.has(id));for(const id of addedRuns)trackedRuns.add(id);
|
||||
if(addedRuns.length)queueMicrotask(()=>runTracker.notify(addedRuns));
|
||||
if (!scheduled) { scheduled = true; queueMicrotask(() => { scheduled = false; void persist().catch(report); }); }
|
||||
// Intents already exist in this root; notification only schedules their IDs.
|
||||
const operationIds = [...commit.operationIds, ...staged.map(resource => resource.operationId)];
|
||||
@@ -192,9 +248,33 @@ export async function openProjectEngine(options: EngineOptions) {
|
||||
digestTasks.set(resource.id, task);
|
||||
}
|
||||
}
|
||||
scheduler.start(); prepareDigests();
|
||||
scheduler.start(); prepareDigests(); runTracker.notify([...initial.runs.keys()]);
|
||||
return Object.freeze({
|
||||
editor, sessions,
|
||||
models(boardId: Id, signal: AbortSignal) {
|
||||
if(closing||closed||!options.runs)return Promise.reject(fault('run_interpreter_unavailable'));
|
||||
if(!editor.getSnapshot().document.canvases.has(boardId))return Promise.reject(fault('canvas_missing'));
|
||||
return network.run(signal,()=>options.runs!.catalog(boardId,signal));
|
||||
},
|
||||
startRun(input: RunInput): Result {
|
||||
try {
|
||||
if(closing||closed)throw fault('engine_closing');
|
||||
if(!options.runs)throw fault('run_interpreter_unavailable');
|
||||
validateRunInput(input);
|
||||
const root=editor.getSnapshot(),element=root.document.elements.get(input.target.id);
|
||||
if(!element||element.incarnation!==input.target.incarnation||element.ownerCanvasId!==input.boardId||element.structure!=='node'||options.registry.resolve(element)!=='known'||!['text','plain-text','script','image','plain-image'].includes(element.componentId))throw fault('target_changed');
|
||||
const allowed=editor.can({type:'element.data',target:input.target,patch:{ai:{enabled:true,modelId:input.modelId,prompt:input.prompt,parameters:input.parameters,propagate:input.propagate}}});
|
||||
if(!allowed.enabled)throw fault(allowed.code,allowed.targets);
|
||||
if([...root.runtime.ledger.operations.values()].some(record=>record.intent.operationType==='run.start'&&record.phase!=='resolved'&&record.intent.run.target.id===input.target.id))throw fault('node_busy');
|
||||
if([...root.runtime.runs.values()].some(run=>runContainsTarget(run,input.target)&&(run.hasMore||run.pendingPage||['queued','running'].includes(run.status))))throw fault('node_busy');
|
||||
const id=createId(),dependencies=[...root.runtime.ledger.operations.values()].filter(record=>record.phase!=='resolved').map(record=>record.intent.id);
|
||||
let sequence=1;for(const record of root.runtime.ledger.operations.values())sequence=Math.max(sequence,record.intent.sequence+1);
|
||||
const result=editor.updateRuntime(previous=>{const ledger=createLedger(previous.ledger);ledger.add(freeze({id,operationType:'run.start',operationSchemaVersion:1,sequence,origin:'user',canvasIds:[...root.document.canvases.keys()].sort(),changes:[],dependencies,run:JSON.parse(JSON.stringify(input))}),{});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]};
|
||||
}catch(error){return reject(error);}
|
||||
},
|
||||
refresh(document: Document, versions: ReadonlyMap<Id, number>): Result {
|
||||
const root = editor.getSnapshot(), previous = root.runtime;
|
||||
try {
|
||||
@@ -243,11 +323,11 @@ export async function openProjectEngine(options: EngineOptions) {
|
||||
},
|
||||
getSnapshot: editor.getSnapshot, persist,
|
||||
getDurability: () => Object.freeze({ rootRevision: editor.getSnapshot().rootRevision, durableRootRevision }),
|
||||
retry: () => scheduler.resume(),
|
||||
async settle() { await Promise.resolve(); while (digestTasks.size) await Promise.all([...digestTasks.values()]); await scheduler.settle(); await persist(); },
|
||||
retry: async () => { 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;
|
||||
closing = true; sessions.dispose(); scheduler.stop();
|
||||
closing = true; sessions.dispose(); scheduler.stop(); runTracker.stop();
|
||||
for (const release of [...displays]) release();
|
||||
// Failure keeps the leader/recovery material available for an explicit retry of close.
|
||||
await persist(); unsubscribe(); editor.dispose(); await leader.release(); closed = true;
|
||||
|
||||
@@ -5,7 +5,8 @@ export type Presence = 'absent' | 'exists' | 'unknown';
|
||||
export type RequestPhase = 'frozen' | 'dispatch-marked' | 'waiting' | 'resolved';
|
||||
export type Outcome = 'unresolved' | 'accepted' | 'applied' | 'not-applied';
|
||||
export type FrozenRequest = {
|
||||
readonly kind?: 'document.change' | 'asset.upload';
|
||||
readonly kind?: OperationIntent['operationType'];
|
||||
readonly ownerCanvasId?: Id;
|
||||
readonly id: Id; readonly protocolVersion: number; readonly generation: number; readonly body: string;
|
||||
readonly operationIds: readonly Id[]; readonly canvasIds: readonly Id[]; readonly expectedVersions: Readonly<Record<Id, number>>;
|
||||
readonly phase: RequestPhase; readonly outcome: Outcome; readonly error?: string; readonly attempt: number;
|
||||
@@ -73,14 +74,15 @@ export function createLedger(initial?: LedgerSnapshot) {
|
||||
return !!operation && operation.phase === 'pending' && !operation.blockers.length && operation.intent.dependencies.every(dependency => operations.get(dependency)?.phase === 'resolved');
|
||||
},
|
||||
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(); },
|
||||
freeze(input: { id: Id; kind?: FrozenRequest['kind']; protocolVersion?: number; operationIds: readonly Id[]; canvasIds: readonly Id[]; expectedVersions: Readonly<Record<Id, number>>; payload: Json }) {
|
||||
freeze(input: { id: Id; kind?: FrozenRequest['kind']; ownerCanvasId?: Id; protocolVersion?: number; operationIds: readonly Id[]; canvasIds: readonly Id[]; expectedVersions: Readonly<Record<Id, number>>; 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') || !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');
|
||||
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]);
|
||||
}
|
||||
const request: FrozenRequest = freeze({ id: input.id, kind: input.kind ?? 'document.change', 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 });
|
||||
const request: FrozenRequest = freeze({ id: input.id, kind: input.kind ?? 'document.change', ...(input.ownerCanvasId ? { ownerCanvasId: input.ownerCanvasId } : {}), 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;
|
||||
|
||||
@@ -0,0 +1,26 @@
|
||||
import { fault } from '../canvas-core/values.ts';
|
||||
import { createDeque } from './structures.ts';
|
||||
|
||||
/** Shared bounded transport capacity for writes, uploads and run reads. */
|
||||
export function createNetworkGate(limit = 2) {
|
||||
if(!Number.isSafeInteger(limit)||limit<1)throw fault('invalid_concurrency');
|
||||
let active = 0;
|
||||
const waiting = createDeque<{ signal: AbortSignal; grant(): void; cancelled: boolean }>();
|
||||
async function enter(signal: AbortSignal) {
|
||||
signal.throwIfAborted();
|
||||
if (active < limit) { active++; return; }
|
||||
await new Promise<void>((resolve,reject) => {
|
||||
const item = { signal, cancelled: false, grant() { signal.removeEventListener('abort',abort);resolve(); } };
|
||||
const abort = () => { item.cancelled=true;signal.removeEventListener('abort',abort);reject(signal.reason); };
|
||||
signal.addEventListener('abort',abort,{once:true});waiting.push(item);
|
||||
});
|
||||
}
|
||||
function release() {
|
||||
let item=waiting.pop();while(item?.cancelled)item=waiting.pop();
|
||||
if(item)item.grant();else active--;
|
||||
}
|
||||
return Object.freeze({ async run<T>(signal: AbortSignal, callback: () => Promise<T>): Promise<T> {
|
||||
await enter(signal);
|
||||
try { signal.throwIfAborted();return await callback(); } finally { release(); }
|
||||
} });
|
||||
}
|
||||
@@ -1,8 +1,9 @@
|
||||
import type { Change, CoreSnapshot, Document, HistoryEntry, Id, Json, OperationIntent, Registry } from '../canvas-core/types.ts';
|
||||
import { createDocument, validateDocument } from '../canvas-core/document.ts';
|
||||
import { fault, freeze, readMap, validJson } from '../canvas-core/values.ts';
|
||||
import { fault, freeze, readMap, validJson, equal } from '../canvas-core/values.ts';
|
||||
import type { EngineSnapshot, RuntimeState } from './contracts.ts';
|
||||
import type { FrozenRequest, LedgerSnapshot, OperationRecord } from './ledger.ts';
|
||||
import { validateRunInput, type RunRecord } from './runs.ts';
|
||||
import type { ResourceRecord } from './resources.ts';
|
||||
import type { RecoveryImage } from './storage.ts';
|
||||
|
||||
@@ -13,7 +14,7 @@ export function encodeRecovery(snapshot: EngineSnapshot): RecoveryImage {
|
||||
document: documentValue(snapshot.document),
|
||||
history: { undo: snapshot.history.undo, redo: snapshot.history.redo, entries: [...snapshot.history.entries] },
|
||||
operations: [...snapshot.operations],
|
||||
runtime: { resources: [...snapshot.runtime.resources], resourcePins: [...snapshot.runtime.resourcePins], acknowledged: documentValue(snapshot.runtime.acknowledged), versions: [...snapshot.runtime.versions], receipts: [...snapshot.runtime.receipts], conflicts: [...snapshot.runtime.conflicts], ledger: { ...snapshot.runtime.ledger, operations: [...snapshot.runtime.ledger.operations], requests: [...snapshot.runtime.ledger.requests] } },
|
||||
runtime: { runs: [...snapshot.runtime.runs], resources: [...snapshot.runtime.resources], resourcePins: [...snapshot.runtime.resourcePins], acknowledged: documentValue(snapshot.runtime.acknowledged), versions: [...snapshot.runtime.versions], receipts: [...snapshot.runtime.receipts], conflicts: [...snapshot.runtime.conflicts], ledger: { ...snapshot.runtime.ledger, operations: [...snapshot.runtime.ledger.operations], requests: [...snapshot.runtime.ledger.requests] } },
|
||||
})) as Json;
|
||||
if (!validJson(payload)) throw fault('invalid_recovery_payload');
|
||||
return freeze({ formatVersion: 1, rootRevision: snapshot.rootRevision, localRevision: snapshot.revision, payload });
|
||||
@@ -56,6 +57,7 @@ function intent(value: unknown, id: Id): OperationIntent {
|
||||
if (string(raw.id) !== id) throw fault('corrupt_recovery_image');
|
||||
const kind = string(raw.operationType);
|
||||
if (kind === 'asset.upload') { string(raw.resourceId); if (array(raw.changes).length || array(raw.canvasIds).length) throw fault('corrupt_recovery_image'); }
|
||||
if (kind === 'run.start') { validateRunInput(raw.run as never);if(array(raw.changes).length || !strings(raw.canvasIds).includes(string(object(raw.run).boardId)))throw fault('corrupt_recovery_image'); }
|
||||
integer(raw.operationSchemaVersion, 1); integer(raw.sequence, 1); string(raw.origin);
|
||||
strings(raw.canvasIds); strings(raw.dependencies); changes(raw.changes);
|
||||
return freeze(value as OperationIntent);
|
||||
@@ -76,10 +78,10 @@ export function decodeRecovery(image: RecoveryImage, registry: Registry): CoreSn
|
||||
if (new Set(stackIds).size !== stackIds.length || stackIds.length !== historyEntries.size || stackIds.some(id => !historyEntries.has(id))) throw fault('corrupt_recovery_image');
|
||||
const records = table<OperationRecord>(ledger.operations, (value, id) => {
|
||||
const raw = object(value); intent(raw.intent, id); strings(raw.blockers); object(raw.basePresence);
|
||||
if ((object(raw.intent).operationType !== 'asset.upload' && !operations.has(id)) || !['pending', 'frozen', 'resolved', 'blocked'].includes(string(raw.phase))) throw fault('corrupt_recovery_image');
|
||||
if ((object(raw.intent).operationType === 'document.change' && !operations.has(id)) || !['pending', 'frozen', 'resolved', 'blocked'].includes(string(raw.phase))) throw fault('corrupt_recovery_image');
|
||||
return freeze(value as OperationRecord);
|
||||
});
|
||||
if (operations.size !== [...records.values()].filter(record => record.intent.operationType !== 'asset.upload').length) throw fault('corrupt_recovery_image');
|
||||
if (operations.size !== [...records.values()].filter(record => record.intent.operationType === 'document.change').length) throw fault('corrupt_recovery_image');
|
||||
// Validate the dependency DAG iteratively, including missing dependencies, without recursion overflow.
|
||||
const counts = new Map<Id, number>(), dependents = new Map<Id, Id[]>(), ready: Id[] = [];
|
||||
for (const [id, record] of records) {
|
||||
@@ -95,9 +97,16 @@ export function decodeRecovery(image: RecoveryImage, registry: Registry): CoreSn
|
||||
if (string(raw.id) !== id || typeof raw.body !== 'string') throw fault('corrupt_recovery_image');
|
||||
integer(raw.protocolVersion, 1);
|
||||
integer(raw.generation, 1); integer(raw.attempt); strings(raw.canvasIds); object(raw.expectedVersions);
|
||||
if (raw.kind === 'run.start' && !strings(raw.canvasIds).includes(string(raw.ownerCanvasId)))throw fault('corrupt_recovery_image');
|
||||
if (raw.kind === 'asset.upload' && (array(raw.canvasIds).length || Object.keys(object(raw.expectedVersions)).length || array(raw.operationIds).length !== 1)) throw fault('corrupt_recovery_image');
|
||||
if (!['frozen', 'dispatch-marked', 'waiting', 'resolved'].includes(string(raw.phase)) || !['unresolved', 'accepted', 'applied', 'not-applied'].includes(string(raw.outcome))) throw fault('corrupt_recovery_image');
|
||||
for (const operationId of strings(raw.operationIds)) if (records.get(operationId)?.requestId !== id || (raw.kind ?? 'document.change') !== records.get(operationId)?.intent.operationType) throw fault('corrupt_recovery_image');
|
||||
if(raw.kind==='run.start'){
|
||||
if(array(raw.operationIds).length!==1)throw fault('corrupt_recovery_image');
|
||||
const operation=records.get(string(array(raw.operationIds)[0]))!.intent;if(operation.operationType!=='run.start')throw fault('corrupt_recovery_image');
|
||||
let body:ObjectValue;try{body=object(JSON.parse(raw.body as string));}catch{throw fault('corrupt_recovery_image');}
|
||||
if(raw.ownerCanvasId!==operation.run.boardId||body.protocol_version!==2||body.request_id!==id||body.operation_id!==operation.id||body.node_id!==operation.run.target.id||body.target_incarnation!==operation.run.target.incarnation||body.model_id!==operation.run.modelId||body.prompt!==operation.run.prompt||body.propagate!==operation.run.propagate||!equal(body.parameters,operation.run.parameters)||!equal(body.expected_versions,raw.expectedVersions)||!equal([...strings(raw.canvasIds)].sort(),[...operation.canvasIds].sort()))throw fault('corrupt_recovery_image');
|
||||
}
|
||||
if (raw.retryAt !== undefined && (typeof raw.retryAt !== 'number' || !Number.isFinite(raw.retryAt))) throw fault('corrupt_recovery_image');
|
||||
if (raw.retryAction !== undefined && !['submit', 'query'].includes(string(raw.retryAction))) throw fault('corrupt_recovery_image');
|
||||
return freeze(value as FrozenRequest);
|
||||
@@ -121,8 +130,21 @@ export function decodeRecovery(image: RecoveryImage, registry: Registry): CoreSn
|
||||
});
|
||||
for (const record of records.values()) if (record.intent.operationType === 'asset.upload' && resources.get(record.intent.resourceId)?.operationId !== record.intent.id) throw fault('corrupt_recovery_image');
|
||||
const resourcePins = table(runtime.resourcePins ?? [], strings);
|
||||
const runs=table<RunRecord>(runtime.runs ?? [],(value,id)=>{
|
||||
const raw=object(value),operation=records.get(string(raw.operationId)),request=requests.get(string(raw.requestId)),target=object(raw.target);
|
||||
if(string(raw.id)!==id||operation?.intent.operationType!=='run.start'||operation.requestId!==raw.requestId||request?.kind!=='run.start'||request.outcome!=='applied'||operation.intent.run.boardId!==string(raw.boardId)||operation.intent.run.target.id!==string(target.id)||operation.intent.run.target.incarnation!==integer(target.incarnation,1))throw fault('corrupt_recovery_image');
|
||||
if(!['queued','running','succeeded','failed','conflict','cancelled','partially_succeeded'].includes(string(raw.status))||typeof raw.hasMore!=='boolean')throw fault('corrupt_recovery_image');
|
||||
integer(raw.cursor);integer(raw.attempt);object(raw.plan);array(raw.outputs);array(raw.steps);array(raw.outputEvents ?? []);
|
||||
if(raw.blocked!==undefined&&typeof raw.blocked!=='boolean')throw fault('corrupt_recovery_image');
|
||||
if(raw.error!==undefined)string(raw.error);
|
||||
if(raw.nextPollAt!==undefined&&(typeof raw.nextPollAt!=='number'||!Number.isFinite(raw.nextPollAt)))throw fault('corrupt_recovery_image');
|
||||
if(raw.pendingPage!==undefined){const page=object(raw.pendingPage);let cursor=integer(raw.cursor);for(const value of array(page.events)){const event=object(value);if(event.run_id!==id||integer(event.seq,1)!==++cursor)throw fault('corrupt_recovery_image');}if(integer(page.next)!==cursor||typeof page.hasMore!=='boolean'||!['queued','running','succeeded','failed','conflict','cancelled','partially_succeeded'].includes(string(page.status)))throw fault('corrupt_recovery_image');array(page.outputs);array(page.steps);}
|
||||
const outputIds=new Set<string>();
|
||||
for(const value of array(raw.outputEvents ?? [])){const event=object(value),step=string(event.step_id);if(event.run_id!==id||event.type!=='step.output'||outputIds.has(step)||integer(event.seq,1)>Math.max(integer(raw.cursor),raw.pendingPage===undefined?0:integer(object(raw.pendingPage).next)))throw fault('corrupt_recovery_image');object(event.output);outputIds.add(step);}
|
||||
return freeze({...value as RunRecord,outputEvents:array(raw.outputEvents ?? []) as never});
|
||||
});
|
||||
const restoredLedger: LedgerSnapshot = Object.freeze({ generation, durableGeneration, operations: records, requests });
|
||||
return Object.freeze({ revision: image.localRevision, rootRevision: image.rootRevision, document, history: Object.freeze({ undo: freeze(undo), redo: freeze(redo), entries: historyEntries }), operations,
|
||||
runtime: Object.freeze({ resources, resourcePins, ledger: restoredLedger, acknowledged, versions, receipts: table(runtime.receipts, value => freeze(value as Json)), conflicts: table(runtime.conflicts, string) }),
|
||||
runtime: Object.freeze({ runs, resources, resourcePins, ledger: restoredLedger, acknowledged, versions, receipts: table(runtime.receipts, value => freeze(value as Json)), conflicts: table(runtime.conflicts, string) }),
|
||||
});
|
||||
}
|
||||
|
||||
@@ -0,0 +1,104 @@
|
||||
import type { Id, Json, OperationIntent, Target, RunInput } from '../canvas-core/types.ts';
|
||||
import type { 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';
|
||||
import { createMinHeap } from './structures.ts';
|
||||
|
||||
export function validateRunInput(input: RunInput) {
|
||||
if (!input || typeof input.boardId!=='string' || !input.boardId || typeof input.target?.id!=='string' || !input.target.id || !Number.isSafeInteger(input.target.incarnation) || input.target.incarnation<1 || typeof input.modelId!=='string' || !input.modelId || typeof input.prompt!=='string' || !input.prompt.trim() || [...input.prompt].length>16000 || typeof input.propagate!=='boolean' || !input.parameters || Array.isArray(input.parameters) || typeof input.parameters!=='object' || !validJson(input.parameters) || new TextEncoder().encode(JSON.stringify(input.parameters)).length>16384) throw fault('invalid_ai_request');
|
||||
}
|
||||
|
||||
export type RunModel = { readonly id: string; readonly label?: string; readonly display_name?: string; readonly output_modalities: readonly string[]; readonly input_modalities: readonly string[]; readonly available: boolean; readonly unavailable_reason?: string };
|
||||
export type RunCatalog = { readonly models: readonly RunModel[]; readonly unavailableReason?: string };
|
||||
export type RunStatus = 'queued' | 'running' | 'succeeded' | 'failed' | 'conflict' | 'cancelled' | 'partially_succeeded';
|
||||
export type RunEvent = { readonly seq: number; readonly run_id: Id; readonly type: string; readonly [key: string]: Json };
|
||||
export type RunPage = { readonly status: RunStatus; readonly events: readonly RunEvent[]; readonly next: number; readonly hasMore: boolean; readonly outputs: readonly Json[]; readonly steps: readonly Json[]; readonly error?: string };
|
||||
export type RunRecord = {
|
||||
readonly id: Id; readonly operationId: Id; readonly requestId: Id; readonly boardId: Id; readonly target: Target;
|
||||
readonly plan: Json; readonly status: RunStatus; readonly cursor: number; readonly hasMore: boolean;
|
||||
readonly outputs: readonly Json[]; readonly outputEvents: readonly RunEvent[]; readonly steps: readonly Json[]; readonly attempt: number;
|
||||
readonly nextPollAt?: number; readonly error?: string; readonly blocked?: boolean; readonly pendingPage?: RunPage;
|
||||
};
|
||||
export type RunPort = {
|
||||
catalog(boardId: Id, signal: AbortSignal): Promise<RunCatalog>;
|
||||
compile(intent: OperationIntent, requestId: Id, snapshot: EngineSnapshot): CompiledRequest;
|
||||
submit(request: FrozenRequest, signal: AbortSignal): Promise<OperationStatus>;
|
||||
query(request: FrozenRequest, signal: AbortSignal): Promise<OperationStatus>;
|
||||
accept(request: FrozenRequest, status: OperationStatus, snapshot: EngineSnapshot): { readonly run: RunRecord; readonly proposal: CanonicalProposal };
|
||||
poll(run: RunRecord, signal: AbortSignal): Promise<RunPage>;
|
||||
canonical(event: RunEvent, snapshot: EngineSnapshot): CanonicalProposal;
|
||||
};
|
||||
export function runContainsTarget(run: RunRecord, target: Target) {
|
||||
if(run.target.id===target.id && run.target.incarnation===target.incarnation)return true;
|
||||
const plan=run.plan;
|
||||
if(!plan||typeof plan!=='object'||Array.isArray(plan)||!Array.isArray(plan.steps))return false;
|
||||
return plan.steps.some(step=>!!step&&typeof step==='object'&&!Array.isArray(step)&&step.node_id===target.id&&step.target_incarnation===target.incarnation);
|
||||
}
|
||||
|
||||
export const runTerminal = (status: RunStatus) => status !== 'queued' && status !== 'running';
|
||||
|
||||
/** Recoverable poll deadlines are facts; heap/tokens/active controllers are disposable indexes. */
|
||||
export function createRunTracker(options: {
|
||||
clock: Clock; getRuns(): ReadonlyMap<Id, RunRecord>; persist(): Promise<void>;
|
||||
poll(run: RunRecord, signal: AbortSignal): Promise<RunPage>;
|
||||
accept(run: RunRecord, page: RunPage): void;
|
||||
update(id: Id, reduce: (run: RunRecord) => RunRecord): void;
|
||||
onError(error: unknown): void;
|
||||
}) {
|
||||
const heap = createMinHeap<{ id: Id; due: number; token: number; sequence: number }>((a,b) => a.due-b.due || a.sequence-b.sequence);
|
||||
const tokens = new Map<Id,number>();
|
||||
let sequence = 0, stopped = false, timer: unknown, active: { controller: AbortController; promise: Promise<void> } | undefined;
|
||||
const report = (error: unknown) => { try { options.onError(error); } catch { /* Reporting cannot alter cursor durability. */ } };
|
||||
const eligible = (run: RunRecord) => !run.blocked && (run.hasMore || !runTerminal(run.status));
|
||||
function arm() {
|
||||
if (timer !== undefined) options.clock.clearTimer(timer); timer = undefined;
|
||||
while (heap.peek() && tokens.get(heap.peek()!.id) !== heap.peek()!.token) heap.pop();
|
||||
if (stopped || active || !heap.peek()) return;
|
||||
timer = options.clock.setTimer(() => { timer = undefined; pump(); }, Math.max(0,heap.peek()!.due-options.clock.now()));
|
||||
}
|
||||
function notify(ids: readonly Id[]) {
|
||||
if (stopped) return;
|
||||
for (const id of ids) {
|
||||
const run = options.getRuns().get(id); if (!run || !eligible(run)) continue;
|
||||
const token = (tokens.get(id) ?? 0)+1; tokens.set(id,token);
|
||||
heap.push({ id, due:run.nextPollAt ?? options.clock.now(), token, sequence:++sequence });
|
||||
}
|
||||
arm();
|
||||
}
|
||||
function pump() {
|
||||
if (stopped || active) return;
|
||||
let item = heap.pop();
|
||||
while (item && tokens.get(item.id)!==item.token) item=heap.pop();
|
||||
if (!item) return;
|
||||
if (item.due>options.clock.now()) { heap.push(item);arm();return; }
|
||||
const run=options.getRuns().get(item.id);if (!run || !eligible(run)) { arm();return; }
|
||||
const controller=new AbortController();
|
||||
const promise=Promise.resolve().then(async()=>{
|
||||
try {
|
||||
// Admission identity and cursor must be durable before a read can trigger coordination.
|
||||
await options.persist();if(stopped)return;
|
||||
const page=run.pendingPage ?? await options.poll(run,controller.signal);if(stopped)return;
|
||||
options.accept(run,page);
|
||||
const latest=options.getRuns().get(run.id)!;
|
||||
options.update(run.id,current=>({...current,attempt:0,nextPollAt:options.clock.now()+(current.pendingPage ? 1000 : current.hasMore ? 1 : 1000)}));
|
||||
// Storage failure blocks this tracker until explicit retry, retaining in-memory material.
|
||||
await options.persist();
|
||||
if(!stopped && !latest.blocked)notify([run.id]);
|
||||
} catch(error) {
|
||||
if(stopped)return;
|
||||
const code=(error as {code?:string}).code ?? (error as Error).message ?? 'run_poll_failed';
|
||||
const blocked=['forbidden','unauthorized','identity_changed','invalid_run_response','run_identity_mismatch','run_receipt_expired','run_interpreter_unavailable'].includes(code);
|
||||
options.update(run.id,current=>({...current,error:code,blocked,attempt:current.attempt+1,nextPollAt:options.clock.now()+Math.min(30000,500*2**Math.min(current.attempt,6))}));
|
||||
try { await options.persist();if(!blocked&&!stopped)notify([run.id]); } catch(storageError) { options.update(run.id,current=>({...current,blocked:true,error:'run_storage_failed'}));report(storageError); }
|
||||
report(error);
|
||||
}
|
||||
}).finally(()=>{active=undefined;arm();});
|
||||
active={controller,promise};
|
||||
}
|
||||
return Object.freeze({ notify,
|
||||
async retry() { await options.persist();for(const id of options.getRuns().keys())options.update(id,run=>({...run,blocked:false,error:undefined,nextPollAt:options.clock.now()}));notify([...options.getRuns().keys()]); },
|
||||
async settle() { while(active)await active.promise; },
|
||||
stop() { stopped=true;if(timer!==undefined)options.clock.clearTimer(timer);timer=undefined;active?.controller.abort();heap.clear(); },
|
||||
});
|
||||
}
|
||||
@@ -4,7 +4,7 @@ import { createLedger, type FrozenRequest, type LedgerSnapshot, type Outcome } f
|
||||
import { createDeque, createMinHeap } from './structures.ts';
|
||||
|
||||
export type OperationStatus = { readonly outcome: Outcome; readonly error?: string; readonly retryAfterMs?: number; readonly canonical?: Json; readonly blocked?: boolean };
|
||||
export type CompiledRequest = { readonly kind?: FrozenRequest['kind']; readonly protocolVersion?: number; readonly payload: Json; readonly canvasIds: readonly Id[]; readonly expectedVersions: Readonly<Record<Id, number>> };
|
||||
export type CompiledRequest = { readonly kind?: FrozenRequest['kind']; readonly ownerCanvasId?: Id; readonly protocolVersion?: number; readonly payload: Json; readonly canvasIds: readonly Id[]; readonly expectedVersions: Readonly<Record<Id, number>> };
|
||||
export type Clock = { now(): number; setTimer(callback: () => void, delay: number): unknown; clearTimer(handle: unknown): void };
|
||||
const browserClock: Clock = { now: () => Date.now(), setTimer: (fn, delay) => setTimeout(fn, delay), clearTimer: handle => clearTimeout(handle as ReturnType<typeof setTimeout>) };
|
||||
type Ledger = ReturnType<typeof createLedger>;
|
||||
|
||||
@@ -7,7 +7,7 @@ export type ProjectEngine = Awaited<ReturnType<typeof openProjectEngine>>;
|
||||
export type WorkspaceOptions = {
|
||||
accountId: Id; workspaceId: Id; authEpoch: number; registry: Registry; storage: StoragePort;
|
||||
isCurrentIdentity(): boolean;
|
||||
adapters(projectId: Id, current: () => boolean): Pick<EngineOptions, 'persistence' | 'assets'>;
|
||||
adapters(projectId: Id, current: () => boolean): Pick<EngineOptions, 'persistence' | 'assets' | 'runs'>;
|
||||
onError?: (error: unknown) => void;
|
||||
};
|
||||
|
||||
|
||||
@@ -131,17 +131,22 @@ export function createFoundryPersistence(options: { workspaceId: Id; rootId: Id;
|
||||
canonical(request: FrozenRequest, status: OperationStatus, snapshot: EngineSnapshot) {
|
||||
const body = record(status.canonical);
|
||||
if (body.request_id !== request.id || body.protocol_version !== 2) throw fault('receipt_identity_mismatch');
|
||||
const chunk = modelChunk(body.model);
|
||||
if (chunk.projectId !== options.rootId || chunk.complete || chunk.versions.size !== request.canvasIds.length || request.canvasIds.some(id => !chunk.versions.has(id))) throw fault('receipt_coverage_mismatch');
|
||||
const previous = snapshot.runtime.acknowledged, canvases = new Map(previous.canvases), elements = new Map(previous.elements), tombstones = new Map(previous.tombstones), serverVersions = new Map(snapshot.runtime.versions);
|
||||
for (const canvas of request.canvasIds) canvases.delete(canvas);
|
||||
for (const canvas of chunk.canvases) canvases.set(canvas.id, canvas);
|
||||
for (const [id, element] of elements) if (request.canvasIds.includes(element.ownerCanvasId)) elements.delete(id);
|
||||
for (const element of chunk.elements) { if (elements.has(element.id)) throw fault('canonical_scope_mismatch', [element.id]); elements.set(element.id, element); }
|
||||
for (const [id, generation] of chunk.tombstones) tombstones.set(id, Math.max(tombstones.get(id) ?? 0, generation));
|
||||
for (const [id, version] of chunk.versions) serverVersions.set(id, version);
|
||||
const document = createDocument(previous.projectId, [...canvases.values()], [...elements.values()], tombstones); validateDocument(document, options.registry);
|
||||
return Object.freeze({ document, versions: readMap(serverVersions) });
|
||||
return mergeFoundryModel(body.model, request.canvasIds, snapshot, options.registry, options.rootId);
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
/** 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);
|
||||
if (chunk.projectId !== rootId || chunk.complete || chunk.versions.size !== participants.length || participants.some(id => !chunk.versions.has(id))) throw fault('receipt_coverage_mismatch');
|
||||
const previous = snapshot.runtime.acknowledged, canvases = new Map(previous.canvases), elements = new Map(previous.elements), tombstones = new Map(previous.tombstones), serverVersions = new Map(snapshot.runtime.versions);
|
||||
for (const canvas of participants) canvases.delete(canvas);
|
||||
for (const canvas of chunk.canvases) canvases.set(canvas.id, canvas);
|
||||
for (const [id, element] of elements) if (participants.includes(element.ownerCanvasId)) elements.delete(id);
|
||||
for (const element of chunk.elements) { if (elements.has(element.id)) throw fault('canonical_scope_mismatch', [element.id]); elements.set(element.id, element); }
|
||||
for (const [id, generation] of chunk.tombstones) tombstones.set(id, Math.max(tombstones.get(id) ?? 0, generation));
|
||||
for (const [id, version] of chunk.versions) serverVersions.set(id, version);
|
||||
const document = createDocument(previous.projectId, [...canvases.values()], [...elements.values()], tombstones); validateDocument(document, registry);
|
||||
return Object.freeze({ document, versions: readMap(serverVersions) });
|
||||
}
|
||||
|
||||
@@ -1,2 +1,4 @@
|
||||
export * from './foundry.ts';
|
||||
export * from './assets.ts';
|
||||
|
||||
export * from './runs.ts';
|
||||
|
||||
@@ -0,0 +1,96 @@
|
||||
import type { Id, Json, Registry, RunInput } from '../canvas-core/types.ts';
|
||||
import { equal, fault, freeze, 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, 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']);
|
||||
const object = (value: unknown): Record<string, unknown> => { if (!value || Array.isArray(value) || typeof value !== 'object') throw fault('invalid_run_response'); return value as Record<string,unknown>; };
|
||||
const array = (value: unknown): unknown[] => { if (!Array.isArray(value)) throw fault('invalid_run_response'); return value; };
|
||||
const string = (value: unknown): string => { if (typeof value !== 'string' || !value) throw fault('invalid_run_response');return value; };
|
||||
const integer = (value: unknown, min=0): number => { if (typeof value!=='number'||!Number.isSafeInteger(value)||value<min)throw fault('invalid_run_response');return value; };
|
||||
|
||||
function payload(request: FrozenRequest) {
|
||||
const raw=object(JSON.parse(request.body));
|
||||
if(!request.ownerCanvasId||!uuid.test(request.ownerCanvasId)||request.kind!=='run.start'||request.protocolVersion!==2||!uuid.test(request.id)||raw.protocol_version!==2||raw.request_id!==request.id||request.operationIds.length!==1||raw.operation_id!==request.operationIds[0]||!uuid.test(string(raw.operation_id))||!request.ownerCanvasId||!request.canvasIds.includes(request.ownerCanvasId)||!equal(raw.expected_versions,request.expectedVersions)||!equal(Object.keys(request.expectedVersions).sort(),[...request.canvasIds].sort()))throw fault('invalid_run_request');
|
||||
for(const version of Object.values(request.expectedVersions))integer(version,1);
|
||||
validateRunInput({target:{id:string(raw.node_id),incarnation:integer(raw.target_incarnation,1)},boardId:request.ownerCanvasId,modelId:string(raw.model_id),prompt:string(raw.prompt),parameters:object(raw.parameters) as RunInput['parameters'],propagate:raw.propagate as boolean});
|
||||
return raw;
|
||||
}
|
||||
function planSteps(plan: unknown, rootId: Id) {
|
||||
const steps=array(object(plan).steps).map(object),ids=new Set<string>();
|
||||
for(const step of steps){const id=string(step.step_id);if(ids.has(id)||step.canonical_root!==rootId||!uuid.test(string(step.board_id)))throw fault('run_identity_mismatch');ids.add(id);string(step.node_id);integer(step.target_incarnation,1);}
|
||||
if(!steps.length)throw fault('invalid_run_response');return steps;
|
||||
}
|
||||
|
||||
/** Admission completion is distinct from paid execution completion. Credentials remain in the transport. */
|
||||
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`;
|
||||
async function execute(request: FrozenRequest,query: boolean,signal: AbortSignal): Promise<OperationStatus> {
|
||||
try{
|
||||
payload(request);
|
||||
const response=await options.transport({method:query?'GET':'POST',path:`${path(request.ownerCanvasId!)}/${query?`operations/${encodeURIComponent(request.id)}`:'runs'}`,requestId:query?undefined:request.id,expectedVersion:query?undefined:request.expectedVersions[request.ownerCanvasId!],body:query?undefined:request.body,signal});
|
||||
const body=object(response.body),matching=body.protocol_version===2&&body.request_id===request.id;
|
||||
if(matching&&body.operation_id===request.operationIds[0]&&body.admission==='accepted'&&response.status>=200&&response.status<300&&validJson(body))return freeze({outcome:'applied',canonical:JSON.parse(JSON.stringify(body)) as Json});
|
||||
if(query&&matching&&response.status===404&&body.status==='unknown')return{outcome:'unresolved',error:'ai_admission_unknown',retryAfterMs:response.retryAfterMs};
|
||||
return{outcome:'unresolved',error:typeof body.error==='string'?body.error:'ai_admission_unknown',retryAfterMs:response.retryAfterMs,...(response.status>=400&&response.status<500?{blocked:true}:{})};
|
||||
}catch(error){const code=(error as {code?:string}).code;return{outcome:'unresolved',error:code??'ai_transport_unknown',...(code?{blocked:true}:{})};}
|
||||
}
|
||||
return Object.freeze({
|
||||
async catalog(boardId,signal){
|
||||
if(!uuid.test(boardId))throw fault('invalid_service_scope');
|
||||
const response=await options.transport({method:'GET',path:`${path(boardId)}/models`,signal}),body=object(response.body);
|
||||
if(response.status!==200)throw fault(typeof body.error==='string'?body.error:'model_catalog_unavailable');
|
||||
const models=array(body.models).map(value=>{const model=object(value);string(model.id);for(const key of ['output_modalities','input_modalities'])for(const modality of array(model[key]))string(modality);if(typeof model.available!=='boolean'||!validJson(model))throw fault('invalid_run_response');return JSON.parse(JSON.stringify(model)) as RunModel;});
|
||||
return freeze({models,...(typeof body.unavailable_reason==='string'?{unavailableReason:body.unavailable_reason}:{})});
|
||||
},
|
||||
compile(intent,requestId,snapshot){
|
||||
if(intent.operationType!=='run.start'||intent.operationSchemaVersion!==1||!uuid.test(intent.id)||!uuid.test(requestId)||snapshot.document.projectId!==options.rootId)throw fault('operation_interpreter_unavailable');
|
||||
validateRunInput(intent.run);
|
||||
const before=snapshot.runtime.acknowledged,element=before.elements.get(intent.run.target.id);
|
||||
if(!element||element.incarnation!==intent.run.target.incarnation||element.ownerCanvasId!==intent.run.boardId||element.structure!=='node'||element.dataVersion!==1||!['text','plain-text','script','image','plain-image'].includes(element.componentId))throw fault('target_changed');
|
||||
const canvasIds=[...before.canvases.keys()].sort(),expectedVersions:Record<Id,number>={};
|
||||
if(!equal(canvasIds,[...intent.canvasIds].sort()))throw fault('run_context_changed');
|
||||
for(const id of canvasIds)expectedVersions[id]=integer(snapshot.runtime.versions.get(id),1);
|
||||
const run=intent.run;
|
||||
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),
|
||||
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.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};
|
||||
},
|
||||
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);
|
||||
if(response.status!==200)throw fault(response.status===404?'run_receipt_expired':typeof body.error==='string'?body.error:'run_poll_failed');
|
||||
if(!validJson(body)||body.run_id!==run.id||!statuses.has(string(body.status))||typeof body.has_more!=='boolean'||!equal(body.plan,run.plan))throw fault('invalid_run_response');
|
||||
const steps=planSteps(run.plan,options.rootId),events=array(body.events).map(object);let cursor=run.cursor;
|
||||
for(const event of events){
|
||||
if(event.run_id!==run.id||integer(event.seq,1)!==cursor+1||typeof event.type!=='string')throw fault('invalid_run_response');cursor++;
|
||||
if(event.type==='step.applied'||event.type==='step.output'){
|
||||
const step=steps.find(step=>step.step_id===event.step_id);
|
||||
if(!step||step.board_id!==event.board_id||step.node_id!==event.node_id||event.type==='step.applied'&&(step.target_incarnation!==event.target_incarnation||event.document_format_version!==1))throw fault('run_identity_mismatch');
|
||||
}
|
||||
}
|
||||
if(integer(body.next_seq)!==cursor||events.length>100||body.has_more&&events.length===0)throw fault('invalid_run_response');
|
||||
return freeze({status:body.status as RunStatus,events:JSON.parse(JSON.stringify(events)) as RunEvent[],next:cursor,hasMore:body.has_more,outputs:array(body.outputs) as Json[],steps:array(body.steps) as Json[],...(typeof body.error==='string'?{error:body.error}:{})});
|
||||
},
|
||||
canonical(event,snapshot){
|
||||
const board=string(event.board_id),version=integer(event.canvas_version,1),current=snapshot.runtime.acknowledged.elements.get(string(event.node_id));
|
||||
if(event.type!=='step.applied'||!current||current.ownerCanvasId!==board||current.incarnation!==event.target_incarnation)throw fault('run_identity_mismatch');
|
||||
const proposal=mergeFoundryModel(event.model,[board],snapshot,options.registry,options.rootId),target=proposal.document.elements.get(string(event.node_id));
|
||||
if(proposal.versions.get(board)!==version||!target||target.ownerCanvasId!==board||target.incarnation!==event.target_incarnation)throw fault('run_identity_mismatch');
|
||||
return proposal;
|
||||
},
|
||||
});
|
||||
}
|
||||
@@ -1,9 +1,39 @@
|
||||
import { createBrowserStorage } from '../../src/lib/canvas-runtime/storage.ts';
|
||||
import { openProjectEngine } from '../../src/lib/canvas-runtime/engine.ts';
|
||||
import { createDocument, createRegistry, builtinComponents, applyChanges } from '../../src/lib/canvas-core/index.ts';
|
||||
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 runRecoveryCase(name,assert){
|
||||
const root=crypto.randomUUID(),workspace=crypto.randomUUID(),account=crypto.randomUUID(),runId=crypto.randomUUID();
|
||||
const registry=createRegistry(builtinComponents()),element={id:'a',incarnation:1,structure:'node',componentId:'text',dataVersion:1,ownerCanvasId:root,geometry:{x:0,y:0,width:300,height:200,rotation:0},data:{title:'a',text:'original'}},original=createDocument(root,[{id:root,parent:null}],[element]);
|
||||
const storage=createBrowserStorage({name:`${name}-runs`}),calls=[];let accepted,after,plan,engine;
|
||||
const chunk=(document,version)=>({documentFormatVersion:1,projectId:root,complete:false,removedCanvasIds:[],canvases:[...document.canvases.values()],elements:[...document.elements.values()],tombstones:{},versions:{[root]:version}});
|
||||
const transport=async request=>{
|
||||
calls.push(request);
|
||||
if(request.method==='POST'){
|
||||
const input=JSON.parse(request.body),configured=applyChanges(original,[{kind:'field',target:{id:'a',incarnation:1},field:'data.ai',before:null,beforePresent:false,after:{enabled:true,modelId:input.model_id,prompt:input.prompt,parameters:{},propagate:false},afterPresent:true}]);
|
||||
plan={steps:[{step_id:'step-a',board_id:root,node_id:'a',target_incarnation:1,canonical_root:root}],edges:[],frozen_versions:{[root]:2}};
|
||||
accepted={protocol_version:2,request_id:request.requestId,operation_id:input.operation_id,admission:'accepted',status:'queued',run_id:runId,plan,canvas_version:2,document_format_version:1,target_incarnation:1,model:chunk(configured,2)};
|
||||
after=applyChanges(configured,[{kind:'field',target:{id:'a',incarnation:1},field:'data.text',before:'original',after:'durable generated',beforePresent:true,afterPresent:true}]);
|
||||
return{status:502,body:{error:'response_lost'}};
|
||||
}
|
||||
if(request.path.includes('/ai/operations/'))return{status:200,body:accepted};
|
||||
return{status:200,body:{run_id:runId,status:'succeeded',plan,events:[{seq:1,run_id:runId,type:'run.queued',status:'queued'},{seq:2,run_id:runId,type:'step.applied',step_id:'step-a',board_id:root,node_id:'a',target_incarnation:1,document_format_version:1,canvas_version:3,model:chunk(after,3)}],next_seq:2,has_more:false,outputs:[],steps:[]}};
|
||||
};
|
||||
const options={scope:{accountId:account,workspaceId:workspace,projectId:root},registry,document:original,versions:new Map([[root,1]]),storage,persistence:createFoundryPersistence({workspaceId:workspace,rootId:root,registry,transport}),runs:createFoundryRuns({workspaceId:workspace,rootId:root,registry,transport})};
|
||||
try{
|
||||
engine=await openProjectEngine(options);const detach=engine.sessions.attach('view');engine.startRun({target:{id:'a',incarnation:1},boardId:root,modelId:'text',prompt:'Generate',parameters:{},propagate:false});detach();await engine.settle();
|
||||
const request=[...engine.getSnapshot().runtime.ledger.requests.values()][0];assert(engine.getSnapshot().runtime.runs.size===0,'lost admission was guessed as resolved');await engine.close();
|
||||
engine=await openProjectEngine({...options,clock:{now:()=>Date.now()+100000,setTimer:(fn,ms)=>setTimeout(fn,ms),clearTimer:id=>clearTimeout(id)}});
|
||||
for(let i=0;i<100&&engine.getSnapshot().runtime.runs.get(runId)?.cursor!==2;i++)await new Promise(resolve=>setTimeout(resolve,10));await engine.settle();
|
||||
assert(calls.filter(call=>call.method==='POST').length===1,'recovery invoked a second generation');assert(calls.some(call=>call.path.endsWith('/ai/operations/'+request.id)),'recovery lost the original admission identity');
|
||||
assert(engine.getSnapshot().runtime.runs.get(runId)?.cursor===2&&engine.getSnapshot().document.elements.get('a').data.text==='durable generated','run and canonical result did not publish together');assert(engine.getSnapshot().history.undo.length===0,'ACK replay created user history');await engine.close();
|
||||
const leader=await storage.claim(options.scope);try{const image=await storage.recover(leader);assert(image.payload.runtime.runs[0][1].cursor===2,'run cursor was not durable');assert(image.payload.document.elements[0].data.text==='durable generated','cursor advanced without durable document');}finally{await leader.release();}
|
||||
}finally{await engine?.close();storage.close();indexedDB.deleteDatabase(`${name}-runs`);}
|
||||
}
|
||||
|
||||
async function resourceRecoveryCase(name, assert) {
|
||||
const scope = { accountId: crypto.randomUUID(), workspaceId: crypto.randomUUID(), projectId: crypto.randomUUID() };
|
||||
const registry = createRegistry(builtinComponents()), document = createDocument(scope.projectId, [{ id: scope.projectId, parent: null }], []);
|
||||
@@ -173,6 +203,8 @@ export async function run() {
|
||||
results.push('staged file and image recover from real IndexedDB; lost upload ACK queries stable resource identity');
|
||||
await protocolRecoveryCase(name,assert);
|
||||
results.push('protocol 2 service adapter recovers exact frozen bytes and reconciles a lost ACK through real IndexedDB');
|
||||
await runRecoveryCase(name,assert);
|
||||
results.push('AI admission and RunRecord recover through real IndexedDB; cursor and generated content share a durable cut without a second invocation');
|
||||
return results;
|
||||
} finally { await leader?.release(); other?.close(); storage.close(); indexedDB.deleteDatabase(name); }
|
||||
}
|
||||
|
||||
@@ -9,7 +9,7 @@ import { setTimeout as pause } from 'node:timers/promises';
|
||||
import { createDocument, applyChanges } from '../../src/lib/canvas-core/index.ts';
|
||||
const source = resolve(import.meta.dirname, '../..'), directory = await mkdtemp(join(tmpdir(), 'endoleap-renderer-')), profile = await mkdtemp(join(tmpdir(), 'endoleap-renderer-browser-'));
|
||||
const id = n => `10000000-0000-4000-8000-${String(n).padStart(12, '0')}`;
|
||||
const account = id(1), workspace = id(2), root = id(3), other = id(4), calls = [], uploads = [], documents = new Map(), versions = new Map([[root, 1], [other, 1]]), assets = new Map();
|
||||
const account = id(1), workspace = id(2), root = id(3), other = id(4), calls = [], uploads = [], documents = new Map(), versions = new Map([[root, 1], [other, 1]]), assets = new Map(), aiRuns = new Map();
|
||||
const node = (id, ownerCanvasId, x) => ({ id, ownerCanvasId, incarnation: 1, componentId: 'text', dataVersion: 1, structure: 'node', data: { title: 'Card ' + x, text: 'Editable content' }, geometry: { x, y: 150, width: 260, height: 200, rotation: 0 } });
|
||||
documents.set(root, createDocument(root, [{ id: root, parent: null }], [node('a', root, 80), node('b', root, 450), { id: 'edge', incarnation: 1, componentId: 'connection', dataVersion: 1, structure: 'connection', ownerCanvasId: root, source: { elementId: 'a' }, target: { elementId: 'b' }, data: {}, path: { type: 'bezier-chain', knots: [{ id: 'k', mode: 'corner', position: { x: 400, y: 250 }, in: { x: -20, y: 0 }, out: { x: 20, y: 0 } }] } }]));
|
||||
documents.set(other, createDocument(other, [{ id: other, parent: null }], []));
|
||||
@@ -39,6 +39,16 @@ const upstream = createServer(async (request, response) => {
|
||||
documents.set(write[1], next); versions.set(write[1], versions.get(write[1]) + 1);
|
||||
return reply(200, { protocol_version: 2, request_id: batch.request_id, status: 'applied', operation_ids: batch.operations.map(operation => operation.id), model: model(next, false) });
|
||||
}
|
||||
if(path.endsWith('/canvas/ai/models'))return reply(200,{models:[{id:'fixture-text',label:'Fixture text',available:true,input_modalities:['text'],output_modalities:['text']}]});
|
||||
if(path.endsWith('/canvas/ai/runs')&&request.method==='POST'){
|
||||
const input=JSON.parse(bytes.toString());assert.equal(request.headers['x-expected-account'],account);assert.equal(request.headers['idempotency-key'],input.request_id);assert.equal(request.headers['if-match'],'"'+versions.get(root)+'"');assert.deepEqual(input.expected_versions,{[root]:versions.get(root)});
|
||||
const previous=documents.get(root),element=previous.elements.get(input.node_id),next=applyChanges(previous,[{kind:'field',target:{id:element.id,incarnation:element.incarnation},field:'data.ai',before:element.data.ai??null,beforePresent:element.data.ai!==undefined,after:{enabled:true,modelId:input.model_id,prompt:input.prompt,parameters:input.parameters,propagate:input.propagate},afterPresent:true}]);documents.set(root,next);versions.set(root,versions.get(root)+1);
|
||||
const runId=id(700),plan={steps:[{step_id:'step-b',board_id:root,node_id:input.node_id,canonical_root:root,target_incarnation:input.target_incarnation,kind:'text',model_id:input.model_id,status:'queued',dependencies:[],static_inputs:[]}],edges:[],frozen_versions:{[root]:versions.get(root)}};
|
||||
const admitted={protocol_version:2,request_id:input.request_id,operation_id:input.operation_id,admission:'accepted',status:'queued',run_id:runId,plan,canvas_version:versions.get(root),document_format_version:1,target_incarnation:input.target_incarnation,model:model(next,false)};
|
||||
aiRuns.set(runId,{status:'queued',plan,admitted,events:[{seq:1,run_id:runId,type:'run.queued',status:'queued'},{seq:2,run_id:runId,type:'step.queued',step_id:'step-b',board_id:root,node_id:input.node_id,status:'queued'}]});return reply(202,admitted);
|
||||
}
|
||||
const aiPoll=path.match(/\/canvas\/ai\/runs\/([^/]+)$/);
|
||||
if(aiPoll){const run=aiRuns.get(aiPoll[1]),after=Number(url.searchParams.get('after_seq')||0),events=run.events.filter(event=>event.seq>after);return reply(200,{run_id:aiPoll[1],status:run.status,plan:run.plan,steps:[],events,next_seq:events.at(-1)?.seq??after,has_more:false,outputs:[]});}
|
||||
if (path.endsWith('/assets') && request.method === 'POST') {
|
||||
const upload = new Request('http://fixture', { method: 'POST', body: bytes, headers: { 'Content-Type': request.headers['content-type'] } }), form = await upload.formData(), file = form.get('image');
|
||||
const asset = { id: request.headers['x-resource-id'], bytes: file.size, mime_type: 'image/png', width: 1, height: 1, url: `https://api.foundry.entroia.com/projects/endoleap/workspaces/${workspace}/assets/${request.headers['x-resource-id']}` };
|
||||
@@ -135,6 +145,24 @@ try {
|
||||
assert.deepEqual(documents.get(root).elements.get('edge').path.knots[0].position,{x:440,y:265},JSON.stringify({inputs:await evaluate('window.inputEvents'),text:await evaluate('document.body.innerText'),writes:calls.filter(call=>call.method==='POST'&&call.path.endsWith('/model/operations')).map(call=>JSON.parse(call.bytes.toString()).operations)}));
|
||||
assert.equal(calls.filter(call => call.method === 'POST' && call.path.endsWith('/model/operations')).length,beforeCurve+1);
|
||||
console.log('PASS actual GraphPage chooses canonical editor; create/undo/redo, drag/resize/curve final coordinates, Escape and IME text history publish correctly');
|
||||
const selectB=async()=>{await call('Input.dispatchMouseEvent',{type:'mousePressed',x:1050,y:650,button:'left',buttons:1,clickCount:1});await call('Input.dispatchMouseEvent',{type:'mouseReleased',x:1050,y:650,button:'left',buttons:0,clickCount:1});await pause(50);const point=await evaluate(`(()=>{const r=document.querySelector('.react-flow__node[data-id=b]').getBoundingClientRect();return {x:r.x+r.width/2,y:r.y+r.height-30}})()`);await call('Input.dispatchMouseEvent',{type:'mousePressed',...point,button:'left',buttons:1,clickCount:1});await call('Input.dispatchMouseEvent',{type:'mouseReleased',...point,button:'left',buttons:0,clickCount:1});await until(`!document.querySelector('[aria-label="AI 创作"]').disabled`,'AI tool did not accept the selected element');};
|
||||
await selectB();await click('AI 创作');await until(`!!document.querySelector('[aria-label="AI 模型"] option[value="fixture-text"]')`,'AI catalog did not load through the host');
|
||||
await evaluate(`(()=>{const input=document.querySelector('[aria-label="AI 指令"]');Object.getOwnPropertyDescriptor(HTMLTextAreaElement.prototype,'value').set.call(input,'Summarize this content');input.dispatchEvent(new Event('input',{bubbles:true}));})()`);
|
||||
await until(`!document.querySelector('[aria-label="执行 AI"]').disabled`,'AI command did not become executable');await click('执行 AI');
|
||||
await until(`document.body.innerText.includes('已受理')`,'AI admission did not persist');assert.equal(aiRuns.size,1);
|
||||
await evaluate(`document.querySelector('a[href="/dashboard/${workspace}/${other}"]').click()`);await until(`location.pathname.endsWith('${other}')&&document.querySelectorAll('.react-flow__node').length===0`,'AI view detach failed');
|
||||
const run=aiRuns.get(id(700)),beforeAI=documents.get(root),beforeNode=beforeAI.elements.get('b');
|
||||
const generated=applyChanges(beforeAI,[{kind:'field',target:{id:'b',incarnation:1},field:'data.text',before:beforeNode.data.text,after:'Generated while another canvas was open',beforePresent:true,afterPresent:true}]);documents.set(root,generated);versions.set(root,versions.get(root)+1);run.status='succeeded';
|
||||
run.events.push({seq:3,run_id:id(700),type:'step.output',step_id:'step-b',board_id:root,node_id:'b',output:{output_id:id(701),kind:'text',text:'Generated while another canvas was open'}},{seq:4,run_id:id(700),type:'step.applied',step_id:'step-b',board_id:root,node_id:'b',canvas_version:versions.get(root),document_format_version:1,target_incarnation:1,model:model(generated,false)},{seq:5,run_id:id(700),type:'run.completed',status:'succeeded'});
|
||||
const pollsBefore=calls.filter(call=>call.path.includes('/ai/runs/')).length;
|
||||
for(let i=0;i<200&&calls.filter(call=>call.path.includes('/ai/runs/')).length===pollsBefore;i++)await pause(20);
|
||||
assert.ok(calls.filter(call=>call.path.includes('/ai/runs/')).length>pollsBefore,'view detach stopped persistent AI polling');await pause(200);
|
||||
assert.equal(await evaluate('document.querySelectorAll(".react-flow__node").length'),0,'AI output leaked into a different project view');
|
||||
await call('Page.reload');await until(`location.pathname.endsWith('${other}')&&!!document.querySelector('[data-headless-canvas]')`,'other project reload failed');
|
||||
await evaluate(`document.querySelector('a[href="/dashboard/${workspace}/${root}"]').click()`);await until(`!!document.querySelector('.react-flow__node[data-id=b]')&&document.querySelector('.react-flow__node[data-id=b] textarea')?.value==='Generated while another canvas was open'`,'AI output did not survive reload');
|
||||
await selectB();await click('AI 创作');await until(`document.body.innerText.includes('已更新当前元素')`,'durable RunRecord did not recover');await click('关闭 AI 工具');
|
||||
assert.equal(calls.filter(call=>call.method==='POST'&&call.path.endsWith('/ai/runs')).length,1,'AI recovery invoked generation twice');assert.equal(documents.get(root).elements.get('b').geometry.width,300);
|
||||
console.log('PASS actual AI panel submits through the durable ledger, polls after navigation, preserves geometry, and restores its RunRecord after reload without re-execution');
|
||||
await evaluate(`(()=>{const bytes=Uint8Array.from(atob('${png.toString('base64')}'),c=>c.charCodeAt(0));const transfer=new DataTransfer();transfer.items.add(new File([bytes],'pixel.png',{type:'image/png'}));const input=document.querySelector('[data-headless-canvas] input[type=file]');input.files=transfer.files;input.dispatchEvent(new Event('change',{bubbles:true}));})()`);
|
||||
await until(`document.querySelectorAll('.react-flow__node').length === 4 && [...document.querySelectorAll('[data-headless-canvas] img')].some(image=>image.getAttribute('src')?.startsWith('blob:'))`, 'pending image did not appear before upload');
|
||||
const screenshot=await call('Page.captureScreenshot',{format:'png'});await writeFile('/private/tmp/headless-canvas-integration.png',Buffer.from(screenshot.data,'base64'));
|
||||
|
||||
@@ -0,0 +1,169 @@
|
||||
import { test } from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import { createDocument, applyChanges, createRegistry, builtinComponents } from '../../src/lib/canvas-core/index.ts';
|
||||
import { openProjectEngine } from '../../src/lib/canvas-runtime/engine.ts';
|
||||
import { decodeRecovery } from '../../src/lib/canvas-runtime/recovery.ts';
|
||||
import { createFoundryPersistence, createFoundryRuns } from '../../src/lib/canvas-service/index.ts';
|
||||
import { createNetworkGate } from '../../src/lib/canvas-runtime/network.ts';
|
||||
|
||||
const id=n=>`10000000-0000-4000-8000-${String(n).padStart(12,'0')}`,root=id(1),child=id(2),workspace=id(3),account=id(4),target={id:'a',incarnation:1};
|
||||
const registry=createRegistry(builtinComponents()),node=(name,owner=root)=>({id:name,incarnation:1,structure:'node',ownerCanvasId:owner,componentId:'text',dataVersion:1,data:{title:name,text:'original'},geometry:{x:0,y:0,width:260,height:200,rotation:0}});
|
||||
const original=createDocument(root,[{id:root,parent:null},{id:child,parent:{canvasId:root,entryElementId:'f'}}],[node('a'),{...node('f'),componentId:'frame'}, {...node('future',child),componentId:'plugin.future',dataVersion:9,data:{opaque:{keep:true}}}]);
|
||||
const runInput={target,boardId:root,modelId:'text-model',prompt:'Generate',parameters:{},propagate:false};
|
||||
const defer=()=>{let resolve;const promise=new Promise(yes=>resolve=yes);return{promise,resolve};};
|
||||
const tick=async()=>{for(let i=0;i<100;i++)await Promise.resolve();};
|
||||
function clockFixture(){let now=1000,next=0;const timers=new Map();return{now:()=>now,setTimer(fn,delay){const id=++next;timers.set(id,{fn,due:now+delay});return id;},clearTimer(id){timers.delete(id);},async advance(ms){now+=ms;for(let count=0;count<100;count++){const ready=[...timers].find(([,value])=>value.due<=now);if(!ready)break;timers.delete(ready[0]);ready[1].fn();await tick();}await tick();},timers};}
|
||||
function storageFixture(){let image,owner;const storage={broken:false,saved:[],async claim(scope){assert.equal(owner,undefined);owner={scope,active:true,fence:1,async release(){owner.active=false;owner=undefined;}};return owner;},async recover(){return image&&structuredClone(image);},async commit(leader,batch){assert.equal(leader,owner);if(storage.broken)throw new Error('quota');image=structuredClone(batch.image);storage.saved.push(image);return{durableRootRevision:image.rootRevision};},async readFile(){},close(){},image:()=>structuredClone(image)};return storage;}
|
||||
function fixture(){
|
||||
let document=original,versions=new Map([[root,1],[child,7]]),next=100,runCounter=500,postGate,docGate;
|
||||
const storage=storageFixture(),clock=clockFixture(),calls=[],receipts=new Map(),runs=new Map(),errors=[];
|
||||
const chunk=board=>({documentFormatVersion:1,projectId:root,complete:false,removedCanvasIds:[],canvases:[document.canvases.get(board)],elements:[...document.elements.values()].filter(element=>element.ownerCanvasId===board),tombstones:Object.fromEntries(document.tombstones),versions:{[board]:versions.get(board)}});
|
||||
const transport=async request=>{
|
||||
calls.push(request);
|
||||
if(request.path.endsWith('/model/operations')){
|
||||
if(docGate)await docGate.promise;
|
||||
const input=JSON.parse(request.body);let body=receipts.get(request.requestId);
|
||||
if(!body){
|
||||
if(Object.entries(input.expected_versions).some(([board,version])=>versions.get(board)!==version))body={protocol_version:2,request_id:request.requestId,status:'not-applied',terminal:true,error:'version_conflict'};
|
||||
else{for(const operation of input.operations)document=applyChanges(document,operation.changes);for(const board of Object.keys(input.expected_versions))versions.set(board,versions.get(board)+1);body={protocol_version:2,request_id:request.requestId,status:'applied',operation_ids:input.operations.map(operation=>operation.id),model:chunk(root)};}
|
||||
receipts.set(request.requestId,structuredClone(body));
|
||||
}
|
||||
return{status:body.status==='applied'?200:412,body};
|
||||
}
|
||||
if(request.method==='POST'&&request.path.endsWith('/ai/runs')){
|
||||
const input=JSON.parse(request.body);let body=receipts.get(request.requestId);
|
||||
if(!body){
|
||||
assert.equal(request.expectedVersion,versions.get(root));assert.deepEqual(input.expected_versions,Object.fromEntries(versions));
|
||||
const element=document.elements.get(input.node_id),before=element.data.ai;
|
||||
document=applyChanges(document,[{kind:'field',target,field:'data.ai',before:before??null,beforePresent:before!==undefined,after:{enabled:true,modelId:input.model_id,prompt:input.prompt,parameters:input.parameters,propagate:input.propagate},afterPresent:true}]);versions.set(root,versions.get(root)+1);
|
||||
const runId=id(++runCounter),plan={steps:[{step_id:'step-a',board_id:root,node_id:'a',target_incarnation:1,canonical_root:root,kind:'text',model_id:'text-model',status:'queued',dependencies:[],static_inputs:[]}],edges:[],frozen_versions:Object.fromEntries(versions)};
|
||||
body={protocol_version:2,request_id:request.requestId,operation_id:input.operation_id,admission:'accepted',status:'queued',run_id:runId,plan,canvas_version:versions.get(root),target_incarnation:1,document_format_version:1,model:chunk(root)};
|
||||
receipts.set(request.requestId,structuredClone(body));runs.set(runId,{status:'queued',plan,events:[{seq:1,run_id:runId,type:'run.queued',status:'queued'}],outputs:[]});
|
||||
}
|
||||
if(postGate)await postGate.promise;
|
||||
if(f.loseAdmission)return{status:503,body:{error:'response_lost'}};
|
||||
return{status:202,body};
|
||||
}
|
||||
if(request.path.includes('/ai/operations/')){const receipt=receipts.get(request.path.split('/').at(-1));return receipt?{status:200,body:receipt}:{status:404,body:{protocol_version:2,request_id:request.path.split('/').at(-1),status:'unknown'}};}
|
||||
if(request.path.includes('/ai/runs/')){
|
||||
const url=new URL(request.path,'https://fixture.test'),runId=url.pathname.split('/').at(-1),run=runs.get(runId),after=Number(url.searchParams.get('after_seq'));
|
||||
if(f.pollGate)await f.pollGate.promise;
|
||||
const events=run.events.filter(event=>event.seq>after).slice(0,f.pageSize??100);
|
||||
return{status:200,body:{run_id:runId,status:run.status,plan:run.plan,events,next_seq:events.at(-1)?.seq??after,has_more:after+events.length<run.events.length,steps:[],outputs:run.outputs}};
|
||||
}
|
||||
throw new Error('Unexpected '+request.path);
|
||||
};
|
||||
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)});},
|
||||
finish(text='generated'){
|
||||
const [runId,run]=[...runs].at(-1),element=document.elements.get('a');
|
||||
document=applyChanges(document,[{kind:'field',target,field:'data.text',before:element.data.text,after:text,beforePresent:true,afterPresent:true}]);versions.set(root,versions.get(root)+1);
|
||||
run.status='succeeded';run.outputs=[{step_id:'step-a',output:{output_id:id(999),kind:'text'}}];
|
||||
run.events.push({seq:2,run_id:runId,type:'step.output',step_id:'step-a',board_id:root,node_id:'a',output:{output_id:id(999),kind:'text',text}},
|
||||
{seq:3,run_id:runId,type:'step.applied',step_id:'step-a',board_id:root,node_id:'a',canvas_version:versions.get(root),target_incarnation:1,document_format_version:1,model:structuredClone(chunk(root))},
|
||||
{seq:4,run_id:runId,type:'run.completed',status:'succeeded'});
|
||||
},server:()=>({document,versions}),
|
||||
};return f;
|
||||
}
|
||||
|
||||
test('AI admission orders prior and later edits, persists its job, and does not hold the writer for execution',async()=>{
|
||||
const f=fixture(),engine=await f.open(),gate=f.holdAdmission();const detach=engine.sessions.attach('view');
|
||||
engine.editor.dispatch({type:'element.data',target,patch:{title:'before click'}});
|
||||
const start=engine.startRun(runInput);assert.equal(start.status,'applied');
|
||||
const later=engine.editor.dispatch({type:'element.geometry',target,patch:{x:200}});
|
||||
assert.equal(engine.getSnapshot().document.elements.get('a').geometry.x,200);
|
||||
assert.ok(engine.getSnapshot().runtime.ledger.operations.get(later.operationIds[0]).intent.dependencies.includes(start.operationIds[0]));
|
||||
await tick();assert.equal(f.calls.filter(call=>call.method==='POST').length,2);
|
||||
const admission=f.calls.find(call=>call.path.endsWith('/ai/runs'));
|
||||
assert.equal(JSON.parse(admission.body).expected_versions[root],2);assert.equal(f.server().document.elements.get('a').data.title,'before click');
|
||||
assert.equal(f.server().document.elements.get('a').geometry.x,0);detach();gate.resolve();await engine.settle();
|
||||
const run=[...engine.getSnapshot().runtime.runs.values()][0];assert.ok(run);assert.equal(run.cursor,0);
|
||||
assert.equal(engine.getSnapshot().runtime.ledger.operations.get(start.operationIds[0]).phase,'resolved');
|
||||
assert.equal(f.server().document.elements.get('a').geometry.x,200);assert.equal(run.status,'queued');
|
||||
assert.equal(engine.startRun(runInput).code,'node_busy');
|
||||
f.finish();f.pageSize=2;await f.clock.advance(0);await engine.settle();
|
||||
assert.equal(engine.getSnapshot().runtime.runs.get(run.id).cursor,2);assert.equal(engine.getSnapshot().runtime.runs.get(run.id).hasMore,true);assert.equal(engine.startRun(runInput).code,'node_busy');
|
||||
await f.clock.advance(1);await engine.settle();
|
||||
const finished=engine.getSnapshot();assert.equal(finished.runtime.runs.get(run.id).status,'succeeded');assert.equal(finished.runtime.runs.get(run.id).cursor,4);
|
||||
assert.equal(finished.document.elements.get('a').data.text,'generated');assert.equal(finished.document.elements.get('a').geometry.x,200);assert.equal(finished.document.elements.get('future').data.opaque.keep,true);assert.equal(finished.history.undo.length,2);
|
||||
assert.equal(finished.operations.size,2);assert.equal(finished.runtime.ledger.operations.size,3);
|
||||
const restored=decodeRecovery(f.storage.image(),registry);assert.equal(restored.runtime.runs.get(run.id).cursor,4);assert.equal(restored.runtime.runs.get(run.id).outputEvents[0].output.text,'generated');
|
||||
await engine.close();
|
||||
});
|
||||
|
||||
test('lost AI admission response recovers its original request and run without a second invocation',async()=>{
|
||||
const f=fixture();f.loseAdmission=true;const engine=await f.open();engine.startRun(runInput);await engine.settle();
|
||||
assert.equal(engine.getSnapshot().runtime.runs.size,0);const frozen=[...engine.getSnapshot().runtime.ledger.requests.values()][0];await engine.close();
|
||||
f.loseAdmission=false;const recovered=await f.open();await f.clock.advance(5000);await recovered.settle();
|
||||
assert.equal(f.calls.filter(call=>call.method==='POST'&&call.path.endsWith('/ai/runs')).length,1);
|
||||
assert.ok(f.calls.some(call=>call.path.endsWith('/ai/operations/'+frozen.id)));
|
||||
assert.equal(recovered.getSnapshot().runtime.runs.size,1);assert.equal([...recovered.getSnapshot().runtime.ledger.requests.values()][0].body,frozen.body);
|
||||
f.finish();await f.clock.advance(1000);await recovered.settle();assert.equal(recovered.getSnapshot().document.elements.get('a').data.text,'generated');await recovered.close();
|
||||
});
|
||||
|
||||
test('run event waits for outstanding receipts and retains a content conflict, output and page across recovery',async()=>{
|
||||
const f=fixture(),engine=await f.open();engine.startRun(runInput);await engine.settle();
|
||||
const gate=f.holdDocument();engine.editor.dispatch({type:'element.data',target,patch:{text:'local edit'}});await tick();f.finish();
|
||||
await f.clock.advance(0);await tick();const run=[...engine.getSnapshot().runtime.runs.values()][0];
|
||||
assert.equal(run.cursor,0);assert.equal(run.error,'run_waiting_for_receipts');assert.ok(run.pendingPage);assert.equal(engine.getSnapshot().document.elements.get('a').data.text,'local edit');
|
||||
gate.resolve();await engine.settle();await f.clock.advance(1000);await engine.settle();
|
||||
const conflict=engine.getSnapshot().runtime.runs.get(run.id);assert.equal(conflict.cursor,0);assert.equal(conflict.blocked,true);assert.ok(conflict.pendingPage);assert.equal(conflict.outputs.length,1);assert.equal(conflict.outputEvents[0].output.text,'generated');
|
||||
assert.equal(engine.getSnapshot().document.elements.get('a').data.text,'local edit');await engine.close();
|
||||
const recovered=await f.open();assert.equal(recovered.getSnapshot().runtime.runs.get(run.id).pendingPage.next,4);assert.equal(recovered.getSnapshot().document.elements.get('a').data.text,'local edit');await recovered.close();
|
||||
});
|
||||
|
||||
test('older run event cannot rewind a later canonical document receipt',async()=>{
|
||||
const f=fixture(),engine=await f.open();engine.startRun(runInput);await engine.settle();f.finish();
|
||||
engine.editor.dispatch({type:'element.geometry',target,patch:{x:500}});await engine.settle();
|
||||
// The server rejects its stale version first. Adopt an externally refreshed full base only after receipt resolution.
|
||||
const server=f.server();const refreshed=engine.refresh(server.document,server.versions);assert.equal(refreshed.status,'applied');
|
||||
await f.clock.advance(0);await engine.settle();
|
||||
assert.equal(engine.getSnapshot().document.elements.get('a').geometry.x,500);assert.equal(engine.getSnapshot().document.elements.get('a').data.text,'generated');
|
||||
assert.equal([...engine.getSnapshot().runtime.runs.values()][0].cursor,4);await engine.close();
|
||||
});
|
||||
|
||||
test('run durability failure prevents admission and close fences late poll responses',async()=>{
|
||||
const f=fixture(),engine=await f.open();await engine.persist();f.storage.broken=true;engine.startRun(runInput);await tick();assert.equal(f.calls.length,0);
|
||||
f.storage.broken=false;await engine.retry();await f.clock.advance(0);await f.clock.advance(1000);await engine.settle();assert.equal(f.runs.size,1);
|
||||
f.pollGate=defer();await f.clock.advance(0);await tick();const revision=engine.getSnapshot().rootRevision;
|
||||
const closing=engine.close();f.finish();f.pollGate.resolve();await closing;await tick();
|
||||
assert.equal(engine.getSnapshot().rootRevision,revision);assert.equal(engine.getSnapshot().document.elements.get('a').data.text,'original');
|
||||
});
|
||||
|
||||
test('transport gate bounds combined run/read/write capacity and drops aborted waiters',async()=>{
|
||||
const gate=createNetworkGate(2),release=defer(),controller=new AbortController();let active=0,max=0;
|
||||
const execute=signal=>gate.run(signal,async()=>{max=Math.max(max,++active);await release.promise;active--;});
|
||||
const first=execute(new AbortController().signal),second=execute(new AbortController().signal),cancelled=execute(controller.signal);controller.abort();await assert.rejects(cancelled);
|
||||
const third=execute(new AbortController().signal);await tick();assert.equal(active,2);release.resolve();await Promise.all([first,second,third]);assert.equal(max,2);assert.equal(active,0);
|
||||
});
|
||||
|
||||
|
||||
test('malformed admission identity keeps its receipt and claims until the original query repairs it',async()=>{
|
||||
const f=fixture(),real=f.transport;
|
||||
const runs=createFoundryRuns({workspaceId:workspace,rootId:root,registry,transport:async request=>{const response=await real(request);return request.method==='POST'?{...response,body:{...response.body,target_incarnation:99}}:response;}});
|
||||
const engine=await openProjectEngine({scope:{accountId:account,workspaceId:workspace,projectId:root},registry,document:original,versions:new Map([[root,1],[child,7]]),storage:f.storage,clock:f.clock,createId:(()=>{let next=900;return()=>id(++next)})(),persistence:createFoundryPersistence({workspaceId:workspace,rootId:root,registry,transport:real}),runs});
|
||||
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);assert.ok(engine.getSnapshot().runtime.receipts.has(request.id));assert.equal(engine.getSnapshot().document.elements.get('a').data.ai,undefined);
|
||||
await engine.retry();await f.clock.advance(0);await engine.settle();assert.equal(engine.getSnapshot().runtime.runs.size,1);assert.equal(f.calls.filter(call=>call.method==='POST').length,1);await engine.close();
|
||||
});
|
||||
|
||||
test('foreign event identity and corrupted persisted run bindings cannot overwrite the working document',async()=>{
|
||||
const f=fixture(),engine=await f.open();engine.startRun(runInput);await engine.settle();f.finish();
|
||||
const remote=[...f.runs.values()][0];remote.events[2].target_incarnation=99;
|
||||
await f.clock.advance(0);await engine.settle();const run=[...engine.getSnapshot().runtime.runs.values()][0];
|
||||
assert.equal(run.blocked,true);assert.equal(run.cursor,0);assert.equal(engine.getSnapshot().document.elements.get('a').data.text,'original');
|
||||
const corrupted=f.storage.image();corrupted.payload.runtime.runs[0][1].target.incarnation=99;
|
||||
assert.throws(()=>decodeRecovery(corrupted,registry),/corrupt_recovery_image/);
|
||||
const wrongOwner=f.storage.image();wrongOwner.payload.runtime.ledger.requests[0][1].ownerCanvasId=child;
|
||||
assert.throws(()=>decodeRecovery(wrongOwner,registry),/corrupt_recovery_image/);await engine.close();
|
||||
});
|
||||
|
||||
|
||||
test('even a higher-version malformed event cannot restore an old incarnation',async()=>{
|
||||
const f=fixture(),engine=await f.open();engine.startRun(runInput);await engine.settle();f.finish();
|
||||
const server=f.server(),records=[...server.document.elements.values()].map(element=>element.id==='a'?{...element,incarnation:2,data:{...element.data,text:'restored content'}}:element),restored=createDocument(root,[...server.document.canvases.values()],records,new Map([['a',1]])),nextVersion=server.versions.get(root)+1;
|
||||
assert.equal(engine.refresh(restored,new Map([[root,nextVersion],[child,7]])).status,'applied');
|
||||
const event=[...f.runs.values()][0].events[2];event.canvas_version=nextVersion+1;event.model.versions[root]=nextVersion+1;
|
||||
await f.clock.advance(0);await engine.settle();assert.equal(engine.getSnapshot().document.elements.get('a').incarnation,2);assert.equal(engine.getSnapshot().document.elements.get('a').data.text,'restored content');assert.equal([...engine.getSnapshot().runtime.runs.values()][0].blocked,true);await engine.close();
|
||||
});
|
||||
Reference in New Issue
Block a user