diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 6e73c41b..23f1834f 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -81,7 +81,14 @@ In TUI chat mode there is no completion gate — the session stays open across t - Wires `ask_operator` to an operator-gate event resolved by a modal - Mounts the OpenTUI host via `mountRunnerHost` (`src/tui/runner-host.ts`), which mounts `mountProductHost` (`src/tui/product-host.ts`) over the shell (`src/tui/shell.ts`) - Bridges reactor events to the OpenTUI host via a plain `EventEmitter` -- **Mid-run injection** — When a message arrives while the agent is running, it is queued in an `InjectionQueue`. On the next `inference.done` event (turn boundary), the queue is drained: each queued message is delivered via `agentProxy.deliver()` and a `"mid-run.delivered"` emitter event is fired so the badge count in the App updates. The queue is cleared on session rotation (`/clear`). +- **Mid-run injection** — Shell `session-queue` items drain at the parent + `tool.boundary` through `SessionPort.deliver`. Production `routeQueuedDelivery` + live-injects in-flight parent-boundary steers via `agentProxy.deliver` + (`Agent.deliver`) into the live reactor. Idle leftover, idle-with-fleet, and + post-interrupt steers, plus follow-ups (`kind === "queue"`), use the existing + send path. `/clear` and `/new` bump a + delivery generation and call `SessionBridge.clearQueuedDelivery()` so queued + input from the previous session cannot enter the new one. - **Session rotation** — Uses a serial session-operation queue (`createSessionOperationQueue`, not a boolean flag) so rotation, compaction continuation, and `agentProxy.deliver` never race a concurrent rebuild. Each operation chains onto the tail, ensuring in-flight work completes before the agent is torn down. ### Exec Runner (`src/exec/runner.ts`) diff --git a/docs/TUI.md b/docs/TUI.md index 55af3bbd..a8e954ab 100644 --- a/docs/TUI.md +++ b/docs/TUI.md @@ -457,17 +457,18 @@ the chord to point an operator at when Shift+Enter doesn't respond. Two mid-run gestures, two delivery times (CL-6290): - **Enter, mid-run** — soft steer: enqueues kind `"steer"` and delivers at the - next **parent** `tool.boundary` (the parent tool finishing, not a child). A + next **parent** `tool.boundary` (the parent tool finishing, not a child) via + `Agent.deliver` into the live reactor, not a new `send`. A long parent `run_shell` or an awaiting `task()` is parent-busy and holds steers. The transcript row says `[will steer next]` while pending and `[steering]` once delivered (`submitPrompt`, `drainSteersAtBoundary` in `runtime-bridge.ts`). - **Alt+Enter, mid-run** — follow-up: enqueues kind `"queue"` and delivers - only on **session-idle** (parent-idle and no live fleet lanes). Does not - interrupt or reinject. The transcript row says `[will follow up]` while - pending and `[following up]` once delivered. Idle, or with an empty prompt, - Alt+Enter does nothing — there is nothing to wait for. (Internal `"reinject"` - remains in the submit API for tests; no product chord wires it.) + only on **session-idle** (parent-idle and no live fleet lanes) as a `send`. + Does not interrupt or reinject. The transcript row says `[will follow up]` + while pending and `[following up]` once delivered. Idle, or with an empty + prompt, Alt+Enter does nothing — there is nothing to wait for. (Internal + `"reinject"` remains in the submit API for tests; no product chord wires it.) When `steer > 0` and a parent tool has been in flight ≥ `STEER_WAIT_NOTICE_MS` (3s), the notice row adds `waiting on ` (e.g. `waiting on run_shell`). @@ -480,7 +481,7 @@ events carrying the live-lane count and the bridge holds the run busy on it. During the hold, Enter upgrades to a new primary turn sent immediately — there is no parent tool left to steer — while Alt+Enter follow-ups keep waiting for true session-idle. A steer still pending when the hold engages -delivers at once (the parent it was steering has stopped), and the last lane +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. diff --git a/src/tui/keybindings.test.ts b/src/tui/keybindings.test.ts index b99cbd06..55491455 100644 --- a/src/tui/keybindings.test.ts +++ b/src/tui/keybindings.test.ts @@ -714,6 +714,7 @@ describe("the runner host does not shadow the prompt bindings the catalog claims eventEmitter: new EventEmitter(), send: () => {}, interrupt: () => {}, + deliver: () => {}, providers: {}, onModelSelect: () => {}, commands: [], diff --git a/src/tui/live-session-port.test.ts b/src/tui/live-session-port.test.ts index 71597de9..25583d23 100644 --- a/src/tui/live-session-port.test.ts +++ b/src/tui/live-session-port.test.ts @@ -8,7 +8,7 @@ type Call = | { op: "interrupt" } | { op: "deliver"; text: string; kind: QueueKind }; -function fakeDeps(opts?: { withDeliver?: boolean }) { +function fakeDeps() { const calls: Call[] = []; const deps = { send: (text: string) => { @@ -17,13 +17,9 @@ function fakeDeps(opts?: { withDeliver?: boolean }) { interrupt: () => { calls.push({ op: "interrupt" }); }, - ...(opts?.withDeliver - ? { - deliver: (text: string, kind: QueueKind) => { - calls.push({ op: "deliver", text, kind }); - }, - } - : {}), + deliver: (text: string, kind: QueueKind) => { + calls.push({ op: "deliver", text, kind }); + }, }; return { calls, deps }; } @@ -69,22 +65,12 @@ describe("createLiveSessionPort", () => { expect(calls).toEqual([{ op: "interrupt" }]); }); - test("deliver without deps.deliver falls back to send", () => { + test("deliver never calls send for steer or queue", () => { const { calls, deps } = fakeDeps(); const port = createLiveSessionPort(deps); port.deliver(item("queued msg", "queue")); port.deliver(item("steer msg", "steer", "q2")); - expect(calls).toEqual([ - { op: "send", text: "queued msg" }, - { op: "send", text: "steer msg" }, - ]); - }); - - test("deliver with deps.deliver passes text and kind", () => { - const { calls, deps } = fakeDeps({ withDeliver: true }); - const port = createLiveSessionPort(deps); - port.deliver(item("queued msg", "queue")); - port.deliver(item("steer msg", "steer", "q2")); + expect(calls.some((c) => c.op === "send")).toBe(false); expect(calls).toEqual([ { op: "deliver", text: "queued msg", kind: "queue" }, { op: "deliver", text: "steer msg", kind: "steer" }, @@ -92,7 +78,7 @@ describe("createLiveSessionPort", () => { }); test("full wiring: immediate → enqueue → deliver → interrupt", () => { - const { calls, deps } = fakeDeps({ withDeliver: true }); + const { calls, deps } = fakeDeps(); const port = createLiveSessionPort(deps); port.sendImmediate("start"); @@ -122,6 +108,7 @@ describe("attachment passthrough", () => { const port = createLiveSessionPort({ send: (_text, attachments) => seen.push(attachments), interrupt: () => {}, + deliver: () => {}, }); port.sendImmediate("look", [image]); expect(seen).toEqual([[image]]); diff --git a/src/tui/live-session-port.ts b/src/tui/live-session-port.ts index 8d44368f..16459c37 100644 --- a/src/tui/live-session-port.ts +++ b/src/tui/live-session-port.ts @@ -23,15 +23,8 @@ export interface LiveSessionPortDeps { ) => SubmitClassification; /** Hard interrupt current run (runner close/rebuild). */ interrupt: () => void; - /** - * Optional: drained queue/steer item at tool boundary (or idle). - * Defaults to `send(text)` for both kinds — v1 runner shares send. - */ - deliver?: ( - text: string, - kind: QueueKind, - attachments?: readonly PendingImageAttachment[], - ) => void; + /** Drained queue/steer item. Kind routing (live inject vs send) is the host's. */ + deliver: (text: string, kind: QueueKind, attachments?: readonly PendingImageAttachment[]) => void; } /** @@ -57,11 +50,7 @@ export function createLiveSessionPort(deps: LiveSessionPortDeps): SessionPort { deps.interrupt(); }, deliver: (item: QueueItem): void => { - if (deps.deliver) { - deps.deliver(item.text, item.kind, item.attachments); - return; - } - deps.send(item.text, item.attachments); + deps.deliver(item.text, item.kind, item.attachments); }, }; } diff --git a/src/tui/product-host.test.ts b/src/tui/product-host.test.ts index e9e1c438..30ac240d 100644 --- a/src/tui/product-host.test.ts +++ b/src/tui/product-host.test.ts @@ -24,15 +24,18 @@ import { buildModelsFirstCatalog, modelOptionId } from "./model-catalog.js"; function makeFakeSessionPort(): { readonly sends: string[]; + readonly delivers: string[]; readonly interrupts: number; readonly send: ProductHostConfig["send"]; readonly interrupt: ProductHostConfig["interrupt"]; - readonly deliver: NonNullable; + readonly deliver: ProductHostConfig["deliver"]; } { const sends: string[] = []; + const delivers: string[] = []; let interrupts = 0; return { sends, + delivers, get interrupts() { return interrupts; }, @@ -43,7 +46,7 @@ function makeFakeSessionPort(): { interrupts += 1; }, deliver: (text) => { - sends.push(text); + delivers.push(text); }, }; } @@ -224,6 +227,25 @@ describe("mountProductHost", () => { } }); + test("session.clear drops queued steers and idles the run (CL-7268)", async () => { + const { host, emitter } = await mountHeadless(); + try { + host.bridge.handle({ type: "run", state: "busy" }); + host.bridge.submit("old steer", "steer"); + expect(host.shell.session.run).toBe("busy"); + expect(host.shell.session.items.length).toBe(1); + + emitter.emit("event", { type: "user", text: "old prompt" }); + emitter.emit("session.clear"); + + expect(host.shell.streamLog).toEqual([]); + expect(host.shell.session.items).toEqual([]); + expect(host.shell.session.run).toBe("idle"); + } finally { + host.dispose(); + } + }); + test("permission.gate opens the overlay and resolves through the emitter's resolve callback", async () => { const { host, emitter } = await mountHeadless(); try { @@ -390,6 +412,7 @@ describe("flat type-to-filter model picker", () => { eventEmitter: new EventEmitter(), send: port.send, interrupt: port.interrupt, + deliver: port.deliver, createRenderer: async () => harness.renderer, models: catalog, onModelSelect: (id) => selected.push(id), @@ -477,6 +500,7 @@ describe("flat type-to-filter model picker", () => { eventEmitter: new EventEmitter(), send: port.send, interrupt: port.interrupt, + deliver: port.deliver, createRenderer: async () => harness.renderer, models: catalog, activeModelId: () => modelOptionId("xai/thegreataxios", "grok-4.5"), @@ -509,6 +533,7 @@ describe("flat type-to-filter model picker", () => { eventEmitter: new EventEmitter(), send: port.send, interrupt: port.interrupt, + deliver: port.deliver, createRenderer: async () => harness.renderer, models: catalog, activeModelId: () => modelOptionId("codex/abk-labs", "gpt-5.5"), @@ -536,6 +561,7 @@ describe("flat type-to-filter model picker", () => { eventEmitter: new EventEmitter(), send: port.send, interrupt: port.interrupt, + deliver: port.deliver, createRenderer: async () => harness.renderer, models: catalog, onModelSelect: () => {}, @@ -865,6 +891,7 @@ describe("mount failure", () => { eventEmitter: emitter, send: port.send, interrupt: port.interrupt, + deliver: port.deliver, createRenderer: async () => harness.renderer, }), ).rejects.toThrow("gate wiring failed"); diff --git a/src/tui/product-host.ts b/src/tui/product-host.ts index 915411e7..eb76cf91 100644 --- a/src/tui/product-host.ts +++ b/src/tui/product-host.ts @@ -119,7 +119,7 @@ export interface ProductHostConfig { */ readonly classifySubmit?: ProductHostClassifySubmit; readonly interrupt: ProductHostInterrupt; - readonly deliver?: ProductHostDeliver; + readonly deliver: ProductHostDeliver; /** Model/provider rows for the picker (id applied on select). */ readonly models?: readonly ProductHostModelOption[]; /** @@ -319,8 +319,8 @@ export async function mountProductHost(config: ProductHostConfig): Promise { }); }); +describe("ingestOperatorPrompt", () => { + test("merges pending attachments and expands a missing @mention", async () => { + const pending = attachment("clip.png"); + const result = await ingestOperatorPrompt( + "use @missing.ts", + "/repo", + async () => { + throw new Error("must not load"); + }, + [pending], + ); + expect(result.text).toContain("@missing.ts (not found)"); + expect(result.attachments).toEqual([pending]); + }); + + test("does not send — only returns ingested text and attachments", async () => { + const result = await ingestOperatorPrompt("just words", "/repo", async () => { + throw new Error("must not load"); + }); + expect(result.text).toBe("just words"); + expect(result.attachments).toEqual([]); + }); +}); + describe("spliceMentionCompletion", () => { test("replaces the typed token and keeps the trailing text", () => { const value = "read @src/tu rest"; diff --git a/src/tui/prompt-attachments.ts b/src/tui/prompt-attachments.ts index 3d6b9382..7491ca7c 100644 --- a/src/tui/prompt-attachments.ts +++ b/src/tui/prompt-attachments.ts @@ -9,6 +9,7 @@ import { type AttachImageResult, type PendingImageAttachment, } from "./image-attachments.js"; +import { resolveAtMentions } from "./mention-resolution.js"; export type { PendingImageAttachment }; @@ -42,6 +43,21 @@ export async function ingestPathMentions( return { text: out, attachments }; } +/** + * Shared operator-prompt ingest for send and live-steer deliver: inline image + * paths become attachments and @mentions are expanded. Does not send. + */ +export async function ingestOperatorPrompt( + text: string, + cwd: string, + load: (path: string) => Promise, + pending: readonly PendingImageAttachment[] = [], +): Promise { + const ingested = await ingestPathMentions(text, cwd, load); + const resolved = await resolveAtMentions(ingested.text, cwd); + return { text: resolved, attachments: [...pending, ...ingested.attachments] }; +} + export interface MentionSplice { readonly value: string; readonly cursor: number; diff --git a/src/tui/queued-delivery-hop.test.ts b/src/tui/queued-delivery-hop.test.ts new file mode 100644 index 00000000..bd59dd67 --- /dev/null +++ b/src/tui/queued-delivery-hop.test.ts @@ -0,0 +1,269 @@ +/** + * Last-hop pins for drained queue items: live parent-boundary steers + * Agent.deliver (deliverSteer); leftover / fleet-hold / interrupt use send. + */ +import { describe, expect, test } from "bun:test"; +import { attachSessionBridge, type SessionBridge } from "./runtime-bridge"; +import { createLiveSessionPort } from "./live-session-port"; +import { createAppShell } from "./shell"; +import { withTestRenderer } from "./harness"; +import { createLiveSteerDeliver, routeQueuedDelivery } from "./queued-delivery.js"; +import { createSessionOperationQueue } from "./session-operation-queue.js"; +import { badgeCount } from "./session-queue"; + +function lastHopPort(bridgeRef: { current: SessionBridge | undefined }) { + const sends: string[] = []; + const steers: string[] = []; + const port = createLiveSessionPort({ + send: (text) => { + sends.push(text); + }, + interrupt: () => {}, + deliver: routeQueuedDelivery({ + send: (text) => { + sends.push(text); + }, + deliverSteer: (text) => { + steers.push(text); + }, + parentCycleLive: () => bridgeRef.current?.parentCycleLive === true, + }), + }); + return { port, sends, steers }; +} + +describe("queued delivery last hop", () => { + test("busy parent tool.boundary steer last-hops to deliverSteer, not send", async () => { + await withTestRenderer( + async (h) => { + const shell = createAppShell(h.renderer, { + terminal: { columns: 80, rows: 24 }, + wireKeys: false, + run: "busy", + }); + const bridgeRef: { current: SessionBridge | undefined } = { current: undefined }; + const { port, sends, steers } = lastHopPort(bridgeRef); + const bridge = attachSessionBridge(shell, port); + bridgeRef.current = bridge; + try { + bridge.submit("asap", "steer"); + expect(badgeCount(shell.session)).toBe(1); + bridge.handle({ type: "tool.boundary" }); + expect(steers).toEqual(["asap"]); + expect(sends).toEqual([]); + } finally { + bridge.dispose(); + shell.dispose(); + } + }, + { width: 80, height: 24 }, + ); + }); + + test("two live steers at one tool.boundary keep drain order through Agent.deliver", async () => { + await withTestRenderer( + async (h) => { + const shell = createAppShell(h.renderer, { + terminal: { columns: 80, rows: 24 }, + wireKeys: false, + run: "busy", + }); + const bridgeRef: { current: SessionBridge | undefined } = { current: undefined }; + const sends: string[] = []; + const delivered: string[] = []; + const { enqueue, awaitTail } = createSessionOperationQueue(); + let resolveSlow!: () => void; + const slow = new Promise((resolve) => { + resolveSlow = resolve; + }); + const port = createLiveSessionPort({ + send: (text) => { + sends.push(text); + }, + interrupt: () => {}, + deliver: routeQueuedDelivery({ + send: (text) => { + sends.push(text); + }, + deliverSteer: createLiveSteerDeliver({ + enqueue, + ingest: async (text) => { + if (text.includes("@mention")) await slow; + return { text, attachments: [] }; + }, + deliver: (text) => { + delivered.push(text); + }, + captureGeneration: () => () => true, + onFailure: (err) => { + throw err; + }, + }), + parentCycleLive: () => bridgeRef.current?.parentCycleLive === true, + }), + }); + const bridge = attachSessionBridge(shell, port); + bridgeRef.current = bridge; + try { + bridge.submit("@mention first", "steer"); + bridge.submit("plain second", "steer"); + expect(badgeCount(shell.session)).toBe(2); + bridge.handle({ type: "tool.boundary" }); + await Promise.resolve(); + await Promise.resolve(); + expect(delivered).toEqual([]); + resolveSlow(); + await awaitTail(); + expect(delivered).toEqual(["@mention first", "plain second"]); + expect(sends).toEqual([]); + } finally { + bridge.dispose(); + shell.dispose(); + } + }, + { width: 80, height: 24 }, + ); + }); + + test("inference.done with outstanding tools last-hops to deliverSteer, not send", async () => { + await withTestRenderer( + async (h) => { + const shell = createAppShell(h.renderer, { + terminal: { columns: 80, rows: 24 }, + wireKeys: false, + run: "busy", + }); + const bridgeRef: { current: SessionBridge | undefined } = { current: undefined }; + const { port, sends, steers } = lastHopPort(bridgeRef); + const bridge = attachSessionBridge(shell, port); + bridgeRef.current = bridge; + try { + bridge.submit("asap", "steer"); + expect(badgeCount(shell.session)).toBe(1); + bridge.handle({ + type: "tool.start", + data: { call: { id: "c1", name: "run_shell" } }, + }); + bridge.handle({ type: "inference.done", data: {} }); + expect(steers).toEqual(["asap"]); + expect(sends).toEqual([]); + } finally { + bridge.dispose(); + shell.dispose(); + } + }, + { width: 80, height: 24 }, + ); + }); + + test("text-only settle leftover steer last-hops to send", async () => { + await withTestRenderer( + async (h) => { + const shell = createAppShell(h.renderer, { + terminal: { columns: 80, rows: 24 }, + wireKeys: false, + run: "busy", + }); + const bridgeRef: { current: SessionBridge | undefined } = { current: undefined }; + const { port, sends, steers } = lastHopPort(bridgeRef); + const bridge = attachSessionBridge(shell, port); + bridgeRef.current = bridge; + try { + bridge.submit("leftover", "steer"); + bridge.handle({ type: "inference.start", data: {} }); + bridge.handle({ type: "inference.text.delta", data: { token: "hi" } }); + bridge.handle({ type: "inference.done", data: {} }); + expect(sends).toEqual(["leftover"]); + expect(steers).toEqual([]); + } finally { + bridge.dispose(); + shell.dispose(); + } + }, + { width: 80, height: 24 }, + ); + }); + + test("interrupt leftover steer last-hops to send", async () => { + await withTestRenderer( + async (h) => { + const shell = createAppShell(h.renderer, { + terminal: { columns: 80, rows: 24 }, + wireKeys: false, + run: "busy", + }); + const bridgeRef: { current: SessionBridge | undefined } = { current: undefined }; + const { port, sends, steers } = lastHopPort(bridgeRef); + const bridge = attachSessionBridge(shell, port); + bridgeRef.current = bridge; + try { + bridge.submit("after stop", "steer"); + bridge.interrupt(); + expect(sends).toEqual(["after stop"]); + expect(steers).toEqual([]); + } finally { + bridge.dispose(); + shell.dispose(); + } + }, + { width: 80, height: 24 }, + ); + }); + + test("idle-with-fleet leftover steer last-hops to send", async () => { + await withTestRenderer( + async (h) => { + const shell = createAppShell(h.renderer, { + terminal: { columns: 80, rows: 24 }, + wireKeys: false, + run: "idle", + }); + const bridgeRef: { current: SessionBridge | undefined } = { current: undefined }; + const { port, sends, steers } = lastHopPort(bridgeRef); + const bridge = attachSessionBridge(shell, port); + bridgeRef.current = bridge; + try { + bridge.submit("dispatch", "immediate"); + bridge.submit("one more worker", "steer"); + bridge.handle({ type: "fleet", running: 1 }); + bridge.handle({ type: "inference.start", data: {} }); + bridge.handle({ type: "inference.done", data: {} }); + expect(sends).toEqual(["dispatch", "one more worker"]); + expect(steers).toEqual([]); + } finally { + bridge.dispose(); + shell.dispose(); + } + }, + { width: 80, height: 24 }, + ); + }); + + test("/clear drops queued steers so a later boundary does not deliver or send", async () => { + await withTestRenderer( + async (h) => { + const shell = createAppShell(h.renderer, { + terminal: { columns: 80, rows: 24 }, + wireKeys: false, + run: "busy", + }); + const bridgeRef: { current: SessionBridge | undefined } = { current: undefined }; + const { port, sends, steers } = lastHopPort(bridgeRef); + const bridge = attachSessionBridge(shell, port); + bridgeRef.current = bridge; + try { + bridge.submit("old steer", "steer"); + expect(badgeCount(shell.session)).toBe(1); + bridge.clearQueuedDelivery(); + bridge.handle({ type: "tool.boundary" }); + expect(sends).toEqual([]); + expect(steers).toEqual([]); + } finally { + bridge.dispose(); + shell.dispose(); + } + }, + { width: 80, height: 24 }, + ); + }); +}); diff --git a/src/tui/queued-delivery.test.ts b/src/tui/queued-delivery.test.ts new file mode 100644 index 00000000..8e4a63ec --- /dev/null +++ b/src/tui/queued-delivery.test.ts @@ -0,0 +1,278 @@ +import { describe, expect, test } from "bun:test"; +import type { PendingImageAttachment } from "./image-attachments.js"; +import { + createDeliveryGeneration, + createLeftoverSend, + createLiveSteerDeliver, + routeQueuedDelivery, +} from "./queued-delivery.js"; +import { createSessionOperationQueue } from "./session-operation-queue.js"; + +const image: PendingImageAttachment = { + id: "img-1", + name: "clipboard.png", + contentType: "image/png", + data: new Uint8Array([1]), + contentHash: "hash-1", +}; + +function recordHops(parentCycleLive: () => boolean) { + const sends: string[] = []; + const steers: string[] = []; + const deliver = routeQueuedDelivery({ + send: (text) => { + sends.push(text); + }, + deliverSteer: (text) => { + steers.push(text); + }, + parentCycleLive, + }); + return { deliver, sends, steers }; +} + +describe("routeQueuedDelivery", () => { + test("live parent-boundary steer calls deliverSteer only", () => { + const { deliver, sends, steers } = recordHops(() => true); + deliver("asap", "steer"); + expect(steers).toEqual(["asap"]); + expect(sends).toEqual([]); + }); + + test("leftover steer (idle / fleet-hold / post-interrupt) calls send only", () => { + const { deliver, sends, steers } = recordHops(() => false); + deliver("leftover", "steer"); + expect(sends).toEqual(["leftover"]); + expect(steers).toEqual([]); + }); + + test("queue calls send only, even while the parent cycle is live", () => { + const { deliver, sends, steers } = recordHops(() => true); + deliver("later", "queue"); + expect(sends).toEqual(["later"]); + expect(steers).toEqual([]); + }); + + test("forwards attachments on both hops", () => { + const sent: (readonly PendingImageAttachment[] | undefined)[] = []; + const steered: (readonly PendingImageAttachment[] | undefined)[] = []; + const live = routeQueuedDelivery({ + send: (_text, attachments) => { + sent.push(attachments); + }, + deliverSteer: (_text, attachments) => { + steered.push(attachments); + }, + parentCycleLive: () => true, + }); + const leftover = routeQueuedDelivery({ + send: (_text, attachments) => { + sent.push(attachments); + }, + deliverSteer: (_text, attachments) => { + steered.push(attachments); + }, + parentCycleLive: () => false, + }); + live("asap", "steer", [image]); + leftover("later", "steer", [image]); + leftover("follow", "queue", [image]); + expect(steered).toEqual([[image]]); + expect(sent).toEqual([[image], [image]]); + }); +}); + +describe("createDeliveryGeneration", () => { + test("capture then bump → predicate false", () => { + const generation = createDeliveryGeneration(); + const stillCurrent = generation.capture(); + generation.bump(); + expect(stillCurrent()).toBe(false); + }); + + test("capture without bump → true", () => { + const generation = createDeliveryGeneration(); + const stillCurrent = generation.capture(); + expect(stillCurrent()).toBe(true); + }); + + test("bump then new capture → true", () => { + const generation = createDeliveryGeneration(); + generation.bump(); + const stillCurrent = generation.capture(); + expect(stillCurrent()).toBe(true); + }); + + test("two captures, bump, both stale", () => { + const generation = createDeliveryGeneration(); + const first = generation.capture(); + const second = generation.capture(); + generation.bump(); + expect(first()).toBe(false); + expect(second()).toBe(false); + }); +}); + +describe("createLiveSteerDeliver", () => { + test("slow first ingest does not let a later steer deliver first", async () => { + const delivered: string[] = []; + const { enqueue, awaitTail } = createSessionOperationQueue(); + let resolveSlow!: () => void; + const slow = new Promise((resolve) => { + resolveSlow = resolve; + }); + const deliverSteer = createLiveSteerDeliver({ + enqueue, + ingest: async (text) => { + if (text.startsWith("@mention")) await slow; + return { text, attachments: [] }; + }, + deliver: (text) => { + delivered.push(text); + }, + captureGeneration: () => () => true, + onFailure: (err) => { + throw err; + }, + }); + + deliverSteer("@mention A"); + deliverSteer("plain B"); + await Promise.resolve(); + await Promise.resolve(); + expect(delivered).toEqual([]); + + resolveSlow(); + await awaitTail(); + expect(delivered).toEqual(["@mention A", "plain B"]); + }); + + test("generation bump during ingest drops both in-flight live steers", async () => { + const delivered: string[] = []; + const { enqueue, awaitTail } = createSessionOperationQueue(); + const generation = createDeliveryGeneration(); + let resolveSlow!: () => void; + const slow = new Promise((resolve) => { + resolveSlow = resolve; + }); + const deliverSteer = createLiveSteerDeliver({ + enqueue, + ingest: async (text) => { + if (text === "A") await slow; + return { text, attachments: [] }; + }, + deliver: (text) => { + delivered.push(text); + }, + captureGeneration: generation.capture, + onFailure: (err) => { + throw err; + }, + }); + + deliverSteer("A"); + deliverSteer("B"); + generation.bump(); + resolveSlow(); + await awaitTail(); + expect(delivered).toEqual([]); + }); +}); + +describe("createLeftoverSend", () => { + test("generation bump during ingest drops leftover send and sent-message record", async () => { + const sent: string[] = []; + const recorded: string[] = []; + const { enqueue, awaitTail } = createSessionOperationQueue(); + const generation = createDeliveryGeneration(); + let resolveSlow!: () => void; + const slow = new Promise((resolve) => { + resolveSlow = resolve; + }); + const leftoverSend = createLeftoverSend({ + enqueue, + ingest: async (text) => { + if (text === "leftover") await slow; + return { text, attachments: [] }; + }, + send: (text) => { + sent.push(text); + }, + recordSent: (text) => { + recorded.push(text); + }, + captureGeneration: generation.capture, + onFailure: (err) => { + throw err; + }, + }); + + leftoverSend("leftover"); + generation.bump(); + resolveSlow(); + await awaitTail(); + expect(sent).toEqual([]); + expect(recorded).toEqual([]); + }); + + test("leftover send without a bump still sends and records", async () => { + const sent: string[] = []; + const recorded: string[] = []; + const { enqueue, awaitTail } = createSessionOperationQueue(); + const leftoverSend = createLeftoverSend({ + enqueue, + ingest: async (text) => ({ text, attachments: [] }), + send: (text) => { + sent.push(text); + }, + recordSent: (text) => { + recorded.push(text); + }, + captureGeneration: () => () => true, + onFailure: (err) => { + throw err; + }, + }); + + leftoverSend("follow-up"); + await awaitTail(); + expect(sent).toEqual(["follow-up"]); + expect(recorded).toEqual(["follow-up"]); + }); + + test("generation bump drops leftover send but not a sibling Enter send", async () => { + const leftoverSent: string[] = []; + const enterSent: string[] = []; + const { enqueue, awaitTail } = createSessionOperationQueue(); + const generation = createDeliveryGeneration(); + let resolveSlow!: () => void; + const slow = new Promise((resolve) => { + resolveSlow = resolve; + }); + const leftoverSend = createLeftoverSend({ + enqueue, + ingest: async (text) => { + await slow; + return { text, attachments: [] }; + }, + send: (text) => { + leftoverSent.push(text); + }, + captureGeneration: generation.capture, + onFailure: (err) => { + throw err; + }, + }); + const enterSend = (text: string) => { + enterSent.push(text); + }; + + leftoverSend("queued"); + generation.bump(); + enterSend("hello"); + resolveSlow(); + await awaitTail(); + expect(leftoverSent).toEqual([]); + expect(enterSent).toEqual(["hello"]); + }); +}); diff --git a/src/tui/queued-delivery.ts b/src/tui/queued-delivery.ts new file mode 100644 index 00000000..b307cb46 --- /dev/null +++ b/src/tui/queued-delivery.ts @@ -0,0 +1,129 @@ +/** + * Kind routing for drained queue items, plus a generation token so a + * /clear|/new rotation can drop in-flight delivers that belonged to the + * previous session. Kind routing lives here, not on SessionPort. + * + * Live inject (`deliverSteer` → Agent.deliver) is only for an in-flight + * parent tool.boundary. Leftover steers at idle, idle-with-fleet, or + * post-interrupt share the send path (sendQueue, inFlight, token refresh). + */ + +import type { PendingImageAttachment } from "./image-attachments.js"; +import type { ProductHostDeliver } from "./product-host.js"; + +export interface RouteQueuedDeliveryArgs { + send: (text: string, attachments?: readonly PendingImageAttachment[]) => void; + deliverSteer: (text: string, attachments?: readonly PendingImageAttachment[]) => void; + /** + * True only while the bridge is draining steers at a live parent + * tool.boundary (or inference.done with tools still outstanding). Read + * when the deliver op runs, not captured at mount. + */ + parentCycleLive: () => boolean; +} + +export function routeQueuedDelivery(args: RouteQueuedDeliveryArgs): ProductHostDeliver { + return (text, kind, attachments) => { + if (kind === "steer" && args.parentCycleLive()) { + args.deliverSteer(text, attachments); + return; + } + args.send(text, attachments); + }; +} + +export function createDeliveryGeneration() { + let generation = 0; + return { + bump(): void { + generation += 1; + }, + capture(): () => boolean { + const captured = generation; + return () => captured === generation; + }, + }; +} + +export interface IngestedSteer { + readonly text: string; + readonly attachments: readonly PendingImageAttachment[]; +} + +export interface CreateLiveSteerDeliverArgs { + /** + * FIFO session queue. Ingest must run on this queue — not in a + * fire-and-forget IIFE — so two steers at one boundary cannot reverse + * if the second ingest finishes first. + */ + enqueue: (op: () => Promise) => Promise; + ingest: (text: string, attachments: readonly PendingImageAttachment[]) => Promise; + /** Agent.deliver (or the sessionOps enqueue that wraps it). */ + deliver: (text: string, attachments: readonly PendingImageAttachment[]) => void; + captureGeneration: () => () => boolean; + onFailure: (err: unknown) => void; +} + +export interface CreateLeftoverSendArgs { + enqueue: (op: () => Promise) => Promise; + ingest: (text: string, attachments: readonly PendingImageAttachment[]) => Promise; + /** + * Post-ingest hop (agentProxy.send). Must not ingest again — leftover + * ingest already ran in this wrapper. + */ + send: (text: string, attachments: readonly PendingImageAttachment[]) => void; + /** + * Up/Down recall. Called with the original text only when the hop is + * still current after ingest, so a /clear|/new drop is not recorded. + */ + recordSent?: (text: string) => void; + captureGeneration: () => () => boolean; + onFailure: (err: unknown) => void; +} + +interface GenerationGatedHopArgs { + enqueue: (op: () => Promise) => Promise; + ingest: (text: string, attachments: readonly PendingImageAttachment[]) => Promise; + hop: (text: string, attachments: readonly PendingImageAttachment[]) => void; + recordSent?: (text: string) => void; + captureGeneration: () => () => boolean; + onFailure: (err: unknown) => void; +} + +function createGenerationGatedHop( + args: GenerationGatedHopArgs, +): (text: string, attachments?: readonly PendingImageAttachment[]) => void { + return (text, attachments) => { + const stillCurrent = args.captureGeneration(); + const pending = attachments ?? []; + void args + .enqueue(async () => { + if (!stillCurrent()) return; + const ingested = await args.ingest(text, pending); + if (!stillCurrent()) return; + args.recordSent?.(text); + args.hop(ingested.text, ingested.attachments); + }) + .catch(args.onFailure); + }; +} + +/** + * Live inject: enqueue ingest, then deliver, in drain order. Previously + * each item started ingest immediately, so Agent.deliver could reverse. + */ +export function createLiveSteerDeliver( + args: CreateLiveSteerDeliverArgs, +): (text: string, attachments?: readonly PendingImageAttachment[]) => void { + return createGenerationGatedHop({ ...args, hop: args.deliver }); +} + +/** + * Leftover / queue drain hop: capture generation at hop time, ingest, then + * send only if /clear|/new has not bumped. Operator Enter must not use this. + */ +export function createLeftoverSend( + args: CreateLeftoverSendArgs, +): (text: string, attachments?: readonly PendingImageAttachment[]) => void { + return createGenerationGatedHop({ ...args, hop: args.send }); +} diff --git a/src/tui/runner-host.test.ts b/src/tui/runner-host.test.ts index 51562fa6..ac499fe2 100644 --- a/src/tui/runner-host.test.ts +++ b/src/tui/runner-host.test.ts @@ -150,6 +150,7 @@ describe("mountRunnerHost chrome wiring", () => { eventEmitter: new EventEmitter(), send: () => {}, interrupt: () => {}, + deliver: () => {}, providers: {}, onModelSelect: () => {}, commands: [], @@ -194,6 +195,7 @@ describe("mountRunnerHost command surfaces", () => { eventEmitter: new EventEmitter(), send: () => {}, interrupt: () => {}, + deliver: () => {}, providers: {}, onModelSelect: () => {}, commands: [], @@ -239,6 +241,7 @@ describe("mountRunnerHost model picker", () => { eventEmitter: new EventEmitter(), send: () => {}, interrupt: () => {}, + deliver: () => {}, providers: { xai: { models: ["grok-4", "grok-3"] } }, activeModel: () => ({ provider: "xai", model: "grok-4" }), onModelSelect: () => {}, @@ -270,6 +273,7 @@ describe("mountRunnerHost model picker", () => { eventEmitter: new EventEmitter(), send: () => {}, interrupt: () => {}, + deliver: () => {}, providers: { xai: { models: ["grok-4"] } }, onModelSelect: () => {}, commands: [], @@ -300,6 +304,7 @@ describe("mountRunnerHost model picker", () => { eventEmitter: new EventEmitter(), send: () => {}, interrupt: () => {}, + deliver: () => {}, providers: { xai: { models: ["grok-4"] } }, onModelSelect: () => {}, onFavoriteToggle: (id) => toggled.push(id), @@ -331,6 +336,7 @@ describe("mountRunnerHost model picker", () => { eventEmitter: new EventEmitter(), send: () => {}, interrupt: () => {}, + deliver: () => {}, providers: { xai: { models: ["grok-4"] } }, onModelSelect: () => {}, onSetDefault: (id) => setDefault.push(id), @@ -360,6 +366,7 @@ describe("mountRunnerHost model picker", () => { eventEmitter: new EventEmitter(), send: () => {}, interrupt: () => {}, + deliver: () => {}, providers: { xai: { models: ["grok-4"] } }, onModelSelect: () => {}, onConnectProvider: (name) => connected.push(name), @@ -397,6 +404,7 @@ describe("bottom border cost run", () => { eventEmitter: new EventEmitter(), send: () => {}, interrupt: () => {}, + deliver: () => {}, providers: {}, onModelSelect: () => {}, commands: [], @@ -425,6 +433,7 @@ describe("bottom border cost run", () => { eventEmitter: new EventEmitter(), send: () => {}, interrupt: () => {}, + deliver: () => {}, providers: {}, onModelSelect: () => {}, commands: [], @@ -457,6 +466,7 @@ describe("bottom border cost run", () => { eventEmitter: new EventEmitter(), send: () => {}, interrupt: () => {}, + deliver: () => {}, providers: { xai: { models: ["grok-4"] }, "codex/abk-labs": { models: ["gpt-5.5"] }, @@ -506,6 +516,7 @@ describe("bottom border cost run", () => { eventEmitter: new EventEmitter(), send: () => {}, interrupt: () => {}, + deliver: () => {}, providers: { "codex/abk-labs": { models: ["gpt-5.5"] }, xai: { models: ["grok-4"] }, @@ -555,6 +566,7 @@ describe("bottom border cost run", () => { eventEmitter: emitter, send: () => {}, interrupt: () => {}, + deliver: () => {}, providers: {}, onModelSelect: () => {}, commands: [], @@ -590,6 +602,7 @@ describe("bottom border cost run", () => { eventEmitter: emitter, send: () => {}, interrupt: () => {}, + deliver: () => {}, providers: {}, onModelSelect: () => {}, commands: [], @@ -623,6 +636,7 @@ describe("bottom border cost run", () => { eventEmitter: emitter, send: () => {}, interrupt: () => {}, + deliver: () => {}, providers: {}, onModelSelect: () => {}, commands: [], @@ -662,6 +676,7 @@ describe("mountRunnerHost quit key", () => { eventEmitter: new EventEmitter(), send: () => {}, interrupt: () => {}, + deliver: () => {}, providers: {}, onModelSelect: () => {}, commands: [], diff --git a/src/tui/runner-host.ts b/src/tui/runner-host.ts index f97cc620..55740a13 100644 --- a/src/tui/runner-host.ts +++ b/src/tui/runner-host.ts @@ -65,7 +65,7 @@ export interface RunnerHostDeps { attachments?: readonly PendingImageAttachment[], ) => "agent" | "local" | "empty"; readonly interrupt: () => void; - readonly deliver?: ( + readonly deliver: ( text: string, kind: QueueKind, attachments?: readonly PendingImageAttachment[], @@ -259,8 +259,8 @@ export async function mountRunnerHost(deps: RunnerHostDeps): Promise eventEmitter: deps.eventEmitter, send: deps.send, interrupt: deps.interrupt, + deliver: deps.deliver, ...(deps.classifySubmit !== undefined ? { classifySubmit: deps.classifySubmit } : {}), - ...(deps.deliver !== undefined ? { deliver: deps.deliver } : {}), ...(deps.onConnectProvider !== undefined ? { onConnectProvider: deps.onConnectProvider } : {}), ...(deps.onFavoriteToggle !== undefined ? { onFavoriteToggle: deps.onFavoriteToggle } : {}), ...(deps.onSetDefault !== undefined ? { onSetDefault: deps.onSetDefault } : {}), diff --git a/src/tui/runner.ts b/src/tui/runner.ts index 82ac5ae5..c76a079d 100644 --- a/src/tui/runner.ts +++ b/src/tui/runner.ts @@ -185,6 +185,12 @@ import { setActiveWebProviderBrand } from "./tool-formatter.js"; import { consumeStream } from "../session/stream-consumer.js"; import { createCycleTextRecorder } from "../session/stream-journal.js"; import { mountRunnerHost } from "./runner-host.js"; +import { + createDeliveryGeneration, + createLeftoverSend, + createLiveSteerDeliver, + routeQueuedDelivery, +} from "./queued-delivery.js"; import { createRuntimeShutdown } from "./runtime-shutdown.js"; import { applyFocus, @@ -206,9 +212,8 @@ import { classifyAgentSendFailure, shouldSettleUiAfterSendFailure, } from "./session-chrome.js"; -import { ingestPathMentions } from "./prompt-attachments.js"; +import { ingestOperatorPrompt } from "./prompt-attachments.js"; import { listPathSuggestions } from "./components/at-mention/list.js"; -import { resolveAtMentions } from "./mention-resolution.js"; import { imageAttachmentFromPath, type PendingImageAttachment } from "./image-attachments.js"; import { appendSentMessage, loadSentMessages } from "../session/sent-messages.js"; import type { OperatorGateEvent } from "./gate-events.js"; @@ -1414,8 +1419,11 @@ export async function runTUI(initialConfig: Config): Promise { // Reload, interrupt, compaction continuation, and proxy deliver share one queue // so a rebuild never races an in-flight deliver. const sessionOps = createSessionOperationQueue(); + const deliveryGeneration = createDeliveryGeneration(); const enqueueAgentDeliver = (deliverToLiveAgent: () => void): void => { + const stillCurrent = deliveryGeneration.capture(); void sessionOps.enqueue(async () => { + if (!stillCurrent()) return; // The shell already popped the queue item and painted it as delivered // by the time this runs, so a failed rebuild must be surfaced here — // otherwise the message silently never reaches the agent. @@ -1910,6 +1918,7 @@ export async function runTUI(initialConfig: Config): Promise { // abort handles → child agent.close) before clearing the session store so // /clear does not leave orphaned child reactors burning tokens. const newSession = (): void => { + deliveryGeneration.bump(); cancelFeedbackCapture(); // Wipe the painted transcript immediately. The product host listens for // session.clear; the Ink App used to clear its own stream unconditionally @@ -2219,10 +2228,13 @@ export async function runTUI(initialConfig: Config): Promise { }); }); } - const ingested = await ingestPathMentions(text, config.cwd, imageAttachmentFromPath); - const resolved = await resolveAtMentions(ingested.text, config.cwd); - const attachments = [...pending, ...ingested.attachments]; - await agentProxy.send(userInboundMessage(resolved, attachments)); + const ingested = await ingestOperatorPrompt( + text, + config.cwd, + imageAttachmentFromPath, + pending, + ); + await agentProxy.send(userInboundMessage(ingested.text, ingested.attachments)); }; const dispatchCommand = (name: string, args: string): void => { @@ -2249,32 +2261,34 @@ export async function runTUI(initialConfig: Config): Promise { const computeAddProviderChoices = () => addProviderSelectorChoices(providerChoices(), config.providers); + const send = createSubmitHandler({ + dispatchCommand: (name, args) => dispatchCommand(name, args), + sendPrompt: (text, attachments) => { + void sendUserPrompt(text, attachments ?? []).catch(handleSendFailure); + }, + onPromptSubmitted: () => { + if (telemetryFirstRun && liveTelemetryIntent) { + void activateHeldTelemetry(trueGlobalSettingsPath, () => liveTelemetryIntent); + } + }, + isFeedbackCapturePending, + cancelFeedbackCapture, + onFeedbackText: (text) => { + takeFeedbackCapture(); + const status = captureFeedback(getTelemetry(), text, { + turnTraceId: getLastTurnTraceId(), + }); + return feedbackResultMessage(status); + }, + onSystemNotice: systemNotice, + }); + const host = await mountRunnerHost({ // An unnamed session shows nothing rather than a placeholder. title: runTaskTitle, cwd: process.cwd(), eventEmitter: emitter, - send: createSubmitHandler({ - dispatchCommand: (name, args) => dispatchCommand(name, args), - sendPrompt: (text, attachments) => { - void sendUserPrompt(text, attachments ?? []).catch(handleSendFailure); - }, - onPromptSubmitted: () => { - if (telemetryFirstRun && liveTelemetryIntent) { - void activateHeldTelemetry(trueGlobalSettingsPath, () => liveTelemetryIntent); - } - }, - isFeedbackCapturePending, - cancelFeedbackCapture, - onFeedbackText: (text) => { - takeFeedbackCapture(); - const status = captureFeedback(getTelemetry(), text, { - turnTraceId: getLastTurnTraceId(), - }); - return feedbackResultMessage(status); - }, - onSystemNotice: systemNotice, - }), + send, classifySubmit: (text, attachments) => classifySubmission(text, { hasAttachments: attachments !== undefined && attachments.length > 0, @@ -2282,6 +2296,38 @@ export async function runTUI(initialConfig: Config): Promise { feedbackCaptureEnabled: true, }), interrupt, + deliver: routeQueuedDelivery({ + send: createLeftoverSend({ + enqueue: sessionOps.enqueue, + ingest: (text, pending) => + ingestOperatorPrompt(text, config.cwd, imageAttachmentFromPath, pending), + send: (text, pending) => { + sendAborted = false; + void agentProxy.send(userInboundMessage(text, pending)).catch(handleSendFailure); + }, + recordSent: (text) => { + if (text.trim().length === 0) return; + void appendSentMessage(config.cwd, sessionId, text).catch((err: unknown) => { + tuiLogger.debug("sent-message append failed: {error}", { + error: err instanceof Error ? err.message : String(err), + }); + }); + }, + captureGeneration: deliveryGeneration.capture, + onFailure: handleSendFailure, + }), + parentCycleLive: () => host.bridge.parentCycleLive, + deliverSteer: createLiveSteerDeliver({ + enqueue: sessionOps.enqueue, + ingest: (text, pending) => + ingestOperatorPrompt(text, config.cwd, imageAttachmentFromPath, pending), + deliver: (text, pending) => { + agentProxy.deliver(userInboundMessage(text, pending)); + }, + captureGeneration: deliveryGeneration.capture, + onFailure: handleSendFailure, + }), + }), // Consent by proceeding requires the disclosure to be on screen before the // first prompt activates the held telemetry instance: the landing shows it, // and the shell re-files it into the transcript when the landing clears. diff --git a/src/tui/runtime-bridge.ts b/src/tui/runtime-bridge.ts index fe9945f8..dc8e7f97 100644 --- a/src/tui/runtime-bridge.ts +++ b/src/tui/runtime-bridge.ts @@ -7,6 +7,7 @@ */ import { + createSessionQueue, drainOne, enqueue, enqueueSteer, @@ -175,6 +176,19 @@ export interface SessionBridge { attachments?: readonly PendingImageAttachment[], ) => void; interrupt: () => void; + /** + * Drop mid-run queue items, pending echoes, and fleet hold so a session + * rotation cannot drain old input into the new reactor. Leaves the run idle + * so the next Enter is a send, not a steer. + */ + clearQueuedDelivery: () => void; + /** + * True only while draining steers at a live parent tool.boundary (or + * inference.done with tools still outstanding). Last-hop routing reads this + * when the deliver op runs: leftover / fleet-hold / post-interrupt drains + * are false and must send(). + */ + readonly parentCycleLive: boolean; /** * A permission or operator gate was raised — queued or already displayed. * Blocks the turn (and exempts it from the stall watchdog) until a matching @@ -386,6 +400,11 @@ interface BridgeBag { * into fragments that each read as half a sentence. */ turnThinking: TurnThinking | null; + /** + * Set only around drainSteersAtBoundary at a live parent tool.boundary. + * Last-hop routing (routeQueuedDelivery) reads this when deliver runs. + */ + liveSteerInject: boolean; } const bridges = new WeakMap(); @@ -782,6 +801,20 @@ function drainSteersAtBoundary(shell: AppShell, bag: BridgeBag): void { paintChrome(shell); } +/** + * Live parent-cycle inject: routeQueuedDelivery reads parentCycleLive while + * this drain's port.deliver runs. Idle leftover, fleet-hold, and interrupt + * use drainSteersAtBoundary / drainAtBoundary without this flag so they send. + */ +function drainLiveSteersAtBoundary(shell: AppShell, bag: BridgeBag): void { + bag.liveSteerInject = true; + try { + drainSteersAtBoundary(shell, bag); + } finally { + bag.liveSteerInject = false; + } +} + /** * 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 @@ -795,8 +828,8 @@ function settleRunToIdle(shell: AppShell, bag: BridgeBag): void { shell.inFlightTool = null; if (bag.liveFleet > 0) { // Hold: the fleet is still live, so the run stays busy. Steers left - // pending deliver now — the parent they were steering has stopped, so - // each one just starts its own turn — while follow-ups keep waiting. + // pending send now — the parent they were steering has stopped, so + // each one starts its own turn — while follow-ups keep waiting. drainSteersAtBoundary(shell, bag); return; } @@ -862,7 +895,7 @@ function applyInbound(shell: AppShell, bag: BridgeBag, event: BridgeInboundEvent if (event.type === "tool.boundary") { // Soft steer only — follow-ups wait until the run goes idle. - drainSteersAtBoundary(shell, bag); + drainLiveSteersAtBoundary(shell, bag); return; } @@ -920,6 +953,7 @@ export function attachSessionBridge( panelOnlyCallIds: new Set(), attemptRow: null, turnThinking: null, + liveSteerInject: false, }; bridges.set(shell, bag); @@ -1092,7 +1126,7 @@ export function attachSessionBridge( // turn (see turn-state.ts) — the cycle continues, but a soft-steer // boundary still passed. Follow-ups wait for idle. if (onTurnBoundary(event) && bag.turn.activeToolCalls.length > 0) { - drainSteersAtBoundary(shell, bag); + drainLiveSteersAtBoundary(shell, bag); } if (settled) settleRun(); return; @@ -1223,6 +1257,14 @@ export function attachSessionBridge( paintPhase(); }; + const clearQueuedDelivery = (): void => { + if (bag.disposed) return; + shell.session = createSessionQueue("idle"); + bag.pendingEchoes.length = 0; + bag.liveFleet = 0; + paintChrome(shell); + }; + /** * A permission or operator gate was raised — queued or already on screen, * the turn does not distinguish. Called from the gate wiring itself, not @@ -1330,6 +1372,10 @@ export function attachSessionBridge( }, submit, interrupt: doInterrupt, + clearQueuedDelivery, + get parentCycleLive() { + return bag.liveSteerInject; + }, gateOpened, gateClosed, get turn() { diff --git a/src/tui/runtime-channels.test.ts b/src/tui/runtime-channels.test.ts index 518c601d..d3934edc 100644 --- a/src/tui/runtime-channels.test.ts +++ b/src/tui/runtime-channels.test.ts @@ -28,6 +28,7 @@ async function mountHeadless(overrides: Partial = {}): Promis eventEmitter: emitter, send: () => {}, interrupt: () => {}, + deliver: () => {}, createRenderer: async () => harness.renderer, ...overrides, }); diff --git a/src/tui/steer-worker-invariant.test.ts b/src/tui/steer-worker-invariant.test.ts index 7995c8e3..e1630f9b 100644 --- a/src/tui/steer-worker-invariant.test.ts +++ b/src/tui/steer-worker-invariant.test.ts @@ -107,6 +107,95 @@ describe("CL-6291 worker-alive invariants", () => { ); }); + test("busy Enter boundary uses deliver, not sendImmediate", 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 { + bridge.submit("steer into the live turn", "steer"); + expect(badgeCount(shell.session)).toBe(1); + port.clear(); + bridge.handle({ type: "tool.boundary" }); + await h.renderOnce(); + expect(port.calls.some((c) => c.op === "deliver")).toBe(true); + expect(port.calls.some((c) => c.op === "sendImmediate")).toBe(false); + expect(port.calls.some((c) => c.op === "interrupt")).toBe(false); + } finally { + bridge.dispose(); + shell.dispose(); + } + }, + { width: 80, height: 24 }, + ); + }); + + test("clearQueuedDelivery drops pending steers instead of draining them", 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 { + bridge.submit("old steer", "steer"); + expect(badgeCount(shell.session)).toBe(1); + port.clear(); + bridge.clearQueuedDelivery(); + expect(badgeCount(shell.session)).toBe(0); + expect(shell.session.run).toBe("idle"); + bridge.handle({ type: "tool.boundary" }); + await h.renderOnce(); + expect(port.calls.some((c) => c.op === "deliver")).toBe(false); + expect(port.calls.some((c) => c.op === "sendImmediate")).toBe(false); + } finally { + bridge.dispose(); + shell.dispose(); + } + }, + { width: 80, height: 24 }, + ); + }); + + test("clearQueuedDelivery forgets pending echoes so inbound user rows paint", 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 { + bridge.submit("hello there", "immediate"); + const before = shell.streamLog.filter( + (r) => r.role === "user" && r.text === "hello there", + ).length; + expect(before).toBe(1); + bridge.clearQueuedDelivery(); + bridge.handle({ type: "user", text: "hello there" }); + await h.renderOnce(); + expect( + shell.streamLog.filter((r) => r.role === "user" && r.text === "hello there"), + ).toHaveLength(2); + } finally { + bridge.dispose(); + shell.dispose(); + } + }, + { width: 80, height: 24 }, + ); + }); + test("Ctrl+C / doInterrupt still calls port.interrupt", async () => { await withTestRenderer( async (h) => {