diff --git a/packages/app/src/components/agent-stream-view.tsx b/packages/app/src/components/agent-stream-view.tsx index 92b5504bf..d55ac4959 100644 --- a/packages/app/src/components/agent-stream-view.tsx +++ b/packages/app/src/components/agent-stream-view.tsx @@ -79,6 +79,30 @@ import { isWeb } from "@/constants/platform"; const isUserMessageItem = (item?: StreamItem) => item?.kind === "user_message"; const isToolSequenceItem = (item?: StreamItem) => item?.kind === "tool_call" || item?.kind === "thought" || item?.kind === "todo_list"; + +const isSameAssistantBlockGroup = (params: { + item: StreamItem | null | undefined; + other: StreamItem | null | undefined; +}) => + params.item?.kind === "assistant_message" && + params.other?.kind === "assistant_message" && + params.item.blockGroupId !== undefined && + params.item.blockGroupId === params.other.blockGroupId; + +const isAssistantBlockFragment = (params: { + item: StreamItem; + aboveItem: StreamItem | null | undefined; + belowItem: StreamItem | null | undefined; +}) => + params.item.kind === "assistant_message" && + (isSameAssistantBlockGroup({ + item: params.item, + other: params.aboveItem, + }) || + isSameAssistantBlockGroup({ + item: params.item, + other: params.belowItem, + })); export interface AgentStreamViewHandle { scrollToBottom(reason?: BottomAnchorLocalRequest["reason"]): void; prepareForViewportChange(): void; @@ -249,6 +273,7 @@ const AgentStreamViewComponent = forwardRef ); + } case "thought": { const nextItem = getStreamNeighborItem({ strategy: streamRenderStrategy, @@ -485,21 +531,27 @@ const AgentStreamViewComponent = forwardRef(() => { - const pendingPermissionsNode = + const pendingPermissionsNode = useMemo( + () => pendingPermissionItems.length > 0 ? ( {pendingPermissionItems.map((permission) => ( ))} - ) : null; - const workingIndicatorNode = showWorkingIndicator ? ( - - - - ) : null; - + ) : null, + [client, pendingPermissionItems], + ); + const workingIndicatorNode = useMemo( + () => + showWorkingIndicator ? ( + + + + ) : null, + [showWorkingIndicator], + ); + const renderModel = useMemo(() => { return { ...baseRenderModel, boundary: { @@ -514,7 +566,7 @@ const AgentStreamViewComponent = forwardRef { if ( diff --git a/packages/app/src/components/message.tsx b/packages/app/src/components/message.tsx index af7b923be..65012488d 100644 --- a/packages/app/src/components/message.tsx +++ b/packages/app/src/components/message.tsx @@ -433,6 +433,7 @@ interface AssistantMessageProps { serverId?: string; client?: DaemonClient | null; disableOuterSpacing?: boolean; + spacing?: "default" | "compact"; } export const assistantMessageStylesheet = StyleSheet.create((theme) => ({ @@ -440,6 +441,10 @@ export const assistantMessageStylesheet = StyleSheet.create((theme) => ({ paddingHorizontal: theme.spacing[2], paddingVertical: theme.spacing[3], }, + containerCompactVertical: { + paddingTop: 0, + paddingBottom: 0, + }, containerSpacing: { marginBottom: theme.spacing[4], }, @@ -952,6 +957,115 @@ const expandableBadgeStylesheet = StyleSheet.create((theme) => ({ }, })); +interface NativeExpandableBadgeShimmerProps { + label: string; + secondaryLabel?: string; + rowWidth: number; + rowHeight: number; + peakWidth: number; + durationSeconds: number; + gradientId: string; +} + +const NativeExpandableBadgeShimmer = memo(function NativeExpandableBadgeShimmer({ + label, + secondaryLabel, + rowWidth, + rowHeight, + peakWidth, + durationSeconds, + gradientId, +}: NativeExpandableBadgeShimmerProps) { + const shimmerTranslateX = useSharedValue(0); + + useEffect(() => { + const startPosition = -peakWidth; + const endPosition = rowWidth + peakWidth; + shimmerTranslateX.value = startPosition; + shimmerTranslateX.value = withRepeat( + withTiming(endPosition, { + duration: durationSeconds * 1000, + easing: Easing.linear, + }), + -1, + false, + ); + return () => { + cancelAnimation(shimmerTranslateX); + }; + }, [durationSeconds, peakWidth, rowWidth, shimmerTranslateX]); + + const nativeShimmerPeakStyle = useAnimatedStyle(() => ({ + transform: [{ translateX: shimmerTranslateX.value }], + })); + + const nativeShimmerTrackStyle = useMemo( + () => [expandableBadgeStylesheet.nativeShimmerTrack, { width: rowWidth, height: rowHeight }], + [rowHeight, rowWidth], + ); + + const nativeShimmerMaskStyle = useMemo( + () => [expandableBadgeStylesheet.shimmerMaskRow, { width: rowWidth, height: rowHeight }], + [rowHeight, rowWidth], + ); + + const nativeLabelMaskStyle = useMemo( + () => [expandableBadgeStylesheet.label, { color: "#000000", opacity: 1 }], + [], + ); + + const nativeSecondaryMaskStyle = useMemo( + () => [expandableBadgeStylesheet.secondaryLabel, { color: "#000000", opacity: 1 }], + [], + ); + + const nativeShimmerPeakCombinedStyle = useMemo( + () => [ + expandableBadgeStylesheet.nativeShimmerPeak, + nativeShimmerPeakStyle, + { width: peakWidth, height: rowHeight }, + ], + [nativeShimmerPeakStyle, peakWidth, rowHeight], + ); + + return ( + + + + {label} + + {secondaryLabel ? ( + + {secondaryLabel} + + ) : ( + + )} + + } + > + + + + + + + + + + + + + + + + + ); +}); + interface MemoizedMarkdownBlockProps { text: string; styles: ReturnType; @@ -989,9 +1103,12 @@ export const AssistantMessage = memo(function AssistantMessage({ serverId, client, disableOuterSpacing, + spacing = "default", }: AssistantMessageProps) { const { theme } = useUnistyles(); - const resolvedDisableOuterSpacing = useDisableOuterSpacing(disableOuterSpacing); + const resolvedDisableOuterSpacing = useDisableOuterSpacing( + disableOuterSpacing ?? spacing === "compact", + ); const markdownStyles = useMemo(() => createMarkdownStyles(theme), [theme]); @@ -1192,6 +1309,7 @@ export const AssistantMessage = memo(function AssistantMessage({ testID="assistant-message" style={[ assistantMessageStylesheet.container, + spacing === "compact" && assistantMessageStylesheet.containerCompactVertical, !resolvedDisableOuterSpacing && assistantMessageStylesheet.containerSpacing, ]} > @@ -1670,7 +1788,6 @@ const ExpandableBadge = memo(function ExpandableBadge({ const [labelWidth, setLabelWidth] = useState(0); const [secondaryOffsetX, setSecondaryOffsetX] = useState(0); const [secondaryWidth, setSecondaryWidth] = useState(0); - const shimmerTranslateX = useSharedValue(0); const totalShimmerChars = label.trim().length + (secondaryLabel?.trim().length ?? 0); const shortTextDurationAdjustment = totalShimmerChars <= 12 ? 0.25 : 0; @@ -1738,28 +1855,6 @@ const ExpandableBadge = memo(function ExpandableBadge({ ensureWebToolCallShimmerKeyframes(); }, [isWebShimmer]); - useEffect(() => { - if (!isNativeShimmer) { - cancelAnimation(shimmerTranslateX); - shimmerTranslateX.value = -nativeShimmerPeakWidth; - return; - } - const startPosition = -nativeShimmerPeakWidth; - const endPosition = labelRowWidth + nativeShimmerPeakWidth; - shimmerTranslateX.value = startPosition; - shimmerTranslateX.value = withRepeat( - withTiming(endPosition, { - duration: shimmerDuration * 1000, - easing: Easing.linear, - }), - -1, - false, - ); - return () => { - cancelAnimation(shimmerTranslateX); - }; - }, [isNativeShimmer, labelRowWidth, nativeShimmerPeakWidth, shimmerDuration, shimmerTranslateX]); - useEffect(() => { if (isNative || !isExpanded || !hasDetailContent) { return; @@ -1784,10 +1879,6 @@ const ExpandableBadge = memo(function ExpandableBadge({ }; }, [hasDetailContent, isExpanded, wheelInvestigationComponentId]); - const nativeShimmerPeakStyle = useAnimatedStyle(() => ({ - transform: [{ translateX: shimmerTranslateX.value }], - })); - const shimmerGradient = "linear-gradient(90deg, rgba(255, 255, 255, 0) 0%, rgba(255, 255, 255, 0.45) 24%, #ffffff 40%, #ffffff 60%, rgba(255, 255, 255, 0.45) 76%, rgba(255, 255, 255, 0) 100%)"; @@ -1887,41 +1978,6 @@ const ExpandableBadge = memo(function ExpandableBadge({ [shimmerSecondaryStyle], ); - const nativeShimmerTrackStyle = useMemo( - () => [ - expandableBadgeStylesheet.nativeShimmerTrack, - { width: labelRowWidth, height: labelRowHeight }, - ], - [labelRowHeight, labelRowWidth], - ); - - const nativeShimmerMaskStyle = useMemo( - () => [ - expandableBadgeStylesheet.shimmerMaskRow, - { width: labelRowWidth, height: labelRowHeight }, - ], - [labelRowHeight, labelRowWidth], - ); - - const nativeLabelMaskStyle = useMemo( - () => [expandableBadgeStylesheet.label, { color: "#000000", opacity: 1 }], - [], - ); - - const nativeSecondaryMaskStyle = useMemo( - () => [expandableBadgeStylesheet.secondaryLabel, { color: "#000000", opacity: 1 }], - [], - ); - - const nativeShimmerPeakCombinedStyle = useMemo( - () => [ - expandableBadgeStylesheet.nativeShimmerPeak, - nativeShimmerPeakStyle, - { width: nativeShimmerPeakWidth, height: labelRowHeight }, - ], - [labelRowHeight, nativeShimmerPeakStyle, nativeShimmerPeakWidth], - ); - const chevronStyle = useMemo( () => [ expandableBadgeStylesheet.chevron, @@ -2003,52 +2059,15 @@ const ExpandableBadge = memo(function ExpandableBadge({ ) : null} {isNativeShimmer ? ( - - - - {label} - - {secondaryLabel ? ( - - {secondaryLabel} - - ) : ( - - )} - - } - > - - - - - - - - - - - - - - - - + ) : null} {isInteractive && isHovered ? ( diff --git a/packages/app/src/components/stream-strategy-native.tsx b/packages/app/src/components/stream-strategy-native.tsx index 97664cd39..afb651310 100644 --- a/packages/app/src/components/stream-strategy-native.tsx +++ b/packages/app/src/components/stream-strategy-native.tsx @@ -8,6 +8,7 @@ import { type NativeSyntheticEvent, } from "react-native"; import type { StreamItem } from "@/types/stream"; +import { useStableEvent } from "@/hooks/use-stable-event"; import { useBottomAnchorController } from "./use-bottom-anchor-controller"; import type { StreamRenderInput, StreamStrategy, StreamViewportHandle } from "./stream-strategy"; import { @@ -37,6 +38,7 @@ function NativeStreamViewport(props: StreamRenderInput & { strategy: StreamStrat baseListContentContainerStyle, strategy, } = props; + const { renderHistoryMountedRow, renderLiveHeadRow, renderLiveAuxiliary } = renderers; const flatListRef = useRef>(null); const streamViewportMetricsRef = useRef({ containerKey: "native-virtualized", @@ -190,105 +192,91 @@ function NativeStreamViewport(props: StreamRenderInput & { strategy: StreamStrat }; }, [agentId, bottomAnchorController, markNativeViewportSettling, viewportRef]); - const handleScroll = useCallback( - (event: NativeSyntheticEvent) => { - const { contentOffset, contentSize, layoutMeasurement } = event.nativeEvent; - const previousOffsetY = scrollOffsetYRef.current; - scrollOffsetYRef.current = contentOffset.y; - streamViewportMetricsRef.current = { - contentHeight: Math.max(0, contentSize.height), - viewportWidth: Math.max(0, layoutMeasurement.width), - viewportHeight: Math.max(0, layoutMeasurement.height), - containerKey: "native-virtualized", - offsetY: contentOffset.y, - viewportMeasuredForKey: "native-virtualized", - contentMeasuredForKey: "native-virtualized", - }; + const handleScroll = useStableEvent((event: NativeSyntheticEvent) => { + const { contentOffset, contentSize, layoutMeasurement } = event.nativeEvent; + const previousOffsetY = scrollOffsetYRef.current; + scrollOffsetYRef.current = contentOffset.y; + streamViewportMetricsRef.current = { + contentHeight: Math.max(0, contentSize.height), + viewportWidth: Math.max(0, layoutMeasurement.width), + viewportHeight: Math.max(0, layoutMeasurement.height), + containerKey: "native-virtualized", + offsetY: contentOffset.y, + viewportMeasuredForKey: "native-virtualized", + contentMeasuredForKey: "native-virtualized", + }; - const nearBottom = isNearBottomForStreamRenderStrategy({ - strategy, - offsetY: contentOffset.y, - threshold: 32, - contentHeight: streamViewportMetricsRef.current.contentHeight, - viewportHeight: streamViewportMetricsRef.current.viewportHeight, + const nearBottom = isNearBottomForStreamRenderStrategy({ + strategy, + offsetY: contentOffset.y, + threshold: 32, + contentHeight: streamViewportMetricsRef.current.contentHeight, + viewportHeight: streamViewportMetricsRef.current.viewportHeight, + }); + onNearBottomChange(nearBottom); + + if (programmaticScrollEventBudgetRef.current > 0 && contentOffset.y <= 8) { + programmaticScrollEventBudgetRef.current -= 1; + } else { + programmaticScrollEventBudgetRef.current = 0; + bottomAnchorController.handleScrollNearBottomChange({ + nextIsNearBottom: nearBottom, + scrollDelta: contentOffset.y - previousOffsetY, }); - onNearBottomChange(nearBottom); + } + }); - if (programmaticScrollEventBudgetRef.current > 0 && contentOffset.y <= 8) { - programmaticScrollEventBudgetRef.current -= 1; - } else { - programmaticScrollEventBudgetRef.current = 0; - bottomAnchorController.handleScrollNearBottomChange({ - nextIsNearBottom: nearBottom, - scrollDelta: contentOffset.y - previousOffsetY, - }); - } - }, - [bottomAnchorController, onNearBottomChange, strategy], - ); + const handleListLayout = useStableEvent((event: LayoutChangeEvent) => { + const previousViewportWidth = streamViewportMetricsRef.current.viewportWidth; + const previousViewportHeight = streamViewportMetricsRef.current.viewportHeight; + const viewportWidth = Math.max(0, event.nativeEvent.layout.width); + const viewportHeight = Math.max(0, event.nativeEvent.layout.height); + const viewportChanged = + (previousViewportWidth > 0 && previousViewportWidth !== viewportWidth) || + (previousViewportHeight > 0 && previousViewportHeight !== viewportHeight); + streamViewportMetricsRef.current = { + ...streamViewportMetricsRef.current, + containerKey: "native-virtualized", + viewportWidth, + viewportHeight, + viewportMeasuredForKey: "native-virtualized", + }; + if (viewportChanged) { + markNativeViewportSettling(); + } + bottomAnchorController.handleViewportMetricsChange({ + previousViewportWidth, + viewportWidth, + previousViewportHeight, + viewportHeight, + }); + }); - const handleListLayout = useCallback( - (event: LayoutChangeEvent) => { - const previousViewportWidth = streamViewportMetricsRef.current.viewportWidth; - const previousViewportHeight = streamViewportMetricsRef.current.viewportHeight; - const viewportWidth = Math.max(0, event.nativeEvent.layout.width); - const viewportHeight = Math.max(0, event.nativeEvent.layout.height); - const viewportChanged = - (previousViewportWidth > 0 && previousViewportWidth !== viewportWidth) || - (previousViewportHeight > 0 && previousViewportHeight !== viewportHeight); - streamViewportMetricsRef.current = { - ...streamViewportMetricsRef.current, - containerKey: "native-virtualized", - viewportWidth, - viewportHeight, - viewportMeasuredForKey: "native-virtualized", - }; - if (viewportChanged) { - markNativeViewportSettling(); - } - bottomAnchorController.handleViewportMetricsChange({ - previousViewportWidth, - viewportWidth, - previousViewportHeight, - viewportHeight, - }); - }, - [bottomAnchorController, markNativeViewportSettling], - ); + const handleContentSizeChange = useStableEvent((_width: number, height: number) => { + const previousContentHeight = streamViewportMetricsRef.current.contentHeight; + const nextContentHeight = Math.max(0, height); + streamViewportMetricsRef.current = { + ...streamViewportMetricsRef.current, + containerKey: "native-virtualized", + contentHeight: nextContentHeight, + contentMeasuredForKey: "native-virtualized", + }; + bottomAnchorController.handleContentSizeChange({ + previousContentHeight, + contentHeight: nextContentHeight, + }); + }); - const handleContentSizeChange = useCallback( - (_width: number, height: number) => { - const previousContentHeight = streamViewportMetricsRef.current.contentHeight; - const nextContentHeight = Math.max(0, height); - streamViewportMetricsRef.current = { - ...streamViewportMetricsRef.current, - containerKey: "native-virtualized", - contentHeight: nextContentHeight, - contentMeasuredForKey: "native-virtualized", - }; - bottomAnchorController.handleContentSizeChange({ - previousContentHeight, - contentHeight: nextContentHeight, - }); - }, - [bottomAnchorController], - ); - - const renderItem = useCallback( - ({ item, index }: ListRenderItemInfo) => { - const rendered = renderers.renderHistoryMountedRow(item, index, historyRows); - return rendered ? {rendered} : null; - }, - [historyRows, renderers], - ); + const renderItem = useStableEvent(({ item, index }: ListRenderItemInfo) => { + const rendered = renderHistoryMountedRow(item, index, historyRows); + return rendered ? {rendered} : null; + }); const liveHeaderContent = useMemo(() => { const liveHeadRows = segments.liveHead.map((item, index) => ( - - {renderers.renderLiveHeadRow(item, index, segments.liveHead)} - + {renderLiveHeadRow(item, index, segments.liveHead)} )); - const liveAuxiliary = renderers.renderLiveAuxiliary(); + const liveAuxiliary = renderLiveAuxiliary(); if ( liveHeadRows.length === 0 && !liveAuxiliary && @@ -303,7 +291,7 @@ function NativeStreamViewport(props: StreamRenderInput & { strategy: StreamStrat {liveAuxiliary} ); - }, [boundary, listEmptyComponent, renderers, segments.liveHead]); + }, [boundary, listEmptyComponent, renderLiveAuxiliary, renderLiveHeadRow, segments.liveHead]); return ( { if (message.type !== "agent_stream") return; const { agentId, event, timestamp, seq, epoch } = message.payload; @@ -1075,96 +1083,13 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider } } - // Read current store state - const session = useSessionStore.getState().sessions[serverId]; - const currentTail = session?.agentStreamTail.get(agentId) ?? []; - const currentHead = session?.agentStreamHead.get(agentId) ?? []; - const currentCursor = session?.agentTimelineCursor.get(agentId); - const currentAgentEntry = session?.agents.get(agentId); - const currentAgent = currentAgentEntry - ? { - status: currentAgentEntry.status, - updatedAt: currentAgentEntry.updatedAt, - lastActivityAt: currentAgentEntry.lastActivityAt, - } - : null; - - // Call pure reducer - const result = processAgentStreamEvent({ + agentStreamReducerQueue.enqueue(agentId, { event: streamEvent, seq, epoch, - currentTail, - currentHead, - currentCursor, - currentAgent, timestamp: parsedTimestamp, }); - // Apply tail/head patches - if (result.changedTail || result.changedHead) { - setAgentStreamState(serverId, agentId, { - ...(result.changedTail ? { tail: result.tail } : {}), - ...(result.changedHead ? { head: result.head } : {}), - }); - } - - // Apply cursor patch - if (result.cursorChanged && result.cursor) { - const nextCursor = result.cursor; - setAgentTimelineCursor(serverId, (prev) => { - const current = prev.get(agentId); - if ( - current && - typeof seq === "number" && - typeof epoch === "string" && - current.epoch === epoch && - seq >= current.startSeq && - seq <= current.endSeq - ) { - // Fast-path: seq stays inside the current range during streaming. - return prev; - } - if ( - current && - current.epoch === nextCursor.epoch && - current.startSeq === nextCursor.startSeq && - current.endSeq === nextCursor.endSeq - ) { - return prev; - } - const next = new Map(prev); - next.set(agentId, nextCursor); - return next; - }); - } - - // Apply agent patch (optimistic lifecycle) - if (result.agentChanged && result.agent) { - const nextAgent = result.agent; - setAgents(serverId, (prev) => { - const current = prev.get(agentId); - if (!current) { - return prev; - } - const next = new Map(prev); - next.set(agentId, { - ...current, - status: nextAgent.status, - updatedAt: nextAgent.updatedAt, - lastActivityAt: nextAgent.lastActivityAt, - }); - return next; - }); - } - - // Execute side effects - for (const effect of result.sideEffects) { - if (effect.type === "catch_up") { - requestCanonicalCatchUp(agentId, effect.cursor); - } - } - // NOTE: We don't update lastActivityAt on every stream event to prevent // cascading rerenders. The agent_update handler updates agent.lastActivityAt // on status changes, which is sufficient for sorting and display purposes. @@ -1172,6 +1097,7 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider const unsubAgentTimeline = client.on("fetch_agent_timeline_response", (message) => { if (message.type !== "fetch_agent_timeline_response") return; + agentStreamReducerQueue.flushAgent(message.payload.agentId); applyTimelineResponse(message.payload); }); @@ -1572,6 +1498,7 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider unsubVoiceInputState(); unsubAgentDeleted(); unsubAgentArchived(); + agentStreamReducerQueue.dispose({ flush: true }); }; }, [ client, diff --git a/packages/app/src/contexts/session-stream-reducers.test.ts b/packages/app/src/contexts/session-stream-reducers.test.ts index 580189da3..ba78f1301 100644 --- a/packages/app/src/contexts/session-stream-reducers.test.ts +++ b/packages/app/src/contexts/session-stream-reducers.test.ts @@ -2,10 +2,13 @@ import { describe, expect, it } from "vitest"; import type { AgentStreamEventPayload } from "@server/shared/messages"; import type { StreamItem } from "@/types/stream"; import { + createAgentStreamReducerQueue, processTimelineResponse, processAgentStreamEvent, + processAgentStreamEvents, type ProcessTimelineResponseInput, type ProcessAgentStreamEventInput, + type AgentStreamReducerEvent, type TimelineCursor, } from "./session-stream-reducers"; @@ -41,6 +44,36 @@ function makeUserTimelineEvent(text: string): AgentStreamEventPayload { } as AgentStreamEventPayload; } +function makeToolCallTimelineEvent(callId: string): AgentStreamEventPayload { + return { + type: "timeline", + provider: "claude", + item: { + type: "tool_call", + callId, + name: "Read", + status: "running", + detail: { + type: "read", + filePath: "/tmp/example.ts", + }, + error: null, + }, + } as AgentStreamEventPayload; +} + +function makeStreamReducerEvent( + event: AgentStreamEventPayload, + seq: number, +): AgentStreamReducerEvent { + return { + event, + seq, + epoch: "epoch-1", + timestamp: new Date(1000 + seq), + }; +} + const baseTimelineInput: ProcessTimelineResponseInput = { payload: { agentId: "agent-1", @@ -647,3 +680,270 @@ describe("processAgentStreamEvent", () => { expect(result.agent).toBe(null); }); }); + +describe("processAgentStreamEvents", () => { + it("coalesces contiguous assistant stream events into one head update and final cursor", () => { + const result = processAgentStreamEvents({ + events: [ + makeStreamReducerEvent(makeTimelineEvent("Hello"), 1), + makeStreamReducerEvent(makeTimelineEvent(" world"), 2), + ], + currentTail: [], + currentHead: [], + currentCursor: undefined, + currentAgent: null, + }); + + expect(result.changedTail).toBe(false); + expect(result.changedHead).toBe(true); + expect(result.tail).toEqual([]); + expect(result.head).toHaveLength(1); + expect(result.head[0]).toMatchObject({ + kind: "assistant_message", + text: "Hello world", + }); + expect(result.cursor).toEqual({ + epoch: "epoch-1", + startSeq: 1, + endSeq: 2, + } satisfies TimelineCursor); + expect(result.sideEffects).toEqual([]); + }); + + it("promotes completed assistant markdown blocks to tail while keeping the live block in head", () => { + const result = processAgentStreamEvents({ + events: [ + makeStreamReducerEvent(makeTimelineEvent("First paragraph"), 1), + makeStreamReducerEvent(makeTimelineEvent("\n\nSecond paragraph"), 2), + ], + currentTail: [], + currentHead: [], + currentCursor: undefined, + currentAgent: null, + }); + + expect(result.changedTail).toBe(true); + expect(result.changedHead).toBe(true); + expect(result.tail).toHaveLength(1); + expect(result.tail[0]).toMatchObject({ + kind: "assistant_message", + text: "First paragraph", + }); + expect(result.head).toHaveLength(1); + expect(result.head[0]).toMatchObject({ + kind: "assistant_message", + text: "Second paragraph", + }); + }); + + it("does not promote a markdown block that is still inside an open code fence", () => { + const result = processAgentStreamEvents({ + events: [ + makeStreamReducerEvent(makeTimelineEvent("Before fence\n\n```ts\nconst a = 1;"), 1), + makeStreamReducerEvent(makeTimelineEvent("\n\nconst b = 2;"), 2), + ], + currentTail: [], + currentHead: [], + currentCursor: undefined, + currentAgent: null, + }); + + expect(result.changedTail).toBe(true); + expect(result.tail).toHaveLength(1); + expect(result.tail[0]).toMatchObject({ + kind: "assistant_message", + text: "Before fence", + }); + expect(result.head).toHaveLength(1); + expect(result.head[0]).toMatchObject({ + kind: "assistant_message", + text: "```ts\nconst a = 1;\n\nconst b = 2;", + }); + }); + + it("flushes the live assistant block before applying a tool call in the same reducer pass", () => { + const result = processAgentStreamEvents({ + events: [ + makeStreamReducerEvent(makeTimelineEvent("Before tool"), 1), + makeStreamReducerEvent(makeToolCallTimelineEvent("call-1"), 2), + ], + currentTail: [], + currentHead: [], + currentCursor: undefined, + currentAgent: null, + }); + + expect(result.changedTail).toBe(true); + expect(result.changedHead).toBe(true); + expect(result.head).toEqual([]); + expect(result.tail.map((item: StreamItem) => item.kind)).toEqual([ + "assistant_message", + "tool_call", + ]); + expect(result.cursor).toEqual({ + epoch: "epoch-1", + startSeq: 1, + endSeq: 2, + } satisfies TimelineCursor); + }); + + it("returns the final optimistic lifecycle patch across a batch", () => { + const result = processAgentStreamEvents({ + events: [ + makeStreamReducerEvent(makeTimelineEvent("Done"), 1), + { + event: { type: "turn_completed", provider: "claude" } as AgentStreamEventPayload, + seq: undefined, + epoch: undefined, + timestamp: new Date(3000), + }, + ], + currentTail: [], + currentHead: [], + currentCursor: undefined, + currentAgent: { + status: "running", + updatedAt: new Date(1000), + lastActivityAt: new Date(1000), + }, + }); + + expect(result.head).toEqual([]); + expect(result.tail).toHaveLength(1); + expect(result.agentChanged).toBe(true); + expect(result.agent).toMatchObject({ + status: "idle", + updatedAt: new Date(3000), + lastActivityAt: new Date(3000), + }); + }); +}); + +describe("createAgentStreamReducerQueue", () => { + function createManualScheduler() { + let nextId = 1; + const callbacks = new Map void>(); + return { + schedule(callback: () => void) { + const id = nextId; + nextId += 1; + callbacks.set(id, callback); + return id; + }, + cancel(id: number) { + callbacks.delete(id); + }, + flushOne() { + const entry = callbacks.entries().next().value; + if (!entry) { + throw new Error("Expected a scheduled callback"); + } + const [id, callback] = entry; + callbacks.delete(id); + callback(); + }, + get size() { + return callbacks.size; + }, + }; + } + + it("coalesces multiple events for one agent into one scheduled commit", () => { + const scheduler = createManualScheduler(); + const commits: Array<{ agentId: string; headText: string; cursorEndSeq: number | null }> = []; + let currentTail: StreamItem[] = []; + let currentHead: StreamItem[] = []; + + const queue = createAgentStreamReducerQueue({ + getSnapshot: () => ({ + currentTail, + currentHead, + currentCursor: undefined, + currentAgent: null, + }), + commit: (agentId, result) => { + currentTail = result.tail; + currentHead = result.head; + commits.push({ + agentId, + headText: result.head[0]?.kind === "assistant_message" ? result.head[0].text : "", + cursorEndSeq: result.cursor?.endSeq ?? null, + }); + }, + handleSideEffects: () => {}, + scheduleFlush: scheduler.schedule, + cancelFlush: scheduler.cancel, + }); + + queue.enqueue("agent-1", makeStreamReducerEvent(makeTimelineEvent("Hello"), 1)); + queue.enqueue("agent-1", makeStreamReducerEvent(makeTimelineEvent(" world"), 2)); + + expect(scheduler.size).toBe(1); + expect(commits).toEqual([]); + + scheduler.flushOne(); + + expect(commits).toEqual([ + { + agentId: "agent-1", + headText: "Hello world", + cursorEndSeq: 2, + }, + ]); + expect(scheduler.size).toBe(0); + }); + + it("flushes queued events synchronously for one agent before canonical history is applied", () => { + const scheduler = createManualScheduler(); + const commits: string[] = []; + const queue = createAgentStreamReducerQueue({ + getSnapshot: () => ({ + currentTail: [], + currentHead: [], + currentCursor: undefined, + currentAgent: null, + }), + commit: (agentId, result) => { + commits.push( + `${agentId}:${result.head[0]?.kind === "assistant_message" ? result.head[0].text : ""}`, + ); + }, + handleSideEffects: () => {}, + scheduleFlush: scheduler.schedule, + cancelFlush: scheduler.cancel, + }); + + queue.enqueue("agent-1", makeStreamReducerEvent(makeTimelineEvent("queued"), 1)); + queue.flushAgent("agent-1"); + + expect(commits).toEqual(["agent-1:queued"]); + expect(scheduler.size).toBe(0); + }); + + it("flushes queued events synchronously before disposal", () => { + const scheduler = createManualScheduler(); + const commits: string[] = []; + const queue = createAgentStreamReducerQueue({ + getSnapshot: () => ({ + currentTail: [], + currentHead: [], + currentCursor: undefined, + currentAgent: null, + }), + commit: (agentId, result) => { + commits.push( + `${agentId}:${result.head[0]?.kind === "assistant_message" ? result.head[0].text : ""}`, + ); + }, + handleSideEffects: () => {}, + scheduleFlush: scheduler.schedule, + cancelFlush: scheduler.cancel, + }); + + queue.enqueue("agent-1", makeStreamReducerEvent(makeTimelineEvent("queued"), 1)); + queue.dispose({ flush: true }); + + expect(commits).toEqual(["agent-1:queued"]); + expect(scheduler.size).toBe(0); + }); +}); diff --git a/packages/app/src/contexts/session-stream-reducers.ts b/packages/app/src/contexts/session-stream-reducers.ts index c7fb61e51..ab31c8b80 100644 --- a/packages/app/src/contexts/session-stream-reducers.ts +++ b/packages/app/src/contexts/session-stream-reducers.ts @@ -1,5 +1,7 @@ import type { AgentStreamEventPayload } from "@server/shared/messages"; import type { AgentLifecycleStatus } from "@server/shared/agent-lifecycle"; +import type { Agent } from "@/stores/session-store"; +import { useSessionStore } from "@/stores/session-store"; import type { StreamItem } from "@/types/stream"; import { applyStreamEvent, @@ -17,6 +19,8 @@ import { } from "@/contexts/session-timeline-bootstrap-policy"; import { deriveOptimisticLifecycleStatus } from "@/contexts/session-stream-lifecycle"; +const AGENT_STREAM_REDUCER_FLUSH_DELAY_MS = 16 * 3; + // --------------------------------------------------------------------------- // Shared cursor type // --------------------------------------------------------------------------- @@ -324,6 +328,62 @@ export interface ProcessAgentStreamEventOutput { sideEffects: AgentStreamReducerSideEffect[]; } +export type AgentStreamReducerEvent = { + event: AgentStreamEventPayload; + seq: number | undefined; + epoch: string | undefined; + timestamp: Date; +}; + +export type AgentStreamReducerAgentSnapshot = { + status: AgentLifecycleStatus; + updatedAt: Date; + lastActivityAt: Date; +}; + +export type ProcessAgentStreamEventsInput = { + events: AgentStreamReducerEvent[]; + currentTail: StreamItem[]; + currentHead: StreamItem[]; + currentCursor: TimelineCursor | undefined; + currentAgent: AgentStreamReducerAgentSnapshot | null; +}; + +export type AgentStreamReducerSnapshot = Omit; + +export type AgentStreamReducerQueue = { + enqueue: (agentId: string, event: AgentStreamReducerEvent) => void; + flush: () => void; + flushAgent: (agentId: string) => void; + dispose: (options?: { flush?: boolean }) => void; +}; + +export type CreateAgentStreamReducerQueueInput = { + getSnapshot: (agentId: string) => AgentStreamReducerSnapshot; + commit: ( + agentId: string, + result: ProcessAgentStreamEventOutput, + events: AgentStreamReducerEvent[], + ) => void; + handleSideEffects: (agentId: string, sideEffects: AgentStreamReducerSideEffect[]) => void; + scheduleFlush: (callback: () => void) => number; + cancelFlush: (id: number) => void; +}; + +function applyAgentPatch( + currentAgent: AgentStreamReducerAgentSnapshot | null, + patch: AgentPatch | null, +): AgentStreamReducerAgentSnapshot | null { + if (!currentAgent || !patch) { + return currentAgent; + } + return { + status: patch.status, + updatedAt: patch.updatedAt, + lastActivityAt: patch.lastActivityAt, + }; +} + export function processAgentStreamEvent( input: ProcessAgentStreamEventInput, ): ProcessAgentStreamEventOutput { @@ -430,3 +490,255 @@ export function processAgentStreamEvent( sideEffects, }; } + +export function processAgentStreamEvents( + input: ProcessAgentStreamEventsInput, +): ProcessAgentStreamEventOutput { + let tail = input.currentTail; + let head = input.currentHead; + let cursor = input.currentCursor; + let agent = input.currentAgent; + let changedTail = false; + let changedHead = false; + let cursorChanged = false; + let agentPatch: AgentPatch | null = null; + let agentChanged = false; + const sideEffects: AgentStreamReducerSideEffect[] = []; + + for (const reducerEvent of input.events) { + const result = processAgentStreamEvent({ + event: reducerEvent.event, + seq: reducerEvent.seq, + epoch: reducerEvent.epoch, + currentTail: tail, + currentHead: head, + currentCursor: cursor, + currentAgent: agent, + timestamp: reducerEvent.timestamp, + }); + + tail = result.tail; + head = result.head; + changedTail = changedTail || result.changedTail; + changedHead = changedHead || result.changedHead; + sideEffects.push(...result.sideEffects); + + if (result.cursorChanged) { + cursor = result.cursor ?? undefined; + cursorChanged = true; + } + + if (result.agentChanged) { + agentPatch = result.agent; + agentChanged = true; + agent = applyAgentPatch(agent, result.agent); + } + } + + return { + tail, + head, + changedTail, + changedHead, + cursor: cursor ?? null, + cursorChanged, + agent: agentPatch, + agentChanged, + sideEffects, + }; +} + +export function createAgentStreamReducerQueue( + input: CreateAgentStreamReducerQueueInput, +): AgentStreamReducerQueue { + const pendingByAgentId = new Map(); + let scheduledFlushId: number | null = null; + + const cancelScheduledFlush = () => { + if (scheduledFlushId === null) { + return; + } + input.cancelFlush(scheduledFlushId); + scheduledFlushId = null; + }; + + const flushAgent = (agentId: string) => { + const events = pendingByAgentId.get(agentId); + if (!events || events.length === 0) { + return; + } + pendingByAgentId.delete(agentId); + if (pendingByAgentId.size === 0) { + cancelScheduledFlush(); + } + + const result = processAgentStreamEvents({ + events, + ...input.getSnapshot(agentId), + }); + + input.commit(agentId, result, events); + if (result.sideEffects.length > 0) { + input.handleSideEffects(agentId, result.sideEffects); + } + }; + + const flush = () => { + const agentIds = Array.from(pendingByAgentId.keys()); + for (const agentId of agentIds) { + flushAgent(agentId); + } + }; + + const scheduleFlush = () => { + if (scheduledFlushId !== null) { + return; + } + scheduledFlushId = input.scheduleFlush(() => { + scheduledFlushId = null; + flush(); + }); + }; + + return { + enqueue(agentId, event) { + const pending = pendingByAgentId.get(agentId); + if (pending) { + pending.push(event); + } else { + pendingByAgentId.set(agentId, [event]); + } + scheduleFlush(); + }, + flush, + flushAgent, + dispose(options) { + cancelScheduledFlush(); + if (options?.flush) { + flush(); + } else { + pendingByAgentId.clear(); + } + }, + }; +} + +type StreamStatePatch = { + tail?: StreamItem[]; + head?: StreamItem[]; +}; + +export type CreateSessionAgentStreamReducerQueueInput = { + serverId: string; + setAgentStreamState: (serverId: string, agentId: string, state: StreamStatePatch) => void; + setAgentTimelineCursor: ( + serverId: string, + state: (prev: Map) => Map, + ) => void; + setAgents: (serverId: string, state: (prev: Map) => Map) => void; + requestCanonicalCatchUp: (agentId: string, cursor: { epoch: string; endSeq: number }) => void; +}; + +function scheduleAgentStreamReducerFlush(callback: () => void): number { + return setTimeout(callback, AGENT_STREAM_REDUCER_FLUSH_DELAY_MS) as unknown as number; +} + +function cancelAgentStreamReducerFlush(id: number) { + clearTimeout(id); +} + +export function createSessionAgentStreamReducerQueue( + input: CreateSessionAgentStreamReducerQueueInput, +): AgentStreamReducerQueue { + const { + serverId, + setAgentStreamState, + setAgentTimelineCursor, + setAgents, + requestCanonicalCatchUp, + } = input; + + return createAgentStreamReducerQueue({ + getSnapshot: (agentId) => { + const session = useSessionStore.getState().sessions[serverId]; + const currentAgentEntry = session?.agents.get(agentId); + return { + currentTail: session?.agentStreamTail.get(agentId) ?? [], + currentHead: session?.agentStreamHead.get(agentId) ?? [], + currentCursor: session?.agentTimelineCursor.get(agentId), + currentAgent: currentAgentEntry + ? { + status: currentAgentEntry.status, + updatedAt: currentAgentEntry.updatedAt, + lastActivityAt: currentAgentEntry.lastActivityAt, + } + : null, + }; + }, + commit: (agentId, result, events) => { + if (result.changedTail || result.changedHead) { + setAgentStreamState(serverId, agentId, { + ...(result.changedTail ? { tail: result.tail } : {}), + ...(result.changedHead ? { head: result.head } : {}), + }); + } + + if (result.cursorChanged && result.cursor) { + const nextCursor = result.cursor; + const lastEvent = events.at(-1); + setAgentTimelineCursor(serverId, (prev) => { + const current = prev.get(agentId); + if ( + current && + lastEvent && + typeof lastEvent.seq === "number" && + typeof lastEvent.epoch === "string" && + current.epoch === lastEvent.epoch && + lastEvent.seq >= current.startSeq && + lastEvent.seq <= current.endSeq + ) { + return prev; + } + if ( + current && + current.epoch === nextCursor.epoch && + current.startSeq === nextCursor.startSeq && + current.endSeq === nextCursor.endSeq + ) { + return prev; + } + const next = new Map(prev); + next.set(agentId, nextCursor); + return next; + }); + } + + if (result.agentChanged && result.agent) { + const nextAgent = result.agent; + setAgents(serverId, (prev) => { + const current = prev.get(agentId); + if (!current) { + return prev; + } + const next = new Map(prev); + next.set(agentId, { + ...current, + status: nextAgent.status, + updatedAt: nextAgent.updatedAt, + lastActivityAt: nextAgent.lastActivityAt, + }); + return next; + }); + } + }, + handleSideEffects: (agentId, sideEffects) => { + for (const effect of sideEffects) { + if (effect.type === "catch_up") { + requestCanonicalCatchUp(agentId, effect.cursor); + } + } + }, + scheduleFlush: scheduleAgentStreamReducerFlush, + cancelFlush: cancelAgentStreamReducerFlush, + }); +} diff --git a/packages/app/src/types/stream.ts b/packages/app/src/types/stream.ts index 7d74e6998..0b37243cd 100644 --- a/packages/app/src/types/stream.ts +++ b/packages/app/src/types/stream.ts @@ -2,6 +2,7 @@ import type { AgentProvider, ToolCallDetail } from "@server/server/agent/agent-s import type { AgentStreamEventPayload } from "@server/shared/messages"; import type { AttachmentMetadata } from "@/attachments/types"; import { extractTaskEntriesFromToolCall } from "../utils/tool-call-parsers"; +import { splitMarkdownBlocks } from "@/utils/split-markdown-blocks"; /** * Simple hash function for deterministic ID generation @@ -66,6 +67,8 @@ export interface AssistantMessageItem { id: string; text: string; timestamp: Date; + blockGroupId?: string; + blockIndex?: number; } export type ThoughtStatus = "loading" | "ready"; @@ -770,10 +773,97 @@ function finalizeHeadItems(head: StreamItem[]): StreamItem[] { if (item.kind === "thought" && item.status !== "ready") { return markThoughtReady(item); } + if (item.kind === "assistant_message" && item.blockGroupId) { + return { + ...item, + id: createAssistantBlockId({ + groupId: item.blockGroupId, + blockIndex: item.blockIndex ?? 0, + }), + }; + } return item; }); } +function createAssistantBlockId(params: { groupId: string; blockIndex: number }): string { + return `${params.groupId}:block:${params.blockIndex}`; +} + +function getActiveAssistantHeadIndex(head: StreamItem[]): number { + for (let index = head.length - 1; index >= 0; index -= 1) { + if (head[index]?.kind === "assistant_message") { + return index; + } + } + return -1; +} + +function promoteCompletedAssistantBlocks(params: { tail: StreamItem[]; head: StreamItem[] }): { + tail: StreamItem[]; + head: StreamItem[]; + changedTail: boolean; + changedHead: boolean; +} { + const assistantIndex = getActiveAssistantHeadIndex(params.head); + const activeItem = params.head[assistantIndex]; + if (assistantIndex < 0 || !activeItem || activeItem.kind !== "assistant_message") { + return { + tail: params.tail, + head: params.head, + changedTail: false, + changedHead: false, + }; + } + + const blocks = splitMarkdownBlocks(activeItem.text); + if (blocks.length < 2) { + return { + tail: params.tail, + head: params.head, + changedTail: false, + changedHead: false, + }; + } + + const blockGroupId = activeItem.blockGroupId ?? activeItem.id; + const firstBlockIndex = activeItem.blockIndex ?? 0; + const completedBlocks = blocks.slice(0, -1); + const liveBlock = blocks[blocks.length - 1] ?? ""; + const promotedItems = completedBlocks.map((block, offset) => ({ + kind: "assistant_message", + id: createAssistantBlockId({ + groupId: blockGroupId, + blockIndex: firstBlockIndex + offset, + }), + blockGroupId, + blockIndex: firstBlockIndex + offset, + text: block, + timestamp: activeItem.timestamp, + })); + + const nextTail = flushHeadToTail(params.tail, promotedItems); + const liveItem: AssistantMessageItem = { + ...activeItem, + id: `${blockGroupId}:head`, + blockGroupId, + blockIndex: firstBlockIndex + completedBlocks.length, + text: liveBlock, + }; + const nextHead = [ + ...params.head.slice(0, assistantIndex), + liveItem, + ...params.head.slice(assistantIndex + 1), + ]; + + return { + tail: nextTail, + head: nextHead, + changedTail: nextTail !== params.tail, + changedHead: true, + }; +} + /** * Flush head items to tail, avoiding duplicates. */ @@ -905,6 +995,16 @@ export function applyStreamEvent(params: { nextHead = reduced; changedHead = true; } + if (incomingKind === "assistant_message") { + const promoted = promoteCompletedAssistantBlocks({ + tail: nextTail, + head: nextHead, + }); + nextTail = promoted.tail; + nextHead = promoted.head; + changedTail = changedTail || promoted.changedTail; + changedHead = changedHead || promoted.changedHead; + } return { tail: nextTail, head: nextHead, changedTail, changedHead }; }