Skip to content

Commit 07ce93e

Browse files
committed
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).
1 parent 474ee30 commit 07ce93e

8 files changed

Lines changed: 171 additions & 33 deletions

File tree

‎src/agent/director.ts‎

Lines changed: 26 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -447,23 +447,24 @@ export interface ChatDirectorOptions {
447447
* uses a switched model still stamp the previous id — the director learns
448448
* the new id from that inference's completion event.
449449
*
450-
* - getLiveFleetCount → static config (allowIdleWithFleet below). The count
451-
* is genuinely external (subagent lane statuses the reactor never sees —
452-
* its own tasks only carry todo/doing/done/cancelled), so neither
453-
* BaseEnv-derived nor reactor-supplied can reproduce its liveness.
454-
* Idle-with-fleet itself is unchanged (fleet-running TUI sessions allow
455-
* the terminal wait). Accepted loss: a drained fleet no longer resumes
456-
* the open-task nudge — bounded, since the nudge capitulates to terminal
457-
* after its cap anyway.
450+
* - getLiveFleetCount → seeded config + live narrow setter
451+
* (setAllowIdleWithFleet below). The count is genuinely external (subagent
452+
* lane statuses the reactor never sees — its own tasks only carry
453+
* todo/doing/done/cancelled), so neither BaseEnv-derived nor
454+
* reactor-supplied can reproduce its liveness. Idle-with-fleet itself is
455+
* unchanged (fleet-running TUI sessions allow the terminal wait); the
456+
* fleet-wake publisher drives the setter on count transitions, so a
457+
* drained fleet resumes the open-task nudge.
458458
*/
459459
/** Explicit retry policy; when set, skips the default Corbits policy. */
460460
retryPolicy?: RetryPolicy | undefined;
461461
/**
462-
* Static idle-with-fleet allowance (CL-7918 replacement for the former
462+
* Initial idle-with-fleet allowance (CL-7918 replacement for the former
463463
* getLiveFleetCount closure). When true the director allows a terminal
464464
* wait/reply with open tasks; when omitted or false it keeps the open-task
465-
* nudge. The TUI sets this (fleet lanes may appear mid-session); exec
466-
* omits it.
465+
* nudge. The TUI seeds this (fleet lanes may appear mid-session); exec
466+
* omits it. The live fleet-wake publisher then keeps it current through
467+
* setAllowIdleWithFleet, so a drained fleet resumes the nudge.
467468
*/
468469
allowIdleWithFleet?: boolean | undefined;
469470
}
@@ -515,8 +516,8 @@ class ChatDirectorImpl extends DefaultDirector {
515516
// refreshed on every inference completion, so mid-session /model switches
516517
// remap without rebuilding the agent.
517518
private currentSourceId: string | undefined;
518-
/** CL-7918 static replacement for the former getLiveFleetCount closure. */
519-
private readonly allowIdleWithFleet: boolean;
519+
/** CL-7918 live replacement for the former getLiveFleetCount closure. */
520+
private allowIdleWithFleet: boolean;
520521
// Consecutive assistant turns that contain tool calls and no text. Reset on
521522
// any turn with text and on every fresh user message — a weak model that
522523
// spins in place on one thread of tool calls still converges to the
@@ -578,6 +579,13 @@ class ChatDirectorImpl extends DefaultDirector {
578579
this.workflowCoordinator = coordinator;
579580
}
580581

582+
// Narrow live setter for the idle-with-fleet allowance (CL-7972): the
583+
// fleet-wake publisher drives this on fleet-count transitions, so a drained
584+
// fleet resumes the open-task nudge instead of holding the seeded value.
585+
setAllowIdleWithFleet(value: boolean): void {
586+
this.allowIdleWithFleet = value;
587+
}
588+
581589
updateToolDefinitions(toolDefinitions: ToolDefinition[]): void {
582590
const before = toolSetDigest(this._toolDefinitions);
583591
const after = toolSetDigest(toolDefinitions);
@@ -1095,9 +1103,10 @@ class ChatDirectorImpl extends DefaultDirector {
10951103
(a) => a.type === "wait" || a.type === "reply",
10961104
);
10971105
if (hasTerminal) {
1098-
// CL-7918: static idle-with-fleet allowance (replaces the former
1099-
// getLiveFleetCount closure). TUI sessions set allowIdleWithFleet;
1100-
// exec omits it and keeps the nudge.
1106+
// CL-7918 live idle-with-fleet allowance (replaces the former
1107+
// getLiveFleetCount closure): seeded at construction, then kept
1108+
// current by the fleet-wake publisher. TUI seeds true (fleet lanes
1109+
// may appear mid-session); exec omits it and keeps the nudge.
11011110
if (this.allowIdleWithFleet) {
11021111
return base;
11031112
}
@@ -1172,6 +1181,7 @@ export function hydrateTasksFromTurns(turns: ConversationTurn[]): Task[] {
11721181
export interface ChatDirector extends ReactorDirector {
11731182
updateToolDefinitions(toolDefinitions: ToolDefinition[]): void;
11741183
setWorkflowCoordinator(coordinator: WorkflowCoordinator | undefined): void;
1184+
setAllowIdleWithFleet(value: boolean): void;
11751185
getTasks(): Task[];
11761186
restoreTasks(tasks: Task[]): void;
11771187
getContextEstimate(): { tokens: number; isEstimate: boolean };

‎src/director.test.ts‎

Lines changed: 35 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -442,6 +442,41 @@ describe("open-task termination guard", () => {
442442
).toBe(true);
443443
});
444444

445+
test("setAllowIdleWithFleet tracks fleet transitions off the seeded value", async () => {
446+
const director = createChatDirector("base", [], {
447+
onTasksChange: () => undefined,
448+
allowIdleWithFleet: true,
449+
});
450+
await director.decide(
451+
manageTasksEvent("doing"),
452+
mockState,
453+
mockCapabilities,
454+
);
455+
456+
// Seeded allowance: terminal reply with open tasks, no nudge spent.
457+
const seeded = actionsArray(
458+
await director.decide(textTurn(), mockState, mockCapabilities),
459+
);
460+
expect(hasReply(seeded)).toBe(true);
461+
expect(hasInfer(seeded)).toBe(false);
462+
463+
// Drained fleet resumes the open-task nudge.
464+
director.setAllowIdleWithFleet(false);
465+
const nudged = actionsArray(
466+
await director.decide(textTurn(), mockState, mockCapabilities),
467+
);
468+
expect(hasInfer(nudged)).toBe(true);
469+
expect(hasReply(nudged)).toBe(false);
470+
471+
// Fleet back: terminal allowed again.
472+
director.setAllowIdleWithFleet(true);
473+
const settled = actionsArray(
474+
await director.decide(textTurn(), mockState, mockCapabilities),
475+
);
476+
expect(hasReply(settled)).toBe(true);
477+
expect(hasInfer(settled)).toBe(false);
478+
});
479+
445480
test("empty model turn settles with a valid empty reply", async () => {
446481
// DefaultDirector ends empty responses with bare wait; without a reply,
447482
// agent.send hangs and the TUI Working spinner sticks forever.

‎src/session/assemble-runtime.ts‎

Lines changed: 7 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -474,9 +474,10 @@ export interface ChatAgentWiring {
474474
totalTimeoutMs?: number | undefined;
475475
onTasksChange: (tasks: Task[]) => void;
476476
/**
477-
* Static idle-with-fleet allowance for the chat director (CL-7918: replaces
478-
* the former getLiveFleetCount closure). Omitted in exec (keeps the
479-
* open-task nudge); the TUI sets it (fleet lanes may appear mid-session).
477+
* Seed for the idle-with-fleet allowance for the chat director (CL-7918:
478+
* replaces the former getLiveFleetCount closure; CL-7972 keeps it live via
479+
* the fleet-wake publisher). Omitted in exec (keeps the open-task nudge);
480+
* the TUI seeds it (fleet lanes may appear mid-session).
480481
*/
481482
allowIdleWithFleet?: boolean;
482483
/** Compaction governor re-entry (the reactor emits no event after compact). */
@@ -542,8 +543,9 @@ export function assembleChatAgent(wiring: ChatAgentWiring): AssembledChatAgent {
542543
onTasksChange: wiring.onTasksChange,
543544
requestContinuation: wiring.requestContinuation,
544545
provider: { ...wiring.getProvider() },
545-
// CL-7918: static idle-with-fleet allowance replaces the former
546-
// getLiveFleetCount closure; the live source id for retry stamping
546+
// CL-7918: idle-with-fleet seed replaces the former
547+
// getLiveFleetCount closure (CL-7972 keeps it live via the
548+
// fleet-wake publisher); the live source id for retry stamping
547549
// is reactor-tracked now (no getProviderId).
548550
allowIdleWithFleet: wiring.allowIdleWithFleet,
549551
},

‎src/tui/runner/session.ts‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -619,8 +619,9 @@ export async function assembleTUISession(
619619
inactivityTimeoutMs: config.inactivityTimeoutMs ?? 750_000,
620620
totalTimeoutMs: config.totalTimeoutMs,
621621
onTasksChange: (tasks) => emitter.emit("tasks", tasks),
622-
// CL-7918: static idle-with-fleet allowance (fleet lanes may appear
623-
// mid-session); retry stamping tracks the live source id in-reactor now.
622+
// CL-7918 seed for the idle-with-fleet allowance (fleet lanes may appear
623+
// mid-session; CL-7972 keeps it live via the fleet-wake publisher);
624+
// retry stamping tracks the live source id in-reactor now.
624625
allowIdleWithFleet: true,
625626
requestContinuation: () => {
626627
const targetAgent = liveAgent(state);

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

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -68,6 +68,36 @@ test("failed reset releases publication without flushing partially cancelled wor
6868
}
6969
});
7070

71+
test("fleet-wake publisher reports fleet-count transitions to the idle-with-fleet follower", () => {
72+
const store = createSubAgentSessionStore();
73+
const emitter = new EventEmitter();
74+
const seen: number[] = [];
75+
const publisher = createFleetWakePublisher(store, emitter, (running) => {
76+
seen.push(running);
77+
});
78+
// Steady empty state: no transition, no callback.
79+
publisher.publish();
80+
expect(seen).toEqual([]);
81+
store.start({
82+
id: "worker-one",
83+
agentId: "builder",
84+
description: "worker-one",
85+
brief: "build",
86+
});
87+
// A started lane counts as running: 0 -> 1 transition.
88+
publisher.publish();
89+
expect(seen).toEqual([1]);
90+
store.markRunning("worker-one");
91+
publisher.publish();
92+
expect(seen).toEqual([1]);
93+
// Steady fleet: no repeat callback.
94+
publisher.publish();
95+
expect(seen).toEqual([1]);
96+
store.cancelAll("Session cleared");
97+
publisher.publish();
98+
expect(seen).toEqual([1, 0]);
99+
});
100+
71101
for (const phase of ["settled", "prequeued", "deferred"] as const) {
72102
test(`rotation suppresses old worker snapshots (${phase}), including repeated resets`, async () => {
73103
await withTestRenderer(

‎src/tui/runner/wiring.ts‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -105,6 +105,10 @@ export function createFleetStallPollTick(
105105
export function createFleetWakePublisher(
106106
sessions: RunnerServices["subAgentSessions"],
107107
emitter: RunnerServices["emitter"],
108+
// Live idle-with-fleet flag (CL-7972): fired on fleet-count transitions so
109+
// the chat director's seeded allowance tracks the live fleet instead of
110+
// holding its construction value. Omitted in tests that only assert events.
111+
onFleetCount?: (running: number) => void,
108112
) {
109113
let lastLiveFleet = 0;
110114
let suspended = false;
@@ -120,6 +124,7 @@ export function createFleetWakePublisher(
120124
if (fleet !== lastLiveFleet) {
121125
lastLiveFleet = fleet;
122126
emitter.emit("event", { type: "fleet", running: fleet });
127+
onFleetCount?.(fleet);
123128
}
124129
return { previousRunning, running: fleet };
125130
};
@@ -261,6 +266,11 @@ export function wirePostStartup(
261266
const fleetWakePublisher = createFleetWakePublisher(
262267
services.subAgentSessions,
263268
services.emitter,
269+
// Liven the seeded idle-with-fleet allowance (CL-7972): the director
270+
// reads the holder live so rebuilds stay tracked, and degrades gracefully
271+
// while the director is not yet built.
272+
(running) =>
273+
services.directorHolder.instance?.setAllowIdleWithFleet(running > 0),
264274
);
265275
state.withFleetPublicationSuspended = fleetWakePublisher.withSuspended;
266276
sessionBridge.setDryOpenTaskDriver(() => {

‎src/workflows/host.ts‎

Lines changed: 7 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -59,8 +59,10 @@ export interface WorkflowHostArgs {
5959
getSessionId: () => string;
6060
getToolDefinitions: () => ToolDefinition[];
6161
// The live chat director; the workflow coordinator is attached to it when a
62-
// workflow starts. Returns undefined before the director is built.
63-
getDirector: () => { setWorkflowCoordinator: SetCoordinator } | undefined;
62+
// workflow starts. Returns undefined before the director is built. The seam
63+
// is optional so the host degrades gracefully against directors without
64+
// workflow support — every attach point presence-checks before calling.
65+
getDirector: () => { setWorkflowCoordinator?: SetCoordinator } | undefined;
6466
// Overrides the state-tree home (defaults to the real user home). Tests
6567
// pass a sandboxed dir here so persist()/resume() never touch ~/.corbits.
6668
home?: string;
@@ -84,7 +86,7 @@ export class WorkflowHost {
8486
// Re-attach the active coordinator to a freshly rebuilt director. Safe to
8587
// call with no active workflow — it just clears any stale coordinator.
8688
reattach(): void {
87-
this.args.getDirector()?.setWorkflowCoordinator(this.coordinator);
89+
this.args.getDirector()?.setWorkflowCoordinator?.(this.coordinator);
8890
}
8991

9092
// Drop the active workflow and history (e.g. on /clear).
@@ -94,7 +96,7 @@ export class WorkflowHost {
9496
this.pendingReplace = undefined;
9597
this.completedWorkflows = [];
9698
this.lastActiveStatus = undefined;
97-
this.args.getDirector()?.setWorkflowCoordinator(undefined);
99+
this.args.getDirector()?.setWorkflowCoordinator?.(undefined);
98100
this.notify();
99101
}
100102

@@ -166,7 +168,7 @@ export class WorkflowHost {
166168
this.listen(runtime);
167169
this.runtime = runtime;
168170
this.coordinator = coordinator;
169-
this.args.getDirector()?.setWorkflowCoordinator(coordinator);
171+
this.args.getDirector()?.setWorkflowCoordinator?.(coordinator);
170172
if (restore !== true) {
171173
runtime.start(workflow);
172174
this.persist();

‎tests/unit/workflow-host.test.ts‎

Lines changed: 53 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -43,6 +43,12 @@ async function withHost(
4343
home: string,
4444
) => void | Promise<void>,
4545
onChange?: () => void,
46+
// Overrides the director seam: defaults to a tracking director, pass
47+
// () => undefined (not yet built) or () => ({}) (no workflow support) to
48+
// exercise graceful degradation.
49+
getDirector?: () =>
50+
| { setWorkflowCoordinator?: (c: WorkflowCoordinator | undefined) => void }
51+
| undefined,
4652
): Promise<void> {
4753
const cwd = await mkdtemp(join(tmpdir(), "wf-host-"));
4854
const home = await mkdtemp(join(tmpdir(), "wf-host-home-"));
@@ -54,11 +60,13 @@ async function withHost(
5460
cwd,
5561
getSessionId: () => "session-1",
5662
getToolDefinitions: () => tools,
57-
getDirector: () => ({
58-
setWorkflowCoordinator: (c) => {
59-
director.coordinator = c;
60-
},
61-
}),
63+
getDirector:
64+
getDirector ??
65+
(() => ({
66+
setWorkflowCoordinator: (c) => {
67+
director.coordinator = c;
68+
},
69+
})),
6270
home,
6371
...(onChange !== undefined ? { onChange } : {}),
6472
});
@@ -201,3 +209,43 @@ test("resume() restores an on-disk workflow snapshot for the session", async ()
201209
expect(director.coordinator).toBeInstanceOf(WorkflowCoordinator);
202210
});
203211
});
212+
213+
test("host degrades gracefully when no director is built yet", async () => {
214+
await withHost(
215+
[],
216+
async (host) => {
217+
expect(host.start("review")).toBe("Started review workflow.");
218+
expect(host.isActive()).toBe(true);
219+
host.reattach();
220+
expect(host.isActive()).toBe(true);
221+
host.reset();
222+
expect(host.isActive()).toBe(false);
223+
},
224+
undefined,
225+
() => undefined,
226+
);
227+
});
228+
229+
test("host degrades gracefully against a director without workflow support", async () => {
230+
await withHost(
231+
[],
232+
async (host, _director, cwd, home) => {
233+
const workflow = findWorkflow("review");
234+
expect(workflow).toBeDefined();
235+
const runtime = new WorkflowRuntime(new Map());
236+
runtime.start(defined(workflow, "workflow"));
237+
await saveWorkflowState(cwd, "session-1", runtime.state(), home);
238+
239+
await host.resume();
240+
expect(host.isActive()).toBe(true);
241+
host.reattach();
242+
expect(host.isActive()).toBe(true);
243+
expect(host.start("build")).toContain("again to replace");
244+
expect(host.start("build")).toBe("Started build workflow.");
245+
host.reset();
246+
expect(host.isActive()).toBe(false);
247+
},
248+
undefined,
249+
() => ({}),
250+
);
251+
});

0 commit comments

Comments
 (0)