From 77df35baa871243128f105f56d72653bae67d47c Mon Sep 17 00:00:00 2001 From: PeaceMaker-best Date: Tue, 6 Oct 2026 11:39:58 +0800 Subject: [PATCH] fix(frontend): keep recovered run cancellation target (#6075) * fix(frontend): keep recovered run cancellation target Signed-off-by: PeaceMaker-best <221849497+PeaceMaker-best@users.noreply.github.com> * docs(frontend): explain recovered stream error metadata contract --------- Signed-off-by: PeaceMaker-best <221849497+PeaceMaker-best@users.noreply.github.com> Co-authored-by: PeaceMaker-best <221849497+PeaceMaker-best@users.noreply.github.com> Co-authored-by: Willem Jiang --- frontend/src/core/threads/hooks.ts | 47 ++++++++++------ .../threads/active-run-rejoin.dom.test.tsx | 56 +++++++++++++++++-- 2 files changed, 79 insertions(+), 24 deletions(-) diff --git a/frontend/src/core/threads/hooks.ts b/frontend/src/core/threads/hooks.ts index 0d5f7e6a73..33297a1c4d 100644 --- a/frontend/src/core/threads/hooks.ts +++ b/frontend/src/core/threads/hooks.ts @@ -1982,24 +1982,35 @@ 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; - } + const scheduleActiveRunRejoinRetry = useCallback( + (run: Pick | undefined) => { + const rejoin = activeRunRejoinRef.current; + // SDK 1.6.0 LGP joinStream errors include { thread_id, run_id }; history + // errors omit it. Re-verify this callback contract when upgrading the SDK. + if ( + !rejoin.inFlight || + !rejoin.threadId || + !rejoin.runId || + run?.thread_id !== rejoin.threadId || + run?.run_id !== 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.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); - }, []); + rejoin.retryTimer = setTimeout(() => { + rejoin.retryTimer = null; + setActiveRunRejoinRetry((current) => current + 1); + }, retryDelay); + }, + [], + ); const settleActiveRunRejoin = useCallback(() => { const rejoin = activeRunRejoinRef.current; @@ -2188,8 +2199,8 @@ export function useThreadStream({ } } }, - onError(error) { - scheduleActiveRunRejoinRetry(); + onError(error, run) { + scheduleActiveRunRejoinRetry(run); setOptimisticMessages([]); setOptimisticThreadId(null); setLiveMessagesThreadId(null); 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 index 3c6b727c5c..af9370c288 100644 --- a/frontend/tests/unit/core/threads/active-run-rejoin.dom.test.tsx +++ b/frontend/tests/unit/core/threads/active-run-rejoin.dom.test.tsx @@ -10,7 +10,10 @@ import { DEFAULT_LOCAL_SETTINGS } from "@/core/settings/local"; import { useThreadStream } from "@/core/threads/hooks"; type StreamOptions = { - onError?: (error: unknown) => void; + onError?: ( + error: unknown, + run?: { thread_id: string; run_id: string }, + ) => void; onFinish?: ( state: { values: { artifacts: never[]; messages: never[]; title: string }; @@ -58,6 +61,10 @@ const ACTIVE_RUN = { run_id: "run-active", status: "running", } as Run; +const ACTIVE_RUN_META = { + thread_id: "thread-1", + run_id: "run-active", +}; function createWrapper(queryClient: QueryClient) { return function ActiveRunRejoinTestWrapper({ @@ -109,6 +116,13 @@ function renderThread(threadId = "thread-1") { return { queryClient, ...rendered }; } +function failActiveRecoveredStream() { + streamMockState.options?.onError?.( + new Error("disconnected"), + ACTIVE_RUN_META, + ); +} + beforeEach(() => { rs.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] }); window.sessionStorage.clear(); @@ -162,12 +176,42 @@ test("leaves a matching reconnect pointer to the SDK without joining twice", asy unmount(); }); +test.each([ + ["has no run metadata", undefined], + [ + "belongs to another thread", + { thread_id: "thread-2", run_id: "run-active" }, + ], + ["belongs to another run", { thread_id: "thread-1", run_id: "run-other" }], +])( + "keeps the recovered-run pointer when an error %s", + async (_caseName, run) => { + const { unmount } = renderThread(); + await flushFrames(); + expect(streamMockState.joinStream).toHaveBeenCalledTimes(1); + + act(() => + streamMockState.options?.onError?.(new Error("history failed"), run), + ); + expect(window.sessionStorage.getItem("lg:stream:thread-1")).toBe( + "run-active", + ); + + await act(async () => { + await rs.advanceTimersByTimeAsync(10_000); + }); + expect(streamMockState.joinStream).toHaveBeenCalledTimes(1); + + 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"))); + act(() => failActiveRecoveredStream()); await act(async () => { await rs.advanceTimersByTimeAsync(999); }); @@ -177,7 +221,7 @@ test("retries a failed recovered stream twice with bounded backoff", async () => }); expect(streamMockState.joinStream).toHaveBeenCalledTimes(2); - act(() => streamMockState.options?.onError?.(new Error("disconnected"))); + act(() => failActiveRecoveredStream()); await act(async () => { await rs.advanceTimersByTimeAsync(1_999); }); @@ -187,7 +231,7 @@ test("retries a failed recovered stream twice with bounded backoff", async () => }); expect(streamMockState.joinStream).toHaveBeenCalledTimes(3); - act(() => streamMockState.options?.onError?.(new Error("disconnected"))); + act(() => failActiveRecoveredStream()); await act(async () => { await rs.advanceTimersByTimeAsync(10_000); }); @@ -219,7 +263,7 @@ test("cancels a pending retry when the recovered stream unmounts", async () => { await flushFrames(); expect(streamMockState.joinStream).toHaveBeenCalledTimes(1); - act(() => streamMockState.options?.onError?.(new Error("disconnected"))); + act(() => failActiveRecoveredStream()); unmount(); await act(async () => { await rs.advanceTimersByTimeAsync(10_000); @@ -234,7 +278,7 @@ test("clears the old retry when the active run changes", async () => { await flushFrames(); expect(streamMockState.joinStream).toHaveBeenCalledWith("run-active"); - act(() => streamMockState.options?.onError?.(new Error("disconnected"))); + act(() => failActiveRecoveredStream()); act(() => { queryClient.setQueryData( ["thread", "thread-1"],