diff --git a/src/components/workspace/canvas-runtime-host.tsx b/src/components/workspace/canvas-runtime-host.tsx index 791e0cb..bc2aeee 100644 --- a/src/components/workspace/canvas-runtime-host.tsx +++ b/src/components/workspace/canvas-runtime-host.tsx @@ -11,7 +11,7 @@ import { createWorkspaceRuntime } from "@/lib/canvas-runtime/workspace"; import { createBrowserDigests } from "@/lib/canvas-runtime/browser-digests"; import {createBrowserImageDecoder} from "@/lib/canvas-runtime/browser-images"; import { createBrowserStorage } from "@/lib/canvas-runtime/storage"; -import { createFoundryAssets, createFoundryPersistence, createFoundryTransport, createFoundryRuns } from "@/lib/canvas-service/index"; +import { createFoundryAssets, createFoundryPersistence, createFoundryTransport, createFoundryRuns,createFoundryModelLoader } from "@/lib/canvas-service/index"; const Context = createContext(null); const Renderers=createContext(defaultCanvasRenderers); @@ -20,6 +20,7 @@ export function CanvasRuntimeHost({ children,extensions=[] }: { children: ReactN const [{host,renderers}] = useState(() => { const registry = createRegistry([...builtinComponents(),...extensions.flatMap(extension=>extension.components??[])]),renderers=createReactRenderers([...defaultCanvasRenderers.definitions,...extensions.flatMap(extension=>extension.renderers??[])],registry); const host=createRuntimeHost({ registry, + modelLoader(accountId,workspaceId,_authEpoch,current){return createFoundryModelLoader({workspaceId,registry,transport:createFoundryTransport({accountId,isCurrentIdentity:current})});}, workspace(accountId, workspaceId, authEpoch, isCurrentIdentity) { return createWorkspaceRuntime({ accountId, workspaceId, authEpoch, isCurrentIdentity, registry, storage: createBrowserStorage(),digests:createBrowserDigests(),images:createBrowserImageDecoder(), adapters(rootId, current) { diff --git a/src/components/workspace/graph-page.tsx b/src/components/workspace/graph-page.tsx index 2db2f77..9250621 100644 --- a/src/components/workspace/graph-page.tsx +++ b/src/components/workspace/graph-page.tsx @@ -1,12 +1,11 @@ "use client"; -import { useEffect, useState } from "react"; +import { useEffect, useRef, useState } from "react"; import { useRouter, useSearchParams } from "next/navigation"; import { api, errorText } from "@/lib/api"; import { readDemoBoards } from "@/lib/demo-workspace"; import { matchesRef, type Board } from "@/lib/workspace"; import { loadCanvasEntry } from "@/lib/canvas-entry"; -import { createFoundryTransport } from "@/lib/canvas-service/foundry"; import type { ProjectEngine } from "@/lib/canvas-runtime/workspace"; import { useCanvasRuntimeHost, useCanvasRenderers } from "./canvas-runtime-host"; import type { CanvasSnapshot } from "@/lib/canvas-sync"; @@ -27,6 +26,7 @@ export default function GraphPage({ workspaceId, boardId }: { workspaceId: strin const [loaded, setLoaded] = useState(null); const [error, setError] = useState(""); const [retry, setRetry] = useState(0); + const navigationGeneration=useRef(0); useEffect(() => { setLoaded(null); setError(""); if (demo) { @@ -37,13 +37,14 @@ export default function GraphPage({ workspaceId, boardId }: { workspaceId: strin } let active = true; const controller = new AbortController(); - loadCanvasEntry({ request: api, workspaceRef: workspaceId, boardRef: boardId, registry: runtimeHost.registry, signal: controller.signal, - transport(accountId) { return createFoundryTransport({ accountId, isCurrentIdentity: runtimeHost.bindAccount(accountId) }); }, + const startingIdentity=runtimeHost.identity();let currentIdentity:(()=>boolean)|undefined; + loadCanvasEntry({ request: api, workspaceRef: workspaceId, boardRef: boardId, registry: runtimeHost.registry, signal: controller.signal,navigationGeneration:++navigationGeneration.current, + readModel:input=>{const identity=runtimeHost.identity();if(identity.authEpoch!==startingIdentity.authEpoch&&identity.accountId!==input.accountId)throw new Error('identity_changed');currentIdentity=runtimeHost.bindAccount(input.accountId);return runtimeHost.loadModel(input.accountId,input.workspaceId,input.canvasId,{signal:input.signal,navigationGeneration:input.navigationGeneration});}, }).then(async result => { - if (!active) return; + if (!active||currentIdentity&&!currentIdentity()) return; if (result.mode === 'legacy') { setLoaded({ board: result.board, snapshot: result.snapshot, route }); return; } const engine = await runtimeHost.workspace(result.profile.id, result.board.workspace_id).open(result.model.document, result.model.versions); - if (active) setLoaded({ board: result.board, engine, route }); + if (active&&(!currentIdentity||currentIdentity())) setLoaded({ board: result.board, engine, route }); }) .catch((error) => { if (active) { setError(errorText(error)); if (error.message === "invalid_session") router.replace("/login"); } }); return () => { active = false; controller.abort(); }; diff --git a/src/lib/canvas-entry.ts b/src/lib/canvas-entry.ts index 27e094d..b91b93a 100644 --- a/src/lib/canvas-entry.ts +++ b/src/lib/canvas-entry.ts @@ -4,17 +4,17 @@ import { decodeSnapshot } from './canvas-sync.ts'; import { CANVAS_ACCEPT } from './canvas-v2.ts'; import { matchesRef, type Board } from './workspace.ts'; import type { Registry } from './canvas-core/types.ts'; -import { loadFoundryModel, type ServiceTransport } from './canvas-service/foundry.ts'; +import { loadFoundryModel, type ModelLoad, type ServiceTransport } from './canvas-service/foundry.ts'; /** Read canonical first: a migrated project must never be opened through snapshot writers. */ -export async function loadCanvasEntry(options: { request: CanvasRequest; workspaceRef: string; boardRef: string; registry: Registry; transport(accountId: string): ServiceTransport; signal?: AbortSignal }) { +export async function loadCanvasEntry(options: { request: CanvasRequest; workspaceRef: string; boardRef: string; registry: Registry; transport?(accountId: string): ServiceTransport; readModel?(input:{accountId:string;workspaceId:string;canvasId:string;signal?:AbortSignal;navigationGeneration:number}):Promise;signal?: AbortSignal;navigationGeneration?:number }) { const path = `workspaces/${encodeURIComponent(options.workspaceRef)}/materials/${encodeURIComponent(options.boardRef)}`; const [board, profile] = await Promise.all([options.request(path, { signal: options.signal }), options.request('profile', { signal: options.signal })]); options.signal?.throwIfAborted(); if (board.kind !== 'board' || !matchesRef(board, options.boardRef) || !board.workspace_id || !profile.id) throw new Error('invalid_reference'); - const transport = options.transport(profile.id); + if(!options.readModel&&!options.transport)throw new Error('model_loader_unavailable'); try { - const model = await loadFoundryModel(transport, board.workspace_id, board.id, options.registry, options.signal); + const model = options.readModel?await options.readModel({accountId:profile.id,workspaceId:board.workspace_id,canvasId:board.id,signal:options.signal,navigationGeneration:options.navigationGeneration??0}):await loadFoundryModel(options.transport!(profile.id), board.workspace_id, board.id, options.registry, options.signal); options.signal?.throwIfAborted(); if (model.status === 'ready') return { mode: 'headless' as const, board, profile, model }; // Existing projects keep their current editor until the user explicitly migrates them. diff --git a/src/lib/canvas-runtime/contracts.ts b/src/lib/canvas-runtime/contracts.ts index 558ca2a..3ef6134 100644 --- a/src/lib/canvas-runtime/contracts.ts +++ b/src/lib/canvas-runtime/contracts.ts @@ -22,6 +22,7 @@ export type RuntimeState = { }; export type EngineSnapshot = Snapshot; export type CanonicalProposal = { readonly document: Document; readonly versions: ReadonlyMap }; +export type CanonicalReadResult = ({ readonly status: 'ready' } & CanonicalProposal) | { readonly status: 'migration-required'; readonly rootId: Id; readonly sourceVersions: ReadonlyMap }; /** A validated historical execution receipt; its old model is never published over a fresh read. */ export type AppliedModelProof = { readonly versions:ReadonlyMap; assertCanonical(proposal:CanonicalProposal):void }; export type PersistencePort = { diff --git a/src/lib/canvas-runtime/host.ts b/src/lib/canvas-runtime/host.ts index 2a76998..aa22ddd 100644 --- a/src/lib/canvas-runtime/host.ts +++ b/src/lib/canvas-runtime/host.ts @@ -1,17 +1,20 @@ import type { Id, Registry } from '../canvas-core/types.ts'; import { fault, readMap } from '../canvas-core/values.ts'; import type { WorkspaceRuntime } from './workspace.ts'; +import { createScopedLoader, type LoadPort, type ScopedLoader } from './loader.ts'; +import type { CanonicalReadResult } from './contracts.ts'; /** Host lifetime matches the signed-in application, independently of navigation. */ -export function createRuntimeHost(options: { registry: Registry; workspace(accountId: Id, workspaceId: Id, authEpoch: number, current: () => boolean): WorkspaceRuntime; onError?: (error: unknown) => void }) { +export function createRuntimeHost(options: { registry: Registry; workspace(accountId: Id, workspaceId: Id, authEpoch: number, current: () => boolean): WorkspaceRuntime; modelLoader?(accountId:Id,workspaceId:Id,authEpoch:number,current:()=>boolean):LoadPort; onError?: (error: unknown) => void }) { let accountId: Id | null = null, authEpoch = 0, disposed = false; const workspaces = new Map(), retired = new Set(); + const readers = new Map>(); const closeRetired = async () => { const results = await Promise.allSettled([...retired].map(async runtime => { await runtime.close(); retired.delete(runtime); })); const errors = results.flatMap(result => result.status === 'rejected' ? [result.reason] : []); if (errors.length) throw new AggregateError(errors, 'host_close_failed'); }; - function retire() { authEpoch++; accountId = null; for (const runtime of workspaces.values()) retired.add(runtime); workspaces.clear(); } + function retire() { authEpoch++; accountId = null; for (const reader of readers.values()) reader.close();readers.clear();for (const runtime of workspaces.values()) retired.add(runtime); workspaces.clear(); } function activate(account: Id) { if (disposed) throw fault('host_disposed'); if (accountId !== account) { @@ -24,6 +27,15 @@ export function createRuntimeHost(options: { registry: Registry; workspace(accou bindAccount(account: Id) { activate(account); const epoch = authEpoch; return () => !disposed && accountId === account && authEpoch === epoch; }, identity: () => Object.freeze({ accountId, authEpoch }), getWorkspaces: () => readMap(workspaces), + async loadModel(account:Id,workspace:Id,canvas:Id,settings:{signal?:AbortSignal;navigationGeneration?:number}={}) { + activate(account); + const readEpoch=authEpoch; + if(!options.modelLoader)throw fault('model_loader_unavailable'); + let reader=readers.get(workspace); + if(!reader){const epoch=authEpoch,current=()=>!disposed&&accountId===account&&authEpoch===epoch;reader=createScopedLoader({load:options.modelLoader(account,workspace,epoch,current),current:()=>current(),maxCacheEntries:0});readers.set(workspace,reader);} + const handle=reader.acquire({accountId:account,workspaceId:workspace,projectId:null,canvasId:canvas,viewKey:'canonical-project',format:'canonical-model-v1',queryFingerprint:'full-project',source:'foundry',authEpoch,navigationGeneration:settings.navigationGeneration??0},{signal:settings.signal}); + try{const snapshot=await handle.ready;if(disposed||accountId!==account||authEpoch!==readEpoch)throw fault('identity_changed');const last=snapshot.chunks.at(-1);if(!last||!last.complete)throw fault('incomplete_load');return last.value;}finally{handle.release();} + }, workspace(account: Id, workspace: Id) { activate(account); let runtime = workspaces.get(workspace); diff --git a/src/lib/canvas-runtime/loader.ts b/src/lib/canvas-runtime/loader.ts new file mode 100644 index 0000000..0deca4e --- /dev/null +++ b/src/lib/canvas-runtime/loader.ts @@ -0,0 +1,141 @@ +import type { Id } from '../canvas-core/types.ts'; +import { equal, fault, freeze } from '../canvas-core/values.ts'; + +export type LoadScope = { + readonly accountId: Id; readonly workspaceId: Id; readonly projectId: Id | null; readonly canvasId: Id; + readonly viewKey: string; readonly format: string; readonly queryFingerprint: string; readonly source: string; + readonly authEpoch: number; readonly navigationGeneration: number; +}; +export type LoadChunk = { + readonly queryKey: string; readonly projectId: Id; readonly snapshotToken: string; + readonly serverVersions: Readonly>; readonly permissionEpoch: number; + readonly sequence: number; readonly coverage: readonly string[]; readonly complete: boolean; + readonly byteLength: number; readonly value: T; +}; +export type LoadSnapshot = { + readonly scope: LoadScope; readonly generation: number; + readonly cacheStatus: 'bypassed' | 'miss' | 'invalid' | 'hit'; + readonly status: 'queued' | 'loading' | 'initial-ready' | 'partial' | 'complete' | 'failed' | 'failed-with-partial' | 'cancelled'; + readonly chunks: readonly LoadChunk[]; readonly coverage: readonly string[]; readonly byteLength: number; + readonly error?: { readonly code: string; readonly message: string }; +}; +export type LoadPort = (scope: LoadScope, signal: AbortSignal) => AsyncIterable>; +export type LoadHandle = { readonly ready: Promise>; getSnapshot(): LoadSnapshot; release(): void }; + +/** Navigation belongs to a subscriber, not the shared query identity. */ +export function loadQueryKey(scope: LoadScope): string { + if (!scope || ![scope.accountId, scope.workspaceId, scope.canvasId, scope.viewKey, scope.format, scope.queryFingerprint, scope.source].every(value => typeof value === 'string' && value.length > 0) + || scope.projectId !== null && (typeof scope.projectId !== 'string' || !scope.projectId) + || ![scope.authEpoch, scope.navigationGeneration].every(value => Number.isSafeInteger(value) && value >= 0)) throw fault('invalid_load_scope'); + return JSON.stringify([scope.accountId, scope.workspaceId, scope.projectId, scope.canvasId, scope.viewKey, scope.format, scope.queryFingerprint, scope.source, scope.authEpoch]); +} + +/** Read copies only. This coordinator never publishes a document or a server ACK. + * Adapters validate payloads and declare their measured byte cost before yielding. + */ +export function createScopedLoader(options: { + load: LoadPort; current(scope: LoadScope): boolean; concurrency?: number; maxQueued?: number; + maxResponseBytes?: number; maxChunks?: number; maxCacheEntries?: number; maxCacheBytes?: number; + cacheMaxAgeMs?: number; now?: () => number; + acceptCached?(scope: LoadScope, snapshot: LoadSnapshot): boolean; + onObserverError?(error: unknown): void; +}) { + type Subscriber = { readonly scope: LoadScope; active: boolean; cancelled?: LoadSnapshot; lastState?: LoadSnapshot; snapshot?: LoadSnapshot;resolve(value: LoadSnapshot): void; reject(error: unknown): void; cleanup(): void; notify?(snapshot: LoadSnapshot): void }; + type Job = { readonly key: string; readonly scope: LoadScope; readonly generation: number; readonly controller: AbortController; readonly subscribers: Set; state: LoadSnapshot; retired: boolean }; + const concurrency = options.concurrency ?? 4, maxQueued = options.maxQueued ?? 128, maxBytes = options.maxResponseBytes ?? 64 * 1024 * 1024, maxChunks = options.maxChunks ?? 4096; + const maxCacheEntries = options.maxCacheEntries ?? 16, maxCacheBytes = options.maxCacheBytes ?? 16 * 1024 * 1024, cacheMaxAge = options.cacheMaxAgeMs ?? 0; + for (const [name, value] of Object.entries({ concurrency, maxQueued, maxBytes, maxChunks, maxCacheEntries, maxCacheBytes, cacheMaxAge })) if (!Number.isSafeInteger(value) || value < (['maxCacheEntries', 'maxCacheBytes', 'cacheMaxAge'].includes(name) ? 0 : 1)) throw fault('invalid_loader_budget'); + const inflight = new Map(), queue = new Map(), cache = new Map; at: number }>(); + let active = 0, generation = 0, closed = false, cacheBytes = 0; + const now = () => { const value = (options.now ?? Date.now)(); if (!Number.isSafeInteger(value) || value < 0) throw fault('invalid_loader_clock'); return value; }; + const current = (scope: LoadScope) => { if (closed) throw fault('loader_closed'); if (!options.current(scope)) throw fault('identity_changed'); }; + const view = (job: Job, subscriber: Subscriber): LoadSnapshot => {if(subscriber.lastState!==job.state){subscriber.lastState=job.state;subscriber.snapshot=Object.freeze({...job.state,scope:subscriber.scope});}return subscriber.snapshot!;}; + const notifyOne = (job: Job, subscriber: Subscriber) => { if (subscriber.active) { try { subscriber.notify?.(view(job, subscriber)); } catch (error) { try { options.onObserverError?.(error); } catch { /* Observers cannot change scheduling. */ } } } }; + const notify = (job: Job) => { for (const subscriber of [...job.subscribers]) notifyOne(job, subscriber); }; + const removeCache = (key: string) => { const entry = cache.get(key); if (entry) { cacheBytes -= entry.state.byteLength; cache.delete(key); } }; + function retire(job: Job) { + if (job.retired) return; job.retired = true; job.controller.abort(); queue.delete(job.generation); + if (inflight.get(job.key) === job) inflight.delete(job.key); + } + function pump() { + while (!closed && active < concurrency && queue.size) { + const job = queue.values().next().value!; queue.delete(job.generation); + if (job.retired || !job.subscribers.size) continue; + active++; job.state = Object.freeze({ ...job.state, status: 'loading' }); notify(job); + void (async () => { + try { + current(job.scope); let terminal = false, projectId: Id | undefined, token: string | undefined, versions: Readonly> | undefined; + for await (const chunk of options.load(job.scope, job.controller.signal)) { + if (job.retired) return; current(job.scope); + if (!chunk || terminal || chunk.queryKey !== job.key || chunk.permissionEpoch !== job.scope.authEpoch || typeof chunk.projectId !== 'string' || !chunk.projectId + || job.scope.projectId !== null && chunk.projectId !== job.scope.projectId || typeof chunk.snapshotToken !== 'string' || !chunk.snapshotToken + || chunk.sequence !== job.state.chunks.length || typeof chunk.complete !== 'boolean' || !Number.isSafeInteger(chunk.byteLength) || chunk.byteLength < 0 + || !Array.isArray(chunk.coverage) || chunk.coverage.some(key => typeof key !== 'string' || !key) || new Set(chunk.coverage).size !== chunk.coverage.length + || !chunk.serverVersions || typeof chunk.serverVersions !== 'object' || Array.isArray(chunk.serverVersions) + || ![Object.prototype,null].includes(Object.getPrototypeOf(chunk.serverVersions)) + || Object.entries(chunk.serverVersions).some(([id, version]) => !id || !Number.isSafeInteger(version) || version < 1)) throw fault('invalid_load_chunk'); + const orderedVersions = Object.fromEntries(Object.entries(chunk.serverVersions).sort(([a], [b]) => a.localeCompare(b))); + if (projectId !== undefined && (chunk.projectId !== projectId || chunk.snapshotToken !== token || !equal(orderedVersions, versions))) throw fault('load_snapshot_changed'); + if (job.state.chunks.length >= maxChunks || chunk.byteLength > maxBytes - job.state.byteLength) throw fault('load_response_budget'); + projectId = chunk.projectId; token = chunk.snapshotToken; versions = orderedVersions; terminal = chunk.complete; + // Own metadata; the adapter supplies an immutable/owned payload value. + const owned = Object.freeze({ ...chunk, serverVersions: freeze(orderedVersions), coverage: freeze([...chunk.coverage]) }); + job.state = Object.freeze({ ...job.state, status: job.state.chunks.length ? 'partial' : 'initial-ready', chunks: Object.freeze([...job.state.chunks, owned]), + coverage: freeze([...new Set([...job.state.coverage, ...chunk.coverage])]), byteLength: job.state.byteLength + chunk.byteLength }); notify(job); + } + if (job.retired) return; current(job.scope); if (!terminal) throw fault('incomplete_load'); + job.state = Object.freeze({ ...job.state, status: 'complete' }); + if (maxCacheEntries && maxCacheBytes && cacheMaxAge && job.state.byteLength <= maxCacheBytes) { + removeCache(job.key); cache.set(job.key, { state: job.state, at: now() }); cacheBytes += job.state.byteLength; + while (cache.size > maxCacheEntries || cacheBytes > maxCacheBytes) removeCache(cache.keys().next().value!); + } + notify(job); for (const subscriber of job.subscribers) if (subscriber.active) { subscriber.cleanup(); subscriber.resolve(view(job, subscriber)); } + } catch (error) { + if (job.retired) return; + const failure = error as { code?: string; message?: string }; + job.state = Object.freeze({ ...job.state, status: job.state.chunks.length ? 'failed-with-partial' : 'failed', error: freeze({ code: typeof failure?.code === 'string' ? failure.code : 'load_failed', message: typeof failure?.message === 'string' ? failure.message : 'Load failed' }) }); + notify(job); for (const subscriber of job.subscribers) if (subscriber.active) { subscriber.cleanup(); subscriber.reject(error); } + } finally { + if (inflight.get(job.key) === job) inflight.delete(job.key); + active--; pump(); + } + })(); + } + } + return Object.freeze({ + acquire(scopeInput: LoadScope, settings: { signal?: AbortSignal; onUpdate?: (snapshot: LoadSnapshot) => void; cache?: 'fresh' | 'allow-validated' } = {}): LoadHandle { + if(!scopeInput)throw fault('invalid_load_scope'); + const scope:LoadScope=Object.freeze({accountId:scopeInput.accountId,workspaceId:scopeInput.workspaceId,projectId:scopeInput.projectId,canvasId:scopeInput.canvasId,viewKey:scopeInput.viewKey,format:scopeInput.format,queryFingerprint:scopeInput.queryFingerprint,source:scopeInput.source,authEpoch:scopeInput.authEpoch,navigationGeneration:scopeInput.navigationGeneration}); + const key = loadQueryKey(scope); current(scope); if (settings.signal?.aborted) throw fault('load_aborted'); + let cacheStatus: LoadSnapshot['cacheStatus'] = settings.cache === 'allow-validated' ? 'miss' : 'bypassed'; + let cached = cache.get(key); + if (cached) { + const age = now() - cached.at; + if (age < 0 || age > cacheMaxAge || settings.cache !== 'allow-validated' || !options.acceptCached || options.acceptCached(scope, cached.state) !== true) { if(settings.cache==='allow-validated')cacheStatus='invalid';removeCache(key); cached = undefined; } + else { cacheStatus='hit';cache.delete(key); cache.set(key, cached); } + } + current(scope); + if(settings.signal?.aborted)throw fault('load_aborted'); + let job = inflight.get(key); + if (!job) { + if (!cached && queue.size >= maxQueued + Math.max(0, concurrency - active)) throw fault('load_capacity'); + if (generation >= Number.MAX_SAFE_INTEGER) throw fault('load_generation_exhausted'); + job = { key, scope, generation: ++generation, controller: new AbortController(), subscribers: new Set(), retired: false, + state: cached ? Object.freeze({ ...cached.state, scope, generation,cacheStatus }) : Object.freeze({ scope, generation,cacheStatus,status: 'queued', chunks: Object.freeze([]), coverage: Object.freeze([]), byteLength: 0 }) }; + if (!cached) { inflight.set(key, job); queue.set(job.generation, job); } + } + const shared = job; let subscriber: Subscriber; + const ready = new Promise>((resolve, reject) => { subscriber = { scope, active: true, resolve, reject, notify: settings.onUpdate, cleanup: () => settings.signal?.removeEventListener('abort', release) }; }); + const release = () => { if (!subscriber.active) return; subscriber.cancelled=Object.freeze({...view(shared,subscriber),status:'cancelled'});subscriber.active = false; subscriber.cleanup(); shared.subscribers.delete(subscriber); subscriber.reject(fault('load_aborted')); if (!shared.subscribers.size) retire(shared); }; + shared.subscribers.add(subscriber!); settings.signal?.addEventListener('abort', release, { once: true }); + if (cached) { subscriber!.resolve(view(shared, subscriber!)); notifyOne(shared, subscriber!); } else { notifyOne(shared, subscriber!); pump(); } + return Object.freeze({ ready, getSnapshot: () => subscriber.cancelled??view(shared,subscriber), release }); + }, + close() { + if (closed) return; closed = true; + for (const job of [...inflight.values()]) { retire(job); for (const subscriber of job.subscribers) if (subscriber.active) { subscriber.cancelled=Object.freeze({...view(job,subscriber),status:'cancelled'});subscriber.active = false; subscriber.cleanup(); subscriber.reject(fault('loader_closed')); } job.subscribers.clear(); } + queue.clear(); cache.clear(); cacheBytes = 0; + }, + }); +} +export type ScopedLoader = ReturnType>; diff --git a/src/lib/canvas-service/foundry.ts b/src/lib/canvas-service/foundry.ts index d96a842..27e853c 100644 --- a/src/lib/canvas-service/foundry.ts +++ b/src/lib/canvas-service/foundry.ts @@ -1,7 +1,7 @@ import type { CanvasRecord, Document, ElementRecord, Id, Json, OperationIntent, Registry } from '../canvas-core/types.ts'; import { createDocument, validateDocument } from '../canvas-core/document.ts'; import { equal, fault, freeze, readMap, validJson } from '../canvas-core/values.ts'; -import type { AppliedModelProof, EngineSnapshot, PersistencePort } from '../canvas-runtime/contracts.ts'; +import type { AppliedModelProof, CanonicalReadResult, EngineSnapshot, PersistencePort } from '../canvas-runtime/contracts.ts'; import type { FrozenRequest } from '../canvas-runtime/ledger.ts'; import { projectBatch } from '../canvas-runtime/batch-projection.ts'; import type { OperationStatus } from '../canvas-runtime/scheduler.ts'; @@ -62,18 +62,20 @@ function modelChunk(value: unknown): ModelChunk { if (new Set(removed).size !== removed.length || removed.some(value => canvasIds.has(value) || !serverVersions.has(value)) || serverVersions.size !== canvasIds.size + removed.length) throw fault('invalid_canonical_response'); return freeze({ projectId, complete: raw.complete, canvases, elements, tombstones: readMap(tombstones), versions: serverVersions, removed }); } -export type ModelLoad = { readonly status: 'ready'; readonly document: Document; readonly versions: ReadonlyMap } | { readonly status: 'migration-required'; readonly rootId: Id; readonly sourceVersions: ReadonlyMap }; +export type ModelLoad = CanonicalReadResult; export async function loadFoundryModel(transport: ServiceTransport, workspaceId: Id, canvasId: Id, registry: Registry, signal?: AbortSignal): Promise { const response = await transport({ method: 'GET', path: pathFor(workspaceId, canvasId), signal }), body = record(response.body); if (response.status === 409 && body.error === 'canonical_migration_required') { if (body.document_format_version !== 1) throw fault('unsupported_document_format'); const rootId = id(body.root_id); if (!uuid.test(rootId)) throw fault('invalid_canonical_response'); - return freeze({ status: 'migration-required', rootId, sourceVersions: versions(body.source_versions) }); + const sourceVersions=versions(body.source_versions);if(!sourceVersions.has(canvasId)||!sourceVersions.has(rootId))throw fault('project_mismatch'); + return freeze({ status: 'migration-required', rootId, sourceVersions }); } if (response.status !== 200) throw fault(typeof body.error === 'string' ? body.error : 'model_load_failed'); const chunk = modelChunk(body.model); if (!chunk.complete) throw fault('incomplete_initial_model'); const document = createDocument(chunk.projectId, chunk.canvases, chunk.elements, chunk.tombstones); validateDocument(document, registry); + if(!document.canvases.has(canvasId))throw fault('project_mismatch'); return Object.freeze({ status: 'ready', document, versions: chunk.versions }); } diff --git a/src/lib/canvas-service/index.ts b/src/lib/canvas-service/index.ts index 9a64e7b..0b6f790 100644 --- a/src/lib/canvas-service/index.ts +++ b/src/lib/canvas-service/index.ts @@ -2,3 +2,4 @@ export * from './foundry.ts'; export * from './assets.ts'; export * from './runs.ts'; +export * from './model-loader.ts'; diff --git a/src/lib/canvas-service/model-loader.ts b/src/lib/canvas-service/model-loader.ts new file mode 100644 index 0000000..e0c960b --- /dev/null +++ b/src/lib/canvas-service/model-loader.ts @@ -0,0 +1,24 @@ +import type { Id, Registry } from '../canvas-core/types.ts'; +import { fault } from '../canvas-core/values.ts'; +import type { CanonicalReadResult } from '../canvas-runtime/contracts.ts'; +import { loadQueryKey, type LoadPort } from '../canvas-runtime/loader.ts'; +import { loadFoundryModel, type ServiceTransport } from './foundry.ts'; + +/** The current endpoint is complete-only. One validated chunk is intentional; + * this adapter does not claim to stream an already-buffered response progressively. + */ +export function createFoundryModelLoader(options: { workspaceId: Id; registry: Registry; transport: ServiceTransport }): LoadPort { + return async function* (scope, signal) { + if (scope.workspaceId !== options.workspaceId || scope.format !== 'canonical-model-v1' || scope.source !== 'foundry' || scope.viewKey !== 'canonical-project') throw fault('invalid_model_load_scope'); + const value = await loadFoundryModel(options.transport, options.workspaceId, scope.canvasId, options.registry, signal); + signal.throwIfAborted(); + const projectId = value.status === 'ready' ? value.document.projectId : value.rootId; + const versions = Object.fromEntries([...(value.status === 'ready' ? value.versions : value.sourceVersions)].sort(([a], [b]) => a.localeCompare(b))); + const token = JSON.stringify([projectId, versions]); + const payload = value.status === 'ready' ? { projectId, canvases: [...value.document.canvases.values()], elements: [...value.document.elements.values()], tombstones: [...value.document.tombstones], versions } : { projectId, versions, migrationRequired: true }; + // A serialized payload cost, not a claim about pre-parse network/heap limits. + const byteLength = new TextEncoder().encode(JSON.stringify(payload)).byteLength; + yield Object.freeze({ queryKey: loadQueryKey(scope), projectId, snapshotToken: token, serverVersions: Object.freeze(versions), permissionEpoch: scope.authEpoch, + sequence: 0, coverage: Object.freeze(value.status === 'ready' ? [...value.document.canvases.keys()].map(id => `canvas:${id}`) : [`migration:${scope.canvasId}`]), complete: true, byteLength, value }); + }; +} diff --git a/tests/headless/browser-storage-cases.mjs b/tests/headless/browser-storage-cases.mjs index e37a597..3715601 100644 --- a/tests/headless/browser-storage-cases.mjs +++ b/tests/headless/browser-storage-cases.mjs @@ -11,6 +11,28 @@ 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'; import { createResourceView } from '../../src/lib/canvas-bindings/resource-view.ts'; +import { createScopedLoader,loadQueryKey } from '../../src/lib/canvas-runtime/loader.ts'; + +async function scopedNetworkLoaderCase(assert) { + const scope={accountId:'loader-account',workspaceId:'loader-workspace',projectId:'loader-project',canvasId:'loader-canvas',viewKey:'canvas',format:'fixture-ndjson-v1',queryFingerprint:'all',source:'network-fixture',authEpoch:1,navigationGeneration:1}; + async function run(fail) { + const key=crypto.randomUUID();let signal,first;const initial=new Promise(resolve=>{first=resolve;}); + const loader=createScopedLoader({current:()=>true,async *load(query,abort){ + signal=abort;const response=await fetch(`/loader-stream?key=${key}`,{signal:abort}),reader=response.body.getReader(),decoder=new TextDecoder();let pending='',sequence=0; + try{while(true){const next=await reader.read();pending+=decoder.decode(next.value??new Uint8Array(),{stream:!next.done});let end;while((end=pending.indexOf('\n'))>=0){const line=pending.slice(0,end);pending=pending.slice(end+1);if(!line)continue;const value=JSON.parse(line);yield{queryKey:loadQueryKey(query),projectId:'loader-project',snapshotToken:key,serverVersions:{'loader-canvas':1},permissionEpoch:query.authEpoch,sequence:sequence++,coverage:[`part-${value.part}`],complete:value.complete,byteLength:new TextEncoder().encode(line).byteLength,value:Object.freeze(value)};}if(next.done)break;}}finally{await reader.cancel().catch(()=>{});} + }}); + const a=loader.acquire(scope,{onUpdate:state=>{if(state.status==='initial-ready')first();}}),b=loader.acquire({...scope,navigationGeneration:2}); + const rejected=fail?b.ready.then(()=>{throw Error('invalid tail became complete');},()=>{}):undefined; + try{ + await initial;assert(a.getSnapshot().chunks.length===1&&b.getSnapshot().status==='initial-ready','first network chunk waited for the tail'); + const cancelled=a.ready.then(()=>{throw Error('cancelled subscriber completed');},error=>assert(error.code==='load_aborted','subscriber cancellation changed'));a.release();await cancelled;assert(!signal.aborted,'one subscriber cancelled the shared fetch'); + const response=await(await fetch(`/loader-release?key=${key}${fail?'&fail=1':''}`)).json();assert(response.reads===1,'subscribers fetched the same query twice'); + if(fail){await rejected;assert(b.getSnapshot().status==='failed-with-partial'&&b.getSnapshot().chunks.length===1,'invalid network tail discarded or completed the prefix');} + else{const result=await b.ready;assert(result.status==='complete'&&result.chunks.length===2&&result.scope.navigationGeneration===2,'stream completion lost subscriber scope or chunks');assert(a.getSnapshot().status==='cancelled'&&a.getSnapshot().chunks.length===1,'cancelled subscriber observed its peer tail');} + }finally{a.release();b.release();loader.close();} + } + await run(false);await run(true); +} async function sharedDisplayReadersCase(engine, storage, assert) { const resource = engine.getSnapshot().document.elements.get('picture').data.assetId, read = storage.readFile, createURL = URL.createObjectURL, revokeURL = URL.revokeObjectURL; @@ -727,6 +749,8 @@ export async function run() { results.push('scoped paged inventory finds legacy orphans without hiding references; orphan v5 deletion rolls back faults, retries, isolates projects and forbids byte identity reuse'); await resourceRecoveryCase(name, assert); results.push('staged file and image recover from real IndexedDB; lost upload ACK queries stable resource identity'); + await scopedNetworkLoaderCase(assert); + results.push('scoped loader shares one real streamed fetch, exposes the first chunk before tail release, keeps the other subscriber alive after cancellation and preserves partial data after a malformed tail'); await defaultImageSizeCase(name, assert); results.push('real image Worker enriches creation history in recovery v6; IndexedDB reopening queries a lost size ACK; concurrent views share one read/URL, retain leases incrementally and release only the final display pin before sized undo/redo'); await imageDecodeWorkerCase(assert); diff --git a/tests/headless/browser-storage.mjs b/tests/headless/browser-storage.mjs index 1c0ee73..4061dcf 100644 --- a/tests/headless/browser-storage.mjs +++ b/tests/headless/browser-storage.mjs @@ -8,9 +8,22 @@ import { stripTypeScriptTypes } from 'node:module'; import { setTimeout as pause } from 'node:timers/promises'; const root = resolve(import.meta.dirname, '../..'), profile = await mkdtemp(join(tmpdir(), 'endoleap-storage-browser-')); +const readStreams=new Map(); const server = createServer(async (request, response) => { try { const path = new URL(request.url, 'http://localhost').pathname; + if(path==='/loader-stream'){ + const url=new URL(request.url,'http://localhost'),key=url.searchParams.get('key'); + if(!key){response.writeHead(400).end();return;} + const state=readStreams.get(key)??{reads:0};state.reads++;state.response=response;readStreams.set(key,state); + response.writeHead(200,{'Content-Type':'application/x-ndjson'});response.write(JSON.stringify({part:0,complete:false})+'\n');return; + } + if(path==='/loader-release'){ + const url=new URL(request.url,'http://localhost'),state=readStreams.get(url.searchParams.get('key')); + if(!state){response.writeHead(404).end();return;} + if(!state.response.destroyed)state.response.end(url.searchParams.get('fail')==='1'?'{invalid tail}\n':JSON.stringify({part:1,complete:true})+'\n'); + response.setHeader('Content-Type','application/json');response.end(JSON.stringify({reads:state.reads}));return; + } if (path === '/') { response.setHeader('Content-Type', 'text/html'); response.end('Headless storage checks'); return; } const filename = resolve(root, `.${path}`); if (!filename.startsWith(`${root}/src/lib/`) && !filename.startsWith(`${root}/tests/headless/`)) { response.writeHead(404).end(); return; } diff --git a/tests/headless/loader.test.mjs b/tests/headless/loader.test.mjs new file mode 100644 index 0000000..8f79217 --- /dev/null +++ b/tests/headless/loader.test.mjs @@ -0,0 +1,102 @@ +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import { createScopedLoader, loadQueryKey } from '../../src/lib/canvas-runtime/loader.ts'; +const scope = { accountId: 'account', workspaceId: 'workspace', projectId: 'project', canvasId: 'canvas', viewKey: 'canvas', format: 'v1', queryFingerprint: 'all', source: 'fixture', authEpoch: 1, navigationGeneration: 1 }; +const deferred = () => { let resolve, reject; const promise = new Promise((yes, no) => { resolve = yes; reject = no; }); return { promise, resolve, reject }; }; +const tick = () => new Promise(resolve => setImmediate(resolve)); +const chunk = (query, sequence = 0, complete = true, extra = {}) => ({ queryKey: loadQueryKey(query), projectId: 'project', snapshotToken: 'snapshot', serverVersions: { canvas: 3 }, permissionEpoch: query.authEpoch, sequence, coverage: [`part-${sequence}`], complete, byteLength: 2, value: Object.freeze({ part: sequence }), ...extra }); + +test('subscribers share a progressive read while retaining navigation scopes; cancelling one preserves the other', async () => { + const gate = deferred(), updates = []; let calls = 0, signal; + const loader = createScopedLoader({ current: () => true, async *load(query, abort) { calls++; signal = abort; yield chunk(query, 0, false); await gate.promise; yield chunk(query, 1); } }); + const a = loader.acquire(scope, { onUpdate: value => updates.push(value.status) }); await tick(); + assert.equal(a.getSnapshot().status, 'initial-ready'); assert.equal(a.getSnapshot().chunks.length, 1); + assert.strictEqual(a.getSnapshot(),a.getSnapshot()); + const b = loader.acquire({ ...scope, navigationGeneration: 2 }); assert.equal(b.getSnapshot().scope.navigationGeneration, 2); + const cancelled = assert.rejects(a.ready, { code: 'load_aborted' }); a.release(); await cancelled;const stopped=a.getSnapshot(); assert.equal(signal.aborted, false); + gate.resolve(); const result = await b.ready; assert.equal(calls, 1); assert.equal(result.status, 'complete'); assert.deepEqual(result.coverage, ['part-0', 'part-1']); + assert.equal(a.getSnapshot().status,'cancelled');assert.equal(a.getSnapshot().chunks.length,1);assert.deepEqual(a.getSnapshot().coverage,['part-0']); + assert.strictEqual(a.getSnapshot(),stopped);assert.strictEqual(b.getSnapshot(),result); + assert.deepEqual(updates, ['queued', 'loading', 'initial-ready']); b.release(); loader.close(); +}); +test('tail failure keeps the accepted prefix and a fresh retry replaces the failed inflight entry', async () => { + let calls = 0; + const loader = createScopedLoader({ current: () => true, async *load(query) { calls++; yield chunk(query, 0, false); if (calls === 1) throw Object.assign(Error('tail_failed'), { code: 'tail_failed' }); yield chunk(query, 1); } }); + const a = loader.acquire(scope); await assert.rejects(a.ready, { code: 'tail_failed' }); assert.equal(a.getSnapshot().status, 'failed-with-partial'); assert.equal(a.getSnapshot().chunks[0].value.part, 0); + const b = loader.acquire(scope); assert.equal((await b.ready).chunks.length, 2); assert.equal(calls, 2); a.release(); b.release(); loader.close(); +}); +test('a complete marker is not authoritative before iterator EOF; a checksum/trailing failure retains partial status', async () => { + const statuses = [], loader = createScopedLoader({ current: () => true, async *load(query) { yield chunk(query); throw Error('trailing_checksum'); } }); + const handle = loader.acquire(scope, { onUpdate: value => statuses.push(value.status) }); await assert.rejects(handle.ready, /trailing_checksum/); + assert.equal(handle.getSnapshot().status, 'failed-with-partial'); assert.equal(statuses.includes('complete'), false); handle.release(); loader.close(); +}); +test('snapshot token, version vector and project changes reject the tail without mixing it into accepted coverage', async () => { + for (const extra of [{ snapshotToken: 'other' }, { serverVersions: { canvas: 4 } }, { projectId: 'other' }]) { + const loader = createScopedLoader({ current: () => true, async *load(query) { yield chunk(query, 0, false); yield chunk(query, 1, true, extra); } }); + const handle = loader.acquire(scope); await assert.rejects(handle.ready); assert.equal(handle.getSnapshot().status, 'failed-with-partial'); assert.equal(handle.getSnapshot().chunks.length, 1); assert.deepEqual(handle.getSnapshot().coverage, ['part-0']); handle.release(); loader.close(); + } +}); +test('query, permission, sequence and malformed coverage headers reject before exposing a chunk', async () => { + for (const extra of [{ queryKey: 'foreign' }, { permissionEpoch: 2 }, { sequence: 3 }, { coverage: ['x', 'x'] }, { byteLength: -1 }, { serverVersions: { canvas: 0 } }]) { + const loader = createScopedLoader({ current: () => true, async *load(query) { yield chunk(query, 0, true, extra); } }); + const handle = loader.acquire(scope); await assert.rejects(handle.ready, { code: 'invalid_load_chunk' }); assert.equal(handle.getSnapshot().chunks.length, 0); assert.equal(handle.getSnapshot().status, 'failed'); handle.release(); loader.close(); + } +}); +test('an incomplete EOF or extra chunk after a complete marker cannot become a complete result', async () => { + for (const extra of [false, true]) { + const loader = createScopedLoader({ current: () => true, async *load(query) { yield chunk(query, 0, extra); if (extra) yield chunk(query, 1); } }); + const handle = loader.acquire(scope); await assert.rejects(handle.ready); assert.equal(handle.getSnapshot().status, 'failed-with-partial'); handle.release(); loader.close(); + } +}); +test('response byte and chunk budgets bound retained read copies while preserving earlier chunks', async () => { + for (const limits of [{ maxResponseBytes: 3 }, { maxChunks: 1 }]) { + const loader = createScopedLoader({ ...limits, current: () => true, async *load(query) { yield chunk(query, 0, false); yield chunk(query, 1); } }); + const handle = loader.acquire(scope); await assert.rejects(handle.ready, { code: 'load_response_budget' }); assert.equal(handle.getSnapshot().byteLength, 2); assert.equal(handle.getSnapshot().chunks.length, 1); handle.release(); loader.close(); + } +}); +test('cancelled active work retains its concurrency slot; its late cleanup cannot delete the newer generation', async () => { + const first = deferred(), second = deferred(), entered = []; let calls = 0; + const loader = createScopedLoader({ concurrency: 1, maxQueued: 1, current: () => true, async *load(query) { const index = ++calls; entered.push(index); await (index === 1 ? first.promise : second.promise); yield chunk(query); } }); + const a = loader.acquire(scope); await tick(); const cancelled = assert.rejects(a.ready, { code: 'load_aborted' }); a.release(); await cancelled; + const b = loader.acquire(scope); await tick(); assert.equal(calls, 1); assert.equal(b.getSnapshot().status, 'queued'); + assert.throws(() => loader.acquire({ ...scope, canvasId: 'third' }), { code: 'load_capacity' }); + first.resolve(); await tick(); assert.equal(calls, 2); const peer = loader.acquire(scope); assert.equal(peer.getSnapshot().generation, b.getSnapshot().generation); + second.resolve(); await Promise.all([b.ready, peer.ready]); assert.deepEqual(entered, [1, 2]); b.release(); peer.release(); loader.close(); +}); +test('queued cancellation frees admission capacity and never executes the abandoned source', async () => { + const gate = deferred(), called = [], loader = createScopedLoader({ concurrency: 1, maxQueued: 1, current: () => true, async *load(query) { called.push(query.canvasId); await gate.promise; yield chunk(query); } }); + const a = loader.acquire(scope), b = loader.acquire({ ...scope, canvasId: 'cancelled' }); const cancelled = assert.rejects(b.ready, { code: 'load_aborted' }); b.release(); await cancelled; + const c = loader.acquire({ ...scope, canvasId: 'next' }); gate.resolve(); await Promise.all([a.ready, c.ready]); assert.deepEqual(called, ['canvas', 'next']); a.release(); c.release(); loader.close(); +}); +test('read cache requires explicit source validation and is bounded by byte LRU and permission age', async () => { + let time = 100, accepted = true; const calls = new Map(); + const loader = createScopedLoader({ maxCacheEntries: 10, maxCacheBytes: 4, cacheMaxAgeMs: 100, now: () => time, acceptCached: () => accepted, current: () => true, + async *load(query) { calls.set(query.canvasId, (calls.get(query.canvasId) ?? 0) + 1); yield chunk(query); } }); + const read = async (canvasId, cached = false) => { const handle = loader.acquire({ ...scope, canvasId }, { cache: cached ? 'allow-validated' : 'fresh' }); await handle.ready; handle.release(); }; + await read('a'); await read('b'); await read('a', true); assert.equal(calls.get('a'), 1); await read('c'); await read('b', true); assert.equal(calls.get('b'), 2); + accepted = false; await read('b', true); assert.equal(calls.get('b'), 3); accepted = true; time = 201; await read('b', true); assert.equal(calls.get('b'), 4); + time = 0; await read('b', true); assert.equal(calls.get('b'), 5); loader.close(); +}); +test('different query fingerprints and permission epochs do not share requests', async () => { + let calls = 0; const loader = createScopedLoader({ current: () => true, async *load(query) { calls++; yield chunk(query); } }); + const handles = [scope, { ...scope, queryFingerprint: 'filtered' }, { ...scope, authEpoch: 2 }].map(query => loader.acquire(query)); + await Promise.all(handles.map(handle => handle.ready)); assert.equal(calls, 3); handles.forEach(handle => handle.release()); loader.close(); +}); +test('scope capture owns only declared metadata without freezing unrelated caller state', async () => { + const extra={filter:'mutable'},input={...scope,extra};const loader=createScopedLoader({current:()=>true,async *load(query){yield chunk(query);}}); + const handle=loader.acquire(input);assert.equal(Object.isFrozen(extra),false);input.canvasId='later';extra.filter='changed'; + const result=await handle.ready;assert.equal(result.scope.canvasId,'canvas');assert.equal(Object.hasOwn(result.scope,'extra'),false);assert.equal(Object.isFrozen(result.scope),true);handle.release();loader.close(); +}); +test('identity changes discard late chunks, and close promptly rejects waiting readers even if the source ignores abort', async () => { + const gate = deferred(); let current = true; + const loader = createScopedLoader({ current: () => current, async *load(query) { await gate.promise; yield chunk(query); } }); + const a = loader.acquire(scope); const rejected = assert.rejects(a.ready, { code: 'identity_changed' }); current = false; gate.resolve(); await rejected; assert.equal(a.getSnapshot().chunks.length, 0); a.release(); loader.close(); + const pending = deferred(), closed = createScopedLoader({ current: () => true, async *load(query) { await pending.promise; yield chunk(query); } }); + const b = closed.acquire(scope); const stopped = assert.rejects(b.ready, { code: 'loader_closed' }); closed.close(); await stopped; pending.resolve(); await tick(); assert.equal(b.getSnapshot().status, 'cancelled'); assert.throws(() => closed.acquire(scope), { code: 'loader_closed' }); +}); +test('observer reentrancy does not re-notify old subscribers or allow observer exceptions to fail a read', async () => { + const errors = []; let peer, updates = 0; + const loader = createScopedLoader({ current: () => true, onObserverError: error => errors.push(error), async *load(query) { yield chunk(query); } }); + const a = loader.acquire(scope, { onUpdate: state => { updates++; if (state.status === 'initial-ready' && !peer) peer = loader.acquire(scope, { onUpdate: () => { throw Error('observer'); } }); } }); + await a.ready; await peer.ready; assert.equal(updates, 4); assert.ok(errors.length > 0); a.release(); peer.release(); loader.close(); +}); diff --git a/tests/headless/model-loader.test.mjs b/tests/headless/model-loader.test.mjs new file mode 100644 index 0000000..833f898 --- /dev/null +++ b/tests/headless/model-loader.test.mjs @@ -0,0 +1,58 @@ +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import { createRegistry, builtinComponents } from '../../src/lib/canvas-core/index.ts'; +import { createRuntimeHost } from '../../src/lib/canvas-runtime/host.ts'; +import { loadQueryKey } from '../../src/lib/canvas-runtime/loader.ts'; +import { createFoundryModelLoader } from '../../src/lib/canvas-service/model-loader.ts'; +import { loadCanvasEntry } from '../../src/lib/canvas-entry.ts'; +const id = n => `10000000-0000-4000-8000-${String(n).padStart(12, '0')}`; +const account = id(1), workspace = id(2), root = id(3), registry = createRegistry(builtinComponents()); +const scope = { accountId: account, workspaceId: workspace, projectId: null, canvasId: root, authEpoch: 1, navigationGeneration: 7, format: 'canonical-model-v1', source: 'foundry', viewKey: 'canonical-project', queryFingerprint: 'full-project' }; +const deferred = () => { let resolve; const promise = new Promise(yes => { resolve = yes; }); return { promise, resolve }; }; +const tick = () => new Promise(resolve => setImmediate(resolve)); +const reply = (version = 1, title = account) => ({ status: 200, body: { model: { documentFormatVersion: 1, projectId: root, complete: true, canvases: [{ id: root, parent: null }], elements: [{ id: 'node', incarnation: 1, ownerCanvasId: root, componentId: 'text', dataVersion: 1, structure: 'node', data: { title, text: '' }, geometry: { x: 0, y: 0, width: 100, height: 80, rotation: 0 } }], tombstones: {}, versions: { [root]: version }, removedCanvasIds: [] } } }); + +test('Foundry adapter emits one validated complete chunk with bound query, project, coverage and version metadata', async () => { + const abort = new AbortController(), calls = []; + const port = createFoundryModelLoader({ workspaceId: workspace, registry, transport: async input => { calls.push(input); return reply(3); } }); + const chunks = []; for await (const chunk of port(scope, abort.signal)) chunks.push(chunk); + assert.equal(chunks.length, 1); assert.equal(chunks[0].complete, true); assert.equal(chunks[0].queryKey, loadQueryKey(scope)); assert.equal(chunks[0].projectId, root); + assert.deepEqual(chunks[0].coverage, [`canvas:${root}`]); assert.equal(chunks[0].serverVersions[root], 3); assert.ok(chunks[0].byteLength > 0); assert.equal(chunks[0].value.document.elements.get('node').data.title, account); + assert.strictEqual(calls[0].signal, abort.signal); assert.equal(calls[0].method, 'GET'); +}); +test('partial canonical replies, foreign requested canvases and unbound migration headers cannot masquerade as initial readiness', async () => { + for (const [response, expected] of [[{ ...reply(), body: { model: { ...reply().body.model, complete: false } } }, 'incomplete_initial_model'], [reply(), 'project_mismatch'], [{ status: 409, body: { error: 'canonical_migration_required', document_format_version: 1, root_id: root, source_versions: { [root]: 1 } } }, 'project_mismatch']]) { + const port = createFoundryModelLoader({ workspaceId: workspace, registry, transport: async () => response }), query = expected === 'incomplete_initial_model' ? scope : { ...scope, canvasId: id(99) }; + await assert.rejects(async () => { for await (const _ of port(query, new AbortController().signal)) assert.fail('invalid response yielded a chunk'); }, { code: expected }); + } +}); +test('Host shares concurrent model reads per identity/workspace but never reuses an idle authoritative snapshot', async () => { + const gate = deferred(), calls = []; let factories = 0, version = 1; + const host = createRuntimeHost({ registry, workspace() { assert.fail('readonly load opened a project/storage runtime'); }, modelLoader(accountId, workspaceId, _epoch, current) { + factories++; assert.equal(accountId, account); return createFoundryModelLoader({ workspaceId, registry, transport: async input => { calls.push(input); if (calls.length === 1) await gate.promise; assert.equal(current(), true); return reply(version); } }); + } }); + try { + const abort = new AbortController(), a = host.loadModel(account, workspace, root, { signal: abort.signal, navigationGeneration: 1 }), b = host.loadModel(account, workspace, root, { navigationGeneration: 2 }); + await tick(); const rejected = assert.rejects(a, { code: 'load_aborted' }); abort.abort(); await rejected; assert.equal(calls[0].signal.aborted, false); + gate.resolve(); assert.equal((await b).versions.get(root), 1); assert.equal(calls.length, 1); assert.equal(factories, 1); + version = 2; assert.equal((await host.loadModel(account, workspace, root)).versions.get(root), 2); assert.equal(calls.length, 2); assert.equal(host.getWorkspaces().size, 0); + } finally { gate.resolve(); await host.close(); } +}); +test('Host retirement aborts the old read and a late response cannot pollute the new account reader', async () => { + const gate = deferred(), calls = []; + const host = createRuntimeHost({ registry, workspace() { assert.fail('unexpected project open'); }, modelLoader(accountId, workspaceId) { + return createFoundryModelLoader({ workspaceId, registry, transport: async input => { calls.push({ accountId, input }); if (accountId === account) await gate.promise; return reply(1, accountId); } }); + } }); + try { + const old = host.loadModel(account, workspace, root); await tick(); const rejected = assert.rejects(old, { code: 'loader_closed' }); await host.retireIdentity(); await rejected; + const next = await host.loadModel(id(8), workspace, root); assert.equal(next.document.elements.get('node').data.title, id(8)); + gate.resolve(); await tick(); assert.equal(host.identity().accountId, id(8)); assert.equal(calls[0].input.signal.aborted, true); assert.equal(calls.length, 2); + } finally { gate.resolve(); await host.close(); } +}); +test('entry injection uses resolved IDs and navigation context while preserving canonical-first legacy policy', async () => { + const board = { id: root, workspace_id: workspace, slug: 'board', kind: 'board', title: 'Board', version: 1 }, inputs = [], reads = []; + const options = { workspaceRef: 'workspace-slug', boardRef: 'board', registry, navigationGeneration: 9, request: async path => { reads.push(path); return path === 'profile' ? { id: account } : board; }, + readModel: async input => { inputs.push(input); const port = createFoundryModelLoader({ workspaceId: workspace, registry, transport: async () => reply() }); for await (const chunk of port(scope, new AbortController().signal)) return chunk.value; }, transport() { assert.fail('entry bypassed the installed loader'); } }; + const result = await loadCanvasEntry(options); assert.equal(result.mode, 'headless'); assert.deepEqual(inputs[0], { accountId: account, workspaceId: workspace, canvasId: root, signal: undefined, navigationGeneration: 9 }); assert.equal(reads.some(path => path.endsWith('/canvas')), false); + await assert.rejects(() => loadCanvasEntry({ ...options, readModel: async () => { throw Object.assign(Error('forbidden'), { code: 'forbidden' }); } }), { code: 'forbidden' }); +});