From 864a948a44b25ac6baaf863055bed54c4a5bb7a7 Mon Sep 17 00:00:00 2001 From: Ben Wachman Date: Sat, 10 Oct 2026 19:49:03 -0400 Subject: [PATCH 1/2] feat(tui): show queued operator input --- docs/TUI.md | 41 +- src/tui/delivery-queue.ts | 39 +- src/tui/geometry/resolve.test.ts | 176 +++++++ src/tui/geometry/resolve.ts | 3 + src/tui/geometry/zones.ts | 10 + src/tui/live-session-port.ts | 9 +- src/tui/product-host.ts | 6 + src/tui/queued-delivery-hop.test.ts | 249 ++++++++++ src/tui/queued-input-chrome.test.ts | 727 ++++++++++++++++++++++++++++ src/tui/queued-input.test.ts | 46 ++ src/tui/queued-input.ts | 69 +++ src/tui/runner/index.ts | 9 + src/tui/runner/state.ts | 6 + src/tui/runner/submit.ts | 45 +- src/tui/runtime-bridge.ts | 129 ++++- src/tui/shell/chrome.ts | 31 ++ src/tui/shell/index.ts | 11 + src/tui/shell/internals.ts | 4 + src/tui/shell/layout.ts | 1 + src/tui/stream-event-map.test.ts | 9 + src/tui/stream-event-map.ts | 13 +- 21 files changed, 1608 insertions(+), 25 deletions(-) create mode 100644 src/tui/geometry/resolve.test.ts create mode 100644 src/tui/queued-input-chrome.test.ts create mode 100644 src/tui/queued-input.test.ts create mode 100644 src/tui/queued-input.ts diff --git a/docs/TUI.md b/docs/TUI.md index 4a18263e7..2073ac17a 100644 --- a/docs/TUI.md +++ b/docs/TUI.md @@ -29,8 +29,9 @@ terminal with nothing optional showing, the transcript floor is 12 rows (`IDLE_TRANSCRIPT_FLOOR`); with an inset overlay open the floor drops to 8 rows (`OVERLAY_TRANSCRIPT_FLOOR`) so the log stays glanceable underneath a permission prompt. When space is scarce, collapse follows a fixed order — transient banners first, then settings/plugin notices, then -task/agents strips, then progress, then the INPUT REQUIRED strip (the last -optional row), then the prompt itself shrinks one +task/agents strips, then progress, then the operator's own queued drafts +(pending), then the QUEUED INPUT strip, then WORKER WAITING, then INPUT +REQUIRED (the last optional row), then the prompt itself shrinks one row at a time down to its 3-row base — never the transcript (`COLLAPSE_ORDER` in `zones.ts`). @@ -311,10 +312,14 @@ clocks. ### WORKER WAITING strip While any root worker is parked on `ask_director`, a one-row **WORKER -WAITING** strip sits directly on the prompt box (zone `worker_wait`): +WAITING** strip sits on the prompt box (zone `worker_wait`). When the INPUT +REQUIRED or QUEUED INPUT strips are also visible, they stack between it and +the prompt box, in that order: ``` ◆ WORKER WAITING · director reply needed (send_input) · builder (copy assets): Which destination path should I use? (+1 more) +◆ INPUT REQUIRED · operator answer needed · Which environment should I deploy to? +◆ QUEUED INPUT · 1 ``` It names the worker, previews its question, and says the director owns the @@ -354,8 +359,34 @@ dispose clears every item (`src/tui/operator-input-required.ts`, fed from question shows at a time, with `(+N more)` for the rest. At narrow widths the question truncates and then drops first, then the routing copy. Below 26 cells the `(+N more)` count is sliced away (cut at 25, gone at 16), leaving -only the label. When WORKER WAITING is also showing, it sits directly above -this strip. +only the label. WORKER WAITING, when showing, sits directly above this +strip, and QUEUED INPUT sits directly below it. + +### QUEUED INPUT strip + +While an operator-originated message has been accepted by the live agent but +not yet read by the director, a one-row **QUEUED INPUT** strip sits directly +on the prompt box (zone `queued_input`): + +``` +◆ QUEUED INPUT · 1 +``` + +It counts accepted-but-unread operator input (mid-run steers and sent +follow-ups) while the runtime owns the message but the director has not yet +dequeued it. It is display only: no routing, mailbox ordering, transcript +row, echo, or recovery behavior changes. The strip reads the accepted-input +ledger and nothing else (`src/tui/queued-input.ts`, painted by `paintChrome` +in `shell/chrome.ts`). An item's identity is the inbound message's header +`messageId`: it enters when the delivery hop reports `accepted`, and it +leaves on the reactor's `message.run.started` event, which fires at the +moment the message is dequeued for the director. A message whose read signal +arrived before its acceptance (the send path) never paints. Worker wakes and +other harness-originated text never enter the ledger, so they never paint +this strip. `/clear` and `/new` clear it with the session, and a stale +acceptance from a cleared session is ignored. The row is one line, never +consumes or edits the prompt, and collapses before WORKER WAITING under +height pressure. ### Transcript spawn_agent rows (history anchors) diff --git a/src/tui/delivery-queue.ts b/src/tui/delivery-queue.ts index cca0f3fd0..52c7dee69 100644 --- a/src/tui/delivery-queue.ts +++ b/src/tui/delivery-queue.ts @@ -192,6 +192,12 @@ export interface QueueItem { readonly enqueuedAt: number; /** Images attached to this message, delivered with it at the boundary. */ readonly attachments?: readonly PendingImageAttachment[]; + /** + * False marks harness-originated text (ask wakes, mailbox mail) so it never + * enters the queued-input ledger as operator input; absent means composer + * input. + */ + readonly operatorOriginated?: boolean; } export type RunState = "idle" | "busy"; @@ -457,11 +463,13 @@ export interface RouteQueuedDeliveryArgs { text: string, attachments?: readonly PendingImageAttachment[], settle?: DeliverySettle, + operatorOriginated?: boolean, ) => void; deliverSteer: ( text: string, attachments?: readonly PendingImageAttachment[], settle?: DeliverySettle, + operatorOriginated?: boolean, ) => void; /** * True only while the bridge is draining steers at a live parent @@ -474,12 +482,12 @@ export interface RouteQueuedDeliveryArgs { export function routeQueuedDelivery( args: RouteQueuedDeliveryArgs, ): ProductHostDeliver { - return (text, kind, attachments, settle) => { + return (text, kind, attachments, settle, operatorOriginated) => { if (kind === "steer" && args.parentCycleLive()) { - args.deliverSteer(text, attachments, settle); + args.deliverSteer(text, attachments, settle, operatorOriginated); return; } - args.send(text, attachments, settle); + args.send(text, attachments, settle, operatorOriginated); }; } @@ -531,6 +539,7 @@ export interface CreateLiveSteerDeliverArgs { text: string, attachments: readonly PendingImageAttachment[], settle?: DeliverySettle, + operatorOriginated?: boolean, ) => void; captureGeneration: () => () => boolean; onFailure: (err: unknown) => void; @@ -549,6 +558,8 @@ export interface CreateLeftoverSendArgs { send: ( text: string, attachments: readonly PendingImageAttachment[], + settle?: DeliverySettle, + operatorOriginated?: boolean, ) => MaybeAsyncDeliveryResult; /** * Up/Down recall. Called with the original text only when the hop is @@ -569,6 +580,7 @@ interface GenerationGatedHopArgs { text: string, attachments: readonly PendingImageAttachment[], settle?: DeliverySettle, + operatorOriginated?: boolean, ) => MaybeAsyncDeliveryResult; /** * Leftover/send settles from the send promise; live steer uses the @@ -593,8 +605,9 @@ function createGenerationGatedHop( text: string, attachments?: readonly PendingImageAttachment[], settle?: DeliverySettle, + operatorOriginated?: boolean, ) => void { - return (text, attachments, settle) => { + return (text, attachments, settle, operatorOriginated) => { const stillCurrent = args.captureGeneration(); const pending = attachments ?? []; let settled = false; @@ -635,14 +648,24 @@ function createGenerationGatedHop( } args.recordSent?.(text); if (args.settleFromHopResult === true) { - const result = await args.hop(ingested.text, ingested.attachments); + const result = await args.hop( + ingested.text, + ingested.attachments, + undefined, + operatorOriginated, + ); finish(result ?? { status: "accepted" }); return; } // Live steer: hop receives settle and reports the eventual result. - args.hop(ingested.text, ingested.attachments, (result) => { - finish(result); - }); + args.hop( + ingested.text, + ingested.attachments, + (result) => { + finish(result); + }, + operatorOriginated, + ); }) .catch((err: unknown) => { finish({ diff --git a/src/tui/geometry/resolve.test.ts b/src/tui/geometry/resolve.test.ts new file mode 100644 index 000000000..c2b9ac314 --- /dev/null +++ b/src/tui/geometry/resolve.test.ts @@ -0,0 +1,176 @@ +/** + * QUEUED INPUT zone geometry: requested height, seating on the prompt box, + * coexistence with WORKER WAITING / pending, collapse before worker_wait and + * restore when room returns, and in-bounds no-overlap on a short terminal. + */ +import { describe, expect, test } from "bun:test"; +import { defined } from "../../../testkit/defined.js"; +import { + COLLAPSE_ORDER, + IDLE_TRANSCRIPT_FLOOR, + PROMPT_BASE_ROWS, + PROMPT_IDLE_ROWS, + ZONE_REGISTRY, + resolveGeometry, + type GeometryInput, +} from "./index.js"; +import { PAINT_ORDER } from "./zones.js"; + +function idle80x24(overrides: Partial = {}) { + return resolveGeometry({ + terminal: { columns: 80, rows: 24 }, + ...overrides, + }); +} + +describe("queued_input zone", () => { + test("requested visibility seats exactly one row directly on the prompt box", () => { + const baseline = idle80x24(); + const layout = idle80x24({ visibility: { queuedInput: true } }); + const strip = defined(layout.regions.queued_input, "queued_input"); + const prompt = defined(layout.regions.prompt, "prompt"); + expect(strip.height).toBe(1); + expect(strip.y + strip.height).toBe(prompt.y); + // The strip takes its row from the transcript, never the composer. + expect(layout.heights.prompt).toBe(baseline.heights.prompt); + expect(layout.transcriptHeight).toBe(baseline.transcriptHeight - 1); + }); + + test("zero visibility costs zero chrome", () => { + const layout = idle80x24(); + expect(layout.heights.queued_input).toBe(0); + expect(layout.regions.queued_input).toBeUndefined(); + }); + + test("sits between worker_wait and prompt in paint order", () => { + expect(PAINT_ORDER.indexOf("worker_wait")).toBeLessThan( + PAINT_ORDER.indexOf("queued_input"), + ); + expect(PAINT_ORDER.indexOf("queued_input")).toBeLessThan( + PAINT_ORDER.indexOf("prompt"), + ); + const layout = idle80x24({ + visibility: { workerWait: true, queuedInput: true }, + }); + const waiting = defined(layout.regions.worker_wait, "worker_wait"); + const queued = defined(layout.regions.queued_input, "queued_input"); + const prompt = defined(layout.regions.prompt, "prompt"); + expect(waiting.height).toBe(1); + expect(queued.height).toBe(1); + expect(waiting.y + waiting.height).toBe(queued.y); + expect(queued.y + queued.height).toBe(prompt.y); + expect(layout.heights.prompt).toBe(PROMPT_IDLE_ROWS); + }); + + test("coexists with pending items above worker_wait", () => { + const layout = idle80x24({ + visibility: { pending: 2, workerWait: true, queuedInput: true }, + }); + const pending = defined(layout.regions.pending, "pending"); + const waiting = defined(layout.regions.worker_wait, "worker_wait"); + const queued = defined(layout.regions.queued_input, "queued_input"); + const prompt = defined(layout.regions.prompt, "prompt"); + expect(pending.height).toBe(2); + expect(waiting.height).toBe(1); + expect(queued.height).toBe(1); + expect(pending.y + pending.height).toBe(waiting.y); + expect(waiting.y + waiting.height).toBe(queued.y); + expect(queued.y + queued.height).toBe(prompt.y); + expect(layout.transcriptHeight).toBeGreaterThanOrEqual( + layout.transcriptFloor, + ); + }); + + test("collapses before worker_wait under height pressure", () => { + // Pending own-input rows cut after banners but before the queued-input + // strip; the queued-input strip cuts before the worker-wait strip. + expect(COLLAPSE_ORDER.indexOf("pending")).toBeLessThan( + COLLAPSE_ORDER.indexOf("queued_input"), + ); + expect(COLLAPSE_ORDER.indexOf("queued_input")).toBeLessThan( + COLLAPSE_ORDER.indexOf("worker_wait"), + ); + expect(COLLAPSE_ORDER.indexOf("worker_wait")).toBeLessThan( + COLLAPSE_ORDER.indexOf("prompt"), + ); + const layout = resolveGeometry({ + terminal: { columns: 80, rows: 18 }, + visibility: { + notice: true, + pending: 6, + agents: 11, + task: 3, + workerWait: true, + queuedInput: true, + }, + }); + expect(layout.heights.notice).toBe(0); + expect(layout.heights.pending).toBe(0); + expect(layout.heights.queued_input).toBe(0); + // A parked worker stays visible: its row is the last optional cut. + expect(layout.heights.worker_wait).toBe(1); + expect(layout.transcriptHeight).toBeGreaterThanOrEqual( + layout.transcriptFloor, + ); + }); + + test("restores when room returns after a constrained-height collapse", () => { + const narrow = resolveGeometry({ + terminal: { columns: 80, rows: 18 }, + visibility: { + notice: true, + pending: 6, + agents: 11, + task: 3, + workerWait: true, + queuedInput: true, + }, + }); + expect(narrow.heights.queued_input).toBe(0); + // The ledger is untouched by collapse; the same visibility on a taller + // terminal seats the strip again. + const restored = resolveGeometry({ + terminal: { columns: 80, rows: 24 }, + visibility: { + notice: true, + pending: 6, + agents: 11, + task: 3, + workerWait: true, + queuedInput: true, + }, + }); + expect(restored.heights.queued_input).toBe(1); + expect(defined(restored.regions.queued_input).height).toBe(1); + expect(restored.transcriptHeight).toBeGreaterThanOrEqual( + IDLE_TRANSCRIPT_FLOOR, + ); + }); + + test("short terminal stays in bounds with no overlap", () => { + const layout = resolveGeometry({ + terminal: { columns: 12, rows: 10 }, + visibility: { pending: 1, workerWait: true, queuedInput: true }, + }); + const prompt = defined(layout.regions.prompt, "prompt"); + expect(prompt.y + prompt.height).toBeLessThanOrEqual(10); + expect(layout.heights.prompt).toBeGreaterThanOrEqual(PROMPT_BASE_ROWS); + // Whatever collapsed, the visible regions y-stack without overlap. + const regions = PAINT_ORDER.map((id) => layout.regions[id]).filter( + (region) => region !== undefined && region.height > 0, + ); + for (let index = 1; index < regions.length; index += 1) { + const previous = regions[index - 1]; + const current = regions[index]; + if (previous === undefined || current === undefined) continue; + expect(previous.y + previous.height).toBe(current.y); + } + expect(layout.regions.queued_input).toBeUndefined(); + }); + + test("registry bounds match the zone declaration", () => { + expect(ZONE_REGISTRY.queued_input.min).toBe(0); + expect(ZONE_REGISTRY.queued_input.max).toBe(1); + expect(ZONE_REGISTRY.queued_input.alwaysOn).toBe(false); + }); +}); diff --git a/src/tui/geometry/resolve.ts b/src/tui/geometry/resolve.ts index b2827a4e5..988eaf9f4 100644 --- a/src/tui/geometry/resolve.ts +++ b/src/tui/geometry/resolve.ts @@ -43,6 +43,8 @@ export interface ZoneVisibility { readonly workerWait?: boolean; /** INPUT REQUIRED strip for marked primary ask_operator gates (default off). */ readonly inputRequired?: boolean; + /** Accepted operator input awaiting the correlated runtime echo. */ + readonly queuedInput?: boolean; /** Progress: false/omit = 0; true = 2; or explicit 1|2. */ readonly progress?: boolean | 1 | 2; /** Progress divider (0–1). Default on when progress is shown. */ @@ -153,6 +155,7 @@ export function desiredHeights(input: GeometryInput): MutableHeights { pending: clamp(boolOrRows(vis.pending, 1), 0, ZONE_REGISTRY.pending.max), worker_wait: vis.workerWait === true ? 1 : 0, input_required: vis.inputRequired === true ? 1 : 0, + queued_input: vis.queuedInput === true ? 1 : 0, prompt: promptRows, task: clamp(boolOrRows(vis.task, 1), 0, ZONE_REGISTRY.task.max), // The board asks for exactly the rows it paints; the fraction stops a large diff --git a/src/tui/geometry/zones.ts b/src/tui/geometry/zones.ts index efde19f0d..3c079ee96 100644 --- a/src/tui/geometry/zones.ts +++ b/src/tui/geometry/zones.ts @@ -10,6 +10,7 @@ const ZONE_IDS = [ "pending", "worker_wait", "input_required", + "queued_input", "prompt", "task", "agents", @@ -100,6 +101,13 @@ export const ZONE_REGISTRY: Readonly> = { idleDefault: 0, alwaysOn: false, }, + queued_input: { + id: "queued_input", + min: 0, + max: 1, + idleDefault: 0, + alwaysOn: false, + }, // Grows with the draft; the resolver caps it at PROMPT_CAP_FRACTION and // collapses toward min when the transcript would breach its floor. prompt: { @@ -212,6 +220,7 @@ export const COLLAPSE_ORDER = [ // Pending items are the operator's own queued words: cut late, after every // banner and strip. "pending", + "queued_input", // A parked worker is standing state the operator cannot see anywhere else // once the wake turn scrolls away, so its single row is cut late. "worker_wait", @@ -239,6 +248,7 @@ export const PAINT_ORDER = [ "pending", "worker_wait", "input_required", + "queued_input", "prompt", ] as const satisfies readonly ZoneId[]; diff --git a/src/tui/live-session-port.ts b/src/tui/live-session-port.ts index bab79764d..47042a2ae 100644 --- a/src/tui/live-session-port.ts +++ b/src/tui/live-session-port.ts @@ -29,6 +29,7 @@ export interface LiveSessionPortDeps { kind: QueueKind, attachments?: readonly PendingImageAttachment[], settle?: DeliverySettle, + operatorOriginated?: boolean, ) => void; } @@ -57,7 +58,13 @@ export function createLiveSessionPort(deps: LiveSessionPortDeps): SessionPort { deps.interrupt(); }, deliver: (item: QueueItem, settle?: DeliverySettle): void => { - deps.deliver(item.text, item.kind, item.attachments, settle); + deps.deliver( + item.text, + item.kind, + item.attachments, + settle, + item.operatorOriginated, + ); }, }; } diff --git a/src/tui/product-host.ts b/src/tui/product-host.ts index 45f7fb801..9c986ee4b 100644 --- a/src/tui/product-host.ts +++ b/src/tui/product-host.ts @@ -138,6 +138,12 @@ export type ProductHostDeliver = ( kind: QueueKind, attachments?: readonly PendingImageAttachment[], settle?: DeliverySettle, + /** + * False marks harness-originated text (ask wakes, mailbox mail) so the + * operator flag is never stamped and the queued-input ledger never records + * it; absent means composer input. + */ + operatorOriginated?: boolean, ) => void; /** diff --git a/src/tui/queued-delivery-hop.test.ts b/src/tui/queued-delivery-hop.test.ts index 28c91a6f6..2fa054f1f 100644 --- a/src/tui/queued-delivery-hop.test.ts +++ b/src/tui/queued-delivery-hop.test.ts @@ -3,12 +3,15 @@ * Agent.deliver (deliverSteer); leftover / fleet-hold / interrupt use send. */ import { describe, expect, test } from "bun:test"; +import { type Agent } from "@intx/agent"; +import type { InboundMessage } from "@intx/types/runtime"; import { attachSessionBridge, type SessionBridge } from "./runtime-bridge"; import { createLiveSessionPort } from "./live-session-port"; import { createAppShell } from "./shell/index"; import { withTestRenderer } from "./harness"; import { badgeCount, + createDeliveryGeneration, createLiveSteerDeliver, createSessionOperationQueue, pause, @@ -17,6 +20,13 @@ import { type DeliverySettle, type QueueItem, } from "./delivery-queue.js"; +import { isOperatorOriginated } from "../agent/message-provenance.js"; +import { createDeliverRouting } from "./runner/submit.js"; +import type { + InferenceAttemptIdentity, + RunnerServices, + RunnerState, +} from "./runner/state.js"; type Shell = ReturnType; @@ -463,3 +473,242 @@ describe("closed-target recovery", () => { }); }); }); + +function received(messageId: string) { + return { + type: "message.received", + data: { + message: { + content: "same text", + flags: ["operator-originated"], + headers: { messageId }, + }, + }, + }; +} + +function started(messageId: string) { + return { type: "message.run.started", data: { messageId } }; +} + +describe("queued-input ledger through the runner observer", () => { + /** + * Wire the real `createDeliverRouting` (submit.ts) onto a bridge, binding + * the accepted-input observers the mounted host installs, so a drained + * steer/follow-up exercises the full submit-to-observer path. + */ + function withDeliverRoutingBridge( + results: AgentDeliveryResult[], + fn: (ctx: { + shell: Shell; + bridge: SessionBridge; + awaitTail: () => Promise; + bumpGeneration: () => void; + }) => void | Promise, + ): Promise { + return 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 sessionOps = createSessionOperationQueue(); + const deliveryGeneration = createDeliveryGeneration(); + const begin = (messageId: string): void => { + bridgeRef.current?.beginQueuedInputDelivery(messageId); + }; + const accept = (messageId: string): void => { + bridgeRef.current?.recordAcceptedInput(messageId); + }; + const discard = (messageId: string): void => { + bridgeRef.current?.discardQueuedInputDelivery(messageId); + }; + const sendWithAttemptIdentity = async ( + message: InboundMessage, + ): Promise => { + const isOperatorMessage = isOperatorOriginated(message.flags); + const messageId = message.headers.messageId; + if (isOperatorMessage && typeof messageId === "string") + begin(messageId); + const next = results.shift() ?? { status: "accepted" }; + if (isOperatorMessage && typeof messageId === "string") { + if (next.status === "accepted") accept(messageId); + else discard(messageId); + } + return next; + }; + const state = { + config: { cwd: "/tmp" }, + currentAgent: { + deliver: () => undefined, + }, + fatalBuildError: null, + sendWithAttemptIdentity, + enqueueAgentDeliver: ( + deliverToLiveAgent: () => void, + onSettle?: (result: AgentDeliveryResult) => void, + ): void => { + const stillCurrent = deliveryGeneration.capture(); + void sessionOps.enqueue(async () => { + if (!stillCurrent()) { + onSettle?.({ + status: "not-delivered", + reason: "superseded", + detail: "session identity changed before delivery", + }); + return; + } + deliverToLiveAgent(); + const next = results.shift() ?? { status: "accepted" }; + onSettle?.(next); + }); + }, + beginAcceptedInputObserver: begin, + acceptedInputObserver: accept, + discardAcceptedInputObserver: discard, + host: { + get bridge() { + return bridgeRef.current; + }, + }, + } as unknown as RunnerState; + const services = { + sessionOps, + deliveryGeneration, + } as unknown as RunnerServices; + const live = { + attemptIdentity: (): InferenceAttemptIdentity => ({ + providerId: "probe", + }), + agentProxy: state.currentAgent as unknown as Agent, + }; + const bridge = attachSessionBridge( + shell, + createLiveSessionPort({ + send: () => undefined, + interrupt: () => undefined, + deliver: createDeliverRouting(state, services, live), + }), + ); + bridgeRef.current = bridge; + try { + await fn({ + shell, + bridge, + awaitTail: () => sessionOps.awaitTail(), + bumpGeneration: () => deliveryGeneration.bump(), + }); + } finally { + bridge.dispose(); + shell.dispose(); + } + }, + { width: 80, height: 24 }, + ); + } + + test("drain-accepted live steer adds exactly one ledger entry through the observer", async () => { + await withDeliverRoutingBridge( + [{ status: "accepted" }], + async ({ shell, bridge, awaitTail }) => { + 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: {} }); + await awaitTail(); + expect(shell.queuedInput.messageIds).toHaveLength(1); + expect(shell.queuedInputRow.visible).toBe(true); + const messageId = shell.queuedInput.messageIds[0]; + expect(messageId).toBeDefined(); + if (messageId === undefined) return; + // Receipt alone never consumes; message.run.started is the read signal. + bridge.handle(received(messageId)); + expect(shell.queuedInput.messageIds).toEqual([messageId]); + bridge.handle(started(messageId)); + expect(shell.queuedInput.messageIds).toEqual([]); + expect(shell.queuedInputRow.visible).toBe(false); + }, + ); + }); + + test("drain-accepted leftover send adds exactly one ledger entry", async () => { + await withDeliverRoutingBridge( + [{ status: "accepted" }], + async ({ shell, bridge, awaitTail }) => { + 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: {} }); + await awaitTail(); + expect(shell.queuedInput.messageIds).toHaveLength(1); + expect(shell.queuedInputRow.visible).toBe(true); + }, + ); + }); + + test("agent-closed delivery never enters the ledger", async () => { + await withDeliverRoutingBridge( + [ + { + status: "not-delivered", + reason: "agent-closed", + detail: "agent is closed", + }, + ], + async ({ shell, bridge, awaitTail }) => { + bridge.submit("asap", "steer"); + bridge.handle({ + type: "tool.start", + data: { call: { id: "c1", name: "run_shell" } }, + }); + bridge.handle({ type: "inference.done", data: {} }); + await awaitTail(); + expect(shell.queuedInput.messageIds).toEqual([]); + expect(shell.queuedInputRow.visible).toBe(false); + }, + ); + }); + + test("uncertain delivery never enters the ledger", async () => { + await withDeliverRoutingBridge( + [{ status: "uncertain", detail: "connection reset" }], + async ({ shell, bridge, awaitTail }) => { + bridge.submit("asap", "steer"); + bridge.handle({ + type: "tool.start", + data: { call: { id: "c1", name: "run_shell" } }, + }); + bridge.handle({ type: "inference.done", data: {} }); + await awaitTail(); + expect(shell.queuedInput.messageIds).toEqual([]); + expect(shell.queuedInputRow.visible).toBe(false); + }, + ); + }); + + test("superseded delivery (generation bump) never enters the ledger", async () => { + await withDeliverRoutingBridge( + [], + async ({ shell, bridge, awaitTail, bumpGeneration }) => { + bridge.submit("asap", "steer"); + bridge.handle({ + type: "tool.start", + data: { call: { id: "c1", name: "run_shell" } }, + }); + bridge.handle({ type: "inference.done", data: {} }); + // The session rotates while the deliver is queued: the stale hop drops. + bumpGeneration(); + await awaitTail(); + expect(shell.queuedInput.messageIds).toEqual([]); + expect(shell.queuedInputRow.visible).toBe(false); + }, + ); + }); +}); diff --git a/src/tui/queued-input-chrome.test.ts b/src/tui/queued-input-chrome.test.ts new file mode 100644 index 000000000..9480ef692 --- /dev/null +++ b/src/tui/queued-input-chrome.test.ts @@ -0,0 +1,727 @@ +import { expect, test } from "bun:test"; +import { createReactor, type ReactorEmittedEvent } from "@intx/inference"; +import { createDefaultDependencies } from "@intx/inference/providers"; +import type { + ContextCommit, + ContextStore, + InboundMessage, + ReactorCapabilities, + ReactorDirector, + ReactorState, + TokenUsage, +} from "@intx/types/runtime"; +import { withTestRenderer } from "./harness.js"; +import { attachSessionBridge, type SessionBridge } from "./runtime-bridge.js"; +import { paintChrome, relayout, setWorkerWaitAsks } from "./shell/chrome.js"; +import { createAppShell } from "./shell/index.js"; +import type { PendingAskWake } from "../subagent/fleet-report.js"; +import { createLiveSessionPort } from "./live-session-port.js"; +import { + createLiveSteerDeliver, + createSessionOperationQueue, + deliverAgentMessage, + routeQueuedDelivery, +} from "./delivery-queue.js"; +import { userInboundMessage } from "./runner/submit.js"; + +const WAITING: PendingAskWake = { + sessionId: "worker-session", + agentId: "worker", + description: "inspect state", + question: "Should I continue?", + questionId: "question", +}; + +function received( + messageId: string, + flags: readonly string[] = ["operator-originated"], +) { + return { + type: "message.received", + data: { message: { content: "same text", flags, headers: { messageId } } }, + }; +} + +function started(messageId: string) { + return { type: "message.run.started", data: { messageId } }; +} + +test("accepted input chrome consumes only matching operator identities and survives redraw", async () => { + await withTestRenderer( + async (h) => { + const shell = createAppShell(h.renderer, { + terminal: { columns: 80, rows: 24 }, + wireKeys: false, + }); + const bridge = attachSessionBridge(shell); + try { + bridge.recordAcceptedInput("first"); + bridge.recordAcceptedInput("second"); + bridge.recordAcceptedInput("first"); + expect(shell.queuedInput.messageIds).toEqual(["first", "second"]); + + bridge.handle(received("foreign", ["system"])); + bridge.handle({ + type: "message.received", + data: { message: { content: "same text" } }, + }); + expect(shell.queuedInput.messageIds).toEqual(["first", "second"]); + + // Receipt alone never consumes the ledger: the read signal is + // message.run.started. + bridge.handle(received("second")); + expect(shell.queuedInput.messageIds).toEqual(["first", "second"]); + shell.prompt.value = "keep this draft"; + paintChrome(shell, { force: true }); + expect(shell.prompt.value).toBe("keep this draft"); + expect(shell.queuedInputRow.visible).toBe(true); + + bridge.handle(started("second")); + expect(shell.queuedInput.messageIds).toEqual(["first"]); + bridge.handle(started("first")); + expect(shell.queuedInput.messageIds).toEqual([]); + expect(shell.queuedInputRow.visible).toBe(false); + } finally { + bridge.dispose(); + shell.dispose(); + } + }, + { width: 80, height: 24 }, + ); +}); + +test("steer path: accept between receive and read keeps depth 1 until message.run.started", async () => { + await withTestRenderer( + async (h) => { + const shell = createAppShell(h.renderer, { + terminal: { columns: 80, rows: 24 }, + wireKeys: false, + }); + const bridge = attachSessionBridge(shell); + try { + bridge.beginQueuedInputDelivery("racing"); + bridge.handle(received("racing")); + expect(shell.queuedInput.messageIds).toEqual([]); + bridge.recordAcceptedInput("racing"); + // The director has not read the steer: the row must be visible. + expect(shell.queuedInput.messageIds).toEqual(["racing"]); + expect(shell.queuedInputRow.visible).toBe(true); + + bridge.handle(started("racing")); + expect(shell.queuedInput.messageIds).toEqual([]); + expect(shell.queuedInputRow.visible).toBe(false); + + bridge.recordAcceptedInput("cleared"); + bridge.clearQueuedDelivery(); + expect(shell.queuedInput.messageIds).toEqual([]); + bridge.dispose(); + bridge.recordAcceptedInput("late"); + expect(shell.queuedInput.messageIds).toEqual([]); + } finally { + shell.dispose(); + } + }, + { width: 80, height: 24 }, + ); +}); + +test("send path: read before accept produces no ledger entry", async () => { + await withTestRenderer( + async (h) => { + const shell = createAppShell(h.renderer, { + terminal: { columns: 80, rows: 24 }, + wireKeys: false, + }); + const bridge = attachSessionBridge(shell); + try { + bridge.beginQueuedInputDelivery("sending"); + bridge.handle(received("sending")); + bridge.handle(started("sending")); + // The send resolves accepted only after the director already read the + // message: never paint a stale row. + bridge.recordAcceptedInput("sending"); + expect(shell.queuedInput.messageIds).toEqual([]); + expect(shell.queuedInputRow.visible).toBe(false); + } finally { + bridge.dispose(); + shell.dispose(); + } + }, + { width: 80, height: 24 }, + ); +}); + +test("a delayed acceptance remains correlated after high-volume receives", async () => { + await withTestRenderer( + async (h) => { + const shell = createAppShell(h.renderer, { + terminal: { columns: 80, rows: 24 }, + wireKeys: false, + }); + const bridge = attachSessionBridge(shell); + try { + bridge.beginQueuedInputDelivery("racing"); + bridge.handle(received("racing")); + for (let index = 0; index <= 1000; index += 1) { + bridge.handle(received(`other-${index}`)); + } + bridge.recordAcceptedInput("racing"); + + expect(shell.queuedInput.messageIds).toEqual(["racing"]); + expect(shell.queuedInputRow.visible).toBe(true); + + bridge.handle(started("racing")); + expect(shell.queuedInput.messageIds).toEqual([]); + expect(shell.queuedInputRow.visible).toBe(false); + } finally { + bridge.dispose(); + shell.dispose(); + } + }, + { width: 80, height: 24 }, + ); +}); + +test("duplicate receives and duplicate read signals are idempotent", async () => { + await withTestRenderer( + async (h) => { + const shell = createAppShell(h.renderer, { + terminal: { columns: 80, rows: 24 }, + wireKeys: false, + }); + const bridge = attachSessionBridge(shell); + try { + bridge.recordAcceptedInput("one"); + bridge.recordAcceptedInput("one"); + expect(shell.queuedInput.messageIds).toEqual(["one"]); + bridge.handle(received("one")); + bridge.handle(received("one")); + expect(shell.queuedInput.messageIds).toEqual(["one"]); + bridge.handle(started("one")); + bridge.handle(started("one")); + expect(shell.queuedInput.messageIds).toEqual([]); + expect(shell.queuedInputRow.visible).toBe(false); + } finally { + bridge.dispose(); + shell.dispose(); + } + }, + { width: 80, height: 24 }, + ); +}); + +test("malformed receive and read payloads never disturb the ledger", async () => { + await withTestRenderer( + async (h) => { + const shell = createAppShell(h.renderer, { + terminal: { columns: 80, rows: 24 }, + wireKeys: false, + }); + const bridge = attachSessionBridge(shell); + try { + bridge.recordAcceptedInput("kept"); + bridge.handle({ type: "message.received", data: null }); + bridge.handle({ type: "message.received", data: {} }); + bridge.handle({ + type: "message.received", + data: { message: { content: "no headers" } }, + }); + bridge.handle({ type: "message.run.started", data: null }); + bridge.handle({ type: "message.run.started", data: {} }); + bridge.handle({ + type: "message.run.started", + data: { messageId: 42 }, + }); + expect(shell.queuedInput.messageIds).toEqual(["kept"]); + bridge.handle(started("kept")); + expect(shell.queuedInput.messageIds).toEqual([]); + } finally { + bridge.dispose(); + shell.dispose(); + } + }, + { width: 80, height: 24 }, + ); +}); + +test("queued input and worker waiting occupy independent bottom-stack rows", async () => { + await withTestRenderer( + async (h) => { + const shell = createAppShell(h.renderer, { + terminal: { columns: 32, rows: 24 }, + wireKeys: false, + }); + const bridge = attachSessionBridge(shell); + try { + bridge.recordAcceptedInput("one"); + setWorkerWaitAsks(shell, [WAITING]); + await h.renderOnce(); + await h.renderOnce(); + const rows = h.captureCharFrame().split("\n"); + const waiting = rows.findIndex((row) => row.includes("WAITING")); + const queued = rows.findIndex((row) => row.includes("QUEUED")); + expect(waiting).toBeGreaterThanOrEqual(0); + expect(queued).toBeGreaterThan(waiting); + expect(shell.layout.heights.worker_wait).toBe(1); + expect(shell.layout.heights.queued_input).toBe(1); + expect(shell.layout.regions.worker_wait?.y).toBeLessThan( + shell.layout.regions.queued_input?.y ?? -1, + ); + + relayout(shell, { columns: 12, rows: 10 }); + expect(shell.layout.regions.prompt?.y ?? 0).toBeLessThan(10); + expect(shell.queuedInput.messageIds).toEqual(["one"]); + } finally { + bridge.dispose(); + shell.dispose(); + } + }, + { width: 32, height: 24 }, + ); +}); + +test("busy-Enter pending-only produces no ledger entry until acceptance", async () => { + await withTestRenderer( + async (h) => { + const shell = createAppShell(h.renderer, { + terminal: { columns: 80, rows: 24 }, + wireKeys: false, + run: "busy", + }); + const bridge = attachSessionBridge(shell); + try { + bridge.submit("held follow-up", "steer"); + // The item sits in the pending column: Enter alone never records an + // accepted input. + expect(shell.session.items).toHaveLength(1); + expect(shell.queuedInput.messageIds).toEqual([]); + expect(shell.queuedInputRow.visible).toBe(false); + } finally { + bridge.dispose(); + shell.dispose(); + } + }, + { width: 80, height: 24 }, + ); +}); + +test("late accept after /clear or /new is inert (generation check)", async () => { + await withTestRenderer( + async (h) => { + const shell = createAppShell(h.renderer, { + terminal: { columns: 80, rows: 24 }, + wireKeys: false, + }); + const bridge = attachSessionBridge(shell); + try { + // A live steer begins before the session rotates (/clear or /new both + // route through clearQueuedDelivery). + bridge.beginQueuedInputDelivery("old-session-steer"); + bridge.clearQueuedDelivery(); + // The stale accept lands after the clear: the row must not paint. + bridge.recordAcceptedInput("old-session-steer"); + expect(shell.queuedInput.messageIds).toEqual([]); + expect(shell.queuedInputRow.visible).toBe(false); + + // A fresh session's delivery still works normally. + bridge.beginQueuedInputDelivery("new-session-steer"); + bridge.recordAcceptedInput("new-session-steer"); + expect(shell.queuedInput.messageIds).toEqual(["new-session-steer"]); + bridge.handle(started("new-session-steer")); + expect(shell.queuedInput.messageIds).toEqual([]); + } finally { + bridge.dispose(); + shell.dispose(); + } + }, + { width: 80, height: 24 }, + ); +}); + +test("delivered ask wake never enters the queued-input ledger", async () => { + await withTestRenderer( + async (h) => { + const shell = createAppShell(h.renderer, { + terminal: { columns: 80, rows: 24 }, + wireKeys: false, + }); + const delivered: { + text: string; + kind: string; + operatorOriginated: boolean | undefined; + }[] = []; + const bridge = attachSessionBridge( + shell, + createLiveSessionPort({ + send: () => undefined, + interrupt: () => undefined, + deliver: (text, kind, _attachments, _settle, operatorOriginated) => { + delivered.push({ text, kind, operatorOriginated }); + }, + }), + ); + try { + bridge.handle({ type: "agent-ask", asks: [WAITING] }); + // The wake leaves through the port marked as harness text, so it can + // never read as operator input or paint a queued-input row. + expect(delivered).toHaveLength(1); + expect(delivered[0]?.operatorOriginated).toBe(false); + expect(delivered[0]?.text.startsWith("ask_director wake")).toBe(true); + expect(shell.queuedInput.messageIds).toEqual([]); + expect(shell.queuedInputRow.visible).toBe(false); + } finally { + bridge.dispose(); + shell.dispose(); + } + }, + { width: 80, height: 24 }, + ); +}); + +function emptyUsage(): TokenUsage { + return { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, thinking: 0 }; +} + +function makeContextStore(): ContextStore { + async function commit( + options: { message: string }, + _signal?: AbortSignal, + ): Promise { + return { hash: "abc", message: options.message, timestamp: Date.now() }; + } + return { + async load() { + return { + turns: [], + pendingOperations: [], + tokenUsage: emptyUsage(), + connectorState: null, + }; + }, + setConnectorState() { + /* noop */ + }, + commit, + async branch() { + /* noop */ + }, + async log() { + return []; + }, + async readAt() { + return []; + }, + async writeBlob() { + /* noop */ + }, + async readBlob(key) { + throw new Error(`no blob for key ${key}`); + }, + async writePrompt() { + /* noop */ + }, + async writeResponse() { + /* noop */ + }, + async writeManifest() { + /* noop */ + }, + async writeTurns() { + /* noop */ + }, + async writeMetadata() { + /* noop */ + }, + async readManifestHistory() { + return []; + }, + }; +} + +function inboundMessage(content: string): InboundMessage { + return { + ref: { uid: 1, mailbox: "INBOX" }, + headers: { + from: "user@local", + to: ["agent@local"], + date: new Date().toISOString(), + messageId: `<${crypto.randomUUID()}@local>`, + interchangeType: "conversation.message", + }, + flags: [], + signatureStatus: "missing", + content, + }; +} + +async function tickUntil( + predicate: () => boolean, + timeoutMs = 8000, +): Promise { + const deadline = Date.now() + timeoutMs; + for (;;) { + if (predicate()) return; + if (Date.now() > deadline) { + throw new Error("tickUntil timed out waiting for condition"); + } + await new Promise((resolve) => setTimeout(resolve, 5)); + } +} + +test("mid-run steer paints QUEUED INPUT until the real director reads it", async () => { + await withTestRenderer( + async (h) => { + const shell = createAppShell(h.renderer, { + terminal: { columns: 80, rows: 24 }, + wireKeys: false, + run: "busy", + }); + let releaseTool: () => void = () => undefined; + const toolGate = new Promise((resolve) => { + releaseTool = resolve; + }); + // The steer becomes visible only while the director has not read it: the + // second inference holds the cycle open, so the steer's message.received + // waits behind it and message.run.started has not fired. Releasing this + // gate lets the cycle settle and the director read the steer. + let releaseSecondInfer: () => void = () => undefined; + const secondInferGate = new Promise((resolve) => { + releaseSecondInfer = resolve; + }); + let receivedCount = 0; + let inferCount = 0; + const events: ReactorEmittedEvent[] = []; + const steerMessageId: { current: string | undefined } = { + current: undefined, + }; + const bridgeRef: { current: SessionBridge | undefined } = { + current: undefined, + }; + const queue = createSessionOperationQueue(); + + const director: ReactorDirector = { + async decide(event, _state: ReactorState, caps: ReactorCapabilities) { + if (event.type === "message.received") { + receivedCount += 1; + if (receivedCount === 1) return [caps.infer()]; + return [caps.reply("steer read")]; + } + if (event.type === "inference.done") { + const toolCall = event.turn.content.find( + (block) => block.type === "tool_call", + ); + if (toolCall !== undefined) { + return [ + caps.executeTools([ + { + id: toolCall.id, + name: toolCall.name, + arguments: toolCall.arguments, + }, + ]), + ]; + } + return [caps.reply("ok")]; + } + if (event.type === "tool.done") return [caps.infer()]; + return [caps.done()]; + }, + }; + + const reactor = createReactor({ + sessionId: "cl-10437-steer-probe", + director, + source: { + id: "probe:model", + provider: "openai", + baseURL: "https://probe.invalid", + credentialId: "probe", + model: "probe-model", + }, + toolRunner: { + async run(call) { + await toolGate; + return { callId: call.id, content: "tool result" }; + }, + }, + contextStore: makeContextStore(), + onEvent: (event) => { + events.push(event); + bridgeRef.current?.handle(event); + }, + deps: createDefaultDependencies(), + doomLoopThreshold: false, + inferenceRunner(opts) { + return (async function* () { + inferCount += 1; + if (inferCount === 2) await secondInferGate; + yield { + type: "inference.start", + seq: opts.nextSeq(), + data: { model: "probe-model" }, + }; + if (inferCount === 1) { + // First inference: the model asks for a tool, which keeps the + // parent cycle live while the operator steers mid-run. + yield { + type: "inference.tool_call.start", + seq: opts.nextSeq(), + data: { + callId: "call-1", + name: "probe_tool", + partial: { text: "" }, + }, + }; + yield { + type: "inference.tool_call.end", + seq: opts.nextSeq(), + data: { + callId: "call-1", + name: "probe_tool", + arguments: {}, + partial: { text: "" }, + }, + }; + } + yield { + type: "inference.done", + seq: opts.nextSeq(), + data: { + turn: + inferCount === 1 + ? { + role: "assistant", + content: [ + { + type: "tool_call", + id: "call-1", + name: "probe_tool", + arguments: {}, + }, + ], + model: "probe-model", + timestamp: Date.now(), + } + : { + role: "assistant", + content: [{ type: "text", text: "ok" }], + model: "probe-model", + timestamp: Date.now(), + }, + usage: emptyUsage(), + source: { + sourceId: "probe:model", + provider: "openai", + model: "probe-model", + }, + }, + }; + })(); + }, + }); + + const agent = { + deliver: (message: InboundMessage) => { + reactor.deliver(message); + }, + }; + + const bridge = attachSessionBridge( + shell, + createLiveSessionPort({ + send: (text) => { + throw new Error( + `probe should not use the leftover send path: ${text}`, + ); + }, + interrupt: () => undefined, + deliver: routeQueuedDelivery({ + send: (text) => { + throw new Error( + `probe should not use the leftover send path: ${text}`, + ); + }, + deliverSteer: createLiveSteerDeliver({ + enqueue: queue.enqueue, + ingest: async (text, pending) => ({ + text, + attachments: pending, + }), + deliver: (text, pending, settle) => { + const message = userInboundMessage(text, pending); + const messageId = message.headers.messageId; + steerMessageId.current = messageId; + bridge.beginQueuedInputDelivery(messageId); + void deliverAgentMessage({ + getFatalBuildError: () => null, + deliverToLiveAgent: () => agent.deliver(message), + }).then((result) => { + if (result.status === "accepted") { + bridge.recordAcceptedInput(messageId); + } else { + bridge.discardQueuedInputDelivery(messageId); + } + settle?.(result); + }); + }, + captureGeneration: () => () => true, + onFailure: (err) => { + throw err; + }, + }), + parentCycleLive: () => bridgeRef.current?.parentCycleLive === true, + }), + }), + ); + bridgeRef.current = bridge; + try { + reactor.start(); + reactor.deliver(inboundMessage("start the task")); + await tickUntil(() => events.some((e) => e.type === "tool.start")); + + bridge.submit("steer the ship", "steer"); + expect(shell.queuedInput.messageIds).toEqual([]); + expect(shell.queuedInputRow.visible).toBe(false); + + // Release the tool: the boundary drains the steer, the live deliver + // accepts it, and the row paints because the director has not read the + // steer (the second inference is gated, so the steer's message.received + // still waits behind the cycle). + releaseTool(); + await tickUntil(() => shell.queuedInput.messageIds.length === 1, 4000); + const steerId = steerMessageId.current; + expect(steerId).toBeDefined(); + const steerIdValue: string = steerId as string; + expect(shell.queuedInput.messageIds).toEqual([steerIdValue]); + expect(shell.queuedInputRow.visible).toBe(true); + expect( + events.some( + (e) => + e.type === "message.run.started" && + (e.data as { messageId?: string }).messageId === steerIdValue, + ), + ).toBe(false); + + // Release the second inference: the cycle settles, the reactor dequeues + // the steer, message.run.started fires, and the row disappears. + releaseSecondInfer(); + await tickUntil( + () => + events.some( + (e) => + e.type === "message.run.started" && + (e.data as { messageId?: string }).messageId === steerIdValue, + ), + 4000, + ); + await tickUntil(() => shell.queuedInput.messageIds.length === 0, 4000); + expect(shell.queuedInputRow.visible).toBe(false); + } finally { + reactor.abort("admin_kill"); + bridge.dispose(); + shell.dispose(); + } + }, + { width: 80, height: 24 }, + ); +}); diff --git a/src/tui/queued-input.test.ts b/src/tui/queued-input.test.ts new file mode 100644 index 000000000..8d64eaa59 --- /dev/null +++ b/src/tui/queued-input.test.ts @@ -0,0 +1,46 @@ +import { expect, test } from "bun:test"; +import { + NO_QUEUED_INPUT, + acceptQueuedInput, + clearQueuedInput, + composeQueuedInputLine, + consumeQueuedInput, + queuedInputCount, +} from "./queued-input.js"; +import { stringWidth } from "./view/height.js"; + +test("accepted input ledger preserves first-acceptance order and consumes exact identities", () => { + const one = acceptQueuedInput(NO_QUEUED_INPUT, "one"); + const two = acceptQueuedInput(one, "two"); + const three = acceptQueuedInput(two, "three"); + expect(acceptQueuedInput(two, "one")).toBe(two); + expect(three.messageIds).toEqual(["one", "two", "three"]); + const consumed = consumeQueuedInput(three, "two"); + expect(consumed.messageIds).toEqual(["one", "three"]); + expect(consumeQueuedInput(consumed, "unknown")).toBe(consumed); + expect(consumeQueuedInput(consumed, "two")).toBe(consumed); + expect(queuedInputCount(clearQueuedInput(consumed))).toBe(0); + expect(clearQueuedInput(NO_QUEUED_INPUT)).toBe(NO_QUEUED_INPUT); + expect(acceptQueuedInput(NO_QUEUED_INPUT, "")).toBe(NO_QUEUED_INPUT); +}); + +test("queued input formatter remains meaningful and width safe at full and narrow widths", () => { + const state = acceptQueuedInput(acceptQueuedInput(NO_QUEUED_INPUT, "a"), "b"); + for (const width of [0, 1, 5, 10, 24, 80]) { + const text = composeQueuedInputLine(state, width) + .map((part) => part.text) + .join(""); + expect(stringWidth(text)).toBeLessThanOrEqual(width); + } + expect( + composeQueuedInputLine(state, 80) + .map((part) => part.text) + .join(""), + ).toContain("QUEUED INPUT"); + expect( + composeQueuedInputLine(state, 10) + .map((part) => part.text) + .join(""), + ).toContain("QUEUED"); + expect(composeQueuedInputLine(NO_QUEUED_INPUT, 80)).toEqual([]); +}); diff --git a/src/tui/queued-input.ts b/src/tui/queued-input.ts new file mode 100644 index 000000000..de88e4737 --- /dev/null +++ b/src/tui/queued-input.ts @@ -0,0 +1,69 @@ +import { sliceToWidth, stringWidth } from "./view/height.js"; + +export interface QueuedInputState { + readonly messageIds: readonly string[]; +} + +export const NO_QUEUED_INPUT: QueuedInputState = { messageIds: [] }; + +export function acceptQueuedInput( + state: QueuedInputState, + messageId: string, +): QueuedInputState { + if (messageId.length === 0 || state.messageIds.includes(messageId)) + return state; + return { messageIds: [...state.messageIds, messageId] }; +} + +export function consumeQueuedInput( + state: QueuedInputState, + messageId: string, +): QueuedInputState { + const index = state.messageIds.indexOf(messageId); + if (index === -1) return state; + return { + messageIds: [ + ...state.messageIds.slice(0, index), + ...state.messageIds.slice(index + 1), + ], + }; +} + +export function clearQueuedInput(state: QueuedInputState): QueuedInputState { + return state.messageIds.length === 0 ? state : NO_QUEUED_INPUT; +} + +export function queuedInputCount(state: QueuedInputState): number { + return state.messageIds.length; +} + +export type QueuedInputRole = "mark" | "label" | "separator" | "count"; + +export interface QueuedInputPart { + readonly text: string; + readonly role: QueuedInputRole; +} + +/** One composer-adjacent row which retains its meaning even in narrow terminals. */ +export function composeQueuedInputLine( + state: QueuedInputState, + width: number, +): readonly QueuedInputPart[] { + if (width <= 0 || state.messageIds.length === 0) return []; + const parts: readonly QueuedInputPart[] = [ + { text: "◆ ", role: "mark" }, + { text: "QUEUED INPUT", role: "label" }, + { text: " · ", role: "separator" }, + { text: String(state.messageIds.length), role: "count" }, + ]; + const text = parts.map((part) => part.text).join(""); + if (stringWidth(text) <= width) return parts; + const compact = `QUEUED ${state.messageIds.length}`; + if (stringWidth(compact) <= width) { + return [ + { text: "QUEUED ", role: "label" }, + { text: String(state.messageIds.length), role: "count" }, + ]; + } + return [{ text: sliceToWidth("QUEUED INPUT", width), role: "label" }]; +} diff --git a/src/tui/runner/index.ts b/src/tui/runner/index.ts index 135f743d4..cf8789005 100644 --- a/src/tui/runner/index.ts +++ b/src/tui/runner/index.ts @@ -105,6 +105,15 @@ export async function runTUI(initialConfig: Config): Promise { const computeAddProviderChoices = () => addProviderSelectorChoices(providerChoices(), state.config.providers); + state.acceptedInputObserver = (messageId) => { + state.host?.bridge.recordAcceptedInput(messageId); + }; + state.beginAcceptedInputObserver = (messageId) => { + state.host?.bridge.beginQueuedInputDelivery(messageId); + }; + state.discardAcceptedInputObserver = (messageId) => { + state.host?.bridge.discardQueuedInputDelivery(messageId); + }; const host = await mountRunnerHost({ // An unnamed session shows nothing rather than a placeholder. title: state.runTaskTitle, diff --git a/src/tui/runner/state.ts b/src/tui/runner/state.ts index b29373e3c..941769d05 100644 --- a/src/tui/runner/state.ts +++ b/src/tui/runner/state.ts @@ -305,6 +305,12 @@ export interface RunnerState { send?: (message: InboundMessage) => Promise, presentFailureNotice?: boolean, ) => Promise; + /** Bound by the mounted host; receives only accepted operator message ids. */ + acceptedInputObserver?: (messageId: string) => void; + /** Bound by the mounted host before an operator delivery can echo. */ + beginAcceptedInputObserver?: (messageId: string) => void; + /** Bound by the mounted host when an operator delivery does not settle accepted. */ + discardAcceptedInputObserver?: (messageId: string) => void; sendUserPrompt?: ( text: string, pending: readonly PendingImageAttachment[], diff --git a/src/tui/runner/submit.ts b/src/tui/runner/submit.ts index d410b4dcb..401853400 100644 --- a/src/tui/runner/submit.ts +++ b/src/tui/runner/submit.ts @@ -6,7 +6,10 @@ import { getLogger } from "@intx/log"; import type { InboundMessage } from "@intx/types/runtime"; -import { OPERATOR_ORIGINATED_FLAG } from "../../agent/message-provenance.js"; +import { + isOperatorOriginated, + OPERATOR_ORIGINATED_FLAG, +} from "../../agent/message-provenance.js"; import { appendSentMessage } from "../../session/sent-messages.js"; import { getTelemetry } from "../../telemetry/singleton.js"; import { @@ -235,10 +238,14 @@ export const IMAGE_ONLY_PROMPT = "Please inspect the attached image."; * OPERATOR_ORIGINATED_FLAG so director.ts's loop-protection backstop can * tell it from system-originated sends (compaction continuations, * retries, nudges). + * `operatorOriginated` is false for harness-originated text (ask wakes, + * mailbox mail) delivered through the same port: those messages must never + * read as operator input and never enter the queued-input ledger. */ export function userInboundMessage( text: string, attachments: readonly PendingImageAttachment[], + operatorOriginated = true, ): InboundMessage { return { ref: { uid: 1, mailbox: "INBOX" }, @@ -249,7 +256,7 @@ export function userInboundMessage( messageId: `<${crypto.randomUUID()}@local>`, interchangeType: "conversation.message", }, - flags: [OPERATOR_ORIGINATED_FLAG], + flags: operatorOriginated === true ? [OPERATOR_ORIGINATED_FLAG] : [], signatureStatus: "missing", content: text.length > 0 ? text : IMAGE_ONLY_PROMPT, attachments: attachments.map((a) => ({ @@ -370,6 +377,10 @@ export function createSubmitPath( attempt.providerId, ); state.reconnectRecoveryAttempts.set(providerFailure, reconnectAttempt); + const messageId = message.headers.messageId; + const isOperatorMessage = + typeof messageId === "string" && isOperatorOriginated(message.flags); + if (isOperatorMessage) state.beginAcceptedInputObserver?.(messageId); try { await runWhileAgentBusy(state, async () => { const result = await send(message); @@ -386,8 +397,12 @@ export function createSubmitPath( // Settled: nothing is left for the stall watchdog to re-present. services.suspendedApprovalRecovery.clear(); }); + if (isOperatorMessage) { + state.acceptedInputObserver?.(messageId); + } return { status: "accepted" }; } catch (error) { + if (isOperatorMessage) state.discardAcceptedInputObserver?.(messageId); handleSendFailure(error, attempt, providerFailure, presentFailureNotice); if (error instanceof AgentClosedError) { return { @@ -510,7 +525,7 @@ export function createDeliverRouting( imageAttachmentFromPath, pending, ), - send: async (text, pending) => { + send: async (text, pending, _settle, operatorOriginated) => { state.sendAborted = false; const send = state.sendWithAttemptIdentity; if (send === undefined) { @@ -522,7 +537,7 @@ export function createDeliverRouting( } const targetAgent = liveAgent(state); return send( - userInboundMessage(text, pending), + userInboundMessage(text, pending, operatorOriginated), (message) => targetAgent.send(message), false, ); @@ -553,11 +568,25 @@ export function createDeliverRouting( imageAttachmentFromPath, pending, ), - deliver: (text, pending, settle) => { + deliver: (text, pending, settle, operatorOriginated) => { const targetAgent = liveAgent(state); - state.enqueueAgentDeliver?.( - () => targetAgent.deliver(userInboundMessage(text, pending)), - settle, + const enqueueAgentDeliver = state.enqueueAgentDeliver; + if (enqueueAgentDeliver === undefined) return; + const message = userInboundMessage(text, pending, operatorOriginated); + const messageId = message.headers.messageId; + const isOperatorMessage = + typeof messageId === "string" && isOperatorOriginated(message.flags); + if (isOperatorMessage) state.beginAcceptedInputObserver?.(messageId); + enqueueAgentDeliver( + () => targetAgent.deliver(message), + (result) => { + if (result.status === "accepted" && isOperatorMessage) { + state.acceptedInputObserver?.(message.headers.messageId); + } else if (isOperatorMessage) { + state.discardAcceptedInputObserver?.(messageId); + } + settle?.(result); + }, ); }, captureGeneration: services.deliveryGeneration.capture, diff --git a/src/tui/runtime-bridge.ts b/src/tui/runtime-bridge.ts index 9cc19b71f..e7c5fac53 100644 --- a/src/tui/runtime-bridge.ts +++ b/src/tui/runtime-bridge.ts @@ -107,6 +107,14 @@ import { } from "../subagent/fleet-report.js"; import { isPromiseLike } from "../subagent/fleet-dry-drive.js"; import { mailboxMailDriveClaimed } from "../subagent/mailbox-mail-drive.js"; +import { isOperatorOriginated } from "../agent/message-provenance.js"; +import { + NO_QUEUED_INPUT, + acceptQueuedInput, + clearQueuedInput, + consumeQueuedInput, + type QueuedInputState, +} from "./queued-input.js"; /** spawn_agent dispatch tool; its row gets live progress. */ const SPAWN_AGENT_TOOL_NAME = "spawn_agent"; @@ -202,6 +210,12 @@ export interface SessionBridge { /** Drop queued items, echoes, and fleet hold so a rotation cannot drain * into the new reactor. */ clearQueuedDelivery: () => void; + /** Starts exact identity correlation before an operator delivery can echo. */ + beginQueuedInputDelivery: (messageId: string) => void; + /** Records an operator message only after its final delivery result is accepted. */ + recordAcceptedInput: (messageId: string) => void; + /** Drops exact identity correlation when a delivery does not settle accepted. */ + discardQueuedInputDelivery: (messageId: string) => void; /** True only while draining steers at a live parent tool.boundary (or * inference.done with tools still outstanding); other drains send. */ readonly parentCycleLive: boolean; @@ -413,6 +427,16 @@ export interface BridgeBag { /** Prompts already echoed locally; else `message.received` replay shows * each twice. */ pendingEchoes: string[]; + queuedInput: QueuedInputState; + /** Exact in-flight operator deliveries, including receives that precede acceptance. */ + pendingQueuedInputDeliveries: Map; + /** + * Generation captured when each pending delivery began. Bumped by + * `clearQueuedDelivery` / dispose so a stale accept from a cleared session + * is ignored instead of painting a row. + */ + queuedInputBegunAt: Map; + queuedInputGeneration: number; /** Popped for delivery but unsettled: a second boundary cannot redispatch * them; rejection can restore. */ pendingDeliveries: Map; @@ -660,6 +684,62 @@ function consumePendingEchoEvent( return content !== undefined && consumeEcho(bag, content); } +function consumeAcceptedInputEvent( + shell: AppShell, + bag: BridgeBag, + event: { readonly type: string; readonly data?: unknown }, +): void { + if (event.type === "message.received") { + const data = event.data; + if (data === null || typeof data !== "object" || Array.isArray(data)) + return; + const message = (data as { readonly message?: unknown }).message; + if ( + message === null || + typeof message !== "object" || + Array.isArray(message) + ) + return; + const flags = (message as { readonly flags?: unknown }).flags; + if (!Array.isArray(flags) || !isOperatorOriginated(flags)) return; + const headers = (message as { readonly headers?: unknown }).headers; + if ( + headers === null || + typeof headers !== "object" || + Array.isArray(headers) + ) + return; + const messageId = (headers as { readonly messageId?: unknown }).messageId; + if (typeof messageId !== "string") return; + // Receipt only: a mid-run steer is accepted before its receive, a send is + // received before its accept. Mark the pending delivery but never consume + // the ledger here — the read signal is message.run.started. + const pendingDelivery = bag.pendingQueuedInputDeliveries.get(messageId); + if (pendingDelivery !== undefined) { + bag.pendingQueuedInputDeliveries.set(messageId, "received"); + } + return; + } + if (event.type !== "message.run.started") return; + const data = event.data; + if (data === null || typeof data !== "object" || Array.isArray(data)) return; + const messageId = (data as { readonly messageId?: unknown }).messageId; + if (typeof messageId !== "string") return; + // The reactor dequeued the message and opened its run: the director is + // about to read it. Mark the pending delivery read so a later accept (send + // path) never paints, then consume the ledger entry (a no-op for ids that + // never entered it — no operator-flag check is needed here). + const pendingDelivery = bag.pendingQueuedInputDeliveries.get(messageId); + if (pendingDelivery !== undefined) { + bag.pendingQueuedInputDeliveries.set(messageId, "read"); + } + const next = consumeQueuedInput(bag.queuedInput, messageId); + if (next === bag.queuedInput) return; + bag.queuedInput = next; + shell.queuedInput = next; + paintChrome(shell); +} + function openRowContent( kind: OpenRowKind, text: string, @@ -1372,6 +1452,10 @@ export function attachSessionBridge( port: resolvePort(handlers), openRow: null, pendingEchoes: [], + queuedInput: NO_QUEUED_INPUT, + pendingQueuedInputDeliveries: new Map(), + queuedInputBegunAt: new Map(), + queuedInputGeneration: 0, pendingDeliveries: new Map(), pendingPromptRecoveries: [], mapCtx: createStreamMapContext(), @@ -1634,6 +1718,7 @@ export function attachSessionBridge( } // Reactor-shaped types always map first (avoids tool.done name collision). if (PRODUCTION_REACTOR_TYPES.has(event.type)) { + consumeAcceptedInputEvent(shell, bag, event); if (consumePendingEchoEvent(bag, event)) { // The echo skips the mapper so it cannot expire a recovery handoff, // but still starts a new turn. @@ -1786,12 +1871,14 @@ export function attachSessionBridge( bag.turn = turnStateOnSubmit(bag.turn, now()); paintChrome(shell); paintPhase(); - // Harness turns are not composer input: /feedback must never consume them. + // Harness turns are not composer input: /feedback must never consume them, + // and they must never enter the queued-input ledger as operator input. bag.port.deliver({ id: crypto.randomUUID(), text, kind: "queue", enqueuedAt: now(), + operatorOriginated: false, }); }; const flushMailboxMail = (): boolean => { @@ -1924,6 +2011,14 @@ export function attachSessionBridge( shell.session = createSessionQueue("idle"); recordLastSent(null); bag.pendingEchoes.length = 0; + bag.queuedInput = clearQueuedInput(bag.queuedInput); + bag.pendingQueuedInputDeliveries.clear(); + // Keep `queuedInputBegunAt` so a stale accept from a delivery that began + // before this clear is still recognized as stale by the generation check + // in `recordAcceptedInput` (Nit 1: a late accept after /clear|/new must + // not paint a row). Entries are removed by accept/discard or dispose. + bag.queuedInputGeneration += 1; + shell.queuedInput = bag.queuedInput; bag.pendingDeliveries.clear(); bag.pendingPromptRecoveries.length = 0; bag.liveFleet = 0; @@ -2110,6 +2205,33 @@ export function attachSessionBridge( submit, interrupt: doInterrupt, clearQueuedDelivery, + beginQueuedInputDelivery: (messageId) => { + if (bag.disposed || messageId.length === 0) return; + bag.pendingQueuedInputDeliveries.set(messageId, "awaiting"); + bag.queuedInputBegunAt.set(messageId, bag.queuedInputGeneration); + }, + recordAcceptedInput: (messageId) => { + if (bag.disposed) return; + const begunAt = bag.queuedInputBegunAt.get(messageId); + bag.queuedInputBegunAt.delete(messageId); + if (begunAt !== undefined && begunAt !== bag.queuedInputGeneration) { + // Stale accept from a cleared session (/clear or /new): inert. + bag.pendingQueuedInputDeliveries.delete(messageId); + return; + } + const pendingDelivery = bag.pendingQueuedInputDeliveries.get(messageId); + bag.pendingQueuedInputDeliveries.delete(messageId); + if (pendingDelivery === "read") return; + const next = acceptQueuedInput(bag.queuedInput, messageId); + if (next === bag.queuedInput) return; + bag.queuedInput = next; + shell.queuedInput = next; + paintChrome(shell); + }, + discardQueuedInputDelivery: (messageId) => { + bag.pendingQueuedInputDeliveries.delete(messageId); + bag.queuedInputBegunAt.delete(messageId); + }, setSuspendedApprovalRecovery: (recovery) => { bag.suspendedApprovalRecovery = recovery; }, @@ -2199,6 +2321,11 @@ export function attachSessionBridge( bag.disposed = true; recordLastSent(null); bag.pendingDeliveries.clear(); + bag.queuedInput = clearQueuedInput(bag.queuedInput); + bag.pendingQueuedInputDeliveries.clear(); + bag.queuedInputBegunAt.clear(); + bag.queuedInputGeneration += 1; + shell.queuedInput = bag.queuedInput; bag.pendingPromptRecoveries.length = 0; bag.pendingAskWake.clear(); clearWorkerWait(shell); diff --git a/src/tui/shell/chrome.ts b/src/tui/shell/chrome.ts index 666d695da..743a9f1c5 100644 --- a/src/tui/shell/chrome.ts +++ b/src/tui/shell/chrome.ts @@ -106,6 +106,10 @@ import { type OperatorInputRequiredRole, } from "../operator-input-required.js"; import type { OperatorGateEvent } from "../gate-events.js"; +import { + composeQueuedInputLine, + type QueuedInputPart, +} from "../queued-input.js"; import { type AppShell, @@ -279,6 +283,14 @@ function operatorInputRequiredChunks( ); } +function queuedInputChunks(parts: readonly QueuedInputPart[]): TextChunk[] { + return parts.map((part) => + fgChunk( + part.role === "label" || part.role === "mark" ? UI.action : UI.textDim, + )(part.text), + ); +} + /** * Key over every input a chrome paint reads; a miss leaves stale chrome. */ @@ -287,12 +299,14 @@ function chromeComposeKey( notice: string, workerWait: string, inputRequired: string, + queuedInput: string, ): string { const meter = shell.costContext; return [ notice, workerWait, inputRequired, + queuedInput, pendingColumnRows(shell.session.items) .map((row) => `${row.tag ?? ""}:${row.text}`) .join("\u0001"), @@ -345,11 +359,16 @@ export function paintChrome( shell.operatorInputRequired, shell.layout.contentWidth, ); + const queuedInput = composeQueuedInputLine( + shell.queuedInput, + shell.layout.contentWidth, + ); const key = chromeComposeKey( shell, notice, workerWait.map((part) => part.text).join(""), inputRequired.map((part) => part.text).join(""), + queuedInput.map((part) => part.text).join(""), ); if (!opts?.force && paintedChromeKey.get(shell) === key) return; paintedChromeKey.set(shell, key); @@ -368,6 +387,8 @@ export function paintChrome( operatorInputRequiredChunks(inputRequired), ); syncOperatorInputRequiredRow(shell, inputRequired.length > 0); + shell.queuedInputRow.content = new StyledText(queuedInputChunks(queuedInput)); + syncQueuedInputRow(shell, queuedInput.length > 0); } /** @@ -389,6 +410,13 @@ function syncOperatorInputRequiredRow(shell: AppShell, wanted: boolean): void { relayout(shell, { visibility: { ...bag.visibility, inputRequired: wanted } }); } +function syncQueuedInputRow(shell: AppShell, wanted: boolean): void { + const bag = shellInternals(shell); + if (bag === undefined) return; + if ((bag.visibility.queuedInput ?? false) === wanted) return; + relayout(shell, { visibility: { ...bag.visibility, queuedInput: wanted } }); +} + /** Rebuild pendingBox rows only when the painted content moved; unchanged * queues cost a signature compare, not a rebuild. */ function syncPendingRows(shell: AppShell): void { @@ -943,6 +971,9 @@ export function applyLayout(shell: AppShell, layout: GeometryLayout): void { const inputRequiredH = Math.max(0, h.input_required); shell.inputRequiredRow.height = inputRequiredH > 0 ? inputRequiredH : 1; shell.inputRequiredRow.visible = inputRequiredH > 0; + const queuedInputH = Math.max(0, h.queued_input); + shell.queuedInputRow.height = queuedInputH > 0 ? queuedInputH : 1; + shell.queuedInputRow.visible = queuedInputH > 0; const promptH = Math.max(0, h.prompt); shell.promptBox.height = promptH > 0 ? promptH : 1; diff --git a/src/tui/shell/index.ts b/src/tui/shell/index.ts index 341f431dd..548a3ed2b 100644 --- a/src/tui/shell/index.ts +++ b/src/tui/shell/index.ts @@ -75,6 +75,7 @@ import { import { armMarkdownLinks } from "../url-links.js"; import { NO_WORKER_WAIT } from "../worker-wait.js"; import { NO_OPERATOR_INPUT_REQUIRED } from "../operator-input-required.js"; +import { NO_QUEUED_INPUT } from "../queued-input.js"; import { createShellKeyHandlers, routePromptWheelToTranscript, @@ -269,6 +270,13 @@ export function createAppShell( fg: UI.textDim, visible: layout.heights.worker_wait > 0, }); + const queuedInputRow = new TextRenderable(ctx, { + id: "shell-queued-input", + height: Math.max(1, layout.heights.queued_input), + content: "", + fg: UI.textDim, + visible: layout.heights.queued_input > 0, + }); const inputRequiredRow = new TextRenderable(ctx, { id: "shell-input-required", @@ -339,6 +347,7 @@ export function createAppShell( root.add(pendingBox); root.add(workerWaitRow); root.add(inputRequiredRow); + root.add(queuedInputRow); root.add(promptBox); root.add(landingBelow); root.add(bottomPad); @@ -445,6 +454,7 @@ export function createAppShell( pendingBox, workerWaitRow, inputRequiredRow, + queuedInputRow, layout, focus: createFocusState(), session, @@ -471,6 +481,7 @@ export function createAppShell( pluginNeedsAttention: false, workerWait: NO_WORKER_WAIT, operatorInputRequired: NO_OPERATOR_INPUT_REQUIRED, + queuedInput: NO_QUEUED_INPUT, lockupNowMs: 0, inFlightTool: null, lockupAnimating: false, diff --git a/src/tui/shell/internals.ts b/src/tui/shell/internals.ts index 7399a4f47..ea352a6ee 100644 --- a/src/tui/shell/internals.ts +++ b/src/tui/shell/internals.ts @@ -36,6 +36,7 @@ import { createOverlayView } from "../overlay-view.js"; import { type KillRing } from "../prompt-kill-ring.js"; import { type WorkerWaitState } from "../worker-wait.js"; import { type OperatorInputRequiredState } from "../operator-input-required.js"; +import { type QueuedInputState } from "../queued-input.js"; export const shellExitHandlers = new WeakMap void>(); @@ -373,6 +374,8 @@ export interface AppShell { readonly workerWaitRow: TextRenderable; /** INPUT REQUIRED strip for live marked primary ask_operator gates. */ readonly inputRequiredRow: TextRenderable; + /** Accepted operator inputs awaiting their correlated runtime receive event. */ + readonly queuedInputRow: TextRenderable; /** Latest geometry resolution (updated on resize / relayout). */ layout: GeometryLayout; /** Focus tree + scroll lease (updated by shell helpers). */ @@ -431,6 +434,7 @@ export interface AppShell { workerWait: WorkerWaitState; /** Display-only state for outstanding primary ask_operator gate ids. */ operatorInputRequired: OperatorInputRequiredState; + queuedInput: QueuedInputState; /** Clock and motion state for the bottom-left status slot; the bridge pushes * it off its monitor tick. */ lockupNowMs: number; diff --git a/src/tui/shell/layout.ts b/src/tui/shell/layout.ts index 6a9fc1456..643c33d6a 100644 --- a/src/tui/shell/layout.ts +++ b/src/tui/shell/layout.ts @@ -57,6 +57,7 @@ export function defaultVisibility(visibility?: ZoneVisibility): ZoneVisibility { notice: false, progress: false, progressDivider: false, + queuedInput: false, // Explicit 0 rather than undefined: task and agents are row counts, and // setChromeZones compares by ===, so undefined forces one needless // relayout the first time either is compared. diff --git a/src/tui/stream-event-map.test.ts b/src/tui/stream-event-map.test.ts index 86a7184f0..3370796fe 100644 --- a/src/tui/stream-event-map.test.ts +++ b/src/tui/stream-event-map.test.ts @@ -39,6 +39,15 @@ describe("mapProductionEvent", () => { ).toEqual([{ type: "system", text: "hello" }]); }); + test("message.run.started reaches the bridge without painting a row", () => { + expect( + mapProductionEvent({ + type: "message.run.started", + data: { messageId: "" }, + }), + ).toEqual([]); + }); + test("background shell exit message.received → system", () => { const message = buildShellBackgroundMessage({ id: "sh_1", diff --git a/src/tui/stream-event-map.ts b/src/tui/stream-event-map.ts index 23eaae7d9..d31d4e906 100644 --- a/src/tui/stream-event-map.ts +++ b/src/tui/stream-event-map.ts @@ -20,7 +20,10 @@ import type { PendingAskWake } from "../subagent/fleet-report.js"; /** Canonical inbound events the bridge understands (fixtures + mapped reactor). */ export type BridgeInboundEvent = - | { readonly type: "user"; readonly text: string } + | { + readonly type: "user"; + readonly text: string; + } | { readonly type: "assistant"; readonly text: string } | { readonly type: "assistant.delta"; readonly text: string } | { readonly type: "thinking.delta"; readonly text: string } @@ -73,6 +76,7 @@ export interface ReactorLikeEvent { * production mapper (avoids collisions like tool.done vs tool_result). */ export const PRODUCTION_REACTOR_TYPES: ReadonlySet = new Set([ "message.received", + "message.run.started", "inference.start", "inference.done", "inference.retry", @@ -349,7 +353,12 @@ function mapEvent( ) { return disarmed; } - return [...disarmed, { type: operator ? "user" : "system", text: full }]; + return [ + ...disarmed, + operator + ? { type: "user", text: full } + : { type: "system", text: full }, + ]; } case "inference.start": { From 6dcb8b928106350230264a62fab17c287437d06e Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Mon, 12 Oct 2026 19:13:54 -0700 Subject: [PATCH 2/2] fix(tui): clear queued input when the steer is aborted --- src/tui/queued-input-chrome.test.ts | 68 +++++++++++++++++++++++++++++ src/tui/runtime-bridge.ts | 27 +++++++----- 2 files changed, 85 insertions(+), 10 deletions(-) diff --git a/src/tui/queued-input-chrome.test.ts b/src/tui/queued-input-chrome.test.ts index 9480ef692..12b3830d3 100644 --- a/src/tui/queued-input-chrome.test.ts +++ b/src/tui/queued-input-chrome.test.ts @@ -17,9 +17,11 @@ import { createAppShell } from "./shell/index.js"; import type { PendingAskWake } from "../subagent/fleet-report.js"; import { createLiveSessionPort } from "./live-session-port.js"; import { + badgeCount, createLiveSteerDeliver, createSessionOperationQueue, deliverAgentMessage, + pause, routeQueuedDelivery, } from "./delivery-queue.js"; import { userInboundMessage } from "./runner/submit.js"; @@ -338,6 +340,72 @@ test("late accept after /clear or /new is inert (generation check)", async () => ); }); +test("interrupt before message.run.started clears the queued-input strip and a late accept stays hidden", async () => { + await withTestRenderer( + async (h) => { + const delivered: string[] = []; + const shell = createAppShell(h.renderer, { + terminal: { columns: 80, rows: 24 }, + wireKeys: false, + run: "busy", + }); + const bridge = attachSessionBridge( + shell, + createLiveSessionPort({ + send: (text) => { + delivered.push(text); + }, + interrupt: () => undefined, + deliver: (text) => { + delivered.push(text); + }, + }), + ); + try { + bridge.submit("held for later", "steer"); + shell.session = pause(shell.session); + bridge.beginQueuedInputDelivery("aborted-steer"); + bridge.recordAcceptedInput("aborted-steer"); + expect(shell.queuedInput.messageIds).toEqual(["aborted-steer"]); + expect(shell.queuedInputRow.visible).toBe(true); + bridge.beginQueuedInputDelivery("late-only"); + + bridge.interrupt(); + + expect(shell.queuedInput.messageIds).toEqual([]); + expect(shell.queuedInputRow.visible).toBe(false); + expect(shell.session.items.map((item) => item.text)).toEqual([ + "held for later", + ]); + expect(delivered).toEqual([]); + + bridge.recordAcceptedInput("aborted-steer"); + bridge.recordAcceptedInput("late-only"); + expect(shell.queuedInput.messageIds).toEqual([]); + expect(shell.queuedInputRow.visible).toBe(false); + expect(delivered).toEqual([]); + + bridge.beginQueuedInputDelivery("next-steer"); + bridge.recordAcceptedInput("next-steer"); + expect(shell.queuedInput.messageIds).toEqual(["next-steer"]); + bridge.handle(started("next-steer")); + expect(shell.queuedInput.messageIds).toEqual([]); + expect(shell.queuedInputRow.visible).toBe(false); + + bridge.submit("fresh", "immediate"); + bridge.handle({ type: "run", state: "idle" }); + expect(delivered).toEqual(["fresh", "held for later"]); + expect(badgeCount(shell.session)).toBe(0); + expect(shell.queuedInput.messageIds).toEqual([]); + } finally { + bridge.dispose(); + shell.dispose(); + } + }, + { width: 80, height: 24 }, + ); +}); + test("delivered ask wake never enters the queued-input ledger", async () => { await withTestRenderer( async (h) => { diff --git a/src/tui/runtime-bridge.ts b/src/tui/runtime-bridge.ts index e7c5fac53..b8fa186e8 100644 --- a/src/tui/runtime-bridge.ts +++ b/src/tui/runtime-bridge.ts @@ -432,8 +432,8 @@ export interface BridgeBag { pendingQueuedInputDeliveries: Map; /** * Generation captured when each pending delivery began. Bumped by - * `clearQueuedDelivery` / dispose so a stale accept from a cleared session - * is ignored instead of painting a row. + * `clearQueuedDelivery`, interrupt, and dispose so a stale accept is + * ignored instead of painting a row. */ queuedInputBegunAt: Map; queuedInputGeneration: number; @@ -1944,6 +1944,16 @@ export function attachSessionBridge( } }; + /** Drop accepted-but-unread input. Keep `queuedInputBegunAt` so a stale + * accept from a delivery that began before this clear is still recognized + * as stale by the generation check in `recordAcceptedInput`. */ + const clearQueuedInputLedger = (): void => { + bag.queuedInput = clearQueuedInput(bag.queuedInput); + bag.pendingQueuedInputDeliveries.clear(); + bag.queuedInputGeneration += 1; + shell.queuedInput = bag.queuedInput; + }; + /** Shared abort for Ctrl+C, monitor tick, and the fleet-poll stall bound: * interrupt the hung inference, then flush occupancy. */ const abortInFlightAndHandoff = (): void => { @@ -1953,6 +1963,10 @@ export function attachSessionBridge( // inference.start cannot roll back the stop. bag.mapCtx.errorRollbackArmed = false; bag.attemptRow = null; + for (const messageId of bag.queuedInput.messageIds) { + bag.queuedInputBegunAt.set(messageId, bag.queuedInputGeneration); + } + clearQueuedInputLedger(); applyShellInterrupt(shell); bag.port.interrupt(); // The stop settles the turn without necessarily producing an idle event to @@ -2011,14 +2025,7 @@ export function attachSessionBridge( shell.session = createSessionQueue("idle"); recordLastSent(null); bag.pendingEchoes.length = 0; - bag.queuedInput = clearQueuedInput(bag.queuedInput); - bag.pendingQueuedInputDeliveries.clear(); - // Keep `queuedInputBegunAt` so a stale accept from a delivery that began - // before this clear is still recognized as stale by the generation check - // in `recordAcceptedInput` (Nit 1: a late accept after /clear|/new must - // not paint a row). Entries are removed by accept/discard or dispose. - bag.queuedInputGeneration += 1; - shell.queuedInput = bag.queuedInput; + clearQueuedInputLedger(); bag.pendingDeliveries.clear(); bag.pendingPromptRecoveries.length = 0; bag.liveFleet = 0;