Skip to content

Commit 633c75a

Browse files
committed
feat(subagent): classify doom-loop worker stops as loop_guard
The reactor's doom-loop guard already stops workers, but the run ended as a plain error and its class survived only in the reactor message. The doom_loop run end reaches the stream after the send rejects, so the run notes it in its stream sink and rethrows a typed error once teardown has drained the stream. A forced-stop reason would have taken the salvage path and turned the failed lane into a done one. The run also reports when its own teardown fails, which is the release step behind a partial cleanup outcome.
1 parent 98e4e5f commit 633c75a

3 files changed

Lines changed: 249 additions & 4 deletions

File tree

Lines changed: 150 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,150 @@
1+
import { describe, expect, test } from "bun:test";
2+
import { rm } from "node:fs/promises";
3+
import { join } from "node:path";
4+
import type { ReactorEmittedEvent } from "@intx/inference";
5+
6+
import {
7+
callFleetTool,
8+
createFleetDeps,
9+
deferred,
10+
fleetTools,
11+
spawnAgentId,
12+
} from "./fleet-test-harness.js";
13+
import {
14+
baseRunParams,
15+
stubAgent,
16+
tmpSubAgentCwd,
17+
withStubbedAgent,
18+
} from "./run-test-harness.js";
19+
import { isSubAgentLoopGuardError } from "./terminal-failure.js";
20+
import type {
21+
RunSubAgentParams,
22+
RunSubAgentResult,
23+
SubAgentRunSettlement,
24+
} from "./types.js";
25+
26+
const DOOM_MESSAGE =
27+
"Doom loop detected: an identical tool batch (grep, glob) executed 3 times consecutively";
28+
29+
type Run = (params: RunSubAgentParams) => Promise<RunSubAgentResult>;
30+
31+
/**
32+
* Replays the vendored reactor's doom-loop shutdown as a worker sees it: the
33+
* fatal reactor.error rejects the send first, and the `doom_loop` run end
34+
* only reaches the stream after that rejection.
35+
*/
36+
async function withRunEndingIn<T>(
37+
kind: string,
38+
body: (run: Run, cwd: string) => Promise<T>,
39+
): Promise<T> {
40+
const cwd = await tmpSubAgentCwd("loop-guard-");
41+
const sendRejected = deferred<undefined>();
42+
try {
43+
return await withStubbedAgent(
44+
stubAgent({
45+
send: async () => {
46+
setTimeout(() => sendRejected.resolve(undefined), 0);
47+
throw new Error(`reactor error: ${DOOM_MESSAGE}`);
48+
},
49+
stream: () =>
50+
(async function* (): AsyncGenerator<ReactorEmittedEvent> {
51+
yield {
52+
type: "inference.start",
53+
seq: 1,
54+
data: { sourceId: "test", model: "test-model", input: [] },
55+
} as unknown as ReactorEmittedEvent;
56+
await sendRejected.promise;
57+
yield {
58+
type: "reactor.error",
59+
seq: 2,
60+
data: { error: DOOM_MESSAGE, fatal: true },
61+
} as unknown as ReactorEmittedEvent;
62+
yield {
63+
type: "message.run.ended",
64+
seq: 3,
65+
data: {
66+
messageRunId: "run-1",
67+
messageId: "msg-1",
68+
status: "failed",
69+
error: { message: DOOM_MESSAGE, kind },
70+
},
71+
} as unknown as ReactorEmittedEvent;
72+
})(),
73+
}),
74+
async () => {
75+
const { runSubAgent } = await import("./run.js");
76+
return body(runSubAgent, cwd);
77+
},
78+
);
79+
} finally {
80+
await rm(cwd, { recursive: true, force: true });
81+
}
82+
}
83+
84+
async function spawnAndCollect(
85+
run: Run,
86+
cwd: string,
87+
): Promise<Record<string, unknown>> {
88+
const deps = createFleetDeps(run, { cwd });
89+
deps.getWorkdirBase = () => join(cwd, ".ctx");
90+
const tools = fleetTools(deps);
91+
const id = await spawnAgentId(tools.spawn, {
92+
description: "discovery",
93+
prompt: "find it",
94+
intent: "explore",
95+
});
96+
const waited = await callFleetTool(tools.wait, {
97+
targets: [id],
98+
timeout_ms: 5_000,
99+
});
100+
const rows = waited.results as Record<string, unknown>[];
101+
return rows[0] ?? {};
102+
}
103+
104+
describe("loop-guard termination", () => {
105+
test("runSubAgent rethrows a doom-loop stop as a typed loop-guard error", async () => {
106+
await withRunEndingIn("doom_loop", async (run, cwd) => {
107+
let settled: SubAgentRunSettlement | undefined;
108+
let caught: unknown;
109+
try {
110+
await run(
111+
baseRunParams(cwd, {
112+
onRunSettled: (summary) => {
113+
settled = summary;
114+
},
115+
}),
116+
);
117+
} catch (error) {
118+
caught = error;
119+
}
120+
expect(isSubAgentLoopGuardError(caught)).toBe(true);
121+
expect((caught as Error).message).toBe(`reactor error: ${DOOM_MESSAGE}`);
122+
// Telemetry stop_reason is unchanged: the run still ends as "error".
123+
expect(settled?.terminal_reason).toBe("error");
124+
});
125+
});
126+
127+
test("the parent sees loop_guard, distinct from a provider failure", async () => {
128+
await withRunEndingIn("doom_loop", async (run, cwd) => {
129+
const row = await spawnAndCollect(run, cwd);
130+
expect(row.status).toBe("failed");
131+
expect(row.error).toBe(`reactor error: ${DOOM_MESSAGE}`);
132+
expect(row.failure).toMatchObject({
133+
failure_class: "loop_guard",
134+
recovery: { available: false, reason: "not_retryable" },
135+
});
136+
expect(row.provider_failure).toBeUndefined();
137+
expect(row.continuable).toBeUndefined();
138+
});
139+
});
140+
141+
test("a failed run end of another kind is not a loop guard", async () => {
142+
await withRunEndingIn("reactor_fatal", async (run, cwd) => {
143+
const row = await spawnAndCollect(run, cwd);
144+
expect(row.status).toBe("failed");
145+
expect((row.failure as { failure_class?: string }).failure_class).toBe(
146+
"error",
147+
);
148+
});
149+
});
150+
});
Lines changed: 64 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,64 @@
1+
import { describe, expect, test } from "bun:test";
2+
import { rm } from "node:fs/promises";
3+
import { join } from "node:path";
4+
5+
import {
6+
callFleetTool,
7+
createFleetDeps,
8+
fleetTools,
9+
spawnAgentId,
10+
} from "./fleet-test-harness.js";
11+
import {
12+
stubAgent,
13+
tmpSubAgentCwd,
14+
withPosixDispose,
15+
withStubbedAgent,
16+
} from "./run-test-harness.js";
17+
18+
describe("terminal failure cleanup outcome", () => {
19+
test.each([
20+
["a clean teardown", async (): Promise<void> => undefined, "released"],
21+
[
22+
"a teardown that leaves children behind",
23+
async (): Promise<void> => {
24+
throw new Error("1 shell child process still live after 2000ms reap");
25+
},
26+
"partial",
27+
],
28+
] as const)("%s records cleanup %s", async (_label, dispose, expected) => {
29+
const cwd = await tmpSubAgentCwd("terminal-cleanup-");
30+
try {
31+
await withPosixDispose(dispose, () =>
32+
withStubbedAgent(
33+
stubAgent({
34+
send: async () => {
35+
throw new Error("send exploded");
36+
},
37+
}),
38+
async () => {
39+
const { runSubAgent } = await import("./run.js");
40+
const deps = createFleetDeps(runSubAgent, { cwd });
41+
deps.getWorkdirBase = () => join(cwd, ".ctx");
42+
const tools = fleetTools(deps);
43+
const id = await spawnAgentId(tools.spawn, {
44+
description: "cleanup probe",
45+
prompt: "fail",
46+
intent: "explore",
47+
});
48+
const waited = await callFleetTool(tools.wait, {
49+
targets: [id],
50+
timeout_ms: 5_000,
51+
});
52+
const row = (waited.results as Record<string, unknown>[])[0];
53+
expect(row?.status).toBe("failed");
54+
expect(
55+
(row?.failure as { cleanup?: string } | undefined)?.cleanup,
56+
).toBe(expected);
57+
},
58+
),
59+
);
60+
} finally {
61+
await rm(cwd, { recursive: true, force: true });
62+
}
63+
});
64+
});

‎src/subagent/run.ts‎

Lines changed: 35 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -207,6 +207,7 @@ import type {
207207

208208
import type { TaskIntent } from "./report.js";
209209
import { runWithSubAgentIdentity } from "./identity-context.js";
210+
import { createSubAgentLoopGuardError } from "./terminal-failure.js";
210211

211212
/**
212213
* Worker authorization denies without suspending, so Agent.send must settle
@@ -600,7 +601,9 @@ export async function runSubAgent(
600601
};
601602
let terminalReason: SubAgentTerminalReason = "error";
602603
let errorCount = 0;
603-
const settlementState = { latestModel: params.provider.model };
604+
const settlementState: RunSettlementState = {
605+
latestModel: params.provider.model,
606+
};
604607

605608
try {
606609
const result = await runSubAgentInner(
@@ -612,8 +615,16 @@ export async function runSubAgent(
612615
return result;
613616
} catch (error) {
614617
errorCount = 1;
615-
if (isSubAgentCancelError(error, params.signal))
618+
if (isSubAgentCancelError(error, params.signal)) {
616619
terminalReason = "cancelled";
620+
throw error;
621+
}
622+
// CL-10202: the inner finally has drained the stream by now, so a
623+
// doom_loop run end the reactor emitted after rejecting the send has
624+
// been seen. Telemetry keeps terminal_reason "error".
625+
if (settlementState.loopGuardTripped === true) {
626+
throw createSubAgentLoopGuardError(error);
627+
}
617628
throw error;
618629
} finally {
619630
try {
@@ -624,6 +635,9 @@ export async function runSubAgent(
624635
duration_ms: Date.now() - startedAt,
625636
model: settlementState.latestModel,
626637
terminal_reason: terminalReason,
638+
...(settlementState.teardownFailed === true
639+
? { teardown_failed: true as const }
640+
: {}),
627641
}),
628642
);
629643
} catch {
@@ -632,10 +646,17 @@ export async function runSubAgent(
632646
}
633647
}
634648

649+
interface RunSettlementState {
650+
latestModel: string;
651+
/** The reactor ended a message run with `error.kind: "doom_loop"`. */
652+
loopGuardTripped?: boolean;
653+
teardownFailed?: boolean;
654+
}
655+
635656
async function runSubAgentInner(
636657
params: RunSubAgentParams,
637658
telemetryRollup: SubAgentTelemetryRollup,
638-
settlementState: { latestModel: string },
659+
settlementState: RunSettlementState,
639660
): Promise<RunSubAgentResult> {
640661
// Fleet-spawned workers skip the pricing-cache seed re-read; the parent
641662
// runtime already applied the process seed at boot.
@@ -1420,6 +1441,13 @@ async function runSubAgentInner(
14201441
const runSettlement = createRunEventSettlement();
14211442
const streamSink = (rawEvent: ReactorEmittedEvent): void => {
14221443
runSettlement.handleEvent(rawEvent);
1444+
if (
1445+
rawEvent.type === "message.run.ended" &&
1446+
rawEvent.data.status === "failed" &&
1447+
rawEvent.data.error?.kind === "doom_loop"
1448+
) {
1449+
settlementState.loopGuardTripped = true;
1450+
}
14231451
const event = sanitizeDiagnosticValue(rawEvent, [
14241452
readSourceCredentialMaterial(workerSource.credentialId).secret,
14251453
]) as ReactorEmittedEvent;
@@ -1911,7 +1939,10 @@ async function runSubAgentInner(
19111939
posixTools,
19121940
}),
19131941
params.teardownDeadlineMs ?? DEFAULT_CLOSE_DEADLINE_MS,
1914-
);
1942+
).catch((teardownError: unknown) => {
1943+
settlementState.teardownFailed = true;
1944+
throw teardownError;
1945+
});
19151946
}
19161947
}
19171948
}

0 commit comments

Comments
 (0)