feat(agent): agent.await suspends the session instead of holding a worker slot

When children are still running, agent.await records the wait, closes
the session (the tool call stays pending in the JSONL) and drive returns
suspended. runAgent waits with waitForFact for the children to settle,
the deadline, or cancellation, then reopens the session; agent.await is
replay-safe, so resume replays it and the model gets the real result.

Co-Authored-By: Claude Opus 5.5 <[email protected]>

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<DriveResult> {
   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<RunOutcome> {
+    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;
   }
 }
 

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<string>; requested: AwaitSuspension | null };
   /** 把子会话的新事件 / 新派生 / 结束镜像进本会话事件流(串行) */
   pump(): Promise<void>;
   /** 子孙会话已用量(同步读缓存,pump 刷新) */
@@ -80,6 +94,7 @@ export async function attachLiveRun(
   env: AgentEnv,
   runId: string,
   handle: AgentRunHandle,
+  opts: { resumedAwaits?: readonly string[] } = {},
 ): Promise<{ run: LiveRun; stop(): Promise<void> }> {
   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<true | null> {
+  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;
   });
 }
 

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<true>(`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")