github/deer-flow
1
0
Fork 0
mirror of https://github.com/bytedance/deer-flow.git synced 2026-10-26 13:16:24 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(frontend): keep recovered run cancellation target (#6075)

Browse Source 
* 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 <willem.jiang@gmail.com>
This commit is contained in:
PeaceMaker-best 2026-10-06 11:39:58 +08:00 committed by GitHub
parent e05ca0c4e5
commit 77df35baa8
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 79 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

47
frontend/src/core/threads/hooks.ts
View File

@ -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<Run, "thread_id" | "run_id"> | 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);

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

@ -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"],