From 34b7a2d13bdc8d6d1970ad7b30087cf456a3f9c4 Mon Sep 17 00:00:00 2001 From: TYY-2005 Date: Sun, 11 Oct 2026 11:51:16 +0800 Subject: [PATCH] fix(frontend): preserve newer reconnect pointers during gap recovery (#6662) * fix(frontend): preserve newer reconnect pointers during gap recovery * test(frontend): cover same-run replay-gap pointer restoration --------- Co-authored-by: TYY-2005 <334029581+TYY-2005@users.noreply.github.com> Co-authored-by: Willem Jiang --- README.md | 2 + frontend/src/AGENTS.md | 6 + frontend/src/core/api/api-client.ts | 7 +- .../tests/unit/core/api/api-client.test.ts | 109 ++++++++++++++++++ 4 files changed, 123 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index f2f125d859..51af509e33 100644 --- a/README.md +++ b/README.md @@ -537,6 +537,8 @@ DeerFlow still uses `Forwarded` / `X-Forwarded-*` headers to recover the browser > [!IMPORTANT] > The Gateway still owns active run tasks in process, so production defaults to a single Gateway worker (`GATEWAY_WORKERS=1`). Multi-worker deployments require Postgres, the Redis stream bridge (`stream_bridge.type: redis`), `run_ownership.heartbeat_enabled: true`, and `run_events.backend: db`; process-local memory/JSONL event stores cannot enforce singleton delivery receipts across workers. Kubernetes replicas run one worker per Pod, which the worker count cannot see: declare them with `deployment.multi_instance: true` (or `DEER_FLOW_MULTI_INSTANCE=1`, which deploy tooling such as a Helm chart can set from its replica count) so the startup gate enforces the same prerequisites instead of staying inert. Declared instances that store credentials (`channel_connections`) must also share one at-rest `DEER_FLOW_CREDENTIALS_KEY` (the Helm chart and `make up` generate it; back it up, see [backend/docs/CONFIGURATION.md](backend/docs/CONFIGURATION.md#credentials-encryption-key-deer_flow_credentials_key)). The bridge shares SSE delivery and bounded `Last-Event-ID` replay across workers. When a valid reconnect cursor has been trimmed, or a subscriber that already established an empty-stream wait falls behind before its first delivery, Memory and Redis emit a machine-readable SSE `gap` event instead of silently returning a partial replay; the Web UI reloads durable thread/event state and resumes from the retained tail. Lease reconciliation marks runs from dead workers as errors, persists their delivery receipts, publishes the terminal stream marker, schedules retained-stream cleanup, and updates the affected thread status. SSE, `/wait`, and internal stream consumers use `stream_bridge.heartbeat_interval_seconds` (default `15`) for idle liveness checks; changing it requires a Gateway restart. Malformed Redis reconnect IDs live-tail new events instead of replaying the retained buffer, and the rolling retained-buffer TTL (`stream_ttl_seconds`) remains a cleanup safety net rather than a run timeout. Failed-login counters and lockouts for `POST /api/v1/auth/login/local` (`auth.local.max_login_attempts` / `lockout_seconds`) are kept in the shared `login_throttle` table whenever the application database is SQLite or Postgres (`auth.local.throttle_storage: auto`, the default), so every replica enforces one lockout per client IP; `memory` keeps the historical per-process counter, which under N replicas hands an attacker N × `max_login_attempts` guesses and logs a startup warning. IM chat-to-thread bindings live in the shared `channel_thread_bindings` table whenever the application database is SQLite or Postgres (an existing `channels/store.json` is imported once at startup and renamed `store.json.migrated`); the remaining IM channel state (each instance's platform connections, follow-up buffers) still needs its own multi-worker coordination. To try this topology on one machine, `make dev-multi` starts two Gateways on shared Postgres, Redis and `DEER_FLOW_HOME`, and `make dev-multi-check` runs the cross-instance checks; see the local two-Gateway harness in [backend/docs/CONFIGURATION.md](backend/docs/CONFIGURATION.md#local-two-gateway-harness). > +> During SSE replay-gap recovery, the Web UI preserves a reconnect pointer claimed by another run while durable state is loading. An older recovery cannot overwrite that pointer when it rejoins its own stream. Even a stale foreign pointer is preserved, so the next reload may first check that run before active-run recovery rejoins, adding a reconnect cycle. +> > In single-process JSONL deployments, cancelling an admitted event-store mutation waits for its background file I/O, rollback, and bookkeeping to settle before releasing the thread write lock. This prevents an older cancelled write from recreating deleted records or rolling back a later successful write. Cancellation can therefore wait on slow storage; it does not stop an in-flight filesystem operation. Callers still waiting to acquire the lock can cancel without starting a mutation. A batch spanning multiple threads drains its current thread group before propagating cancellation; subsequent thread groups do not start. > > After a run publishes its terminal stream marker, its process-local `RunRecord` remains available for the existing five-minute grace period before cleanup; durable run history remains available through `RunStore`, while the stream bridge retains its delivery tail on its separate cleanup schedule. diff --git a/frontend/src/AGENTS.md b/frontend/src/AGENTS.md index 0ec70b6dc7..5a8e90e092 100644 --- a/frontend/src/AGENTS.md +++ b/frontend/src/AGENTS.md @@ -191,6 +191,12 @@ export on a page error or non-advancing cursor; demos keep their fixture path. - **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. 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. + Replay-gap recovery restores its tab-local run pointer only if the slot is empty + or still belongs to that run; never overwrite another run's pointer after an + asynchronous durable-state read. A stale foreign pointer may cost an extra + run lookup/rejoin on reload; recovery adds no server probe. The same-named + `rememberReconnectRun` in `core/threads/hooks.ts` stays unconditional: + active-run rejoin makes a fresh claim that must win over stored metadata. - **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. - Web-search steps accept either a provider result array or a Webz-style `{ query, returned_results, results }` envelope. Unwrap only in the web-search renderer; skip entries without string titles/URLs, preserve other tool payloads and apply the same link allowlist to both formats. UI regressions pin source titles/links and browser reloads, including malformed neighboring entries. diff --git a/frontend/src/core/api/api-client.ts b/frontend/src/core/api/api-client.ts index 1468accafd..8791a064d4 100644 --- a/frontend/src/core/api/api-client.ts +++ b/frontend/src/core/api/api-client.ts @@ -328,7 +328,12 @@ function rememberReconnectRun( if (typeof window === "undefined" || !threadId) return; try { - window.sessionStorage.setItem(`lg:stream:${threadId}`, runId); + const storage = window.sessionStorage; + const key = `lg:stream:${threadId}`; + const rememberedRunId = storage.getItem(key); + // Another run may claim the pointer while durable state is loading. + if (rememberedRunId !== null && rememberedRunId !== runId) return; + storage.setItem(key, runId); } catch { // Ignore storage access failures so gap recovery remains usable. } diff --git a/frontend/tests/unit/core/api/api-client.test.ts b/frontend/tests/unit/core/api/api-client.test.ts index 6b9f1cbd34..ad516c25e5 100644 --- a/frontend/tests/unit/core/api/api-client.test.ts +++ b/frontend/tests/unit/core/api/api-client.test.ts @@ -899,8 +899,116 @@ test("recovers a join stream gap from durable state and resumes after the retain expect(new Headers(recoveryRequests[1]?.headers).get("Last-Event-ID")).toBe( "3-0", ); + expect(sessionStorage.getItem("lg:stream:thread-1")).toBe("run-1"); }); +test.each([ + ["initial", "another run"], + ["join", "another run"], + ["initial", "the same run"], + ["join", "the same run"], +])( + "handles %s stream gap recovery when %s claims the reconnect pointer", + async (kind, claimant) => { + const sessionStorage = makeSessionStorage(); + const threadId = "thread-gap-pointer"; + const runId = "run-gap-pointer"; + const claimedRunId = claimant === "the same run" ? runId : "run-newer"; + const key = "lg:stream:" + threadId; + sessionStorage.setItem(key, runId); + const gap = { + code: "stream_replay_gap", + run_id: runId, + requested_event_id: "1-0", + earliest_available_event_id: "2-0", + latest_available_event_id: "3-0", + recovery: "reload_durable_state", + }; + let startStateRead: () => void = () => undefined; + let finishStateRead: (response: Response) => void = () => undefined; + const stateReadStarted = new Promise((resolve) => { + startStateRead = resolve; + }); + const stateResponse = new Promise((resolve) => { + finishStateRead = resolve; + }); + const recoveryHeaders: Headers[] = []; + const fetchFn = rs.fn(async (url: string | URL, init?: RequestInit) => { + const path = String(url); + if (path.endsWith("/runs/" + runId)) { + return new Response(JSON.stringify({ status: "running" })); + } + if (path.endsWith("/threads/" + threadId + "/state")) { + startStateRead(); + return stateResponse; + } + if ( + path.endsWith("/runs/stream") || + (path.includes("/runs/" + runId + "/stream") && + !new Headers(init?.headers).has("Last-Event-ID")) + ) { + return makeSSEResponse( + "event: gap\ndata: " + JSON.stringify(gap) + "\n\n", + { + "Content-Location": "/threads/" + threadId + "/runs/" + runId, + }, + ); + } + if (path.includes("/runs/" + runId + "/stream")) { + recoveryHeaders.push(new Headers(init?.headers)); + return makeSSEResponse("event: end\ndata: null\n\n"); + } + throw new Error("Unexpected request: " + path); + }); + rs.stubGlobal("window", { + location: { origin: "http://localhost:2026" }, + sessionStorage, + }); + rs.stubGlobal("fetch", fetchFn); + + const client = getAPIClient(true); + const stream = + kind === "initial" + ? client.runs.stream(threadId, "lead_agent", {}) + : client.runs.joinStream(threadId, runId); + await expect(stream.next()).resolves.toEqual({ + done: false, + value: { event: "custom", data: { type: "stream_replay_gap", ...gap } }, + }); + expect(sessionStorage.getItem(key)).toBeNull(); + + const snapshot = stream.next(); + await stateReadStarted; + sessionStorage.setItem(key, claimedRunId); + sessionStorage.setItem.mockClear(); + sessionStorage.removeItem.mockClear(); + finishStateRead(new Response(JSON.stringify({ values: { messages: [] } }))); + + await expect(snapshot).resolves.toEqual({ + done: false, + value: { event: "values", data: { messages: [] } }, + }); + await expect(stream.next()).resolves.toEqual({ + done: false, + value: { event: "end", data: null }, + }); + expect(recoveryHeaders).toHaveLength(1); + expect(recoveryHeaders[0]?.get("Last-Event-ID")).toBe("3-0"); + expect(sessionStorage.getItem(key)).toBe(claimedRunId); + if (claimedRunId === runId) { + expect(sessionStorage.setItem).toHaveBeenCalledTimes(1); + expect(sessionStorage.setItem).toHaveBeenCalledWith(key, runId); + } else { + expect(sessionStorage.setItem).not.toHaveBeenCalled(); + } + expect(sessionStorage.removeItem).not.toHaveBeenCalled(); + await expect(stream.next()).resolves.toEqual({ + done: true, + value: undefined, + }); + }, +); + test("recovers a gap emitted by the initial run stream", async () => { const sessionStorage = makeSessionStorage(); const gap = { @@ -962,6 +1070,7 @@ test("recovers a gap emitted by the initial run stream", async () => { { event: "end", data: null }, ]); expect(recoveryHeaders[0]?.get("Last-Event-ID")).toBe("5-0"); + expect(sessionStorage.getItem("lg:stream:thread-2")).toBe("run-2"); }); test("recovers from stream replay gap with null retained bounds", async () => {