feat(director): liven idle-with-fleet flag and decouple host from workflow seam typing (#1048)

* feat(director): liven idle-with-fleet flag and decouple host from workflow seam typing

Summary: optionalize the host-to-runtime workflow seam so the host degrades gracefully against directors without workflow support; replace the static idle-with-fleet seed with a live narrow setter driven by fleet-wake publisher transitions. Verification: bun run check passes (7393 tests, 0 fail).

* fix(tui): re-sync idle-with-fleet flag on director rebuilds (#1058)

Author: TheGreatAxios

src/agent/director.ts

@@ -447,23 +447,24 @@ export interface ChatDirectorOptions {
    *   uses a switched model still stamp the previous id — the director learns
    *   the new id from that inference's completion event.
    *
-   * - getLiveFleetCount → static config (allowIdleWithFleet below). The count
-   *   is genuinely external (subagent lane statuses the reactor never sees —
-   *   its own tasks only carry todo/doing/done/cancelled), so neither
-   *   BaseEnv-derived nor reactor-supplied can reproduce its liveness.
-   *   Idle-with-fleet itself is unchanged (fleet-running TUI sessions allow
-   *   the terminal wait). Accepted loss: a drained fleet no longer resumes
-   *   the open-task nudge — bounded, since the nudge capitulates to terminal
-   *   after its cap anyway.
+   * - getLiveFleetCount → seeded config + live narrow setter
+   *   (setAllowIdleWithFleet below). The count is genuinely external (subagent
+   *   lane statuses the reactor never sees — its own tasks only carry
+   *   todo/doing/done/cancelled), so neither BaseEnv-derived nor
+   *   reactor-supplied can reproduce its liveness. Idle-with-fleet itself is
+   *   unchanged (fleet-running TUI sessions allow the terminal wait); the
+   *   fleet-wake publisher drives the setter on count transitions, so a
+   *   drained fleet resumes the open-task nudge.
    */
   /** Explicit retry policy; when set, skips the default Corbits policy. */
   retryPolicy?: RetryPolicy | undefined;
   /**
-   * Static idle-with-fleet allowance (CL-7918 replacement for the former
+   * Initial idle-with-fleet allowance (CL-7918 replacement for the former
    * getLiveFleetCount closure). When true the director allows a terminal
    * wait/reply with open tasks; when omitted or false it keeps the open-task
-   * nudge. The TUI sets this (fleet lanes may appear mid-session); exec
-   * omits it.
+   * nudge. The TUI seeds this (fleet lanes may appear mid-session); exec
+   * omits it. The live fleet-wake publisher then keeps it current through
+   * setAllowIdleWithFleet, so a drained fleet resumes the nudge.
    */
   allowIdleWithFleet?: boolean | undefined;
 }
@@ -515,8 +516,8 @@ class ChatDirectorImpl extends DefaultDirector {
   // refreshed on every inference completion, so mid-session /model switches
   // remap without rebuilding the agent.
   private currentSourceId: string | undefined;
-  /** CL-7918 static replacement for the former getLiveFleetCount closure. */
-  private readonly allowIdleWithFleet: boolean;
+  /** CL-7918 live replacement for the former getLiveFleetCount closure. */
+  private allowIdleWithFleet: boolean;
   // 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
@@ -578,6 +579,13 @@ class ChatDirectorImpl extends DefaultDirector {
     this.workflowCoordinator = coordinator;
   }
 
+  // Narrow live setter for the idle-with-fleet allowance (CL-7972): the
+  // fleet-wake publisher drives this on fleet-count transitions, so a drained
+  // fleet resumes the open-task nudge instead of holding the seeded value.
+  setAllowIdleWithFleet(value: boolean): void {
+    this.allowIdleWithFleet = value;
+  }
+
   updateToolDefinitions(toolDefinitions: ToolDefinition[]): void {
     const before = toolSetDigest(this._toolDefinitions);
     const after = toolSetDigest(toolDefinitions);
@@ -1095,9 +1103,10 @@ class ChatDirectorImpl extends DefaultDirector {
         (a) => a.type === "wait" || a.type === "reply",
       );
       if (hasTerminal) {
-        // CL-7918: static idle-with-fleet allowance (replaces the former
-        // getLiveFleetCount closure). TUI sessions set allowIdleWithFleet;
-        // exec omits it and keeps the nudge.
+        // CL-7918 live idle-with-fleet allowance (replaces the former
+        // getLiveFleetCount closure): seeded at construction, then kept
+        // current by the fleet-wake publisher. TUI seeds true (fleet lanes
+        // may appear mid-session); exec omits it and keeps the nudge.
         if (this.allowIdleWithFleet) {
           return base;
         }
@@ -1172,6 +1181,7 @@ export function hydrateTasksFromTurns(turns: ConversationTurn[]): Task[] {
 export interface ChatDirector extends ReactorDirector {
   updateToolDefinitions(toolDefinitions: ToolDefinition[]): void;
   setWorkflowCoordinator(coordinator: WorkflowCoordinator | undefined): void;
+  setAllowIdleWithFleet(value: boolean): void;
   getTasks(): Task[];
   restoreTasks(tasks: Task[]): void;
   getContextEstimate(): { tokens: number; isEstimate: boolean };

src/director.test.ts

@@ -442,6 +442,41 @@ describe("open-task termination guard", () => {
     ).toBe(true);
   });
 
+  test("setAllowIdleWithFleet tracks fleet transitions off the seeded value", async () => {
+    const director = createChatDirector("base", [], {
+      onTasksChange: () => undefined,
+      allowIdleWithFleet: true,
+    });
+    await director.decide(
+      manageTasksEvent("doing"),
+      mockState,
+      mockCapabilities,
+    );
+
+    // Seeded allowance: terminal reply with open tasks, no nudge spent.
+    const seeded = actionsArray(
+      await director.decide(textTurn(), mockState, mockCapabilities),
+    );
+    expect(hasReply(seeded)).toBe(true);
+    expect(hasInfer(seeded)).toBe(false);
+
+    // Drained fleet resumes the open-task nudge.
+    director.setAllowIdleWithFleet(false);
+    const nudged = actionsArray(
+      await director.decide(textTurn(), mockState, mockCapabilities),
+    );
+    expect(hasInfer(nudged)).toBe(true);
+    expect(hasReply(nudged)).toBe(false);
+
+    // Fleet back: terminal allowed again.
+    director.setAllowIdleWithFleet(true);
+    const settled = actionsArray(
+      await director.decide(textTurn(), mockState, mockCapabilities),
+    );
+    expect(hasReply(settled)).toBe(true);
+    expect(hasInfer(settled)).toBe(false);
+  });
+
   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

@@ -474,9 +474,10 @@ export interface ChatAgentWiring {
   totalTimeoutMs?: number | undefined;
   onTasksChange: (tasks: Task[]) => void;
   /**
-   * Static idle-with-fleet allowance for the chat director (CL-7918: replaces
-   * the former getLiveFleetCount closure). Omitted in exec (keeps the
-   * open-task nudge); the TUI sets it (fleet lanes may appear mid-session).
+   * Seed for the idle-with-fleet allowance for the chat director (CL-7918:
+   * replaces the former getLiveFleetCount closure; CL-7972 keeps it live via
+   * the fleet-wake publisher). Omitted in exec (keeps the open-task nudge);
+   * the TUI seeds it (fleet lanes may appear mid-session).
    */
   allowIdleWithFleet?: boolean;
   /** Compaction governor re-entry (the reactor emits no event after compact). */
@@ -542,8 +543,9 @@ export function assembleChatAgent(wiring: ChatAgentWiring): AssembledChatAgent {
           onTasksChange: wiring.onTasksChange,
           requestContinuation: wiring.requestContinuation,
           provider: { ...wiring.getProvider() },
-          // CL-7918: static idle-with-fleet allowance replaces the former
-          // getLiveFleetCount closure; the live source id for retry stamping
+          // CL-7918: idle-with-fleet seed replaces the former
+          // getLiveFleetCount closure (CL-7972 keeps it live via the
+          // fleet-wake publisher); the live source id for retry stamping
           // is reactor-tracked now (no getProviderId).
           allowIdleWithFleet: wiring.allowIdleWithFleet,
         },

src/tui/runner/exit.test.ts

@@ -5,6 +5,14 @@ import { getLogger } from "@intx/log";
 import type { InferenceSource } from "@intx/types/runtime";
 
 import * as codexSession from "../../auth/codex/session.js";
+import { createChatDirector } from "../../agent/director.js";
+import { createSubAgentSessionStore } from "../../subagent/session-store.js";
+import type {
+  ReactorAction,
+  ReactorCapabilities,
+  ReactorInboundEvent,
+  ReactorState,
+} from "@intx/types/runtime";
 import { LOG_NAMESPACE_ROOT } from "../../branding.js";
 import { defined } from "../../../tests/helpers/defined.js";
 import {
@@ -259,3 +267,126 @@ describe("agentProxy.send vs /clear", () => {
     }
   });
 });
+
+const rebuildMockState: ReactorState = {} as unknown as ReactorState;
+
+const rebuildMockCapabilities: ReactorCapabilities = {
+  infer: (options) =>
+    ({
+      type: "infer",
+      ...(options !== undefined ? { options } : {}),
+    }) as ReactorAction,
+  executeTools: (calls) => ({ type: "execute_tools", calls }),
+  suspend: (gate) => ({ type: "suspend", gate }),
+  fork: (mode, forkId) => ({ type: "fork", mode, forkId }),
+  emit: (eventType, data) => ({ type: "emit", eventType, data }),
+  reply: (content) => ({ type: "reply", content }),
+  checkpoint: (message = "") => ({ type: "checkpoint", message }),
+  compact: (compactor, reason) => ({ type: "compact", compactor, reason }),
+  wait: () => ({ type: "wait" }),
+  done: () => ({ type: "done" }),
+};
+
+function rebuildManageTasksEvent(): ReactorInboundEvent {
+  return {
+    type: "inference.done",
+    turn: {
+      role: "assistant",
+      model: "test",
+      timestamp: 0,
+      content: [
+        {
+          type: "tool_call",
+          id: "m",
+          name: "manage_tasks",
+          arguments: {
+            action: "create",
+            tasks: [{ id: "t1", title: "work", status: "doing" }],
+          },
+        },
+      ],
+    },
+    usage: { input: 0, output: 1, cacheRead: 0, cacheWrite: 0, thinking: 0 },
+    source: { model: "test-model" },
+  } as unknown as ReactorInboundEvent;
+}
+
+function rebuildTextTurn(): ReactorInboundEvent {
+  return {
+    type: "inference.done",
+    turn: {
+      role: "assistant",
+      model: "test",
+      timestamp: 0,
+      content: [{ type: "text", text: "all set" }],
+    },
+    usage: { input: 10, output: 1, cacheRead: 0, cacheWrite: 0, thinking: 0 },
+    source: { model: "test-model" },
+  } as unknown as ReactorInboundEvent;
+}
+
+describe("rebuild re-syncs idle-with-fleet while drained", () => {
+  test("reload-if-idle and interrupt rebuilds resume the open-task nudge with no fleet transition", async () => {
+    const store = createSubAgentSessionStore();
+    const directorHolder: RunnerServices["directorHolder"] = {};
+    const agent = recordingAgent([]);
+    const { state, services } = stubSendLifecycle(agent);
+    services.directorHolder =
+      directorHolder as unknown as RunnerServices["directorHolder"];
+    services.subAgentSessions =
+      store as unknown as RunnerServices["subAgentSessions"];
+    services.workflowHost = {
+      reattach: () => undefined,
+    } as unknown as RunnerServices["workflowHost"];
+    services.cycleRecorder = {
+      dispose: async () => "",
+      reset: () => undefined,
+      handleEvent: () => undefined,
+    } as unknown as RunnerServices["cycleRecorder"];
+    services.buildAgent = (async () => {
+      // Every rebuild mints a fresh director from the static true seed (fleet
+      // lanes may appear mid-session), exactly like the TUI session assembly.
+      directorHolder.instance = createChatDirector("base", [], {
+        onTasksChange: () => undefined,
+        allowIdleWithFleet: true,
+      });
+      return agent;
+    }) as unknown as RunnerServices["buildAgent"];
+    const fleetEvents: unknown[] = [];
+    services.emitter.on("event", (event: { type: string }) => {
+      if (event.type === "fleet") fleetEvents.push(event);
+    });
+    await createRunLifecycle(state, services);
+    const expectOpenTaskNudge = async (): Promise<void> => {
+      const director = defined(
+        directorHolder.instance,
+        "directorHolder.instance",
+      );
+      await director.decide(
+        rebuildManageTasksEvent(),
+        rebuildMockState,
+        rebuildMockCapabilities,
+      );
+      const actions = await director.decide(
+        rebuildTextTurn(),
+        rebuildMockState,
+        rebuildMockCapabilities,
+      );
+      const list = Array.isArray(actions) ? actions : [actions];
+      expect(list.some((action) => action.type === "infer")).toBe(true);
+    };
+    // Drained fleet: the idle reload rebuilds onto the static true seed.
+    state.pendingReload = true;
+    defined(state.reloadIfIdle, "reloadIfIdle")();
+    await services.sessionOps.awaitTail();
+    expect(state.fatalBuildError).toBeNull();
+    await expectOpenTaskNudge();
+    // The interrupt rebuild inherits the same seed.
+    defined(state.interrupt, "interrupt")();
+    await services.sessionOps.awaitTail();
+    expect(state.fatalBuildError).toBeNull();
+    await expectOpenTaskNudge();
+    expect(store.list()).toEqual([]);
+    expect(fleetEvents).toEqual([]);
+  });
+});

src/tui/runner/exit.ts

@@ -13,6 +13,7 @@ import {
 import { getLogger } from "@intx/log";
 import type { InferenceSource } from "@intx/types/runtime";
 import { consumeStream } from "../../session/stream-consumer.js";
+import { liveFleetCount } from "../../subagent/index.js";
 import { getTelemetry } from "../../telemetry/singleton.js";
 import { onTurnBoundary } from "../../agent/reactor-events.js";
 import { setAgentSourceUnlessClosed } from "../agent-source-sync.js";
@@ -145,6 +146,18 @@ export async function closeAgentForRebuild(
   }
 }
 
+// Rebuilt directors seed allowIdleWithFleet=true (fleet lanes may appear
+// mid-session), so a rebuild while drained must re-sync the new director from
+// the live fleet count — otherwise the open-task nudge stays suppressed until
+// the next fleet transition, which never comes for an already-drained fleet.
+function resyncIdleWithFleetFlag(
+  services: Pick<RunnerServices, "directorHolder" | "subAgentSessions">,
+): void {
+  services.directorHolder.instance?.setAllowIdleWithFleet(
+    liveFleetCount(services.subAgentSessions.list()) > 0,
+  );
+}
+
 // Every rebuild site funnels its failure (a lock left held by a failed
 // close, or any other buildAgent failure) through here so it surfaces as a
 // plain-language, caught error rather than an unhandled rejection.
@@ -345,6 +358,7 @@ export async function createRunLifecycle(
           throw new AgentContextLockError(state.workdir);
         }
         state.currentAgent = await services.buildAgent();
+        resyncIdleWithFleetFlag(services);
         state.streamPromise = consumeStream(
           liveAgent(state).stream(),
           streamSink,
@@ -547,6 +561,7 @@ export async function createRunLifecycle(
             throw new AgentContextLockError(state.workdir);
           }
           state.currentAgent = await services.buildAgent();
+          resyncIdleWithFleetFlag(services);
           services.cycleRecorder.reset();
           state.streamPromise = consumeStream(
             liveAgent(state).stream(),

src/tui/runner/session.ts

@@ -619,8 +619,9 @@ export async function assembleTUISession(
     inactivityTimeoutMs: config.inactivityTimeoutMs ?? 750_000,
     totalTimeoutMs: config.totalTimeoutMs,
     onTasksChange: (tasks) => emitter.emit("tasks", tasks),
-    // CL-7918: static idle-with-fleet allowance (fleet lanes may appear
-    // mid-session); retry stamping tracks the live source id in-reactor now.
+    // CL-7918 seed for the idle-with-fleet allowance (fleet lanes may appear
+    // mid-session; CL-7972 keeps it live via the fleet-wake publisher);
+    // retry stamping tracks the live source id in-reactor now.
     allowIdleWithFleet: true,
     requestContinuation: () => {
       const targetAgent = liveAgent(state);

src/tui/runner/wiring.ask-wake.test.ts

@@ -68,6 +68,36 @@ test("failed reset releases publication without flushing partially cancelled wor
   }
 });
 
+test("fleet-wake publisher reports fleet-count transitions to the idle-with-fleet follower", () => {
+  const store = createSubAgentSessionStore();
+  const emitter = new EventEmitter();
+  const seen: number[] = [];
+  const publisher = createFleetWakePublisher(store, emitter, (running) => {
+    seen.push(running);
+  });
+  // Steady empty state: no transition, no callback.
+  publisher.publish();
+  expect(seen).toEqual([]);
+  store.start({
+    id: "worker-one",
+    agentId: "builder",
+    description: "worker-one",
+    brief: "build",
+  });
+  // A started lane counts as running: 0 -> 1 transition.
+  publisher.publish();
+  expect(seen).toEqual([1]);
+  store.markRunning("worker-one");
+  publisher.publish();
+  expect(seen).toEqual([1]);
+  // Steady fleet: no repeat callback.
+  publisher.publish();
+  expect(seen).toEqual([1]);
+  store.cancelAll("Session cleared");
+  publisher.publish();
+  expect(seen).toEqual([1, 0]);
+});
+
 for (const phase of ["settled", "prequeued", "deferred"] as const) {
   test(`rotation suppresses old worker snapshots (${phase}), including repeated resets`, async () => {
     await withTestRenderer(