From 8121dbcb572d8248acb8665c9c4ecae0fd5178d8 Mon Sep 17 00:00:00 2001 From: Jholly Date: Fri, 25 Sep 2026 21:46:34 +0800 Subject: [PATCH] fix(frontend): prevent run creation retries (#4725) * fix(frontend): prevent run creation retries * test(frontend): cover safe run stream reconnection --------- Co-authored-by: Willem Jiang --- frontend/src/AGENTS.md | 1 + frontend/src/core/api/api-client.ts | 31 ++- .../tests/unit/core/api/api-client.test.ts | 176 ++++++++++++++++++ 3 files changed, 206 insertions(+), 2 deletions(-) diff --git a/frontend/src/AGENTS.md b/frontend/src/AGENTS.md index a4b98fee51..644a2bc8f5 100644 --- a/frontend/src/AGENTS.md +++ b/frontend/src/AGENTS.md @@ -151,6 +151,7 @@ Array previews coalesce consecutive generated markers only at the end into one o - **Thread hooks** (`useThreadStream`, `useSubmitThread`, `useThreads`) are the primary API interface - **Thread routes** — construct Web UI chat paths through `core/threads/utils.ts::pathOfThread()`, which percent-encodes both custom agent names and thread IDs before inserting them into route segments - **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 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. - **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. diff --git a/frontend/src/core/api/api-client.ts b/frontend/src/core/api/api-client.ts index bda91372cd..1468accafd 100644 --- a/frontend/src/core/api/api-client.ts +++ b/frontend/src/core/api/api-client.ts @@ -1,6 +1,9 @@ "use client"; -import { Client as LangGraphClient } from "@langchain/langgraph-sdk/client"; +import { + Client as LangGraphClient, + RunsClient, +} from "@langchain/langgraph-sdk/client"; import { getLangGraphBaseURL } from "../config"; import { isStaticWebsiteOnly } from "../static-mode"; @@ -440,6 +443,14 @@ async function* handleInactiveRunStream({ } } +// Reuse the SDK's retry budget, backoff and HTTP error handling for recovery +// requests that have already been prepared by the run-creation client. +class StreamRecoveryClient extends RunsClient { + fetchWithRetries(...args: Parameters): Promise { + return this.asyncCaller.fetch(...args); + } +} + function createCompatibleClient(isMock?: boolean): LangGraphClient { if (isStaticWebsiteOnly() && !isMock) { return createStaticClient(); @@ -451,7 +462,23 @@ function createCompatibleClient(isMock?: boolean): LangGraphClient { onRequest: injectCsrfHeader, }); - const originalRunStream = client.runs.stream.bind(client.runs); + // Creating a run is not idempotent. Retrying an ambiguous gateway failure + // can create the same run more than once after the backend accepted the + // original request. The SDK also uses this client's transport for recovery + // GETs, which must retain normal HTTP retries (including transient 5xx). + const streamRecoveryClient = new StreamRecoveryClient({ apiUrl }); + const runCreationClient = new RunsClient({ + apiUrl, + callerOptions: { + maxRetries: 0, + fetch: (...args: Parameters) => + args[1]?.method === "GET" + ? streamRecoveryClient.fetchWithRetries(...args) + : fetch(...args), + }, + onRequest: injectCsrfHeader, + }); + const originalRunStream = runCreationClient.stream.bind(runCreationClient); const originalJoinStream = client.runs.joinStream.bind(client.runs); // Preserve the SDK's lazy AsyncIterable contract. Its StreamManager consumes // this return value with `for await`, so run creation still starts on first diff --git a/frontend/tests/unit/core/api/api-client.test.ts b/frontend/tests/unit/core/api/api-client.test.ts index f4aea4283b..9d738feba6 100644 --- a/frontend/tests/unit/core/api/api-client.test.ts +++ b/frontend/tests/unit/core/api/api-client.test.ts @@ -84,6 +84,182 @@ test("ignores reconnect metadata storage access failures", () => { expect(() => clearReconnectRun("thread-1", "run-1")).not.toThrow(); }); +test.each(["http", "network"])( + "does not retry run creation after an ambiguous %s failure", + async (failure) => { + const sessionStorage = makeSessionStorage(); + let attempts = 0; + const fetchFn = rs.fn(async () => { + attempts += 1; + if (failure === "network") { + throw new TypeError("connection interrupted"); + } + const status = attempts === 1 ? 504 : 400; + return new Response(JSON.stringify({ detail: "request failed" }), { + status, + }); + }); + rs.stubGlobal("window", { + location: { origin: "http://localhost:2026" }, + sessionStorage, + }); + rs.stubGlobal("fetch", fetchFn); + + const consume = async () => { + for await (const entry of getAPIClient(true).runs.stream( + "thread-no-retry", + "lead_agent", + { input: { messages: [] } }, + )) { + void entry; + } + }; + + await expect(consume()).rejects.toThrow( + failure === "http" ? "HTTP 504" : "connection interrupted", + ); + expect(fetchFn).toHaveBeenCalledTimes(1); + }, +); + +test.each([0, 1, 4, 5])( + "preserves the recovery GET retry budget without recreating the run (%s failures)", + async (recoveryFailures) => { + const sessionStorage = makeSessionStorage(); + const encoder = new TextEncoder(); + let interruptStream: (() => void) | undefined; + const interruptedBody = new ReadableStream({ + start(controller) { + controller.enqueue( + encoder.encode( + 'id: 1-0\nevent: custom\ndata: {"phase":"started"}\n\n', + ), + ); + interruptStream = () => { + controller.error(new TypeError("connection interrupted")); + }; + }, + }); + const requests: Array<{ + url: string; + method: string | undefined; + lastEventId: string | null; + }> = []; + const fetchFn = rs.fn(async (_url: string | URL, init?: RequestInit) => { + const headers = new Headers(init?.headers); + requests.push({ + url: String(_url), + method: init?.method, + lastEventId: headers.get("Last-Event-ID"), + }); + if (requests.length === 1) { + return new Response(interruptedBody, { + status: 200, + headers: { + "Content-Type": "text/event-stream", + "Content-Location": "/threads/thread-reconnect/runs/run-reconnect", + Location: + "/threads/thread-reconnect/runs/run-reconnect/stream?stream_mode=custom", + }, + }); + } + if (requests.length <= recoveryFailures + 1) { + return new Response(JSON.stringify({ detail: "gateway timeout" }), { + status: 504, + }); + } + return makeSSEResponse("id: 2-0\nevent: end\ndata: null\n\n"); + }); + rs.stubGlobal("window", { + location: { origin: "http://localhost:2026" }, + sessionStorage, + }); + rs.stubGlobal("fetch", fetchFn); + rs.stubGlobal("setTimeout", (callback: () => void) => { + callback(); + return 0; + }); + + const received: Array<{ id?: string; event: string; data: unknown }> = []; + const consume = async () => { + for await (const entry of getAPIClient(true).runs.stream( + "thread-reconnect", + "lead_agent", + { input: { messages: [] } }, + )) { + received.push(entry); + if ("id" in entry && entry.id === "1-0") { + interruptStream?.(); + } + } + }; + // The SDK retries four times after the initial HTTP attempt. Exhaustion + // must surface the error instead of starting a new run or retrying forever. + if (recoveryFailures === 5) { + await expect(consume()).rejects.toThrow("HTTP 504"); + } else { + await consume(); + } + + expect(received).toEqual([ + { id: "1-0", event: "custom", data: { phase: "started" } }, + ...(recoveryFailures === 5 + ? [] + : [{ id: "2-0", event: "end", data: null }]), + ]); + expect(requests).toEqual([ + { + url: "http://localhost:2026/mock/api/threads/thread-reconnect/runs/stream", + method: "POST", + lastEventId: null, + }, + ...Array.from({ length: Math.min(recoveryFailures + 1, 5) }, () => ({ + url: "http://localhost:2026/mock/api/threads/thread-reconnect/runs/run-reconnect/stream?stream_mode=custom", + method: "GET", + lastEventId: "1-0", + })), + ]); + }, +); + +test("keeps retries for reads and explicit stream joins", async () => { + const attempts = new Map(); + const fetchFn = rs.fn(async (url: string | URL) => { + const path = new URL(url).pathname; + const attempt = (attempts.get(path) ?? 0) + 1; + attempts.set(path, attempt); + if (attempt === 1) { + return new Response("gateway timeout", { status: 504 }); + } + return path.endsWith("/stream") + ? makeSSEResponse("event: end\ndata: null\n\n") + : Response.json({ status: "running" }); + }); + rs.stubGlobal("window", { + location: { origin: "http://localhost:2026" }, + sessionStorage: makeSessionStorage(), + }); + rs.stubGlobal("fetch", fetchFn); + rs.stubGlobal("setTimeout", (callback: () => void) => { + callback(); + return 0; + }); + + const received: Array<{ event: string; data: unknown }> = []; + for await (const entry of getAPIClient(true).runs.joinStream( + "thread-read-retry", + "run-read-retry", + )) { + received.push(entry); + } + + expect(received).toEqual([{ event: "end", data: null }]); + expect([...attempts.entries()]).toEqual([ + ["/mock/api/threads/thread-read-retry/runs/run-read-retry", 2], + ["/mock/api/threads/thread-read-retry/runs/run-read-retry/stream", 2], + ]); +}); + test("clears stale reconnect metadata when join stream cannot be resumed", async () => { const sessionStorage = makeSessionStorage(); sessionStorage.setItem("lg:stream:thread-1", "run-1");