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 <willem.jiang@gmail.com>

Author: TYY-2005

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.

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.

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.
   }

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<void>((resolve) => {
+      startStateRead = resolve;
+    });
+    const stateResponse = new Promise<Response>((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 () => {