From e317f7b8d9b2afb4c3925812d4774da602c9f8f3 Mon Sep 17 00:00:00 2001 From: Wu JiaCheng <134792033+patrick-andstar@users.noreply.github.com> Date: Wed, 29 Jul 2026 14:10:51 +0800 Subject: [PATCH] perf(streaming): batch main thread stream updates (#4410) * perf(streaming): batch stream updates * docs(streaming): clarify throttle runtime default --------- Co-authored-by: Willem Jiang --- frontend/AGENTS.md | 2 +- frontend/src/core/threads/hooks.ts | 3 + .../unit/core/threads/stream-throttle.test.ts | 62 +++++++++++++++++++ 3 files changed, 66 insertions(+), 1 deletion(-) create mode 100644 frontend/tests/unit/core/threads/stream-throttle.test.ts diff --git a/frontend/AGENTS.md b/frontend/AGENTS.md index 149ac5ea8..6cecb4cba 100644 --- a/frontend/AGENTS.md +++ b/frontend/AGENTS.md @@ -66,7 +66,7 @@ The frontend is a stateful chat application. Users create **threads** (conversat ### Data Flow 1. Optional composer helpers such as `core/input-polish` can rewrite the local draft before submission, and `core/voice-input` can transcribe browser microphone input into that same local draft; confirmed user input then flows to thread hooks (`core/threads/hooks.ts`) → LangGraph SDK streaming -2. Stream events update thread state (messages, artifacts, todos, goal) +2. Stream events update thread state (messages, artifacts, todos, goal). The main thread stream uses the LangGraph SDK's `throttle: true` mode so updates received in the same macrotask coalesce before React is notified; do not replace it with a numeric delay without validating the SDK's trailing-debounce behavior on a continuous stream. File-tool artifact auto-open work must run in an effect with timer cleanup; never schedule timers while rendering streamed `write_file` or `str_replace` updates. 3. `useThreadHistory` loads persisted conversation pages from `GET /api/threads/{id}/messages/page`, preserving the backend's thread-global event `seq`; rendering overlays checkpoint/live copies at their matching canonical identities (a summarized checkpoint may contain a protected early input plus a recent tail). Context-compaction rescue diffs every retained visible identity rather than slicing at the first anchor, and keeps a run-scoped ledger of committed visible messages so replacement updates and repeated rolling checkpoint windows cannot erase an already displayed step. The resolver suppresses checkpoint/transient prefixes whose canonical position is still behind an unloaded cursor page instead of collapsing that unknown gap before a recent anchor, then adds optimistic messages without timestamp re-sorting. History invalidation preserves already-loaded pages so their established ordering positions are not discarded. 4. Stop actions call the LangGraph SDK stream stop path; `core/threads/hooks.ts` invalidates current-thread, thread-history, token-usage, and sidebar/search caches immediately and schedules one follow-up refetch because SDK stop may finish via abort + fire-and-forget cancel before backend title finalization commits diff --git a/frontend/src/core/threads/hooks.ts b/frontend/src/core/threads/hooks.ts index f10455c5e..888a4a6c1 100644 --- a/frontend/src/core/threads/hooks.ts +++ b/frontend/src/core/threads/hooks.ts @@ -1467,9 +1467,12 @@ export function useThreadStream({ threadId: onStreamThreadId, reconnectOnMount: true, fetchStateHistory: { limit: 1 }, + // Batch stream updates received in the same macrotask without adding a + // fixed debounce interval that could delay a continuously active stream. // Coalesce same-tick stream events into one React notification. Only the // boolean tier is safe: the SDK's numeric tier is a trailing debounce that // starves UI updates while chunks keep arriving faster than the window. + // Keep explicit: SDK types claim @default true, but runtime uses throttle ?? false. throttle: true, onCreated(meta) { handleStreamStart(meta.thread_id, meta.run_id); diff --git a/frontend/tests/unit/core/threads/stream-throttle.test.ts b/frontend/tests/unit/core/threads/stream-throttle.test.ts new file mode 100644 index 000000000..026015664 --- /dev/null +++ b/frontend/tests/unit/core/threads/stream-throttle.test.ts @@ -0,0 +1,62 @@ +import { expect, rs, test } from "@rstest/core"; +import { QueryClient, QueryClientProvider } from "@tanstack/react-query"; +import { createElement } from "react"; +import { renderToStaticMarkup } from "react-dom/server"; + +import { I18nContext } from "@/core/i18n/context"; +import { DEFAULT_LOCAL_SETTINGS } from "@/core/settings/local"; + +const streamMockState = rs.hoisted(() => ({ + options: undefined as { throttle?: boolean | number } | undefined, +})); + +rs.mock("@langchain/langgraph-sdk/react", () => ({ + useStream: (options: { throttle?: boolean | number }) => { + streamMockState.options = options; + return { + isLoading: false, + messages: [], + stop: async () => undefined, + submit: async () => undefined, + values: { + artifacts: [], + messages: [], + title: "", + todos: [], + }, + }; + }, +})); + +test("batches stream subscription updates in a macrotask", async () => { + const { useThreadStream } = await import("@/core/threads/hooks"); + const queryClient = new QueryClient(); + + function StreamProbe() { + useThreadStream({ + context: DEFAULT_LOCAL_SETTINGS.context, + isMock: true, + threadId: "thread-1", + }); + return null; + } + + renderToStaticMarkup( + createElement( + QueryClientProvider, + { client: queryClient }, + createElement( + I18nContext.Provider, + { + value: { + locale: "en-US", + setLocale: () => undefined, + }, + }, + createElement(StreamProbe), + ), + ), + ); + + expect(streamMockState.options?.throttle).toBe(true); +});