Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 11 additions & 8 deletions src/web-ui/src/flow_chat/components/modern/VirtualMessageList.tsx
Original file line number Diff line number Diff line change
Expand Up @@ -524,6 +524,16 @@ const VirtualMessageListSession = forwardRef<VirtualMessageListRef, VirtualMessa
const interactionKeys = useTimelineInteraction(scrollerElementRef, reader, () => 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,
Expand All @@ -539,14 +549,7 @@ const VirtualMessageListSession = forwardRef<VirtualMessageListRef, VirtualMessa
isHistorical: activeSession?.isHistorical === true,
exploreGroupStates,
} satisfies VirtualItemHeightEstimateContext,
estimateContextRevision: [
viewportWidthPx,
activeSession?.isHistorical === true ? 'historical' : 'live',
[...(exploreGroupStates?.entries() ?? [])]
.sort(([left], [right]) => left.localeCompare(right))
.map(([groupId, expanded]) => `${groupId}:${expanded ? 1 : 0}`)
.join(','),
].join('|'),
estimateContextRevision,
isViewportSuspended: () => isViewportSuspendedRef.current,
measurementSuspended,
scrollPaddingStartPx: FLOWCHAT_TURN_TOP_GAP_PX,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -475,6 +475,51 @@ export function useFlowChatVirtualizer<T>({
const aimStartedAtMsRef = useRef<number | null>(null);
const getItemKeyRef = useRef(getItemKey);
getItemKeyRef.current = getItemKey;
const itemKeyCacheOwnerRef = useRef(getItemKey);
const itemKeyCacheRef = useRef(new WeakMap<object, string>());
/**
* 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<string[]>([]);
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);
Expand Down Expand Up @@ -508,7 +553,7 @@ export function useFlowChatVirtualizer<T>({
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
Expand Down Expand Up @@ -657,9 +702,10 @@ export function useFlowChatVirtualizer<T>({
}
const previous = previousKnownHeightsRef.current;
const next = new Map<string, number>();
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;
Expand Down Expand Up @@ -776,7 +822,7 @@ export function useFlowChatVirtualizer<T>({
/** 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
Expand Down
88 changes: 65 additions & 23 deletions src/web-ui/src/flow_chat/grouping/groupProjection.ts
Original file line number Diff line number Diff line change
@@ -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<typeof getFlowGroupCategory>, 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<string>): 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<string>();
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<ReturnType<typeof getFlowGroupCategory> | undefined>(
(joined, part) => joinedFlowGroupCategory(joined, getFlowGroupCategory(part)), undefined,
) ?? getFlowGroupCategory(first);
return flowGroupPolicies[category].create({
const mergedIds = new Set<string>();
const rounds: ModelRound[] = [];
const allItems: FlowItem[] = [];
let category: ReturnType<typeof getFlowGroupCategory> | 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,
});
Expand All @@ -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());
}
Expand Down Expand Up @@ -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;
}
13 changes: 10 additions & 3 deletions src/web-ui/src/flow_chat/services/EventBatcher.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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');
Expand Down Expand Up @@ -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();
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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);
Expand All @@ -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) {
Expand Down Expand Up @@ -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);
Expand All @@ -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) {
Expand Down
11 changes: 10 additions & 1 deletion src/web-ui/src/flow_chat/store/FlowChatStore.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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 => {
Expand Down
Loading
Loading