Skip to content

Commit 7d1bdcf

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 77fe63b commit 7d1bdcf

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
@@ -667,7 +668,9 @@ export async function runSubAgent(
667668
};
668669
let terminalReason: SubAgentTerminalReason = "error";
669670
let errorCount = 0;
670-
const settlementState = { latestModel: params.provider.model };
671+
const settlementState: RunSettlementState = {
672+
latestModel: params.provider.model,
673+
};
671674

672675
try {
673676
const result = await runSubAgentInner(
@@ -679,8 +682,16 @@ export async function runSubAgent(
679682
return result;
680683
} catch (error) {
681684
errorCount = 1;
682-
if (isSubAgentCancelError(error, params.signal))
685+
if (isSubAgentCancelError(error, params.signal)) {
683686
terminalReason = "cancelled";
687+
throw error;
688+
}
689+
// CL-10202: the inner finally has drained the stream by now, so a
690+
// doom_loop run end the reactor emitted after rejecting the send has
691+
// been seen. Telemetry keeps terminal_reason "error".
692+
if (settlementState.loopGuardTripped === true) {
693+
throw createSubAgentLoopGuardError(error);
694+
}
684695
throw error;
685696
} finally {
686697
try {
@@ -691,6 +702,9 @@ export async function runSubAgent(
691702
duration_ms: Date.now() - startedAt,
692703
model: settlementState.latestModel,
693704
terminal_reason: terminalReason,
705+
...(settlementState.teardownFailed === true
706+
? { teardown_failed: true as const }
707+
: {}),
694708
}),
695709
);
696710
} catch {
@@ -699,10 +713,17 @@ export async function runSubAgent(
699713
}
700714
}
701715

716+
interface RunSettlementState {
717+
latestModel: string;
718+
/** The reactor ended a message run with `error.kind: "doom_loop"`. */
719+
loopGuardTripped?: boolean;
720+
teardownFailed?: boolean;
721+
}
722+
702723
async function runSubAgentInner(
703724
params: RunSubAgentParams,
704725
telemetryRollup: SubAgentTelemetryRollup,
705-
settlementState: { latestModel: string },
726+
settlementState: RunSettlementState,
706727
): Promise<RunSubAgentResult> {
707728
// Fleet-spawned workers skip the pricing-cache seed re-read; the parent
708729
// runtime already applied the process seed at boot.
@@ -1488,6 +1509,13 @@ async function runSubAgentInner(
14881509
const runSettlement = createRunEventSettlement();
14891510
const streamSink = (rawEvent: ReactorEmittedEvent): void => {
14901511
runSettlement.handleEvent(rawEvent);
1512+
if (
1513+
rawEvent.type === "message.run.ended" &&
1514+
rawEvent.data.status === "failed" &&
1515+
rawEvent.data.error?.kind === "doom_loop"
1516+
) {
1517+
settlementState.loopGuardTripped = true;
1518+
}
14911519
const event = sanitizeDiagnosticValue(rawEvent, [
14921520
readSourceCredentialMaterial(workerSource.credentialId).secret,
14931521
]) as ReactorEmittedEvent;
@@ -1979,7 +2007,10 @@ async function runSubAgentInner(
19792007
posixTools,
19802008
}),
19812009
params.teardownDeadlineMs ?? DEFAULT_CLOSE_DEADLINE_MS,
1982-
);
2010+
).catch((teardownError: unknown) => {
2011+
settlementState.teardownFailed = true;
2012+
throw teardownError;
2013+
});
19832014
}
19842015
}
19852016
}

0 commit comments

Comments
 (0)