mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-28 14:10:37 +00:00
fix(frontend): retry a failed same-tab reconnect after refresh (#6400)
* fix(frontend): retry a failed same-tab reconnect after refresh After a refresh the SDK reconnects once from the tab's lg:stream pointer and keeps that pointer when the join fails. Active-run recovery defers to the SDK whenever the pointer matches the active run, and its error handler only scheduled retries for joins it started itself, so a failed reconnect (at connect time or mid-stream) left the tab without a live stream until another refresh. When an SDK reconnect fails, release the pointer and wake the recovery effect, which rejoins the run if the runs read still reports it active, with its existing bounded retries. Runs created by this hook's submit (onCreated) keep their current error handling. The new native-reconnect-recovery test drives the real SDK useStream, so it also pins the SDK behavior this relies on: one reconnect per mount and a pointer that survives a failed join. * docs(changelog): reference #6400 * test(frontend): pin that a released reconnect skips finished runs A run handed to recovery after a failed SDK reconnect was never streamed by this hook, so completedRunIdsRef cannot guard it; only the server-reported status in activeRunId does. Resolve the runs read with each terminal status after the failed reconnect and assert recovery never joins and the pointer stays cleared. --------- Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
This commit is contained in:
Hyeonsang Cho
GitHub
parent
b414a1bbec
commit
2e42242888
@ -697,6 +697,14 @@ This release closes that milestone with **439 merged pull requests**.
|
||||
|
||||
### Fixed
|
||||
|
||||
- **frontend:** A failed reconnect after a page refresh is now retried in the same
|
||||
tab. The SDK reconnects once from the tab's `lg:stream` pointer and keeps that
|
||||
pointer on error, and active-run recovery skipped any run with a matching
|
||||
pointer, so the live stream stayed detached until another refresh. When that
|
||||
reconnect fails, including a drop mid-stream, recovery now releases the pointer
|
||||
and rejoins the run if the server still reports it active, with its existing
|
||||
bounded retries (immediately, then after 1s and 2s). Failed submitted runs are
|
||||
unchanged. ([#6400])
|
||||
- **channels:** Opening an IM-channel conversation on the web while its run is still
|
||||
going no longer shows the user's message twice. Channel run input carried no
|
||||
message id, so the Gateway stored it id-less in the run record while the
|
||||
@ -8974,4 +8982,5 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
|
||||
[#6378]: https://github.com/bytedance/deer-flow/pull/6378
|
||||
[#6386]: https://github.com/bytedance/deer-flow/pull/6386
|
||||
[#6388]: https://github.com/bytedance/deer-flow/pull/6388
|
||||
[#6400]: https://github.com/bytedance/deer-flow/pull/6400
|
||||
[#6401]: https://github.com/bytedance/deer-flow/pull/6401
|
||||
|
||||
@ -615,6 +615,11 @@
|
||||
|
||||
### 修复
|
||||
|
||||
- **前端：** 刷新页面后的首次重连失败时，现在会在同一标签页中重试。SDK 只根据标签页的
|
||||
`lg:stream` 指针重连一次，出错时保留该指针；而活动运行恢复会跳过指针匹配的运行，
|
||||
因此实时流在再次刷新前一直无法恢复。现在该重连失败时（包括流中途断开），恢复逻辑会释放指针；
|
||||
若服务端仍报告该运行处于活动状态，则按现有的有限重试（立即一次，随后 1s、2s）重新加入。
|
||||
提交失败的运行行为不变。([#6400])
|
||||
- **渠道：** 在 IM 渠道会话的运行尚未结束时从网页打开它，用户消息不再显示两次。渠道运行输入此前不带消息 id，
|
||||
Gateway 在运行记录中按原样保存无 id 的消息，而检查点中的副本会被分配新的 uuid；网页客户端重连时按 id
|
||||
合并这两份副本，因此在运行结束前两条都会显示。现在渠道发送的人类消息自带 id，两份副本共享同一 id。([#6401])
|
||||
@ -7392,4 +7397,5 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子
|
||||
[#6370]: https://github.com/bytedance/deer-flow/pull/6370
|
||||
[#6378]: https://github.com/bytedance/deer-flow/pull/6378
|
||||
[#6388]: https://github.com/bytedance/deer-flow/pull/6388
|
||||
[#6400]: https://github.com/bytedance/deer-flow/pull/6400
|
||||
[#6401]: https://github.com/bytedance/deer-flow/pull/6401
|
||||
|
||||
@ -183,7 +183,7 @@ Array previews coalesce consecutive generated markers only at the end into one o
|
||||
- **LangGraph client** is a singleton obtained via `getAPIClient()` in `core/api/`
|
||||
- **Run creation retries** are disabled only for the initial `POST /runs/stream` in `core/api/api-client.ts`, because an ambiguous failure can otherwise create duplicate runs. Automatic SSE recovery keeps the SDK's normal HTTP retry budget for `GET` requests to the server-provided `Location`, including transient 5xx responses, and forwards `Last-Event-ID`. Reads and explicit SSE joins keep their normal retry behavior. Regression tests cover both a failed creation POST and POST success → stream interruption → GET 504 → GET success.
|
||||
- **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, including `tool_artifacts` capture/consumption updates (same-handle replacement preserves order, retention directives trim the oldest entries, and the absolute ceiling is 1000) 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.
|
||||
- **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. The SDK's same-tab reconnect keeps its pointer on error and never retries, so a failed reconnect (not a failed submit) releases the pointer and recovery takes over with its own bounded retries.
|
||||
- **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.
|
||||
|
||||
@ -1947,6 +1947,9 @@ export function useThreadStream({
|
||||
// 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>());
|
||||
// onCreated fires only for runs this hook submitted. A failed submit stays
|
||||
// with the submit flow; only SDK reconnects are handed to recovery.
|
||||
const submittedRunIdRef = useRef<string | null>(null);
|
||||
|
||||
// Keep listeners ref updated with latest callbacks
|
||||
useEffect(() => {
|
||||
@ -2005,6 +2008,19 @@ export function useThreadStream({
|
||||
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 (
|
||||
run &&
|
||||
!rejoin.inFlight &&
|
||||
run.run_id !== submittedRunIdRef.current &&
|
||||
readReconnectRun(run.thread_id) === run.run_id
|
||||
) {
|
||||
// The SDK's same-tab reconnect tries once and keeps its pointer on
|
||||
// error, so the recovery effect would keep deferring to a stream that
|
||||
// no longer exists. Release the pointer and let recovery take over.
|
||||
clearReconnectRun(run.thread_id, run.run_id);
|
||||
setActiveRunRejoinRetry((current) => current + 1);
|
||||
return;
|
||||
}
|
||||
if (
|
||||
!rejoin.inFlight ||
|
||||
!rejoin.threadId ||
|
||||
@ -2075,6 +2091,7 @@ export function useThreadStream({
|
||||
// Keep explicit: SDK types claim @default true, but runtime uses throttle ?? false.
|
||||
throttle: true,
|
||||
onCreated(meta) {
|
||||
submittedRunIdRef.current = meta.run_id;
|
||||
handleStreamStart(meta.thread_id, meta.run_id);
|
||||
const now = new Date().toISOString();
|
||||
upsertThreadInSearchCache(queryClient, {
|
||||
@ -2293,7 +2310,8 @@ export function useThreadStream({
|
||||
}
|
||||
|
||||
// A matching pointer means the SDK's native same-tab reconnect owns this
|
||||
// run. Do not create a second SSE consumer.
|
||||
// run. Do not create a second SSE consumer; if that reconnect fails,
|
||||
// scheduleActiveRunRejoinRetry releases the pointer.
|
||||
if (readReconnectRun(resolvedThreadId) === resolvedRunId) {
|
||||
return;
|
||||
}
|
||||
|
||||
@ -10,6 +10,7 @@ import { DEFAULT_LOCAL_SETTINGS } from "@/core/settings/local";
|
||||
import { useThreadStream } from "@/core/threads/hooks";
|
||||
|
||||
type StreamOptions = {
|
||||
onCreated?: (meta: { thread_id: string; run_id: string }) => void;
|
||||
onError?: (
|
||||
error: unknown,
|
||||
run?: { thread_id: string; run_id: string },
|
||||
@ -176,6 +177,120 @@ test("leaves a matching reconnect pointer to the SDK without joining twice", asy
|
||||
unmount();
|
||||
});
|
||||
|
||||
test("takes over a failed same-tab reconnect with bounded recovery", async () => {
|
||||
window.sessionStorage.setItem("lg:stream:thread-1", "run-active");
|
||||
const { unmount } = renderThread();
|
||||
await flushFrames();
|
||||
expect(streamMockState.joinStream).not.toHaveBeenCalled();
|
||||
|
||||
// The SDK keeps its pointer when its single reconnect fails.
|
||||
act(() => failActiveRecoveredStream());
|
||||
await flushFrames();
|
||||
expect(streamMockState.joinStream).toHaveBeenCalledTimes(1);
|
||||
expect(streamMockState.joinStream).toHaveBeenCalledWith("run-active");
|
||||
expect(window.sessionStorage.getItem("lg:stream:thread-1")).toBe(
|
||||
"run-active",
|
||||
);
|
||||
|
||||
act(() => failActiveRecoveredStream());
|
||||
await act(async () => {
|
||||
await rs.advanceTimersByTimeAsync(1_000);
|
||||
});
|
||||
expect(streamMockState.joinStream).toHaveBeenCalledTimes(2);
|
||||
|
||||
act(() => failActiveRecoveredStream());
|
||||
await act(async () => {
|
||||
await rs.advanceTimersByTimeAsync(2_000);
|
||||
});
|
||||
expect(streamMockState.joinStream).toHaveBeenCalledTimes(3);
|
||||
|
||||
act(() => failActiveRecoveredStream());
|
||||
await act(async () => {
|
||||
await rs.advanceTimersByTimeAsync(10_000);
|
||||
});
|
||||
expect(streamMockState.joinStream).toHaveBeenCalledTimes(3);
|
||||
|
||||
unmount();
|
||||
});
|
||||
|
||||
test("recovers a same-tab reconnect that failed before the runs read", async () => {
|
||||
let resolveRuns!: (runs: Run[]) => void;
|
||||
apiMockState.listRuns.mockImplementation(
|
||||
() =>
|
||||
new Promise<Run[]>((resolve) => {
|
||||
resolveRuns = resolve;
|
||||
}),
|
||||
);
|
||||
window.sessionStorage.setItem("lg:stream:thread-1", "run-active");
|
||||
const { unmount } = renderThread();
|
||||
await flushFrames();
|
||||
|
||||
act(() => failActiveRecoveredStream());
|
||||
expect(window.sessionStorage.getItem("lg:stream:thread-1")).toBeNull();
|
||||
expect(streamMockState.joinStream).not.toHaveBeenCalled();
|
||||
|
||||
act(() => resolveRuns([ACTIVE_RUN]));
|
||||
await flushFrames();
|
||||
|
||||
expect(streamMockState.joinStream).toHaveBeenCalledTimes(1);
|
||||
expect(streamMockState.joinStream).toHaveBeenCalledWith("run-active");
|
||||
unmount();
|
||||
});
|
||||
|
||||
test.each(["success", "error", "timeout", "interrupted"])(
|
||||
"does not rejoin a released same-tab reconnect whose run ended with %s",
|
||||
async (status) => {
|
||||
let resolveRuns!: (runs: Run[]) => void;
|
||||
apiMockState.listRuns.mockImplementation(
|
||||
() =>
|
||||
new Promise<Run[]>((resolve) => {
|
||||
resolveRuns = resolve;
|
||||
}),
|
||||
);
|
||||
window.sessionStorage.setItem("lg:stream:thread-1", "run-active");
|
||||
const { unmount } = renderThread();
|
||||
await flushFrames();
|
||||
|
||||
// This hook never streamed the run, so completedRunIdsRef cannot guard
|
||||
// it; only the server-reported status keeps recovery from rejoining.
|
||||
act(() => failActiveRecoveredStream());
|
||||
act(() => resolveRuns([{ ...ACTIVE_RUN, status } as Run]));
|
||||
await flushFrames();
|
||||
await act(async () => {
|
||||
await rs.advanceTimersByTimeAsync(10_000);
|
||||
});
|
||||
|
||||
expect(streamMockState.joinStream).not.toHaveBeenCalled();
|
||||
expect(window.sessionStorage.getItem("lg:stream:thread-1")).toBeNull();
|
||||
unmount();
|
||||
},
|
||||
);
|
||||
|
||||
test("leaves a failed submitted run to the submit flow", async () => {
|
||||
apiMockState.listRuns.mockResolvedValue([]);
|
||||
const { queryClient, unmount } = renderThread();
|
||||
await flushFrames();
|
||||
|
||||
act(() => {
|
||||
window.sessionStorage.setItem("lg:stream:thread-1", "run-active");
|
||||
streamMockState.options?.onCreated?.(ACTIVE_RUN_META);
|
||||
});
|
||||
act(() => {
|
||||
queryClient.setQueryData(["thread", "thread-1"], [ACTIVE_RUN]);
|
||||
});
|
||||
await flushFrames();
|
||||
act(() => failActiveRecoveredStream());
|
||||
await act(async () => {
|
||||
await rs.advanceTimersByTimeAsync(10_000);
|
||||
});
|
||||
|
||||
expect(streamMockState.joinStream).not.toHaveBeenCalled();
|
||||
expect(window.sessionStorage.getItem("lg:stream:thread-1")).toBe(
|
||||
"run-active",
|
||||
);
|
||||
unmount();
|
||||
});
|
||||
|
||||
test.each([
|
||||
["has no run metadata", undefined],
|
||||
[
|
||||
|
||||
@ -0,0 +1,167 @@
|
||||
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";
|
||||
|
||||
// Drives the real SDK useStream (not a mock) so the test pins the SDK
|
||||
// behavior the recovery relies on: one same-tab reconnect from the
|
||||
// lg:stream pointer, and a pointer that survives a failed join.
|
||||
|
||||
const clientMockState = rs.hoisted(() => ({
|
||||
joinStream: rs.fn(),
|
||||
listRuns: rs.fn(async () => [] as Run[]),
|
||||
}));
|
||||
|
||||
rs.mock("@/core/api", () => ({
|
||||
getAPIClient: () => ({
|
||||
runs: {
|
||||
cancel: async () => undefined,
|
||||
joinStream: clientMockState.joinStream,
|
||||
list: clientMockState.listRuns,
|
||||
},
|
||||
threads: {
|
||||
getHistory: async () => [],
|
||||
getState: async () => ({ values: {} }),
|
||||
},
|
||||
}),
|
||||
}));
|
||||
|
||||
const ACTIVE_RUN = {
|
||||
run_id: "run-active",
|
||||
status: "running",
|
||||
} as Run;
|
||||
|
||||
function createWrapper(queryClient: QueryClient) {
|
||||
return function NativeReconnectTestWrapper({
|
||||
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 < 10; index += 1) {
|
||||
await act(async () => {
|
||||
await rs.advanceTimersByTimeAsync(0);
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
function renderThread() {
|
||||
const queryClient = new QueryClient({
|
||||
defaultOptions: { queries: { retry: false } },
|
||||
});
|
||||
return renderHook(
|
||||
() =>
|
||||
useThreadStream({
|
||||
context: DEFAULT_LOCAL_SETTINGS.context,
|
||||
threadId: "thread-1",
|
||||
}),
|
||||
{ wrapper: createWrapper(queryClient) },
|
||||
);
|
||||
}
|
||||
|
||||
beforeEach(() => {
|
||||
rs.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] });
|
||||
window.sessionStorage.clear();
|
||||
clientMockState.listRuns.mockReset();
|
||||
clientMockState.listRuns.mockResolvedValue([ACTIVE_RUN]);
|
||||
clientMockState.joinStream.mockReset();
|
||||
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" } },
|
||||
),
|
||||
),
|
||||
);
|
||||
rs.spyOn(console, "error").mockImplementation(() => undefined);
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
rs.useRealTimers();
|
||||
rs.unstubAllGlobals();
|
||||
rs.restoreAllMocks();
|
||||
});
|
||||
|
||||
test("rejoins when the SDK's reconnect fails before the runs read", async () => {
|
||||
let resolveRuns!: (runs: Run[]) => void;
|
||||
clientMockState.listRuns.mockImplementation(
|
||||
() =>
|
||||
new Promise<Run[]>((resolve) => {
|
||||
resolveRuns = resolve;
|
||||
}),
|
||||
);
|
||||
clientMockState.joinStream.mockImplementationOnce(async function* () {
|
||||
throw new Error("network down");
|
||||
});
|
||||
clientMockState.joinStream.mockImplementation(async function* () {
|
||||
yield { event: "values", data: { messages: [] } };
|
||||
});
|
||||
window.sessionStorage.setItem("lg:stream:thread-1", "run-active");
|
||||
|
||||
const { unmount } = renderThread();
|
||||
await flushFrames();
|
||||
// The SDK made its single reconnect; it failed and nothing else joined.
|
||||
expect(clientMockState.joinStream).toHaveBeenCalledTimes(1);
|
||||
expect(clientMockState.joinStream.mock.calls[0]?.[1]).toBe("run-active");
|
||||
|
||||
act(() => resolveRuns([ACTIVE_RUN]));
|
||||
await flushFrames();
|
||||
|
||||
expect(clientMockState.joinStream).toHaveBeenCalledTimes(2);
|
||||
expect(clientMockState.joinStream.mock.calls[1]?.[1]).toBe("run-active");
|
||||
unmount();
|
||||
});
|
||||
|
||||
test("rejoins when the SDK's reconnect drops after the runs read", async () => {
|
||||
let dropStream!: () => void;
|
||||
clientMockState.joinStream.mockImplementationOnce(async function* () {
|
||||
yield { event: "values", data: { messages: [] } };
|
||||
await new Promise<void>((resolve) => {
|
||||
dropStream = resolve;
|
||||
});
|
||||
throw new Error("connection reset");
|
||||
});
|
||||
clientMockState.joinStream.mockImplementation(async function* () {
|
||||
yield { event: "values", data: { messages: [] } };
|
||||
});
|
||||
window.sessionStorage.setItem("lg:stream:thread-1", "run-active");
|
||||
|
||||
const { result, unmount } = renderThread();
|
||||
await flushFrames();
|
||||
expect(clientMockState.listRuns).toHaveBeenCalled();
|
||||
expect(result.current.thread.isLoading).toBe(true);
|
||||
expect(clientMockState.joinStream).toHaveBeenCalledTimes(1);
|
||||
|
||||
act(() => dropStream());
|
||||
await flushFrames();
|
||||
|
||||
expect(clientMockState.joinStream).toHaveBeenCalledTimes(2);
|
||||
expect(clientMockState.joinStream.mock.calls[1]?.[1]).toBe("run-active");
|
||||
unmount();
|
||||
});
|
||||
Loading…
x
Reference in New Issue
Block a user