diff --git a/frontend/src/AGENTS.md b/frontend/src/AGENTS.md index 25fbb836d..6d39b8d2d 100644 --- a/frontend/src/AGENTS.md +++ b/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. diff --git a/frontend/src/core/threads/hooks.ts b/frontend/src/core/threads/hooks.ts index 63b30b674..f31ed6709 100644 --- a/frontend/src/core/threads/hooks.ts +++ b/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 = new Set(); +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 | 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( + 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()); // 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, }); } diff --git a/frontend/tests/unit/core/threads/active-run-rejoin.dom.test.tsx b/frontend/tests/unit/core/threads/active-run-rejoin.dom.test.tsx new file mode 100644 index 000000000..3c6b727c5 --- /dev/null +++ b/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((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(); +}); diff --git a/frontend/tests/unit/core/threads/stream-options.test.ts b/frontend/tests/unit/core/threads/stream-options.test.ts index 84e8fcaac..97d40a298 100644 --- a/frontend/tests/unit/core/threads/stream-options.test.ts +++ b/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(),