fix(chat): serialize reconnect-driven resumes to stop activeResponse race (#1837)

.changeset/fix-resume-overlap-race.md

@@ -0,0 +1,7 @@
+---
+"agents": patch
+---
+
+Fix reconnect-driven resume overlap throwing `Cannot read properties of undefined (reading 'state')` in `useAgentChat` (#1837).
+
+With `resume: true` (the default), the hook re-probes the stream from its WebSocket `onAgentOpen` handler on every reconnect. The AI SDK's `Chat.makeRequest` has no concurrency guard — every resume shares the single mutable `this.activeResponse`, and its `finally` finalizer reads `this.activeResponse.state.message` with a bare (unguarded) read before clearing it. Under a reconnect storm (flaky mobile link, or a Durable Object bounce on redeploy), a second resume could overwrite + clear `activeResponse` before an earlier resume's finalizer ran, so the earlier finalizer read `undefined` and threw. The old guard didn't close the window: `isAwaitingResume()` only covers the handshake (it flips false the instant `STREAM_RESUMING` resolves, before the AI SDK sets status to `submitted` in a later microtask) and `statusRef` is lagging React state. Resumes are now serialized via an in-flight flag, so a re-probe `resumeStream()` is never issued while one is still outstanding.

packages/agents/src/chat/react.tsx

@@ -1054,6 +1054,25 @@
   // in the socket effect below (#1784).
   const hasConnectedOnceRef = useRef(false);
 
+  // True while a reconnect re-probe `resumeStream()` we issued is still in
+  // flight. AI SDK's `Chat.makeRequest` has no concurrency guard — every
+  // resume shares the single mutable `this.activeResponse`, and its `finally`
+  // finalizer reads `this.activeResponse.state` with a bare (unguarded) read
+  // before clearing it. Overlapping reconnect-driven resumes therefore race:
+  // a later resume clears `activeResponse` before an earlier one's finalizer
+  // runs, throwing `Cannot read properties of undefined (reading 'state')`
+  // (#1837). The `onAgentOpen` guard's `statusRef` is lagging React state and
+  // `isAwaitingResume()` only covers the handshake, so neither closes the
+  // post-handshake / pre-status-propagation window. Serialize resumes here:
+  // never issue a new re-probe `resumeStream()` while one is outstanding.
+  const resumeInFlightRef = useRef(false);
+  // Generation token for the in-flight resume. Bumped whenever the gate is
+  // force-reset (socket-effect cleanup, e.g. an agent swap on `_pk` change).
+  // A resume's `.finally` captures its generation and only clears the flag if
+  // it still matches — so a stale, orphaned resume settling after a newer one
+  // has taken over can't reopen the gate underneath it.
+  const resumeGenerationRef = useRef(0);
+
   const resumingToolContinuationRef = useRef(false);
   const pendingToolContinuationRef = useRef(false);
   const observedToolContinuationRequestIdRef = useRef<string | null>(null);
@@ -2191,6 +2210,17 @@
     // A closed socket invalidates the per-socket resume-ACK dedupe: after a
     // reconnect the server sees a brand-new connection and must be ACKed
     // (and replay) again, so the previous entries must not suppress it.
+    //
+    // NOTE: deliberately does NOT reset `resumeInFlightRef` (#1837). The flag
+    // is owned by the in-flight resume and cleared only by its own `.finally`
+    // (or invalidated via the generation bump in this effect's cleanup).
+    // Clearing it here would set it false while the resume may still be
+    // mid-flight (post-handshake, before `makeRequest`'s finalizer runs),
+    // re-coupling correctness to close/open task ordering and reopening the
+    // overlap window. It is also unnecessary: a handshake-phase drop is already
+    // gated by `isAwaitingResume()` (and the pending promise's resolver re-binds
+    // to the new socket), and a streaming-phase drop closes the resume stream,
+    // which settles `makeRequest` and clears the flag via its `.finally`.
     function onAgentClose() {
       fallbackAckedResumeRequestIds.clear();
     }
@@ -2212,11 +2242,26 @@
         !resume ||
         statusRef.current !== "ready" ||
         resumingToolContinuationRef.current ||
+        resumeInFlightRef.current ||
         customTransport.isAwaitingResume()
       ) {
         return;
       }
-      void Promise.resolve(resumeStreamRef.current?.()).catch(() => {});
+      const resumeStreamFn = resumeStreamRef.current;
+      if (!resumeStreamFn) return;
+      // Serialize resumes (#1837): hold the in-flight flag for the entire
+      // resume lifetime so a rapid second `open` (reconnect storm) can't
+      // overwrite the AI SDK's shared `activeResponse` mid-flight.
+      resumeInFlightRef.current = true;
+      const myGeneration = resumeGenerationRef.current;
+      void Promise.resolve(resumeStreamFn())
+        .catch(() => {})
+        .finally(() => {
+          // Skip if the gate was force-reset (agent swap) and a newer resume
+          // has since taken over — clearing here would reopen it spuriously.
+          if (resumeGenerationRef.current !== myGeneration) return;
+          resumeInFlightRef.current = false;
+        });
     }
 
     agent.addEventListener("message", onAgentMessage);
@@ -2238,6 +2283,12 @@
       setIsRecovering(false);
       protectedStreamingAssistantRef.current = null;
       localResponseIds.clear();
+      // Don't let an orphaned re-probe (e.g. agent swapped on `_pk` change)
+      // leave the serialization gate stuck closed for the next socket (#1837).
+      // Bump the generation so the orphaned resume's late `.finally` can't
+      // clear the gate out from under a newer resume on the next socket.
+      resumeGenerationRef.current++;
+      resumeInFlightRef.current = false;
     };
   }, [
     agent,

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

@@ -0,0 +1,181 @@
+/**
+ * Regression test for #1837 — reconnect-driven resume overlap.
+ *
+ * With `resume: true` (the default), `useAgentChat` re-probes the stream from
+ * its WebSocket `onAgentOpen` handler on every reconnect. The AI SDK's
+ * `Chat.makeRequest` has NO concurrency guard: every resume shares the single
+ * mutable `this.activeResponse`, and its `finally` finalizer reads
+ * `this.activeResponse.state.message` with a BARE (unguarded) read before
+ * clearing it. If a second resume overwrites + clears `activeResponse` before
+ * an earlier resume's `finally` runs, the earlier finalizer reads `undefined`
+ * and throws `TypeError: Cannot read properties of undefined (reading 'state')`
+ * (caught + logged inside makeRequest's finally try/catch).
+ *
+ * The old `onAgentOpen` guard (`statusRef.current === "ready"` &&
+ * `!isAwaitingResume()`) did not close the window, because:
+ *   - `isAwaitingResume()` flips false the instant the resume handshake
+ *     resolves (STREAM_RESUMING), but the AI SDK only sets status to
+ *     "submitted" in a *later microtask* (behind `await reconnectToStream`), and
+ *   - `statusRef.current` is lagging React state that hasn't re-rendered yet.
+ *
+ * So a socket `open` landing in that window (a reconnect storm: flaky mobile
+ * link / DO bounce on redeploy) sailed past both guards and launched an
+ * overlapping resume. The fix serializes resumes via `resumeInFlightRef`: no
+ * new re-probe `resumeStream()` is issued while one is still outstanding.
+ *
+ * This drives the REAL hook through a fake `EventTarget` agent (no Worker
+ * needed) using exactly the frames a reconnect storm produces.
+ */
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import { render as _render, cleanup } from "vitest-browser-react";
+import type { UIMessage } from "ai";
+import type { useAgent } from "../react";
+import { useAgentChat } from "../chat/react";
+
+// Async WebSocket-driven updates legitimately land outside act() here; disable
+// the act environment after mount (mirrors the other react-tests in this dir).
+const render: typeof _render = async (...args) => {
+  const result = await _render(...args);
+  // @ts-expect-error - globalThis is not typed
+  globalThis.IS_REACT_ACT_ENVIRONMENT = false;
+  return result;
+};
+
+function sleep(ms: number) {
+  return new Promise((resolve) => setTimeout(resolve, ms));
+}
+
+function createFakeAgent({ name, url }: { name: string; url: string }) {
+  const target = new EventTarget();
+  const sentMessages: string[] = [];
+  const agent = {
+    _pkurl: url,
+    _pk: name,
+    _url: null as string | null,
+    addEventListener: target.addEventListener.bind(target),
+    agent: "Chat",
+    close: () => {},
+    id: "fake-agent",
+    name,
+    removeEventListener: target.removeEventListener.bind(target),
+    send: (data: string) => sentMessages.push(data),
+    dispatchEvent: target.dispatchEvent.bind(target),
+    path: [{ agent: "Chat", name }],
+    getHttpUrl: () =>
+      url.replace("ws://", "http://").replace("wss://", "https://")
+  };
+  return {
+    agent: agent as unknown as ReturnType<typeof useAgent>,
+    target,
+    sentMessages
+  };
+}
+
+function dispatch(target: EventTarget, data: Record<string, unknown>) {
+  target.dispatchEvent(
+    new MessageEvent("message", { data: JSON.stringify(data) })
+  );
+}
+
+function open(target: EventTarget) {
+  target.dispatchEvent(new Event("open"));
+}
+
+const RESUMING = "cf_agent_stream_resuming";
+const RESUME_NONE = "cf_agent_stream_resume_none";
+const RESUME_REQUEST = "cf_agent_stream_resume_request";
+const CHAT_RESPONSE = "cf_agent_use_chat_response";
+
+function countType(sentMessages: string[], type: string): number {
+  return sentMessages
+    .map((m) => {
+      try {
+        return JSON.parse(m) as { type?: string };
+      } catch {
+        return {};
+      }
+    })
+    .filter((m) => m.type === type).length;
+}
+
+describe("reconnect-driven resume overlap (#1837)", () => {
+  let errorSpy: ReturnType<typeof vi.spyOn>;
+
+  beforeEach(() => {
+    errorSpy = vi.spyOn(console, "error").mockImplementation(() => {});
+  });
+
+  afterEach(() => {
+    errorSpy.mockRestore();
+    cleanup();
+  });
+
+  it("serializes overlapping resumes and never reads a cleared activeResponse", async () => {
+    const { agent, target, sentMessages } = createFakeAgent({
+      name: "overlap",
+      url: "ws://localhost:3000/agents/chat/overlap?_pk=abc"
+    });
+
+    function TestComponent() {
+      useAgentChat({
+        agent,
+        getInitialMessages: null,
+        messages: [] as UIMessage[]
+      });
+      return <div data-testid="ok">ok</div>;
+    }
+
+    await render(<TestComponent />);
+    await sleep(10);
+
+    // Settle the AI SDK's mount-time resume so the chat is at "ready" and the
+    // transport is no longer awaiting a resume (isAwaitingResume() === false).
+    dispatch(target, { type: RESUME_NONE });
+    await sleep(10);
+
+    // First "open" is consumed by the hook's first-open gate (hasConnectedOnce).
+    open(target);
+    await sleep(10);
+
+    // A real reconnect: the re-probe fires resume A (sends a RESUME_REQUEST).
+    open(target);
+    await sleep(10);
+
+    // Server announces resume A. The handshake resolves SYNCHRONOUSLY here, so
+    // isAwaitingResume() flips false immediately — but the AI SDK only sets
+    // status to "submitted" in a *later microtask*. A second "open" dispatched
+    // in the same synchronous turn therefore sees statusRef === "ready" AND
+    // isAwaitingResume() === false. Pre-fix this launched an overlapping resume
+    // B; post-fix `resumeInFlightRef` suppresses it.
+    const requestsBeforeOverlap = countType(sentMessages, RESUME_REQUEST);
+    dispatch(target, { type: RESUMING, id: "s1" });
+    open(target); // overlapping reconnect — no await before this
+    await sleep(10);
+    const requestsAfterOverlap = countType(sentMessages, RESUME_REQUEST);
+
+    // Server announces a second stream id (the would-be overlapping resume B).
+    dispatch(target, { type: RESUMING, id: "s2" });
+    await sleep(10);
+
+    // B settles first and clears the shared activeResponse...
+    dispatch(target, { type: CHAT_RESPONSE, id: "s2", body: "", done: true });
+    await sleep(10);
+
+    // ...then A's finalizer runs. Pre-fix it read the now-undefined
+    // activeResponse and threw.
+    dispatch(target, { type: CHAT_RESPONSE, id: "s1", body: "", done: true });
+    await sleep(10);
+
+    const captured = (errorSpy.mock.calls.flat() as unknown[]).map(
+      (a: unknown) => (a instanceof Error ? a.message : String(a))
+    );
+    const sawStateTypeError = captured.some((m: string) =>
+      /reading 'state'|Cannot read properties of undefined/.test(m)
+    );
+
+    // The overlapping reconnect must not issue a second resume...
+    expect(requestsAfterOverlap).toBe(requestsBeforeOverlap);
+    // ...and the AI SDK finalizer must never read a cleared activeResponse.
+    expect(sawStateTypeError).toBe(false);
+  });
+});