From 7a2bebc1903877ba43e6a484dcdcfee654776850 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Mon, 31 Aug 2026 13:07:42 -0700 Subject: [PATCH 1/5] Route mid-run Enter through the live reactor Boundary delivery used to fall back to ordinary send when production omitted the transport, so a queued mid-run Enter started a second turn. Require the transport, steer via Agent.deliver, and drop queued input on session rotation. --- docs/ARCHITECTURE.md | 7 +- docs/TUI.md | 13 ++-- src/tui/keybindings.test.ts | 1 + src/tui/live-session-port.test.ts | 29 +++----- src/tui/live-session-port.ts | 17 +---- src/tui/product-host.test.ts | 31 ++++++++- src/tui/product-host.ts | 5 +- src/tui/queued-delivery.test.ts | 93 ++++++++++++++++++++++++++ src/tui/queued-delivery.ts | 36 ++++++++++ src/tui/runner-host.test.ts | 13 ++++ src/tui/runner-host.ts | 4 +- src/tui/runner.ts | 55 +++++++++------ src/tui/runtime-bridge.ts | 16 +++++ src/tui/runtime-channels.test.ts | 1 + src/tui/steer-worker-invariant.test.ts | 89 ++++++++++++++++++++++++ 15 files changed, 341 insertions(+), 69 deletions(-) create mode 100644 src/tui/queued-delivery.test.ts create mode 100644 src/tui/queued-delivery.ts diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 6e73c41bd..0cb954a51 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -81,7 +81,12 @@ 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` + sends steers via `agentProxy.deliver` (`Agent.deliver`) into the live reactor + and idle follow-ups via 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 55af3bbd2..93a56f070 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`). diff --git a/src/tui/keybindings.test.ts b/src/tui/keybindings.test.ts index b99cbd06b..55491455e 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 71597de90..25583d23a 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 8d44368f8..f0abad496 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 at tool boundary (or idle). Always forwarded. */ + 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 e9e1c4388..30ac240de 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 915411e7b..eb76cf910 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 { + test("steer calls deliverSteer only", () => { + const sends: string[] = []; + const steers: string[] = []; + const deliver = routeQueuedDelivery({ + send: (text) => { + sends.push(text); + }, + deliverSteer: (text) => { + steers.push(text); + }, + }); + deliver("asap", "steer"); + expect(steers).toEqual(["asap"]); + expect(sends).toEqual([]); + }); + + test("queue calls send only", () => { + const sends: string[] = []; + const steers: string[] = []; + const deliver = routeQueuedDelivery({ + send: (text) => { + sends.push(text); + }, + deliverSteer: (text) => { + steers.push(text); + }, + }); + deliver("later", "queue"); + expect(sends).toEqual(["later"]); + expect(steers).toEqual([]); + }); + + test("forwards attachments on both kinds", () => { + const sent: (readonly PendingImageAttachment[] | undefined)[] = []; + const steered: (readonly PendingImageAttachment[] | undefined)[] = []; + const deliver = routeQueuedDelivery({ + send: (_text, attachments) => { + sent.push(attachments); + }, + deliverSteer: (_text, attachments) => { + steered.push(attachments); + }, + }); + deliver("asap", "steer", [image]); + deliver("later", "queue", [image]); + expect(steered).toEqual([[image]]); + expect(sent).toEqual([[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); + }); +}); diff --git a/src/tui/queued-delivery.ts b/src/tui/queued-delivery.ts new file mode 100644 index 000000000..227367956 --- /dev/null +++ b/src/tui/queued-delivery.ts @@ -0,0 +1,36 @@ +/** + * 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. + */ + +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; +} + +export function routeQueuedDelivery(args: RouteQueuedDeliveryArgs): ProductHostDeliver { + return (text, kind, attachments) => { + if (kind === "steer") { + 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; + }, + }; +} diff --git a/src/tui/runner-host.test.ts b/src/tui/runner-host.test.ts index 51562fa67..d1b16583e 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: [], @@ -555,6 +564,7 @@ describe("bottom border cost run", () => { eventEmitter: emitter, send: () => {}, interrupt: () => {}, + deliver: () => {}, providers: {}, onModelSelect: () => {}, commands: [], @@ -590,6 +600,7 @@ describe("bottom border cost run", () => { eventEmitter: emitter, send: () => {}, interrupt: () => {}, + deliver: () => {}, providers: {}, onModelSelect: () => {}, commands: [], @@ -623,6 +634,7 @@ describe("bottom border cost run", () => { eventEmitter: emitter, send: () => {}, interrupt: () => {}, + deliver: () => {}, providers: {}, onModelSelect: () => {}, commands: [], @@ -662,6 +674,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 f97cc6205..55740a13a 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 82ac5ae5c..0bebc88e7 100644 --- a/src/tui/runner.ts +++ b/src/tui/runner.ts @@ -185,6 +185,7 @@ 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, routeQueuedDelivery } from "./queued-delivery.js"; import { createRuntimeShutdown } from "./runtime-shutdown.js"; import { applyFocus, @@ -1414,8 +1415,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 +1914,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 @@ -2249,32 +2254,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 +2289,12 @@ export async function runTUI(initialConfig: Config): Promise { feedbackCaptureEnabled: true, }), interrupt, + deliver: routeQueuedDelivery({ + send, + deliverSteer: (text, attachments) => { + agentProxy.deliver(userInboundMessage(text, attachments ?? [])); + }, + }), // 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 fe9945f89..31f62d105 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,12 @@ 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; /** * A permission or operator gate was raised — queued or already displayed. * Blocks the turn (and exempts it from the stall watchdog) until a matching @@ -1223,6 +1230,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 +1345,7 @@ export function attachSessionBridge( }, submit, interrupt: doInterrupt, + clearQueuedDelivery, gateOpened, gateClosed, get turn() { diff --git a/src/tui/runtime-channels.test.ts b/src/tui/runtime-channels.test.ts index 518c601db..d3934edc5 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 7995c8e35..e1630f9b0 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) => { From ff577d9ddfa3002bffa025eea6e0ce11accefd49 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Mon, 31 Aug 2026 19:34:06 -0700 Subject: [PATCH 2/5] Route leftover steers through send not live deliver Live parent tool.boundary still injects via Agent.deliver. Steers left at idle, idle-with-fleet, or after interrupt must send so they share sendQueue, inFlight, and token refresh with follow-ups. --- docs/ARCHITECTURE.md | 6 +- docs/TUI.md | 2 +- src/tui/live-session-port.ts | 2 +- src/tui/prompt-attachments.test.ts | 30 ++++- src/tui/prompt-attachments.ts | 16 +++ src/tui/queued-delivery-hop.test.ts | 172 ++++++++++++++++++++++++++++ src/tui/queued-delivery.test.ts | 69 ++++++----- src/tui/queued-delivery.ts | 12 +- src/tui/runner.ts | 28 +++-- src/tui/runtime-bridge.ts | 38 +++++- 10 files changed, 331 insertions(+), 44 deletions(-) create mode 100644 src/tui/queued-delivery-hop.test.ts diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 0cb954a51..23f1834fc 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -83,8 +83,10 @@ In TUI chat mode there is no completion gate — the session stays open across t - Bridges reactor events to the OpenTUI host via a plain `EventEmitter` - **Mid-run injection** — Shell `session-queue` items drain at the parent `tool.boundary` through `SessionPort.deliver`. Production `routeQueuedDelivery` - sends steers via `agentProxy.deliver` (`Agent.deliver`) into the live reactor - and idle follow-ups via the existing send path. `/clear` and `/new` bump a + 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. diff --git a/docs/TUI.md b/docs/TUI.md index 93a56f070..a8e954ab0 100644 --- a/docs/TUI.md +++ b/docs/TUI.md @@ -481,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/live-session-port.ts b/src/tui/live-session-port.ts index f0abad496..16459c370 100644 --- a/src/tui/live-session-port.ts +++ b/src/tui/live-session-port.ts @@ -23,7 +23,7 @@ export interface LiveSessionPortDeps { ) => SubmitClassification; /** Hard interrupt current run (runner close/rebuild). */ interrupt: () => void; - /** Drained queue/steer item at tool boundary (or idle). Always forwarded. */ + /** Drained queue/steer item. Kind routing (live inject vs send) is the host's. */ deliver: (text: string, kind: QueueKind, attachments?: readonly PendingImageAttachment[]) => void; } diff --git a/src/tui/prompt-attachments.test.ts b/src/tui/prompt-attachments.test.ts index 3e2323389..289afa159 100644 --- a/src/tui/prompt-attachments.test.ts +++ b/src/tui/prompt-attachments.test.ts @@ -1,7 +1,11 @@ import { describe, expect, test } from "bun:test"; import type { AttachImageResult, PendingImageAttachment } from "./image-attachments.js"; -import { ingestPathMentions, spliceMentionCompletion } from "./prompt-attachments.js"; +import { + ingestOperatorPrompt, + ingestPathMentions, + spliceMentionCompletion, +} from "./prompt-attachments.js"; function attachment(name: string): PendingImageAttachment { return { @@ -40,6 +44,30 @@ describe("ingestPathMentions", () => { }); }); +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 3d6b93828..7491ca7c5 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 000000000..f76d7632a --- /dev/null +++ b/src/tui/queued-delivery-hop.test.ts @@ -0,0 +1,172 @@ +/** + * 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 { routeQueuedDelivery } from "./queued-delivery.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("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 index 75f5d30aa..a627765b3 100644 --- a/src/tui/queued-delivery.test.ts +++ b/src/tui/queued-delivery.test.ts @@ -10,54 +10,69 @@ const image: PendingImageAttachment = { 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("steer calls deliverSteer only", () => { - const sends: string[] = []; - const steers: string[] = []; - const deliver = routeQueuedDelivery({ - send: (text) => { - sends.push(text); - }, - deliverSteer: (text) => { - steers.push(text); - }, - }); + 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("queue calls send only", () => { - const sends: string[] = []; - const steers: string[] = []; - const deliver = routeQueuedDelivery({ - send: (text) => { - sends.push(text); - }, - deliverSteer: (text) => { - steers.push(text); - }, - }); + 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 kinds", () => { + test("forwards attachments on both hops", () => { const sent: (readonly PendingImageAttachment[] | undefined)[] = []; const steered: (readonly PendingImageAttachment[] | undefined)[] = []; - const deliver = routeQueuedDelivery({ + 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, }); - deliver("asap", "steer", [image]); - deliver("later", "queue", [image]); + live("asap", "steer", [image]); + leftover("later", "steer", [image]); + leftover("follow", "queue", [image]); expect(steered).toEqual([[image]]); - expect(sent).toEqual([[image]]); + expect(sent).toEqual([[image], [image]]); }); }); diff --git a/src/tui/queued-delivery.ts b/src/tui/queued-delivery.ts index 227367956..cde3f161c 100644 --- a/src/tui/queued-delivery.ts +++ b/src/tui/queued-delivery.ts @@ -2,6 +2,10 @@ * 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"; @@ -10,11 +14,17 @@ 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") { + if (kind === "steer" && args.parentCycleLive()) { args.deliverSteer(text, attachments); return; } diff --git a/src/tui/runner.ts b/src/tui/runner.ts index 0bebc88e7..826aa1c2c 100644 --- a/src/tui/runner.ts +++ b/src/tui/runner.ts @@ -207,9 +207,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"; @@ -2224,10 +2223,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 => { @@ -2291,8 +2293,20 @@ export async function runTUI(initialConfig: Config): Promise { interrupt, deliver: routeQueuedDelivery({ send, + parentCycleLive: () => host.bridge.parentCycleLive, deliverSteer: (text, attachments) => { - agentProxy.deliver(userInboundMessage(text, attachments ?? [])); + const stillCurrent = deliveryGeneration.capture(); + void (async () => { + if (!stillCurrent()) return; + const ingested = await ingestOperatorPrompt( + text, + config.cwd, + imageAttachmentFromPath, + attachments ?? [], + ); + if (!stillCurrent()) return; + agentProxy.deliver(userInboundMessage(ingested.text, ingested.attachments)); + })().catch(handleSendFailure); }, }), // Consent by proceeding requires the disclosure to be on screen before the diff --git a/src/tui/runtime-bridge.ts b/src/tui/runtime-bridge.ts index 31f62d105..dc8e7f979 100644 --- a/src/tui/runtime-bridge.ts +++ b/src/tui/runtime-bridge.ts @@ -182,6 +182,13 @@ export interface SessionBridge { * 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 @@ -393,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(); @@ -789,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 @@ -802,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; } @@ -869,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; } @@ -927,6 +953,7 @@ export function attachSessionBridge( panelOnlyCallIds: new Set(), attemptRow: null, turnThinking: null, + liveSteerInject: false, }; bridges.set(shell, bag); @@ -1099,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; @@ -1346,6 +1373,9 @@ export function attachSessionBridge( submit, interrupt: doInterrupt, clearQueuedDelivery, + get parentCycleLive() { + return bag.liveSteerInject; + }, gateOpened, gateClosed, get turn() { From c821bc361faec8eccc67e775a0e10180226e17f0 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Mon, 31 Aug 2026 19:57:20 -0700 Subject: [PATCH 3/5] Preserve live steer drain order through Agent.deliver Ingest used to start before enqueue, so two steers at one boundary could reverse if the second mention resolved first. --- src/tui/queued-delivery-hop.test.ts | 99 ++++++++++++++++++++++++++++- src/tui/queued-delivery.test.ts | 73 ++++++++++++++++++++- src/tui/queued-delivery.ts | 40 ++++++++++++ src/tui/runner.ts | 30 ++++----- 4 files changed, 225 insertions(+), 17 deletions(-) diff --git a/src/tui/queued-delivery-hop.test.ts b/src/tui/queued-delivery-hop.test.ts index f76d7632a..bd59dd67d 100644 --- a/src/tui/queued-delivery-hop.test.ts +++ b/src/tui/queued-delivery-hop.test.ts @@ -7,7 +7,8 @@ import { attachSessionBridge, type SessionBridge } from "./runtime-bridge"; import { createLiveSessionPort } from "./live-session-port"; import { createAppShell } from "./shell"; import { withTestRenderer } from "./harness"; -import { routeQueuedDelivery } from "./queued-delivery.js"; +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 }) { @@ -59,6 +60,102 @@ describe("queued delivery last hop", () => { ); }); + 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) => { diff --git a/src/tui/queued-delivery.test.ts b/src/tui/queued-delivery.test.ts index a627765b3..9d32d10fe 100644 --- a/src/tui/queued-delivery.test.ts +++ b/src/tui/queued-delivery.test.ts @@ -1,6 +1,11 @@ import { describe, expect, test } from "bun:test"; import type { PendingImageAttachment } from "./image-attachments.js"; -import { createDeliveryGeneration, routeQueuedDelivery } from "./queued-delivery.js"; +import { + createDeliveryGeneration, + createLiveSteerDeliver, + routeQueuedDelivery, +} from "./queued-delivery.js"; +import { createSessionOperationQueue } from "./session-operation-queue.js"; const image: PendingImageAttachment = { id: "img-1", @@ -106,3 +111,69 @@ describe("createDeliveryGeneration", () => { 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([]); + }); +}); diff --git a/src/tui/queued-delivery.ts b/src/tui/queued-delivery.ts index cde3f161c..849231fda 100644 --- a/src/tui/queued-delivery.ts +++ b/src/tui/queued-delivery.ts @@ -44,3 +44,43 @@ export function createDeliveryGeneration() { }, }; } + +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; +} + +/** + * 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 (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.deliver(ingested.text, ingested.attachments); + }) + .catch(args.onFailure); + }; +} diff --git a/src/tui/runner.ts b/src/tui/runner.ts index 826aa1c2c..ae9c93dd2 100644 --- a/src/tui/runner.ts +++ b/src/tui/runner.ts @@ -185,7 +185,11 @@ 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, routeQueuedDelivery } from "./queued-delivery.js"; +import { + createDeliveryGeneration, + createLiveSteerDeliver, + routeQueuedDelivery, +} from "./queued-delivery.js"; import { createRuntimeShutdown } from "./runtime-shutdown.js"; import { applyFocus, @@ -2294,20 +2298,16 @@ export async function runTUI(initialConfig: Config): Promise { deliver: routeQueuedDelivery({ send, parentCycleLive: () => host.bridge.parentCycleLive, - deliverSteer: (text, attachments) => { - const stillCurrent = deliveryGeneration.capture(); - void (async () => { - if (!stillCurrent()) return; - const ingested = await ingestOperatorPrompt( - text, - config.cwd, - imageAttachmentFromPath, - attachments ?? [], - ); - if (!stillCurrent()) return; - agentProxy.deliver(userInboundMessage(ingested.text, ingested.attachments)); - })().catch(handleSendFailure); - }, + 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, From 9d9dacc46de8982c4f5410ffbd0fd4d9b4fd1500 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Mon, 31 Aug 2026 20:22:41 -0700 Subject: [PATCH 4/5] Drop leftover drain send after a generation bump Leftover hops went through sendUserPrompt with no deliveryGeneration check, so a /clear mid-ingest still landed in the new session. Capture generation at leftover hop time and skip send and recall after ingest when the generation has bumped. Operator Enter stays ungated. --- src/tui/queued-delivery.test.ts | 99 +++++++++++++++++++++++++++++++++ src/tui/queued-delivery.ts | 57 ++++++++++++++++--- src/tui/runner.ts | 21 ++++++- 3 files changed, 169 insertions(+), 8 deletions(-) diff --git a/src/tui/queued-delivery.test.ts b/src/tui/queued-delivery.test.ts index 9d32d10fe..8e4a63ec9 100644 --- a/src/tui/queued-delivery.test.ts +++ b/src/tui/queued-delivery.test.ts @@ -2,6 +2,7 @@ import { describe, expect, test } from "bun:test"; import type { PendingImageAttachment } from "./image-attachments.js"; import { createDeliveryGeneration, + createLeftoverSend, createLiveSteerDeliver, routeQueuedDelivery, } from "./queued-delivery.js"; @@ -177,3 +178,101 @@ describe("createLiveSteerDeliver", () => { 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 index 849231fda..b307cb461 100644 --- a/src/tui/queued-delivery.ts +++ b/src/tui/queued-delivery.ts @@ -64,12 +64,34 @@ export interface CreateLiveSteerDeliverArgs { onFailure: (err: unknown) => void; } -/** - * 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, +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(); @@ -79,8 +101,29 @@ export function createLiveSteerDeliver( if (!stillCurrent()) return; const ingested = await args.ingest(text, pending); if (!stillCurrent()) return; - args.deliver(ingested.text, ingested.attachments); + 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.ts b/src/tui/runner.ts index ae9c93dd2..c76a079d7 100644 --- a/src/tui/runner.ts +++ b/src/tui/runner.ts @@ -187,6 +187,7 @@ import { createCycleTextRecorder } from "../session/stream-journal.js"; import { mountRunnerHost } from "./runner-host.js"; import { createDeliveryGeneration, + createLeftoverSend, createLiveSteerDeliver, routeQueuedDelivery, } from "./queued-delivery.js"; @@ -2296,7 +2297,25 @@ export async function runTUI(initialConfig: Config): Promise { }), interrupt, deliver: routeQueuedDelivery({ - send, + 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, From 5454e7ef6596b62d141a96ea0f8132db93e698e6 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Mon, 31 Aug 2026 21:16:13 -0700 Subject: [PATCH 5/5] Pass deliver into host tests that landed with Codex cost The required deliver hop was already on older fixtures. Two Codex hide tests from main did not have it after rebase, so typecheck failed. --- src/tui/runner-host.test.ts | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src/tui/runner-host.test.ts b/src/tui/runner-host.test.ts index d1b16583e..ac499fe20 100644 --- a/src/tui/runner-host.test.ts +++ b/src/tui/runner-host.test.ts @@ -466,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"] }, @@ -515,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"] },