diff --git a/apps/server/src/modules/agent/runner.ts b/apps/server/src/modules/agent/runner.ts index a4ea65ec..9b055236 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 { attachLiveRun, type LiveRun, subagentSettings } from "./subagents.ts"; +import { type AwaitSuspension, attachLiveRun, type LiveRun, subagentSettings } from "./subagents.ts"; // runAgent 任务的执行体:从设置装配 Pi(或 fake)运行时 → 事件先写 agent_run_event 再推 SSE → // 崩溃恢复(pendingOperations>0 时先按幂等键对账未知结果再 resume)→ 消费追加消息 → 汇报以业务表为准。 @@ -108,11 +108,13 @@ export function sessionDirOf(env: AgentEnv, configured?: string | null): string } export interface DriveResult { - status: "completed" | "failed" | "cancelled"; + /** suspended:agent.await 挂起等子会话(会话已关闭,runAgent waitForFact 后以 resumedAwaits 重开) */ + status: "completed" | "failed" | "cancelled" | "suspended"; reason: "budget" | "aborted" | "error" | null; budget: string | null; message: string | null; finalText: string; + suspension?: AwaitSuspension; } /** @@ -305,6 +307,7 @@ export async function drive( registry: CapabilityRegistry, runId: string, signal: AbortSignal, + opts: { resumedAwaits?: readonly string[] } = {}, ): Promise { const { ctx } = env; const run = await ctx.db.selectFrom("agent_run").selectAll().where("id", "=", runId).executeTakeFirstOrThrow(); @@ -406,7 +409,7 @@ export async function drive( } catch (err) { return { status: "failed", reason: "error", budget: null, message: (err as Error).message, finalText: "" }; } - const liveRun = await attachLiveRun(env, runId, handle); + const liveRun = await attachLiveRun(env, runId, handle, opts); external = liveRun.run; // 取消传播:任务取消信号或 agent_run.cancel_requested_at → 中止会话 @@ -428,8 +431,42 @@ export async function drive( void liveRun.run.pump(); }, 1000); + const suspended = (): DriveResult | null => { + const sus = liveRun.run.suspend?.requested; + return sus + ? { status: "suspended", reason: null, budget: null, message: null, finalText: "", suspension: sus } + : null; + }; try { - let outcome: RunOutcome | null = null; + let outcome: RunOutcome; + try { + outcome = await runOutcome(); + } catch (err) { + // agent.await 挂起时会话被主动关闭,prompt / resume 以 HarnessClosed 结束 + const sus = suspended(); + if (sus) return sus; + throw err; + } + const sus = suspended(); + if (sus) return sus; + if (outcome.status === "completed") + return { status: "completed", reason: null, budget: null, message: null, finalText: outcome.finalText }; + return { + status: outcome.reason === "aborted" ? "cancelled" : "failed", + reason: outcome.reason ?? "error", + budget: outcome.budget ?? null, + message: outcome.message ?? null, + finalText: outcome.finalText, + }; + } finally { + clearInterval(watch); + signal.removeEventListener("abort", abortOnce); + await liveRun.stop(); + await handle.close(); + } + + async function runOutcome(): Promise { + let outcome: RunOutcome; if (handle.pendingOperations > 0) { await reconcilePending(env, registry, run, signal); outcome = await handle.resume(); @@ -452,25 +489,12 @@ export async function drive( }; } } - while (outcome.status === "completed" && !aborting) { + while (outcome.status === "completed" && !aborting && !liveRun.run.suspend?.requested) { const text = await claimMessage(ctx, runId); if (text === null) break; outcome = await handle.prompt(text); } - if (outcome.status === "completed") - return { status: "completed", reason: null, budget: null, message: null, finalText: outcome.finalText }; - return { - status: outcome.reason === "aborted" ? "cancelled" : "failed", - reason: outcome.reason ?? "error", - budget: outcome.budget ?? null, - message: outcome.message ?? null, - finalText: outcome.finalText, - }; - } finally { - clearInterval(watch); - signal.removeEventListener("abort", abortOnce); - await liveRun.stop(); - await handle.close(); + return outcome; } } diff --git a/apps/server/src/modules/agent/subagents.ts b/apps/server/src/modules/agent/subagents.ts index fdca137d..c33c0778 100644 --- a/apps/server/src/modules/agent/subagents.ts +++ b/apps/server/src/modules/agent/subagents.ts @@ -31,7 +31,9 @@ import type { AgentEnv, RunScope } from "./runner.ts"; // 能力白名单 ⊆ 父、模式不高于父、预算从父剩余预算划拨;settings.pi.subagents{enabled,maxConcurrent,maxDepth}。 // - 并发:子会话在 runAgent 里先占名额(父会话下同时 running 的子会话 < maxConcurrent,咨询锁内原子判定), // 没名额就 waitForFact 挂起(不占 worker、不占模型)。 -// - agent.await:等子会话结束(轮询业务表,不调用模型),返回脱敏后的汇报摘要。 +// - agent.await:等子会话结束,返回脱敏后的汇报摘要。runAgent 驱动的会话里,子会话还没结束时挂起: +// 关闭本会话(工具调用留在 JSONL 里未完成),runAgent 用 waitForFact 等子会话落定(不占 worker、不占模型), +// 再打开会话 resume —— agent.await 是只读能力(replay=safe),恢复时重放一次拿到结果。见 AwaitSuspension。 // - 事件:父会话运行进程把子会话事件按父 seq 镜像进父事件流(subagent_spawned / subagent_event / subagent_completed); // 父会话已不在运行时,由直接写入器在咨询锁内补记 subagent_completed。 // - 取消:父会话取消递归传播到全部子孙会话;预算:父会话运行时把子孙会话用量计入自己的预算。 @@ -64,8 +66,20 @@ const safe = (s: string | null | undefined, n = 1500) => clip(redactString(s ?? // ── 进程内活跃会话(镜像泵 + 用量缓存) ── +/** agent.await 挂起请求:runAgent 据此 waitForFact,落定后以 resumedAwaits 重开会话 */ +export interface AwaitSuspension { + /** 挂起的工具调用(恢复时重放,同一 invocationId 不再挂起) */ + invocationId: string; + /** 等待的子会话;null = 本会话全部子会话 */ + runIds: string[] | null; + /** 超过这个时间(ISO)不再等,恢复后返回当前状态(未结束的标 pending) */ + deadline: string; +} + export interface LiveRun { handle: AgentRunHandle; + /** runAgent 的 drive 打开的会话才可挂起;已挂起过的 invocationId 恢复时不再挂起 */ + suspend?: { resumed: ReadonlySet; requested: AwaitSuspension | null }; /** 把子会话的新事件 / 新派生 / 结束镜像进本会话事件流(串行) */ pump(): Promise; /** 子孙会话已用量(同步读缓存,pump 刷新) */ @@ -80,6 +94,7 @@ export async function attachLiveRun( env: AgentEnv, runId: string, handle: AgentRunHandle, + opts: { resumedAwaits?: readonly string[] } = {}, ): Promise<{ run: LiveRun; stop(): Promise }> { const db = env.ctx.db; const prior = await db @@ -140,7 +155,12 @@ export async function attachLiveRun( chain = next; return next; }; - const run: LiveRun = { handle, pump, externalUsage: () => usage }; + const run: LiveRun = { + handle, + pump, + externalUsage: () => usage, + suspend: { resumed: new Set(opts.resumedAwaits ?? []), requested: null }, + }; live.set(runId, run); return { run, @@ -553,6 +573,21 @@ export async function awaitChildren( } } +/** runAgent 挂起期间的唤醒条件:等待的子会话全部落定 / 超时 / 本会话被取消 */ +export async function awaitReady(ctx: ServerContext, parentId: string, sus: AwaitSuspension): Promise { + if (Date.now() >= Date.parse(sus.deadline)) return true; + const parent = await ctx.db + .selectFrom("agent_run") + .select("cancel_requested_at") + .where("id", "=", parentId) + .executeTakeFirst(); + if (!parent || parent.cancel_requested_at) return true; + let q = ctx.db.selectFrom("agent_run").select("status").where("parent_run_id", "=", parentId); + if (sus.runIds?.length) q = q.where("id", "in", sus.runIds); + const rows = await q.execute(); + return rows.every((r) => SETTLED.includes(r.status)) ? true : null; +} + // ── 能力处理器 ── export interface SpawnDeps { @@ -636,7 +671,21 @@ export function registerSubagentCapabilities(registry: CapabilityRegistry, ctx: registry.register(capabilitySpec("agent.await"), async (input, c) => { const parent = await parentOf(ctx, c); - return awaitChildren(ctx, parent.id, input.runIds, input.timeoutSec, c.signal); + const live = liveRun(parent.id); + const inv = c.invocationId; + if (!live?.suspend || !inv) return awaitChildren(ctx, parent.id, input.runIds, input.timeoutSec, c.signal); + // 恢复时重放:已挂起等过,直接返回当前状态 + 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(); + return snap; }); } diff --git a/apps/server/src/modules/agent/tasks.ts b/apps/server/src/modules/agent/tasks.ts index 70b09c79..378fdd18 100644 --- a/apps/server/src/modules/agent/tasks.ts +++ b/apps/server/src/modules/agent/tasks.ts @@ -17,12 +17,15 @@ import { import { buildGoalReport } from "./report.ts"; import { type AgentEnv, type DriveResult, drive, goalSettled } from "./runner.ts"; import { finalize } from "./service.ts"; -import { catchUpChildren, claimSubagentSlot } from "./subagents.ts"; +import { awaitReady, catchUpChildren, claimSubagentSlot } from "./subagents.ts"; // 持久任务:runAgent(驱动一次有界 Agent 会话 → 等目标关联的下载/入库落定 → 以业务表汇报) // 与 resolveDecision(decision.created → 自动调查;decision.updated(resolved)→ 执行所选处理)。 const id = z.string().uuid(); +/** agent.await 挂起时的唤醒轮询间隔与单个会话最多挂起次数 */ +const AWAIT_POLL_SEC = 2; +const MAX_AWAIT_SUSPENDS = 100; const DB_RETRY = { maximumAttempts: 3, initialInterval: "1s", backoffCoefficient: 2, maximumInterval: "30s" } as const; export function registerAgentTasks(registry: TaskRegistry, env: () => AgentEnv) { @@ -72,7 +75,30 @@ export function registerAgentTasks(registry: TaskRegistry, env: () => AgentEnv) : "claimed"; if (slot === null) res = { status: "failed", reason: "error", budget: null, message: "等待子会话名额超时", finalText: "" }; - else res = await t.step.run("drive", () => drive(e, e.ctx.capabilities, runId, t.abortSignal)); + else { + res = await t.step.run("drive", () => drive(e, e.ctx.capabilities, runId, t.abortSignal)); + // agent.await 挂起:会话已关闭,waitForFact 等子会话落定 / 超时 / 取消(不占 worker、不占模型), + // 再重开会话 resume(挂起的 agent.await 重放一次拿到结果) + const resumedAwaits: string[] = []; + for (let n = 1; res.status === "suspended" && res.suspension; n++) { + const sus = res.suspension; + if (n > MAX_AWAIT_SUSPENDS) { + res = { status: "failed", reason: "error", budget: null, message: "等待子会话次数过多", finalText: "" }; + break; + } + const waitMs = Math.max(0, Date.parse(sus.deadline) - Date.now()); + await t.step.waitForFact(`subagent-await-${n}`, { + check: () => awaitReady(e.ctx, runId, sus), + timeout: `${AWAIT_POLL_SEC}s`, + maxRounds: Math.max(1, Math.ceil(waitMs / 1000 / AWAIT_POLL_SEC) + 1), + }); + resumedAwaits.push(sus.invocationId); + const resumed = [...resumedAwaits]; + res = await t.step.run(`drive-${n}`, () => + drive(e, e.ctx.capabilities, runId, t.abortSignal, { resumedAwaits: resumed }), + ); + } + } let settled: AgentGoalReport | null = null; const run = await e.ctx.db .selectFrom("agent_run") diff --git a/apps/server/test/e2e/e11-subagents.test.ts b/apps/server/test/e2e/e11-subagents.test.ts index 63b67501..ac2ba032 100644 --- a/apps/server/test/e2e/e11-subagents.test.ts +++ b/apps/server/test/e2e/e11-subagents.test.ts @@ -183,6 +183,12 @@ describe.skipIf(!hasPg)("E11 Pi 子代理(agent.spawn / agent.await)", () => { )!; const lastCompleted = events.filter((x) => x.type === "subagent_completed").at(-1)!; expect(Number(lastCompleted.seq)).toBeLessThan(Number(awaitEnd.seq)); + // agent.await 挂起(不占 worker):会话关闭 → waitForFact 等子会话 → 重开会话重放 agent.await + expect((awaitEnd.payload as Any).recovery).toBe(true); + expect(events.filter((x) => x.type === "run_started").map((x) => (x.payload as Any).resumed)).toEqual([ + false, + true, + ]); // SSE:父会话 runId 上能看到子会话提示 expect(seen.some((x) => x.payload.runId === parentId && x.payload.capabilityId === "agent.spawn")).toBe(true);