From a9402b0c65f4cc15a946080473679c1bfb45d603 Mon Sep 17 00:00:00 2001 From: Jholly2008 Date: Fri, 7 Aug 2026 15:48:06 +0800 Subject: [PATCH 1/2] fix(frontend): prevent run creation retries --- frontend/AGENTS.md | 2 +- frontend/src/core/api/api-client.ts | 13 +++++++- .../tests/unit/core/api/api-client.test.ts | 30 +++++++++++++++++++ 3 files changed, 43 insertions(+), 2 deletions(-) diff --git a/frontend/AGENTS.md b/frontend/AGENTS.md index c71ad74306..84ae496a77 100644 --- a/frontend/AGENTS.md +++ b/frontend/AGENTS.md @@ -124,7 +124,7 @@ Edit-and-rerun is deliberately latest-turn-only. `core/messages/utils.ts::getLat - **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 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. Keep this boundary aligned with the backend request schema; `messages` and `events` are not supported and must not be forwarded. -- **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 rejoins after the server-provided retained tail, 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. +- **Run creation and SSE replay** are handled in `core/api/api-client.ts`. The initial `POST /runs/stream` uses a dedicated zero-retry client because creating a run is not idempotent; reads and SSE joins keep the SDK's normal retries. The stream wrappers cover 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 rejoins after the server-provided retained tail, 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. - **Environment validation** uses `@t3-oss/env-nextjs` with Zod schemas (`src/env.js`). Skip with `SKIP_ENV_VALIDATION=1` diff --git a/frontend/src/core/api/api-client.ts b/frontend/src/core/api/api-client.ts index f557edcee2..e3b57104d6 100644 --- a/frontend/src/core/api/api-client.ts +++ b/frontend/src/core/api/api-client.ts @@ -330,7 +330,18 @@ 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. Keep the default retries for reads and SSE joins, but + // disable them for the initial POST /runs/stream request. + const runCreationClient = new LangGraphClient({ + apiUrl, + callerOptions: { maxRetries: 0 }, + onRequest: injectCsrfHeader, + }); + const originalRunStream = runCreationClient.runs.stream.bind( + runCreationClient.runs, + ); 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 e494569c6f..5d4c77858b 100644 --- a/frontend/tests/unit/core/api/api-client.test.ts +++ b/frontend/tests/unit/core/api/api-client.test.ts @@ -84,6 +84,36 @@ test("ignores reconnect metadata storage access failures", () => { expect(() => clearReconnectRun("thread-1", "run-1")).not.toThrow(); }); +test("does not retry run creation after an ambiguous gateway failure", async () => { + const sessionStorage = makeSessionStorage(); + let attempts = 0; + const fetchFn = rs.fn(async () => { + attempts += 1; + 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("HTTP 504"); + expect(fetchFn).toHaveBeenCalledTimes(1); +}); + test("clears stale reconnect metadata when join stream cannot be resumed", async () => { const sessionStorage = makeSessionStorage(); sessionStorage.setItem("lg:stream:thread-1", "run-1"); From 265c49b3bf33153e62f05e6879d3faed7764fdc1 Mon Sep 17 00:00:00 2001 From: Jholly2008 Date: Fri, 7 Aug 2026 16:37:35 +0800 Subject: [PATCH 2/2] test(frontend): cover safe run stream reconnection --- frontend/AGENTS.md | 2 +- frontend/src/core/api/api-client.ts | 5 +- .../tests/unit/core/api/api-client.test.ts | 69 +++++++++++++++++++ 3 files changed, 73 insertions(+), 3 deletions(-) diff --git a/frontend/AGENTS.md b/frontend/AGENTS.md index 84ae496a77..bb4e987e48 100644 --- a/frontend/AGENTS.md +++ b/frontend/AGENTS.md @@ -124,7 +124,7 @@ Edit-and-rerun is deliberately latest-turn-only. `core/messages/utils.ts::getLat - **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 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. Keep this boundary aligned with the backend request schema; `messages` and `events` are not supported and must not be forwarded. -- **Run creation and SSE replay** are handled in `core/api/api-client.ts`. The initial `POST /runs/stream` uses a dedicated zero-retry client because creating a run is not idempotent; reads and SSE joins keep the SDK's normal retries. The stream wrappers cover 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 rejoins after the server-provided retained tail, 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. +- **Run creation and SSE replay** are handled in `core/api/api-client.ts`. The initial `POST /runs/stream` uses a dedicated client with request-level retries disabled because creating a run is not idempotent. The SDK's independent SSE recovery remains enabled: after an established stream fails, it resumes with a `GET` to the server-provided `Location` and forwards `Last-Event-ID`; reads and explicit SSE joins use the normal client. The stream wrappers cover 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 rejoins after the server-provided retained tail, 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. - **Environment validation** uses `@t3-oss/env-nextjs` with Zod schemas (`src/env.js`). Skip with `SKIP_ENV_VALIDATION=1` diff --git a/frontend/src/core/api/api-client.ts b/frontend/src/core/api/api-client.ts index e3b57104d6..9b28b715fa 100644 --- a/frontend/src/core/api/api-client.ts +++ b/frontend/src/core/api/api-client.ts @@ -332,8 +332,9 @@ function createCompatibleClient(isMock?: boolean): LangGraphClient { // 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. Keep the default retries for reads and SSE joins, but - // disable them for the initial POST /runs/stream request. + // original request. Disable request-level retries for this stream client; + // the SDK's independent SSE recovery still resumes established streams with + // GET requests to the server-provided Location. const runCreationClient = new LangGraphClient({ apiUrl, callerOptions: { maxRetries: 0 }, diff --git a/frontend/tests/unit/core/api/api-client.test.ts b/frontend/tests/unit/core/api/api-client.test.ts index 5d4c77858b..49b748df61 100644 --- a/frontend/tests/unit/core/api/api-client.test.ts +++ b/frontend/tests/unit/core/api/api-client.test.ts @@ -114,6 +114,75 @@ test("does not retry run creation after an ambiguous gateway failure", async () expect(fetchFn).toHaveBeenCalledTimes(1); }); +test("reconnects an interrupted run stream with GET without recreating the run", async () => { + 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<{ + method: string | undefined; + lastEventId: string | null; + }> = []; + const fetchFn = rs.fn(async (_url: string | URL, init?: RequestInit) => { + const headers = new Headers(init?.headers); + requests.push({ + 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", + }, + }); + } + 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 }> = []; + 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?.(); + } + } + + expect(received).toEqual([ + { id: "1-0", event: "custom", data: { phase: "started" } }, + { id: "2-0", event: "end", data: null }, + ]); + expect(requests).toEqual([ + { method: "POST", lastEventId: null }, + { method: "GET", lastEventId: "1-0" }, + ]); +}); + test("clears stale reconnect metadata when join stream cannot be resumed", async () => { const sessionStorage = makeSessionStorage(); sessionStorage.setItem("lg:stream:thread-1", "run-1");