From 01653b042733ef2dfcc9373960919f0a9d7b0887 Mon Sep 17 00:00:00 2001 From: root Date: Wed, 30 Sep 2026 17:58:08 +0800 Subject: [PATCH] feat(agent): dedupe agent.spawn by normalized goal within a parent session (AG-4) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A spawn whose goal (NFKC, case, whitespace and punctuation folded) matches a child of the same parent that is still queued/running, or succeeded/waiting within the last 10 minutes, returns that child instead of creating a new one: output gets reused=true, a message ("已有同目标子会话…") and the child's current result summary. Failed/cancelled/over-budget children are not reused. Spawn decisions are serialized per parent in-process so parallel duplicate calls in one batch create a single child. Replays keyed by idempotency still hit the original child as before. The subagent constraint in the system prompt asks for Chinese sub-goals and no duplicate spawns. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/modules/agent/runner.ts | 2 +- apps/server/src/modules/agent/subagents.ts | 89 +++++++++++++++++++++- apps/server/test/e2e/e11-subagents.test.ts | 82 ++++++++++++++++++++ packages/contracts/src/capabilities-io.ts | 7 ++ packages/contracts/src/capability.ts | 2 +- 5 files changed, 178 insertions(+), 4 deletions(-) diff --git a/apps/server/src/modules/agent/runner.ts b/apps/server/src/modules/agent/runner.ts index 85aed158..4c4fcdb8 100644 --- a/apps/server/src/modules/agent/runner.ts +++ b/apps/server/src/modules/agent/runner.ts @@ -271,7 +271,7 @@ function systemPromptFor( constraints.push( `可以用 agent.spawn 把彼此独立的有界子目标交给子会话（同时最多 ${sub.maxConcurrent} 个，派生层数上限 ${sub.maxDepth}）；` + "子会话的能力、模式与预算都不会超过本会话，且会占用本会话预算；budget 的每一项都必须小于本会话剩余预算（超过会被拒绝，不给则按剩余的一半划拨）；" + - "派生后用 agent.await 等结果再汇总，不要为小事派生。", + "派生后用 agent.await 等结果再汇总，不要为小事派生；子目标用中文写；同一目标已派出过的子会话不要重复派生（会直接返回已有子会话）。", ); if (scope.capabilities.includes("web.search")) constraints.push( diff --git a/apps/server/src/modules/agent/subagents.ts b/apps/server/src/modules/agent/subagents.ts index 9ae30e59..44b2ab2d 100644 --- a/apps/server/src/modules/agent/subagents.ts +++ b/apps/server/src/modules/agent/subagents.ts @@ -49,6 +49,44 @@ const ALL_KINDS: CapabilityKind[] = ["read", "preview", "propose", "approve", "e const PLAN_ONLY_KINDS: CapabilityKind[] = ["read", "preview", "propose"]; /** 一个会话最多派生的子会话数(预算之外的兜底) */ export const MAX_CHILDREN_PER_RUN = 10; +/** 同目标去重窗口(AG-4):已完成的子会话在这段时间内按同目标复用 */ +export const SPAWN_DEDUPE_WINDOW_MS = 10 * 60_000; + +/** 目标归一化:全半角统一、忽略大小写、空白与标点("查一下 观众 站点。" ≡ "查一下观众站点") */ +export function normalizeGoal(goal: string): string { + return goal + .normalize("NFKC") + .toLowerCase() + .replace(/[\s\p{P}\p{S}]+/gu, ""); +} + +/** + * 同一父会话里同目标、可复用的子会话:还没落定(排队 / 运行中),或 10 分钟内成功完成 / 已在等下载入库。 + * 失败、取消、超预算的不复用(父会话可以换个预算或说法重派)。 + */ +export async function findSameGoalChild( + db: Kysely, + parentId: string, + goal: string, + now = Date.now(), +): Promise { + const key = normalizeGoal(goal); + if (!key) return null; + const rows = await db + .selectFrom("agent_run") + .selectAll() + .where("parent_run_id", "=", parentId) + .orderBy("created_at", "desc") + .execute(); + for (const r of rows) { + if (normalizeGoal(r.goal) !== key) continue; + if (!SETTLED.includes(r.status)) return r; + if (r.status !== "succeeded" && r.status !== "waiting") continue; + const at = (r.finished_at ?? r.updated_at)?.getTime() ?? 0; + if (now - at <= SPAWN_DEDUPE_WINDOW_MS) return r; + } + return null; +} const MIRRORED_TYPES = [ "run_started", "tool_call_start", @@ -690,6 +728,23 @@ export async function suspendRun(live: LiveRun, sus: AwaitSuspension): Promise>(); +/** 同一父会话的派生判定串行(进程内);返回释放函数 */ +async function spawnLock(parentId: string): Promise<() => void> { + const prev = spawnLocks.get(parentId) ?? Promise.resolve(); + let release!: () => void; + const mine = new Promise((r) => { + release = r; + }); + const tail = prev.then(() => mine); + spawnLocks.set(parentId, tail); + await prev; + return () => { + release(); + if (spawnLocks.get(parentId) === tail) spawnLocks.delete(parentId); + }; +} + export interface SpawnDeps { /** 建子 agent_run + 任务行并入队 runAgent(service.startRun) */ start(p: { @@ -728,8 +783,19 @@ export function registerSubagentCapabilities(registry: CapabilityRegistry, ctx: .selectAll() .where("dedupe_key", "=", dedupeKey) .executeTakeFirst(); + // AG-4 去重:不是重放(幂等键没命中)时,同目标的子会话还在跑或刚完成就直接复用。 + // 同一父会话的派生串行(同一批并行的两次同目标调用也只建一个);父会话只在一个进程里运行 + const release = await spawnLock(parent.id); let child = existing; - if (!child) { + let reused: AgentRunRow | null = null; + try { + reused = !child ? await findSameGoalChild(ctx.db, parent.id, input.goal) : null; + if (reused) child = reused; + if (!child) child = await create(); + } finally { + release(); + } + async function create(): Promise { if (parent.depth + 1 > cfg.maxDepth) throw new CapabilityError("precondition", `派生层数已达上限 ${cfg.maxDepth}，子会话不能再派生`); const mode: AgentRunMode = input.mode ?? parent.mode; @@ -753,7 +819,7 @@ export function registerSubagentCapabilities(registry: CapabilityRegistry, ctx: reason: input.reason ?? null, dedupeKey, }); - child = r.run; + return r.run; } await liveRun(parent.id)?.pump(); let result: SubagentResult | null = null; @@ -774,6 +840,25 @@ export function registerSubagentCapabilities(registry: CapabilityRegistry, ctx: } } const cs = (child.scope ?? {}) as Partial; + if (reused) { + // 没要求等待时也带上已有子会话的当前结果(还在跑的 summary 为 null、pending = true) + if (!result) { + const row = await ctx.db.selectFrom("agent_run").selectAll().where("id", "=", child.id).executeTakeFirst(); + result = resultOf(row ?? child); + } + return { + reused: true, + message: result.pending + ? "已有同目标子会话，仍在运行，未新建；需要结果时用 agent.await 等它" + : "已有同目标子会话（10 分钟内已完成），未新建；结果见 result", + runId: child.id, + depth: child.depth, + mode: child.mode, + capabilities: cs.capabilities ?? [], + budget: (child.budget ?? {}) as SubagentBudget, + result, + }; + } return { runId: child.id, depth: child.depth, diff --git a/apps/server/test/e2e/e11-subagents.test.ts b/apps/server/test/e2e/e11-subagents.test.ts index ac69ec35..38719d84 100644 --- a/apps/server/test/e2e/e11-subagents.test.ts +++ b/apps/server/test/e2e/e11-subagents.test.ts @@ -358,6 +358,88 @@ describe.skipIf(!hasPg)("E11 Pi 子代理(agent.spawn / agent.await)", () => { 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(); diff --git a/packages/contracts/src/capabilities-io.ts b/packages/contracts/src/capabilities-io.ts index fc53576f..c592eca9 100644 --- a/packages/contracts/src/capabilities-io.ts +++ b/packages/contracts/src/capabilities-io.ts @@ -390,6 +390,13 @@ export const SubagentResult = z.object({ pending: z.boolean(), }); export const AgentSpawnOutput = z.object({ + /** + * AG-4 去重:同一父会话里同一目标(归一化后)的子会话还在运行或 10 分钟内已完成时不新建, + * 返回已有子会话(reused = true,result 带它的当前结果摘要);界面的"派出 N 个子会话"不计这次调用 + */ + reused: z.boolean().optional(), + /** 给模型的一句说明(reused 时:"已有同目标子会话…") */ + message: z.string().optional(), runId: Id, depth: NonNegInt, mode: z.enum(["execute", "plan_only"]), diff --git a/packages/contracts/src/capability.ts b/packages/contracts/src/capability.ts index 2b66d4e1..94b5ec9d 100644 --- a/packages/contracts/src/capability.ts +++ b/packages/contracts/src/capability.ts @@ -347,7 +347,7 @@ export const CAPABILITY_CATALOG: Record "agent.spawn": d( "agent.spawn", "派生子会话", - "把一个有界子目标交给子会话处理：能力白名单须是本会话白名单的子集，预算从本会话剩余预算里划拨，模式不高于本会话；返回子会话 runId。await=true 时等它结束并返回汇报", + "把一个有界子目标交给子会话处理：能力白名单须是本会话白名单的子集，预算从本会话剩余预算里划拨，模式不高于本会话；返回子会话 runId。await=true 时等它结束并返回汇报。同一目标的子会话还在运行或 10 分钟内已完成时不新建，直接返回已有子会话（reused=true）", "propose", "AgentSpawn", "organize.execute",