From b4fc0d1618a667bdeada2acfc1f67dfbcb8071ef Mon Sep 17 00:00:00 2001 From: Raing5Days Date: Sun, 25 Oct 2026 00:33:27 +0800 Subject: [PATCH 1/3] perf(flow-chat): cut per-flush cost on the streaming render path Four fixes found while profiling the Web UI streaming hot path. Each removes work that ran unconditionally on every batched flush (32ms while text streams). - TextChunkModule: newline coalescing re-scanned the whole accumulated response on every chunk, which is O(N^2) over a long answer. Both sides of an append are already clean, so only the two-character join boundary plus the incoming chunk need cleaning. Reused item content is cleaned once when recovered so the buffer invariant still holds. - EventBatcher: build the trace payload only when trace output can reach a sink. The logger drops the entry after its level check, so every flush previously allocated a full event map for a discarded log line. The slow-flush warning keeps its counts. - modernFlowChatStore: drop the immer middleware. The setters only assign whole fields, so produce() and its deep freeze walked the entire session and virtual item tree on every flush for no benefit. Also reset the needsUserAttention projection cache guard in clear(). - useConversationTimeline: remember prewarmed text items in a WeakSet so a streaming flush no longer re-walks the whole timeline. Verification: 133 vitest cases across the touched modules pass, tsc --noEmit is clean, eslint reports 0 errors. Web UI only; no remote-workspace, remote-control, peer-device, or dispatch protocol behavior changes. --- .../src/flow_chat/services/EventBatcher.ts | 13 +++-- .../flow-chat-manager/TextChunkModule.ts | 31 ++++++++++-- .../flow_chat/store/modernFlowChatStore.ts | 49 ++++++++++--------- .../timeline/useConversationTimeline.ts | 16 ++++-- 4 files changed, 75 insertions(+), 34 deletions(-) diff --git a/src/web-ui/src/flow_chat/services/EventBatcher.ts b/src/web-ui/src/flow_chat/services/EventBatcher.ts index 12dcc07763..3f56864a98 100644 --- a/src/web-ui/src/flow_chat/services/EventBatcher.ts +++ b/src/web-ui/src/flow_chat/services/EventBatcher.ts @@ -9,7 +9,7 @@ * - Merge keys are scoped by session, round/tool id, and retry attempt */ -import { areSensitiveDiagnosticsEnabled, createLogger } from '@/shared/utils/logger'; +import { areSensitiveDiagnosticsEnabled, createLogger, logger, LogLevel } from '@/shared/utils/logger'; import { elapsedMs, nowMs } from '@/shared/utils/timing'; const log = createLogger('EventBatcher'); @@ -228,14 +228,21 @@ export class EventBatcher { const startTime = nowMs(); const bufferedEvents = Array.from(this.buffer.values()); - const logPayload = getBatchedEventsLogPayload(bufferedEvents); - const { rawEventCount, mergedEventCount } = logPayload; const events = bufferedEvents.map(({ key, payload }) => ({ key, payload })); + // Build the diagnostic payload only when trace output can reach a sink; + // otherwise every flush pays a full event map for a dropped log entry. + // `rawEventCount` stays 0 when skipped: the count is only consumed by the + // trace entry and the slow-flush warning below reports the duration. + const logPayload = logger.getLevel() > LogLevel.TRACE + ? { rawEventCount: 0, mergedEventCount: bufferedEvents.length } + : getBatchedEventsLogPayload(bufferedEvents); + const { rawEventCount, mergedEventCount } = logPayload; + log.trace('Flushing batched events', logPayload); this.buffer = new Map(); diff --git a/src/web-ui/src/flow_chat/services/flow-chat-manager/TextChunkModule.ts b/src/web-ui/src/flow_chat/services/flow-chat-manager/TextChunkModule.ts index 155a64a862..4f8d123693 100644 --- a/src/web-ui/src/flow_chat/services/flow-chat-manager/TextChunkModule.ts +++ b/src/web-ui/src/flow_chat/services/flow-chat-manager/TextChunkModule.ts @@ -20,6 +20,29 @@ function resolveAttemptStreamKey(roundId: string, attemptId?: string, attemptInd return roundId; } +/** + * Coalesce runs of three or more newlines down to two. + * + * Both sides of an append are already clean, so a new run can only appear at + * the join boundary or inside the incoming chunk. Rescanning the whole + * accumulated response on every chunk made streaming cost quadratic in the + * response length; the overlap only has to cover the longest run a clean + * prefix can end with (two newlines). + */ +const NEWLINE_JOIN_OVERLAP_CHARS = 2; + +function coalesceNewlineRuns(content: string): string { + return content.replace(/\n{3,}/g, '\n\n'); +} + +function appendCoalescingNewlines(current: string, text: string): string { + if (!current) return coalesceNewlineRuns(text); + if (!text) return current; + const splitAt = Math.min(current.length, NEWLINE_JOIN_OVERLAP_CHARS); + return current.slice(0, current.length - splitAt) + + coalesceNewlineRuns(current.slice(current.length - splitAt) + text); +} + function activeTextHasLaterRoundItem( context: FlowChatContext, sessionId: string, @@ -125,7 +148,7 @@ export function processNormalTextChunkInternal( if (reusableTextItem) { textItemId = reusableTextItem.id; - recoveredExistingContent = reusableTextItem.content || ''; + recoveredExistingContent = coalesceNewlineRuns(reusableTextItem.content || ''); sessionActiveTextItems.set(streamKey, textItemId); if (!sessionContentBuffer.has(streamKey) && recoveredExistingContent) { sessionContentBuffer.set(streamKey, recoveredExistingContent); @@ -135,7 +158,7 @@ export function processNormalTextChunkInternal( // Coalesce excessive newlines while appending. const currentContent = sessionContentBuffer.get(streamKey) || recoveredExistingContent; - const cleanedContent = (currentContent + text).replace(/\n{3,}/g, '\n\n'); + const cleanedContent = appendCoalescingNewlines(currentContent, text); sessionContentBuffer.set(streamKey, cleanedContent); if (!textItemId) { @@ -213,7 +236,7 @@ export function processThinkingChunkInternal( if (reusableThinkingItem) { thinkingItemId = reusableThinkingItem.id; - recoveredExistingContent = reusableThinkingItem.content || ''; + recoveredExistingContent = coalesceNewlineRuns(reusableThinkingItem.content || ''); sessionActiveTextItems.set(thinkingKey, thinkingItemId); if (!sessionContentBuffer.has(thinkingKey) && recoveredExistingContent) { sessionContentBuffer.set(thinkingKey, recoveredExistingContent); @@ -222,7 +245,7 @@ export function processThinkingChunkInternal( } const currentContent = sessionContentBuffer.get(thinkingKey) || recoveredExistingContent; - const cleanedContent = (currentContent + text).replace(/\n{3,}/g, '\n\n'); + const cleanedContent = appendCoalescingNewlines(currentContent, text); sessionContentBuffer.set(thinkingKey, cleanedContent); if (!thinkingItemId) { diff --git a/src/web-ui/src/flow_chat/store/modernFlowChatStore.ts b/src/web-ui/src/flow_chat/store/modernFlowChatStore.ts index 2024056099..5a5a6ef7c8 100644 --- a/src/web-ui/src/flow_chat/store/modernFlowChatStore.ts +++ b/src/web-ui/src/flow_chat/store/modernFlowChatStore.ts @@ -1,13 +1,12 @@ /** * Modern FlowChat Store - * High-performance state management using Zustand + Immer + * High-performance state management using Zustand * Preserves original concept: Session → DialogTurn → ModelRound → FlowItem */ import { create, useStore } from 'zustand'; import { createContext, useContext } from 'react'; import { useShallow } from 'zustand/react/shallow'; -import { immer } from 'zustand/middleware/immer'; import type { Session, DialogTurn, ModelRound, ModelRoundAttempt, FlowThinkingItem, FlowUserSteeringItem, AnyFlowItem } from '../types/flow-chat'; import { flowChatStore } from './FlowChatStore'; import { getTurnCompletionNotice } from '../utils/turnCompletionNotice'; @@ -382,7 +381,7 @@ function getInitialModernState(): Pick< } export const createModernFlowChatStore = (initialSession?: Session | null) => create()( - immer((set, get) => ({ + (set, get) => ({ ...(initialSession === undefined ? getInitialModernState() : { activeSession: initialSession, virtualItems: sessionToVirtualItems(initialSession), @@ -391,28 +390,28 @@ export const createModernFlowChatStore = (initialSession?: Session | null) => cr setActiveSession: (session) => { const items = sessionToVirtualItems(session); - set((state) => { - if (state.activeSession?.sessionId !== session?.sessionId) { - state.visibleTurnInfo = null; - } - state.activeSession = session; - state.virtualItems = items; - }); + set((state) => ({ + ...(state.activeSession?.sessionId !== session?.sessionId ? { visibleTurnInfo: null } : null), + activeSession: session, + virtualItems: items, + })); }, updateVirtualItems: () => { const session = get().activeSession; const items = sessionToVirtualItems(session); - - set((state) => { - state.virtualItems = items; - }); + + set((state) => ({ + ...state, + virtualItems: items, + })); }, setVisibleTurnInfo: (info) => { - set((state) => { - state.visibleTurnInfo = info; - }); + set((state) => ({ + ...state, + visibleTurnInfo: info, + })); }, clear: () => { @@ -423,14 +422,16 @@ export const createModernFlowChatStore = (initialSession?: Session | null) => cr cachedTotalTurnCount = undefined; cachedVirtualItems = []; cachedTurnItems = new WeakMap(); - - set((state) => { - state.activeSession = null; - state.virtualItems = []; - state.visibleTurnInfo = null; - }); + cachedNeedsUserAttention = undefined; + + set((state) => ({ + ...state, + activeSession: null, + virtualItems: [], + visibleTurnInfo: null, + })); }, - })) + }) ); const defaultModernFlowChatStore = createModernFlowChatStore(); diff --git a/src/web-ui/src/flow_chat/timeline/useConversationTimeline.ts b/src/web-ui/src/flow_chat/timeline/useConversationTimeline.ts index b0c7735cf8..1b4df07410 100644 --- a/src/web-ui/src/flow_chat/timeline/useConversationTimeline.ts +++ b/src/web-ui/src/flow_chat/timeline/useConversationTimeline.ts @@ -22,16 +22,26 @@ export function useConversationTimeline(sources: VirtualItem[], sessionId: strin } const projection = projectionRef.current.value; const labels = useMemo(() => flowGroupToolLabels(t), [t]); + // Every streamed flush produces a new sources array; skip texts already + // queued or parsed instead of re-walking the whole timeline each time. + const prewarmedTextsRef = useRef>(new WeakSet()); useEffect(() => { + const seen = prewarmedTextsRef.current; + const cancel: Array<() => void> = []; const texts = new Set(); for (const source of sources) { const members = source.type === 'explore-group' ? source.data.allItems : source.type === 'model-round' ? source.projectedGroups?.flatMap(group => group.type === 'critical' ? [group.item] : group.projection?.allItems ?? group.items) ?? source.data.items : []; - for (const member of members) if (member.type === 'text' && !('isStreaming' in member && member.isStreaming) - && 'content' in member && typeof member.content === 'string' && member.content.length >= 2000) texts.add(member.content); + for (const member of members) { + if (member.type !== 'text' || !('content' in member) || typeof member.content !== 'string') continue; + if ('isStreaming' in member && member.isStreaming) continue; + if (seen.has(member)) continue; + seen.add(member); + if (member.content.length >= 2000) texts.add(member.content); + } } - const cancel = [...texts].map(prewarmMarkdownParse); + for (const text of texts) cancel.push(prewarmMarkdownParse(text)); return () => cancel.forEach(stop => stop()); }, [sources]); const items = useMemo(() => { From ce4b0d8b483660d5f048ba88bf174d963231e4ae Mon Sep 17 00:00:00 2001 From: Raing5Days Date: Sun, 25 Oct 2026 20:56:48 +0800 Subject: [PATCH 2/3] perf(flow-chat): stop per-flush work from scaling with transcript size A second pass over the same streaming path, looking for work that grows with the conversation rather than with the token that arrived. - timeline/document: every projection rebuilt a state-sized revision string and embedded it into a fresh per-source revision, then matched the cache by comparing those strings. That is O(sources x |view state|) of string building and comparing per flush. The projection now stamps the reader/group inputs by identity and compares one number per source. Identity is what the caller's own memo already keys on, so the contract is unchanged. - FlowChatStore.updateModelRoundItem: every round ahead of the target re-derived its attempts from items and threw the result away; a streamed token always lands in the last round. Rounds that neither recorded attempts nor hold the item now return before the derivation. - useFlowChatVirtualizer: the key sequence was enumerated twice per flush (a join for the revision, a Set for retired measurements), allocating one string per row each time. Keys are now cached per row object and the sequence is compared by reference, so a quiet flush allocates nothing; the join and the Set only run when the sequence actually changed. The cache bypasses primitives because the hook stays generic. - VirtualMessageList: the estimate revision collator-sorted the disclosure map on every render. It is now memoized on that map's identity. Verification: 4311 vitest cases across src/flow_chat pass, tsc --noEmit clean, eslint 0 errors. Local Web UI only; no remote scenario behavior changes. --- .../components/modern/VirtualMessageList.tsx | 19 ++++--- .../modern/useFlowChatVirtualizer.ts | 52 +++++++++++++++++-- .../src/flow_chat/store/FlowChatStore.ts | 11 +++- src/web-ui/src/flow_chat/timeline/document.ts | 49 ++++++++++++++--- 4 files changed, 112 insertions(+), 19 deletions(-) diff --git a/src/web-ui/src/flow_chat/components/modern/VirtualMessageList.tsx b/src/web-ui/src/flow_chat/components/modern/VirtualMessageList.tsx index 5633716ad4..426f038039 100644 --- a/src/web-ui/src/flow_chat/components/modern/VirtualMessageList.tsx +++ b/src/web-ui/src/flow_chat/components/modern/VirtualMessageList.tsx @@ -524,6 +524,16 @@ const VirtualMessageListSession = forwardRef viewportOwner.currentOwner() === 'follow-output', undefined, kind => readerInteractionRef.current(kind)); const pinnedKeys = useMemo(() => submissionKey ? new Set([...interactionKeys, submissionKey]) : interactionKeys, [interactionKeys, submissionKey]); + // A quiet stream re-renders this list on every flush; collator-sorting the + // disclosure map is only worth paying when the map itself changed. + const estimateContextRevision = useMemo(() => [ + viewportWidthPx, + activeSession?.isHistorical === true ? 'historical' : 'live', + [...(exploreGroupStates?.entries() ?? [])] + .sort(([left], [right]) => left.localeCompare(right)) + .map(([groupId, expanded]) => `${groupId}:${expanded ? 1 : 0}`) + .join(','), + ].join('|'), [viewportWidthPx, activeSession?.isHistorical, exploreGroupStates]); const virtualizer = useFlowChatVirtualizer({ pinnedKeys, items: virtualItems, @@ -539,14 +549,7 @@ const VirtualMessageListSession = forwardRef left.localeCompare(right)) - .map(([groupId, expanded]) => `${groupId}:${expanded ? 1 : 0}`) - .join(','), - ].join('|'), + estimateContextRevision, isViewportSuspended: () => isViewportSuspendedRef.current, measurementSuspended, scrollPaddingStartPx: FLOWCHAT_TURN_TOP_GAP_PX, diff --git a/src/web-ui/src/flow_chat/components/modern/useFlowChatVirtualizer.ts b/src/web-ui/src/flow_chat/components/modern/useFlowChatVirtualizer.ts index e2238e9933..f8f8754c86 100644 --- a/src/web-ui/src/flow_chat/components/modern/useFlowChatVirtualizer.ts +++ b/src/web-ui/src/flow_chat/components/modern/useFlowChatVirtualizer.ts @@ -475,6 +475,51 @@ export function useFlowChatVirtualizer({ const aimStartedAtMsRef = useRef(null); const getItemKeyRef = useRef(getItemKey); getItemKeyRef.current = getItemKey; + const itemKeyCacheOwnerRef = useRef(getItemKey); + const itemKeyCacheRef = useRef(new WeakMap()); + /** + * Keys are derived strings, so a fresh enumeration allocates one string per + * row. A quiet stream reuses every row object but the live tail, so one + * string per row object lets the sequence be compared by reference instead of + * re-joined on every flush. + */ + const itemKeyOf = useCallback((item: T): string => { + // Items are rows in production but the hook stays generic, and a primitive + // cannot be a weak map key. + if (item === null || typeof item !== 'object') return getItemKeyRef.current(item); + const cache = itemKeyCacheRef.current; + const asObject = item as unknown as object; + const cached = cache.get(asObject); + if (cached !== undefined) return cached; + const key = getItemKeyRef.current(item); + cache.set(asObject, key); + return key; + }, []); + const itemKeySequenceRef = useRef<{ keys: string[]; revision: string }>({ keys: [], revision: '' }); + const itemKeysRef = useRef([]); + const itemKeyState = useMemo(() => { + // A new key function invalidates every cached key. + if (itemKeyCacheOwnerRef.current !== getItemKey) { + itemKeyCacheOwnerRef.current = getItemKey; + itemKeyCacheRef.current = new WeakMap(); + } + const previous = itemKeySequenceRef.current; + if (previous.keys.length === items.length) { + let same = true; + for (let index = 0; index < items.length; index += 1) { + if (previous.keys[index] !== itemKeyOf(items[index])) { + same = false; + break; + } + } + if (same) return previous; + } + const keys = items.map(itemKeyOf); + const next = { keys, revision: keys.join('\u0000') }; + itemKeySequenceRef.current = next; + return next; + }, [items, itemKeyOf, getItemKey]); + itemKeysRef.current = itemKeyState.keys; const estimateItemHeightRef = useRef(estimateItemHeightPx); estimateItemHeightRef.current = estimateItemHeightPx; const estimateContextRef = useRef(estimateContext); @@ -508,7 +553,7 @@ export function useFlowChatVirtualizer({ return item === undefined ? 0 : estimateItemHeightRef.current(item, estimateContextRef.current); }, []); - const itemKeyRevision = useMemo(() => items.map(getItemKey).join('\u0000'), [items, getItemKey]); + const itemKeyRevision = itemKeyState.revision; const resolveItemKey = useCallback((index: number) => { // Recreate the callback when estimate inputs change. TanStack uses the // callback identity to invalidate its derived measurement positions while @@ -657,9 +702,10 @@ export function useFlowChatVirtualizer({ } const previous = previousKnownHeightsRef.current; const next = new Map(); + const keys = itemKeysRef.current; for (let index = 0; index < items.length; index += 1) { const item = items[index]; - const key = getItemKeyRef.current(item); + const key = keys[index] ?? getItemKeyRef.current(item); const knownHeight = getKnownItemHeightPx(item); if (knownHeight !== undefined) next.set(key, knownHeight); if (knownHeight === undefined && !previous.has(key)) continue; @@ -776,7 +822,7 @@ export function useFlowChatVirtualizer({ /** A pass owed to a layout that a pane drag had not finished moving. */ const pendingLayoutMeasureRef = useRef(false); useLayoutEffect(() => { - const keys = new Set(items.map(getItemKeyRef.current)); + const keys = new Set(itemKeysRef.current); for (const key of previousKeysRef.current) if (!keys.has(key)) retiredMeasurementsRef.current++; previousKeysRef.current = keys; if (measuredLayoutRef.current === layoutSignature && retiredMeasurementsRef.current < 2048 diff --git a/src/web-ui/src/flow_chat/store/FlowChatStore.ts b/src/web-ui/src/flow_chat/store/FlowChatStore.ts index 747ea212d5..feb737fa65 100644 --- a/src/web-ui/src/flow_chat/store/FlowChatStore.ts +++ b/src/web-ui/src/flow_chat/store/FlowChatStore.ts @@ -6587,7 +6587,16 @@ export class FlowChatStore { const updatedModelRounds = turn.modelRounds.map(modelRound => { if (updated) return modelRound; - const activeAttempts = modelRound.attempts ?? deriveRoundAttemptsFromItems(modelRound.items); + const recordedAttempts = modelRound.attempts; + if (!recordedAttempts && !modelRound.items.some((item: any) => itemMatchesIdentity(item, itemId))) { + // Derived attempts only repackage these same items, so a round that + // does not hold the item cannot hold it after derivation either. A + // streamed token lands in the last round; without this probe every + // earlier round re-derives its attempts on every flush. + return modelRound; + } + + const activeAttempts = recordedAttempts ?? deriveRoundAttemptsFromItems(modelRound.items); if (activeAttempts && activeAttempts.length > 0) { let foundInAttempts = false; const nextAttempts = activeAttempts.map(attempt => { diff --git a/src/web-ui/src/flow_chat/timeline/document.ts b/src/web-ui/src/flow_chat/timeline/document.ts index a2741ec5d6..bf20222373 100644 --- a/src/web-ui/src/flow_chat/timeline/document.ts +++ b/src/web-ui/src/flow_chat/timeline/document.ts @@ -60,9 +60,45 @@ const emptyItems: FlowItem[] = []; */ export class ConversationDocumentProjection { private previous = new Map(); - private sourceCache = new WeakMap(); + private sourceCache = new WeakMap(); private next = new Map(); private labels?: ReadonlyMap; + private viewStamp = 0; + private viewInputs?: { + readerRevision: number; + groupStates?: ReadonlyMap; + expandedToolCapsules?: ReadonlySet; + pendingPermissionToolCallIds?: ReadonlySet; + stamp: number; + }; + + /** + * A stamp for the reader and group inputs, reused while they keep their + * identity. Text streaming refreshes `sources` on every flush while these + * stay the same object, so a quiet flush compares one number per source + * instead of rebuilding a state-sized string and matching it by content. + * Identity is what the caller already keys its own memo on. + */ + private viewStampFor(options: TimelineProjectionOptions): number { + const readerRevision = options.reader.getProjectionRevision(); + const cached = this.viewInputs; + if (cached && cached.readerRevision === readerRevision + && cached.groupStates === options.groupStates + && cached.expandedToolCapsules === options.expandedToolCapsules + && cached.pendingPermissionToolCallIds === options.pendingPermissionToolCallIds) { + return cached.stamp; + } + + const stamp = ++this.viewStamp; + this.viewInputs = { + readerRevision, + groupStates: options.groupStates, + expandedToolCapsules: options.expandedToolCapsules, + pendingPermissionToolCallIds: options.pendingPermissionToolCallIds, + stamp, + }; + return stamp; + } project(sources: readonly VirtualItem[], options: TimelineProjectionOptions): VirtualItem[] { if (this.labels !== options.toolLabels) { @@ -71,7 +107,7 @@ export class ConversationDocumentProjection { } this.next = new Map(); const result: VirtualItem[] = []; - const viewRevision = `${options.reader.getProjectionRevision()}:${[...(options.groupStates ?? [])].map(([id, value]) => `${id}:${value}`).join('|')}:${[...(options.expandedToolCapsules ?? [])].join('|')}:${[...(options.pendingPermissionToolCallIds ?? [])].join('|')}`; + const viewStamp = this.viewStampFor(options); sources.forEach((original, sourceIndex) => { // Legacy whole-round groups obey the same residency contract. This is a // render adapter only; neither their recorded rounds nor ids are changed. @@ -86,19 +122,18 @@ export class ConversationDocumentProjection { if (source.type !== 'model-round') { const key = getVirtualItemStableKey(source); const cached = this.sourceCache.get(source); - const row = cached?.revision === String(sourceIndex) ? cached.items[0] : { + const row = cached?.sourceIndex === sourceIndex ? cached.items[0] : { ...source, timeline: { key, kind: 'content' as const, sourceIndex, memberIds: source.type === 'background-run' ? source.turn.modelRounds.flatMap(round => round.items.map(item => item.id)) : [] }, }; - this.sourceCache.set(source, { revision: String(sourceIndex), items: [row] }); + this.sourceCache.set(source, { viewStamp, sourceIndex, items: [row] }); this.next.set(key, row); result.push(row); return; } // Source identity is already shared by completed immutable Turns. Reader // changes only invalidate their projection, never their payloads. - const revision = `${sourceIndex}:${viewRevision}`; const cached = this.sourceCache.get(original); - if (cached?.revision === revision) { + if (cached?.viewStamp === viewStamp && cached.sourceIndex === sourceIndex) { for (const row of cached.items) this.next.set(row.timeline!.key, row); result.push(...cached.items); return; @@ -178,7 +213,7 @@ export class ConversationDocumentProjection { if ((source.isLastRound && source.isTurnComplete) || source.canvasArtifactItems?.length) { add('round-footer', `${source.data.id}:footer`, emptyItems); } - this.sourceCache.set(original, { revision, items: rows }); + this.sourceCache.set(original, { viewStamp, sourceIndex, items: rows }); result.push(...rows); }); this.previous = this.next; From 5f303ea98d1ff64f54f61e6f43252f36d68e6dc7 Mon Sep 17 00:00:00 2001 From: Raing5Days Date: Sun, 25 Oct 2026 22:55:17 +0800 Subject: [PATCH 3/3] perf(flow-chat): trim per-flush projection allocations on the live round A third pass over the same streaming path, now looking at what the live round re-projects on every batched flush. - timeline/document, disclosure: readerExpanded asked whether any member of a group is an expanded tool capsule by serializing a key per member -- toolCapsuleStateKey is a three-element JSON.stringify. That ran for every member of every group even when nothing was expanded, which is the common case. The expanded keys are now inverted once per projection (turn id to member ids) and the member scan is skipped entirely when the set is empty. A test pins the key format this now depends on. - timeline/document, block writer: add() built the whole row before deciding to reuse the previous one -- the data spread plus a member scan for isStreaming, projectedGroups and memberIds -- so a quiet flush allocated payloads it immediately discarded. The reuse predicate now reads cheap primitives first and only the misses build a row. - grouping/groupProjection: the threshold test copied the item list per group, the merged-run finalizer flattened each part four times while building its source ids, and the trailing pass re-mapped every row. The threshold is now a counting scan, each part is flattened once, and rows whose groups already meet the threshold keep their identity. Verification: 499 vitest cases across flow_chat/{grouping,timeline,store}, 1090 across flow_chat/components/modern, tsc --noEmit clean, eslint 0 errors. Local Web UI only; no remote scenario behavior changes. --- .../src/flow_chat/grouping/groupProjection.ts | 88 +++++++++++++----- .../src/flow_chat/timeline/document.test.ts | 28 ++++++ src/web-ui/src/flow_chat/timeline/document.ts | 89 +++++++++++++------ 3 files changed, 156 insertions(+), 49 deletions(-) diff --git a/src/web-ui/src/flow_chat/grouping/groupProjection.ts b/src/web-ui/src/flow_chat/grouping/groupProjection.ts index 7766a44a3b..0502604e3f 100644 --- a/src/web-ui/src/flow_chat/grouping/groupProjection.ts +++ b/src/web-ui/src/flow_chat/grouping/groupProjection.ts @@ -1,34 +1,68 @@ import type { VirtualItem } from '../types/flow-chat-projection'; -import type { FlowItem } from '../types/flow-chat'; +import type { FlowItem, ModelRound } from '../types/flow-chat'; import type { FlowGroupData, ModelRoundItemGroup } from './types'; import { getFlowGroupCategory, getFlowGroupStateIds } from './types'; -import { canContinueFlowGroup, flowGroupPolicies, hasPendingFlowGroupClassification, joinedFlowGroupCategory, meetsFlowGroupThreshold } from './policies'; +import { canContinueFlowGroup, flowGroupPolicies, hasPendingFlowGroupClassification, joinedFlowGroupCategory } from './policies'; import { flowGroupLifecycle } from './lifecycle'; import { buildInlineFlowGroupData, getModelRoundFlowGroups, getProjectedModelRoundGroups, flowGroupCompanionStart, hasModelRoundLeadingControls, isFlowGroupCompanion, isModelRoundGroupingDisabled, projectedFlowGroup } from './roundGroups'; +/** Count only until the policy minimum; the shared threshold helper copies the item list. */ +function reachesFlowGroupThreshold(category: ReturnType, items: readonly FlowItem[]): boolean { + const minimum = flowGroupPolicies[category].minimumItems; + let collected = 0; + for (const item of items) { + if (item.type === 'tool' && ++collected >= minimum) return true; + } + return false; +} + +/** Add this group's disclosure identities without rebuilding intermediate arrays. */ +function collectSourceGroupIds(group: FlowGroupData, ids: Set): void { + for (const id of getFlowGroupStateIds(group)) ids.add(id); + // A settled pure round may previously have rendered an inline group. + if (group.sourceGroupIds) return; + for (const round of group.rounds) { + ids.add(round.id); + for (const source of getModelRoundFlowGroups(round)) ids.add(source.groupId); + } +} + function sourceGroupIds(group: FlowGroupData): readonly string[] { - if (group.sourceGroupIds) return getFlowGroupStateIds(group); - return [...new Set([ - ...getFlowGroupStateIds(group), - // A settled pure round may previously have rendered an inline group. - ...group.rounds.flatMap(round => [round.id, ...getModelRoundFlowGroups(round).map(source => source.groupId)]), - ])]; + const ids = new Set(); + collectSourceGroupIds(group, ids); + return [...ids]; } /** Finalize once per run instead of repeatedly copying and recounting its growing prefix. */ function mergeFlowGroupRun(parts: readonly FlowGroupData[]): FlowGroupData { const first = parts[0]; const last = parts[parts.length - 1]; - const category = parts.reduce | undefined>( - (joined, part) => joinedFlowGroupCategory(joined, getFlowGroupCategory(part)), undefined, - ) ?? getFlowGroupCategory(first); - return flowGroupPolicies[category].create({ + const mergedIds = new Set(); + const rounds: ModelRound[] = []; + const allItems: FlowItem[] = []; + let category: ReturnType | undefined; + let isGroupStreaming = false; + for (const part of parts) { + category = joinedFlowGroupCategory(category, getFlowGroupCategory(part)); + for (const id of getFlowGroupStateIds(part)) mergedIds.add(id); + // Recorded aliases already cover the rounds they were derived from. + const inferFromRounds = part.sourceGroupIds === undefined; + for (const round of part.rounds) { + rounds.push(round); + if (!inferFromRounds) continue; + mergedIds.add(round.id); + for (const source of getModelRoundFlowGroups(round)) mergedIds.add(source.groupId); + } + for (const item of part.allItems) allItems.push(item); + if (part.isGroupStreaming) isGroupStreaming = true; + } + return flowGroupPolicies[category ?? getFlowGroupCategory(first)].create({ ...first, - sourceGroupIds: [...new Set(parts.flatMap(part => [...sourceGroupIds(part)]))], - rounds: parts.flatMap(part => part.rounds), - allItems: parts.flatMap(part => part.allItems), - isGroupStreaming: parts.some(part => part.isGroupStreaming), + sourceGroupIds: [...mergedIds], + rounds, + allItems, + isGroupStreaming, isLastGroupInTurn: last.isLastGroupInTurn, wasCutByCritical: last.wasCutByCritical, }); @@ -52,7 +86,7 @@ export function projectAdjacentFlowGroups(items: readonly VirtualItem[], options const flush = (sealed = true) => { if (owner) { const data = owner.parts.length > 1 ? mergeFlowGroupRun(owner.parts) : owner.parts[0]; - if (meetsFlowGroupThreshold(getFlowGroupCategory(data), data.allItems)) { + if (reachesFlowGroupThreshold(getFlowGroupCategory(data), data.allItems)) { owner.update({ ...data, ...flowGroupLifecycle(data.allItems, sealed) }); owner.removeCollected.forEach(remove => remove()); } @@ -166,10 +200,18 @@ export function projectAdjacentFlowGroups(items: readonly VirtualItem[], options roundUpdates.forEach(update => update()); // Thresholds apply after cross-round adjacency; a single context call stays native. - return projected.filter((item): item is VirtualItem => item !== undefined).map(item => { - if (item.type !== 'model-round' || !item.projectedGroups?.some(group => group.type !== 'critical' && !meetsFlowGroupThreshold(group.type, group.items))) return item; - return { ...item, projectedGroups: item.projectedGroups.flatMap((group): ModelRoundItemGroup[] => - group.type !== 'critical' && !meetsFlowGroupThreshold(group.type, group.items) - ? group.items.map(member => ({ type: 'critical', item: member })) : [group]) }; - }); + const rendered: VirtualItem[] = []; + for (const item of projected) { + if (item === undefined) continue; + // Rows that already satisfy every group threshold keep their identity. + if (item.type !== 'model-round' || !item.projectedGroups?.some( + group => group.type !== 'critical' && !reachesFlowGroupThreshold(group.type, group.items))) { + rendered.push(item); + continue; + } + rendered.push({ ...item, projectedGroups: item.projectedGroups.flatMap((group): ModelRoundItemGroup[] => + group.type !== 'critical' && !reachesFlowGroupThreshold(group.type, group.items) + ? group.items.map(member => ({ type: 'critical', item: member })) : [group]) }); + } + return rendered; } diff --git a/src/web-ui/src/flow_chat/timeline/document.test.ts b/src/web-ui/src/flow_chat/timeline/document.test.ts index e80866f1d7..a7f78eb609 100644 --- a/src/web-ui/src/flow_chat/timeline/document.test.ts +++ b/src/web-ui/src/flow_chat/timeline/document.test.ts @@ -4,6 +4,8 @@ import { FlowChatReaderState } from './readerState'; import type { FlowItem, FlowToolItem } from '../types/flow-chat'; import type { VirtualItem } from '../types/flow-chat-projection'; import { estimateVirtualItemHeight } from '../components/modern/virtualItemHeightEstimators'; +import { buildInlineToolGroupData } from '../grouping/roundGroups'; +import { toolCapsuleStateKey } from '../tool-cards/toolCapsuleModel'; const tool = (id: string, status: FlowToolItem['status'] = 'completed'): FlowToolItem => ({ id, type: 'tool', toolName: 'Read', timestamp: 1, status, @@ -129,6 +131,32 @@ describe('desktop conversation document', () => { release(); }); + it('opens a group when one member holds an expanded capsule key of this session', () => { + const projection = new ConversationDocumentProjection(); + const input = source([tool('a'), tool('b')]); + const open = (sessionId: string | undefined, keys: string[]) => projection + .project([input], { ...options(), sessionId, expandedToolCapsules: new Set(keys) }) + .some(row => row.timeline?.kind === 'group-members'); + expect(projection.project([input], { ...options(), sessionId: 'session' }) + .some(row => row.timeline?.kind === 'group-members')).toBe(false); + expect(open('session', [toolCapsuleStateKey('session', 'turn', 'b')])).toBe(true); + // A foreign session or an unreadable key never opens this transcript. + expect(open('session', [toolCapsuleStateKey('other', 'turn', 'b'), 'unreadable'])).toBe(false); + }); + + it('reuses unchanged member rows and replaces only the one that changed', () => { + const projection = new ConversationDocumentProjection(); + const items = [tool('a'), tool('b')]; + const groupId = buildInlineToolGroupData('round', { type: 'explore', items, isLast: true }).groupId; + const opts = () => ({ ...options(), groupStates: new Map([[groupId, true]]) }); + const members = (rows: VirtualItem[]) => rows.filter(row => row.timeline?.kind === 'group-members'); + const first = members(projection.project([source(items)], opts())); + expect(first).toHaveLength(2); + expect(members(projection.project([source(items)], opts())).map(row => first.includes(row))).toEqual([true, true]); + const changed = members(projection.project([source([items[0], { ...items[1], partialParams: { file_path: '/b' } }])], opts())); + expect(changed.map(row => first.includes(row))).toEqual([true, false]); + }); + it('applies remembered filters before materialization and supports legacy group sources', () => { const projection = new ConversationDocumentProjection(); const opts = options(); diff --git a/src/web-ui/src/flow_chat/timeline/document.ts b/src/web-ui/src/flow_chat/timeline/document.ts index bf20222373..d4a9904d64 100644 --- a/src/web-ui/src/flow_chat/timeline/document.ts +++ b/src/web-ui/src/flow_chat/timeline/document.ts @@ -4,7 +4,6 @@ import { buildInlineToolGroupData, getProjectedModelRoundGroups, hasModelRoundLe import { getFlowGroupDisclosureChoice, isFlowGroupExpanded, type FlowGroupData } from '../grouping/types'; import { filterFlowGroupItems, type FlowGroupStatusFilter } from '../grouping/browse'; import { isFlowItemVisible } from '../utils/flowItemVisibility'; -import { toolCapsuleStateKey } from '../tool-cards/toolCapsuleModel'; import type { FlowChatReaderState } from './readerState'; import { getVirtualItemStableKey } from '../components/modern/virtualItemIdentity'; import { getConcurrentCapsuleRows } from '../tool-cards/toolCapsuleLayout'; @@ -39,15 +38,43 @@ export function conversationContentRuns(items: readonly FlowItem[]): FlowItem[][ return runs; } -function readerExpanded(group: FlowGroupData, turnId: string, options: TimelineProjectionOptions): boolean { +/** Expanded capsule keys inverted for this session: turn id -> expanded member ids. */ +type ExpandedCapsuleIndex = Map>; + +/** + * Inverts the disclosure keys (`JSON [sessionId, turnId, itemId]`, as built by + * `toolCapsuleStateKey`) once per projection, so a member test is a lookup + * instead of a freshly serialized key per item. Most flushes expand nothing. + */ +function capsuleExpansionIndex(options: TimelineProjectionOptions): ExpandedCapsuleIndex | undefined { + const keys = options.expandedToolCapsules; + if (!keys?.size) return undefined; + const sessionId = options.sessionId ?? ''; + const index: ExpandedCapsuleIndex = new Map(); + keys.forEach(key => { + let parsed: unknown; + try { parsed = JSON.parse(key); } catch { return; } + if (!Array.isArray(parsed) || parsed.length < 3) return; + const [session, turn, item] = parsed; + if (session !== sessionId || typeof turn !== 'string' || typeof item !== 'string') return; + const members = index.get(turn); + if (members) members.add(item); else index.set(turn, new Set([item])); + }); + return index; +} + +function readerExpanded( + group: FlowGroupData, turnId: string, options: TimelineProjectionOptions, expandedCapsules?: ExpandedCapsuleIndex, +): boolean { const choice = getFlowGroupDisclosureChoice(group, options.groupStates); const permission = group.allItems.some(item => item.type === 'tool' && (item.status === 'pending_confirmation' || options.pendingPermissionToolCallIds?.has((item as FlowToolItem).toolCall.id))); if (permission) return true; if (choice !== undefined) return choice; + const expandedMembers = expandedCapsules?.get(turnId); return isFlowGroupExpanded(group, options.groupStates) || options.reader.isGroupHeld(group.groupId) - || group.allItems.some(item => options.expandedToolCapsules?.has(toolCapsuleStateKey(options.sessionId, turnId, item.id))); + || (expandedMembers !== undefined && group.allItems.some(item => expandedMembers.has(item.id))); } type RoundItem = Extract; @@ -108,6 +135,7 @@ export class ConversationDocumentProjection { this.next = new Map(); const result: VirtualItem[] = []; const viewStamp = this.viewStampFor(options); + const expandedCapsules = capsuleExpansionIndex(options); sources.forEach((original, sourceIndex) => { // Legacy whole-round groups obey the same residency contract. This is a // render adapter only; neither their recorded rounds nor ids are changed. @@ -143,21 +171,13 @@ export class ConversationDocumentProjection { const key = `${source.turnId}:${id}`; const previous = this.previous.get(key); const chrome = kind === 'round-header' || kind === 'round-footer'; - const data: ModelRound = chrome ? source.data : { - ...source.data, items: members as AnyFlowItem[], attempts: undefined, historyRounds: undefined, - renderHints: undefined, - isStreaming: members.some(member => member.status === 'streaming' || member.status === 'running' - || ('isStreaming' in member && member.isStreaming === true)), - }; - const row: RoundItem = { - ...source, data, - projectedGroups: members.map(item => ({ type: 'critical', item })), - isLastRound: (kind === 'round-footer' || (kind === 'content' && extra.last === true)) && source.isLastRound, - canvasArtifactItems: kind === 'round-footer' ? source.canvasArtifactItems : undefined, - timeline: { key, kind, sourceIndex, memberIds: members.map(item => item.id), - layout: isAgentCardRun(members) ? 'agent-cards' : undefined, ...extra }, - }; - const reused = previous?.type === 'model-round' && previous.timeline?.kind === kind + const isLastRound = (kind === 'round-footer' || (kind === 'content' && extra.last === true)) && source.isLastRound; + const canvasArtifactItems = kind === 'round-footer' ? source.canvasArtifactItems : undefined; + const layout = isAgentCardRun(members) ? 'agent-cards' : undefined; + // Judge reuse from cheap primitives first; an unchanged block keeps the + // previous row identity the virtualizer memoized, and now also keeps the + // flush from allocating row payloads it would immediately discard. + const reusable = previous?.type === 'model-round' && previous.timeline?.kind === kind && previous.timeline.sourceIndex === sourceIndex && (kind === 'group-members' ? previous.timeline.group?.groupId === extra.group?.groupId @@ -166,16 +186,33 @@ export class ConversationDocumentProjection { && previous.timeline.memberOrdinal === extra.memberOrdinal : previous.timeline.group === extra.group) && previous.timeline.expanded === extra.expanded - && previous.timeline.layout === row.timeline!.layout + && previous.timeline.layout === layout && previous.timeline.first === extra.first && previous.timeline.last === extra.last && previous.timeline.revealThinkingIds?.join('|') === extra.revealThinkingIds?.join('|') - && previous.isLastRound === row.isLastRound && previous.isTurnComplete === row.isTurnComplete - && previous.layoutHints?.expandedThinkingItemIds.join('|') === row.layoutHints?.expandedThinkingItemIds.join('|') - && (chrome ? previous.data === data && previous.canvasArtifactItems === row.canvasArtifactItems + && previous.isLastRound === isLastRound && previous.isTurnComplete === source.isTurnComplete + && previous.layoutHints?.expandedThinkingItemIds.join('|') === source.layoutHints?.expandedThinkingItemIds.join('|') + && (chrome ? previous.data === source.data && previous.canvasArtifactItems === canvasArtifactItems : previous.data.items.length === members.length && previous.data.items.every((item, index) => item === members[index])) - ? previous : row; - this.next.set(key, reused); - rows.push(reused); + ? previous : undefined; + if (reusable) { + this.next.set(key, reusable); + rows.push(reusable); + return; + } + const data: ModelRound = chrome ? source.data : { + ...source.data, items: members as AnyFlowItem[], attempts: undefined, historyRounds: undefined, + renderHints: undefined, + isStreaming: members.some(member => member.status === 'streaming' || member.status === 'running' + || ('isStreaming' in member && member.isStreaming === true)), + }; + const row: RoundItem = { + ...source, data, + projectedGroups: members.map(item => ({ type: 'critical', item })), + isLastRound, canvasArtifactItems, + timeline: { key, kind, sourceIndex, memberIds: members.map(item => item.id), layout, ...extra }, + }; + this.next.set(key, row); + rows.push(row); }; if (hasModelRoundLeadingControls(source.data)) add('round-header', `${source.data.id}:history`, emptyItems); const groups = getProjectedModelRoundGroups(source); @@ -189,7 +226,7 @@ export class ConversationDocumentProjection { if (group.type === 'critical') { if (isFlowItemVisible(group.item)) critical.push(group.item); return; } flush(); const data = buildInlineToolGroupData(source.data.id, group, source.isLastRound && group.isLast); - const expanded = readerExpanded(data, source.turnId, options); + const expanded = readerExpanded(data, source.turnId, options, expandedCapsules); add('group-header', `group:${data.groupId}`, emptyItems, { group: data, expanded }); if (!expanded) return; const prefix = `group:${data.groupId}:`;