diff --git a/apps/server/src/modules/agent/runner.ts b/apps/server/src/modules/agent/runner.ts index 9b055236..bc144e6d 100644 --- a/apps/server/src/modules/agent/runner.ts +++ b/apps/server/src/modules/agent/runner.ts @@ -27,7 +27,7 @@ import { type RunUpstream, resolveRunUpstream } from "../pi/index.ts"; import type { ServerContext } from "../types.ts"; import { DEMO_SCRIPTS, parseFakeScripts } from "./fake-scripts.ts"; import { buildGoalReport } from "./report.ts"; -import { type AwaitSuspension, attachLiveRun, type LiveRun, subagentSettings } from "./subagents.ts"; +import { type AwaitSuspension, attachLiveRun, type LiveRun, subagentSettings, wakeParent } from "./subagents.ts"; // runAgent 任务的执行体:从设置装配 Pi(或 fake)运行时 → 事件先写 agent_run_event 再推 SSE → // 崩溃恢复(pendingOperations>0 时先按幂等键对账未知结果再 resume)→ 消费追加消息 → 汇报以业务表为准。 @@ -502,11 +502,14 @@ export async function drive( export async function goalSettled(env: AgentEnv, runId: string, modelText: string): Promise { const r = await buildGoalReport(env.ctx.db, runId, modelText); if (r.state !== "waiting") return r; - await env.ctx.db + const u = await env.ctx.db .updateTable("agent_run") .set({ status: "waiting", summary: r.summary }) .where("id", "=", runId) .where("status", "in", ["running", "waiting"]) - .execute(); + .returning(["parent_run_id", sql`(select status = 'running' from agent_run o where o.id = ${runId})`.as("was_running")]) + .executeTakeFirst(); + // 子会话刚转 waiting(对父会话来说已有结果):唤醒挂起等待的父会话 + if (u?.parent_run_id && u.was_running) await wakeParent(env.ctx, runId); return null; } diff --git a/apps/server/src/modules/agent/service.ts b/apps/server/src/modules/agent/service.ts index 668bd8a1..cba5a253 100644 --- a/apps/server/src/modules/agent/service.ts +++ b/apps/server/src/modules/agent/service.ts @@ -12,7 +12,7 @@ import { type AgentRunRow, type DB, type Kysely, sql } from "@umt/db"; import type { ServerContext } from "../types.ts"; import { buildGoalReport } from "./report.ts"; import { type AgentEnv, type DriveResult, INVESTIGATE_CAPABILITIES, publishStatus, scopeFor } from "./runner.ts"; -import { cascadeCancel, descendants, descendantUsage, recordChildCompletion } from "./subagents.ts"; +import { cascadeCancel, descendants, descendantUsage, recordChildCompletion, wakeParent } from "./subagents.ts"; // agent_run 生命周期:建行 + 业务任务行 + 入队 runAgent;追加消息 / 中止;终态汇报(业务表核查)与后台调查去重。 @@ -340,6 +340,8 @@ export async function finalize( const fin = await ctx.db.selectFrom("agent_run").selectAll().where("id", "=", runId).executeTakeFirst(); if (fin) await recordChildCompletion(ctx, fin); } + // 子会话落定(含 waiting):唤醒挂起等待它的父会话 + if (run.parent_run_id) await wakeParent(ctx, runId); // 结束前刚到的追加消息:再唤醒一次 const pending = await ctx.db .selectFrom("agent_run_message") diff --git a/apps/server/src/modules/agent/subagents.ts b/apps/server/src/modules/agent/subagents.ts index c33c0778..cdc874d7 100644 --- a/apps/server/src/modules/agent/subagents.ts +++ b/apps/server/src/modules/agent/subagents.ts @@ -31,9 +31,11 @@ import type { AgentEnv, RunScope } from "./runner.ts"; // 能力白名单 ⊆ 父、模式不高于父、预算从父剩余预算划拨;settings.pi.subagents{enabled,maxConcurrent,maxDepth}。 // - 并发:子会话在 runAgent 里先占名额(父会话下同时 running 的子会话 < maxConcurrent,咨询锁内原子判定), // 没名额就 waitForFact 挂起(不占 worker、不占模型)。 -// - agent.await:等子会话结束,返回脱敏后的汇报摘要。runAgent 驱动的会话里,子会话还没结束时挂起: -// 关闭本会话(工具调用留在 JSONL 里未完成),runAgent 用 waitForFact 等子会话落定(不占 worker、不占模型), -// 再打开会话 resume —— agent.await 是只读能力(replay=safe),恢复时重放一次拿到结果。见 AwaitSuspension。 +// - agent.await / agent.spawn{await:true}:等子会话结束,返回脱敏后的汇报摘要。runAgent 驱动的会话里, +// 子会话还没结束时工具调用立即返回(挂起):关闭本会话(工具调用留在 JSONL 里未完成),runAgent 在 agent_run.await_wake +// 登记唤醒信号后用 waitForFact 等子会话落定(不占 worker 槽位、不调用模型);子会话落定时 wakeParent 写 fact +// 唤醒父会话,再打开会话 resume —— 两个能力都是 replay=safe,恢复时重放一次,子会话汇报作为工具结果续跑。 +// 见 AwaitSuspension。 // - 事件:父会话运行进程把子会话事件按父 seq 镜像进父事件流(subagent_spawned / subagent_event / subagent_completed); // 父会话已不在运行时,由直接写入器在咨询锁内补记 subagent_completed。 // - 取消:父会话取消递归传播到全部子孙会话;预算:父会话运行时把子孙会话用量计入自己的预算。 @@ -66,7 +68,23 @@ const safe = (s: string | null | undefined, n = 1500) => clip(redactString(s ?? // ── 进程内活跃会话(镜像泵 + 用量缓存) ── -/** agent.await 挂起请求:runAgent 据此 waitForFact,落定后以 resumedAwaits 重开会话 */ +/** + * 挂起中的工具调用(agent.spawn{await:true})抛出这个错误:账本记 failed、不缓存结果, + * 会话恢复时重放同一调用(幂等键相同 → 同一个子会话)再取汇报。会话此时已关闭,模型看不到它。 + */ +export class AwaitSuspendedError extends CapabilityError { + constructor() { + super("precondition", "会话已挂起等待子会话结束,恢复后重放本调用"); + } +} + +/** runAgent 挂起期间登记在 agent_run.await_wake 的唤醒信号(子会话落定时 wakeParent 按它 signal) */ +export interface AwaitWake { + engineRunId: string; + fact: string; +} + +/** agent.await / agent.spawn{await:true} 挂起请求:runAgent 据此 waitForFact,落定后以 resumedAwaits 重开会话 */ export interface AwaitSuspension { /** 挂起的工具调用(恢复时重放,同一 invocationId 不再挂起) */ invocationId: string; @@ -588,6 +606,53 @@ export async function awaitReady(ctx: ServerContext, parentId: string, sus: Awai return rows.every((r) => SETTLED.includes(r.status)) ? true : null; } +/** 挂起前登记 / 恢复后清除唤醒信号(幂等) */ +export async function setAwaitWake(ctx: ServerContext, runId: string, wake: AwaitWake | null): Promise { + await ctx.db + .updateTable("agent_run") + .set({ await_wake: wake ? JSON.stringify(wake) : null }) + .where("id", "=", runId) + .execute(); + return null; +} + +/** + * 子会话落定(终态或 waiting)后唤醒挂起等待的父会话:按父会话登记的 await_wake 写 fact + 发引擎信号。 + * 父会话没挂起时什么也不做;失败只记日志(父会话还有兜底轮询)。 + */ +export async function wakeParent(ctx: ServerContext, childRunId: string): Promise { + try { + const r = await ctx.db + .selectFrom("agent_run as c") + .innerJoin("agent_run as p", "p.id", "c.parent_run_id") + .select(["p.id as parentId", "p.await_wake as wake"]) + .where("c.id", "=", childRunId) + .executeTakeFirst(); + const w = r?.wake as AwaitWake | null | undefined; + if (!w?.engineRunId || !w.fact) return false; + await ctx.engine.signal(w.engineRunId, w.fact, { childRunId }); + return true; + } catch (err) { + ctx.logger.warn({ childRunId, err: { message: (err as Error).message } }, "唤醒父会话失败(父会话会兜底轮询)"); + return false; + } +} + +/** + * 子会话还没落定时请求挂起:记下等待条件并关闭会话(本调用在 JSONL 里保持未完成),drive 看到 requested 后返回 + * suspended。同一批里已有别的调用在挂起时不再重复登记。 + */ +async function requestSuspension(live: LiveRun, invocationId: string, runIds: string[] | null, timeoutSec: number) { + if (!live.suspend || live.suspend.requested) return; + live.suspend.requested = { + invocationId, + runIds, + deadline: new Date(Date.now() + timeoutSec * 1000).toISOString(), + }; + await new Promise((r) => setTimeout(r, 0)); + await live.handle.close(); +} + // ── 能力处理器 ── export interface SpawnDeps { @@ -614,7 +679,8 @@ async function parentOf(ctx: ServerContext, c: HandlerContext): Promise { + // replay=safe:await=true 挂起后会话恢复时要重放本调用取子会话汇报;幂等键相同 → dedupe_key 命中同一个子会话 + registry.register({ ...capabilitySpec("agent.spawn"), replay: "safe" }, async (input, c) => { const parent = await parentOf(ctx, c); const s = await ctx.settings.all(); const cfg = subagentSettings(s); @@ -655,9 +721,23 @@ export function registerSubagentCapabilities(registry: CapabilityRegistry, ctx: child = r.run; } await liveRun(parent.id)?.pump(); - const result = input.await - ? ((await awaitChildren(ctx, parent.id, [child.id], input.timeoutSec, c.signal)).results[0] ?? null) - : null; + let result: SubagentResult | null = null; + if (input.await) { + const live = liveRun(parent.id); + const inv = c.invocationId; + if (live?.suspend && inv) { + // runAgent 驱动:不在工具调用里等,挂起会话(不占 worker),子会话落定后重放本调用 + const snap = await awaitChildren(ctx, parent.id, [child.id], 0, c.signal); + if (!snap.allDone && !live.suspend.resumed.has(inv)) { + // 崩溃对账(reconcile:*)只确保子会话已建,不挂起;真正的等待交给恢复后的重放 + if (!inv.startsWith("reconcile:")) await requestSuspension(live, inv, [child.id], input.timeoutSec); + throw new AwaitSuspendedError(); + } + result = snap.results[0] ?? null; + } else { + result = (await awaitChildren(ctx, parent.id, [child.id], input.timeoutSec, c.signal)).results[0] ?? null; + } + } const cs = (child.scope ?? {}) as Partial; return { runId: child.id, @@ -677,14 +757,8 @@ export function registerSubagentCapabilities(registry: CapabilityRegistry, ctx: // 恢复时重放:已挂起等过,直接返回当前状态 const snap = await awaitChildren(ctx, parent.id, input.runIds, 0, c.signal); if (snap.allDone || live.suspend.resumed.has(inv) || live.suspend.requested) return snap; - // 挂起:记下等待条件,关闭会话(本调用在 JSONL 里保持未完成);drive 看到 requested 后返回 suspended - live.suspend.requested = { - invocationId: inv, - runIds: input.runIds?.length ? [...input.runIds] : null, - deadline: new Date(Date.now() + input.timeoutSec * 1000).toISOString(), - }; - await new Promise((r) => setTimeout(r, 0)); - await live.handle.close(); + // 挂起:立即返回,会话关闭(本调用在 JSONL 里保持未完成,返回值不会进模型上下文);恢复后重放拿真实结果 + await requestSuspension(live, inv, input.runIds?.length ? [...input.runIds] : null, input.timeoutSec); return snap; }); } diff --git a/apps/server/src/modules/agent/tasks.ts b/apps/server/src/modules/agent/tasks.ts index 378fdd18..68f56260 100644 --- a/apps/server/src/modules/agent/tasks.ts +++ b/apps/server/src/modules/agent/tasks.ts @@ -17,14 +17,17 @@ import { import { buildGoalReport } from "./report.ts"; import { type AgentEnv, type DriveResult, drive, goalSettled } from "./runner.ts"; import { finalize } from "./service.ts"; -import { awaitReady, catchUpChildren, claimSubagentSlot } from "./subagents.ts"; +import { awaitReady, catchUpChildren, claimSubagentSlot, setAwaitWake } from "./subagents.ts"; // 持久任务:runAgent(驱动一次有界 Agent 会话 → 等目标关联的下载/入库落定 → 以业务表汇报) // 与 resolveDecision(decision.created → 自动调查;decision.updated(resolved)→ 执行所选处理)。 const id = z.string().uuid(); -/** agent.await 挂起时的唤醒轮询间隔与单个会话最多挂起次数 */ -const AWAIT_POLL_SEC = 2; +/** + * agent.await / agent.spawn{await:true} 挂起时的兜底轮询间隔(正常由子会话落定时的 wakeParent 信号立即唤醒; + * 信号在等待开始前送达而丢失时靠它兜底)与单个会话最多挂起次数 + */ +const AWAIT_POLL_SEC = 10; const MAX_AWAIT_SUSPENDS = 100; const DB_RETRY = { maximumAttempts: 3, initialInterval: "1s", backoffCoefficient: 2, maximumInterval: "30s" } as const; @@ -77,8 +80,8 @@ export function registerAgentTasks(registry: TaskRegistry, env: () => AgentEnv) res = { status: "failed", reason: "error", budget: null, message: "等待子会话名额超时", finalText: "" }; else { res = await t.step.run("drive", () => drive(e, e.ctx.capabilities, runId, t.abortSignal)); - // agent.await 挂起:会话已关闭,waitForFact 等子会话落定 / 超时 / 取消(不占 worker、不占模型), - // 再重开会话 resume(挂起的 agent.await 重放一次拿到结果) + // agent.await / agent.spawn{await:true} 挂起:会话已关闭;登记唤醒信号后 waitForFact 等子会话落定 / 超时 / 取消 + // (不占 worker 槽位、不调用模型),子会话落定时写 fact 唤醒;再重开会话 resume(挂起的调用重放一次拿到结果) const resumedAwaits: string[] = []; for (let n = 1; res.status === "suspended" && res.suspension; n++) { const sus = res.suspension; @@ -87,11 +90,15 @@ export function registerAgentTasks(registry: TaskRegistry, env: () => AgentEnv) break; } const waitMs = Math.max(0, Date.parse(sus.deadline) - Date.now()); - await t.step.waitForFact(`subagent-await-${n}`, { + const fact = `subagent-await-${n}`; + const engineRunId = t.runId; + await t.step.run(`${fact}-arm`, () => setAwaitWake(e.ctx, runId, { engineRunId, fact })); + await t.step.waitForFact(fact, { check: () => awaitReady(e.ctx, runId, sus), timeout: `${AWAIT_POLL_SEC}s`, maxRounds: Math.max(1, Math.ceil(waitMs / 1000 / AWAIT_POLL_SEC) + 1), }); + await t.step.run(`${fact}-disarm`, () => setAwaitWake(e.ctx, runId, null)); resumedAwaits.push(sus.invocationId); const resumed = [...resumedAwaits]; res = await t.step.run(`drive-${n}`, () => diff --git a/apps/server/test/e2e/e11-subagents.test.ts b/apps/server/test/e2e/e11-subagents.test.ts index ac2ba032..54a4307c 100644 --- a/apps/server/test/e2e/e11-subagents.test.ts +++ b/apps/server/test/e2e/e11-subagents.test.ts @@ -87,6 +87,39 @@ const abortParentScript: FakeScript = { finalText: "不应走到这里", }; +const SPAWN_AWAIT_GOAL = "派生并等待:检查《沙丘》原声版本"; +const SPAWN_AWAIT_CHILD = "子目标:检查《沙丘》原声版本"; +const spawnAwaitScript: FakeScript = { + name: "e11-spawn-await", + match: SPAWN_AWAIT_GOAL, + steps: [ + { + toolCalls: [ + { + capability: "agent.spawn", + input: { + goal: SPAWN_AWAIT_CHILD, + capabilityAllowlist: READONLY, + mode: "plan_only", + await: true, + timeoutSec: 300, + }, + }, + ], + }, + ], + finalText: (results) => { + const r = lastOk(results, "agent.spawn")?.result; + return `子会话 ${r?.status ?? "无结果"}:${r?.summary ?? ""}`; + }, +}; +const spawnAwaitChild: FakeScript = { + name: "e11-spawn-await-child", + match: SPAWN_AWAIT_CHILD, + steps: [{ toolCalls: [{ capability: "inventory.query", input: { query: "沙丘", limit: 1 } }] }], + finalText: "《沙丘》原声版本已在库", +}; + const slowChild: FakeScript = { name: "e11-slow", match: /^子目标:慢任务/, @@ -98,7 +131,16 @@ describe.skipIf(!hasPg)("E11 Pi 子代理(agent.spawn / agent.await)", () => { let e: E2E; beforeAll(async () => { - e = await createE2E("e11", { fakeScripts: [parentScript, duneScript, oppieScript, abortParentScript, slowChild] }); + e = await createE2E("e11", { fakeScripts: [ + parentScript, + duneScript, + oppieScript, + abortParentScript, + slowChild, + spawnAwaitScript, + spawnAwaitChild, + ], + }); await e.settings({ pi: { subagents: { enabled: true, maxConcurrent: 1, maxDepth: 1 } } }); }, 120_000); @@ -106,6 +148,31 @@ describe.skipIf(!hasPg)("E11 Pi 子代理(agent.spawn / agent.await)", () => { await e?.close(); }, 60_000); + /** 父会话的引擎 run(task.engine_run_id) */ + const engineRunOf = async (runId: string) => { + const t = await e.db + .selectFrom("task") + .select("engine_run_id") + .where("subject_type", "=", "agent_run") + .where("subject_id", "=", runId) + .executeTakeFirstOrThrow(); + return t.engine_run_id as string; + }; + /** 等到:父会话已挂起(登记了唤醒信号)且没有任何 run 被 worker 持有(io / agent 槽位全部空闲) */ + const suspendedAndIdle = (parentId: string) => + waitFor( + async () => { + const p = await runRow(e, parentId); + if (!p.await_wake) return null; + const held = (await e.booted.ctx.engine.list({ status: "running", limit: 200 })).filter( + (r) => r.status === "running", + ); + return held.length === 0 ? { wake: p.await_wake as Any } : null; + }, + 30_000, + `父会话 ${parentId} 挂起且 worker 槽位空闲`, + ); + const settle = (runId: string, statuses: string[], ms = 90_000) => waitFor( async () => { @@ -334,6 +401,10 @@ describe.skipIf(!hasPg)("E11 Pi 子代理(agent.spawn / agent.await)", () => { "父会话派生两个子会话并开始等待", ); expect(kids.every((k) => k.status === "queued")).toBe(true); + // agent.await 挂起:父会话不占 worker(引擎 run 未被持有),子会话在等名额也不占 → io / agent 槽位空闲 + const idle = await suspendedAndIdle(parentId); + expect(idle.wake.engineRunId).toBe(await engineRunOf(parentId)); + expect((await e.booted.ctx.engine.get(idle.wake.engineRunId))?.status).toBe("waiting"); const ab = await e.api("POST", `/agent/runs/${parentId}/abort`, { reason: "e11 中止" }); expect(ab.status).toBe(200); @@ -349,4 +420,90 @@ describe.skipIf(!hasPg)("E11 Pi 子代理(agent.spawn / agent.await)", () => { expect(b.cancel_requested_at).not.toBeNull(); await e.db.updateTable("agent_run").set({ status: "cancelled" }).where("id", "=", blocker.id).execute(); }, 120_000); + it("agent.spawn{await:true} 挂起:等待期间 io/agent 槽位空闲,子会话落定写 fact 唤醒父会话,子汇报作为工具结果续跑", async () => { + await e.worker?.stop(); + const r = await e.api("POST", "/agent/runs", { + goal: SPAWN_AWAIT_GOAL, + mode: "plan_only", + capabilityAllowlist: [...READONLY, "agent.spawn", "agent.await"], + }); + expect(r.status).toBe(201); + const parentId = r.body.data.id as string; + // 占住唯一名额:真实子会话排队,父会话必须挂起等待 + const blocker = await e.db + .insertInto("agent_run") + .values({ + trigger: "subagent", + goal: "占名额", + status: "running", + mode: "plan_only", + scope: JSON.stringify({ capabilities: READONLY }), + budget: JSON.stringify({}), + parent_run_id: parentId, + depth: 1, + started_at: new Date(), + }) + .returning("id") + .executeTakeFirstOrThrow(); + await e.startWorker(); + + const idle = await suspendedAndIdle(parentId); + const engineRunId = await engineRunOf(parentId); + expect(idle.wake).toEqual({ engineRunId, fact: "subagent-await-1" }); + expect((await e.booted.ctx.engine.get(engineRunId))?.status).toBe("waiting"); + const child = await e.db + .selectFrom("agent_run") + .selectAll() + .where("parent_run_id", "=", parentId) + .where("id", "<>", blocker.id) + .executeTakeFirstOrThrow(); + expect(child.status).toBe("queued"); + // 挂起的调用没有把"未结束"的结果交给模型:工具调用只有开始、没有结束 + const before = await runEvents(e, parentId); + expect(before.some((x) => x.type === "tool_call_end" && (x.payload as Any).capabilityId === "agent.spawn")).toBe( + false, + ); + + // 放出名额 → 子会话跑完 → 写 fact 唤醒父会话(不用等兜底轮询) + await e.db + .updateTable("agent_run") + .set({ status: "succeeded", finished_at: new Date() }) + .where("id", "=", blocker.id) + .execute(); + const doneChild = await settle(child.id, ["succeeded", "failed", "cancelled", "budget_exceeded"]); + expect(doneChild.status).toBe("succeeded"); + const parent = await settle(parentId, ["succeeded", "failed", "cancelled", "budget_exceeded"], 30_000); + expect(parent.status).toBe("succeeded"); + expect(parent.await_wake).toBeNull(); + const fact = await e.db + .selectFrom("tasks_fact") + .select(["name", "payload"]) + .where("run_id", "=", engineRunId) + .where("name", "=", "subagent-await-1") + .execute(); + expect(fact.map((f) => (f.payload as Any).childRunId)).toContain(child.id); + + const events = await runEvents(e, parentId); + const spawnEnd = events.find( + (x) => x.type === "tool_call_end" && (x.payload as Any).capabilityId === "agent.spawn", + )!; + expect(spawnEnd).toBeTruthy(); + expect((spawnEnd.payload as Any).isError).toBe(false); + expect((spawnEnd.payload as Any).recovery).toBe(true); + expect(events.filter((x) => x.type === "run_started").map((x) => (x.payload as Any).resumed)).toEqual([ + false, + true, + ]); + const done = events.find((x) => x.type === "run_completed")!; + expect((done.payload as Any).finalText).toContain("子会话 succeeded"); + expect((done.payload as Any).finalText).toContain("《沙丘》原声版本已在库"); + // 只派生了一个子会话(重放命中同一个 dedupe_key) + const kids = await e.db + .selectFrom("agent_run") + .select("id") + .where("parent_run_id", "=", parentId) + .where("id", "<>", blocker.id) + .execute(); + expect(kids).toHaveLength(1); + }, 120_000); }); diff --git a/docs/runbook/agent.md b/docs/runbook/agent.md index 475792e7..f617f079 100644 --- a/docs/runbook/agent.md +++ b/docs/runbook/agent.md @@ -200,7 +200,8 @@ OpenRouter 换到的是永久 key,不刷新。 - 预算:子会话预算从父会话剩余预算划拨(缺省剩余的一半,请求超过剩余直接拒绝),子会话超预算只结束它自己;父会话的工具调用 / token / 费用预算把子孙会话的实际用量算在内。 - 取消:中止父会话会递归取消全部未结束的子孙会话(排队中的也一样)。 - 事件:子会话的开始、工具调用、结束按父会话 seq 镜像为 `subagent_spawned` / `subagent_event` / `subagent_completed`,SSE 上以 `agent.tool_call`(能力 `agent.spawn`)提示;会话详情返回 `parent`、`children` 与 `familyUsage`(会话树合计)。 -- `agent.await` 在父会话的工具调用里轮询业务表等待,不调用模型,但等待期间父会话占着一个 Agent worker 槽位,超时(缺省 600 秒)返回 pending。 +- 等待(`agent.await`,以及 `agent.spawn {await:true}`):子会话还没落定时工具调用立即返回,父会话挂起——会话文件里这次调用保持未完成,父 run 在 `agent_run.await_wake` 登记唤醒信号 `{engineRunId, fact}` 后用 `waitForFact` 等待,**不占 io / agent worker 槽位,也不调用模型**。子会话落定(终态或 waiting)时 `wakeParent` 按登记写 fact 并发引擎信号,父会话立刻重开会话续跑:挂起的调用(两个能力都是 replay=safe)重放一次,子会话汇报作为工具结果交给模型。信号丢失(子会话恰在父会话开始等待前落定)时靠 10 秒兜底轮询;超时(缺省 600 秒)后未结束的子会话标 pending,父会话被取消时立即唤醒按取消收尾。`agent.spawn {await:true}` 挂起那次调用在账本里记为 failed(`会话已挂起…`,不缓存结果),重放时幂等键相同,命中同一个子会话。 +- 排查:父会话挂起期间 `agent_run.await_wake` 非空、引擎 run 状态为 waiting(`GET /engine/runs/{id}`);恢复后 `await_wake` 清空,事件流里第二个 `run_started` 的 `resumed=true`,重放调用的 `tool_call_end.recovery=true`。 - 成本:每个子会话都是一次完整的模型会话(独立上下文,有自己的系统提示与工具定义),token 成本大致随子会话数线性增加,也计入每日 Agent 运行次数。 ## fake 模式演示 diff --git a/packages/db/migrations/0037_agent_await_passkey.ts b/packages/db/migrations/0037_agent_await_passkey.ts new file mode 100644 index 00000000..cf048867 --- /dev/null +++ b/packages/db/migrations/0037_agent_await_passkey.ts @@ -0,0 +1,63 @@ +import type { Kysely } from "kysely"; +import { dropTables, entityTable, execDdl, oneOf } from "../src/migration-helpers.ts"; + +// C1-B Agent 与账户遗留收尾: +// - agent_run.await_wake:父会话挂起等待子会话(agent.await / agent.spawn{await:true})时登记的唤醒信号 +// {engineRunId, fact};子会话落定时按它写 fact + 发引擎信号,父会话恢复后清空。 +// - admin_recovery_code:TOTP 备用恢复码(一次性,argon2id 哈希存库,used_at 非空即已用)。 +// - admin_passkey:WebAuthn Passkey(COSE 公钥、签名计数器、传输方式、名称),可多把。 +// - webauthn_challenge:注册 / 登录挑战(一次性、5 分钟有效;条件更新消费,防重放)。 + +// biome-ignore lint/suspicious/noExplicitAny: 迁移不依赖应用表类型 +export async function up(db: Kysely): Promise { + await execDdl( + db, + ` +alter table agent_run add column await_wake jsonb; + +create table admin_recovery_code ( + id uuid primary key default gen_random_uuid(), + admin_user_id uuid not null references admin_user (id) on delete cascade, + code_hash text not null, + used_at timestamptz, + created_at timestamptz not null default now() +); +create index admin_recovery_code_unused_idx on admin_recovery_code (admin_user_id) where used_at is null; + +${entityTable( + "admin_passkey", + ` + admin_user_id uuid not null references admin_user (id) on delete cascade, + credential_id text not null, + public_key bytea not null, + counter bigint not null default 0 check (counter >= 0), + transports jsonb not null default '[]'::jsonb, + device_type text, + backed_up boolean not null default false, + name text not null, + last_used_at timestamptz, + constraint admin_passkey_credential_key unique (credential_id)`, +)} + +create table webauthn_challenge ( + id uuid primary key default gen_random_uuid(), + purpose text not null ${oneOf("purpose", ["register", "login"])}, + challenge text not null, + admin_user_id uuid references admin_user (id) on delete cascade, + expires_at timestamptz not null, + used_at timestamptz, + created_at timestamptz not null default now() +); +create index webauthn_challenge_expires_idx on webauthn_challenge (expires_at); +`, + ); +} + +// biome-ignore lint/suspicious/noExplicitAny: 同上 +export async function down(db: Kysely): Promise { + await execDdl( + db, + `${dropTables(["webauthn_challenge", "admin_passkey", "admin_recovery_code"])} +alter table agent_run drop column if exists await_wake;`, + ); +} diff --git a/packages/db/src/types.ts b/packages/db/src/types.ts index 432e48ab..77d41f26 100644 --- a/packages/db/src/types.ts +++ b/packages/db/src/types.ts @@ -246,6 +246,7 @@ export type TaskStatus = "pending" | "running" | "waiting" | "succeeded" | "fail export type TasksOpLedgerStatus = "started" | "succeeded"; export type TitleAliasKind = "original" | "translated" | "short" | "site_title" | "romanized" | "other"; export type TitleAliasSource = "tmdb" | "user" | "site" | "agent" | "nfo" | "import"; +export type WebauthnChallengePurpose = "register" | "login"; export interface AcquisitionPlanTable { id: Generated; @@ -296,6 +297,29 @@ export interface AcquisitionSlotTable { position: Generated; } +export interface AdminPasskeyTable { + id: Generated; + admin_user_id: string; + credential_id: string; + public_key: Buffer; + counter: Generated; + transports: Generated; + device_type: string | null; + backed_up: Generated; + name: string; + last_used_at: Timestamptz | null; + created_at: GeneratedTimestamptz; + updated_at: GeneratedTimestamptz; +} + +export interface AdminRecoveryCodeTable { + id: Generated; + admin_user_id: string; + code_hash: string; + used_at: Timestamptz | null; + created_at: GeneratedTimestamptz; +} + export interface AdminSessionTable { id: Generated; admin_user_id: string; @@ -355,6 +379,7 @@ export interface AgentRunTable { upstream_id: string | null; depth: Generated; spawn_reason: string | null; + await_wake: Json | null; } export interface AgentRunEventTable { @@ -1552,9 +1577,21 @@ export interface TrashItemTable { updated_at: GeneratedTimestamptz; } +export interface WebauthnChallengeTable { + id: Generated; + purpose: WebauthnChallengePurpose; + challenge: string; + admin_user_id: string | null; + expires_at: Timestamptz; + used_at: Timestamptz | null; + created_at: GeneratedTimestamptz; +} + export interface DB { acquisition_plan: AcquisitionPlanTable; acquisition_slot: AcquisitionSlotTable; + admin_passkey: AdminPasskeyTable; + admin_recovery_code: AdminRecoveryCodeTable; admin_session: AdminSessionTable; admin_user: AdminUserTable; agent_run: AgentRunTable; @@ -1627,6 +1664,7 @@ export interface DB { tasks_run_meta: TasksRunMetaTable; title_alias: TitleAliasTable; trash_item: TrashItemTable; + webauthn_challenge: WebauthnChallengeTable; } export type AcquisitionPlanRow = Selectable; @@ -1635,6 +1673,12 @@ export type AcquisitionPlanUpdate = Updateable; export type AcquisitionSlotRow = Selectable; export type NewAcquisitionSlot = Insertable; export type AcquisitionSlotUpdate = Updateable; +export type AdminPasskeyRow = Selectable; +export type NewAdminPasskey = Insertable; +export type AdminPasskeyUpdate = Updateable; +export type AdminRecoveryCodeRow = Selectable; +export type NewAdminRecoveryCode = Insertable; +export type AdminRecoveryCodeUpdate = Updateable; export type AdminSessionRow = Selectable; export type NewAdminSession = Insertable; export type AdminSessionUpdate = Updateable; @@ -1851,3 +1895,6 @@ export type TitleAliasUpdate = Updateable; export type TrashItemRow = Selectable; export type NewTrashItem = Insertable; export type TrashItemUpdate = Updateable; +export type WebauthnChallengeRow = Selectable; +export type NewWebauthnChallenge = Insertable; +export type WebauthnChallengeUpdate = Updateable;