fix(chat): settle turns correctly across replay, resume, and observers

.changeset/chat-terminal-frame-fixes.md

@@ -0,0 +1,16 @@
+---
+"agents": patch
+"@cloudflare/ai-chat": patch
+---
+
+Fix several chat turn-settlement edge cases in the wire protocol and `useAgentChat`:
+
+- Replayed terminals now carry the stream's `outcome` (`aborted` for orphaned or aborted streams, `recovering` when stall recovery is scheduled), and the outcome survives the stream row being cleaned up.
+- A resume ACK that lands while the response is still being persisted replays the held stream instead of sending an early bare done, including a stream closed for recovery or by an error.
+- The "No response" done is sent to every connection, including the one that sent the message.
+- Observers settle on a `done` for a request other than the one they are watching, and a `recovering` done no longer clears `isRecovering`.
+- `onToolCall` stays held after a `recovering` close until live frames for a later request arrive, recovery ends, or a reconnect finds the server idle with no recovery in progress. Replayed frames no longer release it.
+- A held `onToolCall` is released when a reconnect finds the held turn over. With `resume: false`, it moves to a stream the server offers, and is released when that stream's terminal frame arrives.
+- `onTurnEnd` fires after `status` settles for the tab's own requests, so it can call `sendMessage`.
+- Live chunks no longer merge into a message once divergence is detected.
+- Remembered turn errors are capped and cleared by `clearHistory`.

.changeset/think-recovering-stream-outcome.md

@@ -0,0 +1,7 @@
+---
+"@cloudflare/think": patch
+---
+
+Record the close outcome on Think's chat streams, so a resume that replays a stream closed for a scheduled recovery reports `recovering` (and an aborted stream reports `aborted`) instead of `completed`.
+
+A connection that was offered a stream but never acknowledged it (for example `useAgentChat` with `resume: false`) now gets the stream's terminal frame and later broadcasts on every path, including `chat()` RPC turns.

packages/agents/src/chat/__tests__/broadcast-state.test.ts

@@ -656,6 +656,29 @@ describe("broadcast stream state machine", () => {
     );
   });
 
+  it("a live chunk after a healing snapshot keeps the stored copy", () => {
+    // chunk → snapshot heals the diverged copy → chunk → done
+    let state = observe(["You", " canYou", " see can"]);
+    let messages = stored("You can see");
+    const chunk = transition(state, {
+      type: "response",
+      streamId: "s-heal",
+      messageId: "tmp",
+      chunkData: { type: "text-delta", id: "t1", delta: " more" }
+    });
+    state = chunk.state;
+    messages = chunk.messagesUpdate!(messages);
+    expect(assistantText(messages)).toBe("You can see");
+
+    const done = transition(state, {
+      type: "response",
+      streamId: "s-heal",
+      messageId: "tmp",
+      done: true
+    });
+    expect(assistantText(done.messagesUpdate!(messages))).toBe("You can see");
+  });
+
   it("done still merges an accumulator that extends the stored copy", () => {
     const state = observe(["You", " can", " see"]);
     const done = transition(state, {

packages/agents/src/chat/__tests__/resume-handshake.test.ts

@@ -9,6 +9,7 @@ import { ContinuationState } from "../continuation-state";
 import type { ResumableStream } from "../resumable-stream";
 import { PreStreamTurns } from "../pre-stream-turns";
 import { STREAM_RESUME_NONE_REASONS } from "../protocol";
+import type { ChatTurnOutcome } from "../wire-types";
 import {
   replayDoneFrame,
   streamPendingFrame,
@@ -54,7 +55,12 @@ interface FakeStreamState {
   replayCompletedReturn: boolean;
   replayErroredReturn: boolean;
   originMessageIds?: string[];
-  calls: { replayChunks: string[]; replayErrored: string[] };
+  outcome?: ChatTurnOutcome;
+  calls: {
+    replayChunks: string[];
+    replayErrored: string[];
+    replayPending: string[];
+  };
 }
 
 function makeStream(over: Partial<FakeStreamState> = {}): {
@@ -68,7 +74,7 @@ function makeStream(over: Partial<FakeStreamState> = {}): {
     orphanedStreamId: "",
     replayCompletedReturn: false,
     replayErroredReturn: true,
-    calls: { replayChunks: [], replayErrored: [] },
+    calls: { replayChunks: [], replayErrored: [], replayPending: [] },
     ...over
   };
   const resumableStream = {
@@ -85,9 +91,14 @@ function makeStream(over: Partial<FakeStreamState> = {}): {
     },
     replayCompletedChunksByRequestId: () => state.replayCompletedReturn,
     getOriginMessageIds: () => state.originMessageIds,
+    getOutcome: () => state.outcome,
     replayErroredChunksByRequestId: (_c: Connection, requestId: string) => {
       state.calls.replayErrored.push(requestId);
       return state.replayErroredReturn;
+    },
+    replayClosedStreamChunks: (_c: Connection, requestId: string) => {
+      state.calls.replayPending.push(requestId);
+      return true;
     }
   } as unknown as ResumableStream;
   return { state, resumableStream };
@@ -101,8 +112,12 @@ function makeHost(opts: {
   pendingResumeConnections?: Set<string>;
   persistCalls?: string[];
   presentConnectionIds?: Set<string>;
+  heldTerminalRequestIds?: Set<string>;
 }): ResumeHandshakeHost {
   return {
+    ...(opts.heldTerminalRequestIds && {
+      holdsTerminalFrames: (id: string) => opts.heldTerminalRequestIds!.has(id)
+    }),
     responseMessageType: RESPONSE_TYPE,
     resumableStream: opts.resumableStream,
     continuation: opts.continuation ?? new ContinuationState<Connection>(),
@@ -523,4 +538,42 @@ describe("ResumeHandshake (driver → golden frames)", () => {
 
     expect(frames).toEqual([replayDoneFrame("req-done", RESPONSE_TYPE)]);
   });
+
+  it("ACK with nothing left to replay echoes the stream's outcome", async () => {
+    const frames: SentFrame[] = [];
+    const { resumableStream } = makeStream({
+      active: false,
+      replayCompletedReturn: false,
+      outcome: "aborted"
+    });
+    const handshake = new ResumeHandshake(makeHost({ resumableStream }));
+
+    await handshake.handleResumeAck(makeConnection("c1", frames), "req-done");
+
+    expect(frames).toEqual([
+      { ...replayDoneFrame("req-done", RESPONSE_TYPE), outcome: "aborted" }
+    ]);
+  });
+
+  it("ACK while the host holds the request's terminal frames sends no early done", async () => {
+    const frames: SentFrame[] = [];
+    const { state, resumableStream } = makeStream({
+      active: false,
+      replayCompletedReturn: false
+    });
+    const handshake = new ResumeHandshake(
+      makeHost({
+        resumableStream,
+        heldTerminalRequestIds: new Set(["req-persisting"])
+      })
+    );
+
+    await handshake.handleResumeAck(
+      makeConnection("c1", frames),
+      "req-persisting"
+    );
+
+    expect(frames).toEqual([]);
+    expect(state.calls.replayPending).toEqual(["req-persisting"]);
+  });
 });

packages/agents/src/chat/broadcast-state.ts

@@ -142,20 +142,21 @@ export function transition(
 
       let messagesUpdate: ((prev: UIMessage[]) => UIMessage[]) | undefined;
 
+      const mergeUnlessDiverged = (prev: UIMessage[]) =>
+        observedDivergesFrom(accumulator, prev)
+          ? prev
+          : accumulator.mergeInto(prev);
+
       if (event.done) {
-        messagesUpdate = (prev) =>
-          observedDivergesFrom(accumulator, prev)
-            ? prev
-            : accumulator.mergeInto(prev);
         return {
           state: { status: "idle" },
-          messagesUpdate,
+          messagesUpdate: mergeUnlessDiverged,
           isStreaming: false
         };
       }
 
       if (event.chunkData && !event.replay) {
-        messagesUpdate = (prev) => accumulator.mergeInto(prev);
+        messagesUpdate = mergeUnlessDiverged;
       } else if (event.replayComplete) {
         messagesUpdate = (prev) => accumulator.mergeInto(prev);
       }