From 67a2de8aa2801a4a809c98e600c284b6eaba0707 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=92=89=E8=90=81?= Date: Tue, 18 Aug 2026 19:20:25 +0800 Subject: [PATCH 1/2] perf(web-shell): keep streaming output responsive --- .../web-shell-stream-render-performance.md | 75 ++++ packages/sdk-typescript/scripts/build.js | 3 +- .../src/daemon/ui/transcript.ts | 58 +-- .../sdk-typescript/src/daemon/ui/types.ts | 7 +- .../sdk-typescript/test/unit/daemonUi.test.ts | 37 ++ .../web-shell/client/adapters/messageTypes.ts | 2 +- .../adapters/transcriptToMessages.test.ts | 38 ++ .../client/adapters/transcriptToMessages.ts | 23 +- .../components/MessageList.dom.test.tsx | 154 ++++++- .../client/components/MessageList.tsx | 385 +++++++++++++++--- .../messages/AssistantMessage.test.tsx | 49 +++ .../components/messages/AssistantMessage.tsx | 8 +- .../components/messages/Markdown.module.css | 8 + .../components/messages/Markdown.test.ts | 38 ++ .../client/components/messages/Markdown.tsx | 21 +- .../messages/MarkdownChartRenderer.test.tsx | 30 ++ .../e2e/web-shell.stream-performance.spec.ts | 258 ++++++++++++ ...useAnimationFrameTranscriptBlocks.test.tsx | 166 +++++++- .../useAnimationFrameTranscriptBlocks.ts | 99 ++++- .../client/hooks/useMessages.test.ts | 76 ++++ .../web-shell/client/hooks/useMessages.ts | 95 ++++- packages/web-shell/package.json | 1 + .../session/DaemonSessionProvider.test.tsx | 10 +- .../daemon/session/DaemonSessionProvider.tsx | 13 +- 24 files changed, 1548 insertions(+), 106 deletions(-) create mode 100644 docs/design/web-shell-stream-render-performance.md create mode 100644 packages/web-shell/client/e2e/web-shell.stream-performance.spec.ts diff --git a/docs/design/web-shell-stream-render-performance.md b/docs/design/web-shell-stream-render-performance.md new file mode 100644 index 00000000000..9d0c55ab10d --- /dev/null +++ b/docs/design/web-shell-stream-render-performance.md @@ -0,0 +1,75 @@ +# Web Shell streaming render performance + +## Problem + +Thinking and assistant deltas currently wake the transcript on every animation +frame. Each accepted snapshot runs transcript projection and downstream list +work, while the growing Markdown document is parsed again at every streaming +flush. Although `ChatEditor` is memoized, this main-thread work still competes +with editor input and becomes increasingly expensive as the active response +grows. + +## Evidence + +The transcript projector is linear, but browser profiling with 50,000 retained +messages attributes only 2.5% of sampled time to projection. The dominant +104 ms long task spends 52.1 ms in `applyTurnCollapse`; repeated full-history +derivation in `MessageList` also includes final-answer collection, agent +grouping, pinning, and display-index generation. + +After the tail-only path, two CPU samples reduced `applyTurnCollapse` from +467.8 ms total self time to 26.7–51.5 ms, final-answer collection from 247.2 ms +to 11.4–26.9 ms, grouping from 54 ms to 2.4–7.6 ms, and display-index +generation from 67.5 ms to 3.7–12 ms. The mock SSE disconnected after replay +in that rerun, so these samples establish hotspot reduction but are not used as +end-to-end completion or long-task acceptance evidence. + +Markdown has the opposite shape: every streamed append changes the complete +source string and reparses the complete growing document. Throttling bounds how +often that happens but not the cost of each parse. + +## Design + +1. Batch provider transcript events into a 16 ms macrotask window, with + synchronous flushes before control and terminal events and when the stream + ends. Downstream, coalesce transcript notifications and admit at most one + snapshot every 50 ms. +2. Defer transcript snapshots with session and block-index identities. Urgent + editor work can commit against the previous snapshot, while session switches + and same-session store resets immediately reject stale deferred blocks. +3. Preserve normalized tool-content references with a `WeakMap`, allowing the + existing row comparator's JSON cache to avoid reserializing unchanged + historical tool output. +4. Keep the thinking elapsed timer alive across streamed content appends. +5. Keep live Markdown for short responses so closed charts and ordinary + formatting retain their existing behavior. Once a streaming document + exceeds a fixed parse budget, render its throttled source as escaped plain + text with preserved whitespace. When streaming ends, render the complete + Markdown once. This bounds repeated parsing while only delaying formatting + for responses large enough to cause the observed problem. +6. Preserve projected history object identity when every prior transcript block + is unchanged and only the final ordinary streaming text block grows. Reuse + completed-history `MessageList` derivations under the same narrow condition, + replacing only the rendered tail row. Any earlier block change, terminal + transition, tool/background update, usage change, translation change, or + view-option change takes the existing full calculation path. + +## Non-goals + +- No general incremental transcript projector. Projection is not the measured + bottleneck, and the narrow tail path avoids new invalidation machinery. +- No incremental Markdown AST or Web Worker. Plain streaming text removes the + repeated parse with less code and no cross-thread serialization. +- No changes to daemon event ordering, transcript persistence, or public block + shapes. + +## Verification + +- Unit tests cover notification coalescing, the 50 ms window, cancellation, + session switching, stable projection identity, streamed-tail rendering and + invalidation, stable tool normalization, timer reuse, and the + streaming-text-to-settled-Markdown transition. +- `npm run test:e2e:perf --workspace=@qwen-code/web-shell` deterministically + replays 5,000 historical turns, streams 400 Markdown-heavy chunks while + typing, verifies the final output and composer contents, and records input + latency and browser long-task metrics in the Playwright report. diff --git a/packages/sdk-typescript/scripts/build.js b/packages/sdk-typescript/scripts/build.js index dc3529c22f2..a91bbb0ef1a 100755 --- a/packages/sdk-typescript/scripts/build.js +++ b/packages/sdk-typescript/scripts/build.js @@ -93,7 +93,8 @@ const rootDir = join(__dirname, '..'); // Bumped from 195KB to 196KB for transient-vs-gone media hydration errors and // the reference-only replay placeholder. // Bumped from 196KB to 197KB for the workspace session live-state daemon -// surface (catalog version + live snapshot accessors). +// surface (catalog version + live snapshot accessors) and immutable, +// identity-stable transcript block indexes used by browser renderers. const MAX_DAEMON_BROWSER_BUNDLE_BYTES = 197 * 1024; // The opt-in `daemon/transports` browser bundle legitimately ships the concrete // ACP transports (AcpHttpTransport/AcpWsTransport/AutoReconnect + negotiate), so diff --git a/packages/sdk-typescript/src/daemon/ui/transcript.ts b/packages/sdk-typescript/src/daemon/ui/transcript.ts index 7b3e58066c6..bdb2282122f 100644 --- a/packages/sdk-typescript/src/daemon/ui/transcript.ts +++ b/packages/sdk-typescript/src/daemon/ui/transcript.ts @@ -129,17 +129,17 @@ export function appendLocalUserTranscriptMessage( return trimTranscriptState(next); } -// Freeze retained blocks at the dispatch boundary to catch consumers that -// mutate a COW-shared blocks array in place (see reduceDaemonTranscriptEvents). -// This is a dev/CI safety net; in production it is pure O(blocks) overhead on -// every dispatch and the reducer's own mutation discipline (takeBlocksOwnership) -// does not depend on it, so skip it there. App bundlers statically replace -// `process.env.NODE_ENV`, folding the check to `false`. The `typeof process` -// guard keeps an unbundled browser consumer from throwing a ReferenceError — +// Freeze retained COW collections at the dispatch boundary to catch consumers +// that mutate a shared snapshot (see reduceDaemonTranscriptEvents). This is a +// dev/CI safety net; the reducer's own ownership discipline does not depend on +// it, so skip the O(blocks) freeze in production. App bundlers statically +// replace `process.env.NODE_ENV`, folding the check to `false`. The `typeof +// process` guard keeps an unbundled browser consumer from throwing a +// ReferenceError — // this module sits on the browser-hostile `daemon/ui` surface and Vite lib // builds preserve `process.env.NODE_ENV` in their output — matching the // existing SDK idiom (see ProcessTransport, cliPath). -const FREEZE_TRANSCRIPT_BLOCKS = +const FREEZE_TRANSCRIPT_COLLECTIONS = typeof process !== 'undefined' && process.env.NODE_ENV !== 'production'; export function reduceDaemonTranscriptEvents( @@ -151,17 +151,12 @@ export function reduceDaemonTranscriptEvents( const next = cloneTranscriptState(state, opts); for (const event of events) applyDaemonTranscriptEvent(next, event); const result = trimTranscriptState(next); - // With lazy COW, `state.blocks` is shared across - // sidechannel-only snapshots. A misbehaving consumer doing - // `(state.blocks as DaemonTranscriptBlock[]).sort()` would corrupt - // EVERY snapshot that shares the reference (previously only the - // current one). Freeze the array at the dispatch boundary so external - // in-place mutation throws in strict mode instead of silently - // poisoning future snapshots. Internal reducer mutation goes through - // `takeBlocksOwnership` which copies BEFORE mutating, so the frozen - // shared reference is never touched in-place by the next dispatch. - if (FREEZE_TRANSCRIPT_BLOCKS) { + // With lazy COW, blocks and their index can be shared across snapshots. + // Freeze both at the dispatch boundary so external in-place mutation throws + // in strict mode instead of poisoning every snapshot sharing the reference. + if (FREEZE_TRANSCRIPT_COLLECTIONS) { Object.freeze(result.blocks); + Object.freeze(result.blockIndexById); } return result; } @@ -173,6 +168,7 @@ export function finalizeOfflineDaemonTranscriptState( finishAssistant(next); next.activeUserBlockId = undefined; Object.freeze(next.blocks); + Object.freeze(next.blockIndexById); return next; } @@ -1002,6 +998,8 @@ function discardToolBlock( takeBlocksOwnership(state); state.blocks = state.blocks.filter((block) => block.id !== blockId); state.blockIndexById = rebuildDaemonTranscriptBlockIndex(state.blocks); + ownedBlocks.set(state, state.blocks); + ownedBlockIndexes.set(state, state.blockIndexById); delete state.toolBlockByCallId[toolCallId]; delete state.toolProgress[toolCallId]; if (state.currentToolCallId === toolCallId) { @@ -1443,10 +1441,10 @@ function trimTranscriptState( const keptIds = new Set(blocks.map((block) => block.id)); state.blocks = blocks; state.blockIndexById = rebuildDaemonTranscriptBlockIndex(blocks); - // Trim replaces both arrays with fresh objects; register that this - // state now owns its blocks so future appends in the same dispatch - // don't double-copy. + // Trim replaces both collections with fresh objects; register ownership so + // future appends in the same dispatch don't copy them again. ownedBlocks.set(state, state.blocks); + ownedBlockIndexes.set(state, state.blockIndexById); for (const [toolCallId, blockId] of Object.entries(state.toolBlockByCallId)) { if (!keptIds.has(blockId)) { state.toolBlockByCallId[toolCallId] = TRIMMED_TOOL_BLOCK_ID; @@ -1520,7 +1518,7 @@ function shouldRecreateTrimmedToolBlock( } /** - * Lazy copy-on-write for `state.blocks` / `state.blockIndexById`. + * Lazy copy-on-write for `state.blocks`. * * `cloneTranscriptState` shares the parent's `blocks` reference (not * eager-copies) so non-block-mutating events keep the same array @@ -1538,14 +1536,23 @@ const ownedBlocks = new WeakMap< DaemonTranscriptState, readonly DaemonTranscriptBlock[] >(); +const ownedBlockIndexes = new WeakMap< + DaemonTranscriptState, + Readonly> +>(); function takeBlocksOwnership(state: DaemonTranscriptState): void { if (ownedBlocks.get(state) === state.blocks) return; state.blocks = [...state.blocks]; - state.blockIndexById = createIndex(state.blockIndexById); ownedBlocks.set(state, state.blocks); } +function takeBlockIndexOwnership(state: DaemonTranscriptState): void { + if (ownedBlockIndexes.get(state) === state.blockIndexById) return; + state.blockIndexById = createIndex(state.blockIndexById); + ownedBlockIndexes.set(state, state.blockIndexById); +} + // Applies a daemon rewind event to this in-memory transcript only. The target // user turn and everything after it are removed so the rendered session view // matches the already-rewound backend state. @@ -1579,6 +1586,7 @@ function truncateTranscriptBeforeBlock( state.blocks = state.blocks.slice(0, blockIndex); ownedBlocks.set(state, state.blocks); rebuildTranscriptIndexes(state); + ownedBlockIndexes.set(state, state.blockIndexById); } function rebuildTranscriptIndexes(state: DaemonTranscriptState): void { @@ -1617,7 +1625,9 @@ function appendBlock( block: DaemonTranscriptBlock, ): void { takeBlocksOwnership(state); - state.blockIndexById[block.id] = state.blocks.length; + takeBlockIndexOwnership(state); + (state.blockIndexById as Record)[block.id] = + state.blocks.length; (state.blocks as DaemonTranscriptBlock[]).push(block); } diff --git a/packages/sdk-typescript/src/daemon/ui/types.ts b/packages/sdk-typescript/src/daemon/ui/types.ts index 12d688d5088..0935c9bab57 100644 --- a/packages/sdk-typescript/src/daemon/ui/types.ts +++ b/packages/sdk-typescript/src/daemon/ui/types.ts @@ -1038,8 +1038,9 @@ export interface DaemonTranscriptState // lazy COW). Match the runtime contract at the type level so // consumers get a compile-time error for `state.blocks.sort()` / // `.push()` instead of a runtime `TypeError`. Internal reducer - // mutation goes through `takeBlocksOwnership` which casts away - // readonly after copying — the only place that's allowed. + // mutation goes through the ownership helpers which cast away readonly after + // copying — the only place that's allowed. The block index follows the same + // COW contract. blocks: readonly DaemonTranscriptBlock[]; lastEventId?: number; activeUserBlockId?: string; @@ -1047,7 +1048,7 @@ export interface DaemonTranscriptState activeThoughtBlockId?: string; activeAssistantBlockByParent: Record; activeThoughtBlockByParent: Record; - blockIndexById: Record; + blockIndexById: Readonly>; toolBlockByCallId: Record; trimmedToolNotificationByCallId: Record; permissionBlockByRequestId: Record; diff --git a/packages/sdk-typescript/test/unit/daemonUi.test.ts b/packages/sdk-typescript/test/unit/daemonUi.test.ts index b114b5f622b..067abb2d0f0 100644 --- a/packages/sdk-typescript/test/unit/daemonUi.test.ts +++ b/packages/sdk-typescript/test/unit/daemonUi.test.ts @@ -5808,6 +5808,43 @@ describe('transcriptBlockToTerminalText (wenshao review — coverage)', () => { }); describe('daemon UI WeakMap memo hits (wenshao glm-5.1 review)', () => { + it('shares the block index for text updates and copies it for appends', () => { + let state = createDaemonTranscriptState({ now: 1 }); + state = reduceDaemonTranscriptEvents( + state, + [{ type: 'assistant.text.delta', text: 'first' } as never], + { now: 2 }, + ); + const firstState = state; + + state = reduceDaemonTranscriptEvents( + state, + [{ type: 'assistant.text.delta', text: ' second' } as never], + { now: 3 }, + ); + + expect(state.blocks).not.toBe(firstState.blocks); + expect(state.blockIndexById).toBe(firstState.blockIndexById); + expect(Object.isFrozen(state.blockIndexById)).toBe(true); + expect( + () => + ((state.blockIndexById as Record)['assistant-1'] = 99), + ).toThrow(TypeError); + expect(state.blocks[0]).toMatchObject({ text: 'first second' }); + expect(firstState.blocks[0]).toMatchObject({ text: 'first' }); + + const updatedState = state; + state = reduceDaemonTranscriptEvents( + state, + [{ type: 'status', text: 'done' } as never], + { now: 4 }, + ); + + expect(state.blockIndexById).not.toBe(updatedState.blockIndexById); + expect(updatedState.blockIndexById).not.toHaveProperty('status-2'); + expect(state.blockIndexById).toHaveProperty('status-2', 1); + }); + // wenshao 5-23 13:03: lazy COW means non-block-mutating dispatches // preserve `state.blocks` reference, so the WeakMap caches actually hit // across renders. Verify by checking reference identity. diff --git a/packages/web-shell/client/adapters/messageTypes.ts b/packages/web-shell/client/adapters/messageTypes.ts index aa4fb1c3765..82bc1fb81d3 100644 --- a/packages/web-shell/client/adapters/messageTypes.ts +++ b/packages/web-shell/client/adapters/messageTypes.ts @@ -45,7 +45,7 @@ export interface DaemonMessageToolCall { status: DaemonMessageToolCallStatus; parentToolCallId?: string; title?: string; - content?: DaemonMessageToolCallContent[]; + content?: readonly DaemonMessageToolCallContent[]; rawOutput?: unknown; locations?: DaemonMessageToolCallLocation[]; kind?: DaemonMessageToolKind; diff --git a/packages/web-shell/client/adapters/transcriptToMessages.test.ts b/packages/web-shell/client/adapters/transcriptToMessages.test.ts index 80780e5fc81..b6d4c3c1255 100644 --- a/packages/web-shell/client/adapters/transcriptToMessages.test.ts +++ b/packages/web-shell/client/adapters/transcriptToMessages.test.ts @@ -203,6 +203,44 @@ describe('transcriptBlocksToDaemonMessages', () => { }); }); + it('normalizes an unchanged tool block content to a stable reference', () => { + const block = toolBlock('t1', 'call-1', 'running', 0, { + content: [{ type: 'content', content: { type: 'text', text: 'body' } }], + }); + const first = transcriptBlocksToDaemonMessages([block]); + const second = transcriptBlocksToDaemonMessages([block]); + + const firstContent = (first[0] as { tools: { content: unknown }[] }) + .tools[0].content; + const secondContent = (second[0] as { tools: { content: unknown }[] }) + .tools[0].content; + // The normalizer caches by the original block reference, so a block that + // did not change yields the same content array across frames, + // allowing MessageItem's JSON cache to avoid re-serializing the output. + expect(secondContent).toBe(firstContent); + expect(Object.isFrozen(firstContent)).toBe(true); + }); + + it('renormalizes content when a caller replaces the tool block', () => { + const block = toolBlock('t1', 'call-1', 'running', 0, { + content: [{ type: 'content', content: { type: 'text', text: 'before' } }], + }); + const first = transcriptBlocksToDaemonMessages([block]); + const content = block.content as Array<{ + type: 'content'; + content: { type: 'text'; text: string }; + }>; + content[0].content.text = 'after'; + const second = transcriptBlocksToDaemonMessages([{ ...block }]); + + expect( + (first[0] as { tools: { content: unknown }[] }).tools[0].content, + ).toMatchObject([{ content: { text: 'before' } }]); + expect( + (second[0] as { tools: { content: unknown }[] }).tools[0].content, + ).toMatchObject([{ content: { text: 'after' } }]); + }); + it('preserves user file attachment metadata', () => { const messages = transcriptBlocksToDaemonMessages([ textBlock('user-1', 'user', 'check this', 1, false, { diff --git a/packages/web-shell/client/adapters/transcriptToMessages.ts b/packages/web-shell/client/adapters/transcriptToMessages.ts index 18909771714..73834ab6bf3 100644 --- a/packages/web-shell/client/adapters/transcriptToMessages.ts +++ b/packages/web-shell/client/adapters/transcriptToMessages.ts @@ -1107,7 +1107,7 @@ function daemonToolBlockToToolCall( ): DaemonMessageToolCall { const rawOutput = getToolRawOutput(block); const isBackgroundAgent = isBackgroundAgentBlock(block, rawOutput); - const content = normalizeToolContent(block.content); + const content = normalizeToolContent(block); const statusMap: Record = { running: 'in_progress', pending: 'pending', @@ -1278,11 +1278,23 @@ function getToolRawOutput(block: DaemonToolTranscriptBlock): unknown { }; } +// The transcript store uses copy-on-write: an unchanged tool keeps its block +// identity across frames. Keying by the block, rather than its content array, +// also handles callers that replace a block while reusing its content array. +const normalizedToolContentCache = new WeakMap< + DaemonToolTranscriptBlock, + readonly DaemonMessageToolCallContent[] +>(); + function normalizeToolContent( - value: unknown, -): DaemonMessageToolCallContent[] | undefined { + block: DaemonToolTranscriptBlock, +): readonly DaemonMessageToolCallContent[] | undefined { + const value = block.content; if (!Array.isArray(value)) return undefined; + const cached = normalizedToolContentCache.get(block); + if (cached !== undefined) return cached; + const content = value.flatMap((entry): DaemonMessageToolCallContent[] => { const item = getRecord(entry); if (!item) return []; @@ -1328,7 +1340,10 @@ function normalizeToolContent( return []; }); - return content.length > 0 ? content : undefined; + if (content.length === 0) return undefined; + const frozen = Object.freeze(content); + normalizedToolContentCache.set(block, frozen); + return frozen; } function isAskUserQuestionBlock(block: DaemonToolTranscriptBlock): boolean { diff --git a/packages/web-shell/client/components/MessageList.dom.test.tsx b/packages/web-shell/client/components/MessageList.dom.test.tsx index ff81d673d64..033d75fd9f0 100644 --- a/packages/web-shell/client/components/MessageList.dom.test.tsx +++ b/packages/web-shell/client/components/MessageList.dom.test.tsx @@ -1,6 +1,12 @@ // @vitest-environment jsdom import { afterEach, describe, expect, it, vi } from 'vitest'; -import { act, createRef, type RefObject } from 'react'; +import { + act, + createRef, + startTransition, + Suspense, + type RefObject, +} from 'react'; import { createRoot, type Root } from 'react-dom/client'; import type { Message } from '../adapters/types'; import { @@ -18,6 +24,7 @@ import flashStyles from './MessageLocateFlash.module.css'; import styles from './MessageList.module.css'; const virtualizerTestState = vi.hoisted(() => ({ + getItemKeys: [] as Array<(index: number) => string | number>, itemSizeCache: new Map(), resizeItem: vi.fn(), renderItems: true, @@ -66,6 +73,8 @@ vi.mock('./MessageItem', async () => { 'data-locate-flashing': isLocateFlashing ? 'true' : undefined, 'data-send-failed': sendFailed ? 'true' : undefined, 'data-timestamp': message.timestamp, + 'data-message-content': + 'content' in message ? message.content : undefined, 'data-tool-ids': message.role === 'tool_group' ? message.tools.map((tool) => tool.callId).join(',') @@ -111,6 +120,7 @@ vi.mock('@tanstack/react-virtual', () => ({ enabled: boolean; getItemKey: (index: number) => string | number; }) => { + virtualizerTestState.getItemKeys.push(getItemKey); const virtualItems = enabled && virtualizerTestState.renderItems ? Array.from({ length: Math.min(count, 5) }, (_, index) => ({ @@ -182,6 +192,7 @@ afterEach(() => { virtualizerTestState.itemSizeCache.clear(); virtualizerTestState.resizeItem.mockClear(); virtualizerTestState.renderItems = true; + virtualizerTestState.getItemKeys.length = 0; vi.useRealTimers(); vi.restoreAllMocks(); }); @@ -3199,6 +3210,147 @@ describe('MessageList — turn collapse (DOM)', () => { expect(onLoadOlderHistory).toHaveBeenCalledTimes(1); }); + it.each([false, true])( + 'renders the latest content through the streamed-tail fast path (compact: %s)', + (compactMode) => { + const assistant = { + ...asstMsg('a1'), + content: 'first chunk', + isStreaming: true, + }; + const messages = [userMsg('u1'), assistant]; + const container = mount(messages, undefined, { + isResponding: true, + compactMode, + }); + const getItemKey = virtualizerTestState.getItemKeys.at(-1); + + rerenderMessages( + container, + [messages[0], { ...assistant, content: 'first chunk plus delta' }], + { isResponding: true, compactMode }, + ); + + expect( + container + .querySelector('[data-testid="msg-a1"]') + ?.getAttribute('data-message-content'), + ).toBe('first chunk plus delta'); + expect(virtualizerTestState.getItemKeys.at(-1)).toBe(getItemKey); + }, + ); + + it('falls back safely when streamed assistant content is undefined', () => { + const assistant = { + ...asstMsg('a1'), + content: undefined as unknown as string, + isStreaming: true, + }; + const messages = [userMsg('u1'), assistant]; + const container = mount(messages, undefined, { isResponding: true }); + + rerenderMessages(container, [messages[0], { ...assistant }], { + isResponding: true, + }); + + expect(container.querySelector('[data-testid="msg-a1"]')).not.toBeNull(); + }); + + it('does not reuse streamed-tail derivations when an earlier row changes', () => { + const assistant = { + ...asstMsg('a1'), + content: 'first chunk', + isStreaming: true, + }; + const status = { ...systemMsg('s1'), timestamp: 1 }; + const messages = [userMsg('u1'), status, assistant]; + const container = mount(messages, undefined, { isResponding: true }); + + const changedStatus = { ...status, timestamp: 2 }; + rerenderMessages( + container, + [ + messages[0], + changedStatus, + { ...assistant, content: 'first chunk plus delta' }, + ], + { isResponding: true }, + ); + + expect( + container + .querySelector('[data-testid="msg-s1"]') + ?.getAttribute('data-timestamp'), + ).toBe('2'); + expect( + container + .querySelector('[data-testid="msg-a1"]') + ?.getAttribute('data-message-content'), + ).toBe('first chunk plus delta'); + }); + + it('does not reuse caches written by an abandoned concurrent render', async () => { + const container = document.createElement('div'); + document.body.appendChild(container); + const root = createRoot(container); + mounted.push({ + root, + container, + transcriptRenderMode: 'interactive', + compactMode: false, + }); + const userA = { ...userMsg('u1'), content: 'committed' }; + const assistant = { + ...asstMsg('a1'), + content: 'first chunk', + isStreaming: true, + }; + const never = new Promise(() => {}); + const Suspend = () => { + throw never; + }; + const render = (messages: Message[], suspend = false) => + root.render( + + + + {suspend ? : null} + + , + ); + + act(() => render([userA, assistant])); + const committedGetItemKey = virtualizerTestState.getItemKeys.at(-1); + await act(async () => { + startTransition(() => + render( + [{ ...userA, id: 'u-abandoned', content: 'abandoned' }, assistant], + true, + ), + ); + await Promise.resolve(); + }); + expect(committedGetItemKey?.(0)).toBe('msg:u1'); + act(() => + render([userA, { ...assistant, content: 'latest committed chunk' }]), + ); + + expect( + container + .querySelector('[data-testid="msg-u1"]') + ?.getAttribute('data-message-content'), + ).toBe('committed'); + expect( + container + .querySelector('[data-testid="msg-a1"]') + ?.getAttribute('data-message-content'), + ).toBe('latest committed chunk'); + }); + it('measures newly prepended virtual rows before they can overlap the anchor', async () => { Object.defineProperty(HTMLElement.prototype, 'scrollHeight', { configurable: true, diff --git a/packages/web-shell/client/components/MessageList.tsx b/packages/web-shell/client/components/MessageList.tsx index 918a3c8c41d..526999737aa 100644 --- a/packages/web-shell/client/components/MessageList.tsx +++ b/packages/web-shell/client/components/MessageList.tsx @@ -146,6 +146,55 @@ interface MessageListProps { generateContent?: SessionContentGenerator; } +function isStreamingTailContentOnlyUpdate( + previous: readonly Message[] | undefined, + current: readonly Message[], +): boolean { + if (!previous || previous.length !== current.length || current.length === 0) { + return false; + } + for (let i = 0; i < current.length - 1; i += 1) { + if (previous[i] !== current[i]) return false; + } + const before = previous[previous.length - 1]; + const after = current[current.length - 1]; + if ( + before.id !== after.id || + before.role !== after.role || + (after.role !== 'assistant' && after.role !== 'thinking') || + (before.role !== 'assistant' && before.role !== 'thinking') || + before.isStreaming !== true || + after.isStreaming !== true || + before.timestamp !== after.timestamp + ) { + return false; + } + if (before.role === 'assistant' && after.role === 'assistant') { + if ( + typeof before.content !== 'string' || + typeof after.content !== 'string' + ) { + return false; + } + return ( + before.branchRecordId === after.branchRecordId && + before.usage === after.usage && + Boolean(before.content.trim()) === Boolean(after.content.trim()) + ); + } + return before.role === 'thinking' && after.role === 'thinking'; +} + +function sameIdentities( + left: readonly unknown[], + right: readonly unknown[], +): boolean { + return ( + left.length === right.length && + left.every((value, index) => Object.is(value, right[index])) + ); +} + function getLastUserMessageId(messages: Message[]): string | null { for (let i = messages.length - 1; i >= 0; i--) { const msg = messages[i]; @@ -2624,30 +2673,126 @@ export const MessageList = memo( const { t } = useI18n(); const transcriptRenderMode = useTranscriptRenderMode(); const compactMode = useContext(CompactModeContext); - const mergedMessages = useMemo( - () => - compactMode - ? mergeCompactToolGroups(messages, pendingApproval) - : messages, - [compactMode, messages, pendingApproval], + const previousMessagesRef = useRef(undefined); + const streamingTailContentOnly = isStreamingTailContentOnlyUpdate( + previousMessagesRef.current, + messages, ); - const displayItems = useMemo( - () => - attachTurnOutputs( - groupParallelAgents(mergedMessages), - isResponding, - turnFileChanges, - turnArtifacts, - turnScheduledTasks, - ), - [ - mergedMessages, + useLayoutEffect(() => { + previousMessagesRef.current = messages; + }, [messages]); + const mergedMessagesCache = useRef< + | { + sourceMessages: readonly Message[]; + compactMode: boolean; + pendingApproval: PermissionRequest | null; + value: Message[]; + } + | undefined + >(undefined); + const mergedMessages = useMemo(() => { + const cached = mergedMessagesCache.current; + const tail = messages[messages.length - 1]; + let value: Message[]; + if ( + streamingTailContentOnly && + tail?.role === 'assistant' && + cached?.sourceMessages === previousMessagesRef.current && + cached?.compactMode === compactMode && + cached.pendingApproval === pendingApproval && + cached.value[cached.value.length - 1]?.id === tail.id + ) { + value = cached.value.slice(); + value[value.length - 1] = tail; + } else { + value = compactMode + ? mergeCompactToolGroups(messages, pendingApproval) + : messages; + } + mergedMessagesCache.current = { + sourceMessages: messages, + compactMode, + pendingApproval, + value, + }; + return value; + }, [compactMode, messages, pendingApproval, streamingTailContentOnly]); + const displayItemsCache = useRef< + | { + sourceMessages: readonly Message[]; + compactMode: boolean; + pendingApproval: PermissionRequest | null; + isResponding: boolean; + turnFileChanges?: ReadonlyMap< + string, + readonly TurnOutputFileChange[] + >; + turnArtifacts?: ReadonlyMap; + turnScheduledTasks?: ReadonlyMap< + string, + readonly TurnOutputScheduledTask[] + >; + value: DisplayItem[]; + } + | undefined + >(undefined); + const displayItems = useMemo(() => { + const cached = displayItemsCache.current; + const tail = mergedMessages[mergedMessages.length - 1]; + let value: DisplayItem[] | undefined; + if ( + streamingTailContentOnly && + isResponding && + tail?.role === 'assistant' && + cached?.sourceMessages === previousMessagesRef.current && + cached?.compactMode === compactMode && + cached.pendingApproval === pendingApproval && + cached.isResponding === isResponding && + cached.turnFileChanges === turnFileChanges && + cached.turnArtifacts === turnArtifacts && + cached.turnScheduledTasks === turnScheduledTasks + ) { + const previousTail = cached.value[cached.value.length - 1]; + if ( + previousTail?.type === 'message' && + previousTail.message.id === tail.id + ) { + value = cached.value.slice(); + value[value.length - 1] = { + ...previousTail, + message: tail, + }; + } + } + value ??= attachTurnOutputs( + groupParallelAgents(mergedMessages), isResponding, turnFileChanges, turnArtifacts, turnScheduledTasks, - ], - ); + ); + displayItemsCache.current = { + sourceMessages: messages, + compactMode, + pendingApproval, + isResponding, + turnFileChanges, + turnArtifacts, + turnScheduledTasks, + value, + }; + return value; + }, [ + mergedMessages, + messages, + streamingTailContentOnly, + compactMode, + pendingApproval, + isResponding, + turnFileChanges, + turnArtifacts, + turnScheduledTasks, + ]); const latestBackgroundNotificationId = useMemo(() => { for (let i = mergedMessages.length - 1; i >= 0; i -= 1) { const message = mergedMessages[i]; @@ -2904,20 +3049,51 @@ export const MessageList = memo( ? (sessionTimelineEntries[sessionTimelineRange.currentIndex]?.id ?? fallbackCurrentTimelineTurnId) : fallbackCurrentTimelineTurnId; - const finalAssistantTurnIdByAssistantId = useMemo( - () => - collectFinalAssistantTurnIds(displayItems, { + const finalAssistantTurnIdsCache = useRef< + | { + sourceMessages: readonly Message[]; + isResponding: boolean; + latestTurnAwaitsAgentSummary: boolean; + gateBackgroundAgentStatus: boolean; + value: ReadonlyMap; + } + | undefined + >(undefined); + const finalAssistantTurnIdByAssistantId = useMemo(() => { + const cached = finalAssistantTurnIdsCache.current; + let value: ReadonlyMap; + if ( + streamingTailContentOnly && + isResponding && + cached?.sourceMessages === previousMessagesRef.current && + cached?.isResponding === isResponding && + cached.latestTurnAwaitsAgentSummary === latestTurnAwaitsAgentSummary && + cached.gateBackgroundAgentStatus === gateBackgroundAgentStatus + ) { + value = cached.value; + } else { + value = collectFinalAssistantTurnIds(displayItems, { isResponding, latestTurnAwaitsAgentSummary, gateBackgroundAgentStatus, - }), - [ - displayItems, - gateBackgroundAgentStatus, + }); + } + finalAssistantTurnIdsCache.current = { + sourceMessages: messages, isResponding, latestTurnAwaitsAgentSummary, - ], - ); + gateBackgroundAgentStatus, + value, + }; + return value; + }, [ + displayItems, + messages, + streamingTailContentOnly, + gateBackgroundAgentStatus, + isResponding, + latestTurnAwaitsAgentSummary, + ]); // ── Per-turn collapse ──────────────────────────────────────────────── // Completed turns fold down to their prompt + final answer (toggle on the @@ -3043,7 +3219,73 @@ export const MessageList = memo( }, [scheduleScrollOverflowReport], ); + const visibleItemsCache = useRef< + | { + sourceMessages: readonly Message[]; + dependencies: readonly unknown[]; + value: DisplayItem[]; + virtualizerItems: DisplayItem[]; + } + | undefined + >(undefined); + const reusedVisibleStreamingTailRef = useRef(false); const visibleItems = useMemo(() => { + reusedVisibleStreamingTailRef.current = false; + const dependencies = [ + collapseOverrides, + isResponding, + activeTurnStartedAt, + backgroundSummaryGraceActive, + latestTurnHasActiveBackgroundAgent, + unmatchedCompletionGraceExpired, + collapseEnabled, + hideFirstUserMessage, + firstTurnMetrics, + includeSubagentToolUsageInMetrics, + automaticallyExpandedAgentKeys, + compactMode, + pendingApproval, + turnFileChanges, + turnArtifacts, + turnScheduledTasks, + ] as const; + const cached = visibleItemsCache.current; + const currentTail = displayItems[displayItems.length - 1]; + if ( + streamingTailContentOnly && + isResponding && + cached && + cached.sourceMessages === previousMessagesRef.current && + sameIdentities(cached.dependencies, dependencies) && + currentTail?.type === 'message' + ) { + const key = getDisplayItemVirtualKey(currentTail); + let index = -1; + for (let i = cached.value.length - 1; i >= 0; i -= 1) { + if (getDisplayItemVirtualKey(cached.value[i]) === key) { + index = i; + break; + } + } + if (index >= 0) { + const previousTail = cached.value[index]; + if (previousTail?.type === 'message') { + const value = cached.value.slice(); + value[index] = { + ...previousTail, + message: currentTail.message, + }; + visibleItemsCache.current = { + sourceMessages: messages, + dependencies, + value, + virtualizerItems: cached.virtualizerItems, + }; + reusedVisibleStreamingTailRef.current = true; + return value; + } + } + } const collapsedItems = applyTurnCollapse(displayItems, { overrides: collapseOverrides, isResponding, @@ -3087,45 +3329,82 @@ export const MessageList = memo( itemsWithMetrics, automaticallyExpandedAgentKeys, ); - if (!hideFirstUserMessage) return pinnedItems; + if (!hideFirstUserMessage) { + visibleItemsCache.current = { + sourceMessages: messages, + dependencies, + value: pinnedItems, + virtualizerItems: pinnedItems, + }; + return pinnedItems; + } const firstUserId = mergedMessages.find( (message) => message.role === 'user', )?.id; - return firstUserId + const value = firstUserId ? pinnedItems.filter( (item) => item.type !== 'message' || item.message.id !== firstUserId, ) : pinnedItems; + visibleItemsCache.current = { + sourceMessages: messages, + dependencies, + value, + virtualizerItems: value, + }; + return value; }, [ displayItems, + streamingTailContentOnly, collapseOverrides, isResponding, activeTurnStartedAt, backgroundSummaryGraceActive, latestTurnHasActiveBackgroundAgent, unmatchedCompletionGraceExpired, - pendingApproval?.toolCallId, collapseEnabled, hideFirstUserMessage, firstTurnMetrics, includeSubagentToolUsageInMetrics, mergedMessages, + messages, automaticallyExpandedAgentKeys, + compactMode, + pendingApproval, + turnFileChanges, + turnArtifacts, + turnScheduledTasks, ]); - const visibleItemsRef = useRef(visibleItems); - visibleItemsRef.current = visibleItems; + const virtualizerItems = + visibleItemsCache.current?.sourceMessages === messages + ? visibleItemsCache.current.virtualizerItems + : visibleItems; const hasVisibleRowKey = useCallback( (key: string) => - visibleItemsRef.current.some( + virtualizerItems.some( (item) => String(getDisplayItemVirtualKey(item)) === key, ), - [], - ); - const visibleTurnIdByDisplayIndex = useMemo( - () => getTurnIdByDisplayIndex(visibleItems), - [visibleItems], + [virtualizerItems], ); + const visibleTurnIdsCache = useRef< + | { + length: number; + value: Array; + } + | undefined + >(undefined); + const visibleTurnIdByDisplayIndex = useMemo(() => { + if ( + reusedVisibleStreamingTailRef.current && + visibleTurnIdsCache.current?.length === visibleItems.length + ) { + return visibleTurnIdsCache.current.value; + } + const value = getTurnIdByDisplayIndex(visibleItems); + visibleTurnIdsCache.current = { length: visibleItems.length, value }; + return value; + }, [visibleItems]); const hasEnoughSessionTimelineEntries = sessionTimelineEntries.length >= SESSION_TIMELINE_MIN_VISIBLE_ENTRIES; @@ -3488,7 +3767,7 @@ export const MessageList = memo( if (hasTailContent && index === tailContentIndex) { return `slot:tail:${tailKey}`; } - const item = visibleItems[index - headerOffset]; + const item = virtualizerItems[index - headerOffset]; return item ? getDisplayItemVirtualKey(item) : `slot:row:${index}`; }, [ @@ -3496,10 +3775,26 @@ export const MessageList = memo( hasTailContent, tailContentIndex, tailKey, - visibleItems, + virtualizerItems, headerOffset, ], ); + const estimateItemSize = useCallback( + (index: number) => { + if (hasHeader && index === HEADER_INDEX) return ESTIMATE_HEADER; + if (hasTailContent && index === tailContentIndex) return ESTIMATE_TAIL; + const item = virtualizerItems[index - headerOffset]; + if (item?.type === 'turn_collapse') return ESTIMATE_TURN_COLLAPSE; + return ESTIMATE_MESSAGE; + }, + [ + hasHeader, + hasTailContent, + headerOffset, + tailContentIndex, + virtualizerItems, + ], + ); // Rule 6: skip if content doesn't overflow (no scrollbar). const scrollToBottom = useCallback( @@ -3564,13 +3859,7 @@ export const MessageList = memo( enabled: useVirtualScroll, getScrollElement, getItemKey, - estimateSize: (index) => { - if (hasHeader && index === HEADER_INDEX) return ESTIMATE_HEADER; - if (hasTailContent && index === tailContentIndex) return ESTIMATE_TAIL; - const item = visibleItems[index - headerOffset]; - if (item?.type === 'turn_collapse') return ESTIMATE_TURN_COLLAPSE; - return ESTIMATE_MESSAGE; - }, + estimateSize: estimateItemSize, overscan: 20, anchorTo: 'end', useFlushSync: false, diff --git a/packages/web-shell/client/components/messages/AssistantMessage.test.tsx b/packages/web-shell/client/components/messages/AssistantMessage.test.tsx index f41e7949b3b..79f1ca2e761 100644 --- a/packages/web-shell/client/components/messages/AssistantMessage.test.tsx +++ b/packages/web-shell/client/components/messages/AssistantMessage.test.tsx @@ -147,6 +147,55 @@ describe('AssistantMessage thinking logic', () => { expect(container.textContent).not.toContain('private chain of thought'); }); + it('does not recreate the elapsed timer on every streamed chunk', () => { + vi.useFakeTimers(); + const setIntervalSpy = vi.spyOn(globalThis, 'setInterval'); + const container = document.createElement('div'); + document.body.appendChild(container); + const root = createRoot(container); + mounted.push({ root, container }); + const tree = (content: string, isStreaming: boolean) => ( + + + + ); + act(() => root.render(tree('first', true))); + const intervalCountAfterMount = setIntervalSpy.mock.calls.length; + + act(() => root.render(tree('first second', true))); + act(() => root.render(tree('first second third', true))); + + expect(setIntervalSpy.mock.calls.length).toBe(intervalCountAfterMount); + }); + + it('does not start the elapsed timer for undefined content', () => { + vi.useFakeTimers(); + const setIntervalSpy = vi.spyOn(globalThis, 'setInterval'); + const container = document.createElement('div'); + document.body.appendChild(container); + const root = createRoot(container); + mounted.push({ root, container }); + const intervalCountBeforeRender = setIntervalSpy.mock.calls.length; + + act(() => + root.render( + + + , + ), + ); + + expect(setIntervalSpy.mock.calls.length).toBe(intervalCountBeforeRender); + }); + it('only translates completed thinking and reuses the in-memory result', async () => { const generateContent = vi.fn(async function* () { yield { diff --git a/packages/web-shell/client/components/messages/AssistantMessage.tsx b/packages/web-shell/client/components/messages/AssistantMessage.tsx index 02a5982b155..f1e2c9459ff 100644 --- a/packages/web-shell/client/components/messages/AssistantMessage.tsx +++ b/packages/web-shell/client/components/messages/AssistantMessage.tsx @@ -237,13 +237,17 @@ export const ThinkingMessage = memo(function ThinkingMessage({ const sawActiveRef = useRef(thinkingActive); const [now, setNow] = useState(() => Date.now()); const [finishedAt, setFinishedAt] = useState(null); + // `content` grows on every streamed chunk; keying on the boolean instead of + // the string keeps the timer effect from tearing down and re-creating the + // interval per chunk, while still starting once content first appears. + const hasContent = Boolean(content); useEffect(() => { - if (!content || !thinkingActive) return; + if (!hasContent || !thinkingActive) return; setNow(Date.now()); const id = setInterval(() => setNow(Date.now()), 1000); return () => clearInterval(id); - }, [content, thinkingActive]); + }, [hasContent, thinkingActive]); useEffect(() => { if (!content) return; diff --git a/packages/web-shell/client/components/messages/Markdown.module.css b/packages/web-shell/client/components/messages/Markdown.module.css index 8020a66df40..5c577bea062 100644 --- a/packages/web-shell/client/components/messages/Markdown.module.css +++ b/packages/web-shell/client/components/messages/Markdown.module.css @@ -8,6 +8,14 @@ overflow-wrap: break-word; } +.streamingPlainText { + margin: 0; + color: inherit; + font: inherit; + white-space: pre-wrap; + overflow-wrap: anywhere; +} + .content > :first-child { margin-top: 0; } diff --git a/packages/web-shell/client/components/messages/Markdown.test.ts b/packages/web-shell/client/components/messages/Markdown.test.ts index 7a0cc14b705..1d8ae4ecb43 100644 --- a/packages/web-shell/client/components/messages/Markdown.test.ts +++ b/packages/web-shell/client/components/messages/Markdown.test.ts @@ -1322,6 +1322,44 @@ describe('Markdown custom code block rendering', () => { }); container.remove(); }); + + it('applies transformMarkdown to a large streaming response', async () => { + const container = document.createElement('div'); + document.body.appendChild(container); + const root = createRoot(container); + const rawContent = `raw prefix ${'streaming text '.repeat(3_000)}`; + const transformMarkdown = vi.fn((content: string) => + content.replace('raw prefix', 'transformed prefix'), + ); + + await act(async () => { + root.render( + createElement( + WebShellCustomizationProvider, + { value: { markdown: { transformMarkdown } } }, + createElement(Markdown, { + content: rawContent, + source: 'assistant', + isStreaming: true, + }), + ), + ); + }); + + expect( + container.querySelector('[data-markdown-streaming-plain-text="true"]'), + ).not.toBeNull(); + expect(transformMarkdown).toHaveBeenCalledWith(rawContent, { + source: 'assistant', + }); + expect(container.textContent).toContain('transformed prefix'); + expect(container.textContent).not.toContain('raw prefix'); + + await act(async () => { + root.unmount(); + }); + container.remove(); + }); }); describe('Markdown code highlighting while streaming', () => { diff --git a/packages/web-shell/client/components/messages/Markdown.tsx b/packages/web-shell/client/components/messages/Markdown.tsx index 58e0ea318e4..ed89432cd9d 100644 --- a/packages/web-shell/client/components/messages/Markdown.tsx +++ b/packages/web-shell/client/components/messages/Markdown.tsx @@ -51,6 +51,10 @@ interface MarkdownProps { tableMode?: MarkdownTableMode; } +// Keep the cost of repeatedly parsing a growing stream bounded. Short streams +// retain live Markdown; large ones settle into full Markdown once at the end. +const STREAMING_MARKDOWN_PARSE_LIMIT = 32_000; + const SUPPORTED_LANGUAGES = new Set([ 'javascript', 'typescript', @@ -894,12 +898,15 @@ export const Markdown = memo(function Markdown({ const sourceMarkdown = source ? markdown : undefined; const throttledContent = useThrottledValue(content ?? '', isStreaming); + const renderStreamingPlainText = + isStreaming === true && + throttledContent.length > STREAMING_MARKDOWN_PARSE_LIMIT; const renderedContent = useMemo( () => throttledContent && source && sourceMarkdown?.transformMarkdown ? sourceMarkdown.transformMarkdown(throttledContent, { source }) : throttledContent, - [throttledContent, source, sourceMarkdown], + [source, sourceMarkdown, throttledContent], ); const effectiveTableMode = isStreaming @@ -966,6 +973,18 @@ export const Markdown = memo(function Markdown({ if (!content) return null; + if (renderStreamingPlainText) { + return ( +
+ 
{renderedContent}
+
+ ); + } + const renderedMarkdown = ( { }); describe('Web Shell markdown-chart integration', () => { + it('bounds parsing for a large stream and renders Markdown when it settles', async () => { + const registry = createMarkdownChartRegistry({ + loadECharts: async () => createFakeRuntime().runtime, + resizeObserver: false, + }); + const content = `# Large answer\n\n${'streaming text '.repeat(3_000)}`; + const tree = (isStreaming: boolean) => + chartTree({ content, registry, isStreaming }); + const result = await mount(tree(true)); + + expect( + result.container.querySelector( + '[data-markdown-streaming-plain-text="true"]', + ), + ).not.toBeNull(); + expect(result.container.querySelector('h1')).toBeNull(); + expect(result.container.textContent).toContain('# Large answer'); + + await result.rerender(tree(false)); + + expect( + result.container.querySelector( + '[data-markdown-streaming-plain-text="true"]', + ), + ).toBeNull(); + expect(result.container.querySelector('h1')?.textContent).toBe( + 'Large answer', + ); + }); + it('enables the built-in registry without host chart configuration', async () => { const chart = canonicalChart(); const { container } = await mount( diff --git a/packages/web-shell/client/e2e/web-shell.stream-performance.spec.ts b/packages/web-shell/client/e2e/web-shell.stream-performance.spec.ts new file mode 100644 index 00000000000..d8f312afcde --- /dev/null +++ b/packages/web-shell/client/e2e/web-shell.stream-performance.spec.ts @@ -0,0 +1,258 @@ +import { expect, test, type Page, type TestInfo } from '@playwright/test'; +import type { DaemonEvent } from '@qwen-code/sdk/daemon'; +import { + assistantTextEvent, + createWebShellDaemonScenario, + installMockDaemon, + replayCompleteEvent, + turnCompleteEvent, + userTextEvent, + type MockDaemonController, + type WebShellDaemonScenario, +} from './utils/mockDaemon'; + +interface BrowserPerformanceMetrics { + inputEvents: number[]; + longTasks: number[]; +} + +declare global { + interface Window { + __webShellPerformanceMetrics?: BrowserPerformanceMetrics; + } +} + +const historyTurns = Number(process.env['WEB_SHELL_PERF_TURNS'] ?? 5_000); +const streamChunks = Number(process.env['WEB_SHELL_PERF_CHUNKS'] ?? 400); +const streamIntervalMs = Number( + process.env['WEB_SHELL_PERF_INTERVAL_MS'] ?? 10, +); + +test.skip( + process.env['WEB_SHELL_PERF'] !== '1', + 'Set WEB_SHELL_PERF=1 to run the deterministic performance scenario.', +); + +test('keeps the composer responsive during deterministic streaming @perf', async ({ + page, +}, testInfo) => { + test.setTimeout(180_000); + + await installPerformanceObservers(page); + const history = createHistory(historyTurns); + const scenario = createWebShellDaemonScenario({ events: history }); + const daemon = await installScenario(page, scenario, testInfo); + + const replayStartedAt = Date.now(); + await gotoSession(page, scenario, daemon); + const replayMs = Date.now() - replayStartedAt; + + await fillComposer(page, 'Start deterministic performance stream'); + const submit = page.locator('[data-web-shell-composer-submit]'); + await expect(submit).toBeEnabled(); + await submit.click(); + await expect.poll(() => daemon.promptRequests().length).toBe(1); + + await resetPerformanceMetrics(page); + const expectedInput = 'input remains responsive while output is streaming'; + const streamStartedAt = Date.now(); + let nextEventId = history.length + 1; + let typingMs = 0; + const streamEvents = Array.from({ length: streamChunks }, (_, index) => { + const text = `${index}: ${'deterministic streaming text '.repeat(8)}\n`; + return assistantTextEvent(text, { id: nextEventId++ }); + }); + streamEvents.push( + assistantTextEvent('STREAM_COMPLETE_SENTINEL', { id: nextEventId++ }), + turnCompleteEvent('performance-prompt', { id: nextEventId++ }), + ); + + const type = async () => { + const editor = page.locator('[data-web-shell-composer-editor] .cm-content'); + await editor.click(); + const typingStartedAt = Date.now(); + await page.keyboard.type(expectedInput, { delay: 20 }); + typingMs = Date.now() - typingStartedAt; + await expect(editor).toHaveText(expectedInput); + }; + + await Promise.all([ + streamInBrowser(page, streamEvents, streamIntervalMs), + type(), + ]); + await expect(page.locator('[data-web-shell-message-list]')).toContainText( + 'STREAM_COMPLETE_SENTINEL', + ); + await page.waitForTimeout(100); + + const metrics = await readPerformanceMetrics(page); + const result = { + historyTurns, + streamChunks, + streamIntervalMs, + replayMs, + streamMs: Date.now() - streamStartedAt, + typingMs, + typingOverheadMs: typingMs - expectedInput.length * 20, + longTaskCount: metrics.longTasks.length, + longTaskTotalMs: sum(metrics.longTasks), + longTaskMaxMs: max(metrics.longTasks), + slowInputEventCount: metrics.inputEvents.length, + slowInputEventP95Ms: percentile(metrics.inputEvents, 0.95), + slowInputEventMaxMs: max(metrics.inputEvents), + }; + + console.log(`WEB_SHELL_STREAM_PERF ${JSON.stringify(result)}`); + await testInfo.attach('web-shell-stream-performance.json', { + body: JSON.stringify(result, null, 2), + contentType: 'application/json', + }); +}); + +function createHistory(turns: number): DaemonEvent[] { + const events: DaemonEvent[] = []; + for (let turn = 0; turn < turns; turn += 1) { + events.push( + userTextEvent(`historical user message ${turn}`, { + id: events.length + 1, + }), + assistantTextEvent( + `historical assistant message ${turn} ${'content '.repeat(12)}`, + { id: events.length + 2 }, + ), + turnCompleteEvent(`historical-prompt-${turn}`, { + id: events.length + 3, + }), + ); + } + return events; +} + +async function installPerformanceObservers(page: Page): Promise { + await page.addInitScript(() => { + const metrics: BrowserPerformanceMetrics = { + inputEvents: [], + longTasks: [], + }; + window.__webShellPerformanceMetrics = metrics; + + if (PerformanceObserver.supportedEntryTypes.includes('longtask')) { + new PerformanceObserver((list) => { + for (const entry of list.getEntries()) { + metrics.longTasks.push(entry.duration); + } + }).observe({ type: 'longtask', buffered: true }); + } + + if (PerformanceObserver.supportedEntryTypes.includes('event')) { + new PerformanceObserver((list) => { + for (const entry of list.getEntries()) { + if ( + entry.name === 'keydown' || + entry.name === 'beforeinput' || + entry.name === 'input' + ) { + metrics.inputEvents.push(entry.duration); + } + } + }).observe({ + type: 'event', + buffered: true, + durationThreshold: 16, + } as PerformanceObserverInit); + } + }); +} + +async function resetPerformanceMetrics(page: Page): Promise { + await page.evaluate(() => { + const metrics = window.__webShellPerformanceMetrics; + if (metrics) { + metrics.inputEvents.length = 0; + metrics.longTasks.length = 0; + } + }); +} + +async function streamInBrowser( + page: Page, + events: readonly DaemonEvent[], + intervalMs: number, +): Promise { + await page.evaluate( + async ({ events, intervalMs }) => { + const harness = window.__webShellSseHarness; + if (!harness) { + throw new Error('SSE harness is not installed.'); + } + for (const event of events) { + harness.writeFrame(`data: ${JSON.stringify(event)}\n\n`); + await new Promise((resolve) => window.setTimeout(resolve, intervalMs)); + } + }, + { events, intervalMs }, + ); +} + +async function readPerformanceMetrics( + page: Page, +): Promise { + return page.evaluate(() => + structuredClone( + window.__webShellPerformanceMetrics ?? { + inputEvents: [], + longTasks: [], + }, + ), + ); +} + +async function installScenario( + page: Page, + scenario: WebShellDaemonScenario, + testInfo: TestInfo, +): Promise { + return installMockDaemon(page, scenario, { + baseURL: String(testInfo.project.use.baseURL), + }); +} + +async function gotoSession( + page: Page, + scenario: WebShellDaemonScenario, + daemon: MockDaemonController, +): Promise { + await page.goto(`/session/${encodeURIComponent(scenario.sessionId)}`); + await expect(page.locator('[data-web-shell-root]')).toBeVisible(); + const connection = await daemon.sse.waitForConnection(scenario.sessionId); + await daemon.sendEvent( + replayCompleteEvent({ + sessionId: connection.sessionId, + replayedCount: scenario.events.length, + }), + ); + await expect(page.getByText('Loading...')).toHaveCount(0); +} + +async function fillComposer(page: Page, text: string): Promise { + const editor = page.locator('[data-web-shell-composer-editor] .cm-content'); + await editor.click(); + await page.keyboard.type(text); +} + +function sum(values: readonly number[]): number { + return Math.round(values.reduce((total, value) => total + value, 0)); +} + +function max(values: readonly number[]): number { + return Math.round(Math.max(0, ...values)); +} + +function percentile(values: readonly number[], fraction: number): number { + if (values.length === 0) { + return 0; + } + const sorted = [...values].sort((left, right) => left - right); + const index = Math.ceil(sorted.length * fraction) - 1; + return Math.round(sorted[index] ?? 0); +} diff --git a/packages/web-shell/client/hooks/useAnimationFrameTranscriptBlocks.test.tsx b/packages/web-shell/client/hooks/useAnimationFrameTranscriptBlocks.test.tsx index efa052bf53d..b15b36fd725 100644 --- a/packages/web-shell/client/hooks/useAnimationFrameTranscriptBlocks.test.tsx +++ b/packages/web-shell/client/hooks/useAnimationFrameTranscriptBlocks.test.tsx @@ -10,9 +10,10 @@ Object.assign(globalThis, { IS_REACT_ACT_ENVIRONMENT: true }); const testStore = vi.hoisted(() => { let blocks: readonly DaemonTranscriptBlock[] = []; + let blockIndexById: Readonly> = {}; const listeners = new Set<() => void>(); return { - getSnapshot: () => ({ blocks }), + getSnapshot: () => ({ blocks, blockIndexById }), subscribe: (listener: () => void) => { listeners.add(listener); return () => listeners.delete(listener); @@ -21,25 +22,38 @@ const testStore = vi.hoisted(() => { blocks = nextBlocks; listeners.forEach((listener) => listener()); }, + resetBlocks(nextBlocks: readonly DaemonTranscriptBlock[] = []) { + blocks = nextBlocks; + blockIndexById = {}; + listeners.forEach((listener) => listener()); + }, reset() { blocks = []; + blockIndexById = {}; listeners.clear(); }, }; }); +const testConnection = vi.hoisted(() => ({ + sessionId: 'session-a' as string | undefined, +})); + vi.mock('@qwen-code/webui/daemon-react-sdk', () => ({ useTranscriptStore: () => testStore, + useConnection: () => testConnection, })); let root: Root | null = null; let container: HTMLDivElement | null = null; let renderCount = 0; let latestBlocks: readonly DaemonTranscriptBlock[] = []; +let renderLog: string[][] = []; function Harness() { latestBlocks = useAnimationFrameTranscriptBlocks(); renderCount += 1; + renderLog.push(latestBlocks.map((block) => block.id)); return null; } @@ -50,7 +64,9 @@ afterEach(() => { container = null; renderCount = 0; latestBlocks = []; + renderLog = []; testStore.reset(); + testConnection.sessionId = 'session-a'; vi.restoreAllMocks(); }); @@ -61,6 +77,7 @@ describe('useAnimationFrameTranscriptBlocks', () => { pendingFrame = callback; return 1; }); + vi.spyOn(performance, 'now').mockReturnValue(10); container = document.createElement('div'); document.body.append(container); root = createRoot(container); @@ -79,13 +96,105 @@ describe('useAnimationFrameTranscriptBlocks', () => { expect(pendingFrame).not.toBeNull(); act(() => { - pendingFrame?.(performance.now()); + pendingFrame?.(16); }); - expect(renderCount).toBe(initialRenderCount + 1); + // The deferred value renders once (stale) then once more to catch up to + // the latest snapshot — exactly two renders, never one per store update. + // The upper bound keeps the rAF-coalescing guard: a regression that + // renders once per update (100 renders) still fails. + expect(renderCount).toBeLessThanOrEqual(initialRenderCount + 2); expect(latestBlocks).toHaveLength(100); }); + it('throttles renders to one per throttle window', () => { + let pendingFrame: FrameRequestCallback | null = null; + vi.spyOn(window, 'requestAnimationFrame').mockImplementation((callback) => { + pendingFrame = callback; + return 1; + }); + let now = 1_000; + vi.spyOn(performance, 'now').mockImplementation(() => now); + container = document.createElement('div'); + document.body.append(container); + root = createRoot(container); + act(() => root!.render()); + renderCount = 0; + + // First notification is due immediately (lastNotifyTs starts at 0). + act(() => testStore.update([{ id: 'a' } as DaemonTranscriptBlock])); + act(() => pendingFrame?.(now)); + const afterFirst = renderCount; + expect(latestBlocks.map((block) => block.id)).toEqual(['a']); + expect(afterFirst).toBeGreaterThan(0); + + // A second notification inside the 50ms window must not render. + now = 1_020; + act(() => testStore.update([{ id: 'b' } as DaemonTranscriptBlock])); + act(() => pendingFrame?.(now)); + expect(renderCount).toBe(afterFirst); + expect(latestBlocks.map((block) => block.id)).toEqual(['a']); + + // Once the window elapses, the pending frame renders the latest blocks. + now = 1_060; + act(() => pendingFrame?.(now)); + expect(renderCount).toBeGreaterThan(afterFirst); + expect(latestBlocks.map((block) => block.id)).toEqual(['b']); + }); + + it('waits for a quiet window after composer input', () => { + let pendingFrame: FrameRequestCallback | null = null; + vi.spyOn(window, 'requestAnimationFrame').mockImplementation((callback) => { + pendingFrame = callback; + return 1; + }); + let now = 1_000; + vi.spyOn(performance, 'now').mockImplementation(() => now); + container = document.createElement('div'); + document.body.append(container); + root = createRoot(container); + act(() => root!.render()); + renderCount = 0; + + act(() => testStore.update([{ id: 'a' } as DaemonTranscriptBlock])); + document.dispatchEvent(new Event('beforeinput')); + now = 1_060; + act(() => pendingFrame?.(now)); + expect(renderCount).toBe(0); + + now = 1_101; + act(() => pendingFrame?.(now)); + expect(renderCount).toBeGreaterThan(0); + expect(latestBlocks.map((block) => block.id)).toEqual(['a']); + }); + + it('does not starve transcript updates during continuous input', () => { + let pendingFrame: FrameRequestCallback | null = null; + vi.spyOn(window, 'requestAnimationFrame').mockImplementation((callback) => { + pendingFrame = callback; + return 1; + }); + let now = 1_000; + vi.spyOn(performance, 'now').mockImplementation(() => now); + container = document.createElement('div'); + document.body.append(container); + root = createRoot(container); + act(() => root!.render()); + renderCount = 0; + + act(() => testStore.update([{ id: 'a' } as DaemonTranscriptBlock])); + for (now = 1_050; now < 1_250; now += 50) { + document.dispatchEvent(new Event('beforeinput')); + act(() => pendingFrame?.(now)); + expect(renderCount).toBe(0); + } + + document.dispatchEvent(new Event('beforeinput')); + act(() => pendingFrame?.(1_250)); + expect(renderCount).toBeGreaterThan(0); + expect(latestBlocks.map((block) => block.id)).toEqual(['a']); + }); + it('cancels a pending frame on unmount', () => { vi.spyOn(window, 'requestAnimationFrame').mockReturnValue(7); const cancelFrame = vi.spyOn(window, 'cancelAnimationFrame'); @@ -100,4 +209,55 @@ describe('useAnimationFrameTranscriptBlocks', () => { expect(cancelFrame).toHaveBeenCalledWith(7); }); + + it('returns the new session blocks on the first render after a session switch', () => { + container = document.createElement('div'); + document.body.append(container); + root = createRoot(container); + act(() => root!.render()); + renderLog = []; + + // Session switch: the provider updates the connection sessionId and + // store.reset() together in one batch, so the first render must already + // see the new session's blocks instead of the deferred previous-session + // snapshot. The mocked useConnection is a plain object (no context + // re-render), so re-render manually to simulate the provider batch. + const blockB = { id: 'b1' } as DaemonTranscriptBlock; + act(() => { + testConnection.sessionId = 'session-b'; + testStore.update([blockB]); + root!.render(); + }); + + // Every render of the switched session must carry the new blocks — the + // manual render (bypass) and the deferred catch-up — never the previous + // session's snapshot. + expect(renderLog.length).toBeGreaterThan(0); + for (const entry of renderLog) { + expect(entry).toEqual(['b1']); + } + expect(latestBlocks).toEqual([blockB]); + }); + + it('does not return deferred blocks after a same-session reset', () => { + let pendingFrame: FrameRequestCallback | null = null; + vi.spyOn(window, 'requestAnimationFrame').mockImplementation((callback) => { + pendingFrame = callback; + return 1; + }); + vi.spyOn(performance, 'now').mockReturnValue(1_000); + const oldBlock = { id: 'old' } as DaemonTranscriptBlock; + testStore.update([oldBlock]); + container = document.createElement('div'); + document.body.append(container); + root = createRoot(container); + act(() => root!.render()); + renderLog = []; + + act(() => testStore.resetBlocks()); + act(() => pendingFrame?.(1_000)); + + expect(renderLog).not.toContainEqual(['old']); + expect(latestBlocks).toEqual([]); + }); }); diff --git a/packages/web-shell/client/hooks/useAnimationFrameTranscriptBlocks.ts b/packages/web-shell/client/hooks/useAnimationFrameTranscriptBlocks.ts index ab8387fd390..b80f4c43bbb 100644 --- a/packages/web-shell/client/hooks/useAnimationFrameTranscriptBlocks.ts +++ b/packages/web-shell/client/hooks/useAnimationFrameTranscriptBlocks.ts @@ -1,21 +1,69 @@ -import { useCallback, useSyncExternalStore } from 'react'; +import { + useCallback, + useDeferredValue, + useMemo, + useSyncExternalStore, +} from 'react'; import type { DaemonTranscriptBlock } from '@qwen-code/sdk/daemon'; -import { useTranscriptStore } from '@qwen-code/webui/daemon-react-sdk'; +import { + useConnection, + useTranscriptStore, +} from '@qwen-code/webui/daemon-react-sdk'; + +// Cap transcript re-renders at ~20fps. During streaming every network chunk +// notifies the store; each render then runs the O(transcript) normalization +// pass, so rendering at 60fps triples that cost per second while the visible +// text (itself throttled at 80ms for markdown) cannot change that fast. A +// 50ms window is still smooth for streaming text. +const TRANSCRIPT_RENDER_THROTTLE_MS = 50; +const INPUT_QUIET_WINDOW_MS = 100; +const MAX_INPUT_DEFERRAL_MS = 250; + +function hasPendingInput(): boolean { + const scheduling = ( + navigator as Navigator & { + scheduling?: { isInputPending?: () => boolean }; + } + ).scheduling; + return scheduling?.isInputPending?.() === true; +} export function useAnimationFrameTranscriptBlocks(): readonly DaemonTranscriptBlock[] { const store = useTranscriptStore(); + const { sessionId } = useConnection(); const subscribe = useCallback( (notify: () => void) => { let frame: number | null = null; + let lastNotifyTs = Number.NEGATIVE_INFINITY; + let lastInputTs = Number.NEGATIVE_INFINITY; + let pendingSinceTs: number | null = null; + const recordInput = () => { + lastInputTs = performance.now(); + }; + const dispatchWhenDue = (ts: number) => { + frame = null; + if ( + ts - lastNotifyTs >= TRANSCRIPT_RENDER_THROTTLE_MS && + ((ts - lastInputTs >= INPUT_QUIET_WINDOW_MS && !hasPendingInput()) || + (pendingSinceTs !== null && + ts - pendingSinceTs >= MAX_INPUT_DEFERRAL_MS)) + ) { + lastNotifyTs = ts; + pendingSinceTs = null; + notify(); + } else { + frame = window.requestAnimationFrame(dispatchWhenDue); + } + }; + document.addEventListener('beforeinput', recordInput, true); const unsubscribe = store.subscribe(() => { if (frame !== null) return; - frame = window.requestAnimationFrame(() => { - frame = null; - notify(); - }); + pendingSinceTs = performance.now(); + frame = window.requestAnimationFrame(dispatchWhenDue); }); return () => { unsubscribe(); + document.removeEventListener('beforeinput', recordInput, true); if (frame !== null) { window.cancelAnimationFrame(frame); } @@ -23,6 +71,41 @@ export function useAnimationFrameTranscriptBlocks(): readonly DaemonTranscriptBl }, [store], ); - const getSnapshot = useCallback(() => store.getSnapshot().blocks, [store]); - return useSyncExternalStore(subscribe, getSnapshot, getSnapshot); + const getSnapshot = useMemo(() => { + let cached: + | { + blocks: readonly DaemonTranscriptBlock[]; + blockIndexById: Readonly>; + } + | undefined; + return () => { + const state = store.getSnapshot(); + if ( + !cached || + cached.blocks !== state.blocks || + cached.blockIndexById !== state.blockIndexById + ) { + cached = { + blocks: state.blocks, + blockIndexById: state.blockIndexById, + }; + } + return cached; + }; + }, [store]); + const live = useSyncExternalStore(subscribe, getSnapshot, getSnapshot); + // Defer transcript re-renders so urgent updates — composer keystrokes, + // button presses — are never queued behind a streaming frame. The deferred + // value catches up on the next idle render, so streaming stays smooth while + // typing stays responsive. + // + // Session and block-index identities ride inside the deferred snapshot. The + // session id rejects a previous session, while the index identity rejects a + // same-session store reset without blocking ordinary streamed text updates. + const snapshot = useMemo(() => ({ sessionId, ...live }), [live, sessionId]); + const deferred = useDeferredValue(snapshot); + return deferred.sessionId === sessionId && + deferred.blockIndexById === live.blockIndexById + ? deferred.blocks + : live.blocks; } diff --git a/packages/web-shell/client/hooks/useMessages.test.ts b/packages/web-shell/client/hooks/useMessages.test.ts index 200846e527b..d9997be9be6 100644 --- a/packages/web-shell/client/hooks/useMessages.test.ts +++ b/packages/web-shell/client/hooks/useMessages.test.ts @@ -15,6 +15,7 @@ import { reconcileBackgroundAgentResolutions, transcriptBlocksToLocalizedMessages, useMessages, + useMessagesFromBlocks, } from './useMessages'; import type { Message } from '../adapters/types'; @@ -96,6 +97,81 @@ describe('transcriptBlocksToLocalizedMessages', () => { { content: 'localized:error.loopDetected' }, ]); }); + + it('preserves projected history identity for a streaming tail update only', async () => { + const container = document.createElement('div'); + const root = createRoot(container); + const t = (key: string) => key; + let latest: Message[] = []; + const user = baseBlock({ id: 'user', kind: 'user', text: 'hello' }); + const assistant = baseBlock({ + id: 'assistant', + kind: 'assistant', + text: 'a', + streaming: true, + }); + function Consumer({ blocks }: { blocks: DaemonTranscriptBlock[] }) { + latest = useMessagesFromBlocks(t, blocks); + return null; + } + + await act(async () => + root.render(createElement(Consumer, { blocks: [user, assistant] })), + ); + const firstProjection = latest; + const grownAssistant = { ...assistant, text: 'ab', updatedAt: 2 }; + await act(async () => + root.render(createElement(Consumer, { blocks: [user, grownAssistant] })), + ); + + expect(latest[0]).toBe(firstProjection[0]); + expect(latest[1]).not.toBe(firstProjection[1]); + expect(latest[1]).toMatchObject({ content: 'ab', isStreaming: true }); + + const changedUser = { ...user, text: 'changed', updatedAt: 2 }; + await act(async () => + root.render( + createElement(Consumer, { blocks: [changedUser, grownAssistant] }), + ), + ); + expect(latest[0]).not.toBe(firstProjection[0]); + + await act(async () => root.unmount()); + }); + + it.each([undefined, ''])( + 'falls back safely when a streaming block has empty text (%j)', + async (text) => { + const container = document.createElement('div'); + const root = createRoot(container); + const t = (key: string) => key; + let latest: Message[] = []; + const assistant = baseBlock({ + id: 'assistant', + kind: 'assistant', + text: text as unknown as string, + streaming: true, + }); + function Consumer({ blocks }: { blocks: DaemonTranscriptBlock[] }) { + latest = useMessagesFromBlocks(t, blocks); + return null; + } + + await act(async () => + root.render(createElement(Consumer, { blocks: [assistant] })), + ); + await act(async () => + root.render( + createElement(Consumer, { + blocks: [{ ...assistant, updatedAt: 2 }], + }), + ), + ); + + expect(latest).toEqual([]); + await act(async () => root.unmount()); + }, + ); }); function backgroundAgentMessage(status: 'pending' | 'completed' = 'pending') { diff --git a/packages/web-shell/client/hooks/useMessages.ts b/packages/web-shell/client/hooks/useMessages.ts index ce850357670..6c989d20059 100644 --- a/packages/web-shell/client/hooks/useMessages.ts +++ b/packages/web-shell/client/hooks/useMessages.ts @@ -1,4 +1,4 @@ -import { useEffect, useMemo, useRef, useState } from 'react'; +import { useEffect, useLayoutEffect, useMemo, useRef, useState } from 'react'; import { DaemonHttpError, isSessionLevelNotFound, @@ -61,6 +61,80 @@ export function transcriptBlocksToLocalizedMessages( }); } +function reuseUnchangedProjectedPrefix( + previous: + | { + blocks: readonly DaemonTranscriptBlock[]; + messages: Message[]; + t: Translator; + } + | undefined, + blocks: readonly DaemonTranscriptBlock[], + messages: Message[], + t: Translator, +): Message[] { + if ( + !previous || + previous.t !== t || + previous.blocks.length !== blocks.length || + previous.messages.length !== messages.length || + messages.length === 0 || + blocks.length === 0 + ) { + return messages; + } + for (let i = 0; i < blocks.length - 1; i += 1) { + if (previous.blocks[i] !== blocks[i]) return messages; + } + const before = previous.blocks[blocks.length - 1]; + const after = blocks[blocks.length - 1]; + if ( + (before.kind !== 'assistant' && before.kind !== 'thought') || + after.kind !== before.kind || + before.id !== after.id || + before.streaming !== true || + after.streaming !== true || + before.parentToolCallId !== undefined || + after.parentToolCallId !== undefined || + before.meta !== after.meta || + before.usage !== after.usage || + before.branchRecordId !== after.branchRecordId || + before.serverTimestamp !== after.serverTimestamp || + before.clientReceivedAt !== after.clientReceivedAt || + typeof before.text !== 'string' || + typeof after.text !== 'string' || + !after.text.startsWith(before.text) || + after.text.includes('"insight_') + ) { + return messages; + } + for (let i = 0; i < messages.length - 1; i += 1) { + const previousMessage = previous.messages[i]; + const message = messages[i]; + if ( + previousMessage.id !== message.id || + previousMessage.role !== message.role + ) { + return messages; + } + } + const previousTail = previous.messages[previous.messages.length - 1]; + const tail = messages[messages.length - 1]; + if ( + previousTail.id !== tail.id || + previousTail.role !== tail.role || + (tail.role !== 'assistant' && tail.role !== 'thinking') || + tail.isStreaming !== true + ) { + return messages; + } + const result = messages.slice(); + for (let i = 0; i < result.length - 1; i += 1) { + result[i] = previous.messages[i]; + } + return result; +} + function isTerminalBackgroundAgentStatus(status: string): boolean { return ( status === 'completed' || @@ -183,10 +257,27 @@ export function useMessagesFromBlocks( ): Message[] { const workspace = useWorkspace(); const connection = useConnection(); + const previousProjectionRef = useRef< + | { + blocks: readonly DaemonTranscriptBlock[]; + messages: Message[]; + t: Translator; + } + | undefined + >(undefined); const messages = useMemo( - () => transcriptBlocksToLocalizedMessages(blocks, t), + () => + reuseUnchangedProjectedPrefix( + previousProjectionRef.current, + blocks, + transcriptBlocksToLocalizedMessages(blocks, t), + t, + ), [blocks, t], ); + useLayoutEffect(() => { + previousProjectionRef.current = { blocks, messages, t }; + }, [blocks, messages, t]); const [resolutionSnapshot, setResolutionSnapshot] = useState<{ sessionId: string; resolutions: ReadonlyMap; diff --git a/packages/web-shell/package.json b/packages/web-shell/package.json index 866ca5b12c2..4d51b438d35 100644 --- a/packages/web-shell/package.json +++ b/packages/web-shell/package.json @@ -26,6 +26,7 @@ "test:ci": "vitest run --config vitest.config.ts --coverage", "test:coverage": "vitest run --config vitest.config.ts --coverage", "test:e2e:smoke": "playwright test --config playwright.config.ts --grep @smoke", + "test:e2e:perf": "cross-env WEB_SHELL_PERF=1 playwright test --config playwright.config.ts --grep @perf --project=chromium", "test:e2e": "playwright test --config playwright.config.ts", "test:e2e:visuals": "playwright test --config playwright.visuals.config.ts", "test:e2e:report": "playwright show-report client/e2e/playwright-report", diff --git a/packages/webui/src/daemon/session/DaemonSessionProvider.test.tsx b/packages/webui/src/daemon/session/DaemonSessionProvider.test.tsx index da373ebbdd5..c66b8f8589a 100644 --- a/packages/webui/src/daemon/session/DaemonSessionProvider.test.tsx +++ b/packages/webui/src/daemon/session/DaemonSessionProvider.test.tsx @@ -6707,9 +6707,9 @@ describe('DaemonSessionProvider', () => { await renderWithProvider(, { autoConnect: true }); await act(async () => { await flushPromises(); - // Batched transcript dispatch rides a setTimeout; under fake timers - // advance it so the passive assistant chunk lands before asserting. - await vi.advanceTimersByTimeAsync(0); + // Advance through the transcript batching window so the passive + // assistant chunk lands before asserting. + await vi.advanceTimersByTimeAsync(20); await flushPromises(); }); expect(blocks).toMatchObject([ @@ -12717,14 +12717,14 @@ async function flushPromises(): Promise { await Promise.resolve(); } -// Transcript dispatch is batched onto a macrotask (setTimeout 0) so a burst of +// Transcript dispatch is batched onto a short timer so a burst of // SSE events coalesces into one reducer pass. Stay-alive mock generators never // end the consumer loop (which would flush synchronously), so tests that assert // transcript state mid-stream drain the batched dispatch here. // Two hops are required because the dispatch timer and the first timer can be // registered from concurrently draining microtask chains in either order. async function flushTranscriptDispatch(): Promise { - await new Promise((resolve) => setTimeout(resolve, 0)); + await new Promise((resolve) => setTimeout(resolve, 20)); await new Promise((resolve) => setTimeout(resolve, 0)); await Promise.resolve(); await Promise.resolve(); diff --git a/packages/webui/src/daemon/session/DaemonSessionProvider.tsx b/packages/webui/src/daemon/session/DaemonSessionProvider.tsx index e9688b990b2..1d201ab0f01 100644 --- a/packages/webui/src/daemon/session/DaemonSessionProvider.tsx +++ b/packages/webui/src/daemon/session/DaemonSessionProvider.tsx @@ -500,6 +500,7 @@ interface HeartbeatFailureState { // is a history-preservation tradeoff rather than a claim that large transcripts // are CPU-free. Callers can pass a smaller maxBlocks in constrained contexts. const DEFAULT_MAX_BLOCKS = 200_000; +const TRANSCRIPT_DISPATCH_BATCH_MS = 16; const INITIAL_WORKSPACE_EVENT_SIGNALS: DaemonWorkspaceEventSignals = { memoryVersion: 0, @@ -784,8 +785,11 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { // drains already-buffered events back-to-back via microtasks, so a // microtask flush would run between every event and never coalesce. A // macrotask only runs once the generator blocks on a genuinely new network - // event, so a whole burst collapses into a single dispatch while steady - // streaming stays at ~one dispatch per network chunk. + // event. Holding the batch for one frame also coalesces steady streaming: + // copying a 50k-block immutable snapshot once per token otherwise consumes + // the main thread before the render throttle can help. Control and terminal + // paths call flushTranscriptSync below, so ordering and completion are not + // delayed by the window. let pendingTranscriptEvents: DaemonUiEvent[] = []; let transcriptFlushTimer: ReturnType | undefined; const runTranscriptFlush = (force = false) => { @@ -822,7 +826,10 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { if (events.length === 0) return; for (const event of events) pendingTranscriptEvents.push(event); if (transcriptFlushTimer === undefined) { - transcriptFlushTimer = setTimeout(runTranscriptFlush, 0); + transcriptFlushTimer = setTimeout( + runTranscriptFlush, + TRANSCRIPT_DISPATCH_BATCH_MS, + ); } }; // Apply buffered transcript events immediately. Called before any control From 08ad3ab8dc4fc4b06a9277fc263c20cfe42b8d44 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=92=89=E8=90=81?= Date: Wed, 19 Aug 2026 10:13:10 +0800 Subject: [PATCH 2/2] fix(web-shell): address streaming performance review --- .../components/MessageList.dom.test.tsx | 21 +++++- .../client/components/MessageList.tsx | 36 +++++++-- .../e2e/web-shell.stream-performance.spec.ts | 5 +- .../client/hooks/useMessages.test.ts | 8 +- .../web-shell/client/hooks/useMessages.ts | 6 +- .../session/DaemonSessionProvider.test.tsx | 73 +++++++++++++++++++ 6 files changed, 139 insertions(+), 10 deletions(-) diff --git a/packages/web-shell/client/components/MessageList.dom.test.tsx b/packages/web-shell/client/components/MessageList.dom.test.tsx index 033d75fd9f0..e2a5fb14de0 100644 --- a/packages/web-shell/client/components/MessageList.dom.test.tsx +++ b/packages/web-shell/client/components/MessageList.dom.test.tsx @@ -3217,6 +3217,7 @@ describe('MessageList — turn collapse (DOM)', () => { ...asstMsg('a1'), content: 'first chunk', isStreaming: true, + timestamp: 1_001, }; const messages = [userMsg('u1'), assistant]; const container = mount(messages, undefined, { @@ -3227,7 +3228,14 @@ describe('MessageList — turn collapse (DOM)', () => { rerenderMessages( container, - [messages[0], { ...assistant, content: 'first chunk plus delta' }], + [ + messages[0], + { + ...assistant, + content: 'first chunk plus delta', + timestamp: 1_002, + }, + ], { isResponding: true, compactMode }, ); @@ -3399,6 +3407,8 @@ describe('MessageList — turn collapse (DOM)', () => { expect(onLoadOlderHistory).toHaveBeenCalledTimes(1); virtualizerTestState.resizeItem.mockClear(); + await nextFrame(); + await nextFrame(); act(() => render([...earlierMessages, ...currentMessages])); expect(virtualizerTestState.resizeItem).toHaveBeenCalled(); @@ -3487,6 +3497,15 @@ describe('MessageList — turn collapse (DOM)', () => { await nextFrame(); expect(onLoadOlderHistory).toHaveBeenCalledTimes(2); + + for (let frame = 0; frame < 32; frame += 1) await nextFrame(); + await act(async () => { + list.dispatchEvent(new WheelEvent('wheel', { deltaY: -1 })); + await Promise.resolve(); + }); + await nextFrame(); + + expect(onLoadOlderHistory).toHaveBeenCalledTimes(3); }); it('waits for another upward scroll intent before retrying a failed underfill load', async () => { diff --git a/packages/web-shell/client/components/MessageList.tsx b/packages/web-shell/client/components/MessageList.tsx index 526999737aa..1776a82e3fc 100644 --- a/packages/web-shell/client/components/MessageList.tsx +++ b/packages/web-shell/client/components/MessageList.tsx @@ -164,8 +164,7 @@ function isStreamingTailContentOnlyUpdate( (after.role !== 'assistant' && after.role !== 'thinking') || (before.role !== 'assistant' && before.role !== 'thinking') || before.isStreaming !== true || - after.isStreaming !== true || - before.timestamp !== after.timestamp + after.isStreaming !== true ) { return false; } @@ -2673,6 +2672,9 @@ export const MessageList = memo( const { t } = useI18n(); const transcriptRenderMode = useTranscriptRenderMode(); const compactMode = useContext(CompactModeContext); + // Render-phase caches below are reusable only against this post-commit + // identity. An abandoned render cannot advance it, so its cache writes are + // rejected by the next committed render. const previousMessagesRef = useRef(undefined); const streamingTailContentOnly = isStreamingTailContentOnlyUpdate( previousMessagesRef.current, @@ -3179,6 +3181,10 @@ export const MessageList = memo( } | null>(null); const restoringOlderHistoryRef = useRef(false); restoringOlderHistoryRef.current = olderHistoryAnchor?.virtual === true; + const mergedMessageCountRef = useRef(mergedMessages.length); + useLayoutEffect(() => { + mergedMessageCountRef.current = mergedMessages.length; + }, [mergedMessages.length]); const [ suppressOlderHistoryLoadingStatus, setSuppressOlderHistoryLoadingStatus, @@ -3917,19 +3923,35 @@ export const MessageList = memo( ? mergedMessages.length === olderHistoryAnchor.messageCount : current.scrollHeight === olderHistoryAnchor.scrollHeight; if (unchanged) { + olderHistoryLoadInFlight.current = false; if (olderHistoryAnchorFrame.current !== undefined) return; - olderHistoryAnchorFrame.current = requestAnimationFrame(() => { + // The loader can resolve before the parent commits prepended messages. + // Keep the anchor through a bounded frame grace instead of clearing it + // after one busy frame. + let remainingFrames = OLDER_HISTORY_ANCHOR_WAIT_FRAMES; + const waitForPrepend = () => { olderHistoryAnchorFrame.current = undefined; if ( olderHistoryAnchor.generation !== olderHistoryLoadGeneration.current ) { return; } - olderHistoryLoadInFlight.current = false; + if ( + mergedMessageCountRef.current !== olderHistoryAnchor.messageCount + ) { + return; + } + remainingFrames -= 1; + if (remainingFrames > 0) { + olderHistoryAnchorFrame.current = + requestAnimationFrame(waitForPrepend); + return; + } setOlderHistoryAnchor((anchor) => anchor === olderHistoryAnchor ? null : anchor, ); - }); + }; + olderHistoryAnchorFrame.current = requestAnimationFrame(waitForPrepend); return; } if (olderHistoryAnchorFrame.current !== undefined) { @@ -4239,6 +4261,10 @@ export const MessageList = memo( olderHistoryRetryBlocked.current = false; olderHistoryLoadInFlight.current = true; const generation = ++olderHistoryLoadGeneration.current; + if (olderHistoryAnchorFrame.current !== undefined) { + cancelAnimationFrame(olderHistoryAnchorFrame.current); + olderHistoryAnchorFrame.current = undefined; + } setSuppressOlderHistoryLoadingStatus(!force); let virtualAnchor: | { diff --git a/packages/web-shell/client/e2e/web-shell.stream-performance.spec.ts b/packages/web-shell/client/e2e/web-shell.stream-performance.spec.ts index d8f312afcde..d584800ed90 100644 --- a/packages/web-shell/client/e2e/web-shell.stream-performance.spec.ts +++ b/packages/web-shell/client/e2e/web-shell.stream-performance.spec.ts @@ -64,7 +64,7 @@ test('keeps the composer responsive during deterministic streaming @perf', async }); streamEvents.push( assistantTextEvent('STREAM_COMPLETE_SENTINEL', { id: nextEventId++ }), - turnCompleteEvent('performance-prompt', { id: nextEventId++ }), + turnCompleteEvent('prompt-e2e', { id: nextEventId++ }), ); const type = async () => { @@ -83,6 +83,9 @@ test('keeps the composer responsive during deterministic streaming @perf', async await expect(page.locator('[data-web-shell-message-list]')).toContainText( 'STREAM_COMPLETE_SENTINEL', ); + await expect( + page.locator('[data-markdown-streaming-plain-text="true"]'), + ).toHaveCount(0); await page.waitForTimeout(100); const metrics = await readPerformanceMetrics(page); diff --git a/packages/web-shell/client/hooks/useMessages.test.ts b/packages/web-shell/client/hooks/useMessages.test.ts index d9997be9be6..9c3a1b0aaef 100644 --- a/packages/web-shell/client/hooks/useMessages.test.ts +++ b/packages/web-shell/client/hooks/useMessages.test.ts @@ -109,6 +109,7 @@ describe('transcriptBlocksToLocalizedMessages', () => { kind: 'assistant', text: 'a', streaming: true, + serverTimestamp: 1_001, }); function Consumer({ blocks }: { blocks: DaemonTranscriptBlock[] }) { latest = useMessagesFromBlocks(t, blocks); @@ -119,7 +120,12 @@ describe('transcriptBlocksToLocalizedMessages', () => { root.render(createElement(Consumer, { blocks: [user, assistant] })), ); const firstProjection = latest; - const grownAssistant = { ...assistant, text: 'ab', updatedAt: 2 }; + const grownAssistant = { + ...assistant, + text: 'ab', + updatedAt: 2, + serverTimestamp: 1_002, + }; await act(async () => root.render(createElement(Consumer, { blocks: [user, grownAssistant] })), ); diff --git a/packages/web-shell/client/hooks/useMessages.ts b/packages/web-shell/client/hooks/useMessages.ts index 6c989d20059..e65dea54767 100644 --- a/packages/web-shell/client/hooks/useMessages.ts +++ b/packages/web-shell/client/hooks/useMessages.ts @@ -34,6 +34,9 @@ const BACKGROUND_AGENT_RECONCILIATION_MAX_ATTEMPTS = 8; // call appears in the transcript, so a first `session_not_found` can race // registration. Require repeated misses before treating the agent as gone. const MISSING_BACKGROUND_AGENT_GRACE_MISSES = 2; +// Insight JSON can split one growing text block into multiple projected +// messages, so prefix identity reuse is unsafe once its marker appears. +const INSIGHT_CONTENT_MARKER = '"insight_'; export interface BackgroundAgentResolution { status: string; @@ -99,12 +102,11 @@ function reuseUnchangedProjectedPrefix( before.meta !== after.meta || before.usage !== after.usage || before.branchRecordId !== after.branchRecordId || - before.serverTimestamp !== after.serverTimestamp || before.clientReceivedAt !== after.clientReceivedAt || typeof before.text !== 'string' || typeof after.text !== 'string' || !after.text.startsWith(before.text) || - after.text.includes('"insight_') + after.text.includes(INSIGHT_CONTENT_MARKER) ) { return messages; } diff --git a/packages/webui/src/daemon/session/DaemonSessionProvider.test.tsx b/packages/webui/src/daemon/session/DaemonSessionProvider.test.tsx index c66b8f8589a..af3d6c362cc 100644 --- a/packages/webui/src/daemon/session/DaemonSessionProvider.test.tsx +++ b/packages/webui/src/daemon/session/DaemonSessionProvider.test.tsx @@ -3209,6 +3209,79 @@ describe('DaemonSessionProvider', () => { createStoreSpy.mockRestore(); }); + it('coalesces streamed chunks arriving within the dispatch window', async () => { + vi.useFakeTimers(); + const sdk = await import('@qwen-code/sdk/daemon'); + const realCreateStore = sdk.createDaemonTranscriptStore; + const dispatchBatchSizes: number[] = []; + const createStoreSpy = vi + .spyOn(sdk, 'createDaemonTranscriptStore') + .mockImplementation((seed) => { + const store = realCreateStore(seed); + const realDispatch = store.dispatch.bind(store); + store.dispatch = (event) => { + dispatchBatchSizes.push(Array.isArray(event) ? event.length : 1); + return realDispatch(event); + }; + return store; + }); + try { + const firstQueued = createDeferred(); + const secondQueued = createDeferred(); + const event = (id: number, text: string): DaemonEvent => ({ + id, + v: 1, + type: 'session_update', + data: { + update: { + sessionUpdate: 'agent_message_chunk', + content: { type: 'text', text }, + }, + }, + }); + const session = createMockSession({ + events: async function* spacedEvents( + opts: { signal?: AbortSignal } = {}, + ) { + yield event(1, 'first '); + firstQueued.resolve(); + await new Promise((resolve) => setTimeout(resolve, 5)); + yield event(2, 'second'); + secondQueued.resolve(); + await new Promise((resolve) => { + if (opts.signal?.aborted) { + resolve(); + return; + } + opts.signal?.addEventListener('abort', () => resolve(), { + once: true, + }); + }); + }, + }); + sdkMocks.sessions.push(session); + + await renderWithProvider(null, { autoConnect: true }); + await act(async () => { + await firstQueued.promise; + await flushPromises(); + await vi.advanceTimersByTimeAsync(5); + await secondQueued.promise; + await flushPromises(); + }); + expect(dispatchBatchSizes).toEqual([]); + + await act(async () => { + await vi.advanceTimersByTimeAsync(11); + await flushPromises(); + }); + expect(dispatchBatchSizes).toEqual([2]); + } finally { + createStoreSpy.mockRestore(); + vi.useRealTimers(); + } + }); + it('flushes buffered transcript events on unmount instead of dropping them', async () => { // The SSE client advances lastSeenEventId as each event is yielded, before // the batched dispatch runs. If teardown dropped the pending buffer, a