Skip to content

Commit bc844cd

Browse files
Recover queued messages when the target agent closes (#911)
A queued or steered message that hits a closed agent is restored to the prompt instead of being lost. Delivery settles through a typed result so the transcript row is corrected and the operator sees where the message went.
1 parent c7fc429 commit bc844cd

17 files changed

Lines changed: 802 additions & 117 deletions

‎docs/ARCHITECTURE.md‎

Lines changed: 7 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -84,11 +84,15 @@ In TUI chat mode there is no completion gate — the session stays open across t
8484
- **Mid-run injection** — Shell `session-queue` items drain at the parent
8585
`tool.boundary` through `SessionPort.deliver`. Production `routeQueuedDelivery`
8686
live-injects in-flight parent-boundary steers via `agentProxy.deliver`
87-
(`Agent.deliver`) into the live reactor. Idle leftover, idle-with-fleet, and
87+
(`Agent.deliver`) into one captured agent identity — never re-reading
88+
`currentAgent` after enqueue, so a rebuild cannot silently retarget the
89+
message. Idle leftover, idle-with-fleet, and
8890
post-interrupt steers, plus follow-ups (`kind === "queue"`), use the existing
89-
send path. `/clear` and `/new` bump a
91+
send path. A definitive `AgentClosedError` (or other not-delivered /
92+
uncertain settlement) returns ownership to bridge-owned prompt recovery
93+
rather than forwarding to a successor. `/clear` and `/new` bump a
9094
delivery generation and call `SessionBridge.clearQueuedDelivery()` so queued
91-
input from the previous session cannot enter the new one.
95+
input and deferred recoveries from the previous session cannot enter the new one.
9296
- **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.
9397

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

‎docs/TUI.md‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -570,13 +570,21 @@ Two mid-run gestures, two delivery times (CL-6290):
570570
steer is queued so occupancy can deliver it. The transcript row says
571571
`[will steer next]` while pending and
572572
`[steering]` once delivered (`submitPrompt`, `drainSteersAtBoundary` in
573-
`runtime-bridge.ts`).
573+
`runtime-bridge.ts`). If the captured target agent is already closed when
574+
delivery runs, the bridge restores the exact message (and attachments) to an
575+
empty prompt, or FIFO-defers behind a draft the operator already typed — it
576+
never auto-sends to a rebuilt successor. The transcript row is corrected to
577+
`[not delivered]` (or `[delivery uncertain]` for non-closed failures), and
578+
another explicit Enter is required before any new logical delivery.
574579
- **Alt+Enter, mid-run** — follow-up: enqueues kind `"queue"` and delivers
575580
only on **session-idle** (parent-idle and no live fleet lanes) as a `send`.
576581
Does not interrupt or reinject. The transcript row says `[will follow up]`
577582
while pending and `[following up]` once delivered. Idle, or with an empty
578583
prompt, Alt+Enter does nothing — there is nothing to wait for. (Internal
579584
`"reinject"` remains in the submit API for tests; no product chord wires it.)
585+
Closed-target recovery for follow-ups uses the same prompt-restore / draft-
586+
defer ownership as soft steer; `/clear`, `/new`, and dispose discard both
587+
in-flight deliveries and deferred recoveries with the old session.
580588

581589
When `steer > 0` and a parent tool has been in flight ≥ `STEER_WAIT_NOTICE_MS`
582590
(3s), the notice row adds `waiting on <tool>` (e.g. `waiting on run_shell`).
Lines changed: 71 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -1,58 +1,100 @@
11
import { describe, expect, test } from "bun:test";
2-
import { deliverAgentMessage } from "./deliver-agent-message.js";
2+
import { AgentClosedError } from "@intx/agent";
3+
import {
4+
deliverAgentMessage,
5+
deliveryResultNotice,
6+
} from "./deliver-agent-message.js";
37

48
describe("deliverAgentMessage", () => {
5-
test("surfaces a not-delivered notice instead of throwing when the agent is mid-rebuild", async () => {
6-
const notices: string[] = [];
9+
test("reports session-unavailable without calling deliver when rebuild failed", async () => {
710
const fatal = new Error("agent rebuild failed: provider unreachable");
8-
let delivered = false;
11+
let delivered = 0;
912

10-
await deliverAgentMessage({
13+
const result = await deliverAgentMessage({
1114
getFatalBuildError: () => fatal,
1215
deliverToLiveAgent: () => {
13-
delivered = true;
16+
delivered += 1;
1417
},
15-
onDeliverFailure: (message) => notices.push(message),
1618
});
1719

18-
// The rebuild failed, so currentAgent still points at the closed agent.
19-
// Delivery must never be attempted against it, and the operator must see
20-
// why their message did not go through.
21-
expect(delivered).toBe(false);
22-
expect(notices).toHaveLength(1);
23-
expect(notices[0]).toContain("not delivered");
24-
expect(notices[0]).toContain("provider unreachable");
20+
expect(delivered).toBe(0);
21+
expect(result).toEqual({
22+
status: "not-delivered",
23+
reason: "session-unavailable",
24+
detail: "agent rebuild failed: provider unreachable",
25+
});
2526
});
2627

27-
test("surfaces a not-delivered notice when the live agent throws on delivery", async () => {
28-
const notices: string[] = [];
28+
test("classifies typed AgentClosedError as agent-closed, not by message text", async () => {
29+
let delivered = 0;
30+
31+
const result = await deliverAgentMessage({
32+
getFatalBuildError: () => null,
33+
deliverToLiveAgent: () => {
34+
delivered += 1;
35+
throw new AgentClosedError();
36+
},
37+
});
38+
39+
expect(delivered).toBe(1);
40+
expect(result.status).toBe("not-delivered");
41+
if (result.status === "not-delivered") {
42+
expect(result.reason).toBe("agent-closed");
43+
}
44+
});
2945

30-
await deliverAgentMessage({
46+
test("treats a plain Error whose message says agent is closed as uncertain", async () => {
47+
const result = await deliverAgentMessage({
3148
getFatalBuildError: () => null,
3249
deliverToLiveAgent: () => {
3350
throw new Error("agent is closed");
3451
},
35-
onDeliverFailure: (message) => notices.push(message),
3652
});
3753

38-
expect(notices).toHaveLength(1);
39-
expect(notices[0]).toContain("not delivered");
40-
expect(notices[0]).toContain("agent is closed");
54+
expect(result).toEqual({
55+
status: "uncertain",
56+
detail: "agent is closed",
57+
});
4158
});
4259

43-
test("delivers normally and stays silent when the agent is healthy", async () => {
44-
const notices: string[] = [];
45-
let delivered = false;
60+
test("delivers once and returns accepted when the agent is healthy", async () => {
61+
let delivered = 0;
4662

47-
await deliverAgentMessage({
63+
const result = await deliverAgentMessage({
4864
getFatalBuildError: () => null,
4965
deliverToLiveAgent: () => {
50-
delivered = true;
66+
delivered += 1;
5167
},
52-
onDeliverFailure: (message) => notices.push(message),
5368
});
5469

55-
expect(delivered).toBe(true);
56-
expect(notices).toHaveLength(0);
70+
expect(delivered).toBe(1);
71+
expect(result).toEqual({ status: "accepted" });
72+
});
73+
});
74+
75+
describe("deliveryResultNotice", () => {
76+
test("closed restored and deferred copy is actionable", () => {
77+
const closed = {
78+
status: "not-delivered" as const,
79+
reason: "agent-closed" as const,
80+
detail: "agent is closed",
81+
};
82+
expect(deliveryResultNotice(closed, "restored")).toBe(
83+
"Message not delivered because the agent closed. It is back in the prompt; press Enter to send it.",
84+
);
85+
expect(deliveryResultNotice(closed, "deferred")).toBe(
86+
"Message not delivered because the agent closed. Your current draft is unchanged; the message will return to the prompt after you send it.",
87+
);
88+
});
89+
90+
test("uncertain copy does not claim nondelivery", () => {
91+
expect(
92+
deliveryResultNotice(
93+
{ status: "uncertain", detail: "network reset" },
94+
"restored",
95+
),
96+
).toBe(
97+
"Delivery failed: network reset. Delivery status is uncertain; review the transcript before sending again. It is back in the prompt; press Enter to send it.",
98+
);
5799
});
58100
});

‎src/tui/deliver-agent-message.ts‎

Lines changed: 78 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -1,28 +1,95 @@
11
/**
2-
* Guards a queued/steer deliver against a mid-rebuild agent. The shell paints
3-
* the delivered row and pops the queue item before this runs, so a failure
4-
* here must be surfaced — a swallowed error here means the transcript claims
5-
* delivery for a message that never reached the agent.
2+
* Guards a queued/steer deliver against a mid-rebuild or closed agent. The shell
3+
* paints the delivered row and pops the queue item before this runs, so the
4+
* caller must settle ownership from the structured result — a swallowed failure
5+
* here means the transcript claims delivery for a message that never reached
6+
* the agent.
67
*/
8+
import { AgentClosedError } from "@intx/agent";
9+
10+
export type AgentDeliveryNotDeliveredReason =
11+
| "agent-closed"
12+
| "session-unavailable"
13+
| "superseded"
14+
| "preparation-failed";
15+
16+
export type AgentDeliveryResult =
17+
| { readonly status: "accepted" }
18+
| {
19+
readonly status: "not-delivered";
20+
readonly reason: AgentDeliveryNotDeliveredReason;
21+
readonly detail: string;
22+
}
23+
| {
24+
readonly status: "uncertain";
25+
readonly detail: string;
26+
};
27+
728
export interface DeliverAgentMessageDeps {
829
getFatalBuildError: () => Error | null;
930
deliverToLiveAgent: () => void;
10-
onDeliverFailure: (message: string) => void;
1131
}
1232

1333
export async function deliverAgentMessage(
1434
deps: DeliverAgentMessageDeps,
15-
): Promise<void> {
35+
): Promise<AgentDeliveryResult> {
1636
const fatal = deps.getFatalBuildError();
1737
if (fatal !== null) {
18-
deps.onDeliverFailure(`Message not delivered: ${fatal.message}`);
19-
return;
38+
return {
39+
status: "not-delivered",
40+
reason: "session-unavailable",
41+
detail: fatal.message,
42+
};
2043
}
2144
try {
2245
deps.deliverToLiveAgent();
46+
return { status: "accepted" };
2347
} catch (err) {
24-
deps.onDeliverFailure(
25-
`Message not delivered: ${err instanceof Error ? err.message : String(err)}`,
26-
);
48+
if (err instanceof AgentClosedError) {
49+
return {
50+
status: "not-delivered",
51+
reason: "agent-closed",
52+
detail: err.message,
53+
};
54+
}
55+
return {
56+
status: "uncertain",
57+
detail: err instanceof Error ? err.message : String(err),
58+
};
59+
}
60+
}
61+
62+
/** Operator-facing copy for a settled delivery that did not accept. */
63+
export function deliveryResultNotice(
64+
result: Exclude<AgentDeliveryResult, { status: "accepted" }>,
65+
disposition: "restored" | "deferred" | "none" = "none",
66+
): string {
67+
if (result.status === "uncertain") {
68+
const base = `Delivery failed: ${result.detail}. Delivery status is uncertain; review the transcript before sending again.`;
69+
return appendDisposition(base, disposition);
70+
}
71+
if (result.reason === "agent-closed") {
72+
if (disposition === "restored") {
73+
return "Message not delivered because the agent closed. It is back in the prompt; press Enter to send it.";
74+
}
75+
if (disposition === "deferred") {
76+
return "Message not delivered because the agent closed. Your current draft is unchanged; the message will return to the prompt after you send it.";
77+
}
78+
return "Message not delivered because the agent closed.";
79+
}
80+
const base = `Message not delivered: ${result.detail}`;
81+
return appendDisposition(base, disposition);
82+
}
83+
84+
function appendDisposition(
85+
base: string,
86+
disposition: "restored" | "deferred" | "none",
87+
): string {
88+
if (disposition === "restored") {
89+
return `${base} It is back in the prompt; press Enter to send it.`;
90+
}
91+
if (disposition === "deferred") {
92+
return `${base} Your current draft is unchanged; the message will return to the prompt after you send it.`;
2793
}
94+
return base;
2895
}

‎src/tui/gutter-labels.test.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -43,6 +43,8 @@ const STORED_META_LITERALS = [
4343
"following-up",
4444
"reinject",
4545
"cancelled",
46+
"not-delivered",
47+
"delivery-uncertain",
4648
];
4749

4850
const FORBIDDEN = ["permission", "command", "overlay"];

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

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@
55

66
import type { PendingImageAttachment } from "./image-attachments.js";
77
import type { QueueItem, QueueKind } from "./session-queue.js";
8+
import type { DeliverySettle } from "./queued-delivery.js";
89
import type { SessionPort } from "./runtime-bridge.js";
910

1011
export type SubmitClassification = "agent" | "local" | "empty";
@@ -28,6 +29,7 @@ export interface LiveSessionPortDeps {
2829
text: string,
2930
kind: QueueKind,
3031
attachments?: readonly PendingImageAttachment[],
32+
settle?: DeliverySettle,
3133
) => void;
3234
}
3335

@@ -56,8 +58,8 @@ export function createLiveSessionPort(deps: LiveSessionPortDeps): SessionPort {
5658
interrupt: (): void => {
5759
deps.interrupt();
5860
},
59-
deliver: (item: QueueItem): void => {
60-
deps.deliver(item.text, item.kind, item.attachments);
61+
deliver: (item: QueueItem, settle?: DeliverySettle): void => {
62+
deps.deliver(item.text, item.kind, item.attachments, settle);
6163
},
6264
};
6365
}

‎src/tui/product-host.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -67,6 +67,7 @@ import { hydrateHistoryRows } from "./history-hydrate.js";
6767
import type { StreamRow } from "./stream.js";
6868

6969
import type { PendingImageAttachment } from "./image-attachments.js";
70+
import type { DeliverySettle } from "./queued-delivery.js";
7071

7172
/** Suffix the row matching `activeId` (if any) so it reads as the current pick. */
7273
function annotateCurrent(
@@ -93,6 +94,7 @@ export type ProductHostDeliver = (
9394
text: string,
9495
kind: QueueKind,
9596
attachments?: readonly PendingImageAttachment[],
97+
settle?: DeliverySettle,
9698
) => void;
9799

98100
/**

0 commit comments

Comments
 (0)