diff --git a/src/tui/runner/wiring.stall-poll.test.ts b/src/tui/runner/wiring.stall-poll.test.ts new file mode 100644 index 000000000..1c9ef1e9d --- /dev/null +++ b/src/tui/runner/wiring.stall-poll.test.ts @@ -0,0 +1,133 @@ +import { describe, expect, test } from "bun:test"; +import { attachSessionBridge, createRecordingPort } from "../runtime-bridge.js"; +import { createAppShell } from "../shell/index.js"; +import { withTestRenderer } from "../harness.js"; +import { createFleetStallPollTick } from "./wiring.js"; + +function settleToollessTurn( + bridge: ReturnType, +): void { + bridge.handle({ type: "inference.start", data: {} }); + bridge.handle({ type: "inference.done", data: {} }); +} + +describe("fleet stall poll tick (CL-7676)", () => { + test("tick runs the fleet report and re-flushes mailbox mail", () => { + const order: string[] = []; + const tick = createFleetStallPollTick( + () => { + order.push("report"); + }, + () => { + order.push("flush"); + }, + ); + tick(); + expect(order).toEqual(["report", "flush"]); + }); + + test("missed edge (driver throws once) self-heals on the next poll without duplicates", async () => { + await withTestRenderer( + async (h) => { + const shell = createAppShell(h.renderer, { + terminal: { columns: 80, rows: 24 }, + wireKeys: false, + run: "busy", + }); + const port = createRecordingPort(); + const bridge = attachSessionBridge(shell, port); + try { + let reports = 0; + let drives = 0; + let taken = false; + let failNext = true; + bridge.handle({ type: "fleet", running: 1 }); + settleToollessTurn(bridge); + bridge.setMailboxMailDriver(() => { + if (failNext) { + failNext = false; + throw new Error("send failed"); + } + if (taken) return false; + taken = true; + drives += 1; + return true; + }); + // The subscribe-time edge misses: the driver failure is swallowed as + // retryable, with no later edge while the fleet stays quiet. + bridge.flushMailboxMail(); + expect(drives).toBe(0); + + const tick = createFleetStallPollTick( + () => { + reports += 1; + }, + () => bridge.flushMailboxMail(), + ); + tick(); + expect(reports).toBe(1); + expect(drives).toBe(1); + // The report is taken: a second poll must not re-send. + tick(); + expect(reports).toBe(2); + expect(drives).toBe(1); + } finally { + bridge.dispose(); + shell.dispose(); + } + }, + { width: 80, height: 24 }, + ); + }); + + test("terminal landing mid-turn is delivered on the next poll after settle", async () => { + await withTestRenderer( + async (h) => { + const shell = createAppShell(h.renderer, { + terminal: { columns: 80, rows: 24 }, + wireKeys: false, + run: "idle", + }); + const port = createRecordingPort(); + const bridge = attachSessionBridge(shell, port); + try { + let drives = 0; + let taken = false; + let failSettleFlush = true; + bridge.setMailboxMailDriver(() => { + if (failSettleFlush) { + failSettleFlush = false; + throw new Error("settle send failed"); + } + if (taken) return false; + taken = true; + drives += 1; + return true; + }); + bridge.submit("dispatch workers", "immediate"); + bridge.handle({ type: "fleet", running: 1 }); + // Terminal lands while the parent is mid-turn: flush no-ops. + bridge.flushMailboxMail(); + expect(drives).toBe(0); + // Settle-time flush also misses (send fails, swallowed). No later + // store edge fires while the fleet stays quiet. + settleToollessTurn(bridge); + expect(drives).toBe(0); + + const tick = createFleetStallPollTick( + () => undefined, + () => bridge.flushMailboxMail(), + ); + tick(); + expect(drives).toBe(1); + tick(); + expect(drives).toBe(1); + } finally { + bridge.dispose(); + shell.dispose(); + } + }, + { width: 80, height: 24 }, + ); + }); +}); diff --git a/src/tui/runner/wiring.ts b/src/tui/runner/wiring.ts index 9c1bcaaa2..d9d682891 100644 --- a/src/tui/runner/wiring.ts +++ b/src/tui/runner/wiring.ts @@ -71,6 +71,27 @@ import { steerCount } from "../session-queue.js"; const tuiLogger = getLogger([LOG_NAMESPACE_ROOT, "tui"]); +/** + * One tick of the periodic fleet stall poll. + * + * The store-subscribe edge drives mailbox mail the moment a lane terminalizes, + * but that edge is missable: the parent may be mid-turn (`isProcessing`, so + * `flushMailboxMail` no-ops) or the driver send may fail (swallowed as + * retryable with no later edge when the fleet is otherwise quiet). Re-flushing + * here bounds the stall to one poll interval. Both halves are no-ops when + * there is nothing to say: `reportFleet` diffs, `flushMailboxMail` no-ops + * while processing or when no uncollected terminal waits. + */ +export function createFleetStallPollTick( + reportFleet: () => void, + flushMailboxMail: () => void, +): () => void { + return () => { + reportFleet(); + flushMailboxMail(); + }; +} + export function createFleetWakePublisher( sessions: RunnerServices["subAgentSessions"], emitter: RunnerServices["emitter"], @@ -265,7 +286,10 @@ export function wirePostStartup( }, FLEET_REPORT_SETTLE_MS); if (typeof fleetSettle.unref === "function") fleetSettle.unref(); }); - const fleetStallPoll = setInterval(reportFleet, FLEET_STALL_POLL_MS); + const fleetStallPollTick = createFleetStallPollTick(reportFleet, () => + sessionBridge.flushMailboxMail(), + ); + const fleetStallPoll = setInterval(fleetStallPollTick, FLEET_STALL_POLL_MS); if (typeof fleetStallPoll.unref === "function") fleetStallPoll.unref(); state.stopFleetReporting = (): void => { clearInterval(fleetStallPoll);