Merge pull request #852 from corbitsdev/cl-7540-keep-driving-open-tasks-after-the-fleet-goes-dry

Resume the parent when the fleet goes dry with open tasks

Author: TheGreatAxios

docs/ARCHITECTURE.md

@@ -112,7 +112,10 @@ In TUI chat mode there is no completion gate — the session stays open across t
 
 Two directors, selected by role:
 
-- **ChatDirector** (interactive, `src/agent/director.ts`) — Extends `DefaultDirector` with task list tracking, workflow nudges, LSP auto-activation, and multi-turn chat semantics. It never terminates the session: operator declines are surfaced as replies and the reactor stays alive for the next message. Auto mode is toggled by CLI flags (`--auto` / `--no-auto`); there is currently no in-session key to toggle it (default on; constrained envelope — workspace writes and unconstrained shell auto-allow; installs, recursive rm, force/uncontained worktree changes, sensitive-path and opaque-wrapper shell still ask; contained non-force `git worktree add`/`remove`/`prune` and `list` auto-allow; shell file-mutation denied). It is not a separate edit/plan mode.
+- **ChatDirector** (interactive, `src/agent/director.ts`) — Extends `DefaultDirector` with task list tracking, workflow nudges, LSP auto-activation, and multi-turn chat semantics. It never terminates the session: operator declines are surfaced as replies and the reactor stays alive for the next message. Yielding while a live fleet is running is allowed (idle-with-fleet); the open-task nudge does not rewrite that wait/reply. When the fleet goes dry with tasks still todo/doing, the TUI runtime re-enters the parent with collected worker reports rather than settling idle.
+
+  Auto mode is toggled by CLI flags (`--auto` / `--no-auto`); there is currently no in-session key to toggle it (default on; constrained envelope — workspace writes and unconstrained shell auto-allow; installs, recursive rm, force/uncontained worktree changes, sensitive-path and opaque-wrapper shell still ask; contained non-force `git worktree add`/`remove`/`prune` and `list` auto-allow; shell file-mutation denied). It is not a separate edit/plan mode.
+
 - **SubAgentDirector** (delegated work, `src/subagent/index.ts`) — Drives a dispatched worker until a turn arrives with no tool calls, then replies with the final assistant text and ends the run. A tool-less turn **after tools** completes only with the four-heading envelope (Summary, Findings, Blockers, Paths); a missing envelope nudges once (**incomplete-report**) and a second tool-less turn still without the envelope salvages as **incomplete-report-stop**. Explore/read-only workers that used tools then replied with findings remain normal completes; `requireEvidence` (off by default, set per director) additionally requires at least one read before a tool-less spawn-only reply can complete. Reads done through `run_shell` count as evidence too — `src/subagent/shell-evidence.ts` classifies shell reads (`cat`, `grep`, `sed` without `-i`, …) over the same subject expansion the auto-shell policy uses — but there is no corresponding shell-write evidence or file-write requirement: a run that never touches a file still completes normally once it replies with the envelope. There is no turn budget. Operator/parent cancel after any progress returns a **cancelled** salvage report (partial findings + tool activity) instead of a bare cancel string; cancel before progress still surfaces as cancelled-by-operator. There is no repetition/no-progress/never-acted/never-edited hard stop and no fingerprint-based re-dispatch block — a genuinely stuck worker runs until it completes, stalls, hits an opt-in wall-clock deadline, or is cancelled.
   `spawn_agent` starts each worker and records it in the caller's fleet mailbox; `wait_agents` collects terminal reports from that mailbox. Wait JSON includes `stop_reason` from the session when present so a salvage that is wait-`done` is not mistaken for a clean complete, and so parent-initiated interrupt (`interrupted`) is not mistaken for operator-cancel (`cancelled`). Deadline salvage prepends an advisory parent hint suggesting continuation plus a longer deadline if more wall-clock time is warranted. Failed and incomplete-report salvage tell the parent to diagnose from the report or error and MAY spawn one successor with a changed brief. A parent-initiated interrupt is a resumable pause: wait unblocks with `stop_reason: interrupted` (often while the session is still running and has no report); the parent should `resume_agent` or re-wait, and must not spawn a successor against a still-live worker. Successor only if that session is no longer resumable. Operator-cancelled salvage asks the parent to synthesize Findings and Paths and wait for the operator instead of auto-starting another specialist. Identical re-dispatch of the same brief stays refused at the prompt / spawn-handoff layer; there is no fingerprint-based re-dispatch hard-block. Deadline hints are advisory only — an identical re-dispatch is still admitted at runtime. Parent hints are prepended on salvage reports returned to the parent. The runtime does not auto-spawn successors.
 

docs/TUI.md

@@ -258,7 +258,10 @@ lane finishes
 (`N done · nothing running`; failed and cancelled counts appear only
 when non-zero, e.g. `N done, M failed, K cancelled · nothing running`).
 Per-lane `done — summary` walls and live `dispatched` re-announcements
-are never printed.
+are never printed. That dry-fleet line stays operator-facing. If tasks
+are still todo/doing, the runtime re-enters the parent with collected
+reports as a system continuation — it does not paint the report wall as
+a user message.
 
 `src/subagent/fleet-report.ts` is pure: it reads the same fleet-agent session
 store and the same `agentProgress()` stall definition. Store changes drive it;
@@ -582,7 +585,9 @@ there is no parent tool left to steer — while Alt+Enter follow-ups keep
 waiting for true session-idle. A steer still pending when the hold engages
 sends at once (the parent it was steering has stopped), and the last lane
 terminalizing releases the hold, drains follow-ups, and returns the session
-to idle.
+to idle — unless todo/doing tasks remain, in which case a system
+continuation starts before the fleet-0 event so the run stays busy and
+follow-ups wait one more turn.
 
 Interrupting (Ctrl+C) never discards a queued or steered message. It used to
 — the transcript literally said `interrupt — discarded N pending`, and an

src/agent/director.ts

@@ -377,6 +377,12 @@ export interface ChatDirectorOptions {
   getProviderId?: (() => string | undefined) | undefined;
   /** Explicit retry policy; when set, skips the default Corbits policy. */
   retryPolicy?: RetryPolicy | undefined;
+  /**
+   * Live `status === "running"` fleet-lane count. When greater than zero the
+   * director allows a terminal wait/reply with open tasks (idle-with-fleet).
+   * Omitted or 0 keeps the open-task nudge. Exec omits this.
+   */
+  getLiveFleetCount?: (() => number) | undefined;
 }
 
 // The constructor takes the resolved ModelFamilyPolicy rather than the raw
@@ -415,6 +421,7 @@ class ChatDirectorImpl extends DefaultDirector {
   private readonly compaction: CompactionGovernor;
   private readonly modelFamilyPolicy: ModelFamilyPolicy;
   private readonly retryPolicy: RetryPolicy;
+  private readonly getLiveFleetCount: (() => number) | undefined;
   // Consecutive assistant turns that contain tool calls and no text. Reset on
   // any turn with text and on every fresh user message — a weak model that
   // spins in place on one thread of tool calls still converges to the
@@ -448,6 +455,7 @@ class ChatDirectorImpl extends DefaultDirector {
     this.modelFamilyPolicy =
       options.modelFamilyPolicy ?? resolveModelFamilyPolicy({ providerName: "" });
     this.retryPolicy = options.retryPolicy ?? createCorbitsRetryPolicy();
+    this.getLiveFleetCount = options.getLiveFleetCount;
   }
 
   setWorkflowCoordinator(coordinator: WorkflowCoordinator | undefined): void {
@@ -883,6 +891,9 @@ class ChatDirectorImpl extends DefaultDirector {
     if (!atWorkflowGate && hasActiveTasks(this.tasks)) {
       const hasTerminal = baseActions.some((a) => a.type === "wait" || a.type === "reply");
       if (hasTerminal) {
+        if ((this.getLiveFleetCount?.() ?? 0) > 0) {
+          return base;
+        }
         if (this.idleTerminationNudges < MAX_OPEN_TASK_NUDGES) {
           this.idleTerminationNudges++;
           const passThrough = baseActions.filter(

src/agent/directors/skywalker/package.test.ts

@@ -117,6 +117,8 @@ describe("skywalkerPackage", () => {
     expect(p).not.toContain("task()");
     expect(p).toContain('mode="all"');
     expect(p).toContain("uncollected spawns");
+    expect(p).toContain("When the fleet goes dry the runtime re-enters with collected reports");
+    expect(p).toContain("do not tight-loop wait_agents");
     expect(p).not.toContain("Present the plan when the change is large or ambiguous");
   });
 

src/agent/directors/skywalker/package.ts

@@ -16,7 +16,7 @@ You do not do the specialists' jobs by default. For tiny bounded product edits,
 
 Do not run long-blocking jobs on the parent (evals, full test suites, long installs, long-running implementation). Dispatch intern (mechanical shell), tester (suite / repro), or builder (substantial code). Path tools (write_file/edit_file/delete_file) are the DIY surface; shell file-writes stay denied.
 
-Idle-orchestrator: fire one or more spawn_agent calls in a turn — each returns immediately with an agent_id and does not hold the parent. Then **reply to the operator** with who is running and what happens next before you block. Prefer ending that turn (or calling wait_agents with a short timeout_ms) so Enter can land; do not immediately fuse into a long wait_agents right after spawn. wait_agents later on the targets you need (or omit targets to wait on this session's own uncollected spawns — never a sibling's). list_agents shows that same fleet without blocking. Use mode="all" when you need every target to finish; interrupt_agent unblocks wait_agents immediately. A timeout means still running — do not tight-loop wait_agents hoping for a different answer. Enter mid-run delivers at the next parent tool.boundary — a long parent run_shell or awaiting wait_agents holds those steers. A bare spawn_agent does not.
+Idle-orchestrator: fire one or more spawn_agent calls in a turn — each returns immediately with an agent_id and does not hold the parent. Then **reply to the operator** with who is running and what happens next before you block. Prefer ending that turn (or calling wait_agents with a short timeout_ms) so Enter can land; do not immediately fuse into a long wait_agents right after spawn. wait_agents later on the targets you need (or omit targets to wait on this session's own uncollected spawns — never a sibling's). list_agents shows that same fleet without blocking. Use mode="all" when you need every target to finish; interrupt_agent unblocks wait_agents immediately. A timeout means still running — do not tight-loop wait_agents hoping for a different answer. Enter mid-run delivers at the next parent tool.boundary — a long parent run_shell or awaiting wait_agents holds those steers. A bare spawn_agent does not. When the fleet goes dry the runtime re-enters with collected reports.
 
 # Operator updates (mandatory while fleet is live)
 

src/agent/tools.ts

@@ -59,6 +59,7 @@ import {
   createSpawnAgentTool,
   createWaitAgentsTool,
   createListAgentsTool,
+  type FleetMailboxHandle,
 } from "../subagent/agent-fleet.js";
 import { DEFAULT_CLOSE_DEADLINE_MS } from "../subagent/dispose.js";
 import {
@@ -252,6 +253,12 @@ export interface AgentToolset {
   setToolPromoter: (promote: (names: string[]) => void) => void;
   // Session-start skill snapshot shared with the prompt listing.
   skills: SkillSummary[];
+  /**
+   * The live wait mailbox this toolset already built for spawn_agent /
+   * wait_agents. Optional because a session without sub-agents has none.
+   * Callers must read this each time — do not capture a startup snapshot.
+   */
+  fleetRecords?: FleetMailboxHandle;
   dispose: () => Promise<void>;
 }
 
@@ -377,9 +384,10 @@ export async function createAgentToolset(args: AgentToolsetArgs): Promise<AgentT
   // spawn_agent/wait_agents.
   const orchestratorTools: AgentTool[] = [];
   let fleetSessionsForDispose: SubAgentSessionStore | undefined;
+  let fleetRecords: FleetMailboxHandle | undefined;
   if (subAgentsEnabled && args.subAgent !== undefined) {
     const sa = args.subAgent;
-    const fleetRecords = sa.sessions !== undefined ? createFleetMailbox(sa.sessions) : undefined;
+    fleetRecords = sa.sessions !== undefined ? createFleetMailbox(sa.sessions) : undefined;
     if (sa.profiles !== undefined) {
       orchestratorTools.push(
         createSearchAgentsTool(() => {
@@ -977,6 +985,7 @@ export async function createAgentToolset(args: AgentToolsetArgs): Promise<AgentT
       promoter.promote = promote;
     },
     skills,
+    ...(fleetRecords !== undefined ? { fleetRecords } : {}),
     dispose,
   };
 }

src/director.test.ts

@@ -244,6 +244,48 @@ describe("open-task termination guard", () => {
     expect(hasInfer(exhausted)).toBe(false);
   });
 
+  test("live fleet with open tasks allows terminal wait/reply and does not spend the nudge budget", async () => {
+    let live = 1;
+    const director = createChatDirector("base", [], {
+      onTasksChange: () => {},
+      getLiveFleetCount: () => live,
+    });
+    await director.decide(manageTasksEvent("doing"), mockState, mockCapabilities);
+
+    for (let i = 0; i < 4; i++) {
+      const actions = actionsArray(await director.decide(textTurn(), mockState, mockCapabilities));
+      expect(hasInfer(actions)).toBe(false);
+      expect(hasReply(actions)).toBe(true);
+    }
+
+    live = 0;
+    for (let i = 0; i < 3; i++) {
+      const nudged = actionsArray(await director.decide(textTurn(), mockState, mockCapabilities));
+      expect(hasInfer(nudged)).toBe(true);
+      expect(hasReply(nudged)).toBe(false);
+    }
+    const exhausted = actionsArray(await director.decide(textTurn(), mockState, mockCapabilities));
+    expect(hasReply(exhausted)).toBe(true);
+    expect(hasInfer(exhausted)).toBe(false);
+  });
+
+  test("omitted or zero live fleet count still nudges while a task is open", async () => {
+    const omitted = createChatDirector("base", [], { onTasksChange: () => {} });
+    await omitted.decide(manageTasksEvent("doing"), mockState, mockCapabilities);
+    expect(
+      hasInfer(actionsArray(await omitted.decide(textTurn(), mockState, mockCapabilities))),
+    ).toBe(true);
+
+    const zero = createChatDirector("base", [], {
+      onTasksChange: () => {},
+      getLiveFleetCount: () => 0,
+    });
+    await zero.decide(manageTasksEvent("doing"), mockState, mockCapabilities);
+    expect(hasInfer(actionsArray(await zero.decide(textTurn(), mockState, mockCapabilities)))).toBe(
+      true,
+    );
+  });
+
   test("empty model turn settles with a valid empty reply", async () => {
     // DefaultDirector ends empty responses with bare wait; without a reply,
     // agent.send hangs and the TUI Working spinner sticks forever.

src/session/assemble-runtime.ts

@@ -335,6 +335,11 @@ export interface ChatAgentWiring {
   inactivityTimeoutMs: number;
   totalTimeoutMs?: number | undefined;
   onTasksChange: (tasks: Task[]) => void;
+  /**
+   * Live running-lane count for ChatDirector idle-with-fleet. Omitted in exec
+   * (treated as 0).
+   */
+  getLiveFleetCount?: () => number;
   /** Compaction governor re-entry (the reactor emits no event after compact). */
   requestContinuation: () => void;
   getProvider: () => { providerName: string; model: string };
@@ -392,6 +397,7 @@ export function assembleChatAgent(wiring: ChatAgentWiring): AssembledChatAgent {
           requestContinuation: wiring.requestContinuation,
           provider: { ...wiring.getProvider() },
           getProviderId: wiring.getProviderId,
+          getLiveFleetCount: wiring.getLiveFleetCount,
         },
       );
       directorHolder.instance = d;

src/session/runtime-assembly.ts

@@ -379,3 +379,25 @@ export function buildCompactionContinuationMessage(): InboundMessage {
     signatureStatus: "missing",
   };
 }
+
+/**
+ * System-originated inbound that re-enters the parent after the fleet goes dry
+ * with todo/doing tasks still open. Not operator input, so no
+ * OPERATOR_ORIGINATED_FLAG. ChatDirector still resets idle and tool-only
+ * nudge counters on any message.received — occupancy therefore fires one
+ * deferred shot per dry edge rather than re-driving on every settle.
+ */
+export function buildFleetDryContinuationMessage(text: string): InboundMessage {
+  return {
+    ref: { uid: 0, mailbox: "system" },
+    headers: {
+      from: "user@local",
+      to: ["agent@local"],
+      date: new Date().toISOString(),
+      messageId: `fleet-dry-continue-${Date.now()}@local`,
+    },
+    flags: [],
+    content: text,
+    signatureStatus: "missing",
+  };
+}

src/subagent/agent-fleet.ts

@@ -94,6 +94,7 @@ import { formatSubAgentSpawnAuthFailureMessage } from "./inference-auth-failure.
 import { isResolvedProviderFailureError } from "../inference-error-message.js";
 import { isSubAgentCancelError } from "./dispose.js";
 import { createInterventionLog, type InterventionSink } from "./intervention-log.js";
+import { takeAndProjectMailboxRecord } from "./fleet-dry-drive.js";
 
 const log = getLogger([LOG_NAMESPACE_ROOT, "subagent", "agent-fleet"]);
 
@@ -1398,20 +1399,14 @@ export function createWaitAgentsTool(deps: WaitAgentsDeps): AgentTool {
         if (isLiveWaitStatus(record.status)) {
           return { agent_id: id, status: record.status };
         }
-        const taken = deps.fleetRecords.take(id) ?? record;
+        const projected = takeAndProjectMailboxRecord(deps.fleetRecords, id);
+        if (projected === undefined) {
+          return { agent_id: id, status: "unknown" as const };
+        }
         return {
-          agent_id: id,
-          status: taken.status,
-          ...(taken.question !== undefined ? { question: taken.question } : {}),
-          ...(taken.questionId !== undefined ? { question_id: taken.questionId } : {}),
-          ...(taken.description !== undefined ? { description: taken.description } : {}),
-          ...(taken.status !== "failed" && taken.report !== undefined
-            ? { report: taken.report }
-            : {}),
-          ...(taken.error !== undefined ? { error: taken.error } : {}),
-          ...(taken.stopReason !== undefined ? { stop_reason: taken.stopReason } : {}),
-          ...(taken.providerFailure === true ? { provider_failure: true } : {}),
-          ...(taken.hint !== undefined ? { hint: taken.hint } : {}),
+          ...projected,
+          ...(record.question !== undefined ? { question: record.question } : {}),
+          ...(record.questionId !== undefined ? { question_id: record.questionId } : {}),
         };
       });