From 450fa3f7193eabab060a4fcb11db56dd04954e09 Mon Sep 17 00:00:00 2001 From: Codex Bot Date: Thu, 12 Nov 2026 01:20:49 +0800 Subject: [PATCH] docs(canvas): explain Foundry transport model streams and service receipts --- src/lib/canvas-service/assets.ts | 21 ++++++++ src/lib/canvas-service/foundry.ts | 69 ++++++++++++++++++++++++++ src/lib/canvas-service/index.ts | 6 +++ src/lib/canvas-service/model-loader.ts | 11 ++++ src/lib/canvas-service/model-stream.ts | 26 ++++++++++ src/lib/canvas-service/runs.ts | 42 ++++++++++++++++ 6 files changed, 175 insertions(+) diff --git a/src/lib/canvas-service/assets.ts b/src/lib/canvas-service/assets.ts index beab086..9e36c1a 100644 --- a/src/lib/canvas-service/assets.ts +++ b/src/lib/canvas-service/assets.ts @@ -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 { if (!value || Array.isArray(value) || typeof value !== 'object') throw fault('invalid_asset_response'); return value as Record; } +/** 交叉检查冻结请求与元数据载荷：一个上传操作、无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), diff --git a/src/lib/canvas-service/foundry.ts b/src/lib/canvas-service/foundry.ts index e70ceb6..c9eb650 100644 --- a/src/lib/canvas-service/foundry.ts +++ b/src/lib/canvas-service/foundry.ts @@ -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; readonly retryAfterMs?: number }; +/** 各服务共用的可注入传输端口，测试可模拟；调度、并发和可靠保存不由此类型实现。 */ export type ServiceTransport = (request: ServiceRequest) => Promise; +// 服务协议对账户、工作区、项目、请求等 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 => { if (!value || Array.isArray(value) || typeof value !== 'object') throw fault('invalid_canonical_response'); return value as Record; }; +/** 只接受实际数组，禁止缺失字段被当作空列表。 */ 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 { const result = new Map(); 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[]= [];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; versions: ReadonlyMap; 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 { 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; 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>}; +// 压缩只允许认识的封闭字段集合；遇到扩展字段保留原回执，避免丢失未知证据。 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,known:ReadonlySet){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).localReceiptFormat==='foundry.reconciled-document'&&!Object.hasOwn(value,'model')){if((value as Record).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 { 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(),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(); + // 事件版本低于当前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);} diff --git a/src/lib/canvas-service/index.ts b/src/lib/canvas-service/index.ts index 0b6f790..77cdd88 100644 --- a/src/lib/canvas-service/index.ts +++ b/src/lib/canvas-service/index.ts @@ -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'; diff --git a/src/lib/canvas-service/model-loader.ts b/src/lib/canvas-service/model-loader.ts index ec44bea..b240d1c 100644 --- a/src/lib/canvas-service/model-loader.ts +++ b/src/lib/canvas-service/model-loader.ts @@ -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 { 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 }); }; diff --git a/src/lib/canvas-service/model-stream.ts b/src/lib/canvas-service/model-stream.ts index b0e40c1..9bb8abd 100644 --- a/src/lib/canvas-service/model-stream.ts +++ b/src/lib/canvas-service/model-stream.ts @@ -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=>{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. */ +/** 有界 UTF-8 NDJSON 分帧。按字节找换行，fatal 解码拒绝损坏编码，最后一条也要求换行结束。 */ async function* lines(source:AsyncIterable,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;indexmaxLine)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,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;complete:boolean;byteLength:number;coverage:string[];value:CanonicalReadUpdate}|undefined; let canvases:CanvasRecord[]=[],versions:ReadonlyMap=new Map(),tombstones:ReadonlyMap=new Map(); let elementOwners:ElementOwners|undefined;let canvasElementCounts:CanvasElementCounts|undefined;const ownerCounts=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'); + // 第一条必须是 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,canvas 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'); + // 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,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; } diff --git a/src/lib/canvas-service/runs.ts b/src/lib/canvas-service/runs.ts index e7c3fd6..3119e55 100644 --- a/src/lib/canvas-service/runs.ts +++ b/src/lib/canvas-service/runs.ts @@ -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 => { if (!value || Array.isArray(value) || typeof value !== 'object') throw fault('invalid_run_response'); return value as Record; }; +/** 协议要求数组就必须实际给数组，缺字段不能默默当作空列表。 */ 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(); 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()),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 { 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={}; 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;