fix(frontend): rejoin active runs after reopening chats (#5536)

* fix(frontend): rejoin active runs after reopening chats

* test(frontend): mock thread runs query in stream options test

* fix(frontend): avoid rejoining completed runs from stale cache

* fix(frontend): tighten active run recovery cleanup

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

frontend/src/AGENTS.md

@@ -141,6 +141,7 @@ Array previews coalesce consecutive generated markers only at the end into one o
 - **Thread routes** — construct Web UI chat paths through `core/threads/utils.ts::pathOfThread()`, which percent-encodes both custom agent names and thread IDs before inserting them into route segments
 - **LangGraph client** is a singleton obtained via `getAPIClient()` in `core/api/`
 - **Run stream options** are sanitized by `core/api/stream-mode.ts`: the Gateway-supported set is `values`, `messages-tuple`, `updates`, `debug`, `tasks`, `checkpoints`, and `custom`; any request containing an unsupported mode throws before HTTP instead of being partially forwarded or silently defaulting to `values`. `streamResumable` is retained by thread hooks only for SDK-side reconnect bookkeeping but stripped before the HTTP request because the Gateway does not accept that request option; actual replay uses the SSE `Last-Event-ID` cursor. The main chat client's initial and rejoined streams are forced to incremental `messages-tuple`, `updates`, and `custom` modes so SDK lazy tracking cannot add repeated full-state `values` snapshots; explicitly requested non-snapshot modes such as `debug`, `tasks`, and `checkpoints` are preserved. `values` remains supported outside this chat wrapper for explicit state inspection and by the replay-gap durable-state reload below. The backend file-tool chunk batcher is keyed to `messages-tuple`, not `values`, so omitting snapshots must not regress `write_file` / `str_replace` streaming into one SSE frame per model token. `core/threads/stream-state.ts` folds the user-visible non-message fields from `updates` with the matching DeerFlow reducer semantics and rejects irrelevant frames before calling the SDK mutator, while the SDK's `messages-tuple` manager remains the sole owner of live message chunk assembly and deduplication. Keep this boundary aligned with the backend request schema; `messages` and `events` are not supported and must not be forwarded.
+- **Active-run recovery** in `core/threads/hooks.ts` discovers pending/running runs when a reopened chat has no tab-local reconnect pointer. Remember SDK-completed run IDs for the mounted hook, including submitted and natively reconnected runs: stale or delayed runs reads must not rejoin a completed run and interrupt follow-up suggestions. Keep recovery retries bounded and continue allowing a different active run to reconnect.
 - **SSE replay gaps** are handled in `core/api/api-client.ts`, which wraps both initial and joined run streams because the upstream SDK ignores unknown event names. An id-less backend `gap` control frame clears stale reconnect metadata, emits an internal `stream_replay_gap` custom event, reloads durable thread values, and resumes after the server-provided retained tail when one exists (or rejoins without a cursor if the buffer is empty), with up to five recovery rejoins after the original stream (six total stream calls on an all-gap exhaustion path). The wrapper remains a lazy async iterable because the SDK consumes it with `for await`. `core/threads/hooks.ts` clears optimistic/transient/subtask state, invalidates durable history caches, and shows the localized recovery warning; never let a gap fall through as a normal stream finish or cancel the still-running backend run.
 - **Streaming Markdown rendering** is owned by `core/streamdown`: Streamdown's `animated` / `isAnimating` API handles incremental word animation, while the shared `streamdownRenderingPlugins` config registers the named code-highlighting and Mermaid plugins required by Streamdown 2.5. Keep wrappers and derived configs wired to that shared object; do not reintroduce a rehype plugin that wraps every word, because reparsing a growing block remounts old words and replays their animation.
 - Citation links in message and artifact Markdown must derive their `citation:` label from the full `ReactNode` children tree, since Streamdown may provide element or array children during streaming rather than a plain string.

frontend/src/core/threads/hooks.ts

@@ -268,6 +268,61 @@ const EMPTY_MESSAGES: Message[] = [];
 const EMPTY_RUN_MESSAGES: RunMessage[] = [];
 const EMPTY_MESSAGE_IDENTITIES: readonly string[] = [];
 const EMPTY_MESSAGE_IDENTITIES_SET: ReadonlySet<string> = new Set<string>();
+const ACTIVE_RUN_STATUSES = new Set(["pending", "running"]);
+const ACTIVE_RUN_REJOIN_RETRY_DELAYS_MS = [1_000, 2_000] as const;
+const MAX_ACTIVE_RUN_REJOIN_ATTEMPTS =
+  ACTIVE_RUN_REJOIN_RETRY_DELAYS_MS.length + 1;
+
+type ActiveRunRejoinState = {
+  attempts: number;
+  inFlight: boolean;
+  retryTimer: ReturnType<typeof setTimeout> | null;
+  runId: string | null;
+  settled: boolean;
+  threadId: string | null;
+};
+
+function createActiveRunRejoinState(
+  threadId: string | null = null,
+  runId: string | null = null,
+): ActiveRunRejoinState {
+  return {
+    attempts: 0,
+    inFlight: false,
+    retryTimer: null,
+    runId,
+    settled: false,
+    threadId,
+  };
+}
+
+function readReconnectRun(threadId: string): string | null {
+  try {
+    return window.sessionStorage.getItem(`lg:stream:${threadId}`);
+  } catch {
+    return null;
+  }
+}
+
+function rememberReconnectRun(threadId: string, runId: string): void {
+  try {
+    window.sessionStorage.setItem(`lg:stream:${threadId}`, runId);
+  } catch {
+    // The stream can still be joined, but SDK stop cannot cancel it without
+    // the tab-local run pointer.
+  }
+}
+
+function clearReconnectRun(threadId: string, runId: string): void {
+  try {
+    const key = `lg:stream:${threadId}`;
+    if (window.sessionStorage.getItem(key) === runId) {
+      window.sessionStorage.removeItem(key);
+    }
+  } catch {
+    // Storage access is best-effort and must never block stream cleanup.
+  }
+}
 /**
  * The turn this client submitted, recorded at dispatch time. The visible human
  * input gets one client-generated identity shared by the optimistic display
@@ -1784,6 +1839,23 @@ export function useThreadStream({
     enabled: !isMock,
     pendingSupersededRunIds,
   });
+  const runsQuery = useThreadRuns(onStreamThreadId ?? undefined, {
+    enabled: !isMock,
+  });
+  const activeRunId = useMemo(
+    () =>
+      runsQuery.data?.find((run) => ACTIVE_RUN_STATUSES.has(String(run.status)))
+        ?.run_id,
+    [runsQuery.data],
+  );
+  const activeRunRejoinRef = useRef<ActiveRunRejoinState>(
+    createActiveRunRejoinState(),
+  );
+  const [activeRunRejoinRetry, setActiveRunRejoinRetry] = useState(0);
+  // Runs reads can lag behind SDK completion, including the initial read.
+  // Keep completed IDs across recovery-state resets so stale "running" data
+  // cannot restart a submitted or natively reconnected stream.
+  const completedRunIdsRef = useRef(new Set<string>());
 
   // Keep listeners ref updated with latest callbacks
   useEffect(() => {
@@ -1837,6 +1909,38 @@ export function useThreadStream({
   const { tasksRef, setTasks } = useSubtaskContext();
   const updateSubtask = useUpdateSubtask();
 
+  const scheduleActiveRunRejoinRetry = useCallback(() => {
+    const rejoin = activeRunRejoinRef.current;
+    if (!rejoin.inFlight || !rejoin.threadId || !rejoin.runId) {
+      return;
+    }
+
+    rejoin.inFlight = false;
+    clearReconnectRun(rejoin.threadId, rejoin.runId);
+    const retryDelay = ACTIVE_RUN_REJOIN_RETRY_DELAYS_MS[rejoin.attempts - 1];
+    if (retryDelay === undefined) {
+      return;
+    }
+
+    rejoin.retryTimer = setTimeout(() => {
+      rejoin.retryTimer = null;
+      setActiveRunRejoinRetry((current) => current + 1);
+    }, retryDelay);
+  }, []);
+
+  const settleActiveRunRejoin = useCallback(() => {
+    const rejoin = activeRunRejoinRef.current;
+    if (!rejoin.inFlight) {
+      return;
+    }
+    rejoin.inFlight = false;
+    rejoin.settled = true;
+    if (rejoin.retryTimer !== null) {
+      clearTimeout(rejoin.retryTimer);
+      rejoin.retryTimer = null;
+    }
+  }, []);
+
   const clearPreparedReplayMasks = useCallback(
     (replay: PendingPreparedReplayMask | null) => {
       if (!replay) {
@@ -2012,6 +2116,7 @@ export function useThreadStream({
       }
     },
     onError(error) {
+      scheduleActiveRunRejoinRetry();
       setOptimisticMessages([]);
       setOptimisticThreadId(null);
       setLiveMessagesThreadId(null);
@@ -2033,7 +2138,11 @@ export function useThreadStream({
         });
       }
     },
-    onFinish(state) {
+    onFinish(state, run) {
+      if (run) {
+        completedRunIdsRef.current.add(run.run_id);
+      }
+      settleActiveRunRejoin();
       listeners.current.onFinish?.(state.values);
       pendingPreparedReplayRef.current = null;
       pendingUsageBaselineMessageIdsRef.current = new Set(
@@ -2044,6 +2153,77 @@ export function useThreadStream({
       invalidateStoppedThreadCaches(queryClient, threadIdRef.current, isMock);
     },
   });
+  const { isLoading: isThreadLoading, joinStream } = thread;
+
+  // reconnectOnMount only knows the run id stored in this tab's
+  // sessionStorage. A reopened browser or a new tab has no pointer, so recover
+  // the newest active run from the server and join its resumable SSE stream.
+  useEffect(() => {
+    const resolvedThreadId = onStreamThreadId ?? null;
+    const resolvedRunId = activeRunId ?? null;
+    let rejoin = activeRunRejoinRef.current;
+
+    if (
+      rejoin.threadId !== resolvedThreadId ||
+      rejoin.runId !== resolvedRunId
+    ) {
+      if (rejoin.retryTimer !== null) {
+        clearTimeout(rejoin.retryTimer);
+      }
+      if (rejoin.attempts > 0 && rejoin.threadId && rejoin.runId) {
+        clearReconnectRun(rejoin.threadId, rejoin.runId);
+      }
+      rejoin = createActiveRunRejoinState(resolvedThreadId, resolvedRunId);
+      activeRunRejoinRef.current = rejoin;
+    }
+
+    if (
+      !resolvedThreadId ||
+      !resolvedRunId ||
+      completedRunIdsRef.current.has(resolvedRunId) ||
+      rejoin.inFlight ||
+      rejoin.retryTimer !== null ||
+      rejoin.settled ||
+      rejoin.attempts >= MAX_ACTIVE_RUN_REJOIN_ATTEMPTS ||
+      isThreadLoading
+    ) {
+      return;
+    }
+
+    // A matching pointer means the SDK's native same-tab reconnect owns this
+    // run. Do not create a second SSE consumer.
+    if (readReconnectRun(resolvedThreadId) === resolvedRunId) {
+      return;
+    }
+
+    rejoin.attempts += 1;
+    rejoin.inFlight = true;
+    rememberReconnectRun(resolvedThreadId, resolvedRunId);
+    void joinStream(resolvedRunId);
+  }, [
+    activeRunId,
+    activeRunRejoinRetry,
+    isThreadLoading,
+    joinStream,
+    onStreamThreadId,
+  ]);
+
+  useEffect(
+    () => () => {
+      const rejoin = activeRunRejoinRef.current;
+      if (rejoin.threadId !== (onStreamThreadId ?? null)) {
+        return;
+      }
+      if (rejoin.retryTimer !== null) {
+        clearTimeout(rejoin.retryTimer);
+      }
+      if (rejoin.attempts > 0 && rejoin.threadId && rejoin.runId) {
+        clearReconnectRun(rejoin.threadId, rejoin.runId);
+      }
+      activeRunRejoinRef.current = createActiveRunRejoinState();
+    },
+    [onStreamThreadId],
+  );
 
   const stopThread = useCallback(async () => {
     const stoppedThreadId =
@@ -3287,6 +3467,7 @@ export function useThreadRuns(
     },
     enabled: enabled && Boolean(threadId),
     refetchOnWindowFocus: false,
+    retry: false,
   });
 }
 

frontend/tests/unit/core/threads/active-run-rejoin.dom.test.tsx

@@ -0,0 +1,351 @@
+import type { Run } from "@langchain/langgraph-sdk";
+import { afterEach, beforeEach, expect, rs, test } from "@rstest/core";
+import { QueryClient, QueryClientProvider } from "@tanstack/react-query";
+import { act, renderHook } from "@testing-library/react";
+import { createElement, type ReactNode } from "react";
+
+import { I18nContext } from "@/core/i18n/context";
+import { enUS } from "@/core/i18n/locales/en-US";
+import { DEFAULT_LOCAL_SETTINGS } from "@/core/settings/local";
+import { useThreadStream } from "@/core/threads/hooks";
+
+type StreamOptions = {
+  onError?: (error: unknown) => void;
+  onFinish?: (
+    state: {
+      values: { artifacts: never[]; messages: never[]; title: string };
+    },
+    run?: { thread_id: string; run_id: string },
+  ) => void;
+};
+
+const apiMockState = rs.hoisted(() => ({
+  listRuns: rs.fn(async () => [] as Run[]),
+}));
+
+const streamMockState = rs.hoisted(() => ({
+  isLoading: false,
+  joinStream: rs.fn(async (_runId: string) => undefined),
+  options: undefined as StreamOptions | undefined,
+}));
+
+rs.mock("@/core/api", () => ({
+  getAPIClient: () => ({
+    runs: { list: apiMockState.listRuns },
+  }),
+}));
+
+rs.mock("@langchain/langgraph-sdk/react", () => ({
+  useStream: (options: StreamOptions) => {
+    streamMockState.options = options;
+    return {
+      isLoading: streamMockState.isLoading,
+      joinStream: streamMockState.joinStream,
+      messages: [],
+      stop: async () => undefined,
+      submit: async () => undefined,
+      values: {
+        artifacts: [],
+        messages: [],
+        title: "",
+        todos: [],
+      },
+    };
+  },
+}));
+
+const ACTIVE_RUN = {
+  run_id: "run-active",
+  status: "running",
+} as Run;
+
+function createWrapper(queryClient: QueryClient) {
+  return function ActiveRunRejoinTestWrapper({
+    children,
+  }: {
+    children: ReactNode;
+  }) {
+    return createElement(
+      QueryClientProvider,
+      { client: queryClient },
+      createElement(
+        I18nContext.Provider,
+        {
+          value: {
+            locale: "en-US",
+            setLocale: () => undefined,
+            t: enUS,
+          },
+        },
+        children,
+      ),
+    );
+  };
+}
+
+async function flushFrames() {
+  for (let index = 0; index < 6; index += 1) {
+    await act(async () => {
+      await rs.advanceTimersByTimeAsync(0);
+    });
+  }
+}
+
+function renderThread(threadId = "thread-1") {
+  const queryClient = new QueryClient({
+    defaultOptions: { queries: { retry: false } },
+  });
+  const rendered = renderHook(
+    ({ activeThreadId }: { activeThreadId: string }) =>
+      useThreadStream({
+        context: DEFAULT_LOCAL_SETTINGS.context,
+        threadId: activeThreadId,
+      }),
+    {
+      initialProps: { activeThreadId: threadId },
+      wrapper: createWrapper(queryClient),
+    },
+  );
+  return { queryClient, ...rendered };
+}
+
+beforeEach(() => {
+  rs.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] });
+  window.sessionStorage.clear();
+  apiMockState.listRuns.mockReset();
+  apiMockState.listRuns.mockResolvedValue([ACTIVE_RUN]);
+  streamMockState.isLoading = false;
+  streamMockState.joinStream.mockReset();
+  streamMockState.joinStream.mockResolvedValue(undefined);
+  streamMockState.options = undefined;
+  rs.stubGlobal(
+    "fetch",
+    rs.fn(
+      async () =>
+        new Response(
+          JSON.stringify({ data: [], has_more: false, next_before_seq: null }),
+          { status: 200, headers: { "Content-Type": "application/json" } },
+        ),
+    ),
+  );
+});
+
+afterEach(() => {
+  rs.useRealTimers();
+  rs.unstubAllGlobals();
+});
+
+test("joins the newest active run when a reopened tab has no reconnect pointer", async () => {
+  const { unmount } = renderThread();
+  await flushFrames();
+
+  expect(streamMockState.joinStream).toHaveBeenCalledTimes(1);
+  expect(streamMockState.joinStream).toHaveBeenCalledWith("run-active");
+  expect(window.sessionStorage.getItem("lg:stream:thread-1")).toBe(
+    "run-active",
+  );
+
+  unmount();
+  expect(window.sessionStorage.getItem("lg:stream:thread-1")).toBeNull();
+});
+
+test("leaves a matching reconnect pointer to the SDK without joining twice", async () => {
+  window.sessionStorage.setItem("lg:stream:thread-1", "run-active");
+  const { unmount } = renderThread();
+  await flushFrames();
+
+  expect(streamMockState.joinStream).not.toHaveBeenCalled();
+  expect(window.sessionStorage.getItem("lg:stream:thread-1")).toBe(
+    "run-active",
+  );
+
+  unmount();
+});
+
+test("retries a failed recovered stream twice with bounded backoff", async () => {
+  const { unmount } = renderThread();
+  await flushFrames();
+  expect(streamMockState.joinStream).toHaveBeenCalledTimes(1);
+
+  act(() => streamMockState.options?.onError?.(new Error("disconnected")));
+  await act(async () => {
+    await rs.advanceTimersByTimeAsync(999);
+  });
+  expect(streamMockState.joinStream).toHaveBeenCalledTimes(1);
+  await act(async () => {
+    await rs.advanceTimersByTimeAsync(1);
+  });
+  expect(streamMockState.joinStream).toHaveBeenCalledTimes(2);
+
+  act(() => streamMockState.options?.onError?.(new Error("disconnected")));
+  await act(async () => {
+    await rs.advanceTimersByTimeAsync(1_999);
+  });
+  expect(streamMockState.joinStream).toHaveBeenCalledTimes(2);
+  await act(async () => {
+    await rs.advanceTimersByTimeAsync(1);
+  });
+  expect(streamMockState.joinStream).toHaveBeenCalledTimes(3);
+
+  act(() => streamMockState.options?.onError?.(new Error("disconnected")));
+  await act(async () => {
+    await rs.advanceTimersByTimeAsync(10_000);
+  });
+  expect(streamMockState.joinStream).toHaveBeenCalledTimes(3);
+
+  unmount();
+});
+
+test("does not retry after the recovered run finishes", async () => {
+  const { unmount } = renderThread();
+  await flushFrames();
+  expect(streamMockState.joinStream).toHaveBeenCalledTimes(1);
+
+  act(() =>
+    streamMockState.options?.onFinish?.({
+      values: { artifacts: [], messages: [], title: "Done" },
+    }),
+  );
+  await act(async () => {
+    await rs.advanceTimersByTimeAsync(10_000);
+  });
+
+  expect(streamMockState.joinStream).toHaveBeenCalledTimes(1);
+  unmount();
+});
+
+test("cancels a pending retry when the recovered stream unmounts", async () => {
+  const { unmount } = renderThread();
+  await flushFrames();
+  expect(streamMockState.joinStream).toHaveBeenCalledTimes(1);
+
+  act(() => streamMockState.options?.onError?.(new Error("disconnected")));
+  unmount();
+  await act(async () => {
+    await rs.advanceTimersByTimeAsync(10_000);
+  });
+
+  expect(streamMockState.joinStream).toHaveBeenCalledTimes(1);
+  expect(window.sessionStorage.getItem("lg:stream:thread-1")).toBeNull();
+});
+
+test("clears the old retry when the active run changes", async () => {
+  const { queryClient, unmount } = renderThread();
+  await flushFrames();
+  expect(streamMockState.joinStream).toHaveBeenCalledWith("run-active");
+
+  act(() => streamMockState.options?.onError?.(new Error("disconnected")));
+  act(() => {
+    queryClient.setQueryData(
+      ["thread", "thread-1"],
+      [{ ...ACTIVE_RUN, run_id: "run-next", status: "pending" }],
+    );
+  });
+  await flushFrames();
+  await act(async () => {
+    await rs.advanceTimersByTimeAsync(10_000);
+  });
+
+  expect(streamMockState.joinStream).toHaveBeenCalledTimes(2);
+  expect(streamMockState.joinStream).toHaveBeenLastCalledWith("run-next");
+  expect(window.sessionStorage.getItem("lg:stream:thread-1")).toBe("run-next");
+  unmount();
+});
+
+test.each(["submitted", "same-tab reconnect"])(
+  "does not rejoin a finished %s run while the runs cache is stale",
+  async (kind) => {
+    if (kind === "same-tab reconnect") {
+      window.sessionStorage.setItem("lg:stream:thread-1", "run-active");
+    }
+    streamMockState.isLoading = true;
+    const { rerender, unmount } = renderThread();
+    await flushFrames();
+    expect(streamMockState.joinStream).not.toHaveBeenCalled();
+
+    // The SDK removes its pointer before onFinish. Keep the runs refetch
+    // pending so the effect still sees the previous "running" snapshot.
+    apiMockState.listRuns.mockImplementation(
+      () =>
+        new Promise(() => {
+          // Keep the cached running snapshot until the hook unmounts.
+        }),
+    );
+    act(() => {
+      window.sessionStorage.removeItem("lg:stream:thread-1");
+      streamMockState.options?.onFinish?.(
+        { values: { artifacts: [], messages: [], title: "Done" } },
+        { thread_id: "thread-1", run_id: "run-active" },
+      );
+      streamMockState.isLoading = false;
+    });
+    rerender({ activeThreadId: "thread-1" });
+    await flushFrames();
+
+    expect(streamMockState.joinStream).not.toHaveBeenCalled();
+    expect(window.sessionStorage.getItem("lg:stream:thread-1")).toBeNull();
+    unmount();
+  },
+);
+
+test("does not rejoin a finished run discovered by a delayed initial runs read", async () => {
+  let resolveRuns!: (runs: Run[]) => void;
+  apiMockState.listRuns.mockImplementation(
+    () =>
+      new Promise<Run[]>((resolve) => {
+        resolveRuns = resolve;
+      }),
+  );
+  streamMockState.isLoading = true;
+  const { rerender, unmount } = renderThread();
+  await flushFrames();
+
+  act(() => {
+    streamMockState.options?.onFinish?.(
+      { values: { artifacts: [], messages: [], title: "Done" } },
+      { thread_id: "thread-1", run_id: "run-active" },
+    );
+    streamMockState.isLoading = false;
+    resolveRuns([ACTIVE_RUN]);
+  });
+  rerender({ activeThreadId: "thread-1" });
+  await flushFrames();
+
+  expect(streamMockState.joinStream).not.toHaveBeenCalled();
+  unmount();
+});
+
+test("still recovers a different active run after an earlier run finishes", async () => {
+  streamMockState.isLoading = true;
+  const { queryClient, rerender, unmount } = renderThread();
+  await flushFrames();
+
+  apiMockState.listRuns.mockImplementation(
+    () =>
+      new Promise(() => {
+        // Keep the cached running snapshot until the hook unmounts.
+      }),
+  );
+  act(() => {
+    streamMockState.options?.onFinish?.(
+      { values: { artifacts: [], messages: [], title: "Done" } },
+      { thread_id: "thread-1", run_id: "run-active" },
+    );
+    streamMockState.isLoading = false;
+  });
+  rerender({ activeThreadId: "thread-1" });
+  await flushFrames();
+  expect(streamMockState.joinStream).not.toHaveBeenCalled();
+
+  act(() => {
+    queryClient.setQueryData(
+      ["thread", "thread-1"],
+      [{ ...ACTIVE_RUN, run_id: "run-next", status: "pending" }],
+    );
+  });
+  await flushFrames();
+
+  expect(streamMockState.joinStream).toHaveBeenCalledTimes(1);
+  expect(streamMockState.joinStream).toHaveBeenCalledWith("run-next");
+  unmount();
+});

frontend/tests/unit/core/threads/stream-options.test.ts

@@ -27,7 +27,10 @@ async function captureThreadStreamOptions(assistantId?: string) {
       isLoading: false,
     }),
     useMutation: rs.fn(),
-    useQuery: rs.fn(),
+    useQuery: () => ({
+      data: [],
+      refetch: rs.fn(),
+    }),
     useQueryClient: () => ({
       invalidateQueries: rs.fn(),
       setQueriesData: rs.fn(),