deer-flow/frontend/tests/e2e/artifact-stream-state.spec.ts
Aari f090f01806
perf(frontend): coalesce streaming renders to a frame budget instead of per chunk (#4425)
* 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.
2026-07-26 08:19:13 +08:00

151 lines
4.3 KiB
TypeScript

import { expect, test, type Route } from "@playwright/test";
import { mockLangGraphAPI } from "./utils/mock-api";
const THREAD_ID = "00000000-0000-0000-0000-000000003788";
const RUN_ID = "00000000-0000-0000-0000-000000003789";
const ARTIFACT_PATH = "/artifact-fixtures/report.md";
const THREAD_MESSAGES = [
{
type: "human",
id: "msg-human-artifact",
content: [{ type: "text", text: "Create a markdown report" }],
},
{
type: "ai",
id: "msg-ai-artifact",
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 = [
{
event: "metadata",
data: { run_id: RUN_ID, thread_id: THREAD_ID },
},
{
event: "values",
data: {
messages: FOLLOW_UP_MESSAGES,
},
},
];
return route.fulfill({
status: 200,
contentType: "text/event-stream",
body: events
.map((event) => {
return `event: ${event.event}\ndata: ${JSON.stringify(event.data)}\n\n`;
})
.join(""),
});
}
test("keeps artifact trigger after stream values omit artifacts", async ({
page,
}) => {
mockLangGraphAPI(page, {
threads: [
{
thread_id: THREAD_ID,
title: "Artifact stream state",
artifacts: [ARTIFACT_PATH],
messages: THREAD_MESSAGES,
},
],
});
// 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);
});
await page.goto(`/workspace/chats/${THREAD_ID}`);
const artifactTrigger = page.getByRole("button", { name: /artifacts/i });
await expect(artifactTrigger).toBeVisible({ timeout: 15_000 });
const textarea = page.getByPlaceholder(/how can i assist you/i);
await textarea.fill("Continue");
await textarea.press("Enter");
await expect(
page.getByText("Updated response while the artifact list is omitted."),
).toBeVisible({ timeout: 10_000 });
await expect(artifactTrigger).toBeVisible();
await artifactTrigger.click();
const artifactsPanel = page.locator("#artifacts");
await expect(artifactsPanel.getByText("report.md")).toBeVisible();
await artifactsPanel.getByText("report.md").click();
await expect(artifactsPanel.getByRole("combobox")).toContainText("report.md");
});