Skip to content

Commit a893400

Browse files
kklouzalkklouzal
committed
fix(subagents): credit requester-consumed descendant completions
Keep requester-consumed descendant completions credited without duplicate announcements, and model bundled plugin Knip roots needed by the CI dead-code scan.
1 parent 047232b commit a893400

29 files changed

Lines changed: 901 additions & 99 deletions

‎scripts/audit-seams.mjs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -620,7 +620,7 @@ function describeCronSeamKinds(relativePath, source) {
620620

621621
if (
622622
importsFollowup &&
623-
/\bwaitForDescendantSubagentSummary\b|\breadDescendantSubagentFallbackReply\b|\bexpectsSubagentFollowup\b|\bcallGateway\b|\blistDescendantRunsForRequester\b/.test(
623+
/\bwaitForDescendantSubagentSummary\b|\breadDescendantSubagentFallbackReplyWithRuns\b|\bexpectsSubagentFollowup\b|\bcallGateway\b|\blistDescendantRunsForRequester\b/.test(
624624
source,
625625
)
626626
) {

‎src/agents/subagent-announce-output.test.ts‎

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,7 @@ import {
55
testing,
66
applySubagentWaitOutcome,
77
buildCompactAnnounceStatsLine,
8-
buildChildCompletionFindings,
8+
buildChildCompletionFindingsWithRuns,
99
dedupeLatestChildCompletionRows,
1010
readSubagentOutput,
1111
} from "./subagent-announce-output.test-support.js";
@@ -262,7 +262,13 @@ describe("readSubagentOutput", () => {
262262
});
263263
});
264264

265-
describe("buildChildCompletionFindings", () => {
265+
function buildChildCompletionFindings(
266+
children: Parameters<typeof buildChildCompletionFindingsWithRuns>[0],
267+
): string | undefined {
268+
return buildChildCompletionFindingsWithRuns(children)?.text;
269+
}
270+
271+
describe("buildChildCompletionFindingsWithRuns", () => {
266272
it("does not convert ANNOUNCE_SKIP child completions into no-output findings", () => {
267273
const findings = buildChildCompletionFindings([
268274
{

‎src/agents/subagent-announce-output.ts‎

Lines changed: 21 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -391,6 +391,12 @@ function hasCapturedChildCompletionReply(child: ChildCompletionRow): boolean {
391391
export function buildChildCompletionFindings(
392392
children: Array<ChildCompletionRow>,
393393
): string | undefined {
394+
return buildChildCompletionFindingsWithRuns(children)?.text;
395+
}
396+
397+
export function buildChildCompletionFindingsWithRuns(
398+
children: (ChildCompletionRow & { runId?: string })[],
399+
): { text: string; consumedRunIds: string[] } | undefined {
394400
const sorted = [...children].toSorted((a, b) => {
395401
if (a.createdAt !== b.createdAt) {
396402
return a.createdAt - b.createdAt;
@@ -401,6 +407,7 @@ export function buildChildCompletionFindings(
401407
});
402408

403409
const sections: string[] = [];
410+
const consumedRunIds: string[] = [];
404411
for (const [index, child] of sorted.entries()) {
405412
const resultText = selectChildCompletionResultText(child);
406413
const outcome = describeSubagentOutcome(child.outcome);
@@ -418,39 +425,27 @@ export function buildChildCompletionFindings(
418425
"\n",
419426
),
420427
);
428+
if (typeof child.runId === "string" && child.runId.trim()) {
429+
consumedRunIds.push(child.runId);
430+
}
421431
}
422432

423433
if (sections.length === 0) {
424434
return undefined;
425435
}
426436

427-
return ["Child completion results:", "", ...sections].join("\n\n");
437+
return { text: ["Child completion results:", "", ...sections].join("\n\n"), consumedRunIds };
428438
}
429439

430-
export function dedupeLatestChildCompletionRows(
431-
children: Array<{
432-
runId: string;
433-
childSessionKey: string;
434-
task: string;
435-
label?: string;
436-
generation?: number;
437-
createdAt: number;
438-
endedAt?: number;
439-
frozenResultText?: string | null;
440-
completion?: {
441-
resultText?: string | null;
442-
fallbackResultText?: string | null;
443-
};
444-
delivery?: {
445-
payload?: {
446-
frozenResultText?: string | null;
447-
fallbackFrozenResultText?: string | null;
448-
};
449-
};
450-
outcome?: SubagentRunOutcome;
451-
}>,
452-
) {
453-
const latestByChildSessionKey = new Map<string, (typeof children)[number]>();
440+
type ChildCompletionRunRow = ChildCompletionRow & {
441+
runId: string;
442+
generation?: number;
443+
};
444+
445+
export function dedupeLatestChildCompletionRows<T extends ChildCompletionRunRow>(
446+
children: T[],
447+
): T[] {
448+
const latestByChildSessionKey = new Map<string, T>();
454449
for (const child of children) {
455450
const existing = latestByChildSessionKey.get(child.childSessionKey);
456451
if (!existing || compareSubagentRunGeneration(child, existing) > 0) {
@@ -469,6 +464,7 @@ export function filterCurrentDirectChildCompletionRows(
469464
label?: string;
470465
createdAt: number;
471466
endedAt?: number;
467+
generation?: number;
472468
frozenResultText?: string | null;
473469
completion?: {
474470
resultText?: string | null;

‎src/agents/subagent-announce.format.e2e.test.ts‎

Lines changed: 10 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -2764,12 +2764,16 @@ describe("subagent announce formatting", () => {
27642764
expect(message).toContain("All pending descendants for that run have now settled");
27652765
expect(message).toContain("result from child a");
27662766
expect(message).toContain("result from child b");
2767-
expect(subagentRegistryMock.replaceSubagentRunAfterSteer).toHaveBeenCalledWith({
2768-
previousRunId: "run-parent-phase-1",
2769-
nextRunId: "run-parent-phase-2",
2770-
preserveFrozenResultFallback: true,
2771-
task: expect.stringContaining("All pending descendants for that run have now settled"),
2772-
});
2767+
expect(subagentRegistryMock.replaceSubagentRunAfterSteer).toHaveBeenCalledWith(
2768+
expect.objectContaining({
2769+
previousRunId: "run-parent-phase-1",
2770+
nextRunId: "run-parent-phase-2",
2771+
preserveFrozenResultFallback: true,
2772+
pendingRequesterConsumedDescendantRunIds: ["run-child-a", "run-child-b"],
2773+
pendingRequesterConsumedRunStartedAt: 10,
2774+
task: expect.stringContaining("All pending descendants for that run have now settled"),
2775+
}),
2776+
);
27732777
});
27742778

27752779
it("does not re-wake an already woken run id", async () => {

‎src/agents/subagent-announce.registry.runtime.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -15,3 +15,4 @@ export {
1515
resolveRequesterForChildSession,
1616
shouldIgnorePostCompletionAnnounceForSession,
1717
} from "./subagent-registry-runtime.js";
18+
export { markDescendantCompletionConsumedByRequester } from "./subagent-registry-consumption.js";

‎src/agents/subagent-announce.test.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -48,6 +48,7 @@ const { subagentRegistryRuntimeMock } = vi.hoisted(() => ({
4848
countPendingDescendantRuns: vi.fn(() => 0),
4949
countPendingDescendantRunsExcludingRun: vi.fn(() => 0),
5050
listSubagentRunsForRequester: vi.fn(() => []),
51+
markDescendantCompletionConsumedByRequester: vi.fn(() => 0),
5152
replaceSubagentRunAfterSteer: vi.fn(() => true),
5253
resolveRequesterForChildSession: vi.fn(() => null),
5354
},

‎src/agents/subagent-announce.timeout.test.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -201,6 +201,7 @@ vi.mock("./subagent-announce.registry.runtime.js", () => ({
201201
countPendingDescendantRuns: () => pendingDescendantRuns,
202202
countPendingDescendantRunsExcludingRun: () => 0,
203203
listSubagentRunsForRequester: () => [],
204+
markDescendantCompletionConsumedByRequester: () => 0,
204205
isSubagentSessionRunActive: () => subagentSessionRunActive,
205206
shouldIgnorePostCompletionAnnounceForSession: () => shouldIgnorePostCompletion,
206207
replaceSubagentRunAfterSteer: () => true,

‎src/agents/subagent-announce.ts‎

Lines changed: 119 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,7 @@ import type { SubagentAnnounceDeliveryResult } from "./subagent-announce-dispatc
3434
import { resolveAnnounceOrigin } from "./subagent-announce-origin.js";
3535
import {
3636
applySubagentWaitOutcome,
37-
buildChildCompletionFindings,
37+
buildChildCompletionFindingsWithRuns,
3838
buildCompactAnnounceStatsLine,
3939
dedupeLatestChildCompletionRows,
4040
filterCurrentDirectChildCompletionRows,
@@ -128,6 +128,97 @@ function buildDescendantWakeMessage(params: { findings: string; taskLabel: strin
128128

129129
const WAKE_RUN_SUFFIX = ":wake";
130130

131+
type MarkRequesterConsumedCompletion = (params: {
132+
requesterSessionKey: string;
133+
runStartedAt: number;
134+
runIds: readonly string[];
135+
kind: "subagent_descendant_result";
136+
consumerRunId?: string;
137+
}) => unknown;
138+
139+
type RequesterConsumedRegistryRuntime = {
140+
markDescendantCompletionConsumedByRequester?: MarkRequesterConsumedCompletion;
141+
};
142+
143+
function buildRequesterConsumedChildCompletion(
144+
directChildren: Parameters<typeof filterCurrentDirectChildCompletionRows>[0],
145+
options: Parameters<typeof filterCurrentDirectChildCompletionRows>[1],
146+
): { text?: string; consumedRunIds: string[] } {
147+
const currentDirectChildren = filterCurrentDirectChildCompletionRows(directChildren, options);
148+
const childCompletion = buildChildCompletionFindingsWithRuns(
149+
dedupeLatestChildCompletionRows(currentDirectChildren),
150+
);
151+
return { text: childCompletion?.text, consumedRunIds: childCompletion?.consumedRunIds ?? [] };
152+
}
153+
154+
function normalizeRunIdList(runIds: readonly string[] | undefined): string[] {
155+
const seen = new Set<string>();
156+
const result: string[] = [];
157+
for (const runId of runIds ?? []) {
158+
const trimmed = runId.trim();
159+
if (!trimmed || seen.has(trimmed)) {
160+
continue;
161+
}
162+
seen.add(trimmed);
163+
result.push(trimmed);
164+
}
165+
return result;
166+
}
167+
168+
function deliveryHasRequesterConsumptionCredit(delivery: SubagentAnnounceDeliveryResult): boolean {
169+
if (!delivery.delivered) {
170+
return false;
171+
}
172+
const phases = delivery.phases ?? [];
173+
if (phases.some((phase) => phase.phase === "direct-primary" && phase.delivered)) {
174+
return true;
175+
}
176+
// Older callers/tests may construct a bare direct result without dispatch phase evidence.
177+
// Fallback steering is never represented as path="direct", so keep legacy direct-delivery
178+
// credit while denying transcript-only steer fallback.
179+
return phases.length === 0 && delivery.path === "direct";
180+
}
181+
182+
function recordRequesterConsumedDescendantCompletions(params: {
183+
delivery: SubagentAnnounceDeliveryResult;
184+
subagentRegistryRuntime?: RequesterConsumedRegistryRuntime;
185+
childSessionKey: string;
186+
childRunId: string;
187+
startedAt?: number;
188+
childCompletionFindings?: string;
189+
consumedChildRunIds?: readonly string[];
190+
pendingRequesterConsumedDescendantRunIds?: readonly string[];
191+
pendingRequesterConsumedRunStartedAt?: number;
192+
}): void {
193+
if (!deliveryHasRequesterConsumptionCredit(params.delivery)) {
194+
return;
195+
}
196+
const mark = params.subagentRegistryRuntime?.markDescendantCompletionConsumedByRequester;
197+
const startedAt = params.startedAt ?? 0;
198+
const directRunIds = params.childCompletionFindings?.trim()
199+
? normalizeRunIdList(params.consumedChildRunIds)
200+
: [];
201+
if (directRunIds.length > 0) {
202+
mark?.({
203+
requesterSessionKey: params.childSessionKey,
204+
runStartedAt: startedAt,
205+
runIds: directRunIds,
206+
kind: "subagent_descendant_result",
207+
consumerRunId: params.childRunId,
208+
});
209+
}
210+
const pendingRunIds = normalizeRunIdList(params.pendingRequesterConsumedDescendantRunIds);
211+
if (pendingRunIds.length > 0) {
212+
mark?.({
213+
requesterSessionKey: params.childSessionKey,
214+
runStartedAt: params.pendingRequesterConsumedRunStartedAt ?? startedAt,
215+
runIds: pendingRunIds,
216+
kind: "subagent_descendant_result",
217+
consumerRunId: params.childRunId,
218+
});
219+
}
220+
}
221+
131222
function stripWakeRunSuffixes(runId: string): string {
132223
let next = runId.trim();
133224
while (next.endsWith(WAKE_RUN_SUFFIX)) {
@@ -172,6 +263,8 @@ async function wakeSubagentRunAfterDescendants(params: {
172263
findings: string;
173264
announceId: string;
174265
signal?: AbortSignal;
266+
pendingRequesterConsumedDescendantRunIds?: string[];
267+
pendingRequesterConsumedRunStartedAt?: number;
175268
}): Promise<boolean> {
176269
if (params.signal?.aborted) {
177270
return false;
@@ -231,6 +324,8 @@ async function wakeSubagentRunAfterDescendants(params: {
231324
// Persist the wake message as the replacement run's task so that any
232325
// post-restart redispatch reconstructs the correct prompt.
233326
task: wakeMessage,
327+
pendingRequesterConsumedDescendantRunIds: params.pendingRequesterConsumedDescendantRunIds,
328+
pendingRequesterConsumedRunStartedAt: params.pendingRequesterConsumedRunStartedAt,
234329
});
235330
}
236331

@@ -258,6 +353,8 @@ export async function runSubagentAnnounceFlow(params: {
258353
expectsCompletionMessage?: boolean;
259354
spawnMode?: SpawnSubagentMode;
260355
wakeOnDescendantSettle?: boolean;
356+
pendingRequesterConsumedDescendantRunIds?: string[];
357+
pendingRequesterConsumedRunStartedAt?: number;
261358
signal?: AbortSignal;
262359
bestEffortDeliver?: boolean;
263360
onDeliveryResult?: (delivery: SubagentAnnounceDeliveryResult) => void;
@@ -317,6 +414,7 @@ export async function runSubagentAnnounceFlow(params: {
317414
requesterDepth >= 1 || isCronSessionKey(targetRequesterSessionKey);
318415

319416
let childCompletionFindings: string | undefined;
417+
let consumedChildRunIds: string[] = [];
320418
let subagentRegistryRuntime:
321419
| Awaited<ReturnType<typeof loadSubagentRegistryRuntime>>
322420
| undefined;
@@ -348,15 +446,13 @@ export async function runSubagentAnnounceFlow(params: {
348446
},
349447
);
350448
if (Array.isArray(directChildren) && directChildren.length > 0) {
351-
childCompletionFindings = buildChildCompletionFindings(
352-
dedupeLatestChildCompletionRows(
353-
filterCurrentDirectChildCompletionRows(directChildren, {
354-
requesterSessionKey: params.childSessionKey,
355-
getLatestSubagentRunByChildSessionKey:
356-
subagentRegistryRuntime.getLatestSubagentRunByChildSessionKey,
357-
}),
358-
),
359-
);
449+
const childCompletion = buildRequesterConsumedChildCompletion(directChildren, {
450+
requesterSessionKey: params.childSessionKey,
451+
getLatestSubagentRunByChildSessionKey:
452+
subagentRegistryRuntime.getLatestSubagentRunByChildSessionKey,
453+
});
454+
consumedChildRunIds = childCompletion.consumedRunIds;
455+
childCompletionFindings = childCompletion.text;
360456
}
361457
}
362458
} catch {
@@ -385,6 +481,8 @@ export async function runSubagentAnnounceFlow(params: {
385481
findings: childCompletionFindings,
386482
announceId: wakeAnnounceId,
387483
signal: params.signal,
484+
pendingRequesterConsumedDescendantRunIds: consumedChildRunIds,
485+
pendingRequesterConsumedRunStartedAt: params.startedAt,
388486
});
389487
if (woke) {
390488
shouldDeleteChildSession = false;
@@ -599,6 +697,17 @@ export async function runSubagentAnnounceFlow(params: {
599697
});
600698
params.onDeliveryResult?.(delivery);
601699
didAnnounce = delivery.delivered || delivery.terminal === true;
700+
recordRequesterConsumedDescendantCompletions({
701+
delivery,
702+
subagentRegistryRuntime,
703+
childSessionKey: params.childSessionKey,
704+
childRunId: params.childRunId,
705+
startedAt: params.startedAt,
706+
childCompletionFindings,
707+
consumedChildRunIds,
708+
pendingRequesterConsumedDescendantRunIds: params.pendingRequesterConsumedDescendantRunIds,
709+
pendingRequesterConsumedRunStartedAt: params.pendingRequesterConsumedRunStartedAt,
710+
});
602711
if (!delivery.delivered && delivery.path === "direct" && delivery.error) {
603712
defaultRuntime.log(
604713
`[warn] Subagent completion direct announce failed for run ${params.childRunId}: ${delivery.error}`,

‎src/agents/subagent-delivery-state.test.ts‎

Lines changed: 34 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -211,6 +211,40 @@ describe("normalizeSubagentRunState", () => {
211211
expect(entry.cleanupHandled).toBe(false);
212212
});
213213

214+
it("preserves requester-consumed delivery credit across normalization", () => {
215+
const entry = normalizeSubagentRunState(
216+
baseRun({
217+
cleanupHandled: true,
218+
delivery: {
219+
status: "delivered",
220+
deliveredAt: 500,
221+
requesterConsumedAt: 500,
222+
requesterConsumedKind: "cron_descendant_fallback",
223+
requesterConsumedBySessionKey: "agent:main:cron:daily:run:abc",
224+
requesterConsumedRunStartedAt: 400,
225+
requesterConsumedMetadata: {
226+
consumerRunId: "cron-run",
227+
deliveryTextHash: "abc123",
228+
},
229+
},
230+
}),
231+
);
232+
233+
expect(entry.delivery).toMatchObject({
234+
status: "delivered",
235+
deliveredAt: 500,
236+
requesterConsumedAt: 500,
237+
requesterConsumedKind: "cron_descendant_fallback",
238+
requesterConsumedBySessionKey: "agent:main:cron:daily:run:abc",
239+
requesterConsumedRunStartedAt: 400,
240+
requesterConsumedMetadata: {
241+
consumerRunId: "cron-run",
242+
deliveryTextHash: "abc123",
243+
},
244+
});
245+
expect(entry.cleanupHandled).toBe(false);
246+
});
247+
214248
it("keeps discarded terminal delivery dormant across restart", () => {
215249
const entry = normalizeSubagentRunState(
216250
baseRun({

0 commit comments

Comments
 (0)