perf(streaming): batch main thread stream updates (#4410)

* perf(streaming): batch stream updates

* docs(streaming): clarify throttle runtime default

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

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

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);

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);
+});