refactor(director): resolve provider and fleet state without host closures (#1044)

* feat(director): resolve getProviderId and getLiveFleetCount without host closures (CL-7918)

Decision record:
- getProviderId -> reactor-supplied. Retry stamping needs the live source
  id so mid-session /model switches remap (xAI bare-429 retry vs abort).
  Neither static config (loses remap liveness) nor BaseEnv.sources
  (carries definitions, not the live selected id) suffice. The director
  now seeds from options.provider.providerName and refreshes the stamped
  id from each inference completion's source.sourceId; BaseEnv.sources
  remains the fallback only until the first completion stamps it.
- getLiveFleetCount -> static config allowIdleWithFleet (default false).
  Live counts only ever gated one boolean (non-zero allows terminal
  wait/reply with open tasks); the exact count was never consumed.
  Rejected: BaseEnv-derived (no fleet state there) and reactor-supplied
  (the reactor owns no fleet-lane registry; a live count closure just
  re-adds the host hook this lane removes). TUI chat sets
  allowIdleWithFleet: true (fleet lanes may appear mid-session); exec and
  the default keep the historical nudge-while-open behavior. Static true
  is strictly more permissive than the old count>0 gate once a fleet
  lane drains, which is the safe direction for terminal wait.

Both closures removed from ChatDirectorOptions. No new env key.
Mid-session /model remap and idle-with-fleet behavior covered by
src/director.test.ts (remap abort->retry across a source switch;
idle-with-fleet terminal wait without nudge spend).

* fix(director): ignore empty source ids in live source tracking (#1056)

* fix(director): ignore empty source ids in live source tracking

An empty-string source id on a completion or cycle source no longer clobbers the learned source id used to stamp retry decisions.

* docs(director): align cycle-source comment with override precedence (#1057)

* 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.test.ts

@@ -366,3 +366,172 @@ describe("ChatDirector inference-error recovery (CL-6910)", () => {
     );
   });
 });
+
+// CL-7973: the director's live source id (which stamps retry decisions so a
+// mid-session /model switch remaps the xAI short-429 handling) is observable
+// only through the retry policy it hands to each infer action. An xAI-gated
+// capacity error retries when the tracked id is an xAI source and aborts
+// otherwise, so driving tracking events then invoking the attached policy
+// reads the tracked id without reaching into privates.
+type LiveRetryPolicy = (situation: {
+  attempt: number;
+  elapsedMs: number;
+  error: { category: "protocol_mismatch"; message: string };
+}) => Promise<{ kind: string }> | { kind: string };
+
+function textCompletion(sourceId?: string): ReactorInboundEvent {
+  const turn = {
+    role: "assistant",
+    model: "test",
+    timestamp: 0,
+    content: [{ type: "text", text: "done work" }],
+  };
+  const event: Record<string, unknown> = {
+    type: "inference.done",
+    turn,
+    usage: { input: 0, output: 0 },
+  };
+  if (sourceId !== undefined)
+    event["source"] = { sourceId, provider: "p", model: "test" };
+  return event as unknown as ReactorInboundEvent;
+}
+
+function stateWithCycleSource(sourceId: string): ReactorState {
+  return {
+    turns: [],
+    lastCycleSource: { sourceId, provider: "p", model: "test" },
+  } as unknown as ReactorState;
+}
+
+async function liveRetryPolicy(
+  director: ReturnType<typeof createChatDirector>,
+  capabilities: ReactorCapabilities,
+): Promise<LiveRetryPolicy> {
+  await director.decide(toolOnlyTurn("source-probe"), mockState, capabilities);
+  const actions = actionsArray(
+    await director.decide(
+      toolDoneEvent("source-probe"),
+      mockState,
+      capabilities,
+    ),
+  );
+  const infer = actions.find((a) => a.type === "infer") as
+    | { type: "infer"; options?: { retryPolicy?: LiveRetryPolicy } }
+    | undefined;
+  if (infer?.options?.retryPolicy === undefined)
+    throw new Error("expected an infer action carrying the live retry policy");
+  return infer.options.retryPolicy;
+}
+
+async function isXaiStamped(policy: LiveRetryPolicy): Promise<boolean> {
+  const decision = await policy({
+    attempt: 1,
+    elapsedMs: 0,
+    error: {
+      category: "protocol_mismatch",
+      message: "The model is currently at capacity",
+    },
+  });
+  return decision.kind === "retry";
+}
+
+describe("ChatDirector live source-id tracking (CL-7973)", () => {
+  test("a sourceless or empty-string completion never wipes the learned id", async () => {
+    const director = createChatDirector("system", [], {
+      onTasksChange: () => undefined,
+      provider: { providerName: "test-provider" },
+    });
+    const capabilities = makeCapabilities();
+    const policy = await liveRetryPolicy(director, capabilities);
+
+    // The seed id is not an xAI source, so the capacity error aborts.
+    expect(await isXaiStamped(policy)).toBe(false);
+
+    // A completion stamps the source that served it.
+    await director.decide(
+      textCompletion("xai/learned"),
+      mockState,
+      capabilities,
+    );
+    expect(await isXaiStamped(policy)).toBe(true);
+
+    // A completion carrying no source keeps the learned id.
+    await director.decide(textCompletion(), mockState, capabilities);
+    expect(await isXaiStamped(policy)).toBe(true);
+
+    // An empty-string source id never clobbers the learned id.
+    await director.decide(textCompletion(""), mockState, capabilities);
+    expect(await isXaiStamped(policy)).toBe(true);
+
+    // An empty-string cycle source never clobbers it either.
+    await director.decide(
+      toolDoneEvent("empty-cycle"),
+      stateWithCycleSource(""),
+      capabilities,
+    );
+    expect(await isXaiStamped(policy)).toBe(true);
+  });
+
+  test("a cycle source remaps tracking on a non-inference event", async () => {
+    const director = createChatDirector("system", [], {
+      onTasksChange: () => undefined,
+      provider: { providerName: "test-provider" },
+    });
+    const capabilities = makeCapabilities();
+    const policy = await liveRetryPolicy(director, capabilities);
+    expect(await isXaiStamped(policy)).toBe(false);
+
+    // tool.done carries no source of its own; the harness's cycle source
+    // covers the turn and remaps tracking from it.
+    await director.decide(
+      toolDoneEvent("cycle-remap"),
+      stateWithCycleSource("xai/cycle"),
+      capabilities,
+    );
+    expect(await isXaiStamped(policy)).toBe(true);
+  });
+
+  test("a drained fleet capitulates to the terminal action after the nudge budget", async () => {
+    const director = createChatDirector("system", [], {
+      onTasksChange: () => undefined,
+      provider: { providerName: "test-provider" },
+    });
+    director.restoreTasks([{ id: "t1", title: "keep going", status: "todo" }]);
+    const capabilities = makeCapabilities();
+
+    // Without the idle-with-fleet allowance (drained fleet), a terminal base
+    // action with open tasks re-infers with the open-task nudge a bounded
+    // number of times, then lets the terminal action through — the accepted
+    // loss stays locked in rather than resuming the nudge.
+    for (let i = 0; i < 3; i++) {
+      const actions = actionsArray(
+        await director.decide(textCompletion(), mockState, capabilities),
+      );
+      expect(actions.some((a) => a.type === "infer")).toBe(true);
+      expect(actions.some((a) => a.type === "reply")).toBe(false);
+    }
+    const terminal = actionsArray(
+      await director.decide(textCompletion(), mockState, capabilities),
+    );
+    expect(terminal.some((a) => a.type === "infer")).toBe(false);
+    expect(terminal.some((a) => a.type === "reply")).toBe(true);
+  });
+
+  test("a cycle source wins over a contradictory event source", async () => {
+    const director = createChatDirector("system", [], {
+      onTasksChange: () => undefined,
+      provider: { providerName: "test-provider" },
+    });
+    const capabilities = makeCapabilities();
+    const policy = await liveRetryPolicy(director, capabilities);
+
+    // The harness's call-start snapshot is authoritative over the event's
+    // own stamp, so when both are present the cycle source defines tracking.
+    await director.decide(
+      textCompletion("other/default"),
+      stateWithCycleSource("xai/cycle"),
+      capabilities,
+    );
+    expect(await isXaiStamped(policy)).toBe(true);
+  });
+});

src/agent/director.ts

@@ -434,19 +434,39 @@ export interface ChatDirectorOptions {
   requestContinuation?: (() => void) | undefined;
   provider?: { providerName: string; model?: string } | undefined;
   /**
-   * Live catalog provider id for retry stamping. Resolved on each retry
-   * decision so mid-session `/model` switches remapping without rebuilding
-   * the agent. When set, preferred over static `provider.providerName`.
+   * CL-7918 decisions (both former closures removed, no new env key):
+   *
+   * - getProviderId → reactor-supplied. The retry policy needs the *live*
+   *   source id per retry so mid-session /model switches remap retry stamping
+   *   (bare-429 xAI remap). A BaseEnv-derived id goes stale at the first
+   *   switch and only refreshes on rebuild; a static config id can never
+   *   remap. The reactor already learns the live id on every inference
+   *   completion, so it tracks currentSourceId itself (seeded from the session
+   *   providerName) and hands the policy a getter over it.
+   *   Accepted residual gap: retries during the single inference that first
+   *   uses a switched model still stamp the previous id — the director learns
+   *   the new id from that inference's completion event.
+   *
+   * - 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.
    */
-  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.
+   * 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 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.
    */
-  getLiveFleetCount?: (() => number) | undefined;
+  allowIdleWithFleet?: boolean | undefined;
 }
 
 // The constructor takes the resolved ModelFamilyPolicy rather than the raw
@@ -455,6 +475,8 @@ type ChatDirectorImplOptions = Omit<ChatDirectorOptions, "provider"> & {
   modelFamilyPolicy?: ModelFamilyPolicy | undefined;
   /** Provider-stamped retry policy (xAI short 429 remapping needs providerId). */
   retryPolicy?: RetryPolicy | undefined;
+  /** Session-construction providerName: seeds currentSourceId pre-completion. */
+  sessionProviderName?: string | undefined;
 };
 
 class ChatDirectorImpl extends DefaultDirector {
@@ -489,7 +511,13 @@ class ChatDirectorImpl extends DefaultDirector {
   private readonly compaction: CompactionGovernor;
   private readonly modelFamilyPolicy: ModelFamilyPolicy;
   private readonly retryPolicy: RetryPolicy;
-  private readonly getLiveFleetCount: (() => number) | undefined;
+  // CL-7918: reactor-supplied live source id for retry stamping (replaces the
+  // former getProviderId closure). Seeded from the session providerName and
+  // refreshed on every inference completion, so mid-session /model switches
+  // remap without rebuilding the agent.
+  private currentSourceId: string | undefined;
+  /** 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
@@ -536,14 +564,28 @@ class ChatDirectorImpl extends DefaultDirector {
       toolDefinitions,
     );
     this.modelFamilyPolicy = familyPolicy;
-    this.retryPolicy = options.retryPolicy ?? createCorbitsRetryPolicy();
-    this.getLiveFleetCount = options.getLiveFleetCount;
+    // CL-7918: the default policy stamps the live source id per retry decision
+    // via a getter over currentSourceId (seeded from the session provider,
+    // refreshed on each inference completion) — no host closure needed. An
+    // explicit policy still skips this entirely.
+    this.currentSourceId = options.sessionProviderName;
+    this.retryPolicy =
+      options.retryPolicy ??
+      createCorbitsRetryPolicy({ providerId: () => this.currentSourceId });
+    this.allowIdleWithFleet = options.allowIdleWithFleet === true;
   }
 
   setWorkflowCoordinator(coordinator: WorkflowCoordinator | undefined): void {
     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);
@@ -951,6 +993,16 @@ class ChatDirectorImpl extends DefaultDirector {
     // prefers provider usage when present.
     const turns = state.turns ?? [];
     this.compaction.syncFromTurns(turns);
+    // CL-7918: reactor-supplied live source id (replaces getProviderId). The
+    // completion stamps the source that served it, so a mid-session /model
+    // switch remaps retry stamping from the next completion on; the harness's
+    // lastCycleSource is the call-start snapshot and wins on conflict.
+    if (event.type === "inference.done") {
+      const served = event.source?.sourceId;
+      if (served !== undefined && served !== "") this.currentSourceId = served;
+    }
+    const cycled = state.lastCycleSource?.sourceId;
+    if (cycled !== undefined && cycled !== "") this.currentSourceId = cycled;
     if (onTurnBoundary(event)) {
       this.compaction.noteInferenceDone(event, turns);
     }
@@ -1051,7 +1103,11 @@ class ChatDirectorImpl extends DefaultDirector {
         (a) => a.type === "wait" || a.type === "reply",
       );
       if (hasTerminal) {
-        if ((this.getLiveFleetCount?.() ?? 0) > 0) {
+        // 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;
         }
         if (this.idleTerminationNudges < MAX_OPEN_TASK_NUDGES) {
@@ -1086,25 +1142,21 @@ export function createChatDirector(
   toolDefinitions: ToolDefinition[],
   options: ChatDirectorOptions,
 ): ChatDirector {
-  const { provider, getProviderId, retryPolicy, ...rest } = options;
+  const { provider, retryPolicy, ...rest } = options;
   return new ChatDirectorImpl(systemPrompt, toolDefinitions, {
     ...rest,
     // `provider` is raw {providerName, model} input; the constructor wants
     // the resolved ModelFamilyPolicy, not the input it was resolved from.
     modelFamilyPolicy:
       provider !== undefined ? resolveModelFamilyPolicy(provider) : undefined,
+    // CL-7918: seed the reactor-tracked live source id (replaces
+    // getProviderId). The impl refreshes it on every inference completion so
+    // mid-session `/model` switches remap retry stamping.
+    sessionProviderName: provider?.providerName,
     // Stamp provider id onto retry errors so known-xAI short 429s remap.
-    // Prefer an explicit policy, then a live getter (mid-session `/model`),
-    // then the bootstrap providerName.
-    retryPolicy:
-      retryPolicy ??
-      createCorbitsRetryPolicy(
-        getProviderId !== undefined
-          ? { providerId: getProviderId }
-          : provider !== undefined
-            ? { providerId: provider.providerName }
-            : undefined,
-      ),
+    // Prefer an explicit policy; otherwise the impl builds the default policy
+    // over its live source-id tracker.
+    retryPolicy,
   });
 }
 
@@ -1129,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 };