From f428d213af6f8dcef80be09c8e9ad29ecee1e694 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Thu, 10 Sep 2026 22:33:00 -0700 Subject: [PATCH] Recover queued messages when the target agent closes 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. --- docs/ARCHITECTURE.md | 10 +- docs/TUI.md | 10 +- src/tui/deliver-agent-message.test.ts | 100 ++++++++--- src/tui/deliver-agent-message.ts | 89 +++++++-- src/tui/gutter-labels.test.ts | 2 + src/tui/live-session-port.ts | 6 +- src/tui/product-host.ts | 2 + src/tui/queued-delivery-hop.test.ts | 250 +++++++++++++++++++++++++- src/tui/queued-delivery.test.ts | 25 +++ src/tui/queued-delivery.ts | 122 +++++++++++-- src/tui/runner/exit.ts | 3 +- src/tui/runner/session.ts | 39 +++- src/tui/runner/state.ts | 14 +- src/tui/runner/submit.ts | 55 ++++-- src/tui/runtime-bridge.ts | 142 +++++++++++++-- src/tui/stream.test.ts | 19 ++ src/tui/stream.ts | 31 ++-- 17 files changed, 802 insertions(+), 117 deletions(-) diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 0c5204ea1..87b6745c4 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -84,11 +84,15 @@ In TUI chat mode there is no completion gate — the session stays open across t - **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 + (`Agent.deliver`) into one captured agent identity — never re-reading + `currentAgent` after enqueue, so a rebuild cannot silently retarget the + message. Idle leftover, idle-with-fleet, and post-interrupt steers, plus follow-ups (`kind === "queue"`), use the existing - send path. `/clear` and `/new` bump a + send path. A definitive `AgentClosedError` (or other not-delivered / + uncertain settlement) returns ownership to bridge-owned prompt recovery + rather than forwarding to a successor. `/clear` and `/new` bump a delivery generation and call `SessionBridge.clearQueuedDelivery()` so queued - input from the previous session cannot enter the new one. + input and deferred recoveries from the previous session cannot enter the new one. - **Session rotation** — Uses a serial session-operation queue (`createSessionOperationQueue`, not a boolean flag) so rotation, compaction continuation, and `agentProxy.deliver` never race a concurrent rebuild. Each operation chains onto the tail, ensuring in-flight work completes before the agent is torn down. ### Exec Runner (`src/exec/runner.ts`) diff --git a/docs/TUI.md b/docs/TUI.md index 3c79d8e88..f441518b0 100644 --- a/docs/TUI.md +++ b/docs/TUI.md @@ -571,13 +571,21 @@ Two mid-run gestures, two delivery times (CL-6290): steer is queued so occupancy can deliver it. The transcript row says `[will steer next]` while pending and `[steering]` once delivered (`submitPrompt`, `drainSteersAtBoundary` in - `runtime-bridge.ts`). + `runtime-bridge.ts`). If the captured target agent is already closed when + delivery runs, the bridge restores the exact message (and attachments) to an + empty prompt, or FIFO-defers behind a draft the operator already typed — it + never auto-sends to a rebuilt successor. The transcript row is corrected to + `[not delivered]` (or `[delivery uncertain]` for non-closed failures), and + another explicit Enter is required before any new logical delivery. - **Alt+Enter, mid-run** — follow-up: enqueues kind `"queue"` and delivers 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.) + Closed-target recovery for follow-ups uses the same prompt-restore / draft- + defer ownership as soft steer; `/clear`, `/new`, and dispose discard both + in-flight deliveries and deferred recoveries with the old session. When `steer > 0` and a parent tool has been in flight ≥ `STEER_WAIT_NOTICE_MS` (3s), the notice row adds `waiting on ` (e.g. `waiting on run_shell`). diff --git a/src/tui/deliver-agent-message.test.ts b/src/tui/deliver-agent-message.test.ts index 5f569dcc1..aed319015 100644 --- a/src/tui/deliver-agent-message.test.ts +++ b/src/tui/deliver-agent-message.test.ts @@ -1,58 +1,100 @@ import { describe, expect, test } from "bun:test"; -import { deliverAgentMessage } from "./deliver-agent-message.js"; +import { AgentClosedError } from "@intx/agent"; +import { + deliverAgentMessage, + deliveryResultNotice, +} from "./deliver-agent-message.js"; describe("deliverAgentMessage", () => { - test("surfaces a not-delivered notice instead of throwing when the agent is mid-rebuild", async () => { - const notices: string[] = []; + test("reports session-unavailable without calling deliver when rebuild failed", async () => { const fatal = new Error("agent rebuild failed: provider unreachable"); - let delivered = false; + let delivered = 0; - await deliverAgentMessage({ + const result = await deliverAgentMessage({ getFatalBuildError: () => fatal, deliverToLiveAgent: () => { - delivered = true; + delivered += 1; }, - onDeliverFailure: (message) => notices.push(message), }); - // The rebuild failed, so currentAgent still points at the closed agent. - // Delivery must never be attempted against it, and the operator must see - // why their message did not go through. - expect(delivered).toBe(false); - expect(notices).toHaveLength(1); - expect(notices[0]).toContain("not delivered"); - expect(notices[0]).toContain("provider unreachable"); + expect(delivered).toBe(0); + expect(result).toEqual({ + status: "not-delivered", + reason: "session-unavailable", + detail: "agent rebuild failed: provider unreachable", + }); }); - test("surfaces a not-delivered notice when the live agent throws on delivery", async () => { - const notices: string[] = []; + test("classifies typed AgentClosedError as agent-closed, not by message text", async () => { + let delivered = 0; + + const result = await deliverAgentMessage({ + getFatalBuildError: () => null, + deliverToLiveAgent: () => { + delivered += 1; + throw new AgentClosedError(); + }, + }); + + expect(delivered).toBe(1); + expect(result.status).toBe("not-delivered"); + if (result.status === "not-delivered") { + expect(result.reason).toBe("agent-closed"); + } + }); - await deliverAgentMessage({ + test("treats a plain Error whose message says agent is closed as uncertain", async () => { + const result = await deliverAgentMessage({ getFatalBuildError: () => null, deliverToLiveAgent: () => { throw new Error("agent is closed"); }, - onDeliverFailure: (message) => notices.push(message), }); - expect(notices).toHaveLength(1); - expect(notices[0]).toContain("not delivered"); - expect(notices[0]).toContain("agent is closed"); + expect(result).toEqual({ + status: "uncertain", + detail: "agent is closed", + }); }); - test("delivers normally and stays silent when the agent is healthy", async () => { - const notices: string[] = []; - let delivered = false; + test("delivers once and returns accepted when the agent is healthy", async () => { + let delivered = 0; - await deliverAgentMessage({ + const result = await deliverAgentMessage({ getFatalBuildError: () => null, deliverToLiveAgent: () => { - delivered = true; + delivered += 1; }, - onDeliverFailure: (message) => notices.push(message), }); - expect(delivered).toBe(true); - expect(notices).toHaveLength(0); + expect(delivered).toBe(1); + expect(result).toEqual({ status: "accepted" }); + }); +}); + +describe("deliveryResultNotice", () => { + test("closed restored and deferred copy is actionable", () => { + const closed = { + status: "not-delivered" as const, + reason: "agent-closed" as const, + detail: "agent is closed", + }; + expect(deliveryResultNotice(closed, "restored")).toBe( + "Message not delivered because the agent closed. It is back in the prompt; press Enter to send it.", + ); + expect(deliveryResultNotice(closed, "deferred")).toBe( + "Message not delivered because the agent closed. Your current draft is unchanged; the message will return to the prompt after you send it.", + ); + }); + + test("uncertain copy does not claim nondelivery", () => { + expect( + deliveryResultNotice( + { status: "uncertain", detail: "network reset" }, + "restored", + ), + ).toBe( + "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.", + ); }); }); diff --git a/src/tui/deliver-agent-message.ts b/src/tui/deliver-agent-message.ts index 6a30facf8..7b2b8545b 100644 --- a/src/tui/deliver-agent-message.ts +++ b/src/tui/deliver-agent-message.ts @@ -1,28 +1,95 @@ /** - * Guards a queued/steer deliver against a mid-rebuild agent. The shell paints - * the delivered row and pops the queue item before this runs, so a failure - * here must be surfaced — a swallowed error here means the transcript claims - * delivery for a message that never reached the agent. + * Guards a queued/steer deliver against a mid-rebuild or closed agent. The shell + * paints the delivered row and pops the queue item before this runs, so the + * caller must settle ownership from the structured result — a swallowed failure + * here means the transcript claims delivery for a message that never reached + * the agent. */ +import { AgentClosedError } from "@intx/agent"; + +export type AgentDeliveryNotDeliveredReason = + | "agent-closed" + | "session-unavailable" + | "superseded" + | "preparation-failed"; + +export type AgentDeliveryResult = + | { readonly status: "accepted" } + | { + readonly status: "not-delivered"; + readonly reason: AgentDeliveryNotDeliveredReason; + readonly detail: string; + } + | { + readonly status: "uncertain"; + readonly detail: string; + }; + export interface DeliverAgentMessageDeps { getFatalBuildError: () => Error | null; deliverToLiveAgent: () => void; - onDeliverFailure: (message: string) => void; } export async function deliverAgentMessage( deps: DeliverAgentMessageDeps, -): Promise { +): Promise { const fatal = deps.getFatalBuildError(); if (fatal !== null) { - deps.onDeliverFailure(`Message not delivered: ${fatal.message}`); - return; + return { + status: "not-delivered", + reason: "session-unavailable", + detail: fatal.message, + }; } try { deps.deliverToLiveAgent(); + return { status: "accepted" }; } catch (err) { - deps.onDeliverFailure( - `Message not delivered: ${err instanceof Error ? err.message : String(err)}`, - ); + if (err instanceof AgentClosedError) { + return { + status: "not-delivered", + reason: "agent-closed", + detail: err.message, + }; + } + return { + status: "uncertain", + detail: err instanceof Error ? err.message : String(err), + }; + } +} + +/** Operator-facing copy for a settled delivery that did not accept. */ +export function deliveryResultNotice( + result: Exclude, + disposition: "restored" | "deferred" | "none" = "none", +): string { + if (result.status === "uncertain") { + const base = `Delivery failed: ${result.detail}. Delivery status is uncertain; review the transcript before sending again.`; + return appendDisposition(base, disposition); + } + if (result.reason === "agent-closed") { + if (disposition === "restored") { + return "Message not delivered because the agent closed. It is back in the prompt; press Enter to send it."; + } + if (disposition === "deferred") { + return "Message not delivered because the agent closed. Your current draft is unchanged; the message will return to the prompt after you send it."; + } + return "Message not delivered because the agent closed."; + } + const base = `Message not delivered: ${result.detail}`; + return appendDisposition(base, disposition); +} + +function appendDisposition( + base: string, + disposition: "restored" | "deferred" | "none", +): string { + if (disposition === "restored") { + return `${base} It is back in the prompt; press Enter to send it.`; + } + if (disposition === "deferred") { + return `${base} Your current draft is unchanged; the message will return to the prompt after you send it.`; } + return base; } diff --git a/src/tui/gutter-labels.test.ts b/src/tui/gutter-labels.test.ts index 15c2b3eff..f8a98c9e0 100644 --- a/src/tui/gutter-labels.test.ts +++ b/src/tui/gutter-labels.test.ts @@ -43,6 +43,8 @@ const STORED_META_LITERALS = [ "following-up", "reinject", "cancelled", + "not-delivered", + "delivery-uncertain", ]; const FORBIDDEN = ["permission", "command", "overlay"]; diff --git a/src/tui/live-session-port.ts b/src/tui/live-session-port.ts index 7b6255298..45e6a0087 100644 --- a/src/tui/live-session-port.ts +++ b/src/tui/live-session-port.ts @@ -5,6 +5,7 @@ import type { PendingImageAttachment } from "./image-attachments.js"; import type { QueueItem, QueueKind } from "./session-queue.js"; +import type { DeliverySettle } from "./queued-delivery.js"; import type { SessionPort } from "./runtime-bridge.js"; export type SubmitClassification = "agent" | "local" | "empty"; @@ -28,6 +29,7 @@ export interface LiveSessionPortDeps { text: string, kind: QueueKind, attachments?: readonly PendingImageAttachment[], + settle?: DeliverySettle, ) => void; } @@ -56,8 +58,8 @@ export function createLiveSessionPort(deps: LiveSessionPortDeps): SessionPort { interrupt: (): void => { deps.interrupt(); }, - deliver: (item: QueueItem): void => { - deps.deliver(item.text, item.kind, item.attachments); + deliver: (item: QueueItem, settle?: DeliverySettle): void => { + deps.deliver(item.text, item.kind, item.attachments, settle); }, }; } diff --git a/src/tui/product-host.ts b/src/tui/product-host.ts index 39d01a211..a54a6be6c 100644 --- a/src/tui/product-host.ts +++ b/src/tui/product-host.ts @@ -67,6 +67,7 @@ import { hydrateHistoryRows } from "./history-hydrate.js"; import type { StreamRow } from "./stream.js"; import type { PendingImageAttachment } from "./image-attachments.js"; +import type { DeliverySettle } from "./queued-delivery.js"; /** Suffix the row matching `activeId` (if any) so it reads as the current pick. */ function annotateCurrent( @@ -93,6 +94,7 @@ export type ProductHostDeliver = ( text: string, kind: QueueKind, attachments?: readonly PendingImageAttachment[], + settle?: DeliverySettle, ) => void; /** diff --git a/src/tui/queued-delivery-hop.test.ts b/src/tui/queued-delivery-hop.test.ts index b52a921c5..0bc728476 100644 --- a/src/tui/queued-delivery-hop.test.ts +++ b/src/tui/queued-delivery-hop.test.ts @@ -10,9 +10,11 @@ import { withTestRenderer } from "./harness"; import { createLiveSteerDeliver, routeQueuedDelivery, + type DeliverySettle, } from "./queued-delivery.js"; import { createSessionOperationQueue } from "./session-operation-queue.js"; -import { badgeCount } from "./session-queue"; +import { badgeCount, type QueueItem } from "./session-queue"; +import type { AgentDeliveryResult } from "./deliver-agent-message.js"; function lastHopPort(bridgeRef: { current: SessionBridge | undefined }) { const sends: string[] = []; @@ -287,3 +289,249 @@ describe("queued delivery last hop", () => { ); }); }); + +function makeRecoveryPort(results: AgentDeliveryResult[]) { + const calls: { item: QueueItem; settle: DeliverySettle | undefined }[] = []; + const sentImmediate: string[] = []; + const port = { + sendImmediate: (text: string) => { + sentImmediate.push(text); + }, + deliver: (item: QueueItem, settle?: DeliverySettle) => { + calls.push({ item, settle }); + const next = results.shift(); + if (next !== undefined) settle?.(next); + }, + }; + return { port, calls, sentImmediate }; +} + +const CLOSED_RESULT: AgentDeliveryResult = { + status: "not-delivered", + reason: "agent-closed", + detail: "agent is closed", +}; + +function drainedUserRow(shell: { streamLog: readonly unknown[] }): { + queueItemId?: string; + meta?: string; + deliveryStatus?: string; +} { + // The enqueue-time tag row and the dispatched row share the queueItemId; + // the dispatched row is the one handed to the run. + const rows = shell.streamLog.filter( + (candidate): candidate is { queueItemId?: string } => + typeof candidate === "object" && + candidate !== null && + "queueItemId" in candidate, + ); + const row = rows.at(-1); + if (row === undefined) throw new Error("expected a drained queue row"); + return row as { + queueItemId?: string; + meta?: string; + deliveryStatus?: string; + }; +} + +function recoveryNotices(shell: { + streamLog: readonly { role: string; text: string }[]; +}): string[] { + return shell.streamLog + .filter((row) => row.role === "system") + .map((row) => row.text) + .filter( + (text) => text.includes("not delivered") || text.includes("uncertain"), + ); +} + +describe("closed-target recovery", () => { + test("agent-closed restores the exact message to an empty prompt and corrects the row", async () => { + await withTestRenderer( + async (h) => { + const shell = createAppShell(h.renderer, { + terminal: { columns: 80, rows: 24 }, + wireKeys: false, + run: "busy", + }); + const { port, calls } = makeRecoveryPort([CLOSED_RESULT]); + const bridge = attachSessionBridge(shell, port); + try { + bridge.submit("steer the ship", "steer", [ + { + id: "shot-1", + name: "shot.png", + contentType: "image/png", + data: new Uint8Array([1]), + contentHash: "hash-shot", + }, + ]); + bridge.handle({ type: "tool.boundary" }); + expect(calls).toHaveLength(1); + + // Ownership returns to the composer with the exact payload intact. + expect(shell.prompt.value).toBe("steer the ship"); + expect( + shell.pendingAttachments.map((attachment) => attachment.id), + ).toEqual(["shot-1"]); + + // The row painted as delivered now reads as not delivered. + const row = drainedUserRow(shell); + expect(row.meta).toBe("not-delivered"); + expect(row.deliveryStatus).toBe("not-delivered"); + + // Actionable copy states the delivery status and recovery location. + const notices = recoveryNotices(shell); + expect(notices).toHaveLength(1); + expect(notices[0]).toContain("not delivered"); + expect(notices[0]).toContain("prompt"); + + // The popped item cannot redispatch at a later boundary. + bridge.handle({ type: "tool.boundary" }); + expect(calls).toHaveLength(1); + } finally { + bridge.dispose(); + shell.dispose(); + } + }, + { width: 80, height: 24 }, + ); + }); + + test("agent-closed with a draft in the composer defers recovery behind it", async () => { + await withTestRenderer( + async (h) => { + const shell = createAppShell(h.renderer, { + terminal: { columns: 80, rows: 24 }, + wireKeys: false, + run: "busy", + }); + const { port, calls, sentImmediate } = makeRecoveryPort([ + CLOSED_RESULT, + ]); + const bridge = attachSessionBridge(shell, port); + try { + bridge.submit("steer the ship", "steer"); + shell.prompt.value = "draft in progress"; + bridge.handle({ type: "tool.boundary" }); + expect(calls).toHaveLength(1); + + // The operator's draft is untouched; the notice says where the + // failed message went. + expect(shell.prompt.value).toBe("draft in progress"); + const notices = recoveryNotices(shell); + expect(notices).toHaveLength(1); + expect(notices[0]).toContain("draft is unchanged"); + + // Sending the draft returns the failed message to the prompt. The + // Enter handler clears the composer before submit; model that here. + shell.prompt.value = ""; + bridge.submit("draft in progress", "immediate"); + expect(sentImmediate).toEqual(["draft in progress"]); + expect(shell.prompt.value).toBe("steer the ship"); + expect(calls).toHaveLength(1); + } finally { + bridge.dispose(); + shell.dispose(); + } + }, + { width: 80, height: 24 }, + ); + }); + + test("uncertain delivery marks the row without claiming nondelivery", async () => { + await withTestRenderer( + async (h) => { + const shell = createAppShell(h.renderer, { + terminal: { columns: 80, rows: 24 }, + wireKeys: false, + run: "busy", + }); + const { port, calls } = makeRecoveryPort([ + { status: "uncertain", detail: "connection reset" }, + ]); + const bridge = attachSessionBridge(shell, port); + try { + bridge.submit("steer the ship", "steer"); + bridge.handle({ type: "tool.boundary" }); + expect(calls).toHaveLength(1); + + const row = drainedUserRow(shell); + expect(row.meta).toBe("delivery-uncertain"); + expect(row.deliveryStatus).toBe("uncertain"); + + // Content is still preserved even though delivery is unknown. + expect(shell.prompt.value).toBe("steer the ship"); + + const notices = recoveryNotices(shell); + expect(notices).toHaveLength(1); + expect(notices[0]).toContain("uncertain"); + } finally { + bridge.dispose(); + shell.dispose(); + } + }, + { width: 80, height: 24 }, + ); + }); + + test("accepted delivery leaves the prompt and transcript alone", async () => { + await withTestRenderer( + async (h) => { + const shell = createAppShell(h.renderer, { + terminal: { columns: 80, rows: 24 }, + wireKeys: false, + run: "busy", + }); + const { port, calls } = makeRecoveryPort([{ status: "accepted" }]); + const bridge = attachSessionBridge(shell, port); + try { + bridge.submit("steer the ship", "steer"); + bridge.handle({ type: "tool.boundary" }); + expect(calls).toHaveLength(1); + + expect(shell.prompt.value).toBe(""); + expect(drainedUserRow(shell).meta).toBe("steering"); + expect(recoveryNotices(shell)).toEqual([]); + } finally { + bridge.dispose(); + shell.dispose(); + } + }, + { width: 80, height: 24 }, + ); + }); + + test("a second settle for the same item is dropped, never resent", async () => { + await withTestRenderer( + async (h) => { + const shell = createAppShell(h.renderer, { + terminal: { columns: 80, rows: 24 }, + wireKeys: false, + run: "busy", + }); + const { port, calls } = makeRecoveryPort([]); + const bridge = attachSessionBridge(shell, port); + try { + bridge.submit("steer the ship", "steer"); + bridge.handle({ type: "tool.boundary" }); + expect(calls).toHaveLength(1); + + const settle = calls[0]?.settle; + if (settle === undefined) + throw new Error("expected a settle callback"); + settle(CLOSED_RESULT); + settle(CLOSED_RESULT); + + expect(calls).toHaveLength(1); + expect(shell.prompt.value).toBe("steer the ship"); + expect(recoveryNotices(shell)).toHaveLength(1); + } finally { + bridge.dispose(); + shell.dispose(); + } + }, + { width: 80, height: 24 }, + ); + }); +}); diff --git a/src/tui/queued-delivery.test.ts b/src/tui/queued-delivery.test.ts index 34871ac0b..69b50a548 100644 --- a/src/tui/queued-delivery.test.ts +++ b/src/tui/queued-delivery.test.ts @@ -8,6 +8,7 @@ import { SESSION_IDENTITY_ABORT_REASON, } from "./queued-delivery.js"; import { createSessionOperationQueue } from "./session-operation-queue.js"; +import type { AgentDeliveryResult } from "./deliver-agent-message.js"; const image: PendingImageAttachment = { id: "img-1", @@ -251,6 +252,30 @@ describe("createLeftoverSend", () => { expect(recorded).toEqual(["follow-up"]); }); + test("leftover send preserves an asynchronous closed-agent settlement", async () => { + const { enqueue, awaitTail } = createSessionOperationQueue(); + const settlements: AgentDeliveryResult[] = []; + const closed: AgentDeliveryResult = { + status: "not-delivered", + reason: "agent-closed", + detail: "agent is closed", + }; + const leftoverSend = createLeftoverSend({ + enqueue, + ingest: async (text) => ({ text, attachments: [] }), + send: async () => closed, + captureGeneration: () => () => true, + onFailure: (err) => { + throw err; + }, + }); + + leftoverSend("follow-up", [], (result) => settlements.push(result)); + await awaitTail(); + + expect(settlements).toEqual([closed]); + }); + test("leftover send skips ingest for ask_director wake and mailbox mail", async () => { const sent: string[] = []; const ingested: string[] = []; diff --git a/src/tui/queued-delivery.ts b/src/tui/queued-delivery.ts index fb7dccbed..1b0428e58 100644 --- a/src/tui/queued-delivery.ts +++ b/src/tui/queued-delivery.ts @@ -7,17 +7,27 @@ * parent tool.boundary. Leftover steers at idle, idle-with-fleet, or * post-interrupt share the send path (sendQueue, inFlight, token refresh). */ - import type { PendingImageAttachment } from "./image-attachments.js"; +import type { AgentDeliveryResult } from "./deliver-agent-message.js"; import type { ProductHostDeliver } from "./product-host.js"; import { ASK_DIRECTOR_WAKE_PREFIX } from "../subagent/fleet-report.js"; import { MAILBOX_MAIL_WAKE_PREFIX } from "../subagent/mailbox-mail-drive.js"; +export type DeliverySettle = (result: AgentDeliveryResult) => void; +type MaybeAsyncDeliveryResult = + | Promise + | ReturnType<() => void>; + export interface RouteQueuedDeliveryArgs { - send: (text: string, attachments?: readonly PendingImageAttachment[]) => void; + send: ( + text: string, + attachments?: readonly PendingImageAttachment[], + settle?: DeliverySettle, + ) => void; deliverSteer: ( text: string, attachments?: readonly PendingImageAttachment[], + settle?: DeliverySettle, ) => void; /** * True only while the bridge is draining steers at a live parent @@ -30,12 +40,12 @@ export interface RouteQueuedDeliveryArgs { export function routeQueuedDelivery( args: RouteQueuedDeliveryArgs, ): ProductHostDeliver { - return (text, kind, attachments) => { + return (text, kind, attachments, settle) => { if (kind === "steer" && args.parentCycleLive()) { - args.deliverSteer(text, attachments); + args.deliverSteer(text, attachments, settle); return; } - args.send(text, attachments); + args.send(text, attachments, settle); }; } @@ -79,10 +89,14 @@ export interface CreateLiveSteerDeliverArgs { text: string, attachments: readonly PendingImageAttachment[], ) => Promise; - /** Agent.deliver (or the sessionOps enqueue that wraps it). */ + /** + * Agent.deliver hop. When `settle` is provided, the hop owns reporting the + * eventual AgentDeliveryResult (accepted / closed / uncertain). + */ deliver: ( text: string, attachments: readonly PendingImageAttachment[], + settle?: DeliverySettle, ) => void; captureGeneration: () => () => boolean; onFailure: (err: unknown) => void; @@ -98,7 +112,10 @@ export interface CreateLeftoverSendArgs { * Post-ingest hop (agentProxy.send). Must not ingest again — leftover * ingest already ran in this wrapper. */ - send: (text: string, attachments: readonly PendingImageAttachment[]) => void; + send: ( + text: string, + attachments: readonly PendingImageAttachment[], + ) => MaybeAsyncDeliveryResult; /** * Up/Down recall. Called with the original text only when the hop is * still current after ingest, so a /clear|/new drop is not recorded. @@ -114,27 +131,89 @@ interface GenerationGatedHopArgs { text: string, attachments: readonly PendingImageAttachment[], ) => Promise; - hop: (text: string, attachments: readonly PendingImageAttachment[]) => void; + hop: ( + text: string, + attachments: readonly PendingImageAttachment[], + settle?: DeliverySettle, + ) => MaybeAsyncDeliveryResult; + /** Leftover/send settles from the send promise; live steer uses the callback. */ + settleFromHopResult?: boolean; recordSent?: (text: string) => void; captureGeneration: () => () => boolean; onFailure: (err: unknown) => void; } +function settleOnce( + settle: DeliverySettle | undefined, + result: AgentDeliveryResult, +): void { + settle?.(result); +} + function createGenerationGatedHop( args: GenerationGatedHopArgs, -): (text: string, attachments?: readonly PendingImageAttachment[]) => void { - return (text, attachments) => { +): ( + text: string, + attachments?: readonly PendingImageAttachment[], + settle?: DeliverySettle, +) => void { + return (text, attachments, settle) => { const stillCurrent = args.captureGeneration(); const pending = attachments ?? []; + let settled = false; + const finish = (result: AgentDeliveryResult): void => { + if (settled) return; + settled = true; + settleOnce(settle, result); + }; void args .enqueue(async () => { - if (!stillCurrent()) return; - const ingested = await args.ingest(text, pending); - if (!stillCurrent()) return; + if (!stillCurrent()) { + finish({ + status: "not-delivered", + reason: "superseded", + detail: SESSION_IDENTITY_ABORT_REASON, + }); + return; + } + let ingested: IngestedSteer; + try { + ingested = await args.ingest(text, pending); + } catch (err) { + finish({ + status: "not-delivered", + reason: "preparation-failed", + detail: err instanceof Error ? err.message : String(err), + }); + if (settle === undefined) args.onFailure(err); + return; + } + if (!stillCurrent()) { + finish({ + status: "not-delivered", + reason: "superseded", + detail: SESSION_IDENTITY_ABORT_REASON, + }); + return; + } args.recordSent?.(text); - args.hop(ingested.text, ingested.attachments); + if (args.settleFromHopResult === true) { + const result = await args.hop(ingested.text, ingested.attachments); + finish(result ?? { status: "accepted" }); + return; + } + // Live steer: hop receives settle and reports the eventual result. + args.hop(ingested.text, ingested.attachments, (result) => { + finish(result); + }); }) - .catch(args.onFailure); + .catch((err: unknown) => { + finish({ + status: "uncertain", + detail: err instanceof Error ? err.message : String(err), + }); + if (settle === undefined) args.onFailure(err); + }); }; } @@ -144,7 +223,11 @@ function createGenerationGatedHop( */ export function createLiveSteerDeliver( args: CreateLiveSteerDeliverArgs, -): (text: string, attachments?: readonly PendingImageAttachment[]) => void { +): ( + text: string, + attachments?: readonly PendingImageAttachment[], + settle?: DeliverySettle, +) => void { return createGenerationGatedHop({ ...args, hop: args.deliver }); } @@ -156,10 +239,15 @@ export function createLiveSteerDeliver( */ export function createLeftoverSend( args: CreateLeftoverSendArgs, -): (text: string, attachments?: readonly PendingImageAttachment[]) => void { +): ( + text: string, + attachments?: readonly PendingImageAttachment[], + settle?: DeliverySettle, +) => void { return createGenerationGatedHop({ ...args, hop: args.send, + settleFromHopResult: true, ingest: async (text, pending) => text.startsWith(ASK_DIRECTOR_WAKE_PREFIX) || text.startsWith(MAILBOX_MAIL_WAKE_PREFIX) diff --git a/src/tui/runner/exit.ts b/src/tui/runner/exit.ts index b7877c615..a1d8bc8bb 100644 --- a/src/tui/runner/exit.ts +++ b/src/tui/runner/exit.ts @@ -427,7 +427,8 @@ export async function createRunLifecycle( }, stream: () => liveAgent(state).stream(), deliver: (message) => { - state.enqueueAgentDeliver?.(() => liveAgent(state).deliver(message)); + const targetAgent = liveAgent(state); + state.enqueueAgentDeliver?.(() => targetAgent.deliver(message)); }, close: () => liveAgent(state).close(), setSource: (source) => { diff --git a/src/tui/runner/session.ts b/src/tui/runner/session.ts index 74f16d64f..b96f09bec 100644 --- a/src/tui/runner/session.ts +++ b/src/tui/runner/session.ts @@ -88,7 +88,11 @@ import { detectLanguageServerAvailable } from "../../agent/lsp-availability.js"; import type { SessionMode } from "../../config/session-mode.js"; import { WorkflowHost, type WorkflowHostState } from "../../workflows/host.js"; import type { ToolWatchdogConfig } from "../tool-execution-watchdog.js"; -import { deliverAgentMessage } from "../deliver-agent-message.js"; +import { + deliverAgentMessage, + deliveryResultNotice, + type AgentDeliveryResult, +} from "../deliver-agent-message.js"; import { createProviderFailureAttemptTracker } from "../provider/failure-attempt.js"; import { getTelemetry, liveTelemetry } from "../../telemetry/singleton.js"; import { createChatDirector } from "../../agent/director.js"; @@ -329,8 +333,9 @@ export async function assembleTUISession( // A background run_shell completion re-enters the reactor on a later turn, // so queued steers are not blocked by a long foreground run. onBackgroundShellExit: (exit) => { + const targetAgent = liveAgent(state); state.enqueueAgentDeliver?.(() => - liveAgent(state).deliver(buildShellBackgroundMessage(exit)), + targetAgent.deliver(buildShellBackgroundMessage(exit)), ); }, getEvidenceArchive: () => evidenceArchiveHolder.current, @@ -484,18 +489,33 @@ export async function assembleTUISession( }, gate: permissionGate, }); - state.enqueueAgentDeliver = (deliverToLiveAgent: () => void): void => { + state.enqueueAgentDeliver = ( + deliverToLiveAgent: () => void, + onSettle?: (result: AgentDeliveryResult) => void, + ): void => { const stillCurrent = deliveryGeneration.capture(); void sessionOps.enqueue(async () => { - if (!stillCurrent()) return; + if (!stillCurrent()) { + onSettle?.({ + status: "not-delivered", + reason: "superseded", + detail: "session identity changed before delivery", + }); + return; + } // The shell already popped the queue item and painted it as delivered - // by the time this runs, so a failed rebuild must be surfaced here — - // otherwise the message silently never reaches the agent. - await deliverAgentMessage({ + // by the time this runs, so a failed rebuild or closed agent must settle + // here — otherwise the message silently never reaches the agent. + const result = await deliverAgentMessage({ getFatalBuildError: () => state.fatalBuildError, deliverToLiveAgent, - onDeliverFailure: (text) => state.systemNotice?.(text), }); + if (onSettle !== undefined) { + onSettle(result); + return; + } + if (result.status === "accepted") return; + state.systemNotice?.(deliveryResultNotice(result)); }); }; @@ -536,8 +556,9 @@ export async function assembleTUISession( onTasksChange: (tasks) => emitter.emit("tasks", tasks), getLiveFleetCount: () => liveFleetCount(subAgentSessions.list()), requestContinuation: () => { + const targetAgent = liveAgent(state); state.enqueueAgentDeliver?.(() => - liveAgent(state).deliver(buildCompactionContinuationMessage()), + targetAgent.deliver(buildCompactionContinuationMessage()), ); }, getProvider: () => state.config, diff --git a/src/tui/runner/state.ts b/src/tui/runner/state.ts index 59e7bb630..f90563657 100644 --- a/src/tui/runner/state.ts +++ b/src/tui/runner/state.ts @@ -8,7 +8,7 @@ * need the same contracts. */ -import type { Agent } from "@intx/agent"; +import type { Agent, SendResult } from "@intx/agent"; import type { ContextStore, InboundMessage, @@ -32,6 +32,7 @@ import type { ProviderFailureAttempt } from "../provider/failure-attempt.js"; import type { ScopedApproval } from "../../permission/admin.js"; import type { ConnectedMcpServer, RunState } from "../../session/state.js"; import type { PendingImageAttachment } from "../image-attachments.js"; +import type { AgentDeliveryResult } from "../deliver-agent-message.js"; import type { SubmitOutcome } from "./submit.js"; import type { mountRunnerHost } from "./host.js"; import { EventEmitter } from "node:events"; @@ -252,7 +253,10 @@ export interface RunnerState { approvalPersistNotice: { notify?: (text: string) => void }; // Late-wired cross-module callbacks, in original wiring order. - enqueueAgentDeliver?: (deliverToLiveAgent: () => void) => void; + enqueueAgentDeliver?: ( + deliverToLiveAgent: () => void, + onSettle?: (result: AgentDeliveryResult) => void, + ) => void; reloadIfIdle?: () => void; systemNotice?: (text: string) => void; currentAttemptIdentity?: () => InferenceAttemptIdentity; @@ -261,7 +265,11 @@ export interface RunnerState { attempt: InferenceAttemptIdentity, providerFailure: ProviderFailureAttempt, ) => void; - sendWithAttemptIdentity?: (message: InboundMessage) => Promise; + sendWithAttemptIdentity?: ( + message: InboundMessage, + send?: (message: InboundMessage) => Promise, + presentFailureNotice?: boolean, + ) => Promise; sendUserPrompt?: ( text: string, pending: readonly PendingImageAttachment[], diff --git a/src/tui/runner/submit.ts b/src/tui/runner/submit.ts index 8d4ee8564..c748f5653 100644 --- a/src/tui/runner/submit.ts +++ b/src/tui/runner/submit.ts @@ -36,14 +36,16 @@ import { createLiveSteerDeliver, routeQueuedDelivery, } from "../queued-delivery.js"; +import type { AgentDeliveryResult } from "../deliver-agent-message.js"; import type { InferenceAttemptIdentity } from "./state.js"; import { tuiSendFailureMessage } from "./send-failure-message.js"; import type { ProviderFailureAttempt } from "../provider/failure-attempt.js"; -import type { Agent } from "@intx/agent"; +import { AgentClosedError, type Agent, type SendResult } from "@intx/agent"; import { ASK_DIRECTOR_WAKE_PREFIX } from "../../subagent/fleet-report.js"; import { MAILBOX_MAIL_WAKE_PREFIX } from "../../subagent/mailbox-mail-drive.js"; import { hostOf, + liveAgent, runWhileAgentBusy, type RunnerServices, type RunnerState, @@ -254,6 +256,7 @@ export function createSubmitPath( err: unknown, attempt: InferenceAttemptIdentity, providerFailure: ProviderFailureAttempt, + presentNotice = true, ): void => { const failure = classifyAgentSendFailure( err, @@ -265,7 +268,7 @@ export function createSubmitPath( if (!shouldSettleUiAfterSendFailure(failure.kind)) return; if (failure.kind === "abort") return; state.runError = err instanceof Error ? err.message : String(err); - if (!providerFailure.presented) { + if (presentNotice && !providerFailure.presented) { systemNotice( tuiSendFailureMessage( err, @@ -283,22 +286,35 @@ export function createSubmitPath( const sendWithAttemptIdentity = async ( message: InboundMessage, - ): Promise => { + send: (message: InboundMessage) => Promise = (next) => + live.agentProxy.send(next), + presentFailureNotice = true, + ): Promise => { const attempt = live.attemptIdentity(); const providerFailure = services.providerFailureAttempts.begin(attempt); try { await runWhileAgentBusy(state, async () => { - const result = await live.agentProxy.send(message); + const result = await send(message); // An ask-tier call parked on the reactor's approval gate settles the // send early; resolve the operator surface here and deliver the // decision on the correlationId signal channel so the parked run // resumes. await services.approvalResume.handle(result); }); - return true; + return { status: "accepted" }; } catch (error) { - handleSendFailure(error, attempt, providerFailure); - return false; + handleSendFailure(error, attempt, providerFailure, presentFailureNotice); + if (error instanceof AgentClosedError) { + return { + status: "not-delivered", + reason: "agent-closed", + detail: error.message, + }; + } + return { + status: "uncertain", + detail: error instanceof Error ? error.message : String(error), + }; } finally { services.providerFailureAttempts.sendSettled(providerFailure); } @@ -390,9 +406,22 @@ export function createDeliverRouting( imageAttachmentFromPath, pending, ), - send: (text, pending) => { + send: async (text, pending) => { state.sendAborted = false; - void state.sendWithAttemptIdentity?.(userInboundMessage(text, pending)); + const send = state.sendWithAttemptIdentity; + if (send === undefined) { + return { + status: "not-delivered", + reason: "session-unavailable", + detail: "session send path is unavailable", + }; + } + const targetAgent = liveAgent(state); + return send( + userInboundMessage(text, pending), + (message) => targetAgent.send(message), + false, + ); }, recordSent: (text) => { if (text.trim().length === 0) return; @@ -420,8 +449,12 @@ export function createDeliverRouting( imageAttachmentFromPath, pending, ), - deliver: (text, pending) => { - live.agentProxy.deliver(userInboundMessage(text, pending)); + deliver: (text, pending, settle) => { + const targetAgent = liveAgent(state); + state.enqueueAgentDeliver?.( + () => targetAgent.deliver(userInboundMessage(text, pending)), + settle, + ); }, captureGeneration: services.deliveryGeneration.capture, onFailure: (error) => diff --git a/src/tui/runtime-bridge.ts b/src/tui/runtime-bridge.ts index 09ef042c3..0e3af6fb0 100644 --- a/src/tui/runtime-bridge.ts +++ b/src/tui/runtime-bridge.ts @@ -30,7 +30,7 @@ import { setShellBridgeHooks, type AppShell, } from "./shell/internals.js"; -import { applyShellInterrupt } from "./shell/prompt.js"; +import { applyShellInterrupt, surfaceSystemNotice } from "./shell/prompt.js"; import { streamRowAt, streamRowCount } from "./shell/transcript.js"; import { rampAnimating } from "./ramp.js"; import { onTurnBoundary } from "../agent/reactor-events.js"; @@ -66,6 +66,11 @@ import { userRowText, type PendingImageAttachment, } from "./image-attachments.js"; +import { + deliveryResultNotice, + type AgentDeliveryResult, +} from "./deliver-agent-message.js"; +import type { DeliverySettle } from "./queued-delivery.js"; import { toolCallRow } from "./diff.js"; import { toolResultRow } from "./mcp-view.js"; import { @@ -141,7 +146,7 @@ export interface SessionPort { /** Hard interrupt current run. */ interrupt: () => void; /** Queue item drained at a tool boundary (or idle). */ - deliver: (item: QueueItem) => void; + deliver: (item: QueueItem, settle?: DeliverySettle) => void; } export type SessionPortHandlers = Partial; @@ -416,6 +421,17 @@ export interface BridgeBag { * `message.received`; without this the transcript shows the message twice. */ pendingEchoes: string[]; + /** + * Exact queue items popped for delivery but not yet settled. Ownership moves + * here from `SessionQueueState.items` so a second boundary cannot redispatch + * them, and recovery can restore the original payload on rejection. + */ + pendingDeliveries: Map; + /** + * Failed deliveries waiting for an empty composer. Never merges into a draft + * the operator is already typing; FIFO after each successful submit clears. + */ + pendingPromptRecoveries: QueueItem[]; /** callId→name / delta bookkeeping for production-shaped events. */ mapCtx: StreamMapContext; disposed: boolean; @@ -563,6 +579,100 @@ function consumeEcho(bag: BridgeBag, text: string): boolean { return true; } +function removeOnePendingEcho(bag: BridgeBag, text: string): void { + const index = bag.pendingEchoes.indexOf(promptContent(text)); + if (index === -1) return; + bag.pendingEchoes.splice(index, 1); +} + +function composerIsEmpty(shell: AppShell): boolean { + return ( + shell.prompt.value.trim().length === 0 && + shell.pendingAttachments.length === 0 + ); +} + +function restoreQueueItemToPrompt(shell: AppShell, item: QueueItem): void { + shell.prompt.value = item.text; + shell.pendingAttachments = [...(item.attachments ?? [])]; + paintChrome(shell); +} + +function markDeliveryRows( + shell: AppShell, + queueItemId: string, + deliveryStatus: "not-delivered" | "uncertain", +): void { + for (let local = shell.streamLog.length - 1; local >= 0; local--) { + const row = shell.streamLog[local]; + if (row?.queueItemId !== queueItemId) continue; + const absolute = shell.streamLogBase + local; + const knownNotDelivered = deliveryStatus === "not-delivered"; + replaceStreamRowAt(shell, absolute, { + ...row, + meta: knownNotDelivered ? "not-delivered" : "delivery-uncertain", + deliveryStatus, + }); + } +} + +function settleDrainedDelivery( + shell: AppShell, + bag: BridgeBag, + item: QueueItem, + result: AgentDeliveryResult, +): void { + if (bag.disposed) return; + if (!bag.pendingDeliveries.has(item.id)) return; + bag.pendingDeliveries.delete(item.id); + if (result.status === "accepted") return; + + removeOnePendingEcho(bag, item.text); + markDeliveryRows( + shell, + item.id, + result.status === "uncertain" ? "uncertain" : "not-delivered", + ); + + let disposition: "restored" | "deferred" = "restored"; + if (composerIsEmpty(shell)) { + restoreQueueItemToPrompt(shell, item); + } else { + bag.pendingPromptRecoveries.push(item); + disposition = "deferred"; + } + surfaceSystemNotice(shell, deliveryResultNotice(result, disposition)); + paintChrome(shell); +} + +function dispatchDrainedItem( + shell: AppShell, + bag: BridgeBag, + item: QueueItem, +): void { + bag.pendingDeliveries.set(item.id, item); + appendStreamRow(shell, { + role: "user", + text: userRowText(item.text, item.attachments ?? []), + // Distinct from the still-pending "steer"/"queue" tag — this row is + // being handed to the run right now. Follow-ups must not say steering. + meta: item.kind === "steer" ? "steering" : "following-up", + queueItemId: item.id, + }); + bag.pendingEchoes.push(item.text.trim()); + bag.port.deliver(item, (result) => { + settleDrainedDelivery(shell, bag, item, result); + }); +} + +function restoreNextPromptRecovery(shell: AppShell, bag: BridgeBag): void { + if (bag.disposed) return; + if (!composerIsEmpty(shell)) return; + const next = bag.pendingPromptRecoveries.shift(); + if (next === undefined) return; + restoreQueueItemToPrompt(shell, next); +} + function messageReceivedContent(event: { readonly data?: unknown; }): string | undefined { @@ -993,15 +1103,7 @@ function drainAtBoundary(shell: AppShell, bag: BridgeBag): void { const { state, item } = drainOne(shell.session); if (!item) break; shell.session = state; - appendStreamRow(shell, { - role: "user", - text: userRowText(item.text, item.attachments ?? []), - // Distinct from the still-pending "steer"/"queue" tag — this row is - // being handed to the run right now. Follow-ups must not say steering. - meta: item.kind === "steer" ? "steering" : "following-up", - }); - bag.pendingEchoes.push(item.text.trim()); - bag.port.deliver(item); + dispatchDrainedItem(shell, bag, item); } paintChrome(shell); } @@ -1016,13 +1118,7 @@ function drainSteersAtBoundary(shell: AppShell, bag: BridgeBag): void { const { state, item } = drainOne(shell.session, "steer"); if (!item) break; shell.session = state; - appendStreamRow(shell, { - role: "user", - text: userRowText(item.text, item.attachments ?? []), - meta: "steering", - }); - bag.pendingEchoes.push(item.text.trim()); - bag.port.deliver(item); + dispatchDrainedItem(shell, bag, item); } paintChrome(shell); } @@ -1211,6 +1307,8 @@ export function attachSessionBridge( port: resolvePort(handlers), openRow: null, pendingEchoes: [], + pendingDeliveries: new Map(), + pendingPromptRecoveries: [], mapCtx: createStreamMapContext(), disposed: false, turn: initialTurnState(now()), @@ -1528,6 +1626,7 @@ export function attachSessionBridge( paintChrome(shell); paintPhase(); bag.port.sendImmediate(t, attachments); + restoreNextPromptRecovery(shell, bag); return; } @@ -1535,6 +1634,7 @@ export function attachSessionBridge( kind === "steer" ? enqueueSteer(shell.session, t, undefined, attachments) : enqueue(shell.session, t, "queue", undefined, attachments); + const queued = shell.session.items[shell.session.items.length - 1]; bag.port.enqueue(t, kind); if (kind === "steer") bag.waitYieldWake?.(); // Show the message itself, not the internal transition ("queue +1 → @@ -1544,8 +1644,10 @@ export function attachSessionBridge( role: "user", text: userRowText(t, attached), meta: kind === "steer" ? "steer" : "queue", + ...(queued !== undefined ? { queueItemId: queued.id } : {}), }); paintChrome(shell); + restoreNextPromptRecovery(shell, bag); }; const sendInternalText = (text: string): void => { @@ -1618,6 +1720,8 @@ export function attachSessionBridge( shell.session = createSessionQueue("idle"); recordLastSent(null); bag.pendingEchoes.length = 0; + bag.pendingDeliveries.clear(); + bag.pendingPromptRecoveries.length = 0; bag.liveFleet = 0; bag.pendingAskWake.clear(); bag.deliveredAskWake.clear(); @@ -1823,6 +1927,8 @@ export function attachSessionBridge( flushOpenRow(shell, bag); bag.disposed = true; recordLastSent(null); + bag.pendingDeliveries.clear(); + bag.pendingPromptRecoveries.length = 0; bag.pendingAskWake.clear(); bag.deliveredAskWake.clear(); bag.flushPendingAskWake = null; diff --git a/src/tui/stream.test.ts b/src/tui/stream.test.ts index a492821aa..375fc0f53 100644 --- a/src/tui/stream.test.ts +++ b/src/tui/stream.test.ts @@ -75,6 +75,25 @@ describe("stream paint", () => { ).not.toContain("steering"); }); + test("delivery settlement prefixes keep the original row text", () => { + expect( + userBody({ + role: "user", + text: "exact body", + meta: "not-delivered", + deliveryStatus: "not-delivered", + })[0], + ).toContain("[not delivered] exact body"); + expect( + userBody({ + role: "user", + text: "exact body", + meta: "delivery-uncertain", + deliveryStatus: "uncertain", + })[0], + ).toContain("[delivery uncertain] exact body"); + }); + test("a long operator message wraps as one left-aligned block", () => { const text = "please find every call site of the legacy token helper and tell me which of them still run in production"; diff --git a/src/tui/stream.ts b/src/tui/stream.ts index c2e5fff9e..307ef26cb 100644 --- a/src/tui/stream.ts +++ b/src/tui/stream.ts @@ -93,6 +93,11 @@ export interface StreamRow { * decides how a cancelled row reads. */ readonly cancelled?: boolean; + /** + * Delivery settlement for a drained queue/steer row. Kept as typed state + * rather than baked into `text` so copy/resume still sees the original body. + */ + readonly deliveryStatus?: "not-delivered" | "uncertain"; /** * Row standing for a run of repeated calls. Its subject stays the call the * run repeats (never a total across them, which would be a claim the @@ -693,17 +698,21 @@ export function paintStreamRow( const prefix = row.cancelled === true ? "[cancelled] " - : row.meta === "steer" - ? "[will steer next] " - : row.meta === "queue" - ? "[will follow up] " - : row.meta === "steering" - ? "[steering] " - : row.meta === "following-up" - ? "[following up] " - : row.meta === "reinject" - ? "[restarted here] " - : ""; + : row.deliveryStatus === "not-delivered" + ? "[not delivered] " + : row.deliveryStatus === "uncertain" + ? "[delivery uncertain] " + : row.meta === "steer" + ? "[will steer next] " + : row.meta === "queue" + ? "[will follow up] " + : row.meta === "steering" + ? "[steering] " + : row.meta === "following-up" + ? "[following up] " + : row.meta === "reinject" + ? "[restarted here] " + : ""; return { content: userBubbleLines(`${prefix}${row.text}`, layout.width).join("\n"), fg,