diff --git a/src/browser/components/Messages/AssistantMessage.tsx b/src/browser/components/Messages/AssistantMessage.tsx index 0f28eb8667..9a3becf0cc 100644 --- a/src/browser/components/Messages/AssistantMessage.tsx +++ b/src/browser/components/Messages/AssistantMessage.tsx @@ -103,7 +103,14 @@ export const AssistantMessage: React.FC = ({ // Streaming text gets typewriter effect if (isStreaming) { - const contentElement = ; + const contentElement = ( + + ); // Wrap streaming compaction in special container if (isStreamingCompaction) { diff --git a/src/browser/components/Messages/ReasoningMessage.tsx b/src/browser/components/Messages/ReasoningMessage.tsx index 955bffa039..99610e0b23 100644 --- a/src/browser/components/Messages/ReasoningMessage.tsx +++ b/src/browser/components/Messages/ReasoningMessage.tsx @@ -111,6 +111,8 @@ export const ReasoningMessage: React.FC = ({ message, cla deltas={[normalizeReasoningMarkdown(content)]} isComplete={false} preserveLineBreaks + streamKey={message.historyId} + streamSource={message.streamPresentation?.source} /> ); } diff --git a/src/browser/components/Messages/TypewriterMarkdown.test.tsx b/src/browser/components/Messages/TypewriterMarkdown.test.tsx new file mode 100644 index 0000000000..ab1318bae9 --- /dev/null +++ b/src/browser/components/Messages/TypewriterMarkdown.test.tsx @@ -0,0 +1,77 @@ +import type { UseSmoothStreamingTextOptions } from "@/browser/hooks/useSmoothStreamingText"; +import { afterEach, beforeEach, describe, expect, mock, test } from "bun:test"; +import { cleanup, render } from "@testing-library/react"; +import { GlobalWindow } from "happy-dom"; + +const mockUseSmoothStreamingText = mock( + (options: UseSmoothStreamingTextOptions): { visibleText: string; isCaughtUp: boolean } => ({ + visibleText: options.fullText, + isCaughtUp: !options.isStreaming, + }) +); + +void mock.module("./MarkdownCore", () => ({ + MarkdownCore: (props: { content: string }) => ( +
{props.content}
+ ), +})); + +void mock.module("@/browser/hooks/useSmoothStreamingText", () => ({ + useSmoothStreamingText: mockUseSmoothStreamingText, +})); + +import { TypewriterMarkdown } from "./TypewriterMarkdown"; + +describe("TypewriterMarkdown", () => { + beforeEach(() => { + globalThis.window = new GlobalWindow() as unknown as Window & typeof globalThis; + globalThis.document = globalThis.window.document; + mockUseSmoothStreamingText.mockClear(); + }); + + afterEach(() => { + cleanup(); + mock.restore(); + globalThis.window = undefined as unknown as Window & typeof globalThis; + globalThis.document = undefined as unknown as Document; + }); + + test("passes smoothed visible text to MarkdownCore when streaming", () => { + mockUseSmoothStreamingText.mockImplementationOnce(() => ({ + visibleText: "Hel", + isCaughtUp: false, + })); + + const view = render( + + ); + + expect(view.getByTestId("markdown-core").textContent).toBe("Hel"); + expect(mockUseSmoothStreamingText).toHaveBeenCalledWith({ + fullText: "Hello world", + isStreaming: true, + bypassSmoothing: false, + streamKey: "msg-1", + }); + }); + + test("bypasses smoothing for replay streams", () => { + render( + + ); + + expect(mockUseSmoothStreamingText).toHaveBeenCalledWith( + expect.objectContaining({ bypassSmoothing: true }) + ); + }); +}); diff --git a/src/browser/components/Messages/TypewriterMarkdown.tsx b/src/browser/components/Messages/TypewriterMarkdown.tsx index 775d2142fe..59f192d11c 100644 --- a/src/browser/components/Messages/TypewriterMarkdown.tsx +++ b/src/browser/components/Messages/TypewriterMarkdown.tsx @@ -1,4 +1,5 @@ import React, { useMemo } from "react"; +import { useSmoothStreamingText } from "@/browser/hooks/useSmoothStreamingText"; import { cn } from "@/common/lib/utils"; import { MarkdownCore } from "./MarkdownCore"; import { StreamingContext } from "./StreamingContext"; @@ -13,6 +14,10 @@ interface TypewriterMarkdownProps { * are often intentional. */ preserveLineBreaks?: boolean; + /** Unique key for the current stream β€” reset smooth engine on change. */ + streamKey?: string; + /** Whether this stream originated from live tokens or replay. Defaults to "live". */ + streamSource?: "live" | "replay"; } // Use React.memo to prevent unnecessary re-renders from parent @@ -21,12 +26,21 @@ export const TypewriterMarkdown = React.memo(function T isComplete, className, preserveLineBreaks, + streamKey, + streamSource = "live", }) { - // Simply join all deltas - no artificial delays or character-by-character rendering - const content = deltas.join(""); + const fullContent = deltas.join(""); + const isStreaming = !isComplete && fullContent.length > 0; - // Show cursor only when streaming (not complete) - const isStreaming = !isComplete && content.length > 0; + // Two-clock streaming: ingestion (fullContent) vs presentation (visibleText). + // The jitter buffer reveals text at a steady cadence instead of bursty token clumps. + // Replay and completed streams bypass smoothing entirely. + const { visibleText } = useSmoothStreamingText({ + fullText: fullContent, + isStreaming, + bypassSmoothing: streamSource === "replay", + streamKey: streamKey ?? "", + }); const streamingContextValue = useMemo(() => ({ isStreaming }), [isStreaming]); @@ -34,7 +48,7 @@ export const TypewriterMarkdown = React.memo(function T
diff --git a/src/browser/hooks/useSmoothStreamingText.test.tsx b/src/browser/hooks/useSmoothStreamingText.test.tsx new file mode 100644 index 0000000000..8aa275efe1 --- /dev/null +++ b/src/browser/hooks/useSmoothStreamingText.test.tsx @@ -0,0 +1,255 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "bun:test"; +import { act, cleanup, renderHook } from "@testing-library/react"; +import { GlobalWindow } from "happy-dom"; +import { + useSmoothStreamingText, + type UseSmoothStreamingTextOptions, +} from "./useSmoothStreamingText"; + +const FRAME_MS = 16; + +describe("useSmoothStreamingText", () => { + let originalRequestAnimationFrame: typeof globalThis.requestAnimationFrame; + let originalCancelAnimationFrame: typeof globalThis.cancelAnimationFrame; + let rafHandleCounter = 0; + let currentTimeMs = 0; + const rafCallbacks = new Map(); + + beforeEach(() => { + const domWindow = new GlobalWindow() as unknown as Window & typeof globalThis; + globalThis.window = domWindow; + globalThis.document = domWindow.document; + + originalRequestAnimationFrame = globalThis.requestAnimationFrame; + originalCancelAnimationFrame = globalThis.cancelAnimationFrame; + + vi.useFakeTimers(); + + rafHandleCounter = 0; + currentTimeMs = 0; + rafCallbacks.clear(); + + const requestAnimationFrameMock: typeof requestAnimationFrame = (callback) => { + rafHandleCounter += 1; + rafCallbacks.set(rafHandleCounter, callback); + return rafHandleCounter; + }; + + const cancelAnimationFrameMock: typeof cancelAnimationFrame = (handle) => { + rafCallbacks.delete(handle); + }; + + globalThis.requestAnimationFrame = requestAnimationFrameMock; + globalThis.cancelAnimationFrame = cancelAnimationFrameMock; + globalThis.window.requestAnimationFrame = requestAnimationFrameMock; + globalThis.window.cancelAnimationFrame = cancelAnimationFrameMock; + }); + + afterEach(() => { + cleanup(); + + rafCallbacks.clear(); + + vi.useRealTimers(); + + globalThis.requestAnimationFrame = originalRequestAnimationFrame; + globalThis.cancelAnimationFrame = originalCancelAnimationFrame; + + if (globalThis.window) { + globalThis.window.requestAnimationFrame = originalRequestAnimationFrame; + globalThis.window.cancelAnimationFrame = originalCancelAnimationFrame; + } + + globalThis.window = undefined as unknown as Window & typeof globalThis; + globalThis.document = undefined as unknown as Document; + }); + + function advanceFrames(frameCount: number): void { + act(() => { + for (let i = 0; i < frameCount; i++) { + currentTimeMs += FRAME_MS; + + const callbacks = Array.from(rafCallbacks.values()); + rafCallbacks.clear(); + + for (const callback of callbacks) { + callback(currentTimeMs); + } + } + }); + } + + function hasLoneSurrogate(value: string): boolean { + for (let i = 0; i < value.length; i++) { + const code = value.charCodeAt(i); + const isHigh = code >= 0xd800 && code <= 0xdbff; + const isLow = code >= 0xdc00 && code <= 0xdfff; + + if (isHigh) { + const next = value.charCodeAt(i + 1); + const nextIsLow = next >= 0xdc00 && next <= 0xdfff; + if (!nextIsLow) { + return true; + } + i += 1; + continue; + } + + if (isLow) { + return true; + } + } + + return false; + } + + it("keeps RAF progress stable while fullText updates rapidly", () => { + const { result, rerender } = renderHook( + (hookProps: UseSmoothStreamingTextOptions) => useSmoothStreamingText(hookProps), + { + initialProps: { + fullText: "x".repeat(20), + isStreaming: true, + bypassSmoothing: false, + streamKey: "stream-rapid", + }, + } + ); + + for (let i = 0; i < 12; i++) { + act(() => { + rerender({ + fullText: "x".repeat(20 + i), + isStreaming: true, + bypassSmoothing: false, + streamKey: "stream-rapid", + }); + }); + + advanceFrames(1); + } + + expect(result.current.visibleText.length).toBeGreaterThan(0); + }); + + it("does not emit partial surrogate pairs while smoothing", () => { + const { result } = renderHook( + (hookProps: UseSmoothStreamingTextOptions) => useSmoothStreamingText(hookProps), + { + initialProps: { + fullText: "πŸ™‚πŸ™‚πŸ™‚", + isStreaming: true, + bypassSmoothing: false, + streamKey: "stream-grapheme", + }, + } + ); + + for (let i = 0; i < 10; i++) { + advanceFrames(1); + expect(hasLoneSurrogate(result.current.visibleText)).toBe(false); + } + }); + + it("reveals text progressively while streaming", () => { + const initialProps: UseSmoothStreamingTextOptions = { + fullText: "x".repeat(220), + isStreaming: true, + bypassSmoothing: false, + streamKey: "stream-1", + }; + + const { result } = renderHook( + (hookProps: UseSmoothStreamingTextOptions) => useSmoothStreamingText(hookProps), + { + initialProps, + } + ); + + const initialLength = result.current.visibleText.length; + expect(initialLength).toBeLessThan(initialProps.fullText.length); + + advanceFrames(8); + + const progressedLength = result.current.visibleText.length; + expect(progressedLength).toBeGreaterThan(initialLength); + expect(progressedLength).toBeLessThan(initialProps.fullText.length); + }); + + it("resets reveal progress when stream key changes", () => { + const firstStreamText = "a".repeat(200); + const secondStreamText = "b".repeat(140); + + const { result, rerender } = renderHook( + (hookProps: UseSmoothStreamingTextOptions) => useSmoothStreamingText(hookProps), + { + initialProps: { + fullText: firstStreamText, + isStreaming: true, + bypassSmoothing: false, + streamKey: "stream-1", + }, + } + ); + + advanceFrames(12); + + const firstStreamProgress = result.current.visibleText.length; + expect(firstStreamProgress).toBeGreaterThan(0); + + act(() => { + rerender({ + fullText: secondStreamText, + isStreaming: true, + bypassSmoothing: false, + streamKey: "stream-2", + }); + }); + + const resetLength = result.current.visibleText.length; + expect(resetLength).toBeLessThan(firstStreamProgress); + expect(resetLength).toBeLessThan(secondStreamText.length); + + advanceFrames(6); + + expect(result.current.visibleText.length).toBeGreaterThan(resetLength); + }); + + it("re-arms smoothing after catch-up when new deltas arrive", () => { + const shortText = "x".repeat(40); + const longerText = "x".repeat(200); + + const { result, rerender } = renderHook( + (hookProps: UseSmoothStreamingTextOptions) => useSmoothStreamingText(hookProps), + { + initialProps: { + fullText: shortText, + isStreaming: true, + bypassSmoothing: false, + streamKey: "stream-rearm", + }, + } + ); + + // Advance until fully caught up with the short text. + advanceFrames(60); + expect(result.current.isCaughtUp).toBe(true); + const caughtUpLength = result.current.visibleText.length; + expect(caughtUpLength).toBe(shortText.length); + + // Simulate new deltas arriving (same stream, longer text). + act(() => { + rerender({ + fullText: longerText, + isStreaming: true, + bypassSmoothing: false, + streamKey: "stream-rearm", + }); + }); + + // The hook should re-arm and start revealing the new text. + advanceFrames(4); + expect(result.current.visibleText.length).toBeGreaterThan(caughtUpLength); + expect(result.current.visibleText.length).toBeLessThan(longerText.length); + }); +}); diff --git a/src/browser/hooks/useSmoothStreamingText.ts b/src/browser/hooks/useSmoothStreamingText.ts new file mode 100644 index 0000000000..1f6193bb2a --- /dev/null +++ b/src/browser/hooks/useSmoothStreamingText.ts @@ -0,0 +1,151 @@ +import { useEffect, useRef, useState } from "react"; +import { SmoothTextEngine } from "@/browser/utils/streaming/SmoothTextEngine"; + +export interface UseSmoothStreamingTextOptions { + fullText: string; + isStreaming: boolean; + bypassSmoothing: boolean; + /** Changing this resets the engine (new stream). */ + streamKey: string; +} + +export interface UseSmoothStreamingTextResult { + visibleText: string; + isCaughtUp: boolean; +} + +const graphemeSegmenter = + typeof Intl !== "undefined" && typeof Intl.Segmenter === "function" + ? new Intl.Segmenter(undefined, { granularity: "grapheme" }) + : null; + +function sliceAtGraphemeBoundary(text: string, maxCodeUnitLength: number): string { + if (maxCodeUnitLength <= 0) { + return ""; + } + + if (maxCodeUnitLength >= text.length) { + return text; + } + + if (graphemeSegmenter) { + let safeEnd = 0; + + for (const segment of graphemeSegmenter.segment(text)) { + const segmentEnd = segment.index + segment.segment.length; + if (segmentEnd > maxCodeUnitLength) { + break; + } + safeEnd = segmentEnd; + } + + return text.slice(0, safeEnd); + } + + let safeEnd = 0; + for (const codePoint of Array.from(text)) { + const codePointEnd = safeEnd + codePoint.length; + if (codePointEnd > maxCodeUnitLength) { + break; + } + safeEnd = codePointEnd; + } + + return text.slice(0, safeEnd); +} + +export function useSmoothStreamingText( + options: UseSmoothStreamingTextOptions +): UseSmoothStreamingTextResult { + const engineRef = useRef(new SmoothTextEngine()); + const previousStreamKeyRef = useRef(options.streamKey); + + if (previousStreamKeyRef.current !== options.streamKey) { + engineRef.current.reset(); + previousStreamKeyRef.current = options.streamKey; + } + + const engine = engineRef.current; + engine.update(options.fullText, options.isStreaming, options.bypassSmoothing); + + const [visibleLength, setVisibleLength] = useState(() => engine.visibleLength); + const visibleLengthRef = useRef(visibleLength); + visibleLengthRef.current = visibleLength; + + const rafIdRef = useRef(null); + const previousTimestampRef = useRef(null); + + // Frame callback stored as a ref so effects don't depend on it, preventing + // teardown/restart of the RAF loop on every text delta. Reads from refs and + // the stable engine instance, so the captured closure is always correct. + const frameRef = useRef(null!); + frameRef.current = (timestampMs: number) => { + if (previousTimestampRef.current !== null) { + const nextLength = engine.tick(timestampMs - previousTimestampRef.current); + if (nextLength !== visibleLengthRef.current) { + visibleLengthRef.current = nextLength; + setVisibleLength(nextLength); + } + } + previousTimestampRef.current = timestampMs; + if (!engine.isCaughtUp) { + rafIdRef.current = requestAnimationFrame(frameRef.current); + } else { + rafIdRef.current = null; + previousTimestampRef.current = null; + } + }; + + // Sync engine state β†’ React and re-arm RAF when new deltas arrive after + // catch-up. No cleanup: this effect only observes + one-shot starts; the + // lifecycle effect below owns resource teardown. + useEffect(() => { + if (visibleLengthRef.current !== engine.visibleLength) { + visibleLengthRef.current = engine.visibleLength; + setVisibleLength(engine.visibleLength); + } + + if ( + rafIdRef.current === null && + options.isStreaming && + !options.bypassSmoothing && + !engine.isCaughtUp + ) { + rafIdRef.current = requestAnimationFrame(frameRef.current); + } + }, [engine, options.fullText, options.isStreaming, options.bypassSmoothing, options.streamKey]); + + // Lifecycle: stop RAF when streaming ends or stream key changes, and on unmount. + useEffect(() => { + if (!options.isStreaming || options.bypassSmoothing) { + if (rafIdRef.current !== null) cancelAnimationFrame(rafIdRef.current); + rafIdRef.current = null; + previousTimestampRef.current = null; + } + return () => { + if (rafIdRef.current !== null) cancelAnimationFrame(rafIdRef.current); + rafIdRef.current = null; + previousTimestampRef.current = null; + }; + }, [options.isStreaming, options.bypassSmoothing, options.streamKey]); + + if (!options.isStreaming || options.bypassSmoothing) { + return { + visibleText: options.fullText, + isCaughtUp: true, + }; + } + + const visiblePrefixLength = Math.min( + visibleLength, + engine.visibleLength, + options.fullText.length + ); + + const visibleText = sliceAtGraphemeBoundary(options.fullText, visiblePrefixLength); + + return { + visibleText, + isCaughtUp: visibleText.length === options.fullText.length, + }; +} diff --git a/src/browser/stores/WorkspaceStore.ts b/src/browser/stores/WorkspaceStore.ts index 03743eec73..b7b8356845 100644 --- a/src/browser/stores/WorkspaceStore.ts +++ b/src/browser/stores/WorkspaceStore.ts @@ -778,6 +778,10 @@ export class WorkspaceStore { * slow machines naturally throttle without dropping data. * * Data is always updated immediately in the aggregator - only UI notification is deferred. + * + * NOTE: This is the "ingestion clock" half of the two-clock streaming model. + * The "presentation clock" (useSmoothStreamingText) handles visual cadence + * independently β€” do not collapse them into a single mechanism. */ private scheduleIdleStateBump(workspaceId: string): void { // Skip if already scheduled diff --git a/src/browser/utils/messages/StreamingMessageAggregator.test.ts b/src/browser/utils/messages/StreamingMessageAggregator.test.ts index af30aa5b6a..b7155b1acd 100644 --- a/src/browser/utils/messages/StreamingMessageAggregator.test.ts +++ b/src/browser/utils/messages/StreamingMessageAggregator.test.ts @@ -1104,6 +1104,150 @@ describe("StreamingMessageAggregator", () => { const afterReplayCursor = aggregator.getOnChatCursor(); expect(afterReplayCursor?.stream?.lastTimestamp).toBe(deltaTimestamp); }); + + test("marks streamed assistant rows as replay presentation when stream-start is replayed", () => { + const aggregator = new StreamingMessageAggregator(TEST_CREATED_AT); + + aggregator.handleStreamStart({ + type: "stream-start", + workspaceId: "test-workspace", + messageId: "msg-replay-presentation", + historySequence: 1, + model: "claude-3-5-sonnet-20241022", + startTime: 1_000, + replay: true, + }); + + aggregator.handleStreamDelta({ + type: "stream-delta", + workspaceId: "test-workspace", + messageId: "msg-replay-presentation", + delta: "replayed partial", + tokens: 1, + timestamp: 1_100, + replay: true, + }); + + const displayed = aggregator.getDisplayedMessages(); + const assistant = displayed.find( + (message): message is Extract<(typeof displayed)[number], { type: "assistant" }> => + message.type === "assistant" && message.historyId === "msg-replay-presentation" + ); + + expect(assistant).toBeDefined(); + expect(assistant?.streamPresentation).toEqual({ source: "replay" }); + }); + test("switches streaming presentation from replay to live when non-replay delta arrives", () => { + const aggregator = new StreamingMessageAggregator(TEST_CREATED_AT); + + // Reconnect: stream-start with replay flag + aggregator.handleStreamStart({ + type: "stream-start", + workspaceId: "test-workspace", + messageId: "msg-replay-to-live", + historySequence: 1, + model: "claude-3-5-sonnet-20241022", + startTime: 1_000, + replay: true, + }); + + // Replay catch-up delta (tagged with replay) + aggregator.handleStreamDelta({ + type: "stream-delta", + workspaceId: "test-workspace", + messageId: "msg-replay-to-live", + delta: "cached ", + tokens: 1, + timestamp: 1_100, + replay: true, + }); + + // During replay: source should be "replay" + const duringReplay = aggregator.getDisplayedMessages(); + const replayRow = duringReplay.find( + (message): message is Extract<(typeof duringReplay)[number], { type: "assistant" }> => + message.type === "assistant" && message.historyId === "msg-replay-to-live" + ); + expect(replayRow?.streamPresentation).toEqual({ source: "replay" }); + + // Fresh live delta (no replay flag) β€” catch-up is over + aggregator.handleStreamDelta({ + type: "stream-delta", + workspaceId: "test-workspace", + messageId: "msg-replay-to-live", + delta: "fresh tokens", + tokens: 2, + timestamp: 1_200, + }); + + // After live resume: source should flip to "live" + const afterLive = aggregator.getDisplayedMessages(); + const liveRow = afterLive.find( + (message): message is Extract<(typeof afterLive)[number], { type: "assistant" }> => + message.type === "assistant" && message.historyId === "msg-replay-to-live" + ); + expect(liveRow?.streamPresentation).toEqual({ source: "live" }); + }); + + test("does not exit replay phase on non-replay tool events arriving before replay text drains", () => { + const aggregator = new StreamingMessageAggregator(TEST_CREATED_AT); + + // Reconnect: stream-start with replay flag + aggregator.handleStreamStart({ + type: "stream-start", + workspaceId: "test-workspace", + messageId: "msg-tool-during-replay", + historySequence: 1, + model: "claude-3-5-sonnet-20241022", + startTime: 1_000, + replay: true, + }); + + // Replay catch-up delta + aggregator.handleStreamDelta({ + type: "stream-delta", + workspaceId: "test-workspace", + messageId: "msg-tool-during-replay", + delta: "cached ", + tokens: 1, + timestamp: 1_100, + replay: true, + }); + + // Non-replay tool event arrives before replay text finishes draining. + // Tool events are not buffered by the reconnect relay, so they can + // arrive without the replay flag even while replay text is still in-flight. + aggregator.handleToolCallStart({ + type: "tool-call-start", + workspaceId: "test-workspace", + messageId: "msg-tool-during-replay", + toolCallId: "tool-1", + toolName: "bash", + args: { command: "echo hi" }, + tokens: 1, + timestamp: 1_150, + }); + + // Another replay delta arrives (still part of catch-up) + aggregator.handleStreamDelta({ + type: "stream-delta", + workspaceId: "test-workspace", + messageId: "msg-tool-during-replay", + delta: "tail", + tokens: 1, + timestamp: 1_200, + replay: true, + }); + + // Source must still be "replay" β€” tool event must not have flipped it + const displayed = aggregator.getDisplayedMessages(); + const assistantRows = displayed.filter( + (message): message is Extract<(typeof displayed)[number], { type: "assistant" }> => + message.type === "assistant" && message.historyId === "msg-tool-during-replay" + ); + const assistant = assistantRows.at(-1); + expect(assistant?.streamPresentation).toEqual({ source: "replay" }); + }); }); describe("append replay cache invalidation", () => { diff --git a/src/browser/utils/messages/StreamingMessageAggregator.ts b/src/browser/utils/messages/StreamingMessageAggregator.ts index 5b09c43fd1..955f39eea8 100644 --- a/src/browser/utils/messages/StreamingMessageAggregator.ts +++ b/src/browser/utils/messages/StreamingMessageAggregator.ts @@ -112,6 +112,7 @@ interface StreamingContext { isComplete: boolean; isCompacting: boolean; hasCompactionContinue: boolean; + isReplay: boolean; model: string; routedThroughGateway?: boolean; @@ -501,6 +502,27 @@ export class StreamingMessageAggregator { context.clockOffsetMs = Date.now() - serverTimestamp; } + /** + * Detect the replayβ†’live transition for reconnect streams. + * + * During reconnect, `replayStream()` emits all catch-up events with `replay: true`. + * Once the catch-up phase is over, fresh live deltas arrive without the flag. + * This helper flips `isReplay` to false on the first non-replay event so that + * `streamPresentation.source` correctly transitions to "live" and smoothing + * resumes instead of staying bypassed. + * + * IMPORTANT: Only call from content handlers (handleStreamDelta, handleReasoningDelta). + * Tool events are not buffered by the reconnect relay and can arrive before replay + * text finishes flushing β€” calling this from tool handlers would prematurely end + * replay phase and reclassify catch-up content as live. + */ + private syncReplayPhase(messageId: string, replay?: boolean): void { + const context = this.activeStreams.get(messageId); + if (context && context.isReplay && replay !== true) { + context.isReplay = false; + } + } + private translateServerTime(context: StreamingContext, serverTimestamp: number): number { assert(context, "translateServerTime requires context"); assert(typeof serverTimestamp === "number", "translateServerTime requires serverTimestamp"); @@ -1395,6 +1417,7 @@ export class StreamingMessageAggregator { isComplete: false, isCompacting, hasCompactionContinue, + isReplay: data.replay === true, model: data.model, routedThroughGateway: data.routedThroughGateway, serverFirstTokenTime: null, @@ -1458,6 +1481,8 @@ export class StreamingMessageAggregator { const message = this.messages.get(data.messageId); if (!message) return; + this.syncReplayPhase(data.messageId, data.replay); + const context = this.activeStreams.get(data.messageId); if (context) { this.updateStreamClock(context, data.timestamp); @@ -1989,6 +2014,8 @@ export class StreamingMessageAggregator { const message = this.messages.get(data.messageId); if (!message) return; + this.syncReplayPhase(data.messageId, data.replay); + const context = this.activeStreams.get(data.messageId); if (context) { this.updateStreamClock(context, data.timestamp); @@ -2355,6 +2382,7 @@ export class StreamingMessageAggregator { // Check if this message has an active stream (for inferring streaming status) // Direct Map.has() check - O(1) instead of O(n) iteration const hasActiveStream = this.activeStreams.has(message.id); + const streamContext = hasActiveStream ? this.activeStreams.get(message.id) : undefined; // isPartial from metadata (set by stream-abort event) const isPartial = message.metadata?.partial === true; @@ -2400,6 +2428,9 @@ export class StreamingMessageAggregator { isPartial, isLastPartOfMessage: isLastPart, timestamp: part.timestamp ?? baseTimestamp, + streamPresentation: isStreaming + ? { source: streamContext?.isReplay ? "replay" : "live" } + : undefined, }); } else if (part.type === "text" && part.text) { // Skip empty text parts @@ -2421,6 +2452,9 @@ export class StreamingMessageAggregator { mode: message.metadata?.mode, agentId: message.metadata?.agentId ?? message.metadata?.mode, timestamp: part.timestamp ?? baseTimestamp, + streamPresentation: isStreaming + ? { source: streamContext?.isReplay ? "replay" : "live" } + : undefined, }); } else if (isDynamicToolPart(part)) { // Determine status based on part state and result diff --git a/src/browser/utils/streaming/SmoothTextEngine.test.ts b/src/browser/utils/streaming/SmoothTextEngine.test.ts new file mode 100644 index 0000000000..f25328bb45 --- /dev/null +++ b/src/browser/utils/streaming/SmoothTextEngine.test.ts @@ -0,0 +1,151 @@ +import { describe, it, expect } from "bun:test"; +import { STREAM_SMOOTHING } from "@/constants/streaming"; +import { SmoothTextEngine } from "./SmoothTextEngine"; + +function makeText(length: number): string { + return "x".repeat(length); +} + +describe("SmoothTextEngine", () => { + it("reveals text steadily and reaches full length", () => { + const engine = new SmoothTextEngine(); + const fullText = makeText(200); + + engine.update(fullText, true, false); + + let previousLength = engine.visibleLength; + let reachedFullLength = false; + + for (let i = 0; i < 600; i++) { + const nextLength = engine.tick(16); + expect(nextLength).toBeGreaterThanOrEqual(previousLength); + previousLength = nextLength; + + if (nextLength === fullText.length) { + reachedFullLength = true; + break; + } + } + + expect(reachedFullLength).toBe(true); + expect(engine.visibleLength).toBe(fullText.length); + expect(engine.isCaughtUp).toBe(true); + }); + + it("accelerates reveal speed when backlog is large", () => { + const engine = new SmoothTextEngine(); + const fullText = makeText(500); + + engine.update(fullText, true, false); + + let previousLength = engine.visibleLength; + let revealedCharsInFirst20Ticks = 0; + + for (let i = 0; i < 20; i++) { + const nextLength = engine.tick(16); + revealedCharsInFirst20Ticks += nextLength - previousLength; + previousLength = nextLength; + } + + // Baseline low-backlog behavior reveals ~1 char/frame with MIN_FRAME_CHARS. + // A large backlog should reveal multiple chars/frame on average. + expect(revealedCharsInFirst20Ticks).toBeGreaterThan(20); + }); + + it("caps visual lag when incoming text jumps ahead", () => { + const engine = new SmoothTextEngine(); + + engine.update(makeText(40), true, false); + + while (!engine.isCaughtUp) { + engine.tick(16); + } + + engine.update(makeText(420), true, false); + + expect(420 - engine.visibleLength).toBeLessThanOrEqual(STREAM_SMOOTHING.MAX_VISUAL_LAG_CHARS); + }); + + it("flushes immediately when streaming ends", () => { + const engine = new SmoothTextEngine(); + const fullText = makeText(120); + + engine.update(fullText, true, false); + + for (let i = 0; i < 15; i++) { + engine.tick(16); + } + + expect(engine.visibleLength).toBeLessThan(fullText.length); + + engine.update(fullText, false, false); + + expect(engine.visibleLength).toBe(fullText.length); + expect(engine.isCaughtUp).toBe(true); + }); + + it("bypasses smoothing and returns full length immediately", () => { + const engine = new SmoothTextEngine(); + const fullText = makeText(80); + + engine.update(fullText, true, true); + + expect(engine.visibleLength).toBe(fullText.length); + expect(engine.isCaughtUp).toBe(true); + }); + + it("clamps visible length when content shrinks", () => { + const engine = new SmoothTextEngine(); + + engine.update(makeText(100), true, false); + + while (engine.visibleLength < 50) { + engine.tick(16); + } + + engine.update(makeText(30), true, false); + + expect(engine.visibleLength).toBe(30); + }); + + it("does not force reveal when budget is below one char", () => { + const engine = new SmoothTextEngine(); + // With a 1-char backlog, adaptive rate is at floor (~24 cps). + // At 4ms per tick: 24 * 0.004 = 0.096 budget per tick. + // Budget reaches 1.0 after ceil(1 / 0.096) β‰ˆ 11 ticks. + engine.update("x", true, false); + + // First tick at 4ms should not reveal (budget ~0.10). + const afterFirstTick = engine.tick(4); + expect(afterFirstTick).toBe(0); + + // Several more small ticks should still not reveal. + engine.tick(4); + engine.tick(4); + expect(engine.visibleLength).toBe(0); + + // After enough ticks to accumulate >= 1 char, it should reveal. + for (let i = 0; i < 20; i++) { + engine.tick(4); + } + expect(engine.visibleLength).toBeGreaterThan(0); + }); + + it("keeps reveal near frame-rate invariant over equal wall time", () => { + const run = (frameMs: number) => { + const engine = new SmoothTextEngine(); + engine.update(makeText(400), true, false); + for (let t = 0; t < 1000; t += frameMs) { + engine.tick(frameMs); + } + return engine.visibleLength; + }; + + const at60Hz = run(16); + const at240Hz = run(4); + + // Over 1 second of wall time, both refresh rates should reveal + // approximately the same number of characters. + expect(Math.abs(at60Hz - at240Hz)).toBeLessThanOrEqual(2); + }); +}); diff --git a/src/browser/utils/streaming/SmoothTextEngine.ts b/src/browser/utils/streaming/SmoothTextEngine.ts new file mode 100644 index 0000000000..7813a7fd9a --- /dev/null +++ b/src/browser/utils/streaming/SmoothTextEngine.ts @@ -0,0 +1,126 @@ +import { STREAM_SMOOTHING } from "@/constants/streaming"; + +function clamp(value: number, min: number, max: number): number { + return Math.max(min, Math.min(max, value)); +} + +function getAdaptiveRate(backlog: number): number { + const backlogPressure = clamp(backlog / STREAM_SMOOTHING.CATCHUP_BACKLOG_CHARS, 0, 1); + + const targetRate = + STREAM_SMOOTHING.BASE_CHARS_PER_SEC + + backlogPressure * (STREAM_SMOOTHING.MAX_CHARS_PER_SEC - STREAM_SMOOTHING.BASE_CHARS_PER_SEC); + + return clamp(targetRate, STREAM_SMOOTHING.MIN_CHARS_PER_SEC, STREAM_SMOOTHING.MAX_CHARS_PER_SEC); +} + +/** + * Deterministic text reveal engine for smoothing streamed output. + * + * The ingestion clock (incoming full text) is external; this class manages only + * the presentation clock (visible prefix length) using a character budget model. + */ +export class SmoothTextEngine { + private fullLength = 0; + private visibleLengthValue = 0; + private charBudget = 0; + private isStreaming = false; + private bypassSmoothing = false; + + private enforceMaxVisualLag(): void { + if (!this.isStreaming || this.bypassSmoothing) { + return; + } + + // Keep visible output near the ingested stream so interruption doesn't reveal + // a large hidden tail all at once. + const minVisibleLength = Math.max(0, this.fullLength - STREAM_SMOOTHING.MAX_VISUAL_LAG_CHARS); + if (this.visibleLengthValue < minVisibleLength) { + this.visibleLengthValue = minVisibleLength; + this.charBudget = 0; + } + } + + /** + * Update the ingested text and stream state. + */ + update(fullText: string, isStreaming: boolean, bypassSmoothing: boolean): void { + this.fullLength = fullText.length; + this.isStreaming = isStreaming; + this.bypassSmoothing = bypassSmoothing; + + if (this.fullLength < this.visibleLengthValue) { + this.visibleLengthValue = this.fullLength; + this.charBudget = 0; + } + + if (!isStreaming || bypassSmoothing) { + this.visibleLengthValue = this.fullLength; + this.charBudget = 0; + return; + } + + this.enforceMaxVisualLag(); + } + + /** + * Advance the presentation clock by a timestep. + */ + tick(dtMs: number): number { + if (dtMs <= 0) { + return this.visibleLengthValue; + } + + if (!this.isStreaming || this.bypassSmoothing) { + return this.visibleLengthValue; + } + + if (this.visibleLengthValue > this.fullLength) { + this.visibleLengthValue = this.fullLength; + this.charBudget = 0; + } + + if (this.visibleLengthValue === this.fullLength) { + return this.visibleLengthValue; + } + + const backlog = this.fullLength - this.visibleLengthValue; + const adaptiveRate = getAdaptiveRate(backlog); + + this.charBudget += adaptiveRate * (dtMs / 1000); + + // Budget-gated reveal: only reveal when at least one whole character has + // accrued. This makes cadence frame-rate invariant β€” a 240Hz display + // accumulates budget across several frames before revealing, rather than + // forcing 1 char/frame at any refresh rate. + const wholeCharsReady = Math.floor(this.charBudget); + if (wholeCharsReady < STREAM_SMOOTHING.MIN_FRAME_CHARS) { + return this.visibleLengthValue; + } + + const reveal = Math.min(wholeCharsReady, STREAM_SMOOTHING.MAX_FRAME_CHARS); + this.visibleLengthValue = Math.min(this.fullLength, this.visibleLengthValue + reveal); + this.charBudget -= reveal; + + return this.visibleLengthValue; + } + + get visibleLength(): number { + return this.visibleLengthValue; + } + + get isCaughtUp(): boolean { + return this.visibleLengthValue === this.fullLength; + } + + /** + * Reset all engine state, typically when a new stream starts. + */ + reset(): void { + this.fullLength = 0; + this.visibleLengthValue = 0; + this.charBudget = 0; + this.isStreaming = false; + this.bypassSmoothing = false; + } +} diff --git a/src/common/types/message.ts b/src/common/types/message.ts index 9cf6b382f9..91fcc841a6 100644 --- a/src/common/types/message.ts +++ b/src/common/types/message.ts @@ -531,6 +531,10 @@ export type DisplayedMessage = mode?: AgentMode; timestamp?: number; tokens?: number; + /** Presentation hint for smooth streaming β€” indicates if this is live or replayed content. */ + streamPresentation?: { + source: "live" | "replay"; + }; } | { type: "tool"; @@ -569,6 +573,10 @@ export type DisplayedMessage = isLastPartOfMessage?: boolean; // True if this is the last part of a multi-part message timestamp?: number; tokens?: number; // Reasoning tokens if available + /** Presentation hint for smooth streaming β€” indicates if this is live or replayed content. */ + streamPresentation?: { + source: "live" | "replay"; + }; } | { type: "stream-error"; diff --git a/src/constants/streaming.ts b/src/constants/streaming.ts new file mode 100644 index 0000000000..790a3d29a4 --- /dev/null +++ b/src/constants/streaming.ts @@ -0,0 +1,19 @@ +// Smooth streaming presentation constants. +// These control the jitter buffer that makes streamed text appear at a steady cadence +// instead of bursty token clumps. Internal-only; no user-facing setting. +export const STREAM_SMOOTHING = { + /** Baseline reveal speed in characters per second. */ + BASE_CHARS_PER_SEC: 72, + /** Floor β€” never slower than this even when buffer is nearly empty. */ + MIN_CHARS_PER_SEC: 24, + /** Ceiling β€” hard cap to prevent overwhelming the markdown renderer. */ + MAX_CHARS_PER_SEC: 420, + /** Backlog level where adaptive reveal runs at MAX_CHARS_PER_SEC. */ + CATCHUP_BACKLOG_CHARS: 180, + /** Keep the rendered transcript close to live output even during bursty streams. */ + MAX_VISUAL_LAG_CHARS: 120, + /** Max characters revealed in a single animation frame. */ + MAX_FRAME_CHARS: 48, + /** Min characters revealed per tick once budget permits (avoids sub-character stalls). */ + MIN_FRAME_CHARS: 1, +} as const;