github/deer-flow
1
0
Fork 0
mirror of https://github.com/bytedance/deer-flow.git synced 2026-07-26 16:07:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

perf(frontend): coalesce streaming renders to a frame budget instead of per chunk (#4425)

Browse Source 
* perf(frontend): coalesce streaming renders to a frame budget instead of per chunk

While a run streams, the merge/group/render pipeline consumed every SSE chunk
as its own React update (~60/s), re-rendering the whole thread tree per token.
Enable the SDK's same-tick batching (throttle: true) and publish the
render-facing messages snapshot at most once per 80 ms with a leading edge and
a trailing flush, keyed through a memoized merge so identities stay stable
between flushes. Lifecycle consumers (optimistic clearing, summarization
capture, usage baselines) keep reading the per-chunk array.

* perf(frontend): keep the transient bridge order array identity stable

mergeTransientHistoryBridgeOrder cloned unconditionally, so the render-time
call handed the coalesced merge memo a fresh array identity on every render
while the transient history bridge was open, re-running mergeMessages between
flushes. Clone lazily and return the input order when nothing is appended; the
merge only ever appends, so an unchanged length means unchanged content.

Consumers only read the returned order, so reusing the input is safe.

* perf(frontend): drive the render coalescer from a monotonic clock

The coalescing interval was measured with Date.now(). A backward wall-clock
step (NTP correction, sleep/wake) turns the elapsed term negative, so the
scheduled delay becomes interval + jump and the rendered snapshot stalls for
the length of the jump. Read performance.now() once per effect invocation
instead; the timer callback re-reads it because timers fire late and the next
interval must start from the real flush.

Seed the last-flush marker with -Infinity so the first update of a stream
still takes the leading edge under a page-load-relative clock.

* perf(frontend): reset the coalescer flush baseline when a stream ends

The leading-edge flush was scoped to the hook instance rather than to each
stream: a run starting within one interval of the previous one found a recent
flush baseline and deferred its first frame. Drop the baseline when leaving
the streaming state so every stream opens on the leading edge.

* perf(frontend): disarm the trailing flush when the leading edge wins

decideCoalesce checks the elapsed interval before the pending-timer flag, so
an update arriving past the interval takes the leading edge while a trailing
timer is still armed. Timers fire late under main-thread load -- exactly the
regime this coalescer targets -- so that timer then publishes a second time
and slips the flush baseline forward, breaking the at-most-one-flush-per-
interval property when it matters most.

Disarm the pending timer in the flush-now branch, and cover the previously
untested elapsed >= interval && hasPendingTimer quadrant.

* perf(frontend): stop syncing the render snapshot while idle

The snapshot is only read while streaming, so keeping it current on every
idle messages change costs one wasted render per history refetch or thread
navigation. Dropping that publish outright is not safe either: the leading
edge runs in a passive effect, but the render where isStreaming flips true
paints first and returns the snapshot, so a stale one would be painted --
after a thread switch, another thread's messages, since the chat page
deliberately avoids re-mounting on navigation.

Make the snapshot nullable, where null means no snapshot belongs to the
current stream, and return the live array while it is null. The idle branch
then writes state once per stream end instead of once per idle update, and
the stale-frame window does not exist rather than being short.
This commit is contained in:
Aari 2026-07-26 08:19:13 +08:00 committed by GitHub
parent 7aa314b4c1
commit f090f01806
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 378 additions and 37 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

217
frontend/src/core/threads/hooks.ts
View File

@ -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<Message[] | null>(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<ReturnType<typeof setTimeout> | 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<string[]>([]);
const transientHistoryOrderRef = useRef<readonly string[]>([]);
const transientHistoryThreadIdRef = useRef<string | null>(null);
const summarizedRef = useRef<Set<string>>(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,

78
frontend/tests/e2e/artifact-stream-state.spec.ts
View File

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

97
frontend/tests/unit/core/threads/coalesce.test.ts Normal file
View File

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

23
frontend/tests/unit/core/threads/message-merge.test.ts
View File

@ -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",