From 0326042c0d8b7eeb67801adde5179c23c2c126e1 Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Fri, 31 Jul 2026 10:22:56 +0800 Subject: [PATCH] improve(agents): scale requester settle batching (#116619) --- ...ent-announce.requester-settle-wake.test.ts | 8 +- ...subagent-announce.requester-settle-wake.ts | 82 ++++++++++++------- 2 files changed, 58 insertions(+), 32 deletions(-) diff --git a/src/agents/subagent-announce.requester-settle-wake.test.ts b/src/agents/subagent-announce.requester-settle-wake.test.ts index 86141dec3ffa..60b8cf61ef61 100644 --- a/src/agents/subagent-announce.requester-settle-wake.test.ts +++ b/src/agents/subagent-announce.requester-settle-wake.test.ts @@ -318,11 +318,17 @@ describe("maybeWakeRequesterAfterAllChildrenSettled", () => { }); it("does not wake while other children still await settle", async () => { + const children = [makeSettledChild({ runId: "run-a" }), makeSettledChild({ runId: "run-b" })]; + registryRuntimeMock.listSubagentRunsForRequester.mockReturnValue(children); registryRuntimeMock.hasDescendantRunAwaitingSettle.mockReturnValue(true); - const woke = await maybeWakeRequesterAfterAllChildrenSettled(wakeParams()); + const woke = await maybeWakeRequesterAfterAllChildrenSettled( + wakeParams({ settledEntry: children[1] }), + ); expect(woke).toBe(false); + expect(registryRuntimeMock.hasDescendantRunAwaitingSettle).toHaveBeenCalledOnce(); + expect(transitionBatchSpy).not.toHaveBeenCalled(); expect(deliverSpy).not.toHaveBeenCalled(); }); diff --git a/src/agents/subagent-announce.requester-settle-wake.ts b/src/agents/subagent-announce.requester-settle-wake.ts index fa0af8b78b3e..2c62565c767a 100644 --- a/src/agents/subagent-announce.requester-settle-wake.ts +++ b/src/agents/subagent-announce.requester-settle-wake.ts @@ -52,11 +52,6 @@ export const testing = { }, }; -type SettledRunSummary = Pick< - SubagentRunRecord, - "runId" | "childSessionKey" | "createdAt" | "endedAt" ->; - export type RequesterSettleWakeBatchState = Omit; const REQUESTER_SETTLE_WAKE_MAX_ATTEMPTS = 3; @@ -64,14 +59,6 @@ const REQUESTER_SETTLE_WAKE_MAX_AMBIGUOUS_REPLAYS = 3; const REQUESTER_SETTLE_WAKE_RETRY_DELAYS_MS = [30_000, 120_000] as const; const activeRequesterSettleWakeBatches = new Set(); -function runIntervalsOverlap(a: SettledRunSummary, b: SettledRunSummary): boolean { - const aEnd = typeof a.endedAt === "number" ? a.endedAt : Number.MAX_SAFE_INTEGER; - const bEnd = typeof b.endedAt === "number" ? b.endedAt : Number.MAX_SAFE_INTEGER; - // Fan-out membership begins at spawn, not execution admission. A queued - // sibling can start only after another child ends and still belong to it. - return a.createdAt <= bEnd && b.createdAt <= aEnd; -} - function buildRequesterSettleWakeMessage(params: { findings?: string }): string { return [ "[Subagent Context] Every subagent spawned from this session has now settled — none are still running or awaiting completion delivery.", @@ -88,27 +75,56 @@ function buildConnectedSettledWave( candidates: readonly SubagentRunRecord[], settledEntry: SubagentRunRecord, ): SubagentRunRecord[] { - const unclaimed = new Set(candidates); - const batch: SubagentRunRecord[] = []; - const frontier: SettledRunSummary[] = [settledEntry]; - for (const entry of unclaimed) { - if (entry.runId === settledEntry.runId) { - unclaimed.delete(entry); - batch.push(entry); - frontier.push(entry); - break; - } + const targetIndex = candidates.findIndex((entry) => entry.runId === settledEntry.runId); + const target = candidates[targetIndex]; + if (!target) { + return []; } - for (let pivot = frontier.pop(); pivot; pivot = frontier.pop()) { - for (const entry of unclaimed) { - if (runIntervalsOverlap(entry, pivot)) { - unclaimed.delete(entry); - batch.push(entry); - frontier.push(entry); + + const sorted = candidates + .map((entry, originalIndex) => ({ + entry, + originalIndex, + endedAt: typeof entry.endedAt === "number" ? entry.endedAt : Number.MAX_SAFE_INTEGER, + })) + .toSorted( + (a, b) => + a.entry.createdAt - b.entry.createdAt || + a.endedAt - b.endedAt || + a.originalIndex - b.originalIndex, + ); + const first = sorted[0]; + if (!first) { + return []; + } + + let componentStart = 0; + let componentEnd = first.endedAt; + let containsTarget = first.originalIndex === targetIndex; + for (let index = 1; index <= sorted.length; index += 1) { + const next = sorted[index]; + // Interval-graph components are contiguous after sorting by spawn time. + // Spawn time, rather than execution admission, keeps capacity-queued siblings together. + if (!next || next.entry.createdAt > componentEnd) { + if (containsTarget) { + const component = sorted + .slice(componentStart, index) + .filter((item) => item.originalIndex !== targetIndex) + .toSorted((a, b) => a.originalIndex - b.originalIndex); + return [target, ...component.map((item) => item.entry)]; } + if (!next) { + break; + } + componentStart = index; + componentEnd = next.endedAt; + containsTarget = next.originalIndex === targetIndex; + continue; } + componentEnd = Math.max(componentEnd, next.endedAt); + containsTarget ||= next.originalIndex === targetIndex; } - return batch; + return []; } function readSharedBatchState(batch: readonly SubagentRunRecord[]): RequesterSettleWakeBatchState { @@ -219,6 +235,10 @@ export async function maybeWakeRequesterAfterAllChildrenSettled(params: { const frozenBatchRunIds = currentSettledEntry.requesterSettleWake.batchRunIds; const currentRearmGeneration = currentSettledEntry.requesterSettleWake.rearmGeneration; + const hasUnsettledDescendants = requesterHasUnsettledDescendants(); + if ((!frozenBatchRunIds || frozenBatchRunIds.length === 0) && hasUnsettledDescendants) { + return false; + } let settledBatch: SubagentRunRecord[]; if (frozenBatchRunIds && frozenBatchRunIds.length > 0) { const runsById = new Map(requesterRuns.map((entry) => [entry.runId, entry])); @@ -241,7 +261,7 @@ export async function maybeWakeRequesterAfterAllChildrenSettled(params: { const batchRunIds = settledBatch.map((entry) => entry.runId).toSorted(); const selectedState = readSharedBatchState(settledBatch); - if (requesterHasUnsettledDescendants()) { + if (hasUnsettledDescendants) { if (frozenBatchRunIds && frozenBatchRunIds.length > 0) { deferRequesterSettleWakeBatch({ batchRunIds,