From 37c3c1f583f5386cad2de14293affe2bce840fba Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Wed, 16 Sep 2026 15:13:55 -0700 Subject: [PATCH 1/2] fix(tui): flush mailbox mail before ask wake on settle and gate close MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit settleRunToIdle (idle-with-fleet) and gateClosed flushed the ask wake before mailbox mail, letting a re-surface wake steal the turn occupancy should take. Match the #1080 abort-path rule: mail first, wake only if mail did not start a turn (isProcessing guard, deliveredAskWake dedupe — no duplicate wake). Tests: 4 new CL-8061 order/suppression cases in agent-ask-wake.test.ts (failed pre-fix, pass post-fix). Fixes CL-8061 --- src/tui/agent-ask-wake.test.ts | 80 ++++++++++++++++++++++++++++++++++ src/tui/runtime-bridge.ts | 8 +++- 2 files changed, 86 insertions(+), 2 deletions(-) diff --git a/src/tui/agent-ask-wake.test.ts b/src/tui/agent-ask-wake.test.ts index 67ba9fa95..8bddb8c68 100644 --- a/src/tui/agent-ask-wake.test.ts +++ b/src/tui/agent-ask-wake.test.ts @@ -405,6 +405,86 @@ describe("agent ask wake delivery", () => { }); }); + test("gate close flushes mailbox mail before the ask wake (CL-8061)", async () => { + await withWakeBridge((bridge, sends) => { + const order: string[] = []; + bridge.setOnAskWakeSent(() => { + order.push("wake"); + }); + bridge.setMailboxMailDriver(() => { + order.push("mail"); + return false; + }); + bridge.gateOpened(); + bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] }); + expect(sends).toEqual([]); + bridge.gateClosed(); + expect(order).toEqual(["mail", "wake"]); + expect(sends).toHaveLength(1); + }); + }); + + test("gate close suppresses the wake when mail takes the turn (CL-8061)", async () => { + await withWakeBridge((bridge, sends) => { + const order: string[] = []; + bridge.setOnAskWakeSent(() => { + order.push("wake"); + }); + bridge.setMailboxMailDriver(() => { + order.push("mail"); + bridge.beginSystemContinuation("mailbox occupancy"); + return true; + }); + bridge.gateOpened(); + bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] }); + expect(sends).toEqual([]); + bridge.gateClosed(); + expect(order).toEqual(["mail"]); + expect(sends).toEqual([]); + }); + }); + + test("idle-with-fleet settle flushes mailbox mail before the ask wake (CL-8061)", async () => { + await withWakeBridge((bridge, sends) => { + const order: string[] = []; + bridge.setOnAskWakeSent(() => { + order.push("wake"); + }); + bridge.setMailboxMailDriver(() => { + order.push("mail"); + return false; + }); + bridge.handle({ type: "inference.start", data: {} }); + bridge.handle({ type: "fleet", running: 1 }); + bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] }); + expect(sends).toEqual([]); + bridge.handle({ type: "inference.done", data: {} }); + expect(order).toEqual(["mail", "wake"]); + expect(sends).toHaveLength(1); + }); + }); + + test("idle-with-fleet settle suppresses the wake when mail takes the turn (CL-8061)", async () => { + await withWakeBridge((bridge, sends) => { + const order: string[] = []; + bridge.setOnAskWakeSent(() => { + order.push("wake"); + }); + bridge.setMailboxMailDriver(() => { + order.push("mail"); + bridge.beginSystemContinuation("mailbox occupancy"); + return true; + }); + bridge.handle({ type: "inference.start", data: {} }); + bridge.handle({ type: "fleet", running: 1 }); + bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] }); + expect(sends).toEqual([]); + bridge.handle({ type: "inference.done", data: {} }); + expect(order).toEqual(["mail"]); + expect(sends).toEqual([]); + }); + }); + test("clearQueuedDelivery drops stashed wakes and resets delivered identities", async () => { await withWakeBridge((bridge, sends) => { bridge.handle({ type: "inference.start", data: {} }); diff --git a/src/tui/runtime-bridge.ts b/src/tui/runtime-bridge.ts index aa340223f..e55ddf481 100644 --- a/src/tui/runtime-bridge.ts +++ b/src/tui/runtime-bridge.ts @@ -1322,9 +1322,11 @@ function settleRunToIdle(shell: AppShell, bag: BridgeBag): void { // Hold: the fleet is still live, so the run stays busy. Steers left // pending send now — the parent they were steering has stopped, so // each one starts its own turn — while follow-ups keep waiting. + // Occupancy/mailbox first so a re-surface wake cannot steal that turn; + // the trailing wake flush restates only if mail did not start one. drainSteersAtBoundary(shell, bag); - bag.flushPendingAskWake?.(); bag.flushMailboxMail?.(); + if (!bag.turn.isProcessing) bag.flushPendingAskWake?.(); return; } if (!bag.droveOpenTasksThisDry) { @@ -2032,8 +2034,10 @@ export function attachSessionBridge( if (bag.disposed) return; bag.turn = turnStateGateClosed(bag.turn, now()); paintPhase(); - flushPendingAskWake(); + // Occupancy/mailbox first so a re-surface wake cannot steal that turn; + // the trailing wake flush restates only if mail did not start one. flushMailboxMail(); + if (!bag.turn.isProcessing) flushPendingAskWake(); }; const tick = (): void => { From 2e00ef9bd5079d88d6d35fc4a532855add81bad3 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Wed, 16 Sep 2026 16:04:37 -0700 Subject: [PATCH 2/2] fix(tui): claim the primary turn for occupancy before ask wake Mailbox mail collect is async, so isProcessing is still false while occupancy already owns the slot. A shared mail-then-wake guard uses the latch claim so ask-wake cannot send on that stack. --- src/subagent/mailbox-mail-drive.test.ts | 47 ++++++++- src/subagent/mailbox-mail-drive.ts | 49 ++++++++- src/tui/agent-ask-wake.test.ts | 135 +++++++++++++++++++++++- src/tui/runner/wiring.ts | 1 + src/tui/runtime-bridge.ts | 51 +++++---- 5 files changed, 257 insertions(+), 26 deletions(-) diff --git a/src/subagent/mailbox-mail-drive.test.ts b/src/subagent/mailbox-mail-drive.test.ts index 4b604d75c..f2799b8ab 100644 --- a/src/subagent/mailbox-mail-drive.test.ts +++ b/src/subagent/mailbox-mail-drive.test.ts @@ -132,7 +132,7 @@ describe("driveMailboxMail", () => { }), ).toBe(false); expect( - await driveMailboxMail({ + driveMailboxMail({ parentProcessing: false, mailbox: mapMailbox(new Map()), lanes: [], @@ -315,7 +315,7 @@ describe("driveMailboxMail", () => { ["live", { status: "running" }], ]); expect( - await driveMailboxMail({ + driveMailboxMail({ parentProcessing: false, mailbox: mapMailbox(records), lanes: [], @@ -329,6 +329,28 @@ describe("driveMailboxMail", () => { ).toBe(false); }); + test("does not begin if the parent starts processing during collect", async () => { + const records = new Map([ + ["w1", { status: "done", report: "ok" }], + ]); + let processing = false; + const driven = driveMailboxMail({ + parentProcessing: false, + isParentProcessing: () => processing, + mailbox: mapMailbox(records), + lanes: [], + beginSystemContinuation: () => { + throw new Error("must not begin"); + }, + send: () => { + throw new Error("must not send"); + }, + }); + processing = true; + expect(await driven).toBe(false); + expect(records.get("w1")?.collected).not.toBe(true); + }); + test("not-delivered send leaves the wake retryable", async () => { const records = new Map([ ["w1", { status: "done", report: "ok" }], @@ -395,8 +417,10 @@ describe("latchMailboxMailDrive", () => { }), ); expect(driver()).toBe(true); + expect(driver.claimed()).toBe(true); expect(driver()).toBe(false); expect(driver()).toBe(false); + expect(driver.claimed()).toBe(true); await sent; expect(sends).toHaveLength(1); expect(sends[0]).toContain(MAILBOX_MAIL_WAKE_PREFIX); @@ -410,9 +434,28 @@ describe("latchMailboxMailDrive", () => { }); expect(driver()).toBe(false); expect(driver()).toBe(false); + expect(driver.claimed()).toBe(false); expect(calls).toBe(2); }); + test("an empty mailbox does not claim the occupancy slot", () => { + const driver = latchMailboxMailDrive(() => + driveMailboxMail({ + parentProcessing: false, + mailbox: mapMailbox(new Map()), + lanes: [], + beginSystemContinuation: () => { + throw new Error("must not begin"); + }, + send: () => { + throw new Error("must not send"); + }, + }), + ); + expect(driver()).toBe(false); + expect(driver.claimed()).toBe(false); + }); + test("after the in-flight drive settles, a new terminal can send", async () => { const records = new Map([ ["first", { status: "done", report: "one" }], diff --git a/src/subagent/mailbox-mail-drive.ts b/src/subagent/mailbox-mail-drive.ts index 2210b3f58..7fe018b55 100644 --- a/src/subagent/mailbox-mail-drive.ts +++ b/src/subagent/mailbox-mail-drive.ts @@ -109,8 +109,32 @@ export function buildMailboxMailPrompt( return [mailboxMailWakeLine(), JSON.stringify(reports)].join("\n"); } +function hasDeliverableMailboxMail( + mailbox: FleetDryMailbox | undefined, +): boolean { + if (mailbox === undefined) return false; + const delivering = deliveringByMailbox.get(mailbox); + for (const id of mailbox.ids()) { + if (delivering?.has(id)) continue; + const record = mailbox.peek(id); + if (record === undefined || record.collected === true) continue; + if (isLiveWaitStatus(record.status)) continue; + return true; + } + return false; +} + +function parentIsProcessing(args: { + parentProcessing: boolean; + isParentProcessing?: () => boolean; +}): boolean { + return args.isParentProcessing?.() ?? args.parentProcessing; +} + export function driveMailboxMail(args: { parentProcessing: boolean; + /** Live check after awaited collect; occupancy must not start a turn already claimed. */ + isParentProcessing?: () => boolean; mailbox: FleetDryMailbox | undefined; lanes: readonly FleetDryLane[]; writeBlob?: FleetDryBlobWriter; @@ -118,7 +142,8 @@ export function driveMailboxMail(args: { send: (prompt: string) => unknown; onSendFailure?: () => void; }): boolean | Promise { - if (args.parentProcessing) return false; + if (parentIsProcessing(args)) return false; + if (!hasDeliverableMailboxMail(args.mailbox)) return false; return driveMailboxMailAfterCollect(args); } @@ -138,6 +163,7 @@ async function driveMailboxMailAfterCollect( ) ).filter((report) => !delivering.has(report.agent_id)); if (reports.length === 0) return false; + if (parentIsProcessing(args)) return false; const ids = reports.map((report) => report.agent_id); for (const id of ids) delivering.add(id); const prompt = buildMailboxMailPrompt(reports); @@ -169,12 +195,28 @@ async function driveMailboxMailAfterCollect( * mail. `driveMailboxMail` does not mark the parent busy until after an * awaited collect, so overlapping flushes would each call `send()` and fill * the agent's depth-16 queue. Hold one drive until that promise settles. + * + * `flush()` is true only when this call starts a drive. `claimed()` is true + * while that drive owns the slot — ask-wake admission uses it so a wake + * cannot send on the same stack, or on a later stack before collect lands. */ +export type MailboxMailFlush = (() => boolean) & { + readonly claimed: () => boolean; +}; + +export function mailboxMailDriveClaimed( + flush: (() => boolean) | undefined, +): boolean { + if (flush === undefined) return false; + const claimed = (flush as MailboxMailFlush).claimed; + return typeof claimed === "function" && claimed() === true; +} + export function latchMailboxMailDrive( drive: () => boolean | Promise, -): () => boolean { +): MailboxMailFlush { let inFlight = false; - return () => { + const flush = (): boolean => { if (inFlight) return false; const driven = drive(); if (driven === false) return false; @@ -184,4 +226,5 @@ export function latchMailboxMailDrive( }); return true; }; + return Object.assign(flush, { claimed: () => inFlight }); } diff --git a/src/tui/agent-ask-wake.test.ts b/src/tui/agent-ask-wake.test.ts index 8bddb8c68..799430091 100644 --- a/src/tui/agent-ask-wake.test.ts +++ b/src/tui/agent-ask-wake.test.ts @@ -4,6 +4,7 @@ import { createLiveSessionPort } from "./live-session-port"; import { createAppShell } from "./shell/index"; import { withTestRenderer } from "./harness"; import type { PendingAskWake } from "../subagent/fleet-report.js"; +import { latchMailboxMailDrive } from "../subagent/mailbox-mail-drive.js"; import { classifySubmission, createSubmitHandler } from "./runner/submit.js"; import { createDeliveryGeneration, @@ -33,7 +34,7 @@ async function withWakeBridge( run: ( bridge: ReturnType, sends: string[], - ) => void, + ) => void | Promise, ) { await withTestRenderer( async (h) => { @@ -55,7 +56,7 @@ async function withWakeBridge( }), ); try { - run(bridge, sends); + await run(bridge, sends); } finally { bridge.dispose(); shell.dispose(); @@ -65,6 +66,22 @@ async function withWakeBridge( ); } +function latchAsyncMailboxMail( + bridge: ReturnType, + order: string[], + hold: Promise, +) { + return latchMailboxMailDrive(() => { + order.push("mail"); + return (async () => { + await hold; + order.push("begin"); + bridge.beginSystemContinuation("mailbox occupancy"); + return true; + })(); + }); +} + for (const action of [ "retry", "interrupt", @@ -485,6 +502,120 @@ describe("agent ask wake delivery", () => { }); }); + test("async mailbox drive on gate close claims the slot before wake (CL-8061)", async () => { + await withWakeBridge(async (bridge, sends) => { + const order: string[] = []; + let release: () => void = () => undefined; + const hold = new Promise((resolve) => { + release = resolve; + }); + bridge.setOnAskWakeSent(() => { + order.push("wake"); + }); + bridge.setMailboxMailDriver(latchAsyncMailboxMail(bridge, order, hold)); + bridge.gateOpened(); + bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] }); + expect(sends).toEqual([]); + bridge.gateClosed(); + expect(order).toEqual(["mail"]); + expect(sends).toEqual([]); + expect(bridge.turn.isProcessing).toBe(false); + release(); + await hold; + await Promise.resolve(); + await Promise.resolve(); + expect(order).toEqual(["mail", "begin"]); + expect(bridge.turn.isProcessing).toBe(true); + expect(sends).toEqual([]); + }); + }); + + test("async mailbox drive on idle-with-fleet settle claims the slot before wake (CL-8061)", async () => { + await withWakeBridge(async (bridge, sends) => { + const order: string[] = []; + let release: () => void = () => undefined; + const hold = new Promise((resolve) => { + release = resolve; + }); + bridge.setOnAskWakeSent(() => { + order.push("wake"); + }); + bridge.setMailboxMailDriver(latchAsyncMailboxMail(bridge, order, hold)); + bridge.handle({ type: "inference.start", data: {} }); + bridge.handle({ type: "fleet", running: 1 }); + bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] }); + expect(sends).toEqual([]); + bridge.handle({ type: "inference.done", data: {} }); + expect(order).toEqual(["mail"]); + expect(sends).toEqual([]); + expect(bridge.turn.isProcessing).toBe(false); + release(); + await hold; + await Promise.resolve(); + await Promise.resolve(); + expect(order).toEqual(["mail", "begin"]); + expect(bridge.turn.isProcessing).toBe(true); + expect(sends).toEqual([]); + }); + }); + + test("releaseRunToIdle flushes mailbox mail before the ask wake (CL-8061)", async () => { + await withWakeBridge((bridge, sends) => { + const order: string[] = []; + bridge.setOnAskWakeSent(() => { + order.push("wake"); + }); + bridge.setMailboxMailDriver(() => { + order.push("mail"); + return false; + }); + bridge.handle({ type: "inference.start", data: {} }); + bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] }); + expect(sends).toEqual([]); + bridge.handle({ type: "inference.done", data: {} }); + expect(order).toEqual(["mail", "wake"]); + expect(sends).toHaveLength(1); + }); + }); + + test("releaseRunToIdle suppresses the wake when mail takes the turn (CL-8061)", async () => { + await withWakeBridge((bridge, sends) => { + const order: string[] = []; + bridge.setOnAskWakeSent(() => { + order.push("wake"); + }); + bridge.setMailboxMailDriver(() => { + order.push("mail"); + bridge.beginSystemContinuation("mailbox occupancy"); + return true; + }); + bridge.handle({ type: "inference.start", data: {} }); + bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] }); + expect(sends).toEqual([]); + bridge.handle({ type: "inference.done", data: {} }); + expect(order).toEqual(["mail"]); + expect(sends).toEqual([]); + }); + }); + + test("idle parent subscribe does not flush wake before mailbox mail (CL-8061)", async () => { + await withWakeBridge((bridge, sends) => { + const order: string[] = []; + bridge.setOnAskWakeSent(() => { + order.push("wake"); + }); + bridge.setMailboxMailDriver(() => { + order.push("mail"); + bridge.beginSystemContinuation("mailbox occupancy"); + return true; + }); + bridge.handle({ type: "agent-ask", asks: [wake("a1", "q1")] }); + expect(order).toEqual(["mail"]); + expect(sends).toEqual([]); + expect(bridge.turn.isProcessing).toBe(true); + }); + }); + test("clearQueuedDelivery drops stashed wakes and resets delivered identities", async () => { await withWakeBridge((bridge, sends) => { bridge.handle({ type: "inference.start", data: {} }); diff --git a/src/tui/runner/wiring.ts b/src/tui/runner/wiring.ts index eeb5688a0..473aaead3 100644 --- a/src/tui/runner/wiring.ts +++ b/src/tui/runner/wiring.ts @@ -362,6 +362,7 @@ export function wirePostStartup( const storage = state.currentStorage; return driveMailboxMail({ parentProcessing: sessionBridge.turn.isProcessing, + isParentProcessing: () => sessionBridge.turn.isProcessing, mailbox: services.toolset.fleetRecords, lanes: services.subAgentSessions.list(), ...(storage !== null diff --git a/src/tui/runtime-bridge.ts b/src/tui/runtime-bridge.ts index e55ddf481..03efb5f59 100644 --- a/src/tui/runtime-bridge.ts +++ b/src/tui/runtime-bridge.ts @@ -101,6 +101,7 @@ import { type PendingAskWake, } from "../subagent/fleet-report.js"; import { isPromiseLike } from "../subagent/fleet-dry-drive.js"; +import { mailboxMailDriveClaimed } from "../subagent/mailbox-mail-drive.js"; /** Tool name a sub-agent dispatch call carries — its row gets live progress. */ const SPAWN_AGENT_TOOL_NAME = "spawn_agent"; @@ -551,6 +552,12 @@ export interface BridgeBag { * `attachSessionBridge` so settle and fleet events share one gate. */ flushMailboxMail: (() => void) | null; + /** + * Mail first, then ask-wake only if occupancy did not claim the slot. + * Abort, gate close, idle-with-fleet settle, and true session-idle release + * share this guard — so a wake cannot send while a mailbox drive is in flight. + */ + flushOccupancyThenWake: (() => void) | null; /** * One occupancy shot per dry episode. Reset when a live lane starts. Consumed * only when the driver actually sends a continuation — a no-op (no open @@ -1298,8 +1305,7 @@ function releaseRunToIdle(shell: AppShell, bag: BridgeBag): void { shell.session = setRunState(shell.session, "idle"); bag.awaitingContinuationInference = false; drainAtBoundary(shell, bag); - bag.flushPendingAskWake?.(); - bag.flushMailboxMail?.(); + bag.flushOccupancyThenWake?.(); } /** @@ -1325,8 +1331,7 @@ function settleRunToIdle(shell: AppShell, bag: BridgeBag): void { // Occupancy/mailbox first so a re-surface wake cannot steal that turn; // the trailing wake flush restates only if mail did not start one. drainSteersAtBoundary(shell, bag); - bag.flushMailboxMail?.(); - if (!bag.turn.isProcessing) bag.flushPendingAskWake?.(); + bag.flushOccupancyThenWake?.(); return; } if (!bag.droveOpenTasksThisDry) { @@ -1400,12 +1405,14 @@ function applyInbound( bag.askWakeResurface.delete(sessionId); } } + // Occupancy first on the idle-parent subscribe path. An open gate defers + // both flushes to gateClosed so a wake cannot send before mail. // A fresh snapshot with nothing pending means no wake is owed, but a // still-armed silent turn must stay armed until the poll's expire-abort // (CL-8060, only when expireStaleAsks expired this tick) or stall abort // can interrupt it. Disarming here (expire / send_input) would leave // isProcessing hung; send_input emptying pending must not expire-abort. - bag.flushPendingAskWake?.(); + if (bag.turn.blockedGateCount === 0) bag.flushOccupancyThenWake?.(); return; } @@ -1503,6 +1510,7 @@ export function attachSessionBridge( turnMarkers: [], flushPendingAskWake: null, flushMailboxMail: null, + flushOccupancyThenWake: null, droveOpenTasksThisDry: false, awaitingContinuationInference: false, dryOpenTaskDriver: undefined, @@ -1864,9 +1872,21 @@ export function attachSessionBridge( enqueuedAt: now(), }); }; + const flushMailboxMail = (): boolean => { + if (bag.disposed || bag.turn.isProcessing) return false; + try { + if (bag.mailboxMailDriver?.() === true) return true; + } catch { + // Occupancy miss is retryable on the next idle/subscribe edge. + } + return mailboxMailDriveClaimed(bag.mailboxMailDriver); + }; + bag.flushMailboxMail = flushMailboxMail; + const flushPendingAskWake = (): void => { if (bag.disposed || bag.turn.isProcessing || bag.turn.blockedGateCount > 0) return; + if (mailboxMailDriveClaimed(bag.mailboxMailDriver)) return; const asks = [...bag.pendingAskWake.values()].filter( (ask) => bag.deliveredAskWake.get(ask.sessionId) !== ask.questionId, ); @@ -1892,15 +1912,11 @@ export function attachSessionBridge( }; bag.flushPendingAskWake = flushPendingAskWake; - const flushMailboxMail = (): void => { - if (bag.disposed || bag.turn.isProcessing) return; - try { - bag.mailboxMailDriver?.(); - } catch { - // Occupancy miss is retryable on the next idle/subscribe edge. - } + const flushOccupancyThenWake = (): void => { + if (flushMailboxMail() || bag.turn.isProcessing) return; + flushPendingAskWake(); }; - bag.flushMailboxMail = flushMailboxMail; + bag.flushOccupancyThenWake = flushOccupancyThenWake; /** * A stalled armed wake never landed: drop matching deliveredAskWake entries @@ -1950,8 +1966,7 @@ export function attachSessionBridge( bag.askWakeTurnArmed = false; bag.turn = turnStateOnInterrupt(bag.turn, now()); paintPhase(); - flushMailboxMail(); - if (!bag.turn.isProcessing) flushPendingAskWake(); + flushOccupancyThenWake(); }; /** @@ -2034,10 +2049,7 @@ export function attachSessionBridge( if (bag.disposed) return; bag.turn = turnStateGateClosed(bag.turn, now()); paintPhase(); - // Occupancy/mailbox first so a re-surface wake cannot steal that turn; - // the trailing wake flush restates only if mail did not start one. - flushMailboxMail(); - if (!bag.turn.isProcessing) flushPendingAskWake(); + flushOccupancyThenWake(); }; const tick = (): void => { @@ -2244,6 +2256,7 @@ export function attachSessionBridge( bag.askWakeResurface.clear(); bag.flushPendingAskWake = null; bag.flushMailboxMail = null; + bag.flushOccupancyThenWake = null; applyCadence(null); clearShellBridgeHooks(shell); bridges.delete(shell);