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

docs(canvas): explain Foundry transport model streams and service receipts

Browse Source
This commit is contained in:
Codex Bot
2026-11-12 01:20:49 +08:00
parent 11eaac1f9f
commit 450fa3f719
6 changed files with 175 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/lib/canvas-service/assets.ts
+21
View File
@@ -1,3 +1,7 @@
/**
* Foundry 图片上传与共享资源授权适配。原文件的预分配资源 ID、SHA、长度和操作身份绑定冻结请求，
* 丢响应后查询原请求，而不是重新创建图片。规范 sourceURL 与预览 displayURL 分开，预览不能冒充原件身份。
*/
import {validateSharedAsset} from '../canvas-runtime/asset-depot.ts';
import type { Id, Json } from '../canvas-core/types.ts';
import { fault, freeze, validJson } from '../canvas-core/values.ts';
@@ -6,16 +10,21 @@ import type { FrozenRequest } from '../canvas-runtime/ledger.ts';
import type { OperationStatus } from '../canvas-runtime/scheduler.ts';
import type { ServiceResponse, ServiceTransport } from './foundry.ts';
// 本服务的工作区、请求及资源身份采用 UUID；摘要为小写64位十六进制 SHA-256。
const uuid = /^[a-f\d]{8}-[a-f\d]{4}-[a-f\d]{4}-[a-f\d]{4}-[a-f\d]{12}$/i;
/** 只校验响应最外层对象，回执字段关联、JSON 和资源协议另在对应路径检查。 */
function object(value: unknown): Record<string, unknown> {
if (!value || Array.isArray(value) || typeof value !== 'object') throw fault('invalid_asset_response');
return value as Record<string, unknown>;
}
/** 交叉检查冻结请求与元数据载荷：一个上传操作、无canvas写入版本、固定资源身份和原字节属性。
* 最大原件20MiB；请求body只是可恢复元数据，原始Blob通过上传FormData单独携带。 */
function payload(request: FrozenRequest) {
const raw = object(JSON.parse(request.body));
if (request.kind !== 'asset.upload' || request.protocolVersion !== 2 || raw.protocol_version !== 2 || raw.request_id !== request.id || !uuid.test(request.id) || typeof raw.resource_id !== 'string' || !uuid.test(raw.resource_id) || request.operationIds.length !== 1 || raw.operation_id !== request.operationIds[0] || request.canvasIds.length || Object.keys(request.expectedVersions).length || typeof raw.sha256 !== 'string' || !/^[a-f0-9]{64}$/.test(raw.sha256) || typeof raw.byte_length !== 'number' || !Number.isSafeInteger(raw.byte_length) || raw.byte_length < 1 || raw.byte_length > 20 * 1024 * 1024 || typeof raw.name !== 'string' || !raw.name || typeof raw.mime_type !== 'string') throw fault('invalid_asset_request');
return raw;
}
/** 成功状态须匹配协议/请求身份；未知回执仍为 unresolved，普通拒绝也不能证明先前上传未提交。 */
function status(request: FrozenRequest, response: ServiceResponse, query: boolean): OperationStatus {
const body = object(response.body), matching = body.protocol_version === 2 && body.request_id === request.id;
if (matching && body.status === 'applied' && response.status >= 200 && response.status < 300) {
@@ -29,24 +38,32 @@ function status(request: FrozenRequest, response: ServiceResponse, query: boolea
}
/** Immutable bytes and a preallocated resource ID survive response loss and view changes. */
/** 校验规范HTTPS源站，构造工作区资源服务与授权域。
* 上传与查询复用固定身份；本地资源是否保存、任务如何排队由engine负责。 */
export function createFoundryAssets(options: { workspaceId: Id; transport: ServiceTransport; project?: string; origin?: string; displayURL?: (id: Id) => string }): AssetPort {
if (!uuid.test(options.workspaceId)) throw fault('invalid_service_scope');
const origin = new URL(options.origin ?? 'https://api.foundry.entroia.com');
if (origin.protocol !== 'https:' || origin.username || origin.password || origin.search || origin.hash || origin.pathname !== '/') throw fault('invalid_asset_origin');
const path = `workspaces/${encodeURIComponent(options.workspaceId)}/assets`;
/** 可持久化的规范原件地址；每段编码，默认项目endoleap，不与宿主代理预览URL混用。 */
const sourceURL = (id: Id) => `${origin.origin}/projects/${encodeURIComponent(options.project ?? 'endoleap')}/${path}/${encodeURIComponent(id)}`;
/** 验证已应用回执和本地资源事实一致，包括操作/摘要/长度/规范URL、尺寸和允许MIME。
* 限制正整数尺寸及4000万像素，复制并冻结结果后才能登记StoredAsset。 */
const accept = (request: FrozenRequest, result: OperationStatus, resource: ResourceRecord): StoredAsset => {
const raw = payload(request), receipt = object(result.canonical), asset = object(receipt.asset);
if (result.outcome !== 'applied' || receipt.protocol_version !== 2 || receipt.request_id !== request.id || receipt.status !== 'applied' || resource.id !== raw.resource_id || resource.operationId !== raw.operation_id || resource.digest !== raw.sha256 || resource.byteLength !== raw.byte_length || asset.id !== resource.id || asset.url !== sourceURL(resource.id) || asset.bytes !== resource.byteLength || typeof asset.width !== 'number' || typeof asset.height !== 'number' || !Number.isSafeInteger(asset.width) || !Number.isSafeInteger(asset.height) || asset.width <= 0 || asset.height <= 0 || asset.width * asset.height > 40_000_000 || !['image/png', 'image/jpeg', 'image/gif', 'image/webp'].includes(asset.mime_type as string)) throw fault('asset_receipt_mismatch');
return freeze(JSON.parse(JSON.stringify(asset)) as StoredAsset);
};
/** bytes存在走POST上传，否则GET原操作回执；执行错误保留unresolved，不擅自重置资源身份。 */
async function execute(request: FrozenRequest, bytes: Blob | undefined, signal: AbortSignal) {
try {
const raw = payload(request);
let body: FormData | undefined;
if (bytes) {
// 上传前再核对真实Blob长度和SHA；这里整体读取ArrayBuffer，受20MiB上传上限，不是流式摘要。
const digest = [...new Uint8Array(await crypto.subtle.digest('SHA-256', await bytes.arrayBuffer()))].map(value => value.toString(16).padStart(2, '0')).join('');
if (bytes.size !== raw.byte_length || digest !== raw.sha256) throw fault('resource_bytes_mismatch');
// FormData只携带原图和文件名；幂等请求ID及资源ID通过transport专用头发送。
body = new FormData(); body.append('image', bytes, raw.name as string);
}
const response = await options.transport({ method: bytes ? 'POST' : 'GET', path: bytes ? path : `${path}/operations/${encodeURIComponent(request.id)}`, requestId: bytes ? request.id : undefined, resourceId: bytes ? raw.resource_id as string : undefined, body, signal });
@@ -56,7 +73,11 @@ export function createFoundryAssets(options: { workspaceId: Id; transport: Servi
return { outcome: 'unresolved' as const, error: code ?? 'asset_transport_unknown', ...(code && code !== 'invalid_asset_response' ? { blocked: true } : {}) };
}
}
/** 跨项目复用的服务授权端口。resolve读取版本化元数据并核对ID/规范URL，
* read获取原件字节；字节SHA与授权元数据的一致性由调用它的AssetDepot继续验证。
* scope绑定源站/项目/工作区，缓存命中不能替代一次新的领用授权。 */
const authority=Object.freeze({scope:`${origin.origin}/projects/${encodeURIComponent(options.project??'endoleap')}/${path}`,async resolve(id:Id,signal:AbortSignal){if(!uuid.test(id))throw fault('invalid_resource_id');const response=await options.transport({method:'GET',path:`${path}/metadata/${encodeURIComponent(id)}`,signal}),raw=object(response.body);if(response.status!==200)throw fault(typeof raw.error==='string'?raw.error:'asset_authorization_failed');if(raw.metadata_version!==1)throw fault('asset_metadata_interpreter_unavailable');const value={version:1 as const,authority:authority.scope,digest:raw.sha256 as string,asset:raw.asset as StoredAsset};validateSharedAsset(value);if(value.asset.id!==id||value.asset.url!==sourceURL(id))throw fault('asset_authorization_mismatch');return freeze(value);},async read(id:Id,signal:AbortSignal){if(!uuid.test(id))throw fault('invalid_resource_id');const response=await options.transport({method:'GET',path:`${path}/${encodeURIComponent(id)}`,signal,responseFormat:'asset-bytes-v1'});if(response.status!==200||!response.bytes)throw fault('asset_bytes_unavailable');return response.bytes;}});
// 默认显示地址走同源预览代理；upload/query只委托统一执行器，不复制另一套任务状态。
return Object.freeze({ authority, sourceURL, displayURL: options.displayURL ?? ((id: Id) => `/api/foundry/${path}/${encodeURIComponent(id)}?variant=preview`), accept,
upload: (request: FrozenRequest, bytes: Blob, signal: AbortSignal) => execute(request, bytes, signal),
query: (request: FrozenRequest, signal: AbortSignal) => execute(request, undefined, signal),
src/lib/canvas-service/foundry.ts
+69
View File
@@ -1,3 +1,8 @@
/**
* Foundry HTTP 与文档增量协议的边界。transport 处理账户绑定、媒体类型与流/字节读取；
* persistence 把逻辑操作编译成固定协议请求，并验证回执身份、覆盖范围和版本。
* 本模块返回 ACK 候选/已应用证明，不直接覆盖 editor 的 working；本地未决操作重放由 runtime 负责。
*/
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';
@@ -7,46 +12,65 @@ import type { FrozenRequest } from '../canvas-runtime/ledger.ts';
import { projectBatch } from '../canvas-runtime/batch-projection.ts';
import type { OperationStatus } from '../canvas-runtime/scheduler.ts';
/** 宿主无关服务请求描述；requestId/expectedVersion 分别映射幂等键与条件版本头。
* body 可为已冻结 JSON 字符串或上传 FormData，responseFormat 只选择预期表示，不改变业务身份。 */
export type ServiceRequest = { readonly responseFormat?:'canvas-v2'|'model-stream-v1'|'asset-bytes-v1'; readonly path: string; readonly method: 'GET' | 'POST'; readonly requestId?: Id; readonly expectedVersion?: number; readonly body?: string | FormData; readonly resourceId?: Id; readonly signal?: AbortSignal };
/** 状态与解释后载荷，特定表示额外提供 bytes 或 stream；retryAfterMs 是服务端重试提示。 */
export type ServiceResponse = { readonly status: number; readonly body: unknown; readonly bytes?:Blob; readonly stream?:AsyncIterable<Uint8Array>; readonly retryAfterMs?: number };
/** 各服务共用的可注入传输端口，测试可模拟；调度、并发和可靠保存不由此类型实现。 */
export type ServiceTransport = (request: ServiceRequest) => Promise<ServiceResponse>;
// 服务协议对账户、工作区、项目、请求等 UUID 身份的语法约束。
const uuid = /^[a-f\d]{8}-[a-f\d]{4}-[a-f\d]{4}-[a-f\d]{4}-[a-f\d]{12}$/i;
/** 检查最外层对象形状；需要完整 JSON/业务校验的路径另行执行，不能把此函数当 schema 校验器。 */
const record = (value: unknown): Record<string, unknown> => { if (!value || Array.isArray(value) || typeof value !== 'object') throw fault('invalid_canonical_response'); return value as Record<string, unknown>; };
/** 只接受实际数组，禁止缺失字段被当作空列表。 */
const list = (value: unknown): unknown[] => { if (!Array.isArray(value)) throw fault('invalid_canonical_response'); return value; };
/** 读取非空字符串身份；是否必须 UUID 由各协议字段的调用处决定。 */
const id = (value: unknown): string => { if (typeof value !== 'string' || !value) throw fault('invalid_canonical_response'); return value; };
/** 将版本对象转只读 Map，键必须 UUID，版本必须安全整数且不低于 minimum。 */
function versions(value: unknown, minimum = 1): ReadonlyMap<Id, number> {
const result = new Map<Id, number>();
for (const [key, version] of Object.entries(record(value))) { if (!uuid.test(key) || typeof version !== 'number' || !Number.isSafeInteger(version) || version < minimum) throw fault('invalid_canonical_version'); result.set(key, version); }
return readMap(result);
}
/** 路径段逐个编码，项目模型读取和操作端点共用同一命名空间。 */
const pathFor = (workspace: Id, canvas: Id) => `workspaces/${encodeURIComponent(workspace)}/materials/${encodeURIComponent(canvas)}/canvas/model`;
/** Credentials stay in the host/session. Every write additionally binds the expected account. */
/** 默认将请求发送到同源代理，凭证留在宿主会话，不塞进请求描述或恢复镜像。
* 发出前与收到后均检查当前身份，带 X-Expected-Account 供服务端复查账户；跨身份响应不能交给旧调用者。 */
export function createFoundryTransport(options: { accountId: Id; isCurrentIdentity(): boolean; fetch?: typeof fetch; basePath?: string }): ServiceTransport {
if (!uuid.test(options.accountId)) throw fault('invalid_account_identity');
const requestFetch = options.fetch ?? globalThis.fetch;
return async input => {
if (!options.isCurrentIdentity()) throw fault('identity_changed');
if (input.expectedVersion !== undefined && (!Number.isSafeInteger(input.expectedVersion) || input.expectedVersion < 1)) throw fault('invalid_server_version');
// 禁止缓存和重定向，按表示设置 Accept；FormData 的 boundary 交给浏览器，不能强写 JSON Content-Type。
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: input.responseFormat==='canvas-v2'?'application/vnd.foundry.canvas+json;version=2':input.responseFormat==='model-stream-v1'?modelStreamType:input.responseFormat==='asset-bytes-v1'?'image/*':'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'); }
// 成功的资源字节读取同时约束 Content-Length 提示和实际累计20MiB，不能只信响应头。
if(input.responseFormat==='asset-bytes-v1'&&response.status===200){
const maximum=20*1024*1024,length=response.headers.get('content-length');
if(length!==null&&(!/^\d+$/.test(length)||Number(length)>maximum)){await response.body?.cancel();throw fault('asset_byte_budget_exceeded');}
if(!response.body)throw fault('asset_bytes_unavailable');
const reader=response.body.getReader(),chunks:Uint8Array<ArrayBuffer>[]= [];let size=0;
/** 读流前后检查取消与身份，避免中途登出/换账户后继续交付原资源字节。 */
const check=()=>{input.signal?.throwIfAborted();if(!options.isCurrentIdentity())throw fault('identity_changed');};
/** 尽力取消 reader 以唤醒挂起的 read；最终释放由 finally 统一完成。 */
const abort=()=>{void reader.cancel().catch(()=>{});};input.signal?.addEventListener('abort',abort,{once:true});
try{while(true){check();const next=await reader.read();check();if(next.done)break;size+=next.value.byteLength;if(size>maximum)throw fault('asset_byte_budget_exceeded');chunks.push(next.value);}}
finally{input.signal?.removeEventListener('abort',abort);await reader.cancel().catch(()=>{});reader.releaseLock();}
return{status:200,body:null,bytes:new Blob(chunks,{type:response.headers.get('content-type')?.split(';')[0].trim()??''})};
}
// 只有明确的 NDJSON 媒体类型与version=1 才交给模型流解释器，其它表示继续走 JSON 回退。
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;
/** 惰性读取字节，每个 await 两侧复查身份；迭代结束、抛错或消费者提前退出都归还 reader 锁。
* NDJSON 内容和字节预算由 model-stream 解释器执行，此处只提供有生命周期边界的字节源。 */
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;}}
@@ -54,15 +78,20 @@ export function createFoundryTransport(options: { accountId: Id; isCurrentIdenti
}
return{status:response.status,body:null,stream:stream()};
}
// 普通 JSON 路径仍整体 response.json；不能借用流式协议的64MiB预算宣称这里也有预解析硬上限。
let body: unknown;
try { body = await response.json(); } catch { throw fault('invalid_canonical_response'); }
if (!options.isCurrentIdentity()) throw fault('identity_changed');
// 当前只解释非负数值秒数；HTTP 日期格式不在该适配路径支持范围，不猜测重试时刻。
const retry = response.headers.get('retry-after'), seconds = retry === null ? NaN : Number(retry);
return { status: response.status, body, ...(Number.isFinite(seconds) && seconds >= 0 ? { retryAfterMs: seconds * 1000 } : {}) };
};
}
/** 模型响应的完整或参与者片段，versions 同时覆盖实际 canvas 与 removed ID；不等于已验证完整文档。 */
type ModelChunk = { projectId: Id; complete: boolean; canvases: readonly CanvasRecord[]; elements: readonly ElementRecord[]; tombstones: ReadonlyMap<Id, number>; versions: ReadonlyMap<Id, number>; removed: readonly Id[] };
/** 验证JSON后复制外部响应，检查模型格式、ID唯一性、owner范围、版本覆盖和墓碑值。
* 跨分区树、端点及组件 schema 的完整一致性留给构建最终 Document 后的 validateDocument。 */
function modelChunk(value: unknown): ModelChunk {
if (!validJson(value)) throw fault('invalid_canonical_response');
const raw = record(JSON.parse(JSON.stringify(value)));
@@ -82,12 +111,16 @@ function modelChunk(value: unknown): ModelChunk {
elementIds.add(element.id);
}
for (const [key, value] of Object.entries(record(raw.tombstones))) { if (typeof value !== 'number' || !Number.isSafeInteger(value) || value < 1) throw fault('invalid_canonical_response'); tombstones.set(key, value); }
// 删除列表不得重复、不得与返回 canvas 重叠，版本表也不得含无归属的额外 ID。
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 });
}
/** 完整读取可返回 ready 或 migration-required，迁移要求不能强制断言为新模型。 */
export type ModelLoad = CanonicalReadResult;
/** 完整 JSON 读取入口，校验旧格式迁移提示或完整新模型；部分片段不能作为初始完整文档采用。 */
export async function loadFoundryModel(transport: ServiceTransport, workspaceId: Id, canvasId: Id, registry: Registry, signal?: AbortSignal): Promise<ModelLoad> {
const response = await transport({ method: 'GET', path: pathFor(workspaceId, canvasId), signal }), body = record(response.body);
// 旧数据迁移范围由 root_id/source_versions 指定，并必须覆盖本次访问画布及根项目。
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');
@@ -102,6 +135,8 @@ export async function loadFoundryModel(transport: ServiceTransport, workspaceId:
return Object.freeze({ status: 'ready', document, versions: chunk.versions });
}
/** 把HTTP响应转换为请求结果状态，先匹配协议/请求身份。
* applied 要精确覆盖冻结操作集合；accepted 只表示已受理；未知回执保持 unresolved，不能推断未生效。 */
function statusFor(request: FrozenRequest, response: ServiceResponse, query: boolean): OperationStatus {
const body = record(response.body), error = typeof body.error === 'string' ? body.error : undefined;
const matching = body.protocol_version === 2 && body.request_id === request.id;
@@ -113,10 +148,13 @@ function statusFor(request: FrozenRequest, response: ServiceResponse, query: boo
}
if (matching && body.status === 'accepted' && response.status >= 200 && response.status < 300) return { outcome: 'accepted', retryAfterMs: response.retryAfterMs };
if (matching && body.status === 'not-applied' && body.terminal === true) return { outcome: 'not-applied', error: error ?? 'not_applied' };
// 查询不到回执不等于从未执行，继续使用原请求身份查证；普通4xx未提供终结证据时阻塞。
if (query && response.status === 404 && body.status === 'unknown' && body.request_id === request.id) return { outcome: 'unresolved', error: 'operation_receipt_unknown', retryAfterMs: response.retryAfterMs };
if (response.status >= 400 && response.status < 500) return { outcome: 'unresolved', blocked: true, error: error ?? 'protocol_rejected' };
return { outcome: 'unresolved', error: error ?? 'transport_unknown', retryAfterMs: response.retryAfterMs };
}
/** 发送/查询/解释前交叉核对冻结元数据与请求体：协议、request_id、预期版本、画布集合和操作覆盖一致。
* 不在这里重新编译 payload，重试必须沿用已经保存的精确 body。 */
function validateFrozen(request: FrozenRequest) {
let body: Record<string, unknown>;
try { body = record(JSON.parse(request.body)); } catch { throw fault('invalid_frozen_request'); }
@@ -125,17 +163,26 @@ function validateFrozen(request: FrozenRequest) {
}
/** Operations/conditional fields are sent; full snapshots are only reads and ACK participant views. */
/** 已经协调完成的本地压缩回执，仅保留身份与版本证据；不含可再次合图的原始模型。
* localReceiptVersion 与远端 protocol_version 是独立版本轴，不能拿摘要再次模拟原服务回执。 */
export type FoundryResolvedReceiptSummary={readonly localReceiptFormat:'foundry.reconciled-document';readonly localReceiptVersion:1;readonly protocol_version:2;readonly request_id:Id;readonly status:'applied';readonly operation_ids:readonly Id[];readonly projectId:Id;readonly versions:Readonly<Record<Id,number>>};
// 压缩只允许认识的封闭字段集合；遇到扩展字段保留原回执，避免丢失未知证据。
const receiptKeys=new Set(['protocol_version','request_id','status','operation_ids','model']);
const modelKeys=new Set(['documentFormatVersion','projectId','complete','canvases','elements','tombstones','versions','removedCanvasIds']);
const elementKeys=new Set(['id','incarnation','ownerCanvasId','componentId','dataVersion','structure','geometry','data','source','target','path']);
/** 只检查对象自有可枚举字段，发现任何未知项立即停止有损摘要化。 */
function hasExtraKeys(value:Record<string,unknown>,known:ReadonlySet<string>){for(const key in value)if(Object.hasOwn(value,key)&&!known.has(key))return true;return false;}
/** 识别已经压缩的本地摘要并拒绝重解释；未知摘要版本给出独立解释器错误。 */
function assertRawReceipt(value:unknown){if(value&&typeof value==='object'&&!Array.isArray(value)&&(value as Record<string,unknown>).localReceiptFormat==='foundry.reconciled-document'&&!Object.hasOwn(value,'model')){if((value as Record<string,unknown>).localReceiptVersion!==1)throw fault('receipt_summary_interpreter_unavailable');throw fault('receipt_already_reconciled');}}
/** 绑定一个工作区/根项目的文档协议适配器，支持单/批量增量操作、原请求查询与完整权威读取。
* resolvedReceiptPolicy 默认保留 raw，显式 summary 才安装协调后压缩能力。 */
export function createFoundryPersistence(options: { workspaceId: Id; rootId: Id; registry: Registry; transport: ServiceTransport;resolvedReceiptPolicy?:'raw'|'summary' }): PersistencePort {
if (!uuid.test(options.workspaceId) || !uuid.test(options.rootId)) throw fault('invalid_service_scope');
if(options.resolvedReceiptPolicy!==undefined&&!['raw','summary'].includes(options.resolvedReceiptPolicy))throw fault('invalid_receipt_policy');
const path = pathFor(options.workspaceId, options.rootId);
/** 统一发送/查询已有冻结请求。传输/解释失败保持 unresolved，带可识别错误码时按规则阻塞。
* 不在异常中宣称服务端回滚，也不分配新幂等键来替代原请求。 */
async function execute(request: FrozenRequest, query: boolean, signal: AbortSignal): Promise<OperationStatus> {
try {
validateFrozen(request);
@@ -146,6 +193,8 @@ export function createFoundryPersistence(options: { workspaceId: Id; rootId: Id;
return { outcome: 'unresolved', error: code ?? 'transport_unknown', ...(code && code !== 'invalid_canonical_response' ? { blocked: true } : {}) };
}
}
/** 编译最多64个严格递增的 document.change v1 意图；操作ID不被合成批次ID替换。
* 参与 canvas 版本来自 ACK；仅本批明确新建的 canvas 可用零基线，已存在者缺版本必须拒绝。 */
function compileBatch(intents: readonly OperationIntent[], requestId: Id, snapshot: EngineSnapshot) {
if (!intents.length || intents.length > 64 || snapshot.document.projectId !== options.rootId || !uuid.test(requestId) || new Set(intents.map(intent => intent.id)).size !== intents.length) throw fault('operation_interpreter_unavailable');
for (const intent of intents) if (!uuid.test(intent.id) || intent.operationType !== 'document.change' || intent.operationSchemaVersion !== 1) throw fault('operation_interpreter_unavailable');
@@ -157,12 +206,16 @@ export function createFoundryPersistence(options: { workspaceId: Id; rootId: Id;
expectedVersions[canvas] = version ?? 0;
}
// The body is the durable coverage mapping. Never replace logical IDs with a synthetic batch ID.
// projectBatch 基于 ACK 与 ledger 合并条件字段，同时保留逻辑覆盖；JSON复制使描述脱离调用方引用。
const payload = JSON.parse(JSON.stringify({ protocol_version: 2, request_id: requestId, expected_versions: expectedVersions, operations: projectBatch(intents, snapshot.runtime.acknowledged, snapshot.runtime.ledger, canvasIds) })) as Json;
if (!validJson(payload)) throw fault('invalid_operation_payload');
return freeze({ protocolVersion: 2, canvasIds, expectedVersions, payload });
}
return Object.freeze({
/** 此适配器仅解释远端协议v2，恢复的其它版本不能猜测着继续发送。 */
supportsRequest: (request: FrozenRequest) => request.protocolVersion === 2,
/** 仅在调用方已协调成功且请求resolved/applied后尝试摘要。
* 检查参与版本与已采用结果一致，未知字段/未知组件/额外格式信息一律保留原始回执。 */
...(options.resolvedReceiptPolicy==='summary'?{compactResolvedReceipt(request:FrozenRequest,receipt:Json,applied:import('../canvas-runtime/contracts.ts').CanonicalProposal):Json{
if(request.kind!=='document.change'||request.protocolVersion!==2||request.phase!=='resolved'||request.outcome!=='applied')return receipt;
validateFrozen(request);const body=record(receipt);
@@ -174,12 +227,14 @@ export function createFoundryPersistence(options: { workspaceId: Id; rootId: Id;
if(hasExtraKeys(model,modelKeys)||model.documentFormatVersion!==1||model.projectId!==options.rootId||model.complete!==false)return receipt;
const resultVersions=versions(model.versions);
if(resultVersions.size!==request.canvasIds.length||request.canvasIds.some(id=>!resultVersions.has(id)||(resultVersions.get(id)??0)<=request.expectedVersions[id]||resultVersions.get(id)!==applied.versions.get(id)))return receipt;
// 原始元素含未知类型或扩展字段时可能仍有将来恢复所需信息，不能为了省空间丢掉。
for(const value of list(model.elements)){const element=record(value);if(options.registry.resolve(element as unknown as ElementRecord)!=='known'||hasExtraKeys(element,elementKeys))return receipt;}
const canvasKeys=new Set(['id','parent']),parentKeys=new Set(['canvasId','entryElementId']);
for(const value of list(model.canvases)){const canvas=record(value);if(hasExtraKeys(canvas,canvasKeys)||canvas.parent!==null&&hasExtraKeys(record(canvas.parent),parentKeys))return receipt;}
return freeze({localReceiptFormat:'foundry.reconciled-document',localReceiptVersion:1,protocol_version:2,request_id:request.id,status:'applied',operation_ids:covered,projectId:options.rootId,versions:Object.fromEntries(resultVersions)});
}}:{}),
/** 独立验证请求已生效证据，要求每个参与者版本恰为请求基线+1；返回对后续权威读取的断言。 */
verifyApplied(request:FrozenRequest,receipt:Json){
assertRawReceipt(receipt);
validateFrozen(request);
@@ -188,16 +243,19 @@ export function createFoundryPersistence(options: { workspaceId: Id; rootId: Id;
for(const [canvas,version]of proof.versions)if(version!==request.expectedVersions[canvas]+1)throw fault('receipt_version_mismatch',[canvas]);
return proof;
},
/** 必须取得本根项目的完整新格式文档；迁移提示返回明确错误，留给迁移流程处理。 */
async read(signal: AbortSignal) {
const loaded=await loadFoundryModel(options.transport,options.workspaceId,options.rootId,options.registry,signal);
if(loaded.status!=='ready')throw fault('canonical_migration_required');
if(loaded.document.projectId!==options.rootId)throw fault('project_mismatch');
return loaded;
},
// 单操作复用批量编译器；submit/query 分别以POST原body和GET原requestId走同一结果解释器。
compile: (intent: OperationIntent, requestId: Id, snapshot: EngineSnapshot) => compileBatch([intent], requestId, snapshot),
compileBatch,
submit: (request: FrozenRequest, signal: AbortSignal) => execute(request, false, signal),
query: (request: FrozenRequest, signal: AbortSignal) => execute(request, true, signal),
/** 从原始回执提取参与者模型，合成新的 ACK 候选；不接受本地已压缩摘要代替原模型。 */
canonical(request: FrozenRequest, status: OperationStatus, snapshot: EngineSnapshot) {
assertRawReceipt(status.canonical);const body = record(status.canonical);
if (body.request_id !== request.id || body.protocol_version !== 2) throw fault('receipt_identity_mismatch');
@@ -207,9 +265,12 @@ export function createFoundryPersistence(options: { workspaceId: Id; rootId: Id;
}
/** Verify coverage independently of the current ACK graph. Only a complete fresh model is adopted. */
/** 在不依赖当前 ACK 图的情况下检查回执覆盖，生成 assertCanonical。
* 随后的完整权威读取不得落后于已生效版本；同版本必须内容相等，更高版本可含后续修改。 */
export function verifyFoundryModelProof(value:unknown,participants:readonly Id[],rootId:Id):AppliedModelProof {
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');
/** 断言只验证采用读取是否包含既有证据，不在这里把回执片段直接变成新 working。 */
return Object.freeze({versions:chunk.versions,assertCanonical(proposal){
if(proposal.document.projectId!==rootId)throw fault('project_mismatch');
const unchanged=new Set<Id>(),canvases=new Map(chunk.canvases.map(canvas=>[canvas.id,canvas]));
@@ -219,6 +280,7 @@ export function verifyFoundryModelProof(value:unknown,participants:readonly Id[]
if(!equal(canvases.get(canvas),proposal.document.canvases.get(canvas)))throw fault('canonical_version_mismatch',[canvas]);
unchanged.add(canvas);
}
// 对仍处于同版本的 canvas 逐元素比较，既不能缺元素也不能多元素；墓碑只能单调推进。
const expected=new Map(chunk.elements.filter(element=>unchanged.has(element.ownerCanvasId)).map(element=>[element.id,element]));
let count=0;for(const element of proposal.document.elements.values())if(unchanged.has(element.ownerCanvasId)){count++;if(!equal(expected.get(element.id),element))throw fault('canonical_version_mismatch',[element.ownerCanvasId]);}
if(count!==expected.size)throw fault('canonical_version_mismatch');
@@ -227,12 +289,15 @@ export function verifyFoundryModelProof(value:unknown,participants:readonly Id[]
}
/** Shared participant replacement for document receipts and canonical run events. */
/** 文档回执/普通生成事件的参与者替换：仅替换声明画布及其成员，非参与画布沿用旧 ACK。
* 全树重新验证后返回候选，runtime 另行检查版本推进并叠回本地未决操作。 */
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);
// 先移除旧参与者成员，再加入回执成员；与非参与者现有ID碰撞则拒绝，不能跨范围覆盖。
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));
@@ -243,10 +308,13 @@ export function mergeFoundryModel(value: unknown, participants: readonly Id[], s
/** A relocated result carries a complete ownership context. Retain later ACKs
* per canvas; validate the combined tree before publishing any ownership change. */
/** 生成结果改变归属时采用带完整树的上下文，但逐 canvas 保留比事件更新的 ACK。
* 同版本不同内容是协议冲突；合并树和墓碑再 validate，不能逐节点发布一半的归属变化。 */
export function mergeFoundryRunContext(value: unknown, snapshot: EngineSnapshot, registry: Registry, rootId: Id) {
const chunk=modelChunk(value);
if(!chunk.complete||chunk.projectId!==rootId)throw fault('invalid_run_context');
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),replace=new Set<Id>();
// 事件版本低于当前ACK则不替换该canvas；相等时先排序核对成员与canvas记录完全一致。
for(const [id,version] of chunk.versions)if(version>=(snapshot.runtime.versions.get(id)??0)){
if(version===snapshot.runtime.versions.get(id)){
const incoming=chunk.elements.filter(element=>element.ownerCanvasId===id).sort((a,b)=>a.id.localeCompare(b.id)),known=[...previous.elements.values()].filter(element=>element.ownerCanvasId===id).sort((a,b)=>a.id.localeCompare(b.id));
@@ -254,6 +322,7 @@ export function mergeFoundryRunContext(value: unknown, snapshot: EngineSnapshot,
}
replace.add(id);serverVersions.set(id,version);canvases.delete(id);
}
// 只替换通过版本选择的参与者，加入跨owner元素仍要检测与保留部分的身份碰撞。
for(const canvas of chunk.canvases)if(replace.has(canvas.id))canvases.set(canvas.id,canvas);
for(const [id,element] of elements)if(replace.has(element.ownerCanvasId))elements.delete(id);
for(const element of chunk.elements)if(replace.has(element.ownerCanvasId)){if(elements.has(element.id))throw fault('canonical_scope_mismatch',[element.id]);elements.set(element.id,element);}
src/lib/canvas-service/index.ts
+6
View File
@@ -1,5 +1,11 @@
/**
* Foundry 服务适配层的公共入口，暴露文档协议、资源服务、生成任务和模型加载器。
* 这些模块解释服务端数据与请求，不拥有 editor 状态根、history 或调度队列。
*/
// 底层 transport 与文档增量协议；资源上传和生成各有独立适配器。
export * from './foundry.ts';
export * from './assets.ts';
export * from './runs.ts';
// 共享 loader 使用的流/完整 JSON 协商入口，内部 model-stream 细节不在此单独重导出。
export * from './model-loader.ts';
src/lib/canvas-service/model-loader.ts
+11
View File
@@ -1,3 +1,7 @@
/**
* Foundry 模型读取适配器，把一次服务端读取转换成 runtime LoadPort 的带作用域分块结果。
* 优先协商 NDJSON 流，旧服务器明确返回 JSON 时复用完整模型解释器；不会为了回退再发一次相同请求。
*/
import { decodeModelStream } from './model-stream.ts';
import type { Id, Registry } from '../canvas-core/types.ts';
import { fault } from '../canvas-core/values.ts';
@@ -6,23 +10,30 @@ import { loadQueryKey, type LoadPort } from '../canvas-runtime/loader.ts';
import { loadFoundryModel, type ServiceTransport } from './foundry.ts';
/** Negotiate real streaming; older servers may explicitly return complete JSON. */
/** 绑定工作区和 registry，返回异步分块源；只接受 Foundry canonical-project 查询范围。 */
export function createFoundryModelLoader(options: { workspaceId: Id; registry: Registry; transport: ServiceTransport }): LoadPort<CanonicalReadUpdate> {
return async function* (scope, signal) {
// query scope 是缓存/权限隔离依据，不能用某工作区的 transport 为另一种查询来源提供结果。
if (scope.workspaceId !== options.workspaceId || scope.format !== 'canonical-model-v1' || scope.source !== 'foundry' || scope.viewKey !== 'canonical-project') throw fault('invalid_model_load_scope');
const response=await options.transport({method:'GET',path:`workspaces/${encodeURIComponent(options.workspaceId)}/materials/${encodeURIComponent(scope.canvasId)}/canvas/model`,signal,responseFormat:'model-stream-v1'});
// 流解析器控制序列、来源 token 和 EOF；适配器再补权限代次与 queryKey，供共享 loader 拒绝过期结果。
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;
}
// JSON 回退仅把已有 response 注入解释器，ready 与 migration-required 是不同结果，不能把迁移要求当空文档。
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;
// 对版本排序以稳定完整 JSON 回退的快照标记；这是结果身份，不是原响应字节的密码学摘要。
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.
// byteLength 是已解析后重新序列化的 payload 成本，不能据此宣称 JSON.parse 前已有网络/堆内存硬上限。
const byteLength = new TextEncoder().encode(JSON.stringify(payload)).byteLength;
// 完整 JSON 一次给出 complete；迁移结果使用 migration 覆盖标记，不伪造已加载 canvas 内容。
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 });
};
src/lib/canvas-service/model-stream.ts
+26
View File
@@ -1,40 +1,59 @@
/**
* Foundry NDJSON 模型流的帧与语义校验。header 声明固定项目/版本/可选归属和数量，elements 分批推进接收，
* complete 记录之后还必须真正读到 EOF，才发布 ready，避免把尾部追加/截断的数据误认成完整模型。
* 接收前缀仅是可供预览层继续校验的分块，整份文档的 schema/引用/树一致性在终结时统一验证。
*/
import {validateElementOwners,type ElementOwners,validateCanvasElementCounts,type CanvasElementCounts} from '../canvas-runtime/scope-counts.ts';
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';
/** 与 transport 协商使用的精确媒体类型和流版本，独立于 documentFormatVersion。 */
export const modelStreamType = 'application/vnd.foundry.canvas-model+ndjson;version=1';
// 整流64MiB、单行8MiB、最多4096行；先对原始字节计数，再解析记录，元素批次另限64项。
const maxBytes=64*1024*1024,maxLine=8*1024*1024,maxLines=4096;
// 项目/canvas ID 采用服务端 UUID 约束，元素 ID 则按记录协议单独验证。
const uuid=/^[a-f\d]{8}-[a-f\d]{4}-[a-f\d]{4}-[a-f\d]{4}-[a-f\d]{12}$/i;
/** 外部记录必须是有限 JSON 对象，拒绝数组和非 JSON 值，不靠类型断言信任网络内容。 */
const object=(value:unknown):Record<string,unknown>=>{if(!value||typeof value!=='object'||Array.isArray(value)||!validJson(value))throw fault('invalid_model_stream');return value as Record<string,unknown>;};
/** Bounded NDJSON framing. EOF is required even after a terminal record. */
/** 有界 UTF-8 NDJSON 分帧。按字节找换行，fatal 解码拒绝损坏编码，最后一条也要求换行结束。 */
async function* lines(source:AsyncIterable<Uint8Array>,signal:AbortSignal){
const decoder=new TextDecoder('utf-8',{fatal:true});
/** 统一把解码错误转为协议错误，保持错误码不依赖浏览器具体异常文本。 */
const decode=(bytes?:Uint8Array,stream=false)=>{try{return decoder.decode(bytes,{stream});}catch{throw fault('invalid_model_stream');}};
let pending='',total=0,lineBytes=0,count=0;
for await(const bytes of source){
// 每段先检查取消、字节类型和整流预算，不因 transport 给了大 chunk 就跳过预算。
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;index<bytes.length;index++)if(bytes[index]===10){
lineBytes+=index-start;if(lineBytes>maxLine)throw fault('model_stream_budget');
// 在换行处 flush UTF-8 解码器，完整一行才 JSON.parse，输出成本包含该换行字节。
pending+=decode(bytes.subarray(start,index),true);pending+=decode();if(++count>maxLines||!pending)throw fault('invalid_model_stream');
let raw:unknown;try{raw=JSON.parse(pending);}catch{throw fault('invalid_model_stream');}
yield{raw,byteLength:lineBytes+1};pending='';lineBytes=0;start=index+1;
}
lineBytes+=bytes.length-start;if(lineBytes>maxLine)throw fault('model_stream_budget');pending+=decode(bytes.subarray(start),true);
}
// 真正 EOF 时必须没有未结束行或残余 UTF-8 字节；不能默默接受截断末尾。
signal.throwIfAborted();pending+=decode();if(pending||lineBytes)throw fault('incomplete_model_stream');
}
/** 解释一条固定快照流。sequence 从0连续增长，snapshotToken 全程相同；
* header/元素批次可先 yield，终结结果暂存到 EOF，之后多一条记录也判非法。 */
export async function* decodeModelStream(source:AsyncIterable<Uint8Array>,canvasId:Id,registry:Registry,signal:AbortSignal){
let sequence=0,token='',projectId='',expected=0,ended=false;
// terminal 暂存而不立即 yield，保证 complete 之后的恶意/错误尾部仍能阻止 ready。
let terminal: {sequence:number;snapshotToken:string;projectId:string;versions:ReadonlyMap<Id,number>;complete:boolean;byteLength:number;coverage:string[];value:CanonicalReadUpdate}|undefined;
let canvases:CanvasRecord[]=[],versions:ReadonlyMap<Id,number>=new Map(),tombstones:ReadonlyMap<Id,number>=new Map();
let elementOwners:ElementOwners|undefined;let canvasElementCounts:CanvasElementCounts|undefined;const ownerCounts=new Map<Id,number>();
// 元素累积用于最终构建文档；流式显示不等于恒定内存，完整模型仍需保存在最终文档中。
const elements:ElementRecord[]=[],ids=new Set<Id>(),canvasIds=new Set<Id>();
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');
// 第一条必须是 header，数量及 canvas 集合受限；请求画布和根项目均须出现在这份树中。
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[];
@@ -43,20 +62,26 @@ export async function* decodeModelStream(source:AsyncIterable<Uint8Array>,canvas
const versionMap=new Map<Id,number>(),dead=new Map<Id,number>();
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');
// versions 精确覆盖当前 canvas 加已删除 canvas，禁止靠缺项或多余版本模糊覆盖范围。
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);
// 数量/所有权证据是可选协议能力；elementOwners 依赖 counts，必须与总数及各 canvas 数量匹配。
if(Object.hasOwn(raw,'canvasElementCounts')){validateCanvasElementCounts(raw.canvasElementCounts,[...canvasIds],expected);canvasElementCounts=freeze({...raw.canvasElementCounts});}
if(Object.hasOwn(raw,'elementOwners')){if(!canvasElementCounts)throw fault('invalid_model_identity');validateElementOwners(raw.elementOwners,canvasElementCounts,expected);elementOwners=freeze({...raw.elementOwners});}
const value:CanonicalReadUpdate=freeze({status:'receiving',...(elementOwners?{elementOwners}:{}),...(canvasElementCounts?{canvasElementCounts}:{}),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;
}
// 跨批次换快照必须整体重读，不能把来自两个版本的元素拼接成一个 ready 模型。
if(raw.snapshotToken!==token)throw fault('load_snapshot_changed');
// 非空批次最多64项，不得超过 header 总数；按元素身份、owner 及可选数量/归属声明核对。
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');const count=(ownerCounts.get(element.ownerCanvasId)??0)+1;if(canvasElementCounts&&count>canvasElementCounts[element.ownerCanvasId])throw fault('invalid_model_coverage');ownerCounts.set(element.ownerCanvasId,count);if(elementOwners&&(!Object.hasOwn(elementOwners,element.id)||elementOwners[element.id]!==element.ownerCanvasId))throw fault('invalid_model_identity');ids.add(element.id);elements.push(element);}
// 后续 receiving 只交本批新增元素，不重复传播整份前缀；保留累计进度供显示。
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};
// 总元素数一致才构建并校验完整文档；只在校验成功后设 ended，随后必须等 EOF。
}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;
@@ -64,6 +89,7 @@ export async function* decodeModelStream(source:AsyncIterable<Uint8Array>,canvas
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');
}
// 流正常断开却没有 complete 仍是失败，不能用“已收到一些元素”冒充加载完成。
if(!ended||!terminal)throw fault('incomplete_model_stream');
yield terminal;
}
src/lib/canvas-service/runs.ts
+42
View File
@@ -1,3 +1,8 @@
/**
* Foundry生成任务协议适配。启动受理、执行事件、取消和用户应用输出是不同阶段，分别验证身份与作用范围。
* 受理成功映射为请求applied，只代表启动动作已经生效，不代表付费生成已经完成。
* 原请求/原计划/目标代次贯穿事件追尾，模型结果仅作为ACK提案交回runtime协调。
*/
import type { Id, Json, Registry, RunInput } from '../canvas-core/types.ts';
import { equal, fault, freeze, readMap, validJson } from '../canvas-core/values.ts';
import { validateRunInput } from '../canvas-runtime/runs.ts';
@@ -6,13 +11,21 @@ import type { FrozenRequest } from '../canvas-runtime/ledger.ts';
import type { OperationStatus } from '../canvas-runtime/scheduler.ts';
import { mergeFoundryModel, mergeFoundryRunContext, verifyFoundryModelProof, type ServiceTransport } from './foundry.ts';
// 服务作用域及请求/任务/输出等字段按各处要求使用UUID，不用同名字符串推断身份。
const uuid = /^[a-f\d]{8}-[a-f\d]{4}-[a-f\d]{4}-[a-f\d]{4}-[a-f\d]{12}$/i;
/** 执行状态白名单；与OperationStatus的请求结果状态不是同一状态机。 */
const statuses = new Set(['queued','running','succeeded','failed','conflict','cancelled','partially_succeeded']);
/** 只接受对象外壳，完整JSON及字段语义在具体回执/事件处验证。 */
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; };
/** 读取非空字符串；UUID等更强要求由字段使用处检查。 */
const string = (value: unknown): string => { if (typeof value !== 'string' || !value) throw fault('invalid_run_response');return value; };
/** 游标/版本/代次只接收安全整数，默认允许零，版本/代次调用时提高下限为1。 */
const integer = (value: unknown, min=0): number => { if (typeof value!=='number'||!Number.isSafeInteger(value)||value<min)throw fault('invalid_run_response');return value; };
/** 校验冻结请求的协议、owner、单操作覆盖及载荷；cancel无文档版本，
* apply只写目标画布，start则包含生成输入上下文的预期版本，并复用输入大小校验。 */
function payload(request: FrozenRequest) {
const raw=object(JSON.parse(request.body));
if(!request.ownerCanvasId||!uuid.test(request.ownerCanvasId)||!['run.start','run.cancel','run.apply'].includes(request.kind??'')||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)))throw fault('invalid_run_request');
@@ -31,6 +44,7 @@ function payload(request: FrozenRequest) {
}
return raw;
}
/** 检查计划步骤非空、step_id唯一、同一根项目及有效目标代次/画布；不在客户端重新规划执行DAG。 */
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);}
@@ -38,39 +52,52 @@ function planSteps(plan: unknown, rootId: Id) {
}
/** Admission completion is distinct from paid execution completion. Credentials remain in the transport. */
/** 为一个根项目提供模型目录、请求编译/结果查询、受理回执和事件读取的RunPort。 */
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');
/** 所有任务端点以其入口画布定位，即使输出目标后来移到另一canvas也保留原run所属board。 */
const path=(board: Id)=>`workspaces/${encodeURIComponent(options.workspaceId)}/materials/${encodeURIComponent(board)}/canvas/ai`;
/** 验证启动受理：请求/操作/目标代次、queued任务ID、owner版本+1及冻结计划均匹配。
* frozen_versions仅列实际读入计划的快照；启动预期版本向量则覆盖所有活动canvas。 */
function admissionReceipt(request:FrozenRequest,receipt:Json){
const raw=payload(request),body=object(receipt),runId=string(body.run_id),board=request.ownerCanvasId!;
if(request.kind!=='run.start'||body.protocol_version!==2||body.request_id!==request.id||body.operation_id!==raw.operation_id||body.admission!=='accepted'||body.action!==undefined&&body.action!=='run.start'||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),frozen=object(object(body.plan).frozen_versions),expected={...request.expectedVersions,[board]:request.expectedVersions[board]+1};
// The admission vector covers all active canvases; the execution plan lists only snapshots actually read.
if(frozen[board]!==expected[board]||Object.entries(frozen).some(([id,version])=>!Object.hasOwn(expected,id)||version!==expected[id])||!steps.some(step=>step.board_id===board&&step.node_id===raw.node_id&&step.target_incarnation===raw.target_incarnation)||steps.some(step=>!Object.hasOwn(frozen,step.board_id as Id)))throw fault('run_identity_mismatch');
// 受理只对owner画布给出模型生效证明，并确认计划入口目标仍为同一元素代次。
const model=verifyFoundryModelProof(body.model,[board],options.rootId),target=array(object(body.model).elements).map(object).find(element=>element.id===raw.node_id);
if(model.versions.get(board)!==body.canvas_version||!target||target.ownerCanvasId!==board||target.incarnation!==raw.target_incarnation)throw fault('run_identity_mismatch');
// 新任务从游标0、hasMore=true开始；回执里的plan是后续poll必须保持一致的身份事实。
const admission:RunRecord=freeze({id:runId,requestId:request.id,operationId:request.operationIds[0],boardId:board,target:{id:raw.node_id as Id,incarnation:raw.target_incarnation as number},plan:body.plan as Json,status:'queued',cursor:0,hasMore:true,steps:[],outputs:[],outputEvents:[],attempt:0});
return{model,admission,rawModel:body.model};
}
/** 验证控制动作回执。cancel只返回任务状态与空文档证明；apply还需目标输出/代次及画布版本+1。 */
function actionReceipt(request:FrozenRequest,receipt:Json){
const raw=payload(request),body=object(receipt);
if(!request.runId||body.protocol_version!==2||body.request_id!==request.id||body.operation_id!==request.operationIds[0]||body.run_id!==request.runId||body.action!==request.kind||body.status!=='applied')throw fault('run_identity_mismatch');
// 取消回执不伪造文档改动，后续仍由runtime追读任务尾页。
if(request.kind==='run.cancel'){
if(!statuses.has(string(body.run_status)))throw fault('invalid_run_response');
return{action:{runId:request.runId,runStatus:body.run_status as RunStatus},model:{versions:readMap(new Map<Id,number>()),assertCanonical(){}},rawModel:undefined};
}
if(request.kind!=='run.apply'||body.node_id!==raw.node_id||body.output_id!==raw.output_id||body.board_id!==raw.target_canvas_id||body.target_incarnation!==raw.target_incarnation||body.document_format_version!==1)throw fault('run_identity_mismatch');
const source=object(body.model),board=raw.target_canvas_id as Id;
// 若服务给完整上下文，只抽取本次apply目标画布的成员/版本作局部证明，不声明整树都由此动作改写。
const scoped=source.complete===true?{...source,complete:false,canvases:array(source.canvases).filter(value=>object(value).id===board),elements:array(source.elements).filter(value=>object(value).ownerCanvasId===board),versions:{[board]:object(source.versions)[board]},tombstones:{},removedCanvasIds:[]}:source;
const model=verifyFoundryModelProof(scoped,[board],options.rootId),version=integer(body.canvas_version,1),target=array(scoped.elements).map(object).find(element=>element.id===raw.node_id);
if(model.versions.get(board)!==version||version!==request.expectedVersions[board]+1||target?.incarnation!==raw.target_incarnation||target?.ownerCanvasId!==board)throw fault('run_identity_mismatch');
return{action:{runId:request.runId,outputId:raw.output_id as Id},model,rawModel:scoped};
}
/** submit/query共用解释路径，查询始终用原requestId。终结not-applied须完整匹配动作身份，
* unknown/传输异常保持不确定；启动的accepted仅结清受理动作，执行完成另看事件页。 */
async function execute(request: FrozenRequest,query: boolean,signal: AbortSignal): Promise<OperationStatus> {
try{
const raw=payload(request),suffix=request.kind==='run.start'?'runs':`runs/${encodeURIComponent(request.runId!)}/${request.kind==='run.cancel'?'cancel':'apply'}`;
// 非cancel写入同时带目标版本条件；query不重新携带写入body或创建新幂等身份。
const response=await options.transport({responseFormat:'canvas-v2',method:query?'GET':'POST',path:`${path(request.ownerCanvasId!)}/${query?`operations/${encodeURIComponent(request.id)}`:suffix}`,requestId:query?undefined:request.id,expectedVersion:query||request.kind==='run.cancel'?undefined:request.expectedVersions[request.kind==='run.apply'?raw.target_canvas_id as Id:request.ownerCanvasId!],body:query?undefined:request.body,signal});
const body=object(response.body),matching=body.protocol_version===2&&body.request_id===request.id;
// 只有匹配请求/操作/动作及适用runId的terminal证据，才能认定本次没有生效。
if(matching&&body.operation_id===request.operationIds[0]&&body.action===request.kind&&(request.kind==='run.start'||body.run_id===request.runId)&&body.status==='not-applied'&&body.terminal===true&&validJson(body))return freeze({outcome:'not-applied',error:typeof body.error==='string'?body.error:'not_applied',canonical:JSON.parse(JSON.stringify(body)) as Json});
if(matching&&body.operation_id===request.operationIds[0]&&body.action===request.kind&&body.status==='applied'&&request.kind!=='run.start'&&response.status>=200&&response.status<300&&validJson(body))return freeze({outcome:'applied',canonical:JSON.parse(JSON.stringify(body)) as Json});
if(request.kind==='run.start'&&(body.action===undefined||body.action==='run.start')&&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});
@@ -79,6 +106,7 @@ export function createFoundryRuns(options: {workspaceId: Id; rootId: Id; registr
}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({responseFormat:'canvas-v2',method:'GET',path:`${path(boardId)}/models`,signal}),body=object(response.body);
@@ -86,11 +114,13 @@ export function createFoundryRuns(options: {workspaceId: Id; rootId: Id; registr
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}:{})});
},
/** 从逻辑意图及当前ACK编译请求；不发送网络，也不用尚未确认的working目标替代服务端基线。 */
compile(intent,requestId,snapshot){
if(intent.operationType==='run.cancel'||intent.operationType==='run.apply'){
const action=intent.action,run=snapshot.runtime.runs.get(action.runId);
if(!run||run.boardId!==action.boardId||!uuid.test(intent.id)||!uuid.test(requestId))throw fault('run_identity_mismatch');
const base={protocol_version:2,request_id:requestId,operation_id:intent.id};
// cancel沿原run入口寻址且不写canvas；apply则取当前ACK中的目标owner与版本。
if(intent.operationType==='run.cancel')return freeze({kind:'run.cancel',ownerCanvasId:run.boardId,runId:run.id,protocolVersion:2,canvasIds:[],expectedVersions:{},payload:base});
const target=snapshot.runtime.acknowledged.elements.get(intent.action.target.id);
if(!target||target.incarnation!==intent.action.target.incarnation||target.ownerCanvasId!==intent.action.targetCanvasId)throw fault('target_changed');
@@ -101,42 +131,54 @@ export function createFoundryRuns(options: {workspaceId: Id; rootId: Id; registr
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');
// 启动范围须与入账时的活动canvas集合一致；期间树变了就拒绝旧上下文，不能悄悄扩大生成输入。
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}});
},
// 请求完成与服务端任务执行分离：提交/查询后，accept再核对受理或控制语义。
submit:(request,signal)=>execute(request,false,signal),query:(request,signal)=>execute(request,true,signal),
/** 恢复路径也使用相同回执解释器，生成可对照fresh权威模型的证明。 */
verifyApplied(request,receipt){return request.kind==='run.start'?admissionReceipt(request,receipt):actionReceipt(request,receipt);},
/** 受理回执转为RunRecord与owner ACK候选，实际根发布/本地重放交给engine。 */
accept(request,status,snapshot){
if(status.outcome!=='applied'||status.canonical===undefined)throw fault('run_identity_mismatch');
const parsed=admissionReceipt(request,status.canonical),proposal=mergeFoundryModel(parsed.rawModel,[request.ownerCanvasId!],snapshot,options.registry,options.rootId);
return{run:parsed.admission,proposal};
},
/** 控制回执返回任务状态或输出身份；只有apply带局部ACK提案，cancel不修改文档。 */
acceptAction(request,status,snapshot){
if(status.outcome!=='applied'||status.canonical===undefined)throw fault('run_identity_mismatch');
const parsed=actionReceipt(request,status.canonical);
return{...parsed.action,...(parsed.rawModel?{proposal:mergeFoundryModel(parsed.rawModel,request.canvasIds,snapshot,options.registry,options.rootId)}:{})};
},
/** 从保存游标之后读取最多100条有序事件。验证任务/固定计划、连续seq及迁移后的目标上下文，
* 只返回一页数据，不在适配器推进持久游标；runtime协调成功才与文档一起发布next。 */
async poll(run,signal){
const response=await options.transport({responseFormat:'canvas-v2',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.node_id!==event.node_id)throw fault('run_identity_mismatch');
// 元素跨分区后需携原画布、共同根和代次证据；已应用迁移事件还必须附完整所有权上下文。
const relocated=step.board_id!==event.board_id;
if(relocated&&(!uuid.test(string(event.board_id))||event.origin_board_id!==step.board_id||event.canonical_root!==options.rootId||event.target_incarnation!==step.target_incarnation))throw fault('run_identity_mismatch');
if(event.type==='step.applied'&&(step.target_incarnation!==event.target_incarnation||event.document_format_version!==1||relocated&&object(event.model).complete!==true))throw fault('run_identity_mismatch');
}
}
// next必须恰为最后事件序号；hasMore却没有任何事件会造成空转，作为非法响应拒绝。
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}:{})});
},
/** 解释可合图事件：目标代次必须仍有效，局部结果只能改本owner，完整结果才允许跨分区归属变化。
* 合并后再检查事件目标的owner/代次和版本，防止合法外壳里夹带另一目标的模型。 */
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));
const context=object(event.model).complete===true;