feat(agent): dedupe agent.spawn by normalized goal within a parent session (AG-4)
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 <[email protected]>
This commit is contained in:
rootandClaude Opus 5.5 committed
1 parent
0508f3418f
commit
01653b0427
5 files changed
+178
-4
No files matched your search
@@ -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(
|
||||
|
||||
@@ -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<DB>,
|
||||
parentId: string,
|
||||
goal: string,
|
||||
now = Date.now(),
|
||||
): Promise<AgentRunRow | null> {
|
||||
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<v
|
||||
|
||||
// ── 能力处理器 ──
|
||||
|
||||
const spawnLocks = new Map<string, Promise<void>>();
|
||||
/** 同一父会话的派生判定串行(进程内);返回释放函数 */
|
||||
async function spawnLock(parentId: string): Promise<() => void> {
|
||||
const prev = spawnLocks.get(parentId) ?? Promise.resolve();
|
||||
let release!: () => void;
|
||||
const mine = new Promise<void>((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<AgentRunRow> {
|
||||
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<RunScope>;
|
||||
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,
|
||||
|
||||
@@ -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<string, unknown>, 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();
|
||||
|
||||
@@ -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"]),
|
||||
|
||||
@@ -347,7 +347,7 @@ export const CAPABILITY_CATALOG: Record<KnownCapabilityId, CapabilityDescriptor>
|
||||
"agent.spawn": d(
|
||||
"agent.spawn",
|
||||
"派生子会话",
|
||||
"把一个有界子目标交给子会话处理：能力白名单须是本会话白名单的子集，预算从本会话剩余预算里划拨，模式不高于本会话；返回子会话 runId。await=true 时等它结束并返回汇报",
|
||||
"把一个有界子目标交给子会话处理：能力白名单须是本会话白名单的子集，预算从本会话剩余预算里划拨，模式不高于本会话；返回子会话 runId。await=true 时等它结束并返回汇报。同一目标的子会话还在运行或 10 分钟内已完成时不新建，直接返回已有子会话（reused=true）",
|
||||
"propose",
|
||||
"AgentSpawn",
|
||||
"organize.execute",
|
||||
|
||||
Reference in new issue
Block a user