diff --git a/src/lib/canvas-runtime/runs.ts b/src/lib/canvas-runtime/runs.ts index 9f07b31..0bab615 100644 --- a/src/lib/canvas-runtime/runs.ts +++ b/src/lib/canvas-runtime/runs.ts @@ -3,6 +3,7 @@ * 任务受理和生成完成是两件事:RunPort 解释服务端回执及事件,RunRecord 保存任务事实, * tracker 只负责何时读取下一页;它不直接创建元素,也不自行把输出写进文档。 * nextPollAt/cursor 等随根状态持久化;堆、去重 token、定时器和 AbortController 都可重建。 + * tracker 不提交生成请求:受理与控制操作由 ledger/scheduler 处理,poll 只读取已受理任务。 */ import type { Id, Json, OperationIntent, Target, RunInput } from '../canvas-core/types.ts'; import type { AppliedModelProof, CanonicalProposal, EngineSnapshot } from './contracts.ts'; @@ -12,7 +13,8 @@ import { fault, validJson } from '../canvas-core/values.ts'; import { createMinHeap } from './structures.ts'; /** 校验生成意图的基本形状和大小:提示词按 Unicode 码点计数,参数按 JSON 的 UTF-8 字节计数。 - * 此处不判断模型是否可用或目标是否仍存在,这些属于目录、编辑器和服务适配器的校验。 */ + * 只检查形状,不捕获/冻结输入,也不查询模型目录、目标存活、board 归属或服务权限。 + * JSON 校验/计长可能读取 getter,不能将此检查当成外来可变对象已经隔离的证明。 */ 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'); } @@ -29,7 +31,8 @@ export type RunEvent = { readonly seq: number; readonly run_id: Id; readonly typ 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 要求显式恢复,而不是不断自动重试。 */ + * attempt/nextPollAt 记录重试安排,blocked 要求显式恢复,而不是不断自动重试。 + * 此类型不执行运行时校验;pendingPage 的保存及模型/游标同根协调由 engine.accept 实现。 */ 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; @@ -38,7 +41,8 @@ export type RunRecord = { }; /** 外部生成服务的适配协议。compile 产生请求描述,submit/query 处理请求是否生效, * accept/acceptAction 验证并转换受理/控制回执;poll 追踪执行,canonical 提取事件中的权威文档。 - * verifyApplied 可提供已应用结果的验证证据,不能把任意响应直接当成本地文档覆盖。 */ + * verifyApplied 可提供已应用结果的验证证据,不能把任意响应直接当成本地文档覆盖。 + * 回执/事件的 schema、版本及游标校验是端口与协调器的实现合同;tracker 不重复验证 RunPage。 */ 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; @@ -51,7 +55,8 @@ export type RunPort = { canonical(event: RunEvent, snapshot: EngineSnapshot): CanonicalProposal; }; /** 按 ID + incarnation 查目标:既检查入口目标,也检查计划 steps 中的传播目标。 - * 同 ID 重建后的元素不是旧任务的目标;无法识别的计划不推测它包含哪些节点。 */ + * 同 ID 重建后的元素不是旧任务的目标;只认识 plan.steps 的 node_id/target_incarnation。 + * 不递归搜索任意 JSON、不检查文档中目标是否存活,也不计算所有间接资源依赖。 */ export function runContainsTarget(run: RunRecord, target: Target) { if(run.target.id===target.id && run.target.incarnation===target.incarnation)return true; const plan=run.plan; @@ -59,11 +64,14 @@ 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 判断。 */ +/** 在 RunStatus 合同内,除 queued/running 外均视为终态;是否还有事件要结合 hasMore。 + * 这里只做两个字符串比较,不是验证任意服务状态值的解析器。 */ export const runTerminal = (status: RunStatus) => status !== 'queued' && status !== 'running'; /** 创建串行轮询器:最多一个在途读取,以截止时间小顶堆选下个任务,同刻按入堆顺序。 - * 所有状态修改通过 update/accept 注入,持久化通过 persist 注入,因此调度器不另存业务事实。 */ + * 所有状态修改通过同步 update/accept 注入,持久化通过 persist 注入,因此调度器不另存业务事实。 + * getRuns 提供稳定 RunRecord 快照,clock/persist/accept 的身份与可靠性由宿主保证;构造不自动扫描任务。 + * notify 负责首次及后续排队,堆只索引已通知项;连续重复通知会留惰性失效项,没有独立容量上限。 */ export function createRunTracker(options: { clock: Clock; getRuns(): ReadonlyMap; persist(): Promise; poll(run: RunRecord, signal: AbortSignal): Promise; @@ -72,8 +80,10 @@ 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;旧堆项惰性丢弃,不必在堆中查找删除。 + // token 是每个 ID 最近入堆代次,sequence 仅打破同刻平局;两者不属于持久任务版本。 + // 旧堆项只在到堆顶时惰性丢弃,tokens 不随每次出堆删除,不能把它当作活跃任务清单。 const tokens = new Map(); + // 每个 hold 独占一个 token,释放幂等;影响新读取调度,不改变业务 status 或中止在途采纳。 const holds = new Set(); let sequence = 0, stopped = false, timer: unknown, active: { controller: AbortController; promise: Promise } | undefined; /** 错误展示失败也不能反过来中断游标保存或改变轮询调度。 */ @@ -87,7 +97,8 @@ export function createRunTracker(options: { if (stopped || holds.size || active || !heap.peek()) return; timer = options.clock.setTimer(() => { timer = undefined; pump(); }, Math.max(0,heap.peek()!.due-options.clock.now())); } - /** 从当前事实重新取截止时间;通知只建立索引,不改任务状态、不发网络请求。 */ + /** 从当前事实重新取截止时间;通知只建立索引,不改任务状态、不立即调用 poll。 + * 无 nextPollAt 则按当前时间排队;不合格/缺失项不入堆,旧项由 pump 的实时资格检查挡住。 */ function notify(ids: readonly Id[]) { if (stopped) return; for (const id of ids) { @@ -98,7 +109,9 @@ export function createRunTracker(options: { arm(); } /** 取一个到期任务并执行“先保存身份 → 读/复用页 → 协调 → 保存新游标”。 - * 在途 promise 统一在 finally 释放,以保证失败和成功路径都能继续调度其他任务。 */ + * 在途 promise 统一在 finally 释放,以保证失败和成功路径都能继续调度其他任务。 + * run 在出堆时捕获,await 后不自动刷新;成功响应的 requestId/cursor/controlEpoch 栅栏由 accept 检查。 + * pendingPage 复用也先过 persist,避免新协调越过可靠受理事实;此处不主动将新 poll 页保存为 pendingPage。 */ function pump() { if (stopped || holds.size || active) return; let item = heap.pop(); @@ -113,9 +126,11 @@ export function createRunTracker(options: { await options.persist();if(stopped)return; // 协调未完成的页优先重用;不能跳到下一页而遗漏该页的文档结果。 const page=run.pendingPage ?? await options.poll(run,controller.signal);if(stopped)return; + // accept 必须同步完成发布/保留页/冲突处理,tracker 不 await 异步回调,也不直接修改文档。 options.accept(run,page); // accept 可能将任务标为 blocked 或留下 pendingPage;按照接受后的事实安排下一次读取。 const latest=options.getRuns().get(run.id)!; + // accept 未抛错后清重试次数;pendingPage 优先延迟1000ms,否则 hasMore 用1ms,其余1000ms。 options.update(run.id,current=>({...current,attempt:0,nextPollAt:options.clock.now()+(current.pendingPage ? 1000 : current.hasMore ? 1 : 1000)})); // 保存失败转入下方错误处理;若连错误状态也无法保存,则阻塞该任务等待显式重试。 await options.persist(); @@ -125,8 +140,10 @@ export function createRunTracker(options: { const current=options.getRuns().get(run.id); // 等待期间若取消/应用等控制代次已改变,旧读取的失败不能污染新状态,重新通知即可。 if(current && (current.controlEpoch??0)!==(run.controlEpoch??0)){notify([run.id]);return;} + // 端口须抛 Error 风格对象;此处读取 code/message,不是能归一化任意 thrown 值的通用捕获器。 const code=(error as {code?:string}).code ?? (error as Error).message ?? 'run_poll_failed'; - // 身份、权限和协议解释失败停住;其它错误按指数退避,单次延迟最多 30 秒。 + // 按精确错误码集合阻塞;其它错误按旧 attempt 计算500ms起的指数退避,单次最多30秒。 + // status/cursor 不在错误分支回滚,accept 已发布的冲突或页证据继续由根状态持有。 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))})); // 连错误状态也无法持久化时,在内存中阻塞;不让不可恢复的游标继续向前。 @@ -144,11 +161,15 @@ export function createRunTracker(options: { if(timer!==undefined)options.clock.clearTimer(timer);timer=undefined; return()=>{if(holds.delete(token)&&!holds.size&&!stopped)arm();}; }, - /** 先确认当前状态能保存,再清除阻塞并立即排队;最终是否合格仍由 eligible 决定。 */ + /** 先保存现有根,再清除所有任务(不只 blocked)的阻塞/error 并改 nextPollAt 为现在;不清 attempt。 + * 清除后的状态本方法不再 persist,由下一次 pump 的前置屏障保存;终态无尾页仍不轮询。 + * retry 本身不检查 stopped,关闭后宿主不应再调用;notify 的停止门禁只阻止重新发起读取。 */ 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,撤销计时、请求中止在途网络并丢弃堆索引;任务事实留给恢复。 */ + /** 永久关闭本 tracker,撤销计时、请求中止在途网络并丢弃堆索引;不调用远端取消任务。 + * 同步返回,不等待 active 结束;迟到页在 accept 前被 stopped 挡住,必要时另用 settle 等在途退出。 + * 已进入 accept/persist 的工作不会被回滚,任务事实留给宿主恢复,tokens/holds 随实例回收。 */ stop() { stopped=true;if(timer!==undefined)options.clock.clearTimer(timer);timer=undefined;active?.controller.abort();heap.clear(); }, }); }