fix(chat): handle observed terminal errors

Author: tylergibbs1

.changeset/fix-observer-error-frames.md

@@ -0,0 +1,9 @@
+---
+"agents": patch
+"@cloudflare/ai-chat": patch
+"@cloudflare/think": patch
+---
+
+Treat `useAgentChat` observer error frames as terminal responses.
+
+Plain-text error bodies are no longer parsed as stream chunks or merged into an empty assistant message. Error frames now clear observer streaming, replay, recovery, and tool-continuation state even when they omit `done`, matching the transport-owned stream behavior.

packages/agents/src/chat/react.tsx

@@ -2139,10 +2139,10 @@ export function useAgentChat<
               }
             }
 
-            if (data.done || data.replayComplete) {
+            if (data.done || data.error || data.replayComplete) {
               pendingReplayResumeRequestIdsRef.current.delete(data.id);
             }
-            if (data.done) {
+            if (data.done || data.error) {
               if (
                 streamStateRef.current.status === "observing" &&
                 streamStateRef.current.streamId === data.id
@@ -2170,6 +2170,31 @@ export function useAgentChat<
           ) {
             return;
           }
+          if (data.error) {
+            pendingReplayResumeRequestIdsRef.current.delete(data.id);
+            customTransport.handleServerTurnCompleted(data.id);
+            fallbackAckedResumeRequestIdsRef.current.delete(data.id);
+            setIsRecovering(false);
+
+            if (
+              streamStateRef.current.status === "idle" ||
+              streamStateRef.current.streamId === data.id
+            ) {
+              const result = broadcastTransition(streamStateRef.current, {
+                type: "clear"
+              });
+              streamStateRef.current = result.state;
+              setIsServerStreaming(result.isStreaming);
+            }
+            if (observedToolContinuationRequestIdRef.current === data.id) {
+              resetToolContinuation();
+            }
+            break;
+          }
+
+          // Error bodies are human-readable diagnostics, not UI message
+          // chunks. The transport-owned path short-circuits them before JSON
+          // parsing; observers must do the same.
           if (data.body?.trim()) {
             try {
               chunkData = JSON.parse(data.body);

packages/agents/src/react-tests/resume-overlap-race.test.tsx

@@ -124,16 +124,73 @@ function requireChat(chat: AgentChatResult | null): AgentChatResult {
 
 describe("reconnect-driven stream resume", () => {
   let errorSpy: ReturnType<typeof vi.spyOn>;
+  let warnSpy: ReturnType<typeof vi.spyOn>;
 
   beforeEach(() => {
     errorSpy = vi.spyOn(console, "error").mockImplementation(() => {});
+    warnSpy = vi.spyOn(console, "warn").mockImplementation(() => {});
   });
 
   afterEach(() => {
     errorSpy.mockRestore();
+    warnSpy.mockRestore();
     cleanup();
   });
 
+  it("treats observed error frames as terminal without parsing or appending (#1898)", async () => {
+    const { agent, target, sentMessages } = createFakeAgent({
+      name: "observer-error",
+      url: "ws://localhost:3000/agents/chat/observer-error?_pk=abc"
+    });
+
+    function TestComponent() {
+      const chat = useAgentChat({
+        agent,
+        getInitialMessages: null,
+        messages: [] as UIMessage[]
+      });
+      return (
+        <div>
+          <div data-testid="message-count">{chat.messages.length}</div>
+          <div data-testid="streaming">{String(chat.isServerStreaming)}</div>
+        </div>
+      );
+    }
+
+    const { container } = await render(<TestComponent />);
+    await vi.waitFor(() => {
+      expect(countType(sentMessages, RESUME_REQUEST)).toBe(1);
+    });
+    dispatch(target, { type: RESUME_NONE, reason: "idle" });
+
+    // With no transport waiting, this stream is handled by the observer path.
+    dispatch(target, { type: RESUMING, id: "errored-stream" });
+    await vi.waitFor(() => {
+      expect(
+        container.querySelector('[data-testid="streaming"]')?.textContent
+      ).toBe("true");
+    });
+
+    dispatch(target, {
+      type: CHAT_RESPONSE,
+      id: "errored-stream",
+      body: "Network connection lost.",
+      done: true,
+      error: true,
+      replay: true
+    });
+
+    await vi.waitFor(() => {
+      expect(
+        container.querySelector('[data-testid="streaming"]')?.textContent
+      ).toBe("false");
+      expect(
+        container.querySelector('[data-testid="message-count"]')?.textContent
+      ).toBe("0");
+    });
+    expect(warnSpy).not.toHaveBeenCalled();
+  });
+
   it("serializes overlapping resumes and never reads a cleared activeResponse (#1837)", async () => {
     const { agent, target, sentMessages } = createFakeAgent({
       name: "overlap",