Hold occupancy continuation through a stale connector reply

The occupancy shot re-arms processing during inference.done, so the
same turn's connector.reply settled idle and drained queued follow-ups.
Hold until the continuation's inference.start. Take mailbox reports
only after send actually succeeds on the TUI path.

Author: TheGreatAxios

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

@@ -72,6 +72,13 @@ describe("shouldDriveOpenTasks", () => {
         deferredDryEdge: true,
       }),
     ).toBe(true);
+    expect(
+      shouldDriveOpenTasks({
+        hasOpenTasks: true,
+        parentProcessing: false,
+        deferredDryEdge: true,
+      }),
+    ).toBe(true);
     expect(
       shouldDriveOpenTasks({
         previousRunning: 0,
@@ -158,7 +165,7 @@ describe("collectUncollectedTerminals", () => {
       }),
     ).toBe(true);
 
-    const reports = collectUncollectedTerminals(mailbox, sessions.list());
+    const reports = collectUncollectedTerminals(mailbox, sessions.list(), true);
     expect(reports.map((r) => r.agent_id).sort()).toEqual(["done", "fail"]);
     expect(reports.find((r) => r.agent_id === "done")).toEqual({
       agent_id: "done",
@@ -194,9 +201,11 @@ describe("collectUncollectedTerminals", () => {
         return taken;
       },
     };
-    const reports = collectUncollectedTerminals(mailbox, [
-      { id: "ghost", description: "from store", report: "store report" },
-    ]);
+    const reports = collectUncollectedTerminals(
+      mailbox,
+      [{ id: "ghost", description: "from store", report: "store report" }],
+      true,
+    );
     expect(reports).toEqual([
       {
         agent_id: "ghost",
@@ -217,10 +226,30 @@ describe("collectUncollectedTerminals", () => {
       peek: (id) => records.get(id),
       take: (id) => records.get(id),
     };
-    const reports = collectUncollectedTerminals(mailbox, []);
+    const reports = collectUncollectedTerminals(mailbox, [], true);
     expect(reports[0]?.report?.length).toBe(FLEET_DRY_REPORT_CHARS);
     expect(reports[0]?.report?.endsWith("…")).toBe(true);
   });
+
+  test("consume false peeks without take", () => {
+    const records = new Map<string, FleetDryMailboxRecord>([
+      ["w1", { status: "done", report: "ok" }],
+    ]);
+    const mailbox: FleetDryMailbox = {
+      ids: () => [...records.keys()],
+      peek: (id) => records.get(id),
+      take: (id) => {
+        const existing = records.get(id);
+        if (existing === undefined) return undefined;
+        const taken = { ...existing, collected: true };
+        records.set(id, taken);
+        return taken;
+      },
+    };
+    const reports = collectUncollectedTerminals(mailbox, [], false);
+    expect(reports).toEqual([{ agent_id: "w1", status: "done", report: "ok" }]);
+    expect(records.get("w1")?.collected).not.toBe(true);
+  });
 });
 
 describe("driveOpenTasksAfterFleetDry", () => {
@@ -368,4 +397,109 @@ describe("driveOpenTasksAfterFleetDry", () => {
     expect(driven).toBe(false);
     expect(records.get("w1")?.collected).not.toBe(true);
   });
+
+  test("TUI sendWithAttemptIdentity rejection leaves mailbox uncollected", async () => {
+    const records = new Map<string, FleetDryMailboxRecord>([
+      ["w1", { status: "done", report: "ok" }],
+    ]);
+    const mailbox: FleetDryMailbox = {
+      ids: () => [...records.keys()],
+      peek: (id) => records.get(id),
+      take: (id) => {
+        const existing = records.get(id);
+        if (existing === undefined) return undefined;
+        const taken = { ...existing, collected: true };
+        records.set(id, taken);
+        return taken;
+      },
+    };
+    const sendWithAttemptIdentity = async (): Promise<boolean> => {
+      await Promise.resolve();
+      throw new Error("agentProxy.send failed");
+    };
+    const driven = driveOpenTasksAfterFleetDry({
+      deferredDryEdge: true,
+      openTasks: [openTask],
+      parentProcessing: false,
+      mailbox,
+      lanes: [],
+      beginSystemContinuation: () => undefined,
+      send: () => sendWithAttemptIdentity(),
+    });
+    expect(driven).toBe(true);
+    expect(records.get("w1")?.collected).not.toBe(true);
+    await Promise.resolve();
+    await Promise.resolve();
+    expect(records.get("w1")?.collected).not.toBe(true);
+  });
+
+  test("TUI sendWithAttemptIdentity false after handleSendFailure leaves mailbox uncollected", async () => {
+    const records = new Map<string, FleetDryMailboxRecord>([
+      ["w1", { status: "done", report: "ok" }],
+    ]);
+    const mailbox: FleetDryMailbox = {
+      ids: () => [...records.keys()],
+      peek: (id) => records.get(id),
+      take: (id) => {
+        const existing = records.get(id);
+        if (existing === undefined) return undefined;
+        const taken = { ...existing, collected: true };
+        records.set(id, taken);
+        return taken;
+      },
+    };
+    const sendWithAttemptIdentity = async (): Promise<boolean> => {
+      await Promise.resolve();
+      return false;
+    };
+    const driven = driveOpenTasksAfterFleetDry({
+      deferredDryEdge: true,
+      openTasks: [openTask],
+      parentProcessing: false,
+      mailbox,
+      lanes: [],
+      beginSystemContinuation: () => undefined,
+      send: () => sendWithAttemptIdentity(),
+    });
+    expect(driven).toBe(true);
+    await Promise.resolve();
+    await Promise.resolve();
+    expect(records.get("w1")?.collected).not.toBe(true);
+  });
+
+  test("TUI sendWithAttemptIdentity true takes mailbox after send resolves", async () => {
+    const records = new Map<string, FleetDryMailboxRecord>([
+      ["w1", { status: "done", report: "ok" }],
+    ]);
+    const mailbox: FleetDryMailbox = {
+      ids: () => [...records.keys()],
+      peek: (id) => records.get(id),
+      take: (id) => {
+        const existing = records.get(id);
+        if (existing === undefined) return undefined;
+        const taken = { ...existing, collected: true };
+        records.set(id, taken);
+        return taken;
+      },
+    };
+    let resolveSend: ((ok: boolean) => void) | undefined;
+    const sendWithAttemptIdentity = (): Promise<boolean> =>
+      new Promise((resolve) => {
+        resolveSend = resolve;
+      });
+    const driven = driveOpenTasksAfterFleetDry({
+      deferredDryEdge: true,
+      openTasks: [openTask],
+      parentProcessing: false,
+      mailbox,
+      lanes: [],
+      beginSystemContinuation: () => undefined,
+      send: () => sendWithAttemptIdentity(),
+    });
+    expect(driven).toBe(true);
+    expect(records.get("w1")?.collected).not.toBe(true);
+    resolveSend?.(true);
+    await Promise.resolve();
+    expect(records.get("w1")?.collected).toBe(true);
+  });
 });

src/subagent/fleet-dry-drive.ts

@@ -46,15 +46,21 @@ export interface CollectedWorkerReport {
 }
 
 export function shouldDriveOpenTasks(input: {
-  previousRunning: number;
-  running: number;
+  previousRunning?: number | undefined;
+  running?: number | undefined;
   hasOpenTasks: boolean;
   parentProcessing: boolean;
   deferredDryEdge?: boolean;
 }): boolean {
-  const wentDry = input.running === 0 && input.previousRunning > 0;
+  const running = input.running ?? 0;
+  const previousRunning = input.previousRunning ?? 0;
+  const wentDry = running === 0 && previousRunning > 0;
   const dryEdge = wentDry || input.deferredDryEdge === true;
-  return dryEdge && input.running === 0 && input.hasOpenTasks && !input.parentProcessing;
+  return dryEdge && running === 0 && input.hasOpenTasks && !input.parentProcessing;
+}
+
+function isPromiseLike(value: unknown): value is Promise<unknown> {
+  return typeof value === "object" && value !== null && "then" in value;
 }
 
 function clipField(text: string | undefined): string | undefined {
@@ -112,7 +118,7 @@ function clipCollectedReport(report: CollectedWorkerReport): CollectedWorkerRepo
 export function collectUncollectedTerminals(
   mailbox: FleetDryMailbox | undefined,
   lanes: readonly FleetDryLane[],
-  consume = true,
+  consume: boolean,
 ): CollectedWorkerReport[] {
   if (mailbox === undefined) return [];
   const byId = new Map(lanes.map((lane) => [lane.id, lane]));
@@ -151,15 +157,15 @@ export function buildFleetDryContinuationPrompt(
 }
 
 export function driveOpenTasksAfterFleetDry(args: {
-  previousRunning: number;
-  running: number;
+  previousRunning?: number | undefined;
+  running?: number | undefined;
   openTasks: readonly Task[];
   parentProcessing: boolean;
   deferredDryEdge?: boolean;
   mailbox: FleetDryMailbox | undefined;
   lanes: readonly FleetDryLane[];
   beginSystemContinuation: (prompt: string) => void;
-  send: (prompt: string) => void;
+  send: (prompt: string) => unknown;
 }): boolean {
   const tasks = [...args.openTasks];
   if (
@@ -175,14 +181,26 @@ export function driveOpenTasksAfterFleetDry(args: {
   }
   const reports = collectUncollectedTerminals(args.mailbox, args.lanes, false);
   const prompt = buildFleetDryContinuationPrompt(tasks, reports);
+  const takeReports = (): void => {
+    for (const report of reports) {
+      args.mailbox?.take(report.agent_id);
+    }
+  };
   try {
     args.beginSystemContinuation(prompt);
-    args.send(prompt);
+    const sent = args.send(prompt);
+    if (isPromiseLike(sent)) {
+      void sent.then(
+        (result) => {
+          if (result !== false) takeReports();
+        },
+        () => undefined,
+      );
+      return true;
+    }
+    if (sent !== false) takeReports();
   } catch {
     return false;
   }
-  for (const report of reports) {
-    args.mailbox?.take(report.agent_id);
-  }
   return true;
 }

src/subagent/index.ts

@@ -25,18 +25,7 @@ export {
   type FleetWatch,
   type PendingAskWake,
 } from "./fleet-report.js";
-export {
-  buildFleetDryContinuationPrompt,
-  collectUncollectedTerminals,
-  driveOpenTasksAfterFleetDry,
-  FLEET_DRY_CONTINUATION_PREFIX,
-  projectMailboxRecord,
-  shouldDriveOpenTasks,
-  takeAndProjectMailboxRecord,
-  type CollectedWorkerReport,
-  type FleetDryLane,
-  type FleetDryMailbox,
-} from "./fleet-dry-drive.js";
+export { driveOpenTasksAfterFleetDry } from "./fleet-dry-drive.js";
 export {
   EMPTY_THRASH_STATE,
   nextThrashState,

src/tui/runner/state.ts

@@ -222,7 +222,7 @@ export interface RunnerState {
     attempt: InferenceAttemptIdentity,
     providerFailure: ProviderFailureAttempt,
   ) => void;
-  sendWithAttemptIdentity?: (message: InboundMessage) => Promise<void>;
+  sendWithAttemptIdentity?: (message: InboundMessage) => Promise<boolean>;
   sendUserPrompt?: (text: string, pending: readonly PendingImageAttachment[]) => Promise<void>;
   dispatchCommand?: (name: string, args: string) => void;
   newSession?: () => void;

src/tui/runner/submit.ts

@@ -255,7 +255,7 @@ export function createSubmitPath(
   };
   state.handleSendFailure = handleSendFailure;
 
-  const sendWithAttemptIdentity = async (message: InboundMessage): Promise<void> => {
+  const sendWithAttemptIdentity = async (message: InboundMessage): Promise<boolean> => {
     const attempt = live.attemptIdentity();
     const providerFailure = services.providerFailureAttempts.begin(attempt);
     try {
@@ -265,8 +265,10 @@ export function createSubmitPath(
       // decision on the correlationId signal channel so the parked run
       // resumes.
       await services.approvalResume.handle(result);
+      return true;
     } catch (error) {
       handleSendFailure(error, attempt, providerFailure);
+      return false;
     } finally {
       services.providerFailureAttempts.sendSettled(providerFailure);
     }

src/tui/runner/wiring.ts

@@ -171,8 +171,6 @@ export function wirePostStartup(
     const send = state.sendWithAttemptIdentity;
     if (send === undefined) return false;
     return driveOpenTasksAfterFleetDry({
-      previousRunning: 0,
-      running: 0,
       deferredDryEdge: true,
       openTasks: services.directorHolder.instance?.getTasks() ?? [],
       parentProcessing: false,
@@ -181,9 +179,7 @@ export function wirePostStartup(
       beginSystemContinuation: (prompt) => {
         sessionBridge.beginSystemContinuation(prompt);
       },
-      send: (prompt) => {
-        void send(buildFleetDryContinuationMessage(prompt));
-      },
+      send: (prompt) => send(buildFleetDryContinuationMessage(prompt)),
     });
   });
   const unsubscribeFleetReport = services.subAgentSessions.subscribe(() => {

src/tui/runtime-bridge.test.ts

@@ -1476,6 +1476,17 @@ describe("fleet-dry open-task drive (CL-7540)", () => {
           settleToollessTurn(bridge);
           expect(drives).toBe(1);
           expect(shell.session.run).toBe("busy");
+          bridge.submit("when it finishes, summarize", "queue");
+          expect(badgeCount(shell.session)).toBe(1);
+          port.clear();
+          bridge.handle({ type: "connector.reply", data: { content: "" } });
+          expect(shell.session.run).toBe("busy");
+          expect(badgeCount(shell.session)).toBe(1);
+          expect(port.calls.some((c) => c.op === "deliver")).toBe(false);
+          settleToollessTurn(bridge);
+          expect(drives).toBe(1);
+          expect(shell.session.run).toBe("idle");
+          expect(badgeCount(shell.session)).toBe(0);
         } finally {
           bridge.dispose();
           shell.dispose();