import { type FakeScript, type FakeToolResult, isCapabilityError } from "@umt/agent"; import { afterAll, beforeAll, describe, expect, it } from "vitest"; import { createE2E, type E2E, pgAvailable, runEvents, runRow, waitFor } from "./harness.ts"; // E11 Pi 子代理(agent.spawn / agent.await):父会话"检查两部作品缺什么版本"派生两个只读子会话 → 等待 → 汇总。 // - maxConcurrent=1:两个子会话串行(运行区间不重叠); // - 子会话超预算(划拨 1 次工具调用却调了 2 次)→ budget_exceeded,不影响父会话; // - 父会话合计用量 = 自身 + 子会话;子会话事件按父 seq 镜像进父事件流,SSE 可见; // - 白名单超父 / 只给方案派生执行 / 深度上限 / 开关关闭 → 拒绝; // - 父会话中止 → 子会话 cancelled。 const hasPg = await pgAvailable(); // biome-ignore lint/suspicious/noExplicitAny: 测试断言直接读 JSON type Any = any; const PARENT_GOAL = "检查两部作品缺什么版本"; const ABORT_GOAL = "中止测试:派生后等待"; const DUNE = "子目标:检查《沙丘2》缺什么版本"; const OPPIE = "子目标:检查《奥本海默》缺什么版本"; const READONLY = ["inventory.query", "catalog.lookup"]; const lastOk = (results: FakeToolResult[], cap: string): Any => [...results].reverse().find((r) => r.capabilityId === cap && !r.isError)?.json; const parentScript: FakeScript = { name: "e11-parent", match: PARENT_GOAL, steps: [ { text: "两部作品互不相关,各派一个只读子会话。", toolCalls: [ { capability: "agent.spawn", input: { goal: DUNE, capabilityAllowlist: READONLY, mode: "plan_only", reason: "逐部检查" }, }, { capability: "agent.spawn", input: { goal: OPPIE, capabilityAllowlist: READONLY, mode: "plan_only", reason: "逐部检查(小预算)", budget: { maxToolCalls: 1 }, }, }, ], }, { toolCalls: [{ capability: "agent.await", input: { timeoutSec: 120 } }] }, ], finalText: (results) => { const r = lastOk(results, "agent.await"); const lines = (r?.results ?? []).map((x: Any) => `${x.goal} → ${x.status}`); return `汇总:${lines.join(";")}`; }, }; const duneScript: FakeScript = { name: "e11-dune", match: DUNE, steps: [{ toolCalls: [{ capability: "inventory.query", input: { query: "沙丘2", limit: 5 } }] }], finalText: "《沙丘2》:库里没有 1080p 版本", }; // 只划拨了 1 次工具调用:第二次调用超预算 const oppieScript: FakeScript = { name: "e11-oppie", match: OPPIE, steps: [ { toolCalls: [{ capability: "inventory.query", input: { query: "奥本海默", limit: 5 } }] }, { toolCalls: [{ capability: "catalog.lookup", input: { query: "奥本海默" } }] }, ], finalText: "不应走到这里", }; const abortParentScript: FakeScript = { name: "e11-abort", match: ABORT_GOAL, steps: [ { toolCalls: [ { capability: "agent.spawn", input: { goal: "子目标:慢任务甲", capabilityAllowlist: READONLY } }, { capability: "agent.spawn", input: { goal: "子目标:慢任务乙", capabilityAllowlist: READONLY } }, ], }, { toolCalls: [{ capability: "agent.await", input: { timeoutSec: 600 } }] }, ], 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: /^子目标:慢任务/, steps: [{ toolCalls: [{ capability: "inventory.query", input: { limit: 1 } }] }], finalText: "done", }; describe.skipIf(!hasPg)("E11 Pi 子代理(agent.spawn / agent.await)", () => { let e: E2E; beforeAll(async () => { 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); afterAll(async () => { 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 () => { const r = await runRow(e, runId); return statuses.includes(r.status) ? r : null; }, ms, `run ${runId} → ${statuses.join("/")}`, ); it("派生两个只读子会话并等待 → 汇总;串行、子超预算不影响父、预算合计、子事件进父事件流", async () => { const seen: Any[] = []; const off = e.bus.subscribe((ev) => { if (ev.type === "agent.tool_call") seen.push(ev); }); const r = await e.api("POST", "/agent/runs", { goal: PARENT_GOAL, mode: "plan_only", capabilityAllowlist: [...READONLY, "agent.spawn", "agent.await"], }); expect(r.status).toBe(201); const parentId = r.body.data.id as string; const parent = await settle(parentId, ["succeeded", "failed", "cancelled", "budget_exceeded"]); off(); expect(parent.status).toBe("succeeded"); const kids = await e.db .selectFrom("agent_run") .selectAll() .where("parent_run_id", "=", parentId) .orderBy("created_at") .execute(); expect(kids.map((k) => k.goal)).toEqual([DUNE, OPPIE]); for (const k of kids) { expect(k).toMatchObject({ trigger: "subagent", depth: 1, mode: "plan_only", created_by: parent.created_by }); expect((k.scope as Any).capabilities).toEqual(READONLY); } const [dune, oppie] = kids as [(typeof kids)[0], (typeof kids)[0]]; expect(dune.status).toBe("succeeded"); expect(dune.spawn_reason).toBe("逐部检查"); // 子超预算:budget_exceeded,不影响父会话 expect(oppie.status).toBe("budget_exceeded"); expect((oppie.budget as Any).maxToolCalls).toBe(1); // 缺省划拨:父剩余预算的一半(设置 maxToolCalls 缺省 120,派生时父已用 1 次) expect((dune.budget as Any).maxToolCalls).toBeLessThanOrEqual(60); // maxConcurrent=1:串行(谁先占到名额不定,但两个运行区间不重叠) const [first, second] = [dune, oppie].sort((a, b) => a.started_at!.getTime() - b.started_at!.getTime()) as [ typeof dune, typeof dune, ]; expect(first.finished_at).toBeTruthy(); expect(second.started_at!.getTime()).toBeGreaterThanOrEqual(first.finished_at!.getTime()); // 父会话汇总来自 agent.await 的结果 const events = await runEvents(e, parentId); const done = events.find((x) => x.type === "run_completed"); expect((done!.payload as Any).finalText).toContain(`${DUNE} → succeeded`); expect((done!.payload as Any).finalText).toContain(`${OPPIE} → budget_exceeded`); // 子会话事件进父事件流(父 seq 单调、不冲突) const types = events.map((x) => x.type); expect(types.filter((t) => t === "subagent_spawned")).toHaveLength(2); expect(types.filter((t) => t === "subagent_completed")).toHaveLength(2); const mirrored = events.filter((x) => x.type === "subagent_event"); expect( mirrored.some( (x) => (x.payload as Any).childRunId === dune.id && (x.payload as Any).capabilityId === "inventory.query", ), ).toBe(true); expect(events.map((x) => Number(x.seq))).toEqual(events.map((_, i) => i)); // 子会话结束记录先于 agent.await 的工具结果 const awaitEnd = events.find( (x) => x.type === "tool_call_end" && (x.payload as Any).capabilityId === "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); // 预算合计:父会话合计 = 自身 + 子会话 const detail = await e.api("GET", `/agent/runs/${parentId}?afterSeq=-1&limit=500`); const d = detail.body.data; const ownTokens = (parent.usage as Any).totalTokens as number; const kidTokens = kids.reduce((a, k) => a + ((k.usage as Any).totalTokens ?? 0), 0); expect(kidTokens).toBeGreaterThan(0); expect(d.familyUsage).toMatchObject({ runs: 3, toolCalls: parent.tool_call_count + dune.tool_call_count + oppie.tool_call_count, totalTokens: ownTokens + kidTokens, }); expect(parent.tool_call_count).toBe(3); expect(d.children.map((c: Any) => c.id)).toEqual([dune.id, oppie.id]); expect(parent.summary).toContain("子会话 2 个"); // 子会话详情:父会话链接与继承信息 const child = await e.api("GET", `/agent/runs/${dune.id}`); expect(child.body.data.parent).toMatchObject({ id: parentId, goal: PARENT_GOAL }); expect(child.body.data.run).toMatchObject({ parentRunId: parentId, depth: 1, spawnReason: "逐部检查" }); // 列表可按父会话过滤 const list = await e.api("GET", `/agent/runs?parentRunId=${parentId}`); expect(list.body.data.items).toHaveLength(2); }, 180_000); it("拒绝:白名单超父、只给方案派生执行、深度上限、开关关闭", async () => { const parent = await e.db .insertInto("agent_run") .values({ trigger: "user", goal: "fixture parent", status: "running", mode: "plan_only", scope: JSON.stringify({ capabilities: [...READONLY, "agent.spawn", "agent.await"], allowedKinds: ["read", "preview", "propose"], }), budget: JSON.stringify({}), created_by: "user:admin", }) .returningAll() .executeTakeFirstOrThrow(); const child = await e.db .insertInto("agent_run") .values({ trigger: "subagent", goal: "fixture child", status: "running", mode: "plan_only", scope: JSON.stringify({ capabilities: [...READONLY, "agent.spawn"], allowedKinds: ["read", "preview", "propose"], }), budget: JSON.stringify({}), created_by: "user:admin", parent_run_id: parent.id, depth: 1, }) .returningAll() .executeTakeFirstOrThrow(); const spawn = async (runId: string, input: Record) => { try { await e.booted.ctx.capabilities.invoke("agent.spawn", input, { actorId: "user:admin", taskId: runId, scopeIds: [], signal: new AbortController().signal, source: "agent", agentRunId: runId, }); return null; } catch (err) { if (!isCapabilityError(err)) throw err; return { code: err.code, message: err.message }; } }; expect(await spawn(parent.id, { goal: "越权", capabilityAllowlist: ["release.search"] })).toMatchObject({ code: "unauthorized", message: expect.stringContaining("超出本会话白名单"), }); expect(await spawn(parent.id, { goal: "提权", mode: "execute" })).toMatchObject({ code: "unauthorized", message: expect.stringContaining("只给方案"), }); expect(await spawn(child.id, { goal: "再派生" })).toMatchObject({ code: "precondition", message: expect.stringContaining("派生层数已达上限"), }); await e.settings({ pi: { subagents: { enabled: false } } }); expect(await spawn(parent.id, { goal: "关闭后" })).toMatchObject({ code: "precondition", message: expect.stringContaining("子会话未开启"), }); await e.settings({ pi: { subagents: { enabled: true } } }); // 以上都没有建出子会话 const n = await e.db.selectFrom("agent_run").select("id").where("parent_run_id", "=", parent.id).execute(); expect(n.map((x) => x.id)).toEqual([child.id]); await e.db.updateTable("agent_run").set({ status: "cancelled" }).where("id", "in", [parent.id, child.id]).execute(); }, 60_000); it("同目标去重(AG-4):子会话还在运行或 10 分钟内已完成时返回已有子会话,不新建", async () => { const parent = await e.db .insertInto("agent_run") .values({ trigger: "user", goal: "fixture dedupe parent", status: "running", mode: "plan_only", scope: JSON.stringify({ capabilities: [...READONLY, "agent.spawn", "agent.await"], allowedKinds: ["read", "preview", "propose"], }), budget: JSON.stringify({}), created_by: "user:admin", }) .returningAll() .executeTakeFirstOrThrow(); const child = await e.db .insertInto("agent_run") .values({ trigger: "subagent", goal: "子目标:查《沙丘》 缺什么版本", status: "running", mode: "plan_only", scope: JSON.stringify({ capabilities: READONLY, allowedKinds: ["read", "preview", "propose"] }), budget: JSON.stringify({ maxToolCalls: 5 }), created_by: "user:admin", parent_run_id: parent.id, depth: 1, started_at: new Date(), }) .returningAll() .executeTakeFirstOrThrow(); const spawn = async (input: Record, key: string) => (await e.booted.ctx.capabilities.invoke("agent.spawn", input, { actorId: "user:admin", taskId: parent.id, scopeIds: [], signal: new AbortController().signal, source: "agent", agentRunId: parent.id, invocationId: key, })) as Any; const children = async () => (await e.db.selectFrom("agent_run").select("id").where("parent_run_id", "=", parent.id).execute()).map( (x) => x.id, ); // 仍在运行:归一化后同目标(全半角标点、空白不同)→ 复用,结果摘要 pending const a = await spawn({ goal: "子目标:查《沙丘》缺什么版本。", capabilityAllowlist: READONLY }, "dedupe-1"); expect(a).toMatchObject({ reused: true, runId: child.id, result: { pending: true, summary: null } }); expect(a.message).toContain("已有同目标子会话"); expect(Object.keys(a)[0]).toBe("reused"); // 5 分钟前已完成:复用并带回汇报摘要 await e.db .updateTable("agent_run") .set({ status: "succeeded", summary: "《沙丘》缺 4K 版本", finished_at: new Date(Date.now() - 5 * 60_000) }) .where("id", "=", child.id) .execute(); const b = await spawn({ goal: "子目标:查《沙丘》缺什么版本" }, "dedupe-2"); expect(b).toMatchObject({ reused: true, runId: child.id, result: { status: "succeeded", pending: false, summary: expect.stringContaining("缺 4K") }, }); expect(b.message).toContain("10 分钟内已完成"); // 结果行的子会话计数(会话树)不变 expect(await children()).toEqual([child.id]); // 超过 10 分钟 / 失败的不复用;不同目标不复用 const { findSameGoalChild, normalizeGoal } = await import("../../src/modules/agent/subagents.ts"); expect(normalizeGoal("子目标:查《沙丘》 缺什么版本。")).toBe(normalizeGoal("子目标:查《沙丘》缺什么版本")); const later = Date.now() + 6 * 60_000; expect(await findSameGoalChild(e.db, parent.id, "子目标:查《沙丘》缺什么版本", later)).toBeNull(); expect((await findSameGoalChild(e.db, parent.id, "子目标:查《沙丘》缺什么版本"))?.id).toBe(child.id); expect(await findSameGoalChild(e.db, parent.id, "子目标:查《奥本海默》缺什么版本")).toBeNull(); await e.db.updateTable("agent_run").set({ status: "failed" }).where("id", "=", child.id).execute(); expect(await findSameGoalChild(e.db, parent.id, "子目标:查《沙丘》缺什么版本")).toBeNull(); await e.db.updateTable("agent_run").set({ status: "cancelled" }).where("id", "=", parent.id).execute(); }, 60_000); it("父会话中止 → 全部子会话 cancelled(排队等名额的也一样)", async () => { // 先停 worker 建父会话,再塞一个占着唯一名额的子会话,保证真实子会话排队等名额 await e.worker?.stop(); const r = await e.api("POST", "/agent/runs", { goal: ABORT_GOAL, mode: "plan_only", capabilityAllowlist: [...READONLY, "agent.spawn", "agent.await"], }); 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 kids = await waitFor( async () => { const rows = await e.db .selectFrom("agent_run") .selectAll() .where("parent_run_id", "=", parentId) .where("id", "<>", blocker.id) .execute(); const p = await runEvents(e, parentId); const awaiting = p.some( (x) => x.type === "tool_call_start" && (x.payload as Any).capabilityId === "agent.await", ); return rows.length === 2 && awaiting ? rows : null; }, 60_000, "父会话派生两个子会话并开始等待", ); 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); const parent = await settle(parentId, ["cancelled", "succeeded", "failed"]); expect(parent.status).toBe("cancelled"); for (const k of kids) { const c = await settle(k.id, ["cancelled", "succeeded", "failed"]); expect(c.status).toBe("cancelled"); expect(c.cancel_requested_at).not.toBeNull(); } // 占名额的行也收到取消标志 const b = await runRow(e, blocker.id); 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); });