diff --git a/frontend/src/core/threads/hooks.ts b/frontend/src/core/threads/hooks.ts index efcae9588..4a9525586 100644 --- a/frontend/src/core/threads/hooks.ts +++ b/frontend/src/core/threads/hooks.ts @@ -140,6 +140,10 @@ export function buildThreadSubmitMessages({ ]; } +// Stable identity for "no optimistic messages" so the merged-messages memo +// below is not invalidated by a fresh empty array on every render. +const EMPTY_MESSAGES: Message[] = []; + const EMPTY_THREAD_VALUES: AgentThreadState = { title: "", messages: [], @@ -650,19 +654,22 @@ export function mergeTransientHistoryBridge( export function mergeTransientHistoryBridgeOrder( currentOrder: readonly string[], capturedMessages: Message[], -): string[] { +): readonly string[] { const capturedOrder = dedupeMessagesByIdentity(capturedMessages) .map(messageIdentity) .filter(isNonEmptyString); - const merged = [...currentOrder]; + // Clone lazily and return the input when nothing is appended: this runs per + // render while the bridge is active, and a fresh array would invalidate the + // coalesced render memo on every chunk (#4409 Phase 1). + let merged: string[] | null = null; const seen = new Set(currentOrder); for (const identity of capturedOrder) { if (!seen.has(identity)) { seen.add(identity); - merged.push(identity); + (merged ??= [...currentOrder]).push(identity); } } - return merged; + return merged ?? currentOrder; } export function resolveThreadTransientHistoryBridge( @@ -753,6 +760,136 @@ export function getSummarizationMiddlewareMessages( return undefined; } +export const STREAM_RENDER_COALESCE_MS = 80; + +export type CoalesceDecision = + | { action: "flush-now" } + | { action: "schedule"; delayMs: number } + | { action: "wait" }; + +/** + * Decide how an incoming stream update reaches the rendered snapshot: flush + * immediately once a full interval has elapsed (leading edge), otherwise + * schedule exactly one trailing flush for the interval remainder. Unlike a + * debounce, the delay never extends past the interval, so a dense stream can + * never starve rendering. + */ +export function decideCoalesce( + nowMs: number, + lastFlushMs: number, + intervalMs: number, + hasPendingTimer: boolean, +): CoalesceDecision { + if (nowMs - lastFlushMs >= intervalMs) { + return { action: "flush-now" }; + } + if (hasPendingTimer) { + return { action: "wait" }; + } + return { action: "schedule", delayMs: intervalMs - (nowMs - lastFlushMs) }; +} + +/** + * While a run is streaming, expose the messages array as a snapshot that + * updates at most once per interval instead of once per SSE chunk, so the + * merge/group/render pipeline runs per frame budget rather than per token + * (#4409 Phase 1). When the stream is idle the latest array passes straight + * through, keeping non-stream updates immediate. + */ +function sameMessageArray(a: Message[], b: Message[]): boolean { + return ( + a === b || + (a.length === b.length && a.every((message, index) => message === b[index])) + ); +} + +export function useCoalescedStreamMessages( + messages: Message[], + isStreaming: boolean, + intervalMs: number = STREAM_RENDER_COALESCE_MS, +): Message[] { + // `null` means "no snapshot belongs to the current stream": the live array is + // returned until the leading-edge flush lands, so a snapshot left over from an + // earlier stream can never be painted. This hook outlives thread switches (the + // chat page deliberately avoids re-mounting, see its `onStart` comment), so a + // retained snapshot would otherwise be another thread's messages. + const [snapshot, setSnapshot] = useState(null); + const latestRef = useRef(messages); + latestRef.current = messages; + // Monotonic clock: a wall-clock step (NTP, sleep/wake) between two reads + // would otherwise be added to the remaining interval and stall the flush for + // the length of the jump. -Infinity means "never flushed", so the first + // update of a stream always takes the leading edge. + const lastFlushRef = useRef(Number.NEGATIVE_INFINITY); + const timerRef = useRef | null>(null); + + // Every publication goes through the shallow-equality guard: inputs whose + // identity churns without content change (the SDK getter mints fresh arrays) + // must not re-trigger renders, or this effect would setState-loop. + const publish = useCallback(() => { + setSnapshot((previous) => + previous !== null && sameMessageArray(previous, latestRef.current) + ? previous + : latestRef.current, + ); + }, []); + + const clearPendingFlush = useCallback(() => { + if (timerRef.current !== null) { + clearTimeout(timerRef.current); + timerRef.current = null; + } + }, []); + + useEffect(() => { + if (!isStreaming) { + clearPendingFlush(); + // Drop the flush baseline so the leading edge is per stream rather than + // per hook instance: a run starting within one interval of the previous + // one must not have its first frame deferred. Dropping the snapshot with + // it costs one render per stream end, versus one per idle message change + // if the snapshot were instead kept in sync while nothing reads it. + lastFlushRef.current = Number.NEGATIVE_INFINITY; + setSnapshot((previous) => (previous === null ? previous : null)); + return; + } + const now = performance.now(); + const decision = decideCoalesce( + now, + lastFlushRef.current, + intervalMs, + timerRef.current !== null, + ); + if (decision.action === "flush-now") { + // A trailing timer can still be armed here: timers fire late under + // main-thread load, which is exactly when a chunk overtakes one. Leaving + // it would publish a second time and slip the next interval forward. + clearPendingFlush(); + lastFlushRef.current = now; + publish(); + } else if (decision.action === "schedule") { + timerRef.current = setTimeout(() => { + timerRef.current = null; + // Read the clock again: timers fire late under load, and the next + // interval must start from the real flush. + lastFlushRef.current = performance.now(); + publish(); + }, decision.delayMs); + } + }, [messages, isStreaming, intervalMs, publish, clearPendingFlush]); + + useEffect( + () => () => { + if (timerRef.current !== null) { + clearTimeout(timerRef.current); + } + }, + [], + ); + + return isStreaming && snapshot !== null ? snapshot : messages; +} + export function upsertThreadInSearchCache( queryClient: QueryClient, thread: AgentThread, @@ -1073,6 +1210,10 @@ export function useThreadStream({ threadId: onStreamThreadId, reconnectOnMount: true, fetchStateHistory: { limit: 1 }, + // 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. + throttle: true, onCreated(meta) { handleStreamStart(meta.thread_id, meta.run_id); const now = new Date().toISOString(); @@ -1271,16 +1412,18 @@ export function useThreadStream({ const hasVisibleStreamState = Boolean(threadId) || liveMessagesThreadId === currentViewThreadId; - const persistedMessages = useMemo( - () => - hasVisibleStreamState - ? thread.messages.filter( - (message) => - !message.id || !pendingSupersededMessageIds.has(message.id), - ) - : [], - [hasVisibleStreamState, pendingSupersededMessageIds, thread.messages], - ); + const persistedMessages = useMemo(() => { + if (!hasVisibleStreamState) { + return EMPTY_MESSAGES; + } + const filtered = thread.messages.filter( + (message) => !message.id || !pendingSupersededMessageIds.has(message.id), + ); + // The SDK getter mints a fresh [] on every read while the stream has no + // values; normalize to a stable identity so downstream effects keyed on + // this array cannot re-fire (and setState-loop) on idle renders. + return filtered.length === 0 ? EMPTY_MESSAGES : filtered; + }, [hasVisibleStreamState, pendingSupersededMessageIds, thread.messages]); const visibleHistory = useMemo( () => (threadId ? history : []), [history, threadId], @@ -1298,7 +1441,7 @@ export function useThreadStream({ // Full identity order of each captured checkpoint. Confirmed bridge entries // are pruned from the message buffer, but remain here as non-rendering // anchors so an older rescue can be placed before a newest-first page. - const transientHistoryOrderRef = useRef([]); + const transientHistoryOrderRef = useRef([]); const transientHistoryThreadIdRef = useRef(null); const summarizedRef = useRef>(null); // Track human message count before sending to prevent clearing optimistic @@ -1705,11 +1848,23 @@ export function useThreadStream({ messagesRef.current = persistedMessages; } - const visibleOptimisticMessages = getVisibleOptimisticMessages( + // Render-facing coalesced snapshot. Refs, counters and usage tracking keep + // consuming the per-chunk array above so lifecycle semantics (optimistic + // clearing, summarization capture, token-usage baselines) are unchanged. + const renderMessages = useCoalescedStreamMessages( + persistedMessages, + thread.isLoading, + ); + + const rawVisibleOptimisticMessages = getVisibleOptimisticMessages( optimisticThreadId === currentViewThreadId ? optimisticMessages : [], prevHumanMsgCountRef.current, humanMessageCount, ); + const visibleOptimisticMessages = + rawVisibleOptimisticMessages.length === 0 + ? EMPTY_MESSAGES + : rawVisibleOptimisticMessages; const transientHistoryOrder = transientHistoryBridgeRef.current.length > 0 && @@ -1735,18 +1890,30 @@ export function useThreadStream({ } }, [persistedMessages, threadId]); - const effectiveHistory = resolveThreadTransientHistoryBridge( - visibleHistory, - transientHistoryBridgeRef.current, - transientHistoryThreadIdRef.current, + // The transient-bridge refs mutate in lockstep with stream/history updates + // already captured by these deps, and resolveTransientHistoryBridge is + // idempotent for entries canonical history has absorbed, so memoizing on the + // coalesced snapshot cannot pin a stale bridge. + const mergedMessages = useMemo(() => { + const effectiveHistory = resolveThreadTransientHistoryBridge( + visibleHistory, + transientHistoryBridgeRef.current, + transientHistoryThreadIdRef.current, + threadId, + transientHistoryOrder, + ); + return mergeMessages( + effectiveHistory, + renderMessages, + visibleOptimisticMessages, + ); + }, [ + renderMessages, threadId, transientHistoryOrder, - ); - const mergedMessages = mergeMessages( - effectiveHistory, - persistedMessages, + visibleHistory, visibleOptimisticMessages, - ); + ]); const pendingUsageMessages = thread.isLoading ? getMessagesAfterBaseline( persistedMessages, diff --git a/frontend/tests/e2e/artifact-stream-state.spec.ts b/frontend/tests/e2e/artifact-stream-state.spec.ts index 2768bae6c..aff3f451c 100644 --- a/frontend/tests/e2e/artifact-stream-state.spec.ts +++ b/frontend/tests/e2e/artifact-stream-state.spec.ts @@ -17,6 +17,18 @@ const THREAD_MESSAGES = [ content: "Created a markdown report.", }, ]; +const FOLLOW_UP_MESSAGES = [ + { + type: "human", + id: "msg-human-artifact-follow-up", + content: [{ type: "text", text: "Continue" }], + }, + { + type: "ai", + id: "msg-ai-artifact-follow-up", + content: "Updated response while the artifact list is omitted.", + }, +]; function streamWithoutArtifacts(route: Route) { const events = [ @@ -27,18 +39,7 @@ function streamWithoutArtifacts(route: Route) { { event: "values", data: { - messages: [ - { - type: "human", - id: "msg-human-artifact-follow-up", - content: [{ type: "text", text: "Continue" }], - }, - { - type: "ai", - id: "msg-ai-artifact-follow-up", - content: "Updated response while the artifact list is omitted.", - }, - ], + messages: FOLLOW_UP_MESSAGES, }, }, ]; @@ -68,7 +69,60 @@ test("keeps artifact trigger after stream values omit artifacts", async ({ ], }); + // A real gateway persists the run's messages, so the SDK's end-of-run state + // refetch returns them. This spec's stream route bypasses the shared mock's + // thread upsert, which would otherwise leave post-run state looking like the + // run never happened — an impossible backend state that only a same-tick + // render could paper over. Serve the persisted turn once the run has started; + // stream values still omit artifacts, which is the invariant under test. + let runStarted = false; + await page.route("**/api/langgraph/threads/*/history", (route) => { + if (!runStarted) { + return route.fallback(); + } + return route.fulfill({ + status: 200, + contentType: "application/json", + body: JSON.stringify([ + { + values: { + title: "Artifact stream state", + goal: null, + messages: [...THREAD_MESSAGES, ...FOLLOW_UP_MESSAGES], + artifacts: [ARTIFACT_PATH], + }, + next: [], + metadata: {}, + created_at: "2025-01-01T00:00:00Z", + parent_config: null, + }, + ]), + }); + }); + await page.route(/\/api\/threads\/([^/]+)\/messages\/page/, (route) => { + if (!runStarted) { + return route.fallback(); + } + return route.fulfill({ + status: 200, + contentType: "application/json", + body: JSON.stringify({ + data: [...THREAD_MESSAGES, ...FOLLOW_UP_MESSAGES].map( + (message, index) => ({ + run_id: RUN_ID, + seq: index + 1, + content: message, + metadata: { caller: "lead_agent" }, + created_at: `2025-01-01T00:00:${String(index).padStart(2, "0")}Z`, + }), + ), + has_more: false, + next_before_seq: null, + }), + }); + }); await page.route("**/api/langgraph/threads/*/runs/stream", (route) => { + runStarted = true; return streamWithoutArtifacts(route); }); diff --git a/frontend/tests/unit/core/threads/coalesce.test.ts b/frontend/tests/unit/core/threads/coalesce.test.ts new file mode 100644 index 000000000..862a1ae5d --- /dev/null +++ b/frontend/tests/unit/core/threads/coalesce.test.ts @@ -0,0 +1,97 @@ +import { describe, expect, it } from "@rstest/core"; + +import { + decideCoalesce, + STREAM_RENDER_COALESCE_MS, +} from "@/core/threads/hooks"; + +describe("decideCoalesce", () => { + it("flushes immediately once a full interval has elapsed (leading edge)", () => { + expect(decideCoalesce(1000, 900, 80, false)).toEqual({ + action: "flush-now", + }); + expect(decideCoalesce(1000, 920, 80, false)).toEqual({ + action: "flush-now", + }); + }); + + it("schedules exactly one trailing flush for the interval remainder", () => { + expect(decideCoalesce(1000, 950, 80, false)).toEqual({ + action: "schedule", + delayMs: 30, + }); + }); + + it("waits while a trailing flush is already pending", () => { + expect(decideCoalesce(1000, 950, 80, true)).toEqual({ action: "wait" }); + }); + + it("still takes the leading edge when a late trailing flush is pending", () => { + // Timer callbacks queue behind long tasks, so an update can arrive past the + // interval while the trailing flush is still armed. The decision is + // flush-now, which obliges the call site to disarm that timer — otherwise + // it publishes a second time and slips the next interval forward. + expect(decideCoalesce(1000, 900, 80, true)).toEqual({ + action: "flush-now", + }); + }); + + it("never delays a flush beyond the interval, unlike a debounce", () => { + // Simulate a dense stream: updates every 10ms. A debounce would keep + // resetting its timer and never fire; here the trailing flush scheduled at + // the first update stays put and every later update just waits on it. + let lastFlush = 0; + let pendingUntil: number | null = null; + const flushes: number[] = []; + for (let now = 10; now <= 400; now += 10) { + if (pendingUntil !== null && now >= pendingUntil) { + flushes.push(pendingUntil); + lastFlush = pendingUntil; + pendingUntil = null; + } + const decision = decideCoalesce( + now, + lastFlush, + 80, + pendingUntil !== null, + ); + if (decision.action === "flush-now") { + flushes.push(now); + lastFlush = now; + } else if (decision.action === "schedule") { + pendingUntil = now + decision.delayMs; + } + } + expect(flushes.length).toBeGreaterThanOrEqual(4); + for (let i = 1; i < flushes.length; i++) { + const gap = flushes[i]! - flushes[i - 1]!; + expect(gap).toBeGreaterThanOrEqual(80); + expect(gap).toBeLessThanOrEqual(90); + } + }); + + it("takes the leading edge when nothing has flushed yet", () => { + // The hook seeds lastFlush with -Infinity on a monotonic clock whose epoch + // is page load, so the first update of a stream must not be deferred. + expect(decideCoalesce(5, Number.NEGATIVE_INFINITY, 80, false)).toEqual({ + action: "flush-now", + }); + }); + + it("keeps the scheduled delay inside the interval", () => { + // Holds for every elapsed value a monotonic clock can produce, which is why + // the call site needs no clamp on the timeout delay. + for (let elapsed = 0; elapsed < 80; elapsed++) { + const decision = decideCoalesce(1000, 1000 - elapsed, 80, false); + expect(decision.action).toBe("schedule"); + if (decision.action !== "schedule") continue; + expect(decision.delayMs).toBeGreaterThan(0); + expect(decision.delayMs).toBeLessThanOrEqual(80); + } + }); + + it("exports a frame-scale default interval", () => { + expect(STREAM_RENDER_COALESCE_MS).toBeGreaterThanOrEqual(50); + expect(STREAM_RENDER_COALESCE_MS).toBeLessThanOrEqual(100); + }); +}); diff --git a/frontend/tests/unit/core/threads/message-merge.test.ts b/frontend/tests/unit/core/threads/message-merge.test.ts index 3162ccef7..5a1ba2457 100644 --- a/frontend/tests/unit/core/threads/message-merge.test.ts +++ b/frontend/tests/unit/core/threads/message-merge.test.ts @@ -911,6 +911,29 @@ test("mergeTransientHistoryBridgeOrder retains confirmed overlap as a non-render ]); }); +test("mergeTransientHistoryBridgeOrder returns the same array when nothing is new", () => { + const order = mergeTransientHistoryBridgeOrder( + [], + [summarizationHuman1, summarizationAi1], + ); + + // Identity, not just equality: this runs per render while the bridge is + // active and feeds the coalesced render memo (#4409 Phase 1). + expect(mergeTransientHistoryBridgeOrder(order, [summarizationAi1])).toBe( + order, + ); + expect( + mergeTransientHistoryBridgeOrder(order, [ + summarizationHuman1, + summarizationAi1, + ]), + ).toBe(order); + expect(mergeTransientHistoryBridgeOrder(order, [])).toBe(order); + expect( + mergeTransientHistoryBridgeOrder(order, [summarizationHuman2]), + ).not.toBe(order); +}); + test("mergeTransientHistoryBridgeOrder keeps a recaptured protected prefix in place", () => { const protectedInput = { id: "protected-input",