Skip to content

Commit 7a2bebc

Browse files
committed
Route mid-run Enter through the live reactor
Boundary delivery used to fall back to ordinary send when production omitted the transport, so a queued mid-run Enter started a second turn. Require the transport, steer via Agent.deliver, and drop queued input on session rotation.
1 parent 6b68b94 commit 7a2bebc

15 files changed

Lines changed: 341 additions & 69 deletions

‎docs/ARCHITECTURE.md‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -81,7 +81,12 @@ In TUI chat mode there is no completion gate — the session stays open across t
8181
- Wires `ask_operator` to an operator-gate event resolved by a modal
8282
- 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`)
8383
- Bridges reactor events to the OpenTUI host via a plain `EventEmitter`
84-
- **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`).
84+
- **Mid-run injection** — Shell `session-queue` items drain at the parent
85+
`tool.boundary` through `SessionPort.deliver`. Production `routeQueuedDelivery`
86+
sends steers via `agentProxy.deliver` (`Agent.deliver`) into the live reactor
87+
and idle follow-ups via the existing send path. `/clear` and `/new` bump a
88+
delivery generation and call `SessionBridge.clearQueuedDelivery()` so queued
89+
input from the previous session cannot enter the new one.
8590
- **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.
8691

8792
### Exec Runner (`src/exec/runner.ts`)

‎docs/TUI.md‎

Lines changed: 7 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -457,17 +457,18 @@ the chord to point an operator at when Shift+Enter doesn't respond.
457457
Two mid-run gestures, two delivery times (CL-6290):
458458

459459
- **Enter, mid-run** — soft steer: enqueues kind `"steer"` and delivers at the
460-
next **parent** `tool.boundary` (the parent tool finishing, not a child). A
460+
next **parent** `tool.boundary` (the parent tool finishing, not a child) via
461+
`Agent.deliver` into the live reactor, not a new `send`. A
461462
long parent `run_shell` or an awaiting `task()` is parent-busy and holds
462463
steers. The transcript row says `[will steer next]` while pending and
463464
`[steering]` once delivered (`submitPrompt`, `drainSteersAtBoundary` in
464465
`runtime-bridge.ts`).
465466
- **Alt+Enter, mid-run** — follow-up: enqueues kind `"queue"` and delivers
466-
only on **session-idle** (parent-idle and no live fleet lanes). Does not
467-
interrupt or reinject. The transcript row says `[will follow up]` while
468-
pending and `[following up]` once delivered. Idle, or with an empty prompt,
469-
Alt+Enter does nothing — there is nothing to wait for. (Internal `"reinject"`
470-
remains in the submit API for tests; no product chord wires it.)
467+
only on **session-idle** (parent-idle and no live fleet lanes) as a `send`.
468+
Does not interrupt or reinject. The transcript row says `[will follow up]`
469+
while pending and `[following up]` once delivered. Idle, or with an empty
470+
prompt, Alt+Enter does nothing — there is nothing to wait for. (Internal
471+
`"reinject"` remains in the submit API for tests; no product chord wires it.)
471472

472473
When `steer > 0` and a parent tool has been in flight ≥ `STEER_WAIT_NOTICE_MS`
473474
(3s), the notice row adds `waiting on <tool>` (e.g. `waiting on run_shell`).

‎src/tui/keybindings.test.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -714,6 +714,7 @@ describe("the runner host does not shadow the prompt bindings the catalog claims
714714
eventEmitter: new EventEmitter(),
715715
send: () => {},
716716
interrupt: () => {},
717+
deliver: () => {},
717718
providers: {},
718719
onModelSelect: () => {},
719720
commands: [],

‎src/tui/live-session-port.test.ts‎

Lines changed: 8 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@ type Call =
88
| { op: "interrupt" }
99
| { op: "deliver"; text: string; kind: QueueKind };
1010

11-
function fakeDeps(opts?: { withDeliver?: boolean }) {
11+
function fakeDeps() {
1212
const calls: Call[] = [];
1313
const deps = {
1414
send: (text: string) => {
@@ -17,13 +17,9 @@ function fakeDeps(opts?: { withDeliver?: boolean }) {
1717
interrupt: () => {
1818
calls.push({ op: "interrupt" });
1919
},
20-
...(opts?.withDeliver
21-
? {
22-
deliver: (text: string, kind: QueueKind) => {
23-
calls.push({ op: "deliver", text, kind });
24-
},
25-
}
26-
: {}),
20+
deliver: (text: string, kind: QueueKind) => {
21+
calls.push({ op: "deliver", text, kind });
22+
},
2723
};
2824
return { calls, deps };
2925
}
@@ -69,30 +65,20 @@ describe("createLiveSessionPort", () => {
6965
expect(calls).toEqual([{ op: "interrupt" }]);
7066
});
7167

72-
test("deliver without deps.deliver falls back to send", () => {
68+
test("deliver never calls send for steer or queue", () => {
7369
const { calls, deps } = fakeDeps();
7470
const port = createLiveSessionPort(deps);
7571
port.deliver(item("queued msg", "queue"));
7672
port.deliver(item("steer msg", "steer", "q2"));
77-
expect(calls).toEqual([
78-
{ op: "send", text: "queued msg" },
79-
{ op: "send", text: "steer msg" },
80-
]);
81-
});
82-
83-
test("deliver with deps.deliver passes text and kind", () => {
84-
const { calls, deps } = fakeDeps({ withDeliver: true });
85-
const port = createLiveSessionPort(deps);
86-
port.deliver(item("queued msg", "queue"));
87-
port.deliver(item("steer msg", "steer", "q2"));
73+
expect(calls.some((c) => c.op === "send")).toBe(false);
8874
expect(calls).toEqual([
8975
{ op: "deliver", text: "queued msg", kind: "queue" },
9076
{ op: "deliver", text: "steer msg", kind: "steer" },
9177
]);
9278
});
9379

9480
test("full wiring: immediate → enqueue → deliver → interrupt", () => {
95-
const { calls, deps } = fakeDeps({ withDeliver: true });
81+
const { calls, deps } = fakeDeps();
9682
const port = createLiveSessionPort(deps);
9783

9884
port.sendImmediate("start");
@@ -122,6 +108,7 @@ describe("attachment passthrough", () => {
122108
const port = createLiveSessionPort({
123109
send: (_text, attachments) => seen.push(attachments),
124110
interrupt: () => {},
111+
deliver: () => {},
125112
});
126113
port.sendImmediate("look", [image]);
127114
expect(seen).toEqual([[image]]);

‎src/tui/live-session-port.ts‎

Lines changed: 3 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -23,15 +23,8 @@ export interface LiveSessionPortDeps {
2323
) => SubmitClassification;
2424
/** Hard interrupt current run (runner close/rebuild). */
2525
interrupt: () => void;
26-
/**
27-
* Optional: drained queue/steer item at tool boundary (or idle).
28-
* Defaults to `send(text)` for both kinds — v1 runner shares send.
29-
*/
30-
deliver?: (
31-
text: string,
32-
kind: QueueKind,
33-
attachments?: readonly PendingImageAttachment[],
34-
) => void;
26+
/** Drained queue/steer item at tool boundary (or idle). Always forwarded. */
27+
deliver: (text: string, kind: QueueKind, attachments?: readonly PendingImageAttachment[]) => void;
3528
}
3629

3730
/**
@@ -57,11 +50,7 @@ export function createLiveSessionPort(deps: LiveSessionPortDeps): SessionPort {
5750
deps.interrupt();
5851
},
5952
deliver: (item: QueueItem): void => {
60-
if (deps.deliver) {
61-
deps.deliver(item.text, item.kind, item.attachments);
62-
return;
63-
}
64-
deps.send(item.text, item.attachments);
53+
deps.deliver(item.text, item.kind, item.attachments);
6554
},
6655
};
6756
}

‎src/tui/product-host.test.ts‎

Lines changed: 29 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -24,15 +24,18 @@ import { buildModelsFirstCatalog, modelOptionId } from "./model-catalog.js";
2424

2525
function makeFakeSessionPort(): {
2626
readonly sends: string[];
27+
readonly delivers: string[];
2728
readonly interrupts: number;
2829
readonly send: ProductHostConfig["send"];
2930
readonly interrupt: ProductHostConfig["interrupt"];
30-
readonly deliver: NonNullable<ProductHostConfig["deliver"]>;
31+
readonly deliver: ProductHostConfig["deliver"];
3132
} {
3233
const sends: string[] = [];
34+
const delivers: string[] = [];
3335
let interrupts = 0;
3436
return {
3537
sends,
38+
delivers,
3639
get interrupts() {
3740
return interrupts;
3841
},
@@ -43,7 +46,7 @@ function makeFakeSessionPort(): {
4346
interrupts += 1;
4447
},
4548
deliver: (text) => {
46-
sends.push(text);
49+
delivers.push(text);
4750
},
4851
};
4952
}
@@ -224,6 +227,25 @@ describe("mountProductHost", () => {
224227
}
225228
});
226229

230+
test("session.clear drops queued steers and idles the run (CL-7268)", async () => {
231+
const { host, emitter } = await mountHeadless();
232+
try {
233+
host.bridge.handle({ type: "run", state: "busy" });
234+
host.bridge.submit("old steer", "steer");
235+
expect(host.shell.session.run).toBe("busy");
236+
expect(host.shell.session.items.length).toBe(1);
237+
238+
emitter.emit("event", { type: "user", text: "old prompt" });
239+
emitter.emit("session.clear");
240+
241+
expect(host.shell.streamLog).toEqual([]);
242+
expect(host.shell.session.items).toEqual([]);
243+
expect(host.shell.session.run).toBe("idle");
244+
} finally {
245+
host.dispose();
246+
}
247+
});
248+
227249
test("permission.gate opens the overlay and resolves through the emitter's resolve callback", async () => {
228250
const { host, emitter } = await mountHeadless();
229251
try {
@@ -390,6 +412,7 @@ describe("flat type-to-filter model picker", () => {
390412
eventEmitter: new EventEmitter(),
391413
send: port.send,
392414
interrupt: port.interrupt,
415+
deliver: port.deliver,
393416
createRenderer: async () => harness.renderer,
394417
models: catalog,
395418
onModelSelect: (id) => selected.push(id),
@@ -477,6 +500,7 @@ describe("flat type-to-filter model picker", () => {
477500
eventEmitter: new EventEmitter(),
478501
send: port.send,
479502
interrupt: port.interrupt,
503+
deliver: port.deliver,
480504
createRenderer: async () => harness.renderer,
481505
models: catalog,
482506
activeModelId: () => modelOptionId("xai/thegreataxios", "grok-4.5"),
@@ -509,6 +533,7 @@ describe("flat type-to-filter model picker", () => {
509533
eventEmitter: new EventEmitter(),
510534
send: port.send,
511535
interrupt: port.interrupt,
536+
deliver: port.deliver,
512537
createRenderer: async () => harness.renderer,
513538
models: catalog,
514539
activeModelId: () => modelOptionId("codex/abk-labs", "gpt-5.5"),
@@ -536,6 +561,7 @@ describe("flat type-to-filter model picker", () => {
536561
eventEmitter: new EventEmitter(),
537562
send: port.send,
538563
interrupt: port.interrupt,
564+
deliver: port.deliver,
539565
createRenderer: async () => harness.renderer,
540566
models: catalog,
541567
onModelSelect: () => {},
@@ -865,6 +891,7 @@ describe("mount failure", () => {
865891
eventEmitter: emitter,
866892
send: port.send,
867893
interrupt: port.interrupt,
894+
deliver: port.deliver,
868895
createRenderer: async () => harness.renderer,
869896
}),
870897
).rejects.toThrow("gate wiring failed");

‎src/tui/product-host.ts‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -119,7 +119,7 @@ export interface ProductHostConfig {
119119
*/
120120
readonly classifySubmit?: ProductHostClassifySubmit;
121121
readonly interrupt: ProductHostInterrupt;
122-
readonly deliver?: ProductHostDeliver;
122+
readonly deliver: ProductHostDeliver;
123123
/** Model/provider rows for the picker (id applied on select). */
124124
readonly models?: readonly ProductHostModelOption[];
125125
/**
@@ -319,8 +319,8 @@ export async function mountProductHost(config: ProductHostConfig): Promise<Produ
319319
const port = createLiveSessionPort({
320320
send: config.send,
321321
interrupt: config.interrupt,
322+
deliver: config.deliver,
322323
...(config.classifySubmit !== undefined ? { classifySubmit: config.classifySubmit } : {}),
323-
...(config.deliver !== undefined ? { deliver: config.deliver } : {}),
324324
});
325325
// Empty options accept the defaults (real clock, 250 ms tick, 15 min stall)
326326
// while still opting this host into the quota-retry / stall timers.
@@ -508,6 +508,7 @@ export async function mountProductHost(config: ProductHostConfig): Promise<Produ
508508
function onSessionClear(): void {
509509
if (disposed) return;
510510
clearTranscript(shell);
511+
bridge.clearQueuedDelivery();
511512
}
512513

513514
let currentModels = config.models ?? [];

‎src/tui/queued-delivery.test.ts‎

Lines changed: 93 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,93 @@
1+
import { describe, expect, test } from "bun:test";
2+
import type { PendingImageAttachment } from "./image-attachments.js";
3+
import { createDeliveryGeneration, routeQueuedDelivery } from "./queued-delivery.js";
4+
5+
const image: PendingImageAttachment = {
6+
id: "img-1",
7+
name: "clipboard.png",
8+
contentType: "image/png",
9+
data: new Uint8Array([1]),
10+
contentHash: "hash-1",
11+
};
12+
13+
describe("routeQueuedDelivery", () => {
14+
test("steer calls deliverSteer only", () => {
15+
const sends: string[] = [];
16+
const steers: string[] = [];
17+
const deliver = routeQueuedDelivery({
18+
send: (text) => {
19+
sends.push(text);
20+
},
21+
deliverSteer: (text) => {
22+
steers.push(text);
23+
},
24+
});
25+
deliver("asap", "steer");
26+
expect(steers).toEqual(["asap"]);
27+
expect(sends).toEqual([]);
28+
});
29+
30+
test("queue calls send only", () => {
31+
const sends: string[] = [];
32+
const steers: string[] = [];
33+
const deliver = routeQueuedDelivery({
34+
send: (text) => {
35+
sends.push(text);
36+
},
37+
deliverSteer: (text) => {
38+
steers.push(text);
39+
},
40+
});
41+
deliver("later", "queue");
42+
expect(sends).toEqual(["later"]);
43+
expect(steers).toEqual([]);
44+
});
45+
46+
test("forwards attachments on both kinds", () => {
47+
const sent: (readonly PendingImageAttachment[] | undefined)[] = [];
48+
const steered: (readonly PendingImageAttachment[] | undefined)[] = [];
49+
const deliver = routeQueuedDelivery({
50+
send: (_text, attachments) => {
51+
sent.push(attachments);
52+
},
53+
deliverSteer: (_text, attachments) => {
54+
steered.push(attachments);
55+
},
56+
});
57+
deliver("asap", "steer", [image]);
58+
deliver("later", "queue", [image]);
59+
expect(steered).toEqual([[image]]);
60+
expect(sent).toEqual([[image]]);
61+
});
62+
});
63+
64+
describe("createDeliveryGeneration", () => {
65+
test("capture then bump → predicate false", () => {
66+
const generation = createDeliveryGeneration();
67+
const stillCurrent = generation.capture();
68+
generation.bump();
69+
expect(stillCurrent()).toBe(false);
70+
});
71+
72+
test("capture without bump → true", () => {
73+
const generation = createDeliveryGeneration();
74+
const stillCurrent = generation.capture();
75+
expect(stillCurrent()).toBe(true);
76+
});
77+
78+
test("bump then new capture → true", () => {
79+
const generation = createDeliveryGeneration();
80+
generation.bump();
81+
const stillCurrent = generation.capture();
82+
expect(stillCurrent()).toBe(true);
83+
});
84+
85+
test("two captures, bump, both stale", () => {
86+
const generation = createDeliveryGeneration();
87+
const first = generation.capture();
88+
const second = generation.capture();
89+
generation.bump();
90+
expect(first()).toBe(false);
91+
expect(second()).toBe(false);
92+
});
93+
});

‎src/tui/queued-delivery.ts‎

Lines changed: 36 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,36 @@
1+
/**
2+
* Kind routing for drained queue items, plus a generation token so a
3+
* /clear|/new rotation can drop in-flight delivers that belonged to the
4+
* previous session. Kind routing lives here, not on SessionPort.
5+
*/
6+
7+
import type { PendingImageAttachment } from "./image-attachments.js";
8+
import type { ProductHostDeliver } from "./product-host.js";
9+
10+
export interface RouteQueuedDeliveryArgs {
11+
send: (text: string, attachments?: readonly PendingImageAttachment[]) => void;
12+
deliverSteer: (text: string, attachments?: readonly PendingImageAttachment[]) => void;
13+
}
14+
15+
export function routeQueuedDelivery(args: RouteQueuedDeliveryArgs): ProductHostDeliver {
16+
return (text, kind, attachments) => {
17+
if (kind === "steer") {
18+
args.deliverSteer(text, attachments);
19+
return;
20+
}
21+
args.send(text, attachments);
22+
};
23+
}
24+
25+
export function createDeliveryGeneration() {
26+
let generation = 0;
27+
return {
28+
bump(): void {
29+
generation += 1;
30+
},
31+
capture(): () => boolean {
32+
const captured = generation;
33+
return () => captured === generation;
34+
},
35+
};
36+
}

0 commit comments

Comments
 (0)