From f315ccd25678d291f36ec51755f907e859a4012d Mon Sep 17 00:00:00 2001 From: Ivan Malison Date: Mon, 14 Sep 2026 01:26:07 -0700 Subject: [PATCH] perf(app): stop re-walking the timeline on every streamed delta Every coalesced delta rebuilds the stream render model with the same tail. The model re-sliced the tail at the history-window start on every call, so the rendered tail was a fresh array per delta and every WeakMap keyed on it (ordering, history split, turn timing, the layout history cache, row revision) missed each time. Cache the slice by tail identity and window start. Turn timing then still walked the whole rendered tail per delta and allocated a new assistant-timing Map, which the layout memo depends on. Split the walk: the tail is walked once per tail identity and the head is walked from the carried open turn per delta. The Map keeps its identity while the head adds no completed turn. Adds identity-contract tests and a vitest bench for the per-delta pipeline. 200 deltas over a 2,000-item timeline, mean per run: default mounted window 5.68ms -> 2.82ms, history fully revealed 16.95ms -> 1.81ms. Fixes #4816 Co-Authored-By: Claude Fable 5.1 --- docs/agent-stream-performance.md | 11 ++ packages/app/src/agent-stream/model.test.ts | 43 +++++ packages/app/src/agent-stream/model.ts | 55 +++--- .../app/src/agent-stream/stream-tick.bench.ts | 132 ++++++++++++++ packages/app/src/timeline/turn-time.test.ts | 97 +++++++++- packages/app/src/timeline/turn-time.ts | 168 +++++++++++++----- 6 files changed, 425 insertions(+), 81 deletions(-) create mode 100644 packages/app/src/agent-stream/stream-tick.bench.ts diff --git a/docs/agent-stream-performance.md b/docs/agent-stream-performance.md index b31a3fd9ee..626a9940b2 100644 --- a/docs/agent-stream-performance.md +++ b/docs/agent-stream-performance.md @@ -58,12 +58,23 @@ So arrival sets a _target_ and the reveal rate is derived from the backlog inste through that hook; the web viewport once skipped it and history hosts of a live tool group went stale. A new field on `StreamLayoutItem` must be added to `areLayoutItemsEquivalent`, or sharing silently stops. +- **Per-delta work is bounded by the head, not the timeline.** `buildAgentStreamRenderModel` runs + once per coalesced delta, and everything downstream (ordering, history split, turn timing, + `layoutStream`'s history cache, `useRevisedHistoryRows`) is a `WeakMap` keyed on the rendered + tail's identity. The model caches the history-window slice for that reason; a fresh array there + restarts every cache on every delta. Turn timing walks the tail once per tail identity and only + the head per delta, and returns the same `byAssistantId` Map while the head adds no completed + turn, so the layout memo does not miss on it. Anything else fed the full tail (the chat outline, + the history-window boundary lookup) memoizes on the tail and does per-delta work on the head + only. `stream-tick.bench.ts` measures this pipeline; `model.test.ts` and `turn-time.test.ts` + assert the identity contracts. ## Measuring - **Smoothness (user-perceived):** `packages/app/e2e/browser/agent-stream-smoothness.spec.ts`, gated behind `PASEO_AGENT_STREAM_PERF_E2E=1`. Drives the mock provider's `bursty-stream` model and reports coefficient of variation of characters painted per frame (smoothness) plus p95 gap between visible updates (stalls). Both numbers are needed: a stalled stream is perfectly smooth. - **Reproducing bursty arrival:** the `bursty-stream` model in `mock-load-test-agent.ts` emits uneven runs of tokens separated by idle gaps. Burst sizes come from a seeded generator, so a run repeats exactly. - **Rate policy in isolation:** `computeRevealStep` in `packages/app/src/agent-stream/text-reveal.ts` is pure; `text-reveal.test.ts` covers convergence and burst flattening without a renderer. +- **Per-delta cost on a long timeline:** `npx vitest bench --project unit src/agent-stream/stream-tick.bench.ts` from `packages/app` replays 200 deltas over a 2,000-item timeline through the pure model and layout pipeline, with the default mounted window and with history fully revealed. It measures the JS work per delta, not paint. Healthy numbers (2026-08, Expo web against a local dev daemon, real Claude Haiku agent, ~8.5s samples during active streaming). Setting `TEXT_REVEAL_HORIZON_MS` to 0 makes the reveal paint on arrival, which is how the baseline column was taken: diff --git a/packages/app/src/agent-stream/model.test.ts b/packages/app/src/agent-stream/model.test.ts index e29792ab02..dec4c489b9 100644 --- a/packages/app/src/agent-stream/model.test.ts +++ b/packages/app/src/agent-stream/model.test.ts @@ -240,4 +240,47 @@ describe("buildAgentStreamRenderModel", () => { expect(model.turnTiming.byAssistantId.size).toBe(0); }); + + it.each(["web", "native"] as const)( + "reuses rendered history and timing across live-head updates inside a history window on %s", + (platform) => { + const at = (seed: number) => new Date(Date.UTC(2026, 0, 1, 0, 0, seed)); + const tail: StreamItem[] = []; + for (let index = 0; index < 40; index += 1) { + tail.push( + { ...userMessage(`u${index}`, 0), timestamp: at(index * 2) }, + { ...assistantMessage(`a${index}`, 0), timestamp: at(index * 2 + 1) }, + ); + } + const live = (text: string): StreamItem => ({ + kind: "assistant_message", + id: "live", + text, + timestamp: at(81), + }); + const build = (head: StreamItem[]) => + buildAgentStreamRenderModel({ + isTurnActive: true, + activeTurnStartedAt: tail.at(-2)?.timestamp ?? null, + tail, + head, + platform, + isMobileBreakpoint: false, + historyStart: 60, + }); + + const first = build([live("live")]); + const second = build([live("live grows")]); + + expect(second.history).toBe(first.history); + expect(second.segments.historyMounted).toBe(first.segments.historyMounted); + expect(second.turnTiming.byAssistantId).toBe(first.turnTiming.byAssistantId); + expect(first.history).toHaveLength(20); + expect(first.turnTiming.byAssistantId.has("a29")).toBe(false); + expect(first.turnTiming.byAssistantId.get("a38")).toEqual({ + completedAt: tail[77]?.timestamp, + durationMs: 1000, + }); + }, + ); }); diff --git a/packages/app/src/agent-stream/model.ts b/packages/app/src/agent-stream/model.ts index 89eb73a7be..b320b6b491 100644 --- a/packages/app/src/agent-stream/model.ts +++ b/packages/app/src/agent-stream/model.ts @@ -53,10 +53,27 @@ const splitHistoryCache = new WeakMap< StreamItem[], Map> >(); -const turnTimingCache = new WeakMap< - StreamItem[], - WeakMap> ->(); +// A streamed delta rebuilds the model with the same tail. Everything below is keyed on +// the rendered tail's identity, so the slice has to be reused, not recomputed. +const renderedTailCache = new WeakMap>(); + +function getRenderedTail(tail: StreamItem[], historyStart: number | undefined): StreamItem[] { + if (!historyStart) { + return tail; + } + let byStart = renderedTailCache.get(tail); + if (!byStart) { + byStart = new Map(); + renderedTailCache.set(tail, byStart); + } + const cached = byStart.get(historyStart); + if (cached) { + return cached; + } + const rendered = tail.slice(historyStart); + byStart.set(historyStart, rendered); + return rendered; +} function getOrderedItems(params: { cache: WeakMap>; @@ -129,32 +146,6 @@ function splitOrderedTail(params: { return split; } -function getTurnTiming(params: { - isTurnActive: boolean; - activeTurnStartedAt: Date | null; - tail: StreamItem[]; - head: StreamItem[]; -}): StreamTurnTiming { - let cachedByHead = turnTimingCache.get(params.tail); - if (!cachedByHead) { - cachedByHead = new WeakMap(); - turnTimingCache.set(params.tail, cachedByHead); - } - let cachedByActivity = cachedByHead.get(params.head); - if (!cachedByActivity) { - cachedByActivity = new Map(); - cachedByHead.set(params.head, cachedByActivity); - } - const activityKey = `${params.isTurnActive}:${params.activeTurnStartedAt?.getTime() ?? "none"}`; - const cached = cachedByActivity.get(activityKey); - if (cached) { - return cached; - } - const timing = deriveStreamTurnTiming(params); - cachedByActivity.set(activityKey, timing); - return timing; -} - export function buildAgentStreamRenderModel( input: BuildAgentStreamRenderModelInput, ): AgentStreamRenderModel { @@ -163,7 +154,7 @@ export function buildAgentStreamRenderModel( isMobileBreakpoint: input.isMobileBreakpoint, }); const orderingCacheKey = `${input.platform}:${input.isMobileBreakpoint}`; - const renderedTail = input.historyStart ? input.tail.slice(input.historyStart) : input.tail; + const renderedTail = getRenderedTail(input.tail, input.historyStart); const orderedTail = getOrderedItems({ cache: orderedTailCache, source: renderedTail, @@ -189,7 +180,7 @@ export function buildAgentStreamRenderModel( platform: input.platform, isMobileBreakpoint: input.isMobileBreakpoint, }); - const turnTiming = getTurnTiming({ + const turnTiming = deriveStreamTurnTiming({ isTurnActive: input.isTurnActive, activeTurnStartedAt: input.activeTurnStartedAt, tail: renderedTail, diff --git a/packages/app/src/agent-stream/stream-tick.bench.ts b/packages/app/src/agent-stream/stream-tick.bench.ts new file mode 100644 index 0000000000..dc1fecb881 --- /dev/null +++ b/packages/app/src/agent-stream/stream-tick.bench.ts @@ -0,0 +1,132 @@ +import { bench, describe } from "vitest"; +import type { StreamItem, ToolCallItem } from "@/types/stream"; +import { + prepareToolCallHistory, + projectToolCallDetailLevel, +} from "@/tool-calls/detail-level/projection"; +import { findMountedWindowStart, getMountedRecentStreamItems } from "./history-window"; +import { layoutStream } from "./layout"; +import { buildAgentStreamRenderModel } from "./model"; +import { resolveStreamRenderStrategy } from "./strategy-resolver"; + +// One coalesced delta replaces the stream head. This is the pure work AgentStreamView +// does per delta on a long timeline, without React: +// npx vitest bench --project unit src/agent-stream/stream-tick.bench.ts + +const TIMELINE_ITEMS = 2_000; +const TICKS = 200; + +function timestamp(seed: number): Date { + return new Date(1_700_000_000_000 + seed * 1000); +} + +function toolCall(id: string, seed: number): ToolCallItem { + return { + kind: "tool_call", + id, + timestamp: timestamp(seed), + payload: { + source: "agent", + data: { + provider: "claude", + callId: id, + name: "shell", + status: "completed", + error: null, + detail: { type: "shell", command: id }, + }, + }, + }; +} + +function buildTimeline(count: number): StreamItem[] { + const items: StreamItem[] = []; + let seed = 0; + while (items.length < count) { + const turn = items.length; + items.push({ + kind: "user_message", + id: `u${turn}`, + text: "prompt", + timestamp: timestamp(seed++), + }); + items.push(toolCall(`t${turn}a`, seed++)); + items.push(toolCall(`t${turn}b`, seed++)); + items.push({ + kind: "assistant_message", + id: `a${turn}`, + text: "answer", + timestamp: timestamp(seed++), + }); + } + return items.slice(0, count); +} + +function buildHeads(base: StreamItem[], ticks: number): StreamItem[][] { + let text = ""; + const heads: StreamItem[][] = []; + for (let tick = 0; tick < ticks; tick += 1) { + text += "token "; + heads.push([ + { + kind: "assistant_message", + id: "live", + text, + timestamp: timestamp(base.length + tick), + }, + ]); + } + return heads; +} + +function runTicks(input: { tail: StreamItem[]; historyStart: number; heads: StreamItem[][] }) { + const { tail, historyStart, heads } = input; + const strategy = resolveStreamRenderStrategy({ platform: "web", isMobileBreakpoint: false }); + const prepared = prepareToolCallHistory("overview", tail); + const boundaryId = tail[historyStart]?.id ?? null; + const activeTurnStartedAt = tail.at(-1)?.timestamp ?? null; + for (const head of heads) { + const projected = projectToolCallDetailLevel({ + level: "overview", + tail, + head, + preparedHistory: prepared, + isTurnActive: true, + }); + const boundaryIndex = boundaryId + ? projected.tail.findIndex((item) => item.id === boundaryId) + : -1; + const model = buildAgentStreamRenderModel({ + isTurnActive: true, + activeTurnStartedAt, + tail: projected.tail, + head: projected.head, + platform: "web", + isMobileBreakpoint: false, + historyStart: boundaryIndex >= 0 ? boundaryIndex : historyStart, + }); + layoutStream({ + strategy, + isTurnActive: true, + history: model.history, + liveHead: model.segments.liveHead, + timingByAssistantId: model.turnTiming.byAssistantId, + }); + } +} + +const tail = buildTimeline(TIMELINE_ITEMS); +const heads = buildHeads(tail, TICKS); +const defaultWindowStart = findMountedWindowStart({ + items: tail, + minMountedCount: getMountedRecentStreamItems(), +}); + +describe(`${TICKS} streamed deltas over a ${TIMELINE_ITEMS}-item timeline`, () => { + bench("default mounted window (recent items only)", () => { + runTicks({ tail, historyStart: defaultWindowStart, heads }); + }); + bench("history fully revealed (scrolled to top)", () => { + runTicks({ tail, historyStart: 0, heads }); + }); +}); diff --git a/packages/app/src/timeline/turn-time.test.ts b/packages/app/src/timeline/turn-time.test.ts index 35dacf397b..3ce6942301 100644 --- a/packages/app/src/timeline/turn-time.test.ts +++ b/packages/app/src/timeline/turn-time.test.ts @@ -12,11 +12,11 @@ function user(id: string, timestamp: Date): StreamItem { }; } -function assistant(id: string, timestamp: Date): StreamItem { +function assistant(id: string, timestamp: Date, text = id): StreamItem { return { kind: "assistant_message", id, - text: id, + text, timestamp, }; } @@ -125,4 +125,97 @@ describe("deriveStreamTurnTiming", () => { durationMs: null, }); }); + + it("keeps the timing map identity while a streamed delta grows the live head", () => { + const tail = [ + user("u1", new Date("2026-05-15T00:00:00.000Z")), + assistant("a1", new Date("2026-05-15T00:00:07.000Z")), + user("u2", new Date("2026-05-15T00:01:00.000Z")), + ]; + const derive = (text: string, at: string) => + deriveStreamTurnTiming({ + isTurnActive: true, + activeTurnStartedAt: tail[2]?.timestamp ?? null, + tail, + head: [assistant("a2", new Date(at), text)], + }); + + const first = derive("one", "2026-05-15T00:01:04.000Z"); + const second = derive("one two", "2026-05-15T00:01:05.000Z"); + + assert.equal(second.byAssistantId, first.byAssistantId); + assert.deepEqual(first.byAssistantId.get("a1"), { + completedAt: tail[1]?.timestamp, + durationMs: 7000, + }); + assert.equal(first.byAssistantId.has("a2"), false); + }); + + it("keeps the timing map identity when the head closed the tail's open turn", () => { + const lastTailAt = new Date("2026-05-15T00:00:07.000Z"); + const tail = [user("u1", new Date("2026-05-15T00:00:00.000Z")), assistant("a1", lastTailAt)]; + const derive = (text: string, at: string) => + deriveStreamTurnTiming({ + isTurnActive: true, + activeTurnStartedAt: null, + tail, + head: [ + user("u2", new Date("2026-05-15T00:01:00.000Z")), + assistant("a2", new Date(at), text), + ], + }); + + const first = derive("one", "2026-05-15T00:01:04.000Z"); + const second = derive("one two", "2026-05-15T00:01:05.000Z"); + + assert.equal(second.byAssistantId, first.byAssistantId); + assert.deepEqual(first.byAssistantId.get("a1"), { + completedAt: lastTailAt, + durationMs: 7000, + }); + }); + + it("produces a new timing map when the streamed turn completes", () => { + const tail = [user("u1", new Date("2026-05-15T00:00:00.000Z"))]; + const completedAt = new Date("2026-05-15T00:00:09.000Z"); + const head = [assistant("a1", completedAt)]; + + const active = deriveStreamTurnTiming({ + isTurnActive: true, + activeTurnStartedAt: tail[0]?.timestamp ?? null, + tail, + head, + }); + const completed = deriveStreamTurnTiming({ + isTurnActive: false, + activeTurnStartedAt: null, + tail, + head, + }); + + assert.notEqual(completed.byAssistantId, active.byAssistantId); + assert.equal(active.byAssistantId.has("a1"), false); + assert.deepEqual(completed.byAssistantId.get("a1"), { completedAt, durationMs: 9000 }); + }); + + it("derives fresh timing when the tail itself changes", () => { + const firstTail = [ + user("u1", new Date("2026-05-15T00:00:00.000Z")), + assistant("a1", new Date("2026-05-15T00:00:07.000Z")), + user("u2", new Date("2026-05-15T00:01:00.000Z")), + ]; + const secondTail = [...firstTail, assistant("a2", new Date("2026-05-15T00:01:03.000Z"))]; + const derive = (tail: StreamItem[]) => + deriveStreamTurnTiming({ isTurnActive: false, activeTurnStartedAt: null, tail, head: [] }); + + const first = derive(firstTail); + const second = derive(secondTail); + + assert.notEqual(second.byAssistantId, first.byAssistantId); + assert.equal(first.byAssistantId.has("a2"), false); + assert.deepEqual(second.byAssistantId.get("a2"), { + completedAt: secondTail[3]?.timestamp, + durationMs: 3000, + }); + }); }); diff --git a/packages/app/src/timeline/turn-time.ts b/packages/app/src/timeline/turn-time.ts index 4542962af4..c47a6983cd 100644 --- a/packages/app/src/timeline/turn-time.ts +++ b/packages/app/src/timeline/turn-time.ts @@ -11,61 +11,135 @@ export interface StreamTurnTiming { runningStartedAt: Date | null; } +interface OpenTurn { + userAt: Date | null; + lastItemAt: Date | null; + assistantIds: readonly string[]; + previousItem: StreamItem | null; +} + +interface HeadTurnTiming { + additions: Map; + byAssistantId: Map; +} + +interface TailTurnTiming { + completed: Map; + openTurn: OpenTurn; + lastHead: HeadTurnTiming | null; +} + +const EMPTY_OPEN_TURN: OpenTurn = { + userAt: null, + lastItemAt: null, + assistantIds: [], + previousItem: null, +}; + +function closeTurn(turn: OpenTurn, into: Map): void { + if (!turn.lastItemAt || turn.assistantIds.length === 0) { + return; + } + const timing: TurnTiming = { + completedAt: turn.lastItemAt, + durationMs: turn.userAt ? Math.max(0, turn.lastItemAt.getTime() - turn.userAt.getTime()) : null, + }; + for (const id of turn.assistantIds) { + into.set(id, timing); + } +} + +function walkTurns( + items: readonly StreamItem[], + from: OpenTurn, + into: Map, +): OpenTurn { + let userAt = from.userAt; + let lastItemAt = from.lastItemAt; + let assistantIds = [...from.assistantIds]; + let previousItem = from.previousItem; + for (const item of items) { + if (startsNewTurn(item, previousItem)) { + closeTurn({ userAt, lastItemAt, assistantIds, previousItem }, into); + userAt = item.kind === "user_message" ? item.timestamp : null; + lastItemAt = null; + assistantIds = []; + } + lastItemAt = item.timestamp; + if (item.kind === "assistant_message") { + assistantIds.push(item.id); + } + previousItem = item; + } + return { userAt, lastItemAt, assistantIds, previousItem }; +} + +function areTimingsEqual(a: Map, b: Map): boolean { + if (a.size !== b.size) { + return false; + } + for (const [id, timing] of a) { + const other = b.get(id); + if ( + !other || + other.completedAt.getTime() !== timing.completedAt.getTime() || + other.durationMs !== timing.durationMs + ) { + return false; + } + } + return true; +} + +// The tail walk is cached by array identity so a streamed delta, which only replaces the +// head, walks the head from the carried open turn. The returned Map keeps its identity +// while the head adds no completed turn, so memos keyed on it do not miss per tick. +const tailTurnTimingCache = new WeakMap(); + +function getTailTurnTiming(tail: StreamItem[]): TailTurnTiming { + const cached = tailTurnTimingCache.get(tail); + if (cached) { + return cached; + } + const completed = new Map(); + const openTurn = walkTurns(tail, EMPTY_OPEN_TURN, completed); + const timing: TailTurnTiming = { completed, openTurn, lastHead: null }; + tailTurnTimingCache.set(tail, timing); + return timing; +} + +function resolveByAssistantId( + tail: TailTurnTiming, + additions: Map, +): Map { + if (additions.size === 0) { + return tail.completed; + } + if (tail.lastHead && areTimingsEqual(tail.lastHead.additions, additions)) { + return tail.lastHead.byAssistantId; + } + const byAssistantId = new Map(tail.completed); + for (const [id, timing] of additions) { + byAssistantId.set(id, timing); + } + tail.lastHead = { additions, byAssistantId }; + return byAssistantId; +} + export function deriveStreamTurnTiming(params: { isTurnActive: boolean; activeTurnStartedAt: Date | null; tail: StreamItem[]; head: StreamItem[]; }): StreamTurnTiming { - const byAssistantId = new Map(); - let currentUserAt: Date | null = null; - let currentLastItemAt: Date | null = null; - let currentAssistantIds: string[] = []; - let previousItem: StreamItem | null = null; - - const flushCompletedTurn = () => { - if (!currentLastItemAt || currentAssistantIds.length === 0) { - return; - } - const timing: TurnTiming = { - completedAt: currentLastItemAt, - durationMs: currentUserAt - ? Math.max(0, currentLastItemAt.getTime() - currentUserAt.getTime()) - : null, - }; - for (const id of currentAssistantIds) { - byAssistantId.set(id, timing); - } - }; - - const visitItem = (item: StreamItem) => { - if (startsNewTurn(item, previousItem)) { - flushCompletedTurn(); - currentUserAt = item.kind === "user_message" ? item.timestamp : null; - currentLastItemAt = null; - currentAssistantIds = []; - } - currentLastItemAt = item.timestamp; - if (item.kind === "assistant_message") { - currentAssistantIds.push(item.id); - } - previousItem = item; - }; - - for (const item of params.tail) { - visitItem(item); - } - for (const item of params.head) { - visitItem(item); - } - - const runningStartedAt = params.isTurnActive ? params.activeTurnStartedAt : null; + const tail = getTailTurnTiming(params.tail); + const additions = new Map(); + const openTurn = walkTurns(params.head, tail.openTurn, additions); if (!params.isTurnActive) { - flushCompletedTurn(); + closeTurn(openTurn, additions); } - return { - byAssistantId, - runningStartedAt, + byAssistantId: resolveByAssistantId(tail, additions), + runningStartedAt: params.isTurnActive ? params.activeTurnStartedAt : null, }; }