diff --git a/docs/IMPLEMENTATION.md b/docs/IMPLEMENTATION.md index 0b3afc333..daeb5a553 100644 --- a/docs/IMPLEMENTATION.md +++ b/docs/IMPLEMENTATION.md @@ -212,7 +212,7 @@ Listings are list-free, dumps are dump-locked: a bounded `ls`/`tree` prints name - **Alt+Enter** queues a follow-up (kind `"queue"`) delivered only on **session-idle** — parent-idle **and** no live fleet lanes (`run` goes idle). Session-idle Alt+Enter is a no-op. **Ctrl+C** stops the run. -Idle-with-fleet is shipped: after a non-blocking `spawn_agent` dispatch the parent turn can settle while workers keep running. The runner emits a `fleet` event carrying the live-lane count; the bridge holds the run busy on that count, so mid-hold Enter upgrades to a new primary turn (sent immediately) instead of queueing a steer, follow-ups keep waiting for true session-idle, and any steer left pending at the hold's engagement delivers immediately — the parent it was steering has already stopped. While the hold is up and the parent is not processing, occupancy flushes mailbox mail (`driveMailboxMail` + `buildMailboxMailMessage`) on store subscribe and idle-with-fleet settle — one child done while siblings run is enough. Skip that shot when a fleet-dry open-task continuation is latched. Exec-primary `wait_agents` mounts with no yield predicate (`createWaitAgentsTool({ sessions, fleetRecords })`) and blocks to ready/timeout/abort; parent interrupt/abort cancels that parked waiter via the tool AbortSignal (`timed_out`, workers stay running) rather than a second waiter table. `interrupt_agent` on a child unblocks wait as interrupted through the mailbox overlay. `shouldYieldWait` remains a supported-but-unwired tool option (no production mount passes it). +Idle-with-fleet is shipped: after a non-blocking `spawn_agent` dispatch the parent turn can settle while workers keep running. The runner emits a `fleet` event carrying the live-lane count; the bridge holds the run busy on that count, so mid-hold Enter upgrades to a new primary turn (sent immediately) instead of queueing a steer, follow-ups keep waiting for true session-idle, and any steer left pending at the hold's engagement delivers immediately — the parent it was steering has already stopped. While the hold is up and the parent is not processing, occupancy flushes mailbox mail (`driveMailboxMail` + `buildMailboxMailMessage`) on store subscribe and idle-with-fleet settle — one child done while siblings run is enough. Skip that shot when a fleet-dry open-task continuation is latched. Both wakes wait for send to settle as accepted before taking reports; a pending Promise is not a continuation, and a failed or uncertain send idles the parent. Exec-primary `wait_agents` mounts with no yield predicate (`createWaitAgentsTool({ sessions, fleetRecords })`) and blocks to ready/timeout/abort; parent interrupt/abort cancels that parked waiter via the tool AbortSignal (`timed_out`, workers stay running) rather than a second waiter table. `interrupt_agent` on a child unblocks wait as interrupted through the mailbox overlay. `shouldYieldWait` remains a supported-but-unwired tool option (no production mount passes it). `src/tui/stream-event-map.ts` maps reactor events onto the bridge's inbound events, and `src/tui/turn-state.ts` tracks the turn's status. `src/tui/turns-to-blocks.ts` hydrates a resumed session's stored turns into the same content blocks. diff --git a/docs/TUI.md b/docs/TUI.md index dedb3d2e1..f2e41b747 100644 --- a/docs/TUI.md +++ b/docs/TUI.md @@ -710,7 +710,9 @@ sends at once (the parent it was steering has stopped), and the last lane terminalizing releases the hold, drains follow-ups, and returns the session to idle — unless todo/doing tasks remain, in which case a system continuation starts before the fleet-0 event so the run stays busy and -follow-ups wait one more turn. The mail and that fleet-dry continuation +follow-ups wait one more turn. Occupancy wakes consume mailbox reports only +after send settles as accepted; a failed or uncertain send idles the parent +so a later flush can retry. The mail and that fleet-dry continuation are runtime-to-agent traffic — the fleet board owns worker status — so neither paints a transcript row, and neither rehydrates as one. diff --git a/src/subagent/fleet-dry-drive.test.ts b/src/subagent/fleet-dry-drive.test.ts index b50714794..ba4cacc6c 100644 --- a/src/subagent/fleet-dry-drive.test.ts +++ b/src/subagent/fleet-dry-drive.test.ts @@ -27,6 +27,33 @@ function peekMailbox( }; } +function collectingMailbox( + records: Map, +): FleetDryMailbox { + return { + 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 ACCEPTED_DELIVERY = { status: "accepted" as const }; +const NOT_DELIVERED_RESULT = { + status: "not-delivered" as const, + reason: "agent-closed" as const, + detail: "agent closed", +}; +const UNCERTAIN_DELIVERY = { + status: "uncertain" as const, + detail: "send raced", +}; + function fakeBlobStore() { const blobs = new Map(); return { @@ -512,6 +539,7 @@ describe("driveOpenTasksAfterFleetDry", () => { send: (prompt) => { order.push("send"); sent.push(prompt); + return ACCEPTED_DELIVERY; }, }); expect(driven).toBe(true); @@ -539,7 +567,7 @@ describe("driveOpenTasksAfterFleetDry", () => { beginSystemContinuation: (prompt) => { sent.push(prompt); }, - send: () => undefined, + send: () => ACCEPTED_DELIVERY, }); expect(driven).toBe(true); const parsed = reportsJSONFromPrompt(sent[0] ?? ""); @@ -624,6 +652,7 @@ describe("driveOpenTasksAfterFleetDry", () => { beginSystemContinuation: () => undefined, send: (prompt) => { sent.push(prompt); + return ACCEPTED_DELIVERY; }, }); expect(driven).toBe(true); @@ -677,7 +706,7 @@ describe("driveOpenTasksAfterFleetDry", () => { return taken; }, }; - const sendWithAttemptIdentity = async (): Promise => { + const sendWithAttemptIdentity = async (): Promise => { await Promise.resolve(); throw new Error("agentProxy.send failed"); }; @@ -690,14 +719,11 @@ describe("driveOpenTasksAfterFleetDry", () => { beginSystemContinuation: () => undefined, send: () => sendWithAttemptIdentity(), }); - expect(driven).toBe(true); - expect(records.get("w1")?.collected).not.toBe(true); - await Promise.resolve(); - await Promise.resolve(); + expect(driven).toBe(false); expect(records.get("w1")?.collected).not.toBe(true); }); - test("TUI sendWithAttemptIdentity false after handleSendFailure leaves mailbox uncollected", async () => { + test("TUI sendWithAttemptIdentity not-delivered after handleSendFailure leaves mailbox uncollected", async () => { const records = new Map([ ["w1", { status: "done", report: "ok" }], ]); @@ -712,9 +738,9 @@ describe("driveOpenTasksAfterFleetDry", () => { return taken; }, }; - const sendWithAttemptIdentity = async (): Promise => { + const sendWithAttemptIdentity = async () => { await Promise.resolve(); - return false; + return NOT_DELIVERED_RESULT; }; const driven = await driveOpenTasksAfterFleetDry({ deferredDryEdge: true, @@ -725,13 +751,11 @@ describe("driveOpenTasksAfterFleetDry", () => { beginSystemContinuation: () => undefined, send: () => sendWithAttemptIdentity(), }); - expect(driven).toBe(true); - await Promise.resolve(); - await Promise.resolve(); + expect(driven).toBe(false); expect(records.get("w1")?.collected).not.toBe(true); }); - test("TUI sendWithAttemptIdentity true takes mailbox after send resolves", async () => { + test("TUI sendWithAttemptIdentity accepted takes mailbox after send resolves", async () => { const records = new Map([ ["w1", { status: "done", report: "ok" }], ]); @@ -746,12 +770,18 @@ describe("driveOpenTasksAfterFleetDry", () => { return taken; }, }; - let resolveSend: ((ok: boolean) => void) | undefined; - const sendWithAttemptIdentity = (): Promise => - new Promise((resolve) => { + let resolveSend: ((result: typeof ACCEPTED_DELIVERY) => void) | undefined; + let sendStarted: (() => void) | undefined; + const sendSeen = new Promise((resolve) => { + sendStarted = resolve; + }); + const sendWithAttemptIdentity = (): Promise => { + sendStarted?.(); + return new Promise((resolve) => { resolveSend = resolve; }); - const driven = await driveOpenTasksAfterFleetDry({ + }; + const driven = driveOpenTasksAfterFleetDry({ deferredDryEdge: true, openTasks: [openTask], parentProcessing: false, @@ -760,14 +790,14 @@ describe("driveOpenTasksAfterFleetDry", () => { beginSystemContinuation: () => undefined, send: () => sendWithAttemptIdentity(), }); - expect(driven).toBe(true); + await sendSeen; expect(records.get("w1")?.collected).not.toBe(true); - resolveSend?.(true); - await Promise.resolve(); + resolveSend?.(ACCEPTED_DELIVERY); + expect(await driven).toBe(true); expect(records.get("w1")?.collected).toBe(true); }); - test("sync send false returns false, calls onSendFailure, and leaves mailbox uncollected", async () => { + test("sync send not-delivered returns false, calls onSendFailure, and leaves mailbox uncollected", async () => { const records = new Map([ ["w1", { status: "done", report: "ok" }], ]); @@ -791,7 +821,7 @@ describe("driveOpenTasksAfterFleetDry", () => { mailbox, lanes: [], beginSystemContinuation: () => undefined, - send: () => false, + send: () => NOT_DELIVERED_RESULT, onSendFailure: () => { failures += 1; }, @@ -822,24 +852,21 @@ describe("driveOpenTasksAfterFleetDry", () => { expect(failures).toBe(1); }); - test("TUI send false after handleSendFailure calls onSendFailure", async () => { + test("TUI send not-delivered after handleSendFailure calls onSendFailure", async () => { let failures = 0; - const driven = driveOpenTasksAfterFleetDry({ + const driven = await driveOpenTasksAfterFleetDry({ deferredDryEdge: true, openTasks: [openTask], parentProcessing: false, mailbox: undefined, lanes: [], beginSystemContinuation: () => undefined, - send: async () => false, + send: async () => NOT_DELIVERED_RESULT, onSendFailure: () => { failures += 1; }, }); - expect(failures).toBe(0); - expect(await driven).toBe(true); - await Promise.resolve(); - await Promise.resolve(); + expect(driven).toBe(false); expect(failures).toBe(1); }); @@ -860,9 +887,86 @@ describe("driveOpenTasksAfterFleetDry", () => { failures += 1; }, }); - expect(driven).toBe(true); - await Promise.resolve(); + expect(driven).toBe(false); + expect(failures).toBe(1); + }); + + test("pending send Promise does not resolve as a successful continuation", async () => { + const records = new Map([ + ["w1", { status: "done", report: "ok" }], + ]); + let resolveSend: ((result: typeof ACCEPTED_DELIVERY) => void) | undefined; + let sendStarted: (() => void) | undefined; + const sendSeen = new Promise((resolve) => { + sendStarted = resolve; + }); + const driven = driveOpenTasksAfterFleetDry({ + previousRunning: 1, + running: 0, + openTasks: [openTask], + parentProcessing: false, + mailbox: collectingMailbox(records), + lanes: [], + beginSystemContinuation: () => undefined, + send: () => { + sendStarted?.(); + return new Promise((resolve) => { + resolveSend = resolve; + }); + }, + }); + await sendSeen; + let settled: boolean | undefined; + void Promise.resolve(driven).then((value) => { + settled = value; + }); await Promise.resolve(); + expect(settled).toBeUndefined(); + expect(records.get("w1")?.collected).not.toBe(true); + resolveSend?.(ACCEPTED_DELIVERY); + expect(await driven).toBe(true); + expect(settled).toBe(true); + expect(records.get("w1")?.collected).toBe(true); + }); + + test("resolved not-delivered returns idle and leaves the wake waitable", async () => { + const records = new Map([ + ["w1", { status: "done", report: "ok" }], + ]); + let failures = 0; + const driven = await driveOpenTasksAfterFleetDry({ + previousRunning: 1, + running: 0, + openTasks: [openTask], + parentProcessing: false, + mailbox: collectingMailbox(records), + lanes: [], + beginSystemContinuation: () => undefined, + send: () => Promise.resolve(NOT_DELIVERED_RESULT), + onSendFailure: () => { + failures += 1; + }, + }); + expect(driven).toBe(false); expect(failures).toBe(1); + expect(records.get("w1")?.collected).not.toBe(true); + }); + + test("resolved uncertain returns idle and leaves the wake waitable", async () => { + const records = new Map([ + ["w1", { status: "done", report: "ok" }], + ]); + const driven = await driveOpenTasksAfterFleetDry({ + previousRunning: 1, + running: 0, + openTasks: [openTask], + parentProcessing: false, + mailbox: collectingMailbox(records), + lanes: [], + beginSystemContinuation: () => undefined, + send: () => Promise.resolve(UNCERTAIN_DELIVERY), + }); + expect(driven).toBe(false); + expect(records.get("w1")?.collected).not.toBe(true); }); }); diff --git a/src/subagent/fleet-dry-drive.ts b/src/subagent/fleet-dry-drive.ts index 1df7a07f6..65b3a67ac 100644 --- a/src/subagent/fleet-dry-drive.ts +++ b/src/subagent/fleet-dry-drive.ts @@ -77,10 +77,36 @@ export function shouldDriveOpenTasks(input: { ); } -function isPromiseLike(value: unknown): value is Promise { +export function isPromiseLike(value: unknown): value is Promise { return typeof value === "object" && value !== null && "then" in value; } +/** TUI send returns a delivery-result object. Only `accepted` consumes a wake. */ +function occupancySendSucceeded(result: unknown): boolean { + return ( + typeof result === "object" && + result !== null && + "status" in result && + result.status === "accepted" + ); +} + +export async function settleOccupancySend(args: { + send: () => unknown; + onSuccess: () => void; + onFailure: () => boolean; +}): Promise { + try { + const sent = args.send(); + const result = isPromiseLike(sent) ? await sent : sent; + if (!occupancySendSucceeded(result)) return args.onFailure(); + args.onSuccess(); + return true; + } catch { + return args.onFailure(); + } +} + /** Session blob-store write; same shape as ContextStore.writeBlob. */ export type FleetDryBlobWriter = ( key: string, @@ -285,23 +311,12 @@ async function driveOpenTasksAfterFleetDrySpill( }; try { args.beginSystemContinuation(prompt); - const sent = args.send(prompt); - if (isPromiseLike(sent)) { - void sent.then( - (result) => { - if (result !== false) takeReports(); - else args.onSendFailure?.(); - }, - () => { - args.onSendFailure?.(); - }, - ); - return true; - } - if (sent === false) return fail(); - takeReports(); } catch { return fail(); } - return true; + return settleOccupancySend({ + send: () => args.send(prompt), + onSuccess: takeReports, + onFailure: fail, + }); } diff --git a/src/subagent/mailbox-mail-drive.test.ts b/src/subagent/mailbox-mail-drive.test.ts index f800523a1..4b604d75c 100644 --- a/src/subagent/mailbox-mail-drive.test.ts +++ b/src/subagent/mailbox-mail-drive.test.ts @@ -30,6 +30,17 @@ function mapMailbox( }; } +const ACCEPTED_DELIVERY = { status: "accepted" as const }; +const NOT_DELIVERED_RESULT = { + status: "not-delivered" as const, + reason: "agent-closed" as const, + detail: "agent closed", +}; +const UNCERTAIN_DELIVERY = { + status: "uncertain" as const, + detail: "send raced", +}; + describe("buildMailboxMailPrompt", () => { test("prefixes collected JSON and tells the parent not to wait_agents", () => { const prompt = buildMailboxMailPrompt([ @@ -67,6 +78,7 @@ describe("driveMailboxMail", () => { send: (prompt) => { order.push("send"); sent.push(prompt); + return ACCEPTED_DELIVERY; }, }); expect(driven).toBe(true); @@ -90,6 +102,7 @@ describe("driveMailboxMail", () => { beginSystemContinuation: () => undefined, send: (prompt) => { sent.push(prompt); + return ACCEPTED_DELIVERY; }, }); expect(driven).toBe(true); @@ -173,7 +186,7 @@ describe("driveMailboxMail", () => { expect(records.get("w1")?.collected).not.toBe(true); }); - test("async send false after begin leaves mailbox uncollected", async () => { + test("async send not-delivered after begin leaves mailbox uncollected", async () => { const records = new Map([ ["w1", { status: "done", report: "ok" }], ]); @@ -182,11 +195,9 @@ describe("driveMailboxMail", () => { mailbox: mapMailbox(records), lanes: [], beginSystemContinuation: () => undefined, - send: () => Promise.resolve(false), + send: () => Promise.resolve(NOT_DELIVERED_RESULT), }); - expect(driven).toBe(true); - expect(records.get("w1")?.collected).not.toBe(true); - await Promise.resolve(); + expect(driven).toBe(false); expect(records.get("w1")?.collected).not.toBe(true); }); @@ -194,21 +205,27 @@ describe("driveMailboxMail", () => { const records = new Map([ ["w1", { status: "done", report: "ok" }], ]); - let resolveSend: ((ok: boolean) => void) | undefined; - const driven = await driveMailboxMail({ + let resolveSend: ((result: typeof ACCEPTED_DELIVERY) => void) | undefined; + let sendStarted: (() => void) | undefined; + const sendSeen = new Promise((resolve) => { + sendStarted = resolve; + }); + const driven = driveMailboxMail({ parentProcessing: false, mailbox: mapMailbox(records), lanes: [], beginSystemContinuation: () => undefined, - send: () => - new Promise((resolve) => { + send: () => { + sendStarted?.(); + return new Promise((resolve) => { resolveSend = resolve; - }), + }); + }, }); - expect(driven).toBe(true); + await sendSeen; expect(records.get("w1")?.collected).not.toBe(true); - resolveSend?.(true); - await Promise.resolve(); + resolveSend?.(ACCEPTED_DELIVERY); + expect(await driven).toBe(true); expect(records.get("w1")?.collected).toBe(true); }); @@ -218,20 +235,25 @@ describe("driveMailboxMail", () => { ]); const mailbox = mapMailbox(records); const sends: string[] = []; - let resolveSend: ((ok: boolean) => void) | undefined; - const driven = await driveMailboxMail({ + let resolveSend: ((result: typeof ACCEPTED_DELIVERY) => void) | undefined; + let sendStarted: (() => void) | undefined; + const sendSeen = new Promise((resolve) => { + sendStarted = resolve; + }); + const driven = driveMailboxMail({ parentProcessing: false, mailbox, lanes: [], beginSystemContinuation: () => undefined, send: (prompt) => { sends.push(prompt); - return new Promise((resolve) => { + sendStarted?.(); + return new Promise((resolve) => { resolveSend = resolve; }); }, }); - expect(driven).toBe(true); + await sendSeen; expect(sends).toHaveLength(1); expect(records.get("w1")?.collected).not.toBe(true); expect( @@ -248,8 +270,8 @@ describe("driveMailboxMail", () => { }), ).toBe(false); expect(sends).toHaveLength(1); - resolveSend?.(true); - await Promise.resolve(); + resolveSend?.(ACCEPTED_DELIVERY); + expect(await driven).toBe(true); expect(records.get("w1")?.collected).toBe(true); }); @@ -279,6 +301,7 @@ describe("driveMailboxMail", () => { beginSystemContinuation: () => undefined, send: (prompt) => { sends.push(prompt); + return ACCEPTED_DELIVERY; }, }), ).toBe(true); @@ -305,6 +328,47 @@ describe("driveMailboxMail", () => { }), ).toBe(false); }); + + test("not-delivered send leaves the wake retryable", async () => { + const records = new Map([ + ["w1", { status: "done", report: "ok" }], + ]); + const driven = await driveMailboxMail({ + parentProcessing: false, + mailbox: mapMailbox(records), + lanes: [], + beginSystemContinuation: () => undefined, + send: () => Promise.resolve(NOT_DELIVERED_RESULT), + }); + expect(driven).toBe(false); + expect(records.get("w1")?.collected).not.toBe(true); + expect( + await driveMailboxMail({ + parentProcessing: false, + mailbox: mapMailbox(records), + lanes: [], + beginSystemContinuation: () => undefined, + send: () => ACCEPTED_DELIVERY, + }), + ).toBe(true); + expect(records.get("w1")?.collected).toBe(true); + }); + + test("uncertain send leaves the wake retryable", async () => { + const records = new Map([ + ["w1", { status: "done", report: "ok" }], + ]); + expect( + await driveMailboxMail({ + parentProcessing: false, + mailbox: mapMailbox(records), + lanes: [], + beginSystemContinuation: () => undefined, + send: () => Promise.resolve(UNCERTAIN_DELIVERY), + }), + ).toBe(false); + expect(records.get("w1")?.collected).not.toBe(true); + }); }); describe("latchMailboxMailDrive", () => { @@ -326,6 +390,7 @@ describe("latchMailboxMailDrive", () => { send: (prompt) => { sends.push(prompt); resolveSend?.(); + return ACCEPTED_DELIVERY; }, }), ); @@ -368,6 +433,7 @@ describe("latchMailboxMailDrive", () => { send: (prompt) => { sends.push(prompt); sawSend?.(); + return ACCEPTED_DELIVERY; }, }), ); diff --git a/src/subagent/mailbox-mail-drive.ts b/src/subagent/mailbox-mail-drive.ts index 87cf8d1a7..2210b3f58 100644 --- a/src/subagent/mailbox-mail-drive.ts +++ b/src/subagent/mailbox-mail-drive.ts @@ -8,6 +8,7 @@ import { isLiveWaitStatus } from "./lifecycle.js"; import { collectUncollectedTerminals, + settleOccupancySend, type CollectedWorkerReport, type FleetDryBlobWriter, type FleetDryLane, @@ -61,10 +62,6 @@ export function isPersistedOccupancyWakeText(text: string): boolean { return isMailboxMailText(bare) || isFleetDryContinuationText(bare); } -function isPromiseLike(value: unknown): value is Promise { - return typeof value === "object" && value !== null && "then" in value; -} - /** * Ids occupancy has snapshotted and handed to send, but not yet taken. * A second flush must not start another parent turn for the same reports. @@ -157,25 +154,14 @@ async function driveMailboxMailAfterCollect( }; try { args.beginSystemContinuation(prompt); - const sent = args.send(prompt); - if (isPromiseLike(sent)) { - void sent.then( - (result) => { - if (result !== false) takeReports(); - else fail(); - }, - () => { - fail(); - }, - ); - return true; - } - if (sent === false) return fail(); - takeReports(); } catch { return fail(); } - return true; + return settleOccupancySend({ + send: () => args.send(prompt), + onSuccess: takeReports, + onFailure: fail, + }); } /** diff --git a/src/tui/runner/wiring.ts b/src/tui/runner/wiring.ts index 5c276169f..0c80ab94a 100644 --- a/src/tui/runner/wiring.ts +++ b/src/tui/runner/wiring.ts @@ -279,7 +279,7 @@ export function wirePostStartup( const send = state.sendWithAttemptIdentity; if (send === undefined) return false; const storage = state.currentStorage; - const driven = driveOpenTasksAfterFleetDry({ + return driveOpenTasksAfterFleetDry({ deferredDryEdge: true, openTasks: services.directorHolder.instance?.getTasks() ?? [], parentProcessing: false, @@ -299,9 +299,6 @@ export function wirePostStartup( sessionBridge.abortSystemContinuation(); }, }); - if (driven === false) return false; - void driven; - return true; }); sessionBridge.setMailboxMailDriver( latchMailboxMailDrive(() => { diff --git a/src/tui/runtime-bridge.test.ts b/src/tui/runtime-bridge.test.ts index c39a605a7..8457e7b51 100644 --- a/src/tui/runtime-bridge.test.ts +++ b/src/tui/runtime-bridge.test.ts @@ -2178,6 +2178,90 @@ describe("fleet-dry open-task drive (CL-7540)", () => { ); }); + test("pending occupancy Promise is not a continuation; failed send idles", 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 { + const prompt = + "The fleet has gone dry. Remaining open tasks:\n- t1: keep going (todo)\n"; + let resolveDrive: ((ok: boolean) => void) | undefined; + bridge.setDryOpenTaskDriver(() => { + bridge.beginSystemContinuation(prompt); + return new Promise((resolve) => { + resolveDrive = resolve; + }); + }); + bridge.submit("dispatch workers", "immediate"); + bridge.handle({ type: "fleet", running: 1 }); + settleToollessTurn(bridge); + expect(shell.session.run).toBe("busy"); + bridge.handle({ type: "fleet", running: 0 }); + expect(shell.session.run).toBe("busy"); + expect(resolveDrive).toBeDefined(); + resolveDrive?.(false); + await Promise.resolve(); + expect(shell.session.run).toBe("idle"); + port.clear(); + bridge.submit("operator turn", "immediate"); + expect(shell.session.run).toBe("busy"); + expect(port.calls.some((c) => c.op === "sendImmediate")).toBe(true); + } finally { + bridge.dispose(); + shell.dispose(); + } + }, + { width: 80, height: 24 }, + ); + }); + + test("pending occupancy abort after dispose does not idle or drain", 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); + const prompt = + "The fleet has gone dry. Remaining open tasks:\n- t1: keep going (todo)\n"; + let resolveDrive: ((ok: boolean) => void) | undefined; + bridge.setDryOpenTaskDriver(() => { + bridge.beginSystemContinuation(prompt); + return new Promise((resolve) => { + resolveDrive = resolve; + }); + }); + bridge.submit("dispatch workers", "immediate"); + bridge.handle({ type: "fleet", running: 1 }); + settleToollessTurn(bridge); + bridge.handle({ type: "fleet", running: 0 }); + expect(shell.session.run).toBe("busy"); + bridge.submit("when it finishes, summarize", "queue"); + expect(badgeCount(shell.session)).toBe(1); + port.clear(); + bridge.dispose(); + const runAfterDispose = shell.session.run; + expect(runAfterDispose).toBe("busy"); + resolveDrive?.(false); + await Promise.resolve(); + expect(shell.session.run).toBe(runAfterDispose); + expect(badgeCount(shell.session)).toBe(1); + expect(port.calls.some((c) => c.op === "deliver")).toBe(false); + shell.dispose(); + }, + { width: 80, height: 24 }, + ); + }); + test("occupancy send abort does not swallow a later matching inbound", async () => { await withTestRenderer( async (h) => { diff --git a/src/tui/runtime-bridge.ts b/src/tui/runtime-bridge.ts index c978bd3c1..2e86669c6 100644 --- a/src/tui/runtime-bridge.ts +++ b/src/tui/runtime-bridge.ts @@ -99,6 +99,7 @@ import { pendingAskWakeText, type PendingAskWake, } from "../subagent/fleet-report.js"; +import { isPromiseLike } from "../subagent/fleet-dry-drive.js"; /** Tool name a sub-agent dispatch call carries — its row gets live progress. */ const SPAWN_AGENT_TOOL_NAME = "spawn_agent"; @@ -277,9 +278,12 @@ export interface SessionBridge { /** * Occupancy owner for dry+open continuation. Called once per dry episode * from settleRunToIdle when the fleet is dry. Return true if a continuation - * was sent (run stays busy). + * was sent (run stays busy). A pending send Promise is not a continuation — + * settle waits for it and idles on failed/uncertain delivery. */ - setDryOpenTaskDriver: (driver: (() => boolean) | undefined) => void; + setDryOpenTaskDriver: ( + driver: (() => boolean | Promise) | undefined, + ) => void; /** * Occupancy owner for per-item mailbox mail. Called from idle-with-fleet * settle (like flushPendingAskWake) and from the store-subscribe driver. @@ -493,7 +497,7 @@ export interface BridgeBag { */ awaitingContinuationInference: boolean; /** Occupancy driver: collect+send when settle takes a dry-episode shot. */ - dryOpenTaskDriver: (() => boolean) | undefined; + dryOpenTaskDriver: (() => boolean | Promise) | undefined; /** * Occupancy driver for per-item mailbox mail (worker terminal/fail) while * the parent is idle, including idle-with-fleet. Not the fleet-0+open-tasks @@ -1222,6 +1226,14 @@ function occupancyHold(bag: BridgeBag): boolean { return bag.liveFleet > 0 || bag.awaitingContinuationInference; } +function releaseRunToIdle(shell: AppShell, bag: BridgeBag): void { + shell.session = setRunState(shell.session, "idle"); + bag.awaitingContinuationInference = false; + drainAtBoundary(shell, bag); + bag.flushPendingAskWake?.(); + bag.flushMailboxMail?.(); +} + /** * Release the run to idle and drain everything queued — but only at true * session-idle. A live fleet holds the run busy after the parent turn settles @@ -1248,25 +1260,39 @@ function settleRunToIdle(shell: AppShell, bag: BridgeBag): void { return; } if (!bag.droveOpenTasksThisDry) { - let driven = false; + let driven: boolean | Promise = false; try { - driven = bag.dryOpenTaskDriver?.() === true; + driven = bag.dryOpenTaskDriver?.() ?? false; } catch { driven = false; } + if (isPromiseLike(driven)) { + bag.droveOpenTasksThisDry = true; + const abortPendingDrive = (): void => { + if (bag.disposed) return; + bag.droveOpenTasksThisDry = false; + if (shell.session.run !== "busy") return; + bag.lastSentMessage = ""; + flushOpenRow(shell, bag); + bag.turnThinking = null; + shell.inFlightTool = null; + bag.turn = turnStateOnInterrupt(bag.turn, bag.now()); + releaseRunToIdle(shell, bag); + }; + void driven.then((ok) => { + if (ok === true) return; + abortPendingDrive(); + }, abortPendingDrive); + return; + } // Consume only on a real continuation. A false/no-op leaves the latch // open so a later missed-edge settle with open tasks can still fire. - if (driven) { + if (driven === true) { bag.droveOpenTasksThisDry = true; return; } } - shell.session = setRunState(shell.session, "idle"); - bag.awaitingContinuationInference = false; - // Full drain: soft steers first, then follow-ups (drainOrder). - drainAtBoundary(shell, bag); - bag.flushPendingAskWake?.(); - bag.flushMailboxMail?.(); + releaseRunToIdle(shell, bag); } function applyInbound(