diff --git a/packages/chat-ui/src/streaming-reply.test.ts b/packages/chat-ui/src/streaming-reply.test.ts index 4ad61b539..2605a2f7e 100644 --- a/packages/chat-ui/src/streaming-reply.test.ts +++ b/packages/chat-ui/src/streaming-reply.test.ts @@ -217,6 +217,66 @@ describe("nextStreamingReplyState (CL-6376: the typing pulse clears on a dispatc }); }); +describe("nextStreamingReplyState (CL-false-no-reply: rendered content, not a lifecycle event, ends the turn)", () => { + test("a chat.message from the awaiting turn's agent moves straight to replied — the reply already rendered, connector.reply or not", () => { + const state = awaiting("Full answer."); + expect( + nextStreamingReplyState(state, { + eventType: "chat.message", + data: { + id: "msg_1", + sender: { name: null, address: MYRA.address }, + parts: [{ kind: "text", text: "Full answer." }], + }, + }), + ).toEqual({ phase: "replied" }); + }); + + test("a chat.message from the agent's own undelivered-notice address (turnFailed) is never mistaken for a rendered reply", () => { + const state = awaiting(""); + expect( + nextStreamingReplyState(state, { + eventType: "chat.message", + data: { + id: "msg_1", + sender: { name: null, address: MYRA.address }, + parts: [ + { kind: "text", text: "I didn't get that one", turnFailed: true }, + ], + }, + }), + ).toBeNull(); + }); + + test("a chat.message from a human sender never ends the turn — it's not the agent's reply", () => { + const state = awaiting(""); + expect( + nextStreamingReplyState(state, { + eventType: "chat.message", + data: { + id: "msg_1", + sender: { name: null, address: HUMAN.address }, + parts: [{ kind: "text", text: "hi" }], + }, + }), + ).toBe(state); + }); + + test("a chat.message with no content parts never ends the turn", () => { + const state = awaiting(""); + expect( + nextStreamingReplyState(state, { + eventType: "chat.message", + data: { + id: "msg_1", + sender: { name: null, address: MYRA.address }, + parts: [], + }, + }), + ).toBe(state); + }); +}); + describe("nextStreamingReplyState (CL-6432 reopened: a folded run parks after the reply — post-reply tool rounds never re-open the pulse)", () => { test("connector.reply moves the turn to the replied phase — the persisted message takes over the timeline", () => { expect( diff --git a/packages/chat-ui/src/streaming-reply.ts b/packages/chat-ui/src/streaming-reply.ts index fe352d013..99b8bf894 100644 --- a/packages/chat-ui/src/streaming-reply.ts +++ b/packages/chat-ui/src/streaming-reply.ts @@ -23,11 +23,15 @@ import { displayNameFromHandle } from "./timeline"; * - `"awaiting"` — a turn is in flight and its visible reply hasn't posted * yet: an empty `text` renders the typing pulse, streamed tokens render * the growing bubble. - * - `"replied"` — this turn's `connector.reply` has posted. A live folded - * run PARKS after the turn (no `message.run.ended`), and its post-reply - * tool-only rounds (memory writes) still emit `inference.start`/ - * `inference.done` — none of which may re-open the pulse. Renders - * nothing; only a genuinely new turn leaves it. + * - `"replied"` — this turn's reply has rendered: either its `chat.agent` + * stream carried `connector.reply`, or (CL-false-no-reply) its `chat.message` + * posted straight from the reply pipeline while awaiting — whichever is + * observed first, since a parked folded run's `chat.agent` stream may + * never carry the former at all. A live folded run PARKS after the turn + * (no `message.run.ended`), and its post-reply tool-only rounds (memory + * writes) still emit `inference.start`/`inference.done` — none of which + * may re-open the pulse. Renders nothing; only a genuinely new turn + * leaves it. * - `null` — idle, no turn in flight. */ export type StreamingReplyState = @@ -98,6 +102,29 @@ function hasTurnFailedPart(data: unknown): boolean { ); } +/** Whether a `chat.message` payload is the awaiting turn's own reply + * actually rendering on screen — the one observable fact this module can + * trust over any `chat.agent` lifecycle event. `postReply` + * (`chat-orchestrator.ts`) posts this event from its own dispatch + * pipeline, entirely separate from the raw `connector.reply` event + * `chat.agent` otherwise carries — so a dropped, delayed, or (parked + * folded run) never-sent `connector.reply` can no longer leave the + * backstop armed once the reply the reader is looking at has already + * posted. Requires an agent sender (never the reader's own echoed + * message) and at least one part; `postUndeliveredNotice` also posts from + * the agent's own address with a real text part, so `hasTurnFailedPart` + * rules that one out explicitly rather than by accident. */ +function isRenderedAgentReply(data: unknown): boolean { + if (hasTurnFailedPart(data)) return false; + if (typeof data !== "object" || data === null) return false; + const sender = (data as Record).sender; + if (typeof sender !== "object" || sender === null) return false; + const address = (sender as Record).address; + if (typeof address !== "string" || !isAgentAddress(address)) return false; + const parts = (data as Record).parts; + return Array.isArray(parts) && parts.length > 0; +} + /** * The streaming reply's whole state machine, pure and turn-phase aware * (CL-6432 reopened). `message.run.started` — the harness's per-dequeued- @@ -107,12 +134,17 @@ function hasTurnFailedPart(data: unknown): boolean { * open the empty pulse (never wiping streamed tokens), each * `inference.text.delta` replaces the text with its cumulative snapshot, * and a textless `inference.done` keeps the pulse up across pre-reply tool - * rounds. `connector.reply` — the event the orchestrator posts the - * persisted reply off — moves the turn to `"replied"`: a live folded run - * PARKS here (no `message.run.ended`), and its post-reply tool-only rounds - * (memory writes) still emit `inference.start`/`inference.done`, so in - * `"replied"` every inference/reactor event is inert rather than - * re-opening the pulse. The hard-terminal events — + * rounds. Two independent signals move the turn to `"replied"`: + * `connector.reply` on the `chat.agent` stream (the event the orchestrator + * posts the persisted reply off), and — CL-false-no-reply, since a parked + * folded run's `chat.agent` stream may never carry that event at all — an + * awaiting turn's own `chat.message` actually rendering the agent's reply + * (see `isRenderedAgentReply`). Whichever arrives first wins; the other is + * then a no-op against an already-`"replied"` turn. A live folded run + * PARKS after replying (no `message.run.ended`), and its post-reply + * tool-only rounds (memory writes) still emit `inference.start`/ + * `inference.done`, so in `"replied"` every inference/reactor event is + * inert rather than re-opening the pulse. The hard-terminal events — * `reactor.done`/`reactor.error`, `message.run.ended`, `inference.error`, * and a `chat.message` carrying `postUndeliveredNotice`'s `turnFailed` * part (see `hasTurnFailedPart`, the one failure path with no `chat.agent` @@ -124,7 +156,15 @@ export function nextStreamingReplyState( event: { readonly eventType: string; readonly data: unknown }, ): StreamingReplyState { if (event.eventType === "chat.message") { - return hasTurnFailedPart(event.data) ? null : current; + if (hasTurnFailedPart(event.data)) return null; + if ( + current !== null && + current.phase === "awaiting" && + isRenderedAgentReply(event.data) + ) { + return REPLIED; + } + return current; } if (event.eventType !== "chat.agent") return current; diff --git a/packages/chat-ui/test/use-streaming-reply.test.tsx b/packages/chat-ui/test/use-streaming-reply.test.tsx index eb174ed7f..7bc9efac8 100644 --- a/packages/chat-ui/test/use-streaming-reply.test.tsx +++ b/packages/chat-ui/test/use-streaming-reply.test.tsx @@ -215,6 +215,71 @@ describe("useStreamingReply's reply-timeout backstop (CL-6252 #6)", () => { }); }); +describe("useStreamingReply (CL-false-no-reply: the notice must never fire once a reply has rendered)", () => { + const MYRA_ADDRESS = "myra@agents.example"; + + test("a full reply renders and the run then goes quiet with no terminal event (parked folded run) — no notice", async () => { + const harness = mount("chan_a", 30); + harness.awaitReply(); + harness.send("chat.agent", delta("Full answer.")); + // The persisted reply posts as its own chat.message — the real signal + // the reader sees on screen — entirely independent of whichever + // chat.agent event (or none, if this run parks) follows it. + harness.send("chat.message", { + id: "msg_1", + sender: { name: null, address: MYRA_ADDRESS }, + parts: [{ kind: "text", text: "Full answer." }], + }); + expect(harness.get()).toEqual({ phase: "replied" }); + + // No message.run.ended, no connector.reply — mimics a folded run that + // parks instead of ending. The backstop must not have armed for a + // "replied" phase, so it must never fire. + await harness.settle(60); + expect(harness.timedOut()).toBe(false); + harness.unmount(); + }); + + test("a full reply renders and then a post-reply tool round runs — no notice", async () => { + const harness = mount("chan_a", 30); + harness.awaitReply(); + harness.send("chat.agent", delta("Full answer.")); + harness.send("chat.message", { + id: "msg_1", + sender: { name: null, address: MYRA_ADDRESS }, + parts: [{ kind: "text", text: "Full answer." }], + }); + + // A memory-write tool round after the reply — must stay inert. + harness.send("chat.agent", { + type: "inference.start", + seq: 10, + data: { model: "x" }, + }); + harness.send("chat.agent", { + type: "inference.done", + seq: 11, + data: { turn: {}, usage: {}, source: "primary" }, + }); + + await harness.settle(60); + expect(harness.get()).toEqual({ phase: "replied" }); + expect(harness.timedOut()).toBe(false); + harness.unmount(); + }); + + test("a turn that never produces any content still times out after the backstop window", async () => { + const harness = mount("chan_a", 30); + harness.awaitReply(); + expect(harness.get()).toEqual({ phase: "awaiting", text: "" }); + + await harness.settle(60); + expect(harness.get()).toBeNull(); + expect(harness.timedOut()).toBe(true); + harness.unmount(); + }); +}); + describe("useStreamingReply.resumeFromTurn (CL-6380: catch-up on remount)", () => { test("hydrates the reply from a running turn's committed text on a fresh mount", () => { const harness = mount("chan_a");