From a92ae72c8b1ea9c92a7ccc4ea2ec6e48eef44737 Mon Sep 17 00:00:00 2001 From: Codex Bot Date: Mon, 9 Nov 2026 02:06:07 +0800 Subject: [PATCH] Negotiate and decode snapshot-bound canonical model streams --- src/app/api/foundry/[...path]/route.ts | 4 +- src/lib/canvas-runtime/contracts.ts | 4 +- src/lib/canvas-runtime/host.ts | 8 ++-- src/lib/canvas-service/foundry.ts | 18 +++++-- src/lib/canvas-service/model-loader.ts | 17 ++++--- src/lib/canvas-service/model-stream.ts | 62 +++++++++++++++++++++++++ tests/headless/model-stream.test.mjs | 56 ++++++++++++++++++++++ tests/headless/proxy.integration.mjs | 8 ++++ tests/headless/renderer.integration.mjs | 10 ++++ 9 files changed, 172 insertions(+), 15 deletions(-) create mode 100644 src/lib/canvas-service/model-stream.ts create mode 100644 tests/headless/model-stream.test.mjs diff --git a/src/app/api/foundry/[...path]/route.ts b/src/app/api/foundry/[...path]/route.ts index f3e7511..7af1d3d 100644 --- a/src/app/api/foundry/[...path]/route.ts +++ b/src/app/api/foundry/[...path]/route.ts @@ -72,6 +72,8 @@ async function handler( const project = process.env.NEXT_PUBLIC_FOUNDRY_PROJECT || "endoleap"; const headers: Record = { "Content-Type": asset ? (req.headers.get("content-type") || "application/octet-stream") : "application/json" }; if (canvas) headers.Accept = "application/vnd.foundry.canvas+json;version=2"; + const modelStream = canonical && path.endsWith('/canvas/model') && req.method==='GET' && req.headers.get('accept')==='application/vnd.foundry.canvas-model+ndjson;version=1'; + if(modelStream)headers.Accept='application/vnd.foundry.canvas-model+ndjson;version=1'; const session = req.cookies.get("foundry_session")?.value; if (session) headers.Authorization = "Bearer " + session; for (const h of ["x-foundry-code", "x-foundry-verifier", "if-match", "idempotency-key", "x-expected-account", "x-resource-id"]) { @@ -133,7 +135,7 @@ async function handler( // Snapshots contain no login exchange payload. Forward them without buffering // and parsing/re-encoding multi-megabyte inline image JSON in the proxy. if (canonical || canvas && path.endsWith("/canvas") && req.method === "GET" && upstream.ok) { - responseHeaders.set("Content-Type", "application/json; charset=utf-8"); + responseHeaders.set("Content-Type", modelStream && upstream.ok && upstream.headers.get('content-type')?.startsWith('application/vnd.foundry.canvas-model+ndjson;') ? upstream.headers.get('content-type')! : "application/json; charset=utf-8"); responseHeaders.set("Server-Timing", `upstream;dur=${(headersReceived - started).toFixed(1)}, proxy_headers;dur=${(performance.now() - headersReceived).toFixed(1)}`); return new NextResponse(upstream.body, { status: upstream.status, headers: responseHeaders }); } diff --git a/src/lib/canvas-runtime/contracts.ts b/src/lib/canvas-runtime/contracts.ts index 3ef6134..c5734b3 100644 --- a/src/lib/canvas-runtime/contracts.ts +++ b/src/lib/canvas-runtime/contracts.ts @@ -1,6 +1,6 @@ import type {OrphanFile} from './orphan-files.ts'; import type { ImportJob } from './imports.ts'; -import type { Document, Id, Json, OperationIntent, Snapshot } from '../canvas-core/types.ts'; +import type { CanvasRecord, ElementRecord, 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'; @@ -22,6 +22,8 @@ export type RuntimeState = { }; export type EngineSnapshot = Snapshot; export type CanonicalProposal = { readonly document: Document; readonly versions: ReadonlyMap }; +/** Receiving records are read deltas, not a validated Document or an ACK base. */ +export type CanonicalReadUpdate = CanonicalReadResult | { readonly status:'receiving'; readonly projectId:Id; readonly canvases:readonly CanvasRecord[]; readonly elements:readonly ElementRecord[]; readonly tombstones:ReadonlyMap; readonly versions:ReadonlyMap; readonly receivedElements:number; readonly totalElements:number }; 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 }; diff --git a/src/lib/canvas-runtime/host.ts b/src/lib/canvas-runtime/host.ts index aa22ddd..8d689cd 100644 --- a/src/lib/canvas-runtime/host.ts +++ b/src/lib/canvas-runtime/host.ts @@ -2,13 +2,13 @@ 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'; +import type { CanonicalReadUpdate } 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; modelLoader?(accountId:Id,workspaceId:Id,authEpoch:number,current:()=>boolean):LoadPort; 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 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] : []); @@ -34,7 +34,7 @@ export function createRuntimeHost(options: { registry: Registry; workspace(accou 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();} + 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||last.value.status==='receiving')throw fault('incomplete_load');return last.value;}finally{handle.release();} }, workspace(account: Id, workspace: Id) { activate(account); diff --git a/src/lib/canvas-service/foundry.ts b/src/lib/canvas-service/foundry.ts index 27e853c..183a853 100644 --- a/src/lib/canvas-service/foundry.ts +++ b/src/lib/canvas-service/foundry.ts @@ -1,3 +1,4 @@ +import { modelStreamType } from './model-stream.ts'; 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'; @@ -6,8 +7,8 @@ import type { FrozenRequest } from '../canvas-runtime/ledger.ts'; import { projectBatch } from '../canvas-runtime/batch-projection.ts'; import type { OperationStatus } from '../canvas-runtime/scheduler.ts'; -export type ServiceRequest = { readonly path: string; readonly method: 'GET' | 'POST'; readonly requestId?: Id; readonly expectedVersion?: number; readonly body?: string | FormData; readonly resourceId?: Id; readonly signal?: AbortSignal }; -export type ServiceResponse = { readonly status: number; readonly body: unknown; readonly retryAfterMs?: number }; +export type ServiceRequest = { readonly responseFormat?:'model-stream-v1'; readonly path: string; readonly method: 'GET' | 'POST'; readonly requestId?: Id; readonly expectedVersion?: number; readonly body?: string | FormData; readonly resourceId?: Id; readonly signal?: AbortSignal }; +export type ServiceResponse = { readonly status: number; readonly body: unknown; readonly stream?:AsyncIterable; readonly retryAfterMs?: number }; export type ServiceTransport = (request: ServiceRequest) => Promise; const uuid = /^[a-f\d]{8}-[a-f\d]{4}-[a-f\d]{4}-[a-f\d]{4}-[a-f\d]{12}$/i; const record = (value: unknown): Record => { if (!value || Array.isArray(value) || typeof value !== 'object') throw fault('invalid_canonical_response'); return value as Record; }; @@ -28,9 +29,20 @@ export function createFoundryTransport(options: { accountId: Id; isCurrentIdenti if (!options.isCurrentIdentity()) throw fault('identity_changed'); if (input.expectedVersion !== undefined && (!Number.isSafeInteger(input.expectedVersion) || input.expectedVersion < 1)) throw fault('invalid_server_version'); const response = await requestFetch(`${options.basePath ?? '/api/foundry/'}${input.path}`, { method: input.method, body: input.body, signal: input.signal, credentials: 'same-origin', cache: 'no-store', redirect: 'error', headers: { - Accept: 'application/json', ...(input.body instanceof FormData ? {} : { 'Content-Type': 'application/json' }), ...(input.resourceId ? { 'X-Resource-Id': input.resourceId } : {}), 'X-Expected-Account': options.accountId, ...(input.requestId ? { 'Idempotency-Key': input.requestId } : {}), ...(input.expectedVersion !== undefined ? { 'If-Match': `"${input.expectedVersion}"` } : {}), + Accept: input.responseFormat==='model-stream-v1'?modelStreamType:'application/json', ...(input.body instanceof FormData ? {} : { 'Content-Type': 'application/json' }), ...(input.resourceId ? { 'X-Resource-Id': input.resourceId } : {}), 'X-Expected-Account': options.accountId, ...(input.requestId ? { 'Idempotency-Key': input.requestId } : {}), ...(input.expectedVersion !== undefined ? { 'If-Match': `"${input.expectedVersion}"` } : {}), } }); if (!options.isCurrentIdentity()) { await response.body?.cancel(); throw fault('identity_changed'); } + if(input.responseFormat==='model-stream-v1'&&response.status===200&&response.headers.get('content-type')?.split(';')[0].trim()==='application/vnd.foundry.canvas-model+ndjson'){ + if(!/;\s*version=1(?:;|$)/.test(response.headers.get('content-type')??'')){await response.body?.cancel();throw fault('unsupported_model_stream');} + if(!response.body)throw fault('incomplete_model_stream'); + const body=response.body; + async function* stream(){ + const reader=body.getReader(),abort=()=>{void reader.cancel().catch(()=>{});};input.signal?.addEventListener('abort',abort,{once:true}); + try{while(true){input.signal?.throwIfAborted();if(!options.isCurrentIdentity())throw fault('identity_changed');const next=await reader.read();input.signal?.throwIfAborted();if(!options.isCurrentIdentity())throw fault('identity_changed');if(next.done)break;yield next.value;}} + finally{input.signal?.removeEventListener('abort',abort);await reader.cancel().catch(()=>{});reader.releaseLock();} + } + return{status:response.status,body:null,stream:stream()}; + } let body: unknown; try { body = await response.json(); } catch { throw fault('invalid_canonical_response'); } if (!options.isCurrentIdentity()) throw fault('identity_changed'); diff --git a/src/lib/canvas-service/model-loader.ts b/src/lib/canvas-service/model-loader.ts index e0c960b..ec44bea 100644 --- a/src/lib/canvas-service/model-loader.ts +++ b/src/lib/canvas-service/model-loader.ts @@ -1,16 +1,21 @@ +import { decodeModelStream } from './model-stream.ts'; 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 type { CanonicalReadUpdate } 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 { +/** Negotiate real streaming; older servers may explicitly return complete JSON. */ +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); + const response=await options.transport({method:'GET',path:`workspaces/${encodeURIComponent(options.workspaceId)}/materials/${encodeURIComponent(scope.canvasId)}/canvas/model`,signal,responseFormat:'model-stream-v1'}); + if(response.stream){ + if(response.status!==200)throw fault('invalid_model_stream'); + for await(const chunk of decodeModelStream(response.stream,scope.canvasId,options.registry,signal))yield Object.freeze({...chunk,serverVersions:Object.freeze(Object.fromEntries(chunk.versions)),permissionEpoch:scope.authEpoch,queryKey:loadQueryKey(scope)}); + return; + } + const value = await loadFoundryModel(async()=>response, 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))); diff --git a/src/lib/canvas-service/model-stream.ts b/src/lib/canvas-service/model-stream.ts new file mode 100644 index 0000000..4533264 --- /dev/null +++ b/src/lib/canvas-service/model-stream.ts @@ -0,0 +1,62 @@ +import type { CanvasRecord, ElementRecord, Id, 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 type { CanonicalReadUpdate } from '../canvas-runtime/contracts.ts'; + +export const modelStreamType = 'application/vnd.foundry.canvas-model+ndjson;version=1'; +const maxBytes=64*1024*1024,maxLine=8*1024*1024,maxLines=4096; +const uuid=/^[a-f\d]{8}-[a-f\d]{4}-[a-f\d]{4}-[a-f\d]{4}-[a-f\d]{12}$/i; +const object=(value:unknown):Record=>{if(!value||typeof value!=='object'||Array.isArray(value)||!validJson(value))throw fault('invalid_model_stream');return value as Record;}; +/** Bounded NDJSON framing. EOF is required even after a terminal record. */ +async function* lines(source:AsyncIterable,signal:AbortSignal){ + const decoder=new TextDecoder('utf-8',{fatal:true});let pending='',total=0,lineBytes=0,count=0; + for await(const bytes of source){ + signal.throwIfAborted();if(!(bytes instanceof Uint8Array))throw fault('invalid_model_stream');total+=bytes.byteLength;if(total>maxBytes)throw fault('model_stream_budget'); + let start=0; + for(let index=0;indexmaxLine)throw fault('model_stream_budget'); + pending+=decoder.decode(bytes.subarray(start,index),{stream:true});pending+=decoder.decode();if(++count>maxLines||!pending)throw fault('invalid_model_stream'); + yield{raw:JSON.parse(pending),byteLength:lineBytes+1};pending='';lineBytes=0;start=index+1; + } + lineBytes+=bytes.length-start;if(lineBytes>maxLine)throw fault('model_stream_budget');pending+=decoder.decode(bytes.subarray(start),{stream:true}); + } + signal.throwIfAborted();pending+=decoder.decode();if(pending||lineBytes)throw fault('incomplete_model_stream'); +} +export async function* decodeModelStream(source:AsyncIterable,canvasId:Id,registry:Registry,signal:AbortSignal){ + let sequence=0,token='',projectId='',expected=0,ended=false; + let terminal: {sequence:number;snapshotToken:string;projectId:string;versions:ReadonlyMap;complete:boolean;byteLength:number;coverage:string[];value:CanonicalReadUpdate}|undefined; + let canvases:CanvasRecord[]=[],versions:ReadonlyMap=new Map(),tombstones:ReadonlyMap=new Map(); + const elements:ElementRecord[]=[],ids=new Set(),canvasIds=new Set(); + for await(const line of lines(source,signal)){ + const raw=object(line.raw); + if(ended||raw.streamVersion!==1||raw.sequence!==sequence++||typeof raw.snapshotToken!=='string'||!raw.snapshotToken||raw.snapshotToken.length>256)throw fault('invalid_model_stream'); + if(sequence===1){ + if(raw.kind!=='header'||raw.documentFormatVersion!==1||typeof raw.projectId!=='string'||!uuid.test(raw.projectId)||!Number.isSafeInteger(raw.elementCount)||(raw.elementCount as number)<0||!Array.isArray(raw.canvases)||raw.canvases.length>256)throw fault('invalid_model_stream'); + token=raw.snapshotToken;projectId=raw.projectId;expected=raw.elementCount as number;canvases=raw.canvases as CanvasRecord[]; + for(const canvas of canvases){object(canvas);if(typeof canvas.id!=='string'||!uuid.test(canvas.id)||canvasIds.has(canvas.id))throw fault('invalid_model_stream');canvasIds.add(canvas.id);} + if(!canvasIds.has(canvasId)||!canvasIds.has(projectId))throw fault('project_mismatch'); + const versionMap=new Map(),dead=new Map(); + for(const [id,value]of Object.entries(object(raw.versions))){if(!uuid.test(id)||!Number.isSafeInteger(value)||(value as number)<1)throw fault('invalid_canonical_version');versionMap.set(id,value as number);} + if([...canvasIds].some(id=>!versionMap.has(id)))throw fault('invalid_model_stream'); + if(!Array.isArray(raw.removedCanvasIds)||new Set(raw.removedCanvasIds).size!==raw.removedCanvasIds.length||raw.removedCanvasIds.some(id=>typeof id!=='string'||canvasIds.has(id)||!versionMap.has(id))||versionMap.size!==canvasIds.size+raw.removedCanvasIds.length)throw fault('invalid_model_stream'); + for(const [id,value]of Object.entries(object(raw.tombstones))){if(!id||!Number.isSafeInteger(value)||(value as number)<1)throw fault('invalid_model_stream');dead.set(id,value as number);} + versions=readMap(versionMap);tombstones=readMap(dead); + const value:CanonicalReadUpdate=freeze({status:'receiving',projectId,canvases,elements:[],tombstones,versions,receivedElements:0,totalElements:expected}); + yield{sequence:0,snapshotToken:token,projectId,versions,complete:false,byteLength:line.byteLength,coverage:['model-header'],value};continue; + } + if(raw.snapshotToken!==token)throw fault('load_snapshot_changed'); + if(raw.kind==='elements'){ + if(!Array.isArray(raw.elements)||!raw.elements.length||raw.elements.length>64||elements.length+raw.elements.length>expected)throw fault('invalid_model_stream'); + for(const item of raw.elements){const element=object(item) as unknown as ElementRecord;if(typeof element.id!=='string'||!element.id||ids.has(element.id)||!canvasIds.has(element.ownerCanvasId))throw fault('invalid_model_stream');ids.add(element.id);elements.push(element);} + const value:CanonicalReadUpdate=freeze({status:'receiving',projectId,canvases:[],elements:raw.elements as ElementRecord[],tombstones:readMap(new Map()),versions,receivedElements:elements.length,totalElements:expected}); + yield{sequence:sequence-1,snapshotToken:token,projectId,versions,complete:false,byteLength:line.byteLength,coverage:raw.elements.map(item=>`element:${(item as ElementRecord).id}`),value}; + }else if(raw.kind==='complete'){ + if(raw.elementCount!==expected||elements.length!==expected)throw fault('incomplete_model_stream'); + const document=createDocument(projectId,canvases,elements,tombstones);validateDocument(document,registry);ended=true; + const value:CanonicalReadUpdate=Object.freeze({status:'ready',document,versions}); + terminal={sequence:sequence-1,snapshotToken:token,projectId,versions,complete:true,byteLength:line.byteLength,coverage:[...canvasIds].map(id=>`canvas:${id}`),value}; + }else throw fault('invalid_model_stream'); + } + if(!ended||!terminal)throw fault('incomplete_model_stream'); + yield terminal; +} diff --git a/tests/headless/model-stream.test.mjs b/tests/headless/model-stream.test.mjs new file mode 100644 index 0000000..b1fc49d --- /dev/null +++ b/tests/headless/model-stream.test.mjs @@ -0,0 +1,56 @@ +import {test} from 'node:test'; +import assert from 'node:assert/strict'; +import {createRegistry,builtinComponents} from '../../src/lib/canvas-core/index.ts'; +import {decodeModelStream,modelStreamType} from '../../src/lib/canvas-service/model-stream.ts'; +import {createFoundryTransport} from '../../src/lib/canvas-service/foundry.ts'; +import {createFoundryModelLoader} from '../../src/lib/canvas-service/model-loader.ts'; +import {createScopedLoader} from '../../src/lib/canvas-runtime/loader.ts'; +const root='10000000-0000-4000-8000-000000000001',workspace='10000000-0000-4000-8000-000000000002',account='10000000-0000-4000-8000-000000000003'; +const registry=createRegistry(builtinComponents()),signal=()=>new AbortController().signal; +const node={id:'a',incarnation:1,ownerCanvasId:root,componentId:'text',dataVersion:1,structure:'node',data:{title:'中文',text:'🙂'},geometry:{x:0,y:0,width:100,height:80,rotation:0}}; +const records=()=>[{streamVersion:1,snapshotToken:'snapshot-a',sequence:0,kind:'header',documentFormatVersion:1,projectId:root,canvases:[{id:root,parent:null}],versions:{[root]:3},tombstones:{old:2},removedCanvasIds:[],elementCount:1},{streamVersion:1,snapshotToken:'snapshot-a',sequence:1,kind:'elements',elements:[structuredClone(node)]},{streamVersion:1,snapshotToken:'snapshot-a',sequence:2,kind:'complete',elementCount:1}]; +const encode=rows=>new TextEncoder().encode(rows.map(row=>JSON.stringify(row)).join('\n')+'\n'); +async function* source(rows){yield encode(rows);} +const collect=async iterable=>{const result=[];for await(const item of iterable)result.push(item);return result;}; +const scope={accountId:account,workspaceId:workspace,projectId:null,canvasId:root,authEpoch:1,navigationGeneration:1,viewKey:'canonical-project',format:'canonical-model-v1',queryFingerprint:'full-project',source:'foundry'}; + +test('stream decoder frames split UTF8/newlines, emits deltas and validates one final complete model',async()=>{ + const bytes=encode(records());async function* split(){for(const byte of bytes)yield Uint8Array.of(byte);} + const chunks=await collect(decodeModelStream(split(),root,registry,signal())); + assert.deepEqual(chunks.map(chunk=>[chunk.sequence,chunk.complete,chunk.value.status]),[[0,false,'receiving'],[1,false,'receiving'],[2,true,'ready']]); + assert.equal(chunks[1].value.elements[0].data.text,'🙂');assert.equal(chunks[2].value.document.elements.get('a').data.title,'中文');assert.equal(chunks[2].value.document.tombstones.get('old'),2);assert.equal(chunks.reduce((sum,chunk)=>sum+chunk.byteLength,0),bytes.length);assert.ok(Object.isFrozen(chunks[1].value.elements[0])); +}); +test('snapshot switches, sequence gaps, duplicate IDs, wrong counts, late records and invalid topology cannot finish',async()=>{ + const cases=[rows=>rows[1].snapshotToken='other',rows=>rows[1].sequence=2,rows=>rows[0].elementCount=2,rows=>rows[2].elementCount=2,rows=>rows.push({...rows[2],sequence:3}),rows=>rows[1].elements.push(rows[1].elements[0]),rows=>rows[1].elements[0].ownerCanvasId=workspace,rows=>rows[1].elements[0].geometry.width=-1,rows=>rows[0].versions={},rows=>rows[0].canvases[0].parent={canvasId:root,entryElementId:'a'}]; + for(const mutate of cases){const rows=records();mutate(rows);await assert.rejects(()=>collect(decodeModelStream(source(rows),root,registry,signal())));} + for(const rows of [[],records().slice(0,1),records().slice(0,2)])await assert.rejects(()=>collect(decodeModelStream(source(rows),root,registry,signal()))); + async function* truncated(){yield encode(records()).slice(0,-1);}await assert.rejects(()=>collect(decodeModelStream(truncated(),root,registry,signal())),{code:'incomplete_model_stream'}); + async function* huge(){yield new Uint8Array(8*1024*1024+1).fill(65);}await assert.rejects(()=>collect(decodeModelStream(huge(),root,registry,signal())),{code:'model_stream_budget'}); +}); +test('Foundry transport negotiates streaming without buffering; coordinator retains first delta before gated tail',async()=>{ + let release,opened;const gate=new Promise(resolve=>release=resolve),started=new Promise(resolve=>opened=resolve);let requested; + const rows=records(),stream=new ReadableStream({async start(controller){controller.enqueue(encode(rows.slice(0,2)));opened();await gate;controller.enqueue(encode(rows.slice(2)));controller.close();}}); + const transport=createFoundryTransport({accountId:account,isCurrentIdentity:()=>true,fetch:async(_url,input)=>{requested=input;return new Response(stream,{headers:{'Content-Type':modelStreamType}});}}); + const seen=[],loader=createScopedLoader({load:createFoundryModelLoader({workspaceId:workspace,registry,transport}),current:()=>true}); + const handle=loader.acquire(scope,{onUpdate:snapshot=>seen.push(snapshot)});try{ + await started;while(!seen.some(snapshot=>snapshot.chunks.length===2))await new Promise(resolve=>setImmediate(resolve)); + assert.equal(requested.headers.Accept,modelStreamType);assert.equal(handle.getSnapshot().status,'partial');assert.equal(handle.getSnapshot().chunks.at(-1).value.status,'receiving'); + release();const done=await handle.ready;assert.equal(done.status,'complete');assert.equal(done.chunks.at(-1).value.document.elements.size,1); + }finally{release();handle.release();loader.close();} +}); +test('malformed tails retain read deltas without yielding an authoritative model',async()=>{ + async function* broken(){yield encode(records().slice(0,2));yield new TextEncoder().encode('{broken}\n');} + const states=[],loader=createScopedLoader({load:createFoundryModelLoader({workspaceId:workspace,registry,transport:async()=>({status:200,body:null,stream:broken()})}),current:()=>true});const handle=loader.acquire(scope,{onUpdate:value=>states.push(value)}); + try{await assert.rejects(handle.ready);assert.equal(handle.getSnapshot().status,'failed-with-partial');assert.equal(handle.getSnapshot().chunks.length,2);assert.equal(handle.getSnapshot().chunks.at(-1).value.status,'receiving');}finally{handle.release();loader.close();} +}); +test('terminal record still requires EOF; trailing garbage rejects rather than opening a baseline',async()=>{ + const gate={},rows=records();let release;gate.promise=new Promise(resolve=>release=resolve); + async function* trailing(){yield encode(rows);await gate.promise;yield new TextEncoder().encode('x\n');} + const loader=createScopedLoader({load:createFoundryModelLoader({workspaceId:workspace,registry,transport:async()=>({status:200,body:null,stream:trailing()})}),current:()=>true}),handle=loader.acquire(scope);let ready=false;handle.ready.then(()=>ready=true,()=>{}); + try{while(handle.getSnapshot().chunks.length<2)await new Promise(resolve=>setImmediate(resolve));assert.equal(ready,false);release();await assert.rejects(handle.ready);assert.equal(handle.getSnapshot().status,'failed-with-partial');}finally{release();handle.release();loader.close();} +}); +test('stream transport fences changed identity and abort cancels a blocked body reader',async()=>{ + let current=true,cancelled=0;const abort=new AbortController();const transport=createFoundryTransport({accountId:account,isCurrentIdentity:()=>current,fetch:async()=>new Response(new ReadableStream({cancel(){cancelled++;}}),{headers:{'Content-Type':modelStreamType}})}); + const reply=await transport({method:'GET',path:'model',responseFormat:'model-stream-v1',signal:abort.signal});const iterator=reply.stream[Symbol.asyncIterator]();const pending=iterator.next();await Promise.resolve();abort.abort();await assert.rejects(pending);assert.equal(cancelled,1); + const other=await transport({method:'GET',path:'model',responseFormat:'model-stream-v1'});current=false;await assert.rejects(()=>other.stream[Symbol.asyncIterator]().next(),{code:'identity_changed'});assert.equal(cancelled,2); +}); diff --git a/tests/headless/proxy.integration.mjs b/tests/headless/proxy.integration.mjs index e80e2ab..6d7b71a 100644 --- a/tests/headless/proxy.integration.mjs +++ b/tests/headless/proxy.integration.mjs @@ -10,9 +10,12 @@ import { setTimeout as pause } from 'node:timers/promises'; const source=resolve(import.meta.dirname,'../..'),directory=await mkdtemp(join(tmpdir(),'endoleap-proxy-check-')),received=[]; const account='10000000-0000-4000-8000-000000000004',key='90000000-0000-4000-8000-000000000002'; const responseBytes=' { "protocol_version":2, "request_id":"'+key+'", "status":"not-applied", "terminal":true, "error":"fixture_rejection", "extension":{"retained":true} }\n'; +let streamRelease;const streamGate=new Promise(resolve=>streamRelease=resolve); +const streamPrefix='{"streamVersion":1,"kind":"header"}\n',streamTail='{"streamVersion":1,"kind":"complete"}\n'; const upstream=createServer(async(request,response)=>{ const chunks=[];for await(const chunk of request)chunks.push(chunk); received.push({path:request.url,body:Buffer.concat(chunks),headers:request.headers}); + if(request.url.endsWith('/model')&&request.headers.accept==='application/vnd.foundry.canvas-model+ndjson;version=1'){response.writeHead(200,{'Content-Type':'application/vnd.foundry.canvas-model+ndjson;version=1'});response.write(streamPrefix);await streamGate;response.end(streamTail);return;} response.writeHead(request.url.endsWith('/model/operations')?409:200,{'Content-Type':'application/json'}); response.end(request.url.endsWith('/model/operations')?responseBytes:'{"id":"fixture"}'); }); @@ -60,10 +63,15 @@ try{ const aiEvents=await fetch(`http://127.0.0.1:${port}/api/foundry/${aiPath}/${key}?after_seq=41&limit=100`);await aiEvents.text();assert.equal(aiEvents.status,200); assert.ok(received.some(item=>item.path.endsWith('/ai/runs/'+key+'?limit=100&after_seq=41'))); console.log('PASS frozen AI bytes, identity, version header, admission receipt and event cursor cross the actual Next proxy'); + const streamed=await fetch(`http://127.0.0.1:${port}/api/foundry/${path.replace('/operations','')}`,{headers:{Accept:'application/vnd.foundry.canvas-model+ndjson;version=1'}}); + assert.equal(streamed.headers.get('content-type'),'application/vnd.foundry.canvas-model+ndjson;version=1'); + const reader=streamed.body.getReader(),first=await reader.read();assert.equal(new TextDecoder().decode(first.value),streamPrefix);streamRelease();let rest='';while(true){const next=await reader.read();if(next.done)break;rest+=new TextDecoder().decode(next.value);}assert.equal(rest,streamTail);reader.releaseLock(); + console.log('PASS negotiated canonical NDJSON and first chunk cross the actual Next proxy before the gated tail'); const before=received.length,invalid=await fetch(`http://127.0.0.1:${port}/api/foundry/${path}/unexpected/extra`,{signal:AbortSignal.timeout(10000)});await invalid.text();assert.equal(invalid.status,404);assert.equal(received.length,before); console.log('PASS actual Next proxy preserves frozen request bytes/key/account and raw terminal receipt above 1 MiB'); console.log('PASS canonical route allowlist rejects unexpected paths without contacting the upstream'); }finally{ + streamRelease(); if(next?.pid){const exited=new Promise(resolve=>{if(next.exitCode!==null)resolve();else next.once('exit',resolve)});try{process.kill(-next.pid,'SIGTERM')}catch{}await Promise.race([exited,pause(3000)]);if(next.exitCode===null){try{process.kill(-next.pid,'SIGKILL')}catch{}}} upstream.closeAllConnections();await new Promise(resolve=>upstream.close(resolve));await rm(directory,{recursive:true,force:true,maxRetries:5}); } diff --git a/tests/headless/renderer.integration.mjs b/tests/headless/renderer.integration.mjs index 23dfc83..7a09b05 100644 --- a/tests/headless/renderer.integration.mjs +++ b/tests/headless/renderer.integration.mjs @@ -39,6 +39,14 @@ const upstream = createServer(async (request, response) => { if (path.endsWith('/materials')) return reply(200, { items: boards, next_offset: null }); const board = boards.find(board => path.endsWith('/materials/' + board.id)); if (board) return reply(200, board); const read = path.match(/\/materials\/([^/]+)\/canvas\/model$/); + if(read&&request.headers.accept==='application/vnd.foundry.canvas-model+ndjson;version=1'){ + const value=model(documents.get(read[1])),token='fixture-snapshot-'+calls.length;let sequence=0; + response.writeHead(200,{'Content-Type':'application/vnd.foundry.canvas-model+ndjson;version=1'}); + const line=record=>response.write(JSON.stringify({streamVersion:1,snapshotToken:token,sequence:sequence++,...record})+'\n'); + line({kind:'header',documentFormatVersion:1,projectId:value.projectId,canvases:value.canvases,versions:value.versions,tombstones:value.tombstones,removedCanvasIds:value.removedCanvasIds,elementCount:value.elements.length}); + for(let offset=0;offsetbutton.textContent.trim()==='退出登录').click()`); await until(`location.pathname==='/login'&&!document.body.innerText.includes('This page couldn’t load')`,'logout did not safely detach the view'); + assert.ok(calls.some(call=>call.path.endsWith('/canvas/model')&&call.headers.accept==='application/vnd.foundry.canvas-model+ndjson;version=1')); + console.log('PASS actual GraphPage negotiates and assembles canonical NDJSON through the real Next proxy'); console.log('PASS logout retires the runtime before dropping its view without a render error'); } } finally {