一次 24.3 秒的摘要不能挂在请求上
回声的内容发布链里，长文档摘要的示例参数是读 48,000 token、生成 1,200 token；在同一示例样本上从开始执行到结束用了 24.3 秒，而全书共享的同步编排延迟基线是 1,269ms。批量审核更极端：示例参数是 500 条内容逐条分类，预计需要几分钟。
这两类工作只能异步交付。同步请求负责创建任务并立即返回任务 ID，执行、重试、通知和结果交付走另一条路径。异步化不是把函数改成 setTimeout，而是引入一套新的状态与协议：任务现在处于什么状态、什么时候算完成、通知发几次、回调重复了怎么办、客户端断线后从哪里续。
本节仍然落在回声链路上：长文档摘要与批量审核都依赖 content.summarize 与 moderation.classify，只是把调用从请求内移到任务队列。所有数字来自 ai-lab 的示例样本运行结果。
五个状态与一次过期判定
| 状态 | 含义 | 可迁移到 |
|---|---|---|
| queued | 已创建，等待执行 | running、expired |
| running | 执行中，可上报进度 | succeeded、failed、queued、expired |
| succeeded | 成功，结果可交付 | 无 |
| failed | 重试耗尽或不可重试 | 无 |
| expired | 超过 TTL，不再接受结果 | 无 |
过期判定放在所有回调之前。任务创建时写入 expiresAtMs，任何事件到达时先看时间：超过 TTL 一律转 expired，不因为事件是成功就网开一面。TTL 是「结果还能交付」的截止时间，不是执行超时；执行可以晚一点结束，但结果过了交付窗口就没有意义。
可重试失败回到 queued，attempts 加一；重试次数用尽或错误不可重试时进入 failed。成功与失败都是一次终态迁移，之后的事件只能作为重复或冲突处理。
通知：终态只发一次
任务进入终态时写一条通知到 outbox，键是 taskId:state:version。同一条终态重复到达不会产生第二条通知，即使回调本身被投递了多次。
通知只告诉下游「去看任务」，不携带完整结果。结果放在对象存储或结果表里，按 resultRef 与 resultHash 读取。这样通知可以很小，重发也不会把大结果复制多份。
回调幂等：同一个成功只能落一次
执行器到编排层的回调是至少一次投递。apply 用 callbackId 去重：同一个回调 ID 第二次到达时返回 duplicate-callback，不改状态、不发通知。
即使回调 ID 不同，成功事件也要按结果哈希判重：同一个任务收到两次 succeed，哈希相同返回 duplicate-result；哈希不同说明两次执行产出了不同内容，返回冲突，交给人或上层策略处理。
失败事件分两类。可重试失败把任务放回 queued，保留错误信息；不可重试失败直接进 failed。示例里批量任务 A 第一次因配额失败、第二次成功，任务 B 因输入非法一次失败即终止。
断线重连：游标、增量与快照
任务每次状态变化把 version 加一，事件写进历史。客户端重连时带上最后看到的版本号：差距在保留窗口内返回增量事件，差距过大返回快照。
示例的保留窗口是 8 个版本。客户端看到版本 3 后断线，任务完成时版本是 5，重连拿到版本 4 和 5 两条增量。若客户端拿的是 0 且任务已经跑了 20 个版本，增量太长，直接返回快照，让客户端整体替换本地状态。
进度必须单调。低百分比的进度事件返回 stale-progress，不写入历史；相同百分比返回重复。重连协议只保证「状态最终一致」，不保证每条中间进度都送达。
结果交付幂等
结果交付按 idempotencyKey 记账。同一个键第二次交付返回同一张小票，不重复执行下游动作；不同键会再次投递，但 resultHash 相同，下游可以按内容哈希去重。一个键被复用到另一个任务时必须拒绝，否则两个任务的结果会串在一起。
这三层幂等——回调 ID、结果哈希、交付键——对应三个不同的重试来源：执行器重发、任务重复执行、客户端重试。少一层，就会在某一种重试下产生重复发布或重复计费。
实现
async-task.ts 把状态迁移、回调去重、通知 outbox、游标重连与结果交付放在同一个确定性模块里。时间由事件携带，不依赖系统时钟。
// file: src/orch/async-task.ts
// tier: L2
// verify: node --test test/orch/human.test.ts
import { AiError } from "../core/errors.ts";
export type AsyncState = "queued" | "running" | "succeeded" | "failed" | "expired";
export type AsyncEvent =
| { type: "start"; atMs: number; worker: string }
| { type: "progress"; atMs: number; progress: number; cursor: string }
| { type: "succeed"; atMs: number; resultRef: string; resultHash: string }
| { type: "fail"; atMs: number; error: string; retryable: boolean }
| { type: "expire"; atMs: number };
export interface AsyncTask {
id: string;
state: AsyncState;
/** 结果可交付截止时间；过了这个点不再接受成功回调。 */
expiresAtMs: number;
attempts: number;
maxAttempts: number;
progress: number;
version: number;
resultRef?: string;
resultHash?: string;
error?: string;
history: readonly { version: number; event: AsyncEvent }[];
}
export interface AsyncTransition {
task: AsyncTask;
changed: boolean;
duplicate: boolean;
notify: boolean;
reason: string;
}
export interface CallbackResult extends AsyncTransition {
duplicateCallback: boolean;
}
export interface DeliveryReceipt {
taskId: string;
resultRef: string;
resultHash: string;
idempotencyKey: string;
deliveredAtMs: number;
}
const TERMINAL: readonly AsyncState[] = ["succeeded", "failed", "expired"];
function fail(message: string): never {
throw new AiError("invalid-request", `async-task: ${message}`);
}
function integer(value: number, field: string, min = 0): void {
if (!Number.isInteger(value) || value < min) fail(`${field} 必须是 >= ${min} 的整数`);
}
function positive(value: number, field: string): void {
if (!Number.isFinite(value) || value <= 0) fail(`${field} 必须是正数`);
}
function nonNegative(value: number, field: string): void {
if (!Number.isFinite(value) || value < 0) fail(`${field} 必须是非负数`);
}
function isTerminal(state: AsyncState): boolean {
return TERMINAL.includes(state);
}
function advance(task: AsyncTask, event: AsyncEvent, patch: Partial<AsyncTask>): AsyncTask {
const version = task.version + 1;
return { ...task, ...patch, version, history: [...task.history, { version, event }] };
}
function unchanged(task: AsyncTask, reason: string, duplicate = false): AsyncTransition {
return { task, changed: false, duplicate, notify: false, reason };
}
export function createAsyncTask(input: {
id: string;
nowMs: number;
ttlMs: number;
maxAttempts: number;
}): AsyncTask {
if (input.id.trim() === "") fail("task.id 不能为空");
nonNegative(input.nowMs, "nowMs");
positive(input.ttlMs, "ttlMs");
integer(input.maxAttempts, "maxAttempts", 1);
return {
id: input.id,
state: "queued",
expiresAtMs: input.nowMs + input.ttlMs,
attempts: 0,
maxAttempts: input.maxAttempts,
progress: 0,
version: 0,
history: [],
};
}
/** 过期是系统动作，时间固定落在 expiresAtMs，重放可复现。 */
export function expireTask(task: AsyncTask): AsyncTransition {
if (isTerminal(task.state)) {
return unchanged(task, task.state === "expired" ? "duplicate-expire" : "terminal-conflict",
task.state === "expired");
}
return {
task: advance(task, { type: "expire", atMs: task.expiresAtMs }, { state: "expired" }),
changed: true,
duplicate: false,
notify: true,
reason: "expired",
};
}
/** 纯状态转移：过期判定先于回调处理；TTL 是结果可交付截止时间。 */
export function applyAsyncEvent(task: AsyncTask, event: AsyncEvent): AsyncTransition {
nonNegative(event.atMs, "event.atMs");
if (isTerminal(task.state)) {
if (event.type === "succeed" && task.state === "succeeded" && task.resultHash === event.resultHash) {
return unchanged(task, "duplicate-result", true);
}
if (event.type === "expire" && task.state === "expired") {
return unchanged(task, "duplicate-expire", true);
}
return unchanged(task, "terminal-conflict");
}
if (event.type === "expire" || event.atMs >= task.expiresAtMs) return expireTask(task);
if (event.type === "start") {
if (task.state === "running") return unchanged(task, "already-running", true);
if (task.state !== "queued" || task.attempts >= task.maxAttempts) {
return unchanged(task, "not-startable");
}
return {
task: advance(task, event, { state: "running", attempts: task.attempts + 1, progress: 0 }),
changed: true,
duplicate: false,
notify: false,
reason: "started",
};
}
if (event.type === "progress") {
if (task.state !== "running") return unchanged(task, "not-running");
integer(event.progress, "progress", 0);
if (event.progress > 100) fail("progress 不能超过 100");
if (event.progress < task.progress) return unchanged(task, "stale-progress");
if (event.progress === task.progress) return unchanged(task, "duplicate-progress", true);
return {
task: advance(task, event, { progress: event.progress }),
changed: true,
duplicate: false,
notify: false,
reason: "progress",
};
}
if (task.state !== "running") return unchanged(task, "not-running");
if (event.type === "succeed") {
if (event.resultRef.trim() === "" || event.resultHash.trim() === "") {
fail("resultRef 与 resultHash 不能为空");
}
return {
task: advance(task, event, {
state: "succeeded",
progress: 100,
resultRef: event.resultRef,
resultHash: event.resultHash,
}),
changed: true,
duplicate: false,
notify: true,
reason: "succeeded",
};
}
if (event.retryable && task.attempts < task.maxAttempts) {
return {
task: advance(task, event, { state: "queued", progress: 0, error: event.error }),
changed: true,
duplicate: false,
notify: false,
reason: "requeued",
};
}
return {
task: advance(task, event, { state: "failed", error: event.error }),
changed: true,
duplicate: false,
notify: true,
reason: "failed",
};
}
/** 回调去重、通知 outbox、游标重连与结果交付都收在这里。 */
export class AsyncBook {
tasks = new Map<string, AsyncTask>();
callbacks = new Map<string, { taskId: string; atMs: number }>();
notifications = new Map<string, { taskId: string; state: AsyncState; atMs: number }>();
deliveries = new Map<string, DeliveryReceipt>();
/** 超过这个版本差就返回快照，避免增量过长。 */
retentionVersions = 8;
create(input: {
id: string;
nowMs: number;
ttlMs: number;
maxAttempts: number;
}): AsyncTask {
const task = createAsyncTask(input);
this.tasks.set(task.id, task);
return task;
}
get(id: string): AsyncTask | undefined {
return this.tasks.get(id);
}
/** callbackId 相同即同一次回调，重复投递不产生第二次状态变更。 */
apply(taskId: string, event: AsyncEvent, callbackId?: string): CallbackResult {
const task = this.tasks.get(taskId);
if (task === undefined) fail(`未知任务 ${taskId}`);
if (callbackId !== undefined && this.callbacks.has(callbackId)) {
return { ...unchanged(task, "duplicate-callback", true), duplicateCallback: true };
}
const result = applyAsyncEvent(task, event);
if (result.changed) this.tasks.set(taskId, result.task);
if (callbackId !== undefined) this.callbacks.set(callbackId, { taskId, atMs: event.atMs });
if (result.notify) {
const key = `${taskId}:${result.task.state}:${result.task.version}`;
if (!this.notifications.has(key)) {
this.notifications.set(key, { taskId, state: result.task.state, atMs: event.atMs });
}
}
return { ...result, duplicateCallback: false };
}
/** 客户端带版本游标重连：窗口内给增量，太旧给快照。 */
resume(taskId: string, cursorVersion: number): {
mode: "caught-up" | "delta" | "snapshot";
events: readonly { version: number; event: AsyncEvent }[];
task: AsyncTask;
} {
const task = this.tasks.get(taskId);
if (task === undefined) fail(`未知任务 ${taskId}`);
integer(cursorVersion, "cursorVersion");
if (task.version <= cursorVersion) return { mode: "caught-up", events: [], task };
if (task.version - cursorVersion > this.retentionVersions) {
return { mode: "snapshot", events: [], task };
}
return {
mode: "delta",
events: task.history.filter((entry) => entry.version > cursorVersion),
task,
};
}
/** 同键返回同一张小票；不同键会再次投递，但 resultHash 相同。 */
deliverResult(taskId: string, idempotencyKey: string, nowMs: number): {
ok: boolean;
duplicate: boolean;
reason: string;
receipt?: DeliveryReceipt;
} {
if (idempotencyKey.trim() === "") fail("idempotencyKey 不能为空");
nonNegative(nowMs, "nowMs");
const existing = this.deliveries.get(idempotencyKey);
if (existing !== undefined) {
return existing.taskId === taskId
? { ok: true, duplicate: true, reason: "duplicate-delivery", receipt: existing }
: { ok: false, duplicate: true, reason: "key-reused-for-other-task", receipt: existing };
}
const task = this.tasks.get(taskId);
if (task === undefined) fail(`未知任务 ${taskId}`);
if (task.state !== "succeeded" || task.resultRef === undefined || task.resultHash === undefined) {
return { ok: false, duplicate: false, reason: "not-succeeded" };
}
const receipt: DeliveryReceipt = {
taskId,
resultRef: task.resultRef,
resultHash: task.resultHash,
idempotencyKey,
deliveredAtMs: nowMs,
};
this.deliveries.set(idempotencyKey, receipt);
return { ok: true, duplicate: false, reason: "delivered", receipt };
}
/** 取出并清空通知 outbox；一次终态最多一条。 */
drainNotifications(): { taskId: string; state: AsyncState; atMs: number }[] {
const list = [...this.notifications.values()].sort((a, b) =>
a.taskId === b.taskId ? a.atMs - b.atMs : a.taskId < b.taskId ? -1 : 1
);
this.notifications.clear();
return list;
}
}
applyAsyncEvent 先判过期，再按事件类型迁移；AsyncBook 在状态迁移之外维护回调表、通知表与交付小票。所有判断只依赖传入的事件与任务快照，因此可以在测试里完整重放。
示例样本上的运行结果
长文档摘要任务的时间线：
| 时间 | 事件 | 版本 |
|---|---|---|
| 0ms | 创建，TTL 1 小时 | 0 |
| 1,200ms | 开始执行 | 1 |
| 6,400ms | 进度 25% | 2 |
| 13,100ms | 进度 55% | 3 |
| 20,700ms | 进度 82% | 4 |
| 24,300ms | 成功，结果哈希 sha256:ab12 | 5 |
客户端在版本 3 断线，重连拿到版本 4、5 两条增量；重复的成功回调返回 duplicate-result，通知只发 1 条。批量审核三任务样本的结果是：1 个重试后成功、1 个永久失败、1 个过期，通知 3 条；2 次重复回调被去重，没有产生额外状态变更。
回声链路里的异步边界
内容发布链 A 的长文档摘要与人工复核都应该异步：摘要走任务状态机，复核走 6.2 的复核状态机，两者通过任务 ID 关联。私信客服链 B 的 chat.reply 必须留在 1,269ms 的同步路径上，不能因为「异步更稳」就把对话回复也推到后台。站内下单咨询链 C 的检索与回复是同步的，下单动作可以由异步结果触发，但授权与幂等必须在动作入口完成。
判断标准不是任务长短，而是用户是否在等这次结果。用户在等，就守同步预算；用户可以先离开，才用异步交付。
小结
异步任务需要五个状态、一次过期判定、一套通知 outbox、三层幂等和一条游标重连协议。示例样本里，24.3 秒的摘要在版本 3 断线后只补两条增量，重复成功回调没有产生第二条通知。
异步化真正难的不是把任务放进队列，而是让「完成」这件事在重复、断线与重放下仍然只发生一次。
阅读导航：上一节：6.2 人工介入节点 · 下一节：7.1 端到端评测集 。
常见错误与边界
- 用
setTimeout代替任务状态机。进程重启后任务丢失，也没有重试与过期。 - 成功回调到了就写结果，不看 TTL。过期任务会覆盖已经交付的旧结果。
- 通知携带完整结果。通知重发会把大对象复制多份，还会让重试成本失控。
- 只按回调 ID 去重。执行器换一个 ID 重发同一结果时，仍会产生第二次状态变更。
- 结果交付不记账。客户端重试会重复触发下游动作，比如重复发送邮件或重复扣费。
- 断线后从头发全量事件。事件一多，重连成本会超过任务本身。
- 进度允许回退。客户端会看到进度条倒退，状态判断也会出现竞态。
- 把对话回复也异步化。用户在等答案，异步只会把延迟换成一次额外的轮询。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。