docs(runtime): explain generation polling state and lifecycle
This commit is contained in:
@@ -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<RunCatalog>;
|
||||
@@ -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<Id, RunRecord>; persist(): Promise<void>;
|
||||
poll(run: RunRecord, signal: AbortSignal): Promise<RunPage>;
|
||||
@@ -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<Id,number>();
|
||||
// 每个 hold 独占一个 token,释放幂等;影响新读取调度,不改变业务 status 或中止在途采纳。
|
||||
const holds = new Set<symbol>();
|
||||
let sequence = 0, stopped = false, timer: unknown, active: { controller: AbortController; promise: Promise<void> } | 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(); },
|
||||
});
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user