improve(agents): scale requester settle batching (#116619)

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();
   });
 

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<RequesterSettleWakeState, "retireAfterSettle">;
 
 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<string>();
 
-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,