Merge pull request #744 from corbitsdev/cl-7268-route-tui-boundary-steers-through-the-active-reactor

Route mid-run Enter through the live reactor

Author: TheGreatAxios

docs/ARCHITECTURE.md

@@ -81,7 +81,14 @@ In TUI chat mode there is no completion gate — the session stays open across t
 - Wires `ask_operator` to an operator-gate event resolved by a modal
 - Mounts the OpenTUI host via `mountRunnerHost` (`src/tui/runner-host.ts`), which mounts `mountProductHost` (`src/tui/product-host.ts`) over the shell (`src/tui/shell.ts`)
 - Bridges reactor events to the OpenTUI host via a plain `EventEmitter`
-- **Mid-run injection** — When a message arrives while the agent is running, it is queued in an `InjectionQueue`. On the next `inference.done` event (turn boundary), the queue is drained: each queued message is delivered via `agentProxy.deliver()` and a `"mid-run.delivered"` emitter event is fired so the badge count in the App updates. The queue is cleared on session rotation (`/clear`).
+- **Mid-run injection** — Shell `session-queue` items drain at the parent
+  `tool.boundary` through `SessionPort.deliver`. Production `routeQueuedDelivery`
+  live-injects in-flight parent-boundary steers via `agentProxy.deliver`
+  (`Agent.deliver`) into the live reactor. Idle leftover, idle-with-fleet, and
+  post-interrupt steers, plus follow-ups (`kind === "queue"`), use the existing
+  send path. `/clear` and `/new` bump a
+  delivery generation and call `SessionBridge.clearQueuedDelivery()` so queued
+  input from the previous session cannot enter the new one.
 - **Session rotation** — Uses a serial session-operation queue (`createSessionOperationQueue`, not a boolean flag) so rotation, compaction continuation, and `agentProxy.deliver` never race a concurrent rebuild. Each operation chains onto the tail, ensuring in-flight work completes before the agent is torn down.
 
 ### Exec Runner (`src/exec/runner.ts`)

docs/TUI.md

@@ -468,17 +468,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 <tool>` (e.g. `waiting on run_shell`).
@@ -491,7 +492,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.
 

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: [],

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,30 +65,20 @@ 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" },
     ]);
   });
 
   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]]);

src/tui/live-session-port.ts

@@ -23,15 +23,8 @@ export interface LiveSessionPortDeps {
   ) => SubmitClassification;
   /** Hard interrupt current run (runner close/rebuild). */
   interrupt: () => void;
-  /**
-   * Optional: drained queue/steer item at tool boundary (or idle).
-   * Defaults to `send(text)` for both kinds — v1 runner shares send.
-   */
-  deliver?: (
-    text: string,
-    kind: QueueKind,
-    attachments?: readonly PendingImageAttachment[],
-  ) => void;
+  /** Drained queue/steer item. Kind routing (live inject vs send) is the host's. */
+  deliver: (text: string, kind: QueueKind, attachments?: readonly PendingImageAttachment[]) => void;
 }
 
 /**
@@ -57,11 +50,7 @@ export function createLiveSessionPort(deps: LiveSessionPortDeps): SessionPort {
       deps.interrupt();
     },
     deliver: (item: QueueItem): void => {
-      if (deps.deliver) {
-        deps.deliver(item.text, item.kind, item.attachments);
-        return;
-      }
-      deps.send(item.text, item.attachments);
+      deps.deliver(item.text, item.kind, item.attachments);
     },
   };
 }

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<ProductHostConfig["deliver"]>;
+  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");

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<Produ
   const port = createLiveSessionPort({
     send: config.send,
     interrupt: config.interrupt,
+    deliver: config.deliver,
     ...(config.classifySubmit !== undefined ? { classifySubmit: config.classifySubmit } : {}),
-    ...(config.deliver !== undefined ? { deliver: config.deliver } : {}),
   });
   // Empty options accept the defaults (real clock, 250 ms tick, 15 min stall)
   // while still opting this host into the quota-retry / stall timers.
@@ -508,6 +508,7 @@ export async function mountProductHost(config: ProductHostConfig): Promise<Produ
   function onSessionClear(): void {
     if (disposed) return;
     clearTranscript(shell);
+    bridge.clearQueuedDelivery();
   }
 
   let currentModels = config.models ?? [];

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";

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<AttachImageResult>,
+  pending: readonly PendingImageAttachment[] = [],
+): Promise<PathMentionIngestion> {
+  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;