Make director task a spawn plus wait wrapper (#685)

* Make director task() a spawn_agent plus wait_agents wrapper

Closed-director task() was a second full spawn engine. When a session
store is present it now starts the worker through spawn_agent and blocks
on wait_agents, so one mailbox owns completion. Custom AgentProfile
lookup still uses the legacy await-run path.

* Preserve task cancel, auth, and abort contracts through the fleet wrapper

Director task() now routes through spawn_agent + wait_agents, which was
misclassifying AbortError as failed:aborted, dropping Re-authenticate
auth wording, and leaving the child running when the parent tool aborted.
Map those outcomes back to the legacy fused-task parent contract and
prettier the inherited agent-progress tip.

Author: TheGreatAxios

src/agent/tools.ts

@@ -282,6 +282,7 @@ export async function createAgentToolset(args: AgentToolsetArgs): Promise<AgentT
   const orchestratorTools: AgentTool[] = [];
   if (subAgentsEnabled && args.subAgent !== undefined) {
     const sa = args.subAgent;
+    const fleetRecords = sa.sessions !== undefined ? createFleetRecords() : undefined;
     orchestratorTools.push(
       createTaskTool({
         cwd,
@@ -302,6 +303,7 @@ export async function createAgentToolset(args: AgentToolsetArgs): Promise<AgentT
         ...(args.getBlobReader !== undefined ? { getBlobReader: args.getBlobReader } : {}),
         ...(sa.useWorktree !== undefined ? { useWorktree: sa.useWorktree } : {}),
         ...(args.telemetry !== undefined ? { telemetry: args.telemetry } : {}),
+        ...(fleetRecords !== undefined ? { fleetRecords } : {}),
       }),
     );
     if (sa.profiles !== undefined) {
@@ -321,9 +323,8 @@ export async function createAgentToolset(args: AgentToolsetArgs): Promise<AgentT
     // Mirror nested runSubAgent's orchestrator fleet mount (run.ts), but
     // reuse the existing TUI/exec session store — do not allocate a private
     // store only for these verbs. spawnAllowlist stays unwired on primary.
-    if (sa.sessions !== undefined) {
+    if (sa.sessions !== undefined && fleetRecords !== undefined) {
       const fleetSessions = sa.sessions;
-      const fleetRecords = createFleetRecords();
       const fleetDeps = {
         permissionGate,
         inheritMcpTools: () => inheritedMcpTools,

src/subagent/agent-fleet.ts

@@ -63,7 +63,7 @@ import type { Settings } from "../config/settings.js";
 import { resolveEffortForRole } from "../provider/reasoning-effort.js";
 import { isCodexProviderName } from "../config/codex-providers.js";
 import { buildDispatchBrief, type TaskIntent } from "./report.js";
-import type { SubAgentSessionStore } from "./session-store.js";
+import { DEFAULT_CANCEL_REASON, type SubAgentSessionStore } from "./session-store.js";
 import type {
   NestedDispatchDeps,
   RunSubAgentParams,
@@ -75,6 +75,8 @@ import { cleanupSubAgentWorktree, createSubAgentWorktree, WorktreeError } from "
 import { NOOP_TELEMETRY, type Telemetry } from "../telemetry/index.js";
 import { classifyAgentName } from "../telemetry/classify.js";
 import type { DirectorPackage } from "../agent/directors/types.js";
+import { formatSubAgentTaskAuthFailureMessage } from "./inference-auth-failure.js";
+import { isSubAgentCancelError } from "./dispose.js";
 
 const log = getLogger([LOG_NAMESPACE_ROOT, "subagent", "agent-fleet"]);
 
@@ -363,6 +365,8 @@ export type AgentFleetDeps = SubAgentSandboxDeps & {
   useWorktree?: boolean;
   /** Optional wall-clock budget (ms) forwarded to runSubAgent. */
   deadlineMs?: number;
+  /** When false, tear the worker down on completion (task wrapper). Default true. */
+  persist?: boolean;
   settings?: Settings | (() => Settings | undefined);
   catalog?: readonly ProviderCatalogEntry[] | (() => readonly ProviderCatalogEntry[]);
   onEvent?: (event: ReactorEmittedEvent) => void;
@@ -660,7 +664,7 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool {
         // stays alive for followup (agentRetained / interrupt keep-alive) —
         // matching run.ts's persisting gate so followup_task does not hit a
         // removed cwd.
-        persist: true,
+        persist: deps.persist !== false,
         onAgentReady: ({ close, interrupt, followup, deliver }) => {
           deps.sessions.registerClose(session.id, async (deadlineMs) => {
             try {
@@ -716,11 +720,24 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool {
         })
         .catch((err) => {
           // Always terminalize fleetRecords — including pre-progress cancel that
-          // rethrows with no salvage — so wait_agents does not hang. fail()
-          // no-ops when cancel already flipped the strip status.
-          const message = err instanceof Error ? err.message : String(err);
-          deps.fleetRecords.reject(session.id, message);
-          deps.sessions.fail(session.id, message);
+          // rethrows with no salvage — so wait_agents does not hang. Prefer
+          // cancel semantics over fail when the strip already cancelled or the
+          // throw is an AbortError (legacy task() parent contract).
+          const alreadyCancelled = deps.sessions.get(session.id)?.status === "cancelled";
+          if (alreadyCancelled || isSubAgentCancelError(err, childCtl.signal)) {
+            if (!alreadyCancelled) {
+              deps.sessions.cancel(session.id, DEFAULT_CANCEL_REASON);
+            }
+            const message = err instanceof Error ? err.message : String(err);
+            deps.fleetRecords.reject(session.id, message);
+            return;
+          }
+          // Auth failures keep the actionable Re-authenticate wording that
+          // task()'s fused path surfaces via formatSubAgentTaskAuthFailureMessage.
+          const authMessage = formatSubAgentTaskAuthFailureMessage(description, err);
+          const failReason = authMessage ?? (err instanceof Error ? err.message : String(err));
+          deps.fleetRecords.reject(session.id, failReason);
+          deps.sessions.fail(session.id, failReason);
         })
         .finally(() => {
           telemetry.capture("subagent_end", {

src/subagent/run.ts

@@ -521,6 +521,8 @@ export async function runSubAgent(params: RunSubAgentParams): Promise<RunSubAgen
         );
       }
       const nd = params.nestedDispatch;
+      const fleetSessions = nd.sessions ?? createSubAgentSessionStore();
+      const fleetRecords = createFleetRecords();
       tools = [
         ...tools,
         createTaskTool({
@@ -542,7 +544,8 @@ export async function runSubAgent(params: RunSubAgentParams): Promise<RunSubAgen
           telemetry: liveTelemetry,
           ...(nd.onEvent !== undefined ? { onEvent: nd.onEvent } : {}),
           ...(nd.onProgress !== undefined ? { onProgress: nd.onProgress } : {}),
-          ...(nd.sessions !== undefined ? { sessions: nd.sessions } : {}),
+          sessions: fleetSessions,
+          fleetRecords,
           ...(nd.settings !== undefined ? { settings: nd.settings } : {}),
           ...(nd.catalog !== undefined ? { catalog: nd.catalog } : {}),
           ...(nd.profiles !== undefined ? { profiles: nd.profiles } : {}),
@@ -569,16 +572,9 @@ export async function runSubAgent(params: RunSubAgentParams): Promise<RunSubAgen
         createReadAgentTraceTool(nd.getWorkdirBase, {
           actorId: params.id,
           tier,
-          getNodes: () => nd.sessions?.list() ?? [],
+          getNodes: () => fleetSessions.list(),
         }),
       ];
-      // spawn_agent/wait_agents need a session store as their mailbox;
-      // reuse the orchestrator's if it has one, else give this install its
-      // own. fleetRecords holds terminal results the session store's
-      // display cap would otherwise evict before wait_agents collects them
-      // (see agent-fleet.ts).
-      const fleetSessions = nd.sessions ?? createSubAgentSessionStore();
-      const fleetRecords = createFleetRecords();
       const lifecycleAuthority = {
         actorId: params.id,
         tier,