Fail parent dispose when persist workers leave children

Author: TheGreatAxios

src/agent/fleet-verbs-mount.test.ts

@@ -85,6 +85,80 @@ describe("primary fleet verb mount", () => {
     await expect(toolset.dispose()).rejects.toThrow(/still live after 2000ms reap/);
   });
 
+  test("createAgentToolset dispose rejects when a retained completed persist worker leaves children", async () => {
+    const cwd = mkdtempSync(join(tmpdir(), "corbits-fleet-mount-"));
+    const { createAgentToolset } = await import("./tools.js");
+    const permissionGate = {
+      check: async () => ({ allowed: true }),
+      getSkipPermissions: () => false,
+    } as never;
+    const sessions = createSubAgentSessionStore();
+    const worker = sessions.start({
+      description: "d",
+      agentId: "a",
+      brief: "b",
+      retained: true,
+    });
+    sessions.registerClose(worker.id, async () => {
+      throw new Error("1 shell child process still live after 2000ms reap");
+    });
+    sessions.complete(worker.id, "done", { agentRetained: true });
+
+    const toolset = await createAgentToolset({
+      cwd,
+      permissionGate,
+      onOperatorGate: async () => ({ kind: "option", index: 0 }),
+      subAgent: {
+        provider: {
+          providerName: "test",
+          baseURL: "http://127.0.0.1:0",
+          model: "test-model",
+        },
+        getWorkdirBase: () => cwd,
+        sessions,
+      },
+    });
+
+    await expect(toolset.dispose()).rejects.toThrow(/still live after 2000ms reap/);
+  });
+
+  test("createAgentToolset dispose rejects when a retained running persist worker leaves children", async () => {
+    const cwd = mkdtempSync(join(tmpdir(), "corbits-fleet-mount-"));
+    const { createAgentToolset } = await import("./tools.js");
+    const permissionGate = {
+      check: async () => ({ allowed: true }),
+      getSkipPermissions: () => false,
+    } as never;
+    const sessions = createSubAgentSessionStore();
+    const worker = sessions.start({
+      description: "d",
+      agentId: "a",
+      brief: "b",
+      retained: true,
+    });
+    sessions.markRunning(worker.id);
+    sessions.registerClose(worker.id, async () => {
+      throw new Error("1 shell child process still live after 2000ms reap");
+    });
+
+    const toolset = await createAgentToolset({
+      cwd,
+      permissionGate,
+      onOperatorGate: async () => ({ kind: "option", index: 0 }),
+      subAgent: {
+        provider: {
+          providerName: "test",
+          baseURL: "http://127.0.0.1:0",
+          model: "test-model",
+        },
+        getWorkdirBase: () => cwd,
+        sessions,
+      },
+    });
+
+    await expect(toolset.dispose()).rejects.toThrow(/still live after 2000ms reap/);
+  });
+
   test("createAgentToolset omits fleet verbs when subAgent is not set", async () => {
     const cwd = mkdtempSync(join(tmpdir(), "corbits-fleet-mount-"));
     const { createAgentToolset } = await import("./tools.js");

src/agent/tools.ts

@@ -945,7 +945,7 @@ export async function createAgentToolset(args: AgentToolsetArgs): Promise<AgentT
     disposal = (async () => {
       const fleetSessions = fleetSessionsForDispose;
       if (fleetSessions !== undefined) {
-        fleetSessions.cancelAll("parent session closed");
+        await fleetSessions.cancelAll("parent session closed");
         for (const session of [...fleetSessions.list()].reverse()) {
           await fleetSessions.closeOne(session.id, DEFAULT_CLOSE_DEADLINE_MS);
         }

src/exec/runner.ts

@@ -127,9 +127,10 @@ export function execUserFailureMessage(
 
 /**
  * Headless analogue of TUI `runtime-shutdown`: abort live workers, then close
- * the primary agent and dispose the toolset. `cancelAll` is fire-and-forget —
- * it does not serialize `closeOne`. Once-only per runtime object so the send
- * path, `finally`, and signal host cannot double-dispose.
+ * the primary agent and dispose the toolset. `cancelAll` is awaited so a
+ * leftover-child throw is visible; hang-forever close is still deadline-bounded.
+ * Once-only per runtime object so the send path, `finally`, and signal host
+ * cannot double-dispose.
  */
 const execDisposeInFlight = new WeakMap<object, Promise<void>>();
 
@@ -163,7 +164,7 @@ async function runExecDispose(args: {
 }): Promise<void> {
   const failures: unknown[] = [];
   try {
-    args.subAgentSessions?.cancelAll("Session closed");
+    await args.subAgentSessions?.cancelAll("Session closed");
   } catch (err) {
     failures.push(err);
   }

src/subagent/lifecycle-tools.test.ts

@@ -90,6 +90,45 @@ describe("close_agent", () => {
     expect(Date.now() - started).toBeLessThan(500);
     expect(childStatus).toBe("shutdown");
   });
+
+  test("closes remaining siblings after a leftover-child throw, then fails", async () => {
+    const sessions = createSubAgentSessionStore();
+    const parent = sessions.start({ description: "parent", agentId: "a", brief: "b" });
+    const leftover = sessions.start({
+      description: "leftover",
+      agentId: "a",
+      brief: "b",
+      parentSessionId: parent.id,
+    });
+    const sibling = sessions.start({
+      description: "sibling",
+      agentId: "a",
+      brief: "b",
+      parentSessionId: parent.id,
+    });
+    const closedOrder: string[] = [];
+    sessions.registerClose(leftover.id, async () => {
+      closedOrder.push(leftover.id);
+      throw new Error("1 shell child process still live after 2000ms reap");
+    });
+    sessions.registerClose(sibling.id, async () => {
+      closedOrder.push(sibling.id);
+    });
+    sessions.registerClose(parent.id, async () => {
+      closedOrder.push(parent.id);
+    });
+
+    const closeAgent = createCloseAgentTool({
+      sessions,
+      fleetRecords: createFleetMailbox(sessions),
+    });
+    await expect(callTool(closeAgent, { target: parent.id })).rejects.toThrow(
+      /still live after 2000ms reap/,
+    );
+    expect(closedOrder).toContain(leftover.id);
+    expect(closedOrder).toContain(sibling.id);
+    expect(closedOrder).toContain(parent.id);
+  });
 });
 
 describe("resume_agent", () => {

src/subagent/lifecycle-tools.ts

@@ -184,14 +184,28 @@ export function createCloseAgentTool(deps: CloseAgentToolDeps): AgentTool {
         .map((s) => ({ id: s.id, parentSessionId: s.parentSessionId }));
       const order = descendantsClosingOrder(nodes, target);
       const closed: { agent_id: string; status: AgentLifecycleStatus }[] = [];
+      const failures: unknown[] = [];
       for (const id of order) {
         // Terminalize the wait mailbox before teardown. closeOne flips strip
         // status to "cancelled", which kills the soft-interrupt fallback that
         // still requires status === "running" — without this, in-flight
         // wait_agents hangs until timeout.
         deps.fleetRecords.interrupt(id);
-        const status = await deps.sessions.closeOne(id, DEFAULT_CLOSE_DEADLINE_MS);
-        closed.push({ agent_id: id, status });
+        try {
+          const status = await deps.sessions.closeOne(id, DEFAULT_CLOSE_DEADLINE_MS);
+          closed.push({ agent_id: id, status });
+        } catch (err: unknown) {
+          failures.push(err);
+          const after = deps.sessions.get(id);
+          closed.push({
+            agent_id: id,
+            status: after === undefined ? "not_found" : after.lifecycleStatus,
+          });
+        }
+      }
+      if (failures.length === 1) throw failures[0];
+      if (failures.length > 1) {
+        throw new AggregateError(failures, "close_agent leftover dispose failed");
       }
       const own = closed.find((c) => c.agent_id === target);
       return lifecycleResult(

src/subagent/retain-salvage.test.ts

@@ -24,7 +24,7 @@ describe("retained session lifecycle", () => {
     expect(outcome.ok).toBe(false);
   });
 
-  test("cancelAll does not close retained completed sessions", () => {
+  test("cancelAll does not close retained completed sessions", async () => {
     const store = createSubAgentSessionStore({ maxCompleted: 5 });
     const s = store.start({
       description: "worker",
@@ -37,7 +37,7 @@ describe("retained session lifecycle", () => {
       closed = true;
     });
     store.complete(s.id, "done");
-    const cancelled = store.cancelAll("parent stop");
+    const cancelled = await store.cancelAll("parent stop");
     console.log("cancelAll returned:", cancelled, "| close invoked:", closed);
     expect(closed).toBe(true);
   });
@@ -67,7 +67,7 @@ describe("retained session lifecycle", () => {
     expect(store.list().length).toBeLessThanOrEqual(3);
   });
 
-  test("a genuinely retained clean completion IS resumable, and cancelAll releases it", () => {
+  test("a genuinely retained clean completion IS resumable, and cancelAll releases it", async () => {
     const store = createSubAgentSessionStore({ maxCompleted: 5 });
     const s = store.start({
       description: "worker",
@@ -86,7 +86,7 @@ describe("retained session lifecycle", () => {
     store.registerFollowup(s.id, async () => "next");
     expect(store.resumeOne(s.id, "more").ok).toBe(true);
     expect(closed).toBe(false);
-    store.cancelAll("parent stop");
+    await store.cancelAll("parent stop");
     expect(closed).toBe(true);
   });
 

src/subagent/session-store.test.ts

@@ -534,6 +534,41 @@ describe("CL-6943 reusable worker sessions", () => {
     expect(store.resumeOne("missing", "more")).toEqual({ ok: false, status: "not_found" });
   });
 
+  test("cancelAll then closeOne does not swallow a leftover-child throw as shutdown success", async () => {
+    const store = createSubAgentSessionStore();
+    const session = store.start({
+      description: "d",
+      agentId: "a",
+      brief: "b",
+      retained: true,
+    });
+    store.registerClose(session.id, async () => {
+      throw new Error("1 shell child process still live after 2000ms reap");
+    });
+    store.complete(session.id, "done", { agentRetained: true });
+
+    const leftover = /still live after 2000ms reap/;
+    let cancelThrew = false;
+    try {
+      await store.cancelAll("parent stop");
+    } catch (err) {
+      expect(err).toBeInstanceOf(Error);
+      expect((err as Error).message).toMatch(leftover);
+      cancelThrew = true;
+    }
+    let closeThrew = false;
+    let closeStatus: string | undefined;
+    try {
+      closeStatus = await store.closeOne(session.id, 1000);
+    } catch (err) {
+      expect(err).toBeInstanceOf(Error);
+      expect((err as Error).message).toMatch(leftover);
+      closeThrew = true;
+    }
+    expect(cancelThrew || closeThrew).toBe(true);
+    expect(cancelThrew === false && closeThrew === false && closeStatus === "shutdown").toBe(false);
+  });
+
   test("closeOne is bounded by its deadline when the registered close hangs forever", async () => {
     const store = createSubAgentSessionStore();
     const session = store.start({ description: "d", agentId: "a", brief: "b", retained: true });

src/subagent/session-store.ts

@@ -22,6 +22,26 @@ import type { AdmissionQueue, AdmissionStatus } from "./admission.js";
 
 const log = getLogger([LOG_NAMESPACE_ROOT, "subagent", "session-store"]);
 
+async function invokeCloseBounded(
+  close: (deadlineMs?: number) => Promise<void>,
+  deadlineMs: number,
+): Promise<void> {
+  let closeError: unknown;
+  await Promise.race([
+    close(deadlineMs).then(
+      () => undefined,
+      (err: unknown) => {
+        closeError = err;
+        log.warn("session close raced deadline: {error}", {
+          error: err instanceof Error ? err.message : String(err),
+        });
+      },
+    ),
+    new Promise<void>((resolve) => setTimeout(resolve, deadlineMs)),
+  ]);
+  if (closeError !== undefined) throw closeError;
+}
+
 export type SubAgentSessionStatus = "running" | "done" | "failed" | "cancelled";
 
 /**
@@ -194,8 +214,10 @@ export interface SubAgentSessionStore {
   // Abort a running session and mark it cancelled. Returns true when a running
   // session was cancelled; false if missing or already terminal.
   cancel(id: string, reason?: string): boolean;
-  // Cancel every running session. Returns the ids that transitioned.
-  cancelAll(reason?: string): string[];
+  // Cancel every running session. Closes retained workers with the same
+  // deadline race as closeOne: leftover-child throws reject, hang-forever
+  // resolves without throwing. Returns the ids that transitioned to cancelled.
+  cancelAll(reason?: string): Promise<string[]>;
   // CL-6943: flips a "pending_init" session to "running" once its agent
   // object actually exists. No-op on an unknown id or one already past init.
   markRunning(id: string): void;
@@ -1219,18 +1241,11 @@ export function createSubAgentSessionStore(
       // close that does not honor its own deadline argument — a wedged
       // descendant must not hang the whole close_agent call.
       let closeError: unknown;
-      await Promise.race([
-        close(deadlineMs).then(
-          () => undefined,
-          (err: unknown) => {
-            closeError = err;
-            log.warn("session close raced deadline: {error}", {
-              error: err instanceof Error ? err.message : String(err),
-            });
-          },
-        ),
-        new Promise<void>((resolve) => setTimeout(resolve, deadlineMs)),
-      ]);
+      try {
+        await invokeCloseBounded(close, deadlineMs);
+      } catch (err: unknown) {
+        closeError = err;
+      }
       if (keepFailed) {
         // fail() already stamped failed; invoke leftover teardown without
         // rewriting that to shutdown.
@@ -1465,7 +1480,7 @@ export function createSubAgentSessionStore(
       return cancelSession(id, reason);
     },
 
-    cancelAll(reason = DEFAULT_CANCEL_REASON): string[] {
+    async cancelAll(reason = DEFAULT_CANCEL_REASON): Promise<string[]> {
       // Snapshot before cancelSession: markCancelled clears retained, and a
       // resumed retained worker is strip-live so the first loop would otherwise
       // skip the close-handle pass (CL-7001).
@@ -1477,10 +1492,17 @@ export function createSubAgentSessionStore(
       for (const session of running) {
         if (cancelSession(session.id, reason)) cancelled.push(session.id);
       }
+      const pendingCloses: Promise<void>[] = [];
       for (const id of retainedIds) {
         const session = sessions.get(id);
         if (session === undefined || session.lifecycle.state === "shutdown") continue;
-        releaseHandles(id);
+        const close = closeHandles.get(id);
+        cancelAskInternal(id, "session handles released");
+        closeHandles.delete(id);
+        cancelHandles.delete(id);
+        interruptHandles.delete(id);
+        followupHandles.delete(id);
+        deliverHandles.delete(id);
         mutate(id, (s) => {
           s.lifecycle = {
             state: "shutdown",
@@ -1489,6 +1511,15 @@ export function createSubAgentSessionStore(
           };
           s.retained = false;
         });
+        if (close !== undefined) {
+          pendingCloses.push(invokeCloseBounded(close, DEFAULT_CLOSE_DEADLINE_MS));
+        }
+      }
+      const results = await Promise.allSettled(pendingCloses);
+      const failures = results.flatMap((r) => (r.status === "rejected" ? [r.reason] : []));
+      if (failures.length === 1) throw failures[0];
+      if (failures.length > 1) {
+        throw new AggregateError(failures, "session cancelAll close failed");
       }
       return cancelled;
     },

src/tui/runner/exit.ts

@@ -440,13 +440,16 @@ export async function createRunLifecycle(
     // and that path never moved to OpenTUI.
     services.emitter.emit("session.clear");
     // Cancel live workers before rotation so /clear does not leave orphaned
-    // child reactors burning tokens under the old session id.
-    services.subAgentSessions.cancelAll("Session cleared");
+    // child reactors burning tokens under the old session id. Start immediately
+    // (do not wait on the op queue) and await on the rotation path so a leftover
+    // throw is not an unhandled rejection.
+    const cancelledWorkers = services.subAgentSessions.cancelAll("Session cleared");
     // Backend rotation is always enqueued regardless of contention; the queue
     // serialises it behind any in-progress op. Sub-agents nest under the new
     // session automatically because getWorkdirBase reads the live sessionId.
     void enqueueOp(async () => {
       try {
+        await cancelledWorkers;
         // Tear the old agent down and dispose the recorder before workdir is
         // repointed: the pump can deliver stray deltas until the stream
         // settles, and a dead cycle's partial must land in the session that