docs(canvas): explain run polling and explicit conflict recovery
This commit is contained in:
@@ -1,4 +1,10 @@
|
||||
/** Opt-in browser adapters. Importing the portable runtime does not load these. */
|
||||
/**
|
||||
* 浏览器专用适配入口。调用方显式选择 IndexedDB 存储、摘要 Worker 或图片解码实现,
|
||||
* 可移植入口不直接重导出这些模块;此文件只导出工厂,不在这里打开数据库或启动 Worker。
|
||||
*/
|
||||
// 项目持久化及共享资源存储的浏览器实现,由工厂调用方提供身份和生命周期配置。
|
||||
export { createBrowserStorage } from './storage.ts';
|
||||
// 原文件摘要计算适配器,管理浏览器 Worker 与中止/关闭。
|
||||
export { createBrowserDigests } from './browser-digests.ts';
|
||||
// 图片尺寸解码适配器,与资源上传、显示 URL 生命周期分别管理。
|
||||
export { createBrowserImageDecoder } from './browser-images.ts';
|
||||
|
||||
@@ -1,3 +1,9 @@
|
||||
/**
|
||||
* 用户明确选择后的冲突恢复规划,不是后台静默重试。
|
||||
* 以校验过的服务端文档为起点,逐项保留或丢弃本地未决意图,重建文档、历史及 ledger,
|
||||
* 同时保存旧文档备份和决策审计。此模块只返回方案,真正发布和持久化由 engine 协调。
|
||||
* 保留本地意图时创建新操作身份;原操作/请求证据不会被改写成一次看似未发生的请求。
|
||||
*/
|
||||
import {validateLoadingCompletion} from './loading.ts';
|
||||
import { withLocalLifecycles } from './elision.ts';
|
||||
import type { Change, Document, FieldChange, HistoryState, Id, Json, OperationIntent, Registry } from '../canvas-core/types.ts';
|
||||
@@ -14,36 +20,52 @@ import { replanTraversals } from './traversal-recovery.ts';
|
||||
import { operationCanvases, validateCanonicalRead } from './reconcile.ts';
|
||||
import { referencedResources } from './resources.ts';
|
||||
|
||||
/** 每个待处理操作都必须有用户选择;不隐式替未选择的依赖作决定。 */
|
||||
export type RecoveryChoice = 'keep-local' | 'discard-local';
|
||||
/** 旧操作的结论:被新操作替代、被丢弃,或目标已满足而不需要新操作。 */
|
||||
export type RecoveryDecision = { readonly operationId: Id; readonly choice: RecoveryChoice; readonly replacementId?: Id; readonly resolution: 'superseded' | 'discarded' | 'satisfied' };
|
||||
/** 恢复审计记录:绑定规划时的根 revision,保留备份及可用的历史重放证据。 */
|
||||
export type RecoveryAudit = { readonly id: Id; readonly formatVersion: 1; readonly baseRootRevision: number; readonly decisions: readonly RecoveryDecision[]; readonly backup: Json; readonly historyTransition?:HistoryTransition; readonly historyReplay?:HistoryReplay };
|
||||
/** 待一次性发布的候选根及引用资源;baseRootRevision 供提交方拒绝过时方案。 */
|
||||
export type ConflictPlan = { readonly id: Id; readonly baseRootRevision: number; readonly document: Document; readonly history: HistoryState; readonly operations: EngineSnapshot['operations']; readonly ledger: EngineSnapshot['runtime']['ledger']; readonly audit: RecoveryAudit; readonly resourceIds: readonly Id[]; readonly operationIds: readonly Id[] };
|
||||
/** 字段身份包括元素代次,避免把删除后重建的同 ID 元素算成同一字段历史。 */
|
||||
const fieldKey = (change: FieldChange) => JSON.stringify([change.target.id,change.target.incarnation,change.field]);
|
||||
/** 读取恢复起点的字段值,并区分 data 字段缺失与显式 null。
|
||||
* allowMissing 仅供历史分析记录目标缺失,不将旧代次目标重定向到新元素。 */
|
||||
function value(document: Document, change: FieldChange, allowMissing=false): {before:Json;beforePresent?:boolean;missing?:boolean} {
|
||||
const element=document.elements.get(change.target.id);
|
||||
if(!element||element.incarnation!==change.target.incarnation){if(allowMissing)return{before:null,beforePresent:false,missing:true};throw fault('stale_incarnation',[change.target.id]);}
|
||||
const key=change.field.startsWith('data.')?change.field.slice(5):undefined;
|
||||
return {before:(key===undefined?(element as unknown as Record<string,Json>)[change.field]:element.data[key]??null),beforePresent:key===undefined?undefined:Object.hasOwn(element.data,key)};
|
||||
}
|
||||
/** 把文档 Map 转成备份用的 JSON 形状;序列化的是数据,不是运行时方法或索引。 */
|
||||
const documentValue = (document: Document) => ({projectId:document.projectId,canvases:[...document.canvases.values()],elements:[...document.elements.values()],tombstones:[...document.tombstones]});
|
||||
|
||||
/** Pure proposal. Explicit choices cover the complete pending dependency closure; no key/body is reused. */
|
||||
/** 纯规划入口:先校验权威读取与回执已全部落定,再按操作顺序投影用户决策。
|
||||
* 不复用旧请求键/请求体;未知结果尚未确认时拒绝规划,防止重复执行已经生效的修改。 */
|
||||
export function planConflictRecovery(snapshot: EngineSnapshot, canonical: CanonicalProposal, choices: ReadonlyMap<Id,RecoveryChoice>, registry: Registry, createId:()=>Id,historyLimit=100): ConflictPlan {
|
||||
if(snapshot.runtime.loading)validateLoadingCompletion(snapshot.runtime.acknowledged,snapshot.runtime.versions,snapshot.runtime.loading,canonical.document,canonical.versions,registry,true);else validateCanonicalRead(snapshot.runtime.acknowledged,snapshot.runtime.versions,canonical.document,canonical.versions,registry);
|
||||
if([...snapshot.runtime.ledger.requests.values()].some(request=>request.phase!=='resolved'))throw fault('recovery_waiting_for_receipts');
|
||||
// 上传资源不在这次文档取舍中;其它所有未决操作都纳入选择集合,并检查依赖的时间顺序。
|
||||
const records=[...snapshot.runtime.ledger.operations.values()].filter(record=>record.phase!=='resolved'&&record.intent.operationType!=='asset.upload').sort((a,b)=>a.intent.sequence-b.intent.sequence);
|
||||
if(!records.length)throw fault('recovery_nothing_to_resolve');
|
||||
for(const record of records)for(const dependency of record.intent.dependencies)if((snapshot.runtime.ledger.operations.get(dependency)?.intent.sequence??Infinity)>=record.intent.sequence)throw fault('recovery_dependency_order',[record.intent.id]);
|
||||
if(choices.size!==records.length||records.some(record=>!['keep-local','discard-local'].includes(choices.get(record.intent.id)??'')))throw fault('recovery_choices_incomplete');
|
||||
// 新审计和替代操作都在同一命名空间避碰,包含旧历史、请求和恢复记录中的身份。
|
||||
const used=new Set([...snapshot.operations.keys(),...snapshot.runtime.ledger.operations.keys(),...snapshot.runtime.ledger.requests.keys(),...snapshot.history.entries.keys(),...(snapshot.runtime.recoveries?.keys()??[])]);
|
||||
/** 每次分配都登记使用过的 ID,拒绝生成器重复值;不靠覆盖 Map 来解决碰撞。 */
|
||||
function unique(){const id=createId();if(!id||used.has(id))throw fault('duplicate_operation_id');used.add(id);return id;}
|
||||
const id=unique(),ledger=createLedger(snapshot.runtime.ledger),operations=new Map(snapshot.operations),decisions:RecoveryDecision[]=[],replacements=new Map<Id,OperationIntent|undefined>();
|
||||
// 边界记录历史单元恢复前后的文档;starts 保存首次触及字段时的基线,供混合 ACK 历史核对。
|
||||
const boundaries=new Map<Id,{before:Document;after:Document}>(),starts=new Map<string,ReturnType<typeof value>>(),historyOwner=new Map<Id,Id>(),contexts=intentContexts(snapshot.document,records.map(record=>record.intent));
|
||||
for(const entry of snapshot.history.entries.values())for(const operationId of entry.operationIds){if(historyOwner.has(operationId))throw fault('recovery_history_ambiguous');historyOwner.set(operationId,entry.id);}
|
||||
// 有历史转移证据或之前已采用重放的操作时,优先用日志恢复保持 undo/redo 的语义。
|
||||
const adopted=new Set([...snapshot.runtime.recoveries.values()].filter(audit=>audit.historyReplay!==undefined).flatMap(audit=>audit.decisions.flatMap(decision=>decision.replacementId?[decision.replacementId]:[])));
|
||||
const content=records.filter(record=>record.intent.changes.length),hasEvidence=content.length>0&&content.every(record=>record.intent.historyTransition)||records.some(record=>['undo','redo'].includes(record.intent.origin)&&record.intent.historyTransition)||content.some(record=>adopted.has(record.intent.id));
|
||||
const journal=hasEvidence?createHistoryJournalRecovery(snapshot,records.map(record=>record.intent),registry,historyLimit):undefined;
|
||||
// 权威文档叠加本地生命周期事实,避免尚未发出的创建/删除身份被错误复活。
|
||||
let document=withLocalLifecycles(canonical.document,snapshot.runtime.ledger),sequence=0;for(const record of snapshot.runtime.ledger.operations.values())sequence=Math.max(sequence,record.intent.sequence);
|
||||
// 串行投影使后一个意图看到前一项保留/丢弃的结果,而不是分别套在同一旧快照上。
|
||||
for(const record of records){
|
||||
const intent=record.intent,choice=choices.get(intent.id)!;
|
||||
if(intent.operationSchemaVersion!==1)throw fault('operation_interpreter_unavailable',[intent.id]);
|
||||
@@ -54,19 +76,24 @@ export function planConflictRecovery(snapshot: EngineSnapshot, canonical: Canoni
|
||||
const owner=historyOwner.get(intent.id),key=owner?JSON.stringify([owner,fieldKey(original)]):undefined;
|
||||
if(key&&!starts.has(key))starts.set(key,value(document,original,true));
|
||||
}
|
||||
// 用户新建 canvas 的身份若已被服务端占用,不能悄悄视为同一对象。
|
||||
if(choice==='keep-local'&&intent.origin==='user')for(const change of intent.changes)if(change.kind==='canvas'&&!change.before&&canonical.versions.has(change.id))throw fault('recovery_identity_collision',[change.id]);
|
||||
// keep-local 在新基线上重新计算改动;discard-local 不施加该意图,日志仍记录该决定。
|
||||
const rebased=journal?journal.project(document,intent,contexts.get(intent.id)!,choice==='keep-local'):choice==='keep-local'?rebaseIntent(document,intent,contexts.get(intent.id)!,registry):{document,changes:[]};
|
||||
const changes=rebased.changes;document=rebased.document;
|
||||
if(entryId)boundaries.get(entryId)!.after=document;
|
||||
let replacement:OperationIntent|undefined;
|
||||
if(choice==='keep-local'){
|
||||
// 生成启动、取消等动作需重新发起命令;冲突审阅不能顺手重新执行可能付费的任务。
|
||||
if(intent.operationType!=='document.change'&&intent.operationType!=='run.apply')throw fault('recovery_action_requires_new_command',[intent.id]);
|
||||
if(changes.length){
|
||||
// 依赖替换为对应的新操作,丢弃或已满足的依赖不再排队;再串联此前所有替代操作。
|
||||
const dependencyIds=intent.dependencies.flatMap(dependency=>replacements.has(dependency)?replacements.get(dependency)?[replacements.get(dependency)!.id]:[]:[dependency]);
|
||||
// New dependencies reference new logical facts; a discarded parent never silently starts a paid run.
|
||||
const dependencySet=new Set(dependencyIds);
|
||||
for(const previous of replacements.values())if(previous)dependencySet.add(previous.id);
|
||||
const replacementId=unique();
|
||||
// 旧历史转移绑定旧状态,不能直接复制给新操作;保留其余逻辑属性并重算画布范围。
|
||||
const {historyTransition:_oldHistory,...originalFact}=intent;
|
||||
const basic={...originalFact,id:replacementId,sequence:++sequence,changes:freeze(changes),dependencies:[...dependencySet],canvasIds:operationCanvases({...intent,canvasIds:[],changes},before,document)};
|
||||
replacement=intent.operationType==='run.apply'?freeze({...basic,operationType:'run.apply',action:{...intent.action,targetCanvasId:document.elements.get(intent.action.target.id)!.ownerCanvasId}}) as OperationIntent:freeze(basic);
|
||||
@@ -77,7 +104,9 @@ export function planConflictRecovery(snapshot: EngineSnapshot, canonical: Canoni
|
||||
replacements.set(intent.id,replacement);
|
||||
decisions.push(freeze({operationId:intent.id,choice,resolution:choice==='discard-local'?'discarded':replacement?'superseded':'satisfied',...(replacement?{replacementId:replacement.id}:{})}));
|
||||
}
|
||||
// 根据现有证据选日志重放、撤销/重做重规划或兼容历史重写,拒绝无法证明安全的路径。
|
||||
const pending=records.map(record=>record.intent),history=journal?journal.finish():pending.some(intent=>['undo','redo'].includes(intent.origin)&&intent.changes.length)?replanTraversals(snapshot,pending,replacements,new Set(decisions.filter(decision=>decision.choice==='keep-local').map(decision=>decision.operationId)),boundaries,registry):rewriteHistory(snapshot,pending,replacements,starts,boundaries,registry);
|
||||
// 旧逻辑事实转为带审计的终结状态,新替代操作单独加入账本并记录存在/不存在基线。
|
||||
ledger.retire(id,decisions);
|
||||
for(const replacement of replacements.values())if(replacement){
|
||||
const presence:Record<Id,Presence>={};for(const change of replacement.changes)if(change.kind==='field')presence[change.target.id]??='exists';else if(change.kind==='element')presence[change.id]??=change.before?'exists':'absent';
|
||||
@@ -89,22 +118,27 @@ export function planConflictRecovery(snapshot: EngineSnapshot, canonical: Canoni
|
||||
for(const record of records)for(const resource of referencedResources(record.intent.changes,registry))resourceIds.add(resource);
|
||||
for(const replacement of replacements.values())if(replacement)for(const resource of referencedResources(replacement.changes,registry))resourceIds.add(resource);
|
||||
if(historyReplay)for(const transition of [historyReplay.anchor,...historyReplay.events.map(event=>event.transition)])for(const resource of referencedResources(historyTransitionChanges(transition),registry))resourceIds.add(resource);
|
||||
// 同时备份旧 working、ACK 基线和历史;资源 pin 保护恢复证据,防止审阅后被清理。
|
||||
const backup=JSON.parse(JSON.stringify({document:documentValue(snapshot.document),acknowledged:documentValue(snapshot.runtime.acknowledged),versions:[...snapshot.runtime.versions],resourceIds:[...resourceIds],history:{undo:snapshot.history.undo,redo:snapshot.history.redo,entries:[...snapshot.history.entries]}})) as Json;
|
||||
const historyTransition=describeHistoryTransition(snapshot.history,history,snapshot.rootRevision+1,undefined,!!historyReplay);
|
||||
const audit=freeze({id,formatVersion:1 as const,baseRootRevision:snapshot.rootRevision,decisions,backup,...(historyTransition?{historyTransition}:{}),...(historyReplay?{historyReplay}:{})});
|
||||
return Object.freeze({id,baseRootRevision:snapshot.rootRevision,document,history,operations:readMap(operations),ledger:ledger.getSnapshot(),audit,resourceIds:Object.freeze([...resourceIds]),operationIds:Object.freeze([...replacements.values()].flatMap(intent=>intent?[intent.id]:[]))});
|
||||
}
|
||||
|
||||
/** 无完整日志证据时的兼容历史重写。仅支持能够证明前缀/字段基线的撤销栈项,
|
||||
* 不推测交错 ACK、结构变更或不支持的历史遍历;无法验证时抛错让整个规划保持未发布。 */
|
||||
function rewriteHistory(snapshot:EngineSnapshot,pending:readonly OperationIntent[],replacements:ReadonlyMap<Id,OperationIntent|undefined>,starts:ReadonlyMap<string,ReturnType<typeof value>>,boundaries:ReadonlyMap<Id,{before:Document;after:Document}>,registry:Registry):HistoryState {
|
||||
const entries=new Map(snapshot.history.entries),affected=new Set(pending.map(intent=>intent.id));
|
||||
for(const entry of snapshot.history.entries.values()){
|
||||
const intents=pending.filter(intent=>entry.operationIds.includes(intent.id));if(!intents.length)continue;
|
||||
if(!snapshot.history.undo.includes(entry.id)||entry.operationIds.some(id=>snapshot.operations.get(id)?.origin!=='user'))throw fault('recovery_history_traversal_not_supported',[entry.id]);
|
||||
// 整个历史单元都未确认时,可直接由替代 changes 重建;没有剩余改动则移除该单元。
|
||||
if(entry.operationIds.every(id=>affected.has(id))){
|
||||
const changes=intents.flatMap(intent=>replacements.get(intent.id)?.changes??[]),operationIds=intents.flatMap(intent=>replacements.get(intent.id)?[replacements.get(intent.id)!.id]:[]);
|
||||
if(changes.length)entries.set(entry.id,freeze({...entry,changes,operationIds}));else entries.delete(entry.id);
|
||||
continue;
|
||||
}
|
||||
// 所有权和连线端点也属于结构,不能按普通字段独立拼接。
|
||||
const structural=(changes:readonly Change[])=>changes.some(change=>change.kind!=='field'||['ownerCanvasId','source','target'].includes(change.field));
|
||||
if(structural(entry.changes)||intents.some(intent=>structural(intent.changes))){
|
||||
// A grouped entry is chronological. Never splice a confirmed operation from after a pending one into its prefix.
|
||||
@@ -121,6 +155,7 @@ function rewriteHistory(snapshot:EngineSnapshot,pending:readonly OperationIntent
|
||||
if(changes.length)entries.set(entry.id,freeze({...entry,changes,operationIds}));else entries.delete(entry.id);
|
||||
continue;
|
||||
}
|
||||
// 纯字段混合历史按“目标代次 + 字段”分组;确认部分是否保留取决于恢复起点的值和存在性。
|
||||
const keys=new Map<string,FieldChange>();for(const intent of intents)for(const change of intent.changes as readonly FieldChange[])keys.set(fieldKey(change),change);
|
||||
const fields:FieldChange[]=[];
|
||||
for(const [key,example]of keys){
|
||||
@@ -129,6 +164,7 @@ function rewriteHistory(snapshot:EngineSnapshot,pending:readonly OperationIntent
|
||||
const oldPending=intents.flatMap(intent=>intent.changes as readonly FieldChange[]).filter(change=>fieldKey(change)===key);
|
||||
const prefix=original.slice(0,original.length-oldPending.length);
|
||||
if(original.length<oldPending.length||!equal(original.slice(prefix.length),oldPending))throw fault('recovery_history_mismatch',[entry.id]);
|
||||
// 若服务端基线已改变,原先确认前缀不再代表当前起点,不盲目把它纳入新的撤销内容。
|
||||
const start=starts.get(JSON.stringify([entry.id,key]))!,first=oldPending[0];
|
||||
if(!start.missing&&equal(start.before,first.before)&&(start.beforePresent??true)===(first.beforePresent??true))fields.push(...prefix);
|
||||
fields.push(...intents.flatMap(intent=>replacements.get(intent.id)?.changes as readonly FieldChange[]??[]).filter(change=>fieldKey(change)===key));
|
||||
@@ -137,5 +173,6 @@ function rewriteHistory(snapshot:EngineSnapshot,pending:readonly OperationIntent
|
||||
const operationIds=entry.operationIds.flatMap(operationId=>affected.has(operationId)?replacements.get(operationId)?[replacements.get(operationId)!.id]:[]:[operationId]);
|
||||
if(changes.length)entries.set(entry.id,freeze({...entry,changes,operationIds}));else entries.delete(entry.id);
|
||||
}
|
||||
// 只移除已消失的 undo 项;redo 的安全重规划不在此兼容分支中暗自执行。
|
||||
return Object.freeze({undo:Object.freeze(snapshot.history.undo.filter(id=>entries.has(id))),redo:snapshot.history.redo,entries:readMap(entries)});
|
||||
}
|
||||
|
||||
@@ -1,10 +1,15 @@
|
||||
/** Portable runtime entry. Browser adapters are an explicit separate entry. */
|
||||
/**
|
||||
* 可移植 runtime 的公共入口:集中公开宿主/工作区/项目 engine、协议类型及纯计算服务。
|
||||
* 这里只做模块重导出,不创建项目实例;浏览器存储、Worker 等具体适配器放在 browser.ts 显式选择。
|
||||
*/
|
||||
// 项目 → 工作区 → 宿主三级生命周期装配入口;各层的可变状态由对应实例持有。
|
||||
export { openProjectEngine } from './engine.ts';
|
||||
export type { EngineOptions, RecoverySession } from './engine.ts';
|
||||
export { createWorkspaceRuntime } from './workspace.ts';
|
||||
export type { WorkspaceOptions, WorkspaceRuntime, ProjectEngine } from './workspace.ts';
|
||||
export { createRuntimeHost } from './host.ts';
|
||||
export type { RuntimeHost } from './host.ts';
|
||||
// 状态根、存储和各任务端口的公共契约,类型导出不会实例化服务。
|
||||
export type * from './contracts.ts';
|
||||
export type * from './storage-port.ts';
|
||||
export type { LoadingAssetAdmission, AssetPort, ResourceRecord, ResourceHandle, StagedFile, StoredAsset, DisplayHandle } from './resources.ts';
|
||||
@@ -13,6 +18,7 @@ export type { FrozenRequest, LedgerSnapshot, OperationRecord, Presence } from '.
|
||||
export type { Clock, OperationStatus, CompiledRequest } from './scheduler.ts';
|
||||
export type { LoadingImportAdmission,ImportJob, ImportSource, ImportLine, ImportUnit, ImportFailure, NativeImportSource } from './imports.ts';
|
||||
export type { RecoveryChoice, ConflictPlan, RecoveryAudit } from './conflicts.ts';
|
||||
// 恢复编解码与共享读取/显示/计算池,可由宿主注入并按身份作用域管理。
|
||||
export { encodeRecovery, decodeRecovery } from './recovery.ts';
|
||||
export { createScopedLoader, loadQueryKey } from './loader.ts';
|
||||
export type { LoadScope, LoadChunk, LoadSnapshot, LoadPort, LoadHandle, ScopedLoader } from './loader.ts';
|
||||
@@ -23,6 +29,7 @@ export type { DigestPort } from './digest-pool.ts';
|
||||
export { createImageDecodePool } from './image-decoder.ts';
|
||||
export type { ImageDecoderPort, ImageDimensions, ImageDecodeState } from './image-decoder.ts';
|
||||
|
||||
// 渐进加载预览及加载状态,不代表另一份可以独立写入的业务文档。
|
||||
export {createModelPreview} from './model-preview.ts';
|
||||
export type {ModelPreview,LiveModelPreview} from './model-preview.ts';
|
||||
|
||||
@@ -30,6 +37,7 @@ export type {LoadingState} from './loading.ts';
|
||||
|
||||
export type {CanvasElementCounts,ElementOwners} from './scope-counts.ts';
|
||||
|
||||
// 跨项目资源仓库与导入源/调度配置;资源授权仍由具体端口与作用域约束。
|
||||
export {createAssetDepot} from './asset-depot.ts';
|
||||
export type {AssetDepot,AssetAuthority,SharedAsset,SharedAssetBinding,AssetDepotPort,DepotCut,DepotCollection} from './asset-depot.ts';
|
||||
export {createJsonArrayImportSource} from './import-json-array.ts';
|
||||
|
||||
@@ -1,3 +1,9 @@
|
||||
/**
|
||||
* 生成任务的协议边界与可恢复轮询调度。
|
||||
* 任务受理和生成完成是两件事:RunPort 解释服务端回执及事件,RunRecord 保存任务事实,
|
||||
* tracker 只负责何时读取下一页;它不直接创建元素,也不自行把输出写进文档。
|
||||
* nextPollAt/cursor 等随根状态持久化;堆、去重 token、定时器和 AbortController 都可重建。
|
||||
*/
|
||||
import type { Id, Json, OperationIntent, Target, RunInput } from '../canvas-core/types.ts';
|
||||
import type { AppliedModelProof, CanonicalProposal, EngineSnapshot } from './contracts.ts';
|
||||
import type { FrozenRequest } from './ledger.ts';
|
||||
@@ -5,21 +11,34 @@ import type { CompiledRequest, OperationStatus, Clock } from './scheduler.ts';
|
||||
import { fault, validJson } from '../canvas-core/values.ts';
|
||||
import { createMinHeap } from './structures.ts';
|
||||
|
||||
/** 校验生成意图的基本形状和大小:提示词按 Unicode 码点计数,参数按 JSON 的 UTF-8 字节计数。
|
||||
* 此处不判断模型是否可用或目标是否仍存在,这些属于目录、编辑器和服务适配器的校验。 */
|
||||
export function validateRunInput(input: RunInput) {
|
||||
if (!input || typeof input.boardId!=='string' || !input.boardId || typeof input.target?.id!=='string' || !input.target.id || !Number.isSafeInteger(input.target.incarnation) || input.target.incarnation<1 || typeof input.modelId!=='string' || !input.modelId || typeof input.prompt!=='string' || !input.prompt.trim() || [...input.prompt].length>16000 || typeof input.propagate!=='boolean' || !input.parameters || Array.isArray(input.parameters) || typeof input.parameters!=='object' || !validJson(input.parameters) || new TextEncoder().encode(JSON.stringify(input.parameters)).length>16384) throw fault('invalid_ai_request');
|
||||
}
|
||||
|
||||
/** 模型目录的单项能力声明;可用性和输入/输出模态供界面选择及后续校验。 */
|
||||
export type RunModel = { readonly id: string; readonly label?: string; readonly display_name?: string; readonly output_modalities: readonly string[]; readonly input_modalities: readonly string[]; readonly available: boolean; readonly unavailable_reason?: string };
|
||||
/** 一次目录读取的结果;目录整体不可用时仍可给出原因。 */
|
||||
export type RunCatalog = { readonly models: readonly RunModel[]; readonly unavailableReason?: string };
|
||||
/** 服务端执行状态。终态仍可能有未读事件,不能仅凭状态停止追尾。 */
|
||||
export type RunStatus = 'queued' | 'running' | 'succeeded' | 'failed' | 'conflict' | 'cancelled' | 'partially_succeeded';
|
||||
/** 同一任务的有序事件;扩展字段保持 JSON,具体含义由端口解释。 */
|
||||
export type RunEvent = { readonly seq: number; readonly run_id: Id; readonly type: string; readonly [key: string]: Json };
|
||||
/** 一页事件及当前结果摘要;next 是事件游标,hasMore 决定是否继续读尾页。 */
|
||||
export type RunPage = { readonly status: RunStatus; readonly events: readonly RunEvent[]; readonly next: number; readonly hasMore: boolean; readonly outputs: readonly Json[]; readonly steps: readonly Json[]; readonly error?: string };
|
||||
/** 可恢复的任务事实:关联受理操作/请求、目标代次、计划、事件游标与输出。
|
||||
* controlEpoch 区分控制操作前后的读取;pendingPage 保留尚待协调的页,避免贸然越过游标。
|
||||
* attempt/nextPollAt 记录重试安排,blocked 要求显式恢复,而不是不断自动重试。 */
|
||||
export type RunRecord = {
|
||||
readonly id: Id; readonly operationId: Id; readonly requestId: Id; readonly boardId: Id; readonly target: Target;
|
||||
readonly plan: Json; readonly status: RunStatus; readonly cursor: number; readonly hasMore: boolean;
|
||||
readonly outputs: readonly Json[]; readonly outputEvents: readonly RunEvent[]; readonly steps: readonly Json[]; readonly attempt: number;
|
||||
readonly controlEpoch?: number; readonly appliedOutputIds?: readonly Id[]; readonly nextPollAt?: number; readonly error?: string; readonly blocked?: boolean; readonly pendingPage?: RunPage;
|
||||
};
|
||||
/** 外部生成服务的适配协议。compile 产生请求描述,submit/query 处理请求是否生效,
|
||||
* accept/acceptAction 验证并转换受理/控制回执;poll 追踪执行,canonical 提取事件中的权威文档。
|
||||
* verifyApplied 可提供已应用结果的验证证据,不能把任意响应直接当成本地文档覆盖。 */
|
||||
export type RunPort = {
|
||||
verifyApplied?: (request:FrozenRequest,receipt:Json)=>{readonly model:AppliedModelProof;readonly admission?:RunRecord;readonly action?:{readonly runId:Id;readonly runStatus?:RunStatus;readonly outputId?:Id}};
|
||||
catalog(boardId: Id, signal: AbortSignal): Promise<RunCatalog>;
|
||||
@@ -31,6 +50,8 @@ export type RunPort = {
|
||||
poll(run: RunRecord, signal: AbortSignal): Promise<RunPage>;
|
||||
canonical(event: RunEvent, snapshot: EngineSnapshot): CanonicalProposal;
|
||||
};
|
||||
/** 按 ID + incarnation 查目标:既检查入口目标,也检查计划 steps 中的传播目标。
|
||||
* 同 ID 重建后的元素不是旧任务的目标;无法识别的计划不推测它包含哪些节点。 */
|
||||
export function runContainsTarget(run: RunRecord, target: Target) {
|
||||
if(run.target.id===target.id && run.target.incarnation===target.incarnation)return true;
|
||||
const plan=run.plan;
|
||||
@@ -38,9 +59,11 @@ export function runContainsTarget(run: RunRecord, target: Target) {
|
||||
return plan.steps.some(step=>!!step&&typeof step==='object'&&!Array.isArray(step)&&step.node_id===target.id&&step.target_incarnation===target.incarnation);
|
||||
}
|
||||
|
||||
/** 除排队/执行中外均视为终态;是否还有事件要结合 hasMore 判断。 */
|
||||
export const runTerminal = (status: RunStatus) => status !== 'queued' && status !== 'running';
|
||||
|
||||
/** Recoverable poll deadlines are facts; heap/tokens/active controllers are disposable indexes. */
|
||||
/** 创建串行轮询器:最多一个在途读取,以截止时间小顶堆选下个任务,同刻按入堆顺序。
|
||||
* 所有状态修改通过 update/accept 注入,持久化通过 persist 注入,因此调度器不另存业务事实。 */
|
||||
export function createRunTracker(options: {
|
||||
clock: Clock; getRuns(): ReadonlyMap<Id, RunRecord>; persist(): Promise<void>;
|
||||
poll(run: RunRecord, signal: AbortSignal): Promise<RunPage>;
|
||||
@@ -49,17 +72,22 @@ export function createRunTracker(options: {
|
||||
onError(error: unknown): void;
|
||||
}) {
|
||||
const heap = createMinHeap<{ id: Id; due: number; token: number; sequence: number }>((a,b) => a.due-b.due || a.sequence-b.sequence);
|
||||
// 同一任务重复通知只更新 token;旧堆项惰性丢弃,不必在堆中查找删除。
|
||||
const tokens = new Map<Id,number>();
|
||||
const holds = new Set<symbol>();
|
||||
let sequence = 0, stopped = false, timer: unknown, active: { controller: AbortController; promise: Promise<void> } | undefined;
|
||||
/** 错误展示失败也不能反过来中断游标保存或改变轮询调度。 */
|
||||
const report = (error: unknown) => { try { options.onError(error); } catch { /* Reporting cannot alter cursor durability. */ } };
|
||||
/** 已终结但仍有尾页的任务也要读取;blocked 具有更高优先级。 */
|
||||
const eligible = (run: RunRecord) => !run.blocked && (run.hasMore || !runTerminal(run.status));
|
||||
/** 清理失效堆顶,维护唯一计时器;暂停、关闭或已有读取时不启动另一读取。 */
|
||||
function arm() {
|
||||
if (timer !== undefined) options.clock.clearTimer(timer); timer = undefined;
|
||||
while (heap.peek() && tokens.get(heap.peek()!.id) !== heap.peek()!.token) heap.pop();
|
||||
if (stopped || holds.size || active || !heap.peek()) return;
|
||||
timer = options.clock.setTimer(() => { timer = undefined; pump(); }, Math.max(0,heap.peek()!.due-options.clock.now()));
|
||||
}
|
||||
/** 从当前事实重新取截止时间;通知只建立索引,不改任务状态、不发网络请求。 */
|
||||
function notify(ids: readonly Id[]) {
|
||||
if (stopped) return;
|
||||
for (const id of ids) {
|
||||
@@ -69,6 +97,8 @@ export function createRunTracker(options: {
|
||||
}
|
||||
arm();
|
||||
}
|
||||
/** 取一个到期任务并执行“先保存身份 → 读/复用页 → 协调 → 保存新游标”。
|
||||
* 在途 promise 统一在 finally 释放,以保证失败和成功路径都能继续调度其他任务。 */
|
||||
function pump() {
|
||||
if (stopped || holds.size || active) return;
|
||||
let item = heap.pop();
|
||||
@@ -81,20 +111,25 @@ export function createRunTracker(options: {
|
||||
try {
|
||||
// Admission identity and cursor must be durable before a read can trigger coordination.
|
||||
await options.persist();if(stopped)return;
|
||||
// 协调未完成的页优先重用;不能跳到下一页而遗漏该页的文档结果。
|
||||
const page=run.pendingPage ?? await options.poll(run,controller.signal);if(stopped)return;
|
||||
options.accept(run,page);
|
||||
// accept 可能将任务标为 blocked 或留下 pendingPage;按照接受后的事实安排下一次读取。
|
||||
const latest=options.getRuns().get(run.id)!;
|
||||
options.update(run.id,current=>({...current,attempt:0,nextPollAt:options.clock.now()+(current.pendingPage ? 1000 : current.hasMore ? 1 : 1000)}));
|
||||
// Storage failure blocks this tracker until explicit retry, retaining in-memory material.
|
||||
// 保存失败转入下方错误处理;若连错误状态也无法保存,则阻塞该任务等待显式重试。
|
||||
await options.persist();
|
||||
if(!stopped && !latest.blocked)notify([run.id]);
|
||||
} catch(error) {
|
||||
if(stopped)return;
|
||||
const current=options.getRuns().get(run.id);
|
||||
// 等待期间若取消/应用等控制代次已改变,旧读取的失败不能污染新状态,重新通知即可。
|
||||
if(current && (current.controlEpoch??0)!==(run.controlEpoch??0)){notify([run.id]);return;}
|
||||
const code=(error as {code?:string}).code ?? (error as Error).message ?? 'run_poll_failed';
|
||||
// 身份、权限和协议解释失败停住;其它错误按指数退避,单次延迟最多 30 秒。
|
||||
const blocked=['forbidden','unauthorized','identity_changed','invalid_run_response','run_identity_mismatch','run_receipt_expired','run_interpreter_unavailable'].includes(code);
|
||||
options.update(run.id,current=>({...current,error:code,blocked,attempt:current.attempt+1,nextPollAt:options.clock.now()+Math.min(30000,500*2**Math.min(current.attempt,6))}));
|
||||
// 连错误状态也无法持久化时,在内存中阻塞;不让不可恢复的游标继续向前。
|
||||
try { await options.persist();if(!blocked&&!stopped)notify([run.id]); } catch(storageError) { options.update(run.id,current=>({...current,blocked:true,error:'run_storage_failed'}));report(storageError); }
|
||||
report(error);
|
||||
}
|
||||
@@ -102,15 +137,18 @@ export function createRunTracker(options: {
|
||||
active={controller,promise};
|
||||
}
|
||||
return Object.freeze({ notify,
|
||||
/** Holds compose; completing a poll still commits its cursor before settle() returns. */
|
||||
/** 可嵌套暂停新轮询,不中止已经开始的读取;释放最后一个 hold 才恢复计时。 */
|
||||
hold() {
|
||||
if(stopped)throw fault('run_tracker_closed');
|
||||
const token=Symbol();holds.add(token);
|
||||
if(timer!==undefined)options.clock.clearTimer(timer);timer=undefined;
|
||||
return()=>{if(holds.delete(token)&&!holds.size&&!stopped)arm();};
|
||||
},
|
||||
/** 先确认当前状态能保存,再清除阻塞并立即排队;最终是否合格仍由 eligible 决定。 */
|
||||
async retry() { await options.persist();for(const id of options.getRuns().keys())options.update(id,run=>({...run,blocked:false,error:undefined,nextPollAt:options.clock.now()}));notify([...options.getRuns().keys()]); },
|
||||
/** 等当前在途处理及其保存结束;不保证所有未来轮询或生成任务已结束。 */
|
||||
async settle() { while(active)await active.promise; },
|
||||
/** 永久关闭本 tracker,撤销计时、请求中止在途网络并丢弃堆索引;任务事实留给恢复。 */
|
||||
stop() { stopped=true;if(timer!==undefined)options.clock.clearTimer(timer);timer=undefined;active?.controller.abort();heap.clear(); },
|
||||
});
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user