diff --git a/src/agent/background-shell-tool.test.ts b/src/agent/background-shell-tool.test.ts index b745c39b2..12ee2af99 100644 --- a/src/agent/background-shell-tool.test.ts +++ b/src/agent/background-shell-tool.test.ts @@ -129,6 +129,81 @@ describe("background shell through the agent toolset", () => { } }); + test("shell_collect with an aborted signal releases as running without killing the child", async () => { + const toolset = await createAgentToolset({ + cwd: process.cwd(), + permissionGate: gate(process.cwd()), + }); + try { + const started = await toolset.dynamicRunner.run( + { + id: "a-start", + name: "run_shell", + arguments: { + command: "sleep 60", + background: true, + }, + }, + new AbortController().signal, + ); + const { shell_id } = JSON.parse(String(started.content)) as { + shell_id: string; + }; + const aborted = new AbortController(); + aborted.abort(new Error("interrupted by interrupt_agent")); + const out = await toolset.dynamicRunner.run( + { + id: "a-collect", + name: "shell_collect", + arguments: { shell_id, action: "collect", wait_ms: 60_000 }, + }, + aborted.signal, + ); + expect(JSON.parse(String(out.content))).toMatchObject({ + status: "running", + }); + // A live-signal collect still sees the child running: the abort above + // released the waiter instead of killing the process. Had the abort + // killed it, this would already report completed. + const stillThere = await toolset.dynamicRunner.run( + { + id: "a-collect2", + name: "shell_collect", + arguments: { shell_id, action: "collect" }, + }, + new AbortController().signal, + ); + expect(JSON.parse(String(stillThere.content))).toMatchObject({ + status: "running", + }); + // And the child is still killable: cancel settles it as completed. + const cancelled = await toolset.dynamicRunner.run( + { + id: "a-cancel", + name: "shell_collect", + arguments: { shell_id, action: "cancel" }, + }, + new AbortController().signal, + ); + expect(JSON.parse(String(cancelled.content))).toMatchObject({ + status: "cancelling", + }); + const final = await toolset.dynamicRunner.run( + { + id: "a-collect3", + name: "shell_collect", + arguments: { shell_id, action: "collect", wait_ms: 5_000 }, + }, + new AbortController().signal, + ); + expect(JSON.parse(String(final.content))).toMatchObject({ + status: "completed", + }); + } finally { + await toolset.dispose(); + } + }); + test("toolset dispose kills every live background process group", async () => { const token = `ic_toolset_dispose_${randomUUID()}`; const toolset = await createAgentToolset({ diff --git a/src/agent/background-shell-tool.ts b/src/agent/background-shell-tool.ts index 8f79a4d15..b78170666 100644 --- a/src/agent/background-shell-tool.ts +++ b/src/agent/background-shell-tool.ts @@ -79,7 +79,10 @@ export function createShellCollectTool( ) { return { definition: shellCollectDefinition, - handler: async (rawArgs: Record): Promise => { + handler: async ( + rawArgs: Record, + signal?: AbortSignal, + ): Promise => { const parsed = ShellCollectArgs(rawArgs); if (parsed instanceof type.errors) { return "Error: shell_collect requires shell_id (string) and action ('collect' | 'cancel')."; @@ -91,7 +94,11 @@ export function createShellCollectTool( } return JSON.stringify({ shell_id, status: "cancelling" }); } - const snapshot = await registry.collect(shell_id, parsed.wait_ms ?? 0); + const snapshot = await registry.collect( + shell_id, + parsed.wait_ms ?? 0, + signal, + ); if (snapshot.state === "running") { return JSON.stringify({ shell_id, status: "running" }); } diff --git a/src/shell/background-shell.test.ts b/src/shell/background-shell.test.ts index 51dc053ff..c26458d6b 100644 --- a/src/shell/background-shell.test.ts +++ b/src/shell/background-shell.test.ts @@ -122,4 +122,53 @@ describe("background shell registry", () => { const after = await registry.collect(started.id, 0); expect(after.state).toBe("not-found"); }); + + test("collect with an aborted signal releases as running without killing the child", async () => { + const token = `ic_bg_abort_${randomUUID()}`; + const registry = createBackgroundShellRegistry(); + const started = registry.start({ + command: `bash -c 'exec -a ${token} sleep 600'`, + cwd: tmpCwd, + }); + if ("error" in started) throw new Error(started.error); + try { + const aborted = new AbortController(); + aborted.abort(new Error("interrupted by interrupt_agent")); + const snapshot = await registry.collect( + started.id, + 60_000, + aborted.signal, + ); + expect(snapshot.state).toBe("running"); + const stillThere = await registry.collect(started.id, 0); + expect(stillThere.state).toBe("running"); + const probe = spawnSync("pgrep", ["-f", token], { encoding: "utf8" }); + expect(probe.status).toBe(0); + } finally { + registry.disposeAll("test done"); + } + }); + + test("releaseWaiters wakes a parked collect without killing the child", async () => { + const token = `ic_bg_release_${randomUUID()}`; + const registry = createBackgroundShellRegistry(); + const started = registry.start({ + command: `bash -c 'exec -a ${token} sleep 600'`, + cwd: tmpCwd, + }); + if ("error" in started) throw new Error(started.error); + try { + const pending = registry.collect(started.id, 60_000); + await new Promise((r) => setTimeout(r, 100)); + registry.releaseWaiters(); + const snapshot = await pending; + expect(snapshot.state).toBe("running"); + const stillThere = await registry.collect(started.id, 0); + expect(stillThere.state).toBe("running"); + const probe = spawnSync("pgrep", ["-f", token], { encoding: "utf8" }); + expect(probe.status).toBe(0); + } finally { + registry.disposeAll("test done"); + } + }); }); diff --git a/src/shell/background-shell.ts b/src/shell/background-shell.ts index f22bf5fde..cbe5b3f30 100644 --- a/src/shell/background-shell.ts +++ b/src/shell/background-shell.ts @@ -64,9 +64,14 @@ export type BackgroundShellSnapshot = export interface BackgroundShellRegistry { start: (args: StartBackgroundShellArgs) => { id: string } | { error: string }; - collect: (id: string, waitMs?: number) => Promise; + collect: ( + id: string, + waitMs?: number, + signal?: AbortSignal, + ) => Promise; cancel: (id: string) => boolean; disposeAll: (reason: string) => void; + releaseWaiters: () => void; runningCount: () => number; } @@ -151,14 +156,33 @@ export function createBackgroundShellRegistry( const collect = async ( id: string, waitMs = 0, + signal?: AbortSignal, ): Promise => { const done = completed.get(id); if (done !== undefined) return { state: "completed", exit: done }; if (!running.has(id)) return { state: "not-found" }; if (waitMs > 0) { + // An already-aborted collect releases immediately as still-running: + // interrupt must not park the session on a live descendant, and must + // not kill it either — the child belongs to the still-alive session. + if (signal?.aborted === true) return { state: "running" }; await new Promise((resolve) => { - exitWaiters.set(id, resolve); - setTimeout(resolve, waitMs); + const timer = setTimeout(() => { + exitWaiters.delete(id); + signal?.removeEventListener("abort", onAbort); + resolve(); + }, waitMs); + const onAbort = (): void => { + clearTimeout(timer); + exitWaiters.delete(id); + resolve(); + }; + signal?.addEventListener("abort", onAbort, { once: true }); + exitWaiters.set(id, () => { + clearTimeout(timer); + signal?.removeEventListener("abort", onAbort); + resolve(); + }); }).finally(() => exitWaiters.delete(id)); const finished = completed.get(id); if (finished !== undefined) return { state: "completed", exit: finished }; @@ -173,12 +197,22 @@ export function createBackgroundShellRegistry( return true; }; + /** + * Wake every parked `collect` waiter as still-running without touching the + * children. Interrupt paths call this so a live descendant releases the + * session instead of wedging teardown; close paths use `disposeAll`, which + * wakes waiters and then kills the trees. + */ + const releaseWaiters = (): void => { + for (const wake of exitWaiters.values()) wake(); + exitWaiters.clear(); + }; + const disposeAll = (reason: string): void => { for (const child of running.values()) killProcessTree(child); running.clear(); completed.clear(); - for (const wake of exitWaiters.values()) wake(); - exitWaiters.clear(); + releaseWaiters(); // onExit is intentionally not fired for disposed shells: the session is // gone, so there is no later turn to deliver to (`reason` is for callers // that log it). @@ -190,6 +224,7 @@ export function createBackgroundShellRegistry( collect, cancel, disposeAll, + releaseWaiters, runningCount: () => running.size, }; } diff --git a/src/subagent/run-shell-child-reap.test.ts b/src/subagent/run-shell-child-reap.test.ts new file mode 100644 index 000000000..c2a1e3bfc --- /dev/null +++ b/src/subagent/run-shell-child-reap.test.ts @@ -0,0 +1,312 @@ +/** + * CL-7990: a session holding a live shell child must still settle. + * + * Teardown path (run.ts finally → disposeSubAgentSession under a 30s + * awaitBoundedTeardown): posixTools.dispose() runs the shell-guard + * reapLiveChildren (SIGKILL the child process groups, 2s reap wait), then + * agent.close() and the session-stream drain are awaited — voided, not hung + * on, when the reap reports leftovers. Interrupt additionally releases + * parked shell_collect waiters so a worker blocked in shell output + * collection comes back as still-running instead of wedging the run. + * + * These tests drive the real `runSubAgent` with a stub agent whose `send` + * holds a REAL live `sleep` child (killed on abort, like runGuardedShell's + * onAbort) and whose `stream()` stays open until the session closes — the + * production shape of a worker parked in shell collection. + */ +import { describe, expect, test } from "bun:test"; +import { spawn, type ChildProcess } from "node:child_process"; +import { mkdtemp } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +import { withMockedModuleDuring } from "../../tests/helpers/mock-module.js"; +import { defined } from "../../tests/helpers/defined.js"; +import { createPermissionGate } from "../permission/gate.js"; +import type { RunSubAgentParams, RunSubAgentResult } from "./types.js"; + +const testPermissionGate = createPermissionGate({ + approvals: [], + interactive: false, + skipPermissions: true, + reactorGated: false, +}); + +async function tmpCwd(): Promise { + return mkdtemp(join(tmpdir(), "cl7990-shell-child-")); +} + +async function waitForChildExit( + child: ChildProcess, + timeoutMs = 5_000, +): Promise { + const started = Date.now(); + while (child.exitCode === null && child.signalCode === null) { + if (Date.now() - started > timeoutMs) { + throw new Error( + `shell child pid=${child.pid} still live after ${timeoutMs}ms`, + ); + } + await new Promise((resolve) => setTimeout(resolve, 10)); + } +} + +/** + * Stub agent modeling a worker parked behind a live shell child: `send` + * spawns a real `sleep` descendant and pends (like shell collection output); + * the abort listener kills the child first, mirroring runGuardedShell's + * onAbort. `stream()` stays open until `releaseStream` — the session cannot + * drain while the descendant is wedged. + */ +function createShellChildAgent(opts?: { wedgeClose?: boolean }) { + const children: ChildProcess[] = []; + let releaseStream: () => void = () => { + // Replaced by the streamGate resolver below; the initializer only + // satisfies definite assignment before the executor runs. + }; + const streamGate = new Promise((resolve) => { + releaseStream = resolve; + }); + const killAll = (): void => { + for (const child of children) { + try { + if (child.pid !== undefined) process.kill(-child.pid, "SIGKILL"); + } catch { + // Already dead — the kill is best-effort. + } + try { + child.kill("SIGKILL"); + } catch { + // Already dead — the kill is best-effort. + } + } + }; + return { + children, + releaseStream, + async send(_content: string, optsSend?: { signal?: AbortSignal }) { + const child = spawn("sleep", ["60"], { + stdio: "ignore", + detached: true, + }); + child.unref(); + children.push(child); + return await new Promise((_resolve, reject) => { + const abort = (reason: unknown): void => { + killAll(); + reject(reason instanceof Error ? reason : new Error("aborted")); + }; + if (optsSend?.signal?.aborted === true) { + abort(optsSend.signal.reason); + return; + } + optsSend?.signal?.addEventListener( + "abort", + () => { + abort(defined(optsSend.signal).reason); + }, + { once: true }, + ); + }); + }, + stream: () => + (async function* () { + await streamGate; + yield* []; + })(), + deliver: () => undefined, + close: async () => { + // close() initiates the kill, but a wedged descendant holds the stream + // open and the close itself never completes. + killAll(); + releaseStream(); + if (opts?.wedgeClose === true) { + await new Promise(() => { + // Wedged: the close never completes while the descendant holds on. + }); + } + }, + setSource: () => undefined, + setSources: () => undefined, + history: async () => [], + checkpoints: async () => [], + readAt: async () => [], + blobReader: {}, + }; +} + +type Handles = { + close: (ms?: number) => Promise; + interrupt: () => void; + followup: (message: string) => Promise; +}; + +async function runWithShellChildAgent( + agent: ReturnType, + params?: Partial, +): Promise<{ runPromise: Promise; handles: Handles }> { + const { runSubAgent } = await import("./run.js"); + const cwd = await tmpCwd(); + let handles: Handles | undefined; + const runParams: RunSubAgentParams = { + cwd, + workdirBase: join(cwd, ".ctx"), + permissionGate: testPermissionGate, + provider: { + providerName: "test", + baseURL: "http://localhost", + model: "test-model", + }, + description: "shell-child reap probe", + prompt: "run the thing that hangs in shell collection", + persist: true, + onAgentReady: (h) => { + handles = h; + }, + ...params, + }; + const runPromise = runSubAgent(runParams); + for (let i = 0; i < 500 && handles === undefined; i++) { + await new Promise((resolve) => setTimeout(resolve, 1)); + } + if (handles === undefined) throw new Error("onAgentReady never fired"); + // Wait until the stub send has spawned its live shell child. + for (let i = 0; i < 500 && agent.children.length === 0; i++) { + await new Promise((resolve) => setTimeout(resolve, 1)); + } + if (agent.children.length === 0) { + throw new Error("stub send never spawned its shell child"); + } + return { runPromise, handles }; +} + +describe("CL-7990 shell-child reap: sessions holding a live shell child settle", () => { + test("interrupt settles a session parked behind a live shell child", async () => { + const agent = createShellChildAgent(); + const outcome = await withMockedModuleDuring( + import.meta.resolve("../agent/live-tool-dispatch.js"), + (real: typeof import("../agent/live-tool-dispatch.js")) => ({ + ...real, + createAgentWithLiveToolDispatch: async () => + agent as unknown as Awaited< + ReturnType + >, + }), + async () => { + const { runPromise, handles } = await runWithShellChildAgent(agent, { + persist: true, + }); + handles.interrupt(); + const result = (await runPromise) as { interrupted?: boolean }; + return result; + }, + ); + expect(outcome.interrupted).toBe(true); + for (const child of agent.children) { + await waitForChildExit(child); + } + }); + + test("interrupt mid-wedge with a queued steer settles the run and launches the stash", async () => { + const agent = createShellChildAgent(); + const { createSubAgentSessionStore } = await import("./session-store.js"); + const store = createSubAgentSessionStore(); + const session = store.start({ + description: "d", + agentId: "a", + brief: "b", + retained: true, + }); + const followupCalls: string[] = []; + const outcome = await withMockedModuleDuring( + import.meta.resolve("../agent/live-tool-dispatch.js"), + (real: typeof import("../agent/live-tool-dispatch.js")) => ({ + ...real, + createAgentWithLiveToolDispatch: async () => + agent as unknown as Awaited< + ReturnType + >, + }), + async () => { + const { runPromise, handles } = await runWithShellChildAgent(agent, { + persist: true, + }); + // Wire the live run's handles like agent-fleet's onAgentReady. + store.registerInterrupt(session.id, handles.interrupt); + store.registerFollowup(session.id, async (message: string) => { + followupCalls.push(message); + return "steer reply"; + }); + store.markRunning(session.id); + // The steer arrives while the worker is wedged: it stashes and fires + // the interrupt; the bounded teardown must still settle the run. + expect( + store.sendInputOne(session.id, "steer mid-wedge", { + interrupt: true, + }), + ).toEqual({ ok: true, status: "interrupted" }); + const result = (await runPromise) as RunSubAgentResult; + // agent-fleet settlement: attach the salvage so the stashed steer + // launches instead of stranding on a phantom turn. + store.attachReport( + session.id, + result.report, + result.stopReason !== undefined + ? { stopReason: result.stopReason } + : undefined, + ); + return result; + }, + ); + expect(outcome.interrupted).toBe(true); + // The stash launched synchronously from attachReport: the queued steer + // was delivered, not dropped, and the shell child was reaped. + expect(followupCalls).toEqual(["steer mid-wedge"]); + for (const child of agent.children) { + await waitForChildExit(child); + } + }); + + test( + "close settles a session whose stream is wedged by a shell descendant", + async () => { + const agent = createShellChildAgent({ wedgeClose: true }); + const outcome = await withMockedModuleDuring( + import.meta.resolve("../agent/live-tool-dispatch.js"), + (real: typeof import("../agent/live-tool-dispatch.js")) => ({ + ...real, + createAgentWithLiveToolDispatch: async () => + agent as unknown as Awaited< + ReturnType + >, + }), + async () => { + const { runPromise, handles } = await runWithShellChildAgent(agent, { + persist: false, + }); + const closeError = await handles.close(500).then( + () => undefined, + (err: unknown) => err, + ); + expect(String(defined(closeError))).toMatch(/session close exceeded/); + const settled = await Promise.race([ + runPromise.then( + (result) => ({ state: "resolved" as const, result }), + (error: unknown) => ({ state: "rejected" as const, error }), + ), + new Promise<{ state: "timeout" }>((resolve) => { + setTimeout(() => resolve({ state: "timeout" }), 45_000); + }), + ]); + expect(settled.state).not.toBe("timeout"); + return settled; + }, + ); + expect(outcome.state).not.toBe("timeout"); + for (const child of agent.children) { + await waitForChildExit(child); + } + }, + { timeout: 60_000 }, + ); +}); diff --git a/src/subagent/run.ts b/src/subagent/run.ts index 473fbf22d..0162563cb 100644 --- a/src/subagent/run.ts +++ b/src/subagent/run.ts @@ -230,6 +230,35 @@ export function assertReplySend( const MAX_OUTER_ATTEMPTS = 2; const OUTER_RETRY_DELAY_MS = 500; +/** + * CL-7990 reap deadline for the session-stream drain on teardown. A worker + * parked behind a live shell descendant holds `streamPromise` open; awaiting + * it unbounded wedges interrupt/close forever. Abandoning the drain after + * this deadline settles the session — late stream events just go unsalvaged. + * Calibrated to the shell-guard reap scale (2s), not the 30s close deadline: + * the drain is salvage bookkeeping, not teardown. + */ +const SUBAGENT_STREAM_DRAIN_REAP_MS = 2_000; + +function drainStreamWithReapDeadline( + drain: Promise | undefined, +): Promise { + if (drain === undefined) return Promise.resolve(); + let timer: ReturnType | undefined; + const bounded = Promise.race([ + drain.then( + () => undefined, + () => undefined, + ), + new Promise((resolve) => { + timer = setTimeout(resolve, SUBAGENT_STREAM_DRAIN_REAP_MS); + }), + ]).finally(() => { + if (timer !== undefined) clearTimeout(timer); + }); + return bounded; +} + function sleepUnlessAborted( ms: number, signals: readonly AbortSignal[], @@ -1393,8 +1422,13 @@ async function runSubAgentInner( } }; // Interrupt only fires interruptController — never runController/ - // close, so it cannot hang teardown on a wedged agent.close. + // close, so it cannot hang teardown on a wedged agent.close. Release + // parked shell_collect waiters too: the interrupt settles the turn + // while the session (and its live shell children) stays alive, so a + // worker parked in shell output collection comes back as + // still-running instead of wedging the run past every deadline. const interrupt = (): void => { + backgroundShells.releaseWaiters(); if (!interruptController.signal.aborted) { interruptController.abort( new Error("interrupted by interrupt_agent"), @@ -1593,9 +1627,11 @@ async function runSubAgentInner( // deadline salvage path or rethrow as a bare AbortError. if (thisTurnInterrupt.signal.aborted && !runController.signal.aborted) { interruptedKeepAlive = true; - const abortedCycleText = await cycleRecorder.dispose("cancelled", { - drain: streamPromise, - }); + // No stream drain here: the session stays alive for followup, so a + // stream held open by a live shell descendant would park this + // settlement forever. The recorder already snapshotted the buffer at + // entry; late events are simply unsalvaged. + const abortedCycleText = await cycleRecorder.dispose("cancelled"); const tail = salvageFindingsText( accumulatedProse, lastPartialText, @@ -1620,10 +1656,12 @@ async function runSubAgentInner( // own bookkeeping (lastPartialText) catch late tool.start / inference.done // events before bare-vs-salvage is decided. // Deadline is already known here; a parent cancel is labeled cancelled - // even if the outcome below resolves to rethrow. + // even if the outcome below resolves to rethrow. The drain is reaped + // on a bounded deadline: a stream wedged by a live shell descendant + // settles the run instead of parking it past every deadline. const abortedCycleText = await cycleRecorder.dispose( runController.deadlineHit() ? "deadline" : "cancelled", - { drain: streamPromise }, + { drain: drainStreamWithReapDeadline(streamPromise) }, ); // Deadline always salvages (even with zero output). Cancel after any // tools or assistant prose salvages so the parent keeps partial work; @@ -1686,13 +1724,18 @@ async function runSubAgentInner( // down. if (!persisting) { backgroundShells.disposeAll("sub-agent closed"); - await disposeSubAgentSession({ - signal: runController.signal, - ...(closeOnAbort !== undefined ? { closeOnAbort } : {}), - agent, - ...(streamPromise !== undefined ? { streamPromise } : {}), - posixTools, - }); + // Bounded like close_agent: a session teardown wedged by a live shell + // descendant fails the run instead of parking it forever. + await awaitBoundedTeardown( + disposeSubAgentSession({ + signal: runController.signal, + ...(closeOnAbort !== undefined ? { closeOnAbort } : {}), + agent, + ...(streamPromise !== undefined ? { streamPromise } : {}), + posixTools, + }), + DEFAULT_CLOSE_DEADLINE_MS, + ); } } }