feat(agent): agent.spawn{await:true} suspends too; children wake the parent via fact
agent.await and agent.spawn{await:true} now return immediately when
children are still running: the session closes, runAgent registers a
wake signal in agent_run.await_wake and waits with waitForFact without
holding an io/agent worker slot. A child that settles (terminal or
waiting) signals the parent's registered fact, so the parent resumes at
once instead of polling; the suspended call replays and the child
report becomes the tool result. agent.spawn is registered replay-safe;
a suspended spawn is recorded as failed in the ledger so the replay
re-runs it against the same child.
Migration 0037 also adds the admin_recovery_code, admin_passkey and
webauthn_challenge tables used by the account work.
E11 asserts the parent is suspended with no run held by any worker
while it waits, and covers spawn{await:true} end to end.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
This commit is contained in:
rootandClaude Opus 5.5 committed
1 parent
f9b30b7e72
commit
5c6f483420
8 files changed
+382
-28
No files matched your search
@@ -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<null> {
|
||||
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<boolean> {
|
||||
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<AgentRun
|
||||
}
|
||||
|
||||
export function registerSubagentCapabilities(registry: CapabilityRegistry, ctx: ServerContext, deps: SpawnDeps) {
|
||||
registry.register(capabilitySpec("agent.spawn"), async (input, c) => {
|
||||
// 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<RunScope>;
|
||||
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;
|
||||
});
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user