Cancel live workers when a headless exec run ends

Headless exec finally cancels live sub-agents the same way TUI runtime shutdown does, then closes the primary agent and disposes the toolset. cancelAll is fire-and-forget and does not serialize closeOne.

Author: TheGreatAxios

docs/ARCHITECTURE.md

@@ -93,7 +93,7 @@ In TUI chat mode there is no completion gate — the session stays open across t
 - Entry: `corbits exec "prompt"` (alias `corbits run`); `loadConfig` sets `command: "exec"`
 - Streams assistant text deltas to stdout; lifecycle errors to stderr
 - Shares ChatDirector compaction continuation (`requestContinuation` → content-less deliver after compact) so long runs do not stall post-compact
-- Single primary `agent.send(task)` turn; samples run-sink status/error **before** close (close emits `reactor.done` which would clear sticky errors); then closes the agent before draining the stream so the process exits; toolset is always disposed in `finally`
+- Single primary `agent.send(task)` turn; samples run-sink status/error **before** close (close emits `reactor.done` which would clear sticky errors); then closes the agent before draining the stream so the process exits; `finally` cancels live sub-agents (`subAgentSessions.cancelAll("Session closed")`, matching TUI runtime-shutdown), closes the agent, and always disposes the toolset
 - Status: chat sessions rarely emit `reactor.done` before close, so a completed `send()` maps to `done` unless the pre-close run sink holds a real error
 - Used by `scripts/demo.ts` (mode `exec`) and the capability eval suite (`scripts/eval-capability.ts` / `evals/capability/`)
 

src/exec/runner.ts

@@ -45,7 +45,11 @@ import {
 import { detectLanguageServerAvailable } from "../agent/lsp-availability.js";
 import { normalizeToolDefinitionsForProvider } from "../agent/tool-schema-normalize.js";
 import { resolveSessionMode, type SessionMode } from "../config/session-mode.js";
-import { createSubAgentSessionStore, type SubAgentProvider } from "../subagent/index.js";
+import {
+  createSubAgentSessionStore,
+  type SubAgentProvider,
+  type SubAgentSessionStore,
+} from "../subagent/index.js";
 import type {
   ContextStore,
   InferenceSource,
@@ -116,6 +120,33 @@ export function formatCaughtError(err: unknown): string {
   return err instanceof Error ? err.message : String(err);
 }
 
+/**
+ * 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`.
+ */
+export async function disposeExecRuntime(args: {
+  agent: { close: () => Promise<unknown> } | null;
+  toolset: { dispose: () => Promise<unknown> } | null;
+  subAgentSessions: Pick<SubAgentSessionStore, "cancelAll"> | null;
+}): Promise<void> {
+  args.subAgentSessions?.cancelAll("Session closed");
+  if (args.agent !== null) {
+    await args.agent.close().catch((err: unknown) => {
+      logger.debug("agent.close during exec finally failed: {error}", {
+        error: formatCaughtError(err),
+      });
+    });
+  }
+  if (args.toolset !== null) {
+    await args.toolset.dispose().catch((err: unknown) => {
+      logger.debug("toolset.dispose during exec finally failed: {error}", {
+        error: formatCaughtError(err),
+      });
+    });
+  }
+}
+
 /**
  * Exec-primary director overlay. Omit / skywalker keep the product default
  * (`loadSessionChatPrompt` + advertised session tools). Any other closed-fleet
@@ -248,6 +279,7 @@ export async function runExec(config: Config): Promise<ExecResult> {
   let connectedMcp: ConnectedMcpServer[] = [];
   let agent: Agent | null = null;
   let toolset: AgentToolset | null = null;
+  let subAgentSessions: SubAgentSessionStore | null = null;
   let textOut = "";
   let finalized = false;
   let turnsUsed = 0;
@@ -392,7 +424,8 @@ export async function runExec(config: Config): Promise<ExecResult> {
     const liveSubAgentProvider: { current: SubAgentProvider } = {
       current: buildSubAgentProvider(config),
     };
-    const subAgentSessions = createSubAgentSessionStore();
+    const fleetSessions = createSubAgentSessionStore();
+    subAgentSessions = fleetSessions;
     const shellTimeout = shellTimeoutFromSettings(config.settings);
     const toolWatchdog = toolWatchdogFromSettings(config.settings);
     const toolAvailability: ToolAvailability = {
@@ -447,7 +480,7 @@ export async function runExec(config: Config): Promise<ExecResult> {
         ? {
             subAgent: {
               provider: () => liveSubAgentProvider.current,
-              sessions: subAgentSessions,
+              sessions: fleetSessions,
               getWorkdirBase: () => sessionDir(config.cwd, sessionId),
               onProgress: () => undefined,
               ...(config.settings !== undefined ? { settings: () => config.settings! } : {}),
@@ -888,21 +921,7 @@ export async function runExec(config: Config): Promise<ExecResult> {
       model: config.model,
     };
   } finally {
-    if (agent !== null) {
-      await agent.close().catch((err: unknown) => {
-        logger.debug("agent.close during exec finally failed: {error}", {
-          error: formatCaughtError(err),
-        });
-      });
-    }
-    // Match TUI: always dispose toolset (MCP clients + posix/plugin resources).
-    if (toolset !== null) {
-      await toolset.dispose().catch((err: unknown) => {
-        logger.debug("toolset.dispose during exec finally failed: {error}", {
-          error: formatCaughtError(err),
-        });
-      });
-    }
+    await disposeExecRuntime({ agent, toolset, subAgentSessions });
   }
 }
 

tests/unit/exec/runner.test.ts

@@ -1,11 +1,13 @@
 import { describe, expect, test } from "bun:test";
 import type { Config } from "../../../src/config/index.js";
 import {
+  disposeExecRuntime,
   formatCaughtError,
   resolveExecDirectorOverlay,
   runExec,
 } from "../../../src/exec/runner.js";
 import { BUILD_TOOLS } from "../../../src/agent/directors/tool-sets.js";
+import { createSubAgentSessionStore } from "../../../src/subagent/session-store.js";
 
 function bareConfig(task: string): Config {
   // Minimal unconfigured-shaped object is not enough — runExec only needs
@@ -54,6 +56,36 @@ describe("runExec", () => {
   });
 });
 
+describe("disposeExecRuntime", () => {
+  test("cancels fire-and-forget workers when exec finishes", async () => {
+    const store = createSubAgentSessionStore();
+    const worker = store.start({ description: "bg", agentId: "w", brief: "b" });
+    let aborted = 0;
+    store.registerCancel(worker.id, () => {
+      aborted += 1;
+    });
+
+    const calls: string[] = [];
+    await disposeExecRuntime({
+      agent: {
+        close: async () => {
+          calls.push("agent");
+        },
+      },
+      toolset: {
+        dispose: async () => {
+          calls.push("toolset");
+        },
+      },
+      subAgentSessions: store,
+    });
+
+    expect(aborted).toBe(1);
+    expect(store.get(worker.id)?.status).toBe("cancelled");
+    expect(calls).toEqual(["agent", "toolset"]);
+  });
+});
+
 describe("resolveExecDirectorOverlay", () => {
   test("builder exec primary does not mount task", () => {
     const overlay = resolveExecDirectorOverlay("builder");