Handle terminal error frames in useAgentChat observers

.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,79 @@ 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.each([
+    { label: "with done", done: true },
+    { label: "without done", done: undefined }
+  ])(
+    "treats observed error frames as terminal $label (#1898)",
+    async ({ done }) => {
+      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,
+        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",