fix(tui): skip occupancy begin while the operator pause holds

Author: TheGreatAxios

src/subagent/fleet-dry-drive.test.ts

@@ -680,6 +680,31 @@ describe("driveOpenTasksAfterFleetDry", () => {
     ).toBe(false);
   });
 
+  test("does not send if begin does not claim the parent turn", async () => {
+    const records = recordsOf({
+      w1: { status: "done", report: "ok" },
+    });
+    const sends: string[] = [];
+    expect(
+      await driveOpenTasksAfterFleetDry({
+        previousRunning: 1,
+        running: 0,
+        openTasks: [openTask],
+        parentProcessing: false,
+        isParentProcessing: () => false,
+        mailbox: peekMailbox(records),
+        lanes: [],
+        beginSystemContinuation: () => undefined,
+        send: (prompt) => {
+          sends.push(prompt);
+          return ACCEPTED_DELIVERY;
+        },
+      }),
+    ).toBe(false);
+    expect(sends).toEqual([]);
+    expect(records.get("w1")?.collected).not.toBe(true);
+  });
+
   test("deferred dry edge after parentProcessing still collects and sends", async () => {
     const records = recordsOf({
       w1: { status: "done", report: "ok" },

src/subagent/fleet-dry-drive.ts

@@ -543,6 +543,8 @@ export function driveOpenTasksAfterFleetDry(args: {
   running?: number | undefined;
   openTasks: readonly Task[];
   parentProcessing: boolean;
+  /** Live check after awaited collect; occupancy must not start a turn already claimed. */
+  isParentProcessing?: () => boolean;
   deferredDryEdge?: boolean;
   mailbox: FleetDryMailbox | undefined;
   lanes: readonly FleetDryLane[];
@@ -601,6 +603,13 @@ async function driveOpenTasksAfterFleetDrySpill(
     const reports = dedupeByAgentId([...digested, ...stubs]);
     prompt = buildFleetDryContinuationPrompt(tasks, reports);
     args.beginSystemContinuation(prompt);
+    // begin is a no-op while paused. Do not send; release the delivering
+    // claim so a later explicit send can still deliver. The dry latch
+    // unclaims when this drive returns false.
+    if (args.isParentProcessing?.() === false) {
+      releaseOccupancyDelivering(args.mailbox, takeIds);
+      return false;
+    }
   } catch {
     return fail();
   }

src/subagent/mailbox-mail-drive.test.ts

@@ -578,6 +578,28 @@ describe("driveMailboxMail", () => {
     expect(records.get("w1")?.collected).not.toBe(true);
   });
 
+  test("does not send if begin does not claim the parent turn", async () => {
+    const records = recordsOf({
+      w1: { status: "done", report: "ok" },
+    });
+    const sends: string[] = [];
+    expect(
+      await driveMailboxMail({
+        parentProcessing: false,
+        isParentProcessing: () => false,
+        mailbox: collectingMailbox(records),
+        lanes: [],
+        beginSystemContinuation: () => undefined,
+        send: (prompt) => {
+          sends.push(prompt);
+          return ACCEPTED_DELIVERY;
+        },
+      }),
+    ).toBe(false);
+    expect(sends).toEqual([]);
+    expect(records.get("w1")?.collected).not.toBe(true);
+  });
+
   test("not-delivered send leaves the wake retryable", async () => {
     const records = await driveMailNotDelivered();
     expect(

src/subagent/mailbox-mail-drive.ts

@@ -179,6 +179,13 @@ async function driveMailboxMailAfterCollect(
       await digestCollectedReports(reports, args.writeBlob),
     );
     args.beginSystemContinuation(prompt);
+    // begin is a no-op while paused (and if disposed/empty). Do not send a
+    // primary, and release the delivering claim so a later explicit send can
+    // still deliver. The mailbox latch unclaims when this drive settles.
+    if (args.isParentProcessing?.() === false) {
+      releaseOccupancyDelivering(args.mailbox, ids);
+      return false;
+    }
   } catch {
     return fail();
   }

src/tui/agent-ask-wake.test.ts

@@ -669,6 +669,62 @@ describe("agent ask wake delivery", () => {
     );
   });
 
+  test("paused beginSystemContinuation does not mark the turn processing", async () => {
+    await withWakeBridge((bridge, sends, shell) => {
+      handleCtrlC(shell);
+      expect(isPaused(shell.session)).toBe(true);
+      bridge.beginSystemContinuation("mailbox occupancy");
+      expect(bridge.turn.isProcessing).toBe(false);
+      expect(shell.session.run).not.toBe("busy");
+      expect(sends).toEqual([]);
+    });
+  });
+
+  test("in-flight mailbox collect after pause does not start a primary", async () => {
+    await withWakeBridge(async (bridge, sends, shell) => {
+      const order: string[] = [];
+      let release: () => void = () => undefined;
+      const hold = new Promise<void>((resolve) => {
+        release = resolve;
+      });
+      const driver = latchAsyncMailboxMail(bridge, order, hold);
+      bridge.setMailboxMailDriver(driver);
+      bridge.flushMailboxMail();
+      expect(order).toEqual(["mail"]);
+      expect(driver.claimed()).toBe(true);
+      expect(bridge.turn.isProcessing).toBe(false);
+
+      handleCtrlC(shell);
+      expect(isPaused(shell.session)).toBe(true);
+
+      release();
+      await hold;
+      await Promise.resolve();
+      await Promise.resolve();
+
+      expect(order).toEqual(["mail", "begin"]);
+      expect(bridge.turn.isProcessing).toBe(false);
+      expect(sends).toEqual([]);
+      expect(shell.session.run).not.toBe("busy");
+      expect(isPaused(shell.session)).toBe(true);
+      expect(driver.claimed()).toBe(false);
+    });
+  });
+
+  test("unpaused interrupt still begins a system continuation", async () => {
+    await withWakeBridge((bridge, sends, shell) => {
+      bridge.handle({ type: "inference.start", data: {} });
+      expect(bridge.turn.isProcessing).toBe(true);
+      bridge.interrupt();
+      expect(isPaused(shell.session)).toBe(false);
+      expect(bridge.turn.isProcessing).toBe(false);
+      bridge.beginSystemContinuation("mailbox occupancy");
+      expect(bridge.turn.isProcessing).toBe(true);
+      expect(shell.session.run).toBe("busy");
+      expect(sends).toEqual([]);
+    });
+  });
+
   test("a wake question with bracket lines does not spoof attachment-echo matching", async () => {
     await withWakeBridge((bridge, sends, shell) => {
       const ask = {

src/tui/runner/wiring.ts

@@ -407,6 +407,7 @@ export function wirePostStartup(
       deferredDryEdge: true,
       openTasks: services.directorHolder.instance?.getTasks() ?? [],
       parentProcessing: false,
+      isParentProcessing: () => sessionBridge.turn.isProcessing,
       mailbox: services.toolset.fleetRecords,
       lanes: services.subAgentSessions.list(),
       ...(storage !== null

src/tui/runtime-bridge.ts

@@ -2482,6 +2482,10 @@ export function attachSessionBridge(
     },
     beginSystemContinuation: (text) => {
       if (bag.disposed) return;
+      // Occupancy that already claimed a latch can still call begin after an
+      // awaited collect. Operator pause must not start a primary; stall/expire
+      // do not set paused, so they still begin.
+      if (isPaused(shell.session)) return;
       const t = text.trim();
       if (t.length === 0) return;
       bag.lastSentMessage = t;