Skip to content

Make subagent follow-up lifecycle race safe #910

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
TheGreatAxios merged 1 commit into from
Sep 11, 2026
Merged

Make subagent follow-up lifecycle race safe #910

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
125 changes: 53 additions & 72 deletions src/subagent/agent-fleet.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1502,13 +1502,16 @@ describe("interrupt_agent unblocks wait_agents", () => {

test("close overlay without in-flight wait stays interrupted after followup complete", async () => {
const gate = deferred<RunSubAgentResult>();
const followupGate = deferred<string>();
const closeHold = deferred<undefined>();
const followupCalls: string[] = [];
const deps = makeDeps(async (params) => {
params.onAgentReady?.({
close: async () => closeHold.promise,
interrupt: () => undefined,
followup: async () => followupGate.promise,
followup: async (message: string) => {
followupCalls.push(message);
return "should never run";
},
deliver: () => undefined,
});
return gate.promise;
Expand Down Expand Up @@ -1552,25 +1555,16 @@ describe("interrupt_agent unblocks wait_agents", () => {
new AbortController().signal,
);

followupGate.resolve("followup after close overlay");
await new Promise<void>((resolve) => {
const done = (): boolean =>
deps.sessions.get(id)?.lifecycle.state === "completed";
if (done()) {
resolve();
return;
}
const unsub = deps.sessions.subscribe(() => {
if (done()) {
unsub();
resolve();
}
});
if (done()) {
unsub();
resolve();
}
});
closeHold.resolve(undefined);
await closing;
// CL-7344: close drops the stashed follow-up — it never runs against
// the closed agent, and the close overlay survives the run settling.
expect(followupCalls).toEqual([]);
gate.resolve({
report: "original interrupted",
interrupted: true,
} as RunSubAgentResult);
await new Promise((resolve) => setTimeout(resolve, 20));

expect(deps.fleetRecords.peek(id)?.status).toBe("interrupted");
const listed = await callTool(list, {});
Expand All @@ -1583,13 +1577,6 @@ describe("interrupt_agent unblocks wait_agents", () => {
expect(waited.timed_out).toBe(false);
const results = waited.results as { status: string }[];
expect(defined(results[0]).status).toBe("interrupted");

closeHold.resolve(undefined);
await closing;
gate.resolve({
report: "original interrupted",
interrupted: true,
} as RunSubAgentResult);
});

test("completeAfterInterrupt does not clear a close overlay", () => {
Expand Down Expand Up @@ -1642,14 +1629,17 @@ describe("interrupt_agent unblocks wait_agents", () => {
message: "stop that",
interrupt: true,
});
followupGate.reject(new Error("followup failed"));
await new Promise((resolve) => setTimeout(resolve, 20));
// CL-7344: the follow-up is stashed until the original run settles; the
// salvage handoff launches it, and its rejection restores the
// interrupted lane so wait collects the salvage.
gate.resolve({
report:
"## Summary\nStopped.\n## Findings\nsalvage\n## Blockers\ninterrupted\n## Paths\n",
interrupted: true,
} as RunSubAgentResult);
await new Promise((resolve) => setTimeout(resolve, 20));
followupGate.reject(new Error("followup failed"));
await new Promise((resolve) => setTimeout(resolve, 20));

const waited = await callTool(wait, { targets: [id], timeout_ms: 5000 });
expect(waited.timed_out).toBe(false);
Expand All @@ -1658,21 +1648,14 @@ describe("interrupt_agent unblocks wait_agents", () => {
expect(defined(results[0]).report).toContain("salvage");
});

test("send_input interrupt queued overlay clears when the followup is admitted", async () => {
const admission = createAdmissionQueue({ capacity: 1 });
const sessions = createSubAgentSessionStore({ admission });
test("send_input interrupt on a live lane stashes the followup until the run settles", async () => {
const sessions = createSubAgentSessionStore();
const fleetRecords = createFleetMailbox(sessions);
admission.enqueue({
id: "holder",
provider: "p",
start: () => undefined,
});
const worker = sessions.start({
description: "looping",
agentId: "explorer",
brief: "b",
retained: true,
provider: "p",
});
sessions.markRunning(worker.id);
fleetRecords.register(worker.id);
Expand All @@ -1690,51 +1673,44 @@ describe("interrupt_agent unblocks wait_agents", () => {
interrupt: true,
});
expect(sent.status).toBe("interrupted");
expect(sessions.get(worker.id)?.lifecycleStatus).toBe("pending_init");
// CL-7344: the interrupt stashes the follow-up instead of starting it;
// the lane is still the live run, so the session stays running while
// wait/list stay live until the run settles and hands off.
expect(sessions.get(worker.id)?.lifecycleStatus).toBe("running");

const queuedWait = await callTool(wait, {
const liveWait = await callTool(wait, {
targets: [worker.id],
timeout_ms: 50,
});
expect(queuedWait.timed_out).toBe(true);
expect(
defined((queuedWait.results as { status: string }[])[0]).status,
).toBe("queued");
const queuedList = await callTool(list, {});
const queuedEntry = (
queuedList.agents as {
expect(liveWait.timed_out).toBe(true);
expect(defined((liveWait.results as { status: string }[])[0]).status).toBe(
"running",
);
const liveList = await callTool(list, {});
const liveEntry = (
liveList.agents as {
agent_id: string;
status: string;
lifecycle: string;
}[]
).find((a) => a.agent_id === worker.id);
expect(queuedEntry?.status).toBe("queued");
expect(queuedEntry?.lifecycle).toBe("pending_init");

admission.release("holder");
await new Promise((resolve) => setTimeout(resolve, 20));
expect(liveEntry?.status).toBe("running");
expect(liveEntry?.lifecycle).toBe("running");

// Settling the original run launches the stashed follow-up.
sessions.attachReport(worker.id, "interrupted salvage", {
stopReason: "interrupted",
});
expect(sessions.get(worker.id)?.lifecycleStatus).toBe("running");
const runningWait = await callTool(wait, {
followupGate.resolve("later");
const done = await callTool(wait, {
targets: [worker.id],
timeout_ms: 50,
timeout_ms: 5000,
});
expect(runningWait.timed_out).toBe(true);
expect(
defined((runningWait.results as { status: string }[])[0]).status,
).toBe("running");
const runningList = await callTool(list, {});
const runningEntry = (
runningList.agents as {
agent_id: string;
status: string;
lifecycle: string;
}[]
).find((a) => a.agent_id === worker.id);
expect(runningEntry?.status).toBe("running");
expect(runningEntry?.lifecycle).toBe("running");

followupGate.resolve("later");
expect(done.timed_out).toBe(false);
const results = done.results as { status: string; report?: string }[];
expect(defined(results[0]).status).toBe("done");
expect(defined(results[0]).report).toBe("later");
});

test("soft-interrupt wait path collects so omitted re-wait does not re-deliver", async () => {
Expand Down Expand Up @@ -1910,8 +1886,13 @@ describe("interrupt_agent unblocks wait_agents", () => {
message: "stop that",
interrupt: true,
});
// CL-7344: the follow-up is stashed until the original run settles; the
// salvage handoff launches it, so the run must settle first.
gate.resolve({
report: "original interrupted",
interrupted: true,
} as RunSubAgentResult);
followupGate.resolve("followup report");
await new Promise((resolve) => setTimeout(resolve, 20));
const waited = await callTool(wait, { targets: [id], timeout_ms: 5000 });
expect(waited.timed_out).toBe(false);
const results = waited.results as { status: string; report?: string }[];
Expand Down
36 changes: 13 additions & 23 deletions src/subagent/agent-fleet.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -249,11 +249,8 @@ class FleetMailbox {
/**
* CL-7331: mark that a send_input interrupt:true followup owns this lane.
* Suppresses any interrupt overlay so wait stays live (running/queued)
* until the followup settles, and tells the spawn settlement to swallow
* the original run's interrupted result instead of attaching salvage over
* the live followup. No-op on an unknown id; safe to call on a collected
* mailbox (frozen status still wins for projection, but the settlement
* swallow still applies).
* until the followup settles. No-op on an unknown id; safe to call on a
* collected mailbox (frozen status still wins for projection).
*/
noteFollowup(id: string): void {
const existing = this.records.get(id);
Expand All @@ -263,11 +260,6 @@ class FleetMailbox {
this.sessions?.wake();
}

/** True while a send_input interrupt:true followup owns this lane. */
hasLiveFollowup(id: string): boolean {
return this.records.get(id)?.followupLive === true;
}

/**
* send_input interrupt:true followup finished. Clear the followup lane flag
* (and any admission queued overlay) so wait can project the settled session.
Expand Down Expand Up @@ -1387,19 +1379,17 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool {
if (result.interrupted === true) {
keepWorktreeAlive = true;
runInterrupted = true;
// CL-7331: a followup started via send_input interrupt owns
// this lane now — the settling original turn must not attach
// salvage over it. The mailbox flag (set by send_input, not
// by interrupt_agent or a bare settle) identifies that lane;
// lifecycle alone cannot, since a never-started run and a
// queued followup both read pending_init.
if (!deps.fleetRecords.hasLiveFollowup(session.id)) {
deps.sessions.attachReport(session.id, result.report, {
...(result.stopReason !== undefined
? { stopReason: result.stopReason }
: {}),
});
}
// CL-7344: a followup stashed via send_input interrupt launches
// from attachReport, so the settling original turn must
// attach its salvage here — attachReport records the salvage
// and launches the stashed followup in one atomic step.
// Skipping the attach would strand the stash and hang
// wait_agents on a phantom turn.
deps.sessions.attachReport(session.id, result.report, {
...(result.stopReason !== undefined
? { stopReason: result.stopReason }
: {}),
});
return;
}
const alreadyCancelled =
Expand Down
Loading
Loading