diff --git a/CLAUDE.md b/CLAUDE.md index d22e2b2381..ddb63d268d 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -21,46 +21,47 @@ This is an npm workspace monorepo: At the start of non-trivial work, list `docs/` and skim anything relevant to the task. -| Doc | What's in it | -| ------------------------------------------------------------------ | ------------------------------------------------------------------------------------------------------------------------------ | -| [docs/product.md](docs/product.md) | What Paseo is, who it's for, where it's going | -| [docs/architecture.md](docs/architecture.md) | System design, package layering, WebSocket protocol, agent lifecycle, data flow | -| [docs/agent-lifecycle.md](docs/agent-lifecycle.md) | Agent states, parent/child relationships, archive semantics, tabs vs archive, subagents track | -| [docs/data-model.md](docs/data-model.md) | File-based JSON persistence, Zod schemas, atomic writes, no migrations | -| [docs/glossary.md](docs/glossary.md) | Authoritative terminology — UI label wins, no synonyms | -| [docs/coding-standards.md](docs/coding-standards.md) | Type hygiene, error handling, state design, React patterns, file organization | -| [docs/design.md](docs/design.md) | Design system — tokens, buttons, hierarchy, density, alignment rails, states, what's forbidden | -| [docs/forms.md](docs/forms.md) | Form architecture — non-React form model, form kit, load-state gating; the schedule form is the golden example | -| [docs/hover.md](docs/hover.md) | Hover — the canonical pattern (plain View + onPointerEnter/Leave, separate inner Pressable) and the three ways agents break it | -| [docs/unistyles.md](docs/unistyles.md) | Unistyles gotchas — `useUnistyles()` is forbidden, alternatives in order | -| [docs/floating-panels.md](docs/floating-panels.md) | Anchored popovers — Portal/Modal escape for Android, lifecycle gates, keyboard-shared-value, status-bar offset, the flash | -| [docs/menus.md](docs/menus.md) | The menu engine — popover vs sheet, submenu pages, hover intent, when a decision earns a submenu | -| [docs/expo-router.md](docs/expo-router.md) | Expo Router route ownership, startup restore, and native blank-screen gotchas | -| [docs/file-icons.md](docs/file-icons.md) | Material icon theme integration for the file explorer | -| [docs/providers.md](docs/providers.md) | Adding a new agent provider end-to-end | -| [docs/forge-providers.md](docs/forge-providers.md) | Adding a git forge: registry/manifest, drop-in checklist, self-host/GHES, the two facts tiers | -| [docs/custom-providers.md](docs/custom-providers.md) | Custom provider config: Z.AI, Alibaba/Qwen, ACP agents, profiles, custom binaries | -| [docs/plugins.md](docs/plugins.md) | Local plugin manifest, directory source config, RPCs, native surfaces, and attachment sources | -| [docs/service-proxy.md](docs/service-proxy.md) | Service proxy: exposing workspace scripts at public URLs, DNS setup, reverse proxy config | -| [docs/development.md](docs/development.md) | Dev server, build sync gotchas, CLI reference, agent state, Playwright MCP | -| [docs/rpc-namespacing.md](docs/rpc-namespacing.md) | WebSocket RPC naming convention — dotted namespaces and `.request`/`.response` pairs | -| [docs/protocol-compatibility.md](docs/protocol-compatibility.md) | Why app/daemon versions drift, protocol vs feature contract, capability gating, COMPAT tagging | -| [docs/protocol-validation.md](docs/protocol-validation.md) | zod-aot generated inbound WebSocket validation, patched compiler regressions, schema-purity rules | -| [docs/terminal-performance.md](docs/terminal-performance.md) | Terminal latency pipeline, coalescing/backpressure invariants, benchmark + perf spec usage | -| [docs/file-observation.md](docs/file-observation.md) | Recursive watcher ownership, Linux constraints, teardown invariants, and Parcel comparison | -| [docs/testing.md](docs/testing.md) | TDD workflow, determinism, real dependencies over mocks, test organization | -| [docs/qa.md](docs/qa.md) | QA evidence bar for pull requests — platform matrix, version drift, performance, UI proof | -| [docs/mobile-testing.md](docs/mobile-testing.md) | Maestro and mobile test workflows | -| [docs/mobile-panels.md](docs/mobile-panels.md) | Compact left/center/right panel ownership, worklet motion, gesture revisions, and Fabric constraints | -| [docs/explorer-sidebar.md](docs/explorer-sidebar.md) | Explorer sidebar and ordinary side-pane host contracts, lifecycle, placement, and routing preferences | -| [docs/ad-hoc-daemon-testing.md](docs/ad-hoc-daemon-testing.md) | Isolated in-process daemon test harness | -| [docs/browser-capture-harness.md](docs/browser-capture-harness.md) | Real-Electron browser screenshot harness and compositor-surface gotcha | -| [docs/android.md](docs/android.md) | App variants, local/cloud builds, EAS workflows | -| [docs/docker.md](docs/docker.md) | Running the daemon and bundled web UI in Docker, volumes, agent images, security | -| [docs/release.md](docs/release.md) | Release playbook, draft releases, completion checklist | -| [docs/terminal-activity.md](docs/terminal-activity.md) | Terminal activity indicators — source-agnostic tracker, agent hook reporting, adding a new hook provider | -| [SECURITY.md](SECURITY.md) | Relay threat model, E2E encryption, DNS rebinding, agent auth | -| [public-docs/hub/security.md](public-docs/hub/security.md) | Public Hub guide — trust boundaries, untrusted triggers, provider controls, and output authority | +| Doc | What's in it | +| -------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------ | +| [docs/product.md](docs/product.md) | What Paseo is, who it's for, where it's going | +| [docs/architecture.md](docs/architecture.md) | System design, package layering, WebSocket protocol, agent lifecycle, data flow | +| [docs/agent-lifecycle.md](docs/agent-lifecycle.md) | Agent states, parent/child relationships, archive semantics, tabs vs archive, subagents track | +| [docs/data-model.md](docs/data-model.md) | File-based JSON persistence, Zod schemas, atomic writes, no migrations | +| [docs/glossary.md](docs/glossary.md) | Authoritative terminology — UI label wins, no synonyms | +| [docs/coding-standards.md](docs/coding-standards.md) | Type hygiene, error handling, state design, React patterns, file organization | +| [docs/design.md](docs/design.md) | Design system — tokens, buttons, hierarchy, density, alignment rails, states, what's forbidden | +| [docs/forms.md](docs/forms.md) | Form architecture — non-React form model, form kit, load-state gating; the schedule form is the golden example | +| [docs/hover.md](docs/hover.md) | Hover — the canonical pattern (plain View + onPointerEnter/Leave, separate inner Pressable) and the three ways agents break it | +| [docs/unistyles.md](docs/unistyles.md) | Unistyles gotchas — `useUnistyles()` is forbidden, alternatives in order | +| [docs/floating-panels.md](docs/floating-panels.md) | Anchored popovers — Portal/Modal escape for Android, lifecycle gates, keyboard-shared-value, status-bar offset, the flash | +| [docs/menus.md](docs/menus.md) | The menu engine — popover vs sheet, submenu pages, hover intent, when a decision earns a submenu | +| [docs/expo-router.md](docs/expo-router.md) | Expo Router route ownership, startup restore, and native blank-screen gotchas | +| [docs/file-icons.md](docs/file-icons.md) | Material icon theme integration for the file explorer | +| [docs/providers.md](docs/providers.md) | Adding a new agent provider end-to-end | +| [docs/forge-providers.md](docs/forge-providers.md) | Adding a git forge: registry/manifest, drop-in checklist, self-host/GHES, the two facts tiers | +| [docs/custom-providers.md](docs/custom-providers.md) | Custom provider config: Z.AI, Alibaba/Qwen, ACP agents, profiles, custom binaries | +| [docs/plugins.md](docs/plugins.md) | Local plugin manifest, directory source config, RPCs, native surfaces, and attachment sources | +| [docs/service-proxy.md](docs/service-proxy.md) | Service proxy: exposing workspace scripts at public URLs, DNS setup, reverse proxy config | +| [docs/development.md](docs/development.md) | Dev server, build sync gotchas, CLI reference, agent state, Playwright MCP | +| [docs/rpc-namespacing.md](docs/rpc-namespacing.md) | WebSocket RPC naming convention — dotted namespaces and `.request`/`.response` pairs | +| [docs/protocol-compatibility.md](docs/protocol-compatibility.md) | Why app/daemon versions drift, protocol vs feature contract, capability gating, COMPAT tagging | +| [docs/protocol-validation.md](docs/protocol-validation.md) | zod-aot generated inbound WebSocket validation, patched compiler regressions, schema-purity rules | +| [docs/terminal-performance.md](docs/terminal-performance.md) | Terminal latency pipeline, coalescing/backpressure invariants, benchmark + perf spec usage | +| [docs/agent-stream-performance.md](docs/agent-stream-performance.md) | Assistant text pipeline — coalescing window, paced reveal, why arrival lumps are smoothed at render | +| [docs/file-observation.md](docs/file-observation.md) | Recursive watcher ownership, Linux constraints, teardown invariants, and Parcel comparison | +| [docs/testing.md](docs/testing.md) | TDD workflow, determinism, real dependencies over mocks, test organization | +| [docs/qa.md](docs/qa.md) | QA evidence bar for pull requests — platform matrix, version drift, performance, UI proof | +| [docs/mobile-testing.md](docs/mobile-testing.md) | Maestro and mobile test workflows | +| [docs/mobile-panels.md](docs/mobile-panels.md) | Compact left/center/right panel ownership, worklet motion, gesture revisions, and Fabric constraints | +| [docs/explorer-sidebar.md](docs/explorer-sidebar.md) | Explorer sidebar and ordinary side-pane host contracts, lifecycle, placement, and routing preferences | +| [docs/ad-hoc-daemon-testing.md](docs/ad-hoc-daemon-testing.md) | Isolated in-process daemon test harness | +| [docs/browser-capture-harness.md](docs/browser-capture-harness.md) | Real-Electron browser screenshot harness and compositor-surface gotcha | +| [docs/android.md](docs/android.md) | App variants, local/cloud builds, EAS workflows | +| [docs/docker.md](docs/docker.md) | Running the daemon and bundled web UI in Docker, volumes, agent images, security | +| [docs/release.md](docs/release.md) | Release playbook, draft releases, completion checklist | +| [docs/terminal-activity.md](docs/terminal-activity.md) | Terminal activity indicators — source-agnostic tracker, agent hook reporting, adding a new hook provider | +| [SECURITY.md](SECURITY.md) | Relay threat model, E2E encryption, DNS rebinding, agent auth | +| [public-docs/hub/security.md](public-docs/hub/security.md) | Public Hub guide — trust boundaries, untrusted triggers, provider controls, and output authority | ### Writing docs diff --git a/docs/agent-stream-performance.md b/docs/agent-stream-performance.md new file mode 100644 index 0000000000..95e0c23706 --- /dev/null +++ b/docs/agent-stream-performance.md @@ -0,0 +1,52 @@ +# Agent stream performance + +How assistant text gets from a provider to the screen, and why it is paced on the way. Read this before changing `packages/server/src/server/agent/agent-stream-coalescer.ts`, the reducer queue in `packages/app/src/timeline/session-stream-reducers.ts`, or the reveal in `packages/app/src/hooks/use-revealed-text.ts`. + +For terminal output, which is a separate pipeline with separate budgets, see [terminal-performance.md](terminal-performance.md). + +## The pipeline + +``` +provider deltas (every provider streams incrementally) + → AgentStreamCoalescer (daemon, leading + trailing, ≤1 message per 60ms per agent) + → recordTimeline: one canonical row per flushed item + → agent_stream ws message + → reducer queue (app, one commit per frame) → session store + → paced reveal (app, per assistant/reasoning item) → markdown blocks → paint +``` + +Every provider delivers incremental text, so there is no provider that needs special handling: Claude via `includePartialMessages`, Codex via `agent_message_delta`, ACP agents via `agent_message_chunk`, Pi and OMP via `text_delta`. + +## Why the reveal is paced + +Arrival is lumpy and there is no fixing that at the source. A 60ms coalescing window carries however many characters the model produced in those 60ms, which swings by an order of magnitude within a single turn. Painting each delta as it lands makes the size of those lumps visible, and that is what reads as jagged. + +So arrival sets a _target_ and the reveal rate is derived from the backlog instead. A burst makes the text catch up faster; it does not make the text jump. Shrinking the coalescing window does not fix this — it makes the lumps smaller and more frequent, at the cost of message rate on a daemon loop that already contends with terminal frames and per-message relay encryption. + +## Invariants + +- **The coalescer is leading + trailing.** The first delta after an idle window flushes synchronously; only the rest of the burst waits for the trailing timer. Reverting to trailing-only adds a full window to the first character of every turn. Same shape and the same reason as `TerminalOutputCoalescer`. +- **The leading flush adds a canonical row, and that is fine.** A burst's first chunk lands as its own timeline row. `mergeAssistantChunks` / `mergeReasoningChunks` in `timeline-projection.ts` join contiguous same-turn rows, and clients read the projected timeline, so history is unaffected. Tests that assert on raw rows have to account for the extra row; tests that assert on what a client sees do not. +- **The store holds the full text; only the rendered slice is paced.** Copy, selection, the chat outline, and scroll geometry all read the same string the user can see. Pacing the store instead would leave the bottom anchor chasing a content height that is ahead of the reveal. +- **First sight of a text is revealed whole.** Only growth is paced. This is what makes history hydration, timeline replay, a virtualized row remounting on scroll, and an already-finished message all render complete on first paint without a special case for each. +- **Leaving `phase: "streaming"` snaps the reveal.** A completed turn must never be left holding characters. `layoutStream` sets the phase, so anything outside the live head with an active turn is already complete. +- **The reducer queue commits on a frame, with a timer as the ceiling.** A frame callback never fires in a hidden tab, so a timer races it and wins when nothing is painting — the store has to keep advancing either way. + +## 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. + +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: + +| | paint on arrival | paced | +| -------------------------------- | ---------------- | ----- | +| frames that advanced the text | 6% | 87% | +| chars-per-frame CV | 4.11 | 1.86 | +| gap between visible updates, p50 | 317ms | 17ms | +| gap between visible updates, p95 | 383ms | 17ms | + +Total characters painted is roughly the same either way — the reveal changes when they land, not how many arrive. + +Measure the **total** length across every `assistant-message` element, not the last one. A turn emits many assistant messages, so the tail element keeps changing identity and its length is not monotonic; sampling only the tail reads those handovers as resets and reports almost no growth. `sampleStreamFrames` does this correctly. diff --git a/docs/terminal-performance.md b/docs/terminal-performance.md index a5bda20ee2..7e65482add 100644 --- a/docs/terminal-performance.md +++ b/docs/terminal-performance.md @@ -43,6 +43,6 @@ Terminal frames share the daemon main event loop with all agent traffic. The `ev ## Known remaining contention (follow-up candidates) -- A single large `agent_stream` message (e.g. a 250KB diff payload) measurably delays terminal echo (~100ms-class dips) — cost is split between daemon serialization and app-side parse/render on the shared browser main thread. +- A single large `agent_stream` message (e.g. a 250KB diff payload) measurably delays terminal echo (~100ms-class dips) — cost is split between daemon serialization and app-side parse/render on the shared browser main thread. See [agent-stream-performance.md](agent-stream-performance.md) for that pipeline's own budgets. - Relay-attached clients pay pure-JS tweetnacl encryption on the daemon main loop (`packages/relay/src/encrypted-channel.ts`). Negotiated binary application frames stay binary ciphertext and avoid base64 encode/decode; text and mixed-version traffic remain base64 WebSocket text frames. - `sendToClient` re-stringifies session messages per socket; only matters for multi-socket connections. diff --git a/packages/app/e2e/browser/agent-stream-smoothness.spec.ts b/packages/app/e2e/browser/agent-stream-smoothness.spec.ts new file mode 100644 index 0000000000..74069d3561 --- /dev/null +++ b/packages/app/e2e/browser/agent-stream-smoothness.spec.ts @@ -0,0 +1,151 @@ +import { test, expect } from "../support/fixtures"; +import { awaitAssistantMessage } from "../support/helpers/agent-stream"; +import { startRunningMockAgent } from "../support/helpers/composer"; +import { + sampleStreamFrames, + summarizeStreamSmoothness, +} from "../support/helpers/stream-smoothness"; + +/** + * Streaming smoothness, measured on the path the user actually sees. + * + * The mock provider's "bursty-stream" model emits tokens in uneven runs separated + * by idle gaps, which is how real models arrive. Sampling and the statistics live + * in ../support/helpers/stream-smoothness.ts, which also documents what the two + * numbers mean. + * + * Budgets are deliberately loose. This spec exists to catch a regression in the + * shape of the stream, not to pin exact numbers on shared CI hardware. For + * reference, on a local dev daemon the paced reveal measures a CV around 1.3–1.5 + * against 2.7 when painting deltas as they arrive. + */ + +const SAMPLE_WINDOW_MS = 6_000; +const CHARS_PER_FRAME_CV_BUDGET = 2; +const UPDATE_GAP_P95_BUDGET_MS = 250; +const REACT_COMMIT_P95_BUDGET_MS = 12; +const REACT_COMMIT_COUNT_BUDGET = 600; +const RUN_AGENT_STREAM_PERF = process.env.PASEO_AGENT_STREAM_PERF_E2E === "1"; +const agentStreamPerfDescribe = RUN_AGENT_STREAM_PERF ? test.describe : test.describe.skip; + +interface ReactCommit { + id: string; + phase: "mount" | "update" | "nested-update"; + actualDuration: number; + baseDuration: number; + startTime: number; + commitTime: number; +} + +agentStreamPerfDescribe("Agent stream smoothness", () => { + test("reveals bursty model output at a steady rate", async ({ page }, testInfo) => { + test.setTimeout(120_000); + await page.addInitScript(() => { + Reflect.set(globalThis, "__PASEO_RENDER_PROFILE_ENABLED__", true); + }); + + const agent = await startRunningMockAgent(page, { + prefix: "stream-smoothness-", + model: "bursty-stream", + prompt: "Stream bursty output for smoothness measurement.", + }); + try { + await awaitAssistantMessage(page); + await page.evaluate(() => { + const reset = Reflect.get(globalThis, "__PASEO_RESET_RENDER_PROFILE__"); + if (typeof reset !== "function") { + throw new Error("Render profiler did not initialize"); + } + reset(); + }); + + const samples = await sampleStreamFrames(page, { sampleWindowMs: SAMPLE_WINDOW_MS }); + const report = summarizeStreamSmoothness(samples, SAMPLE_WINDOW_MS); + const reactCommits = await readStreamReactCommits(page); + const reactReport = summarizeReactCommits(reactCommits); + + await testInfo.attach("agent-stream-smoothness-report", { + body: JSON.stringify(report, null, 2), + contentType: "application/json", + }); + await testInfo.attach("agent-stream-react-profile", { + body: JSON.stringify(reactReport, null, 2), + contentType: "application/json", + }); + console.log(`[perf] Agent stream React: ${JSON.stringify(reactReport)}`); + console.log(`[perf] Agent stream smoothness: ${JSON.stringify(report)}`); + + expect( + report.growthFrames, + "the stream must actually produce text during the sample window", + ).toBeGreaterThan(20); + expect( + report.charsPerFrameCv, + `characters painted per frame should be steady (cv < ${CHARS_PER_FRAME_CV_BUDGET})`, + ).toBeLessThan(CHARS_PER_FRAME_CV_BUDGET); + expect( + report.updateGapP95Ms, + `the text should not stall (p95 gap < ${UPDATE_GAP_P95_BUDGET_MS}ms)`, + ).toBeLessThan(UPDATE_GAP_P95_BUDGET_MS); + expect(reactReport.count, "streaming should produce measured React commits").toBeGreaterThan( + 20, + ); + expect( + reactReport.count, + `the reveal should cap commits below ${REACT_COMMIT_COUNT_BUDGET} per sample window`, + ).toBeLessThan(REACT_COMMIT_COUNT_BUDGET); + expect( + reactReport.actualDurationP95Ms, + `React commits should stay below ${REACT_COMMIT_P95_BUDGET_MS}ms at p95`, + ).toBeLessThan(REACT_COMMIT_P95_BUDGET_MS); + } finally { + await agent.cleanup(); + } + }); +}); + +async function readStreamReactCommits( + page: Parameters[0], +): Promise { + return await page.evaluate(() => { + const samples = Reflect.get(globalThis, "__PASEO_RENDER_PROFILE__"); + if (!Array.isArray(samples)) { + return []; + } + return samples.filter( + (sample): sample is ReactCommit => + typeof sample === "object" && + sample !== null && + typeof Reflect.get(sample, "id") === "string" && + Reflect.get(sample, "id").startsWith("AgentStreamSection:"), + ); + }); +} + +function summarizeReactCommits(commits: readonly ReactCommit[]) { + const durations = commits + .map((commit) => commit.actualDuration) + .sort((left, right) => left - right); + const total = durations.reduce((sum, duration) => sum + duration, 0); + return { + count: commits.length, + actualDurationTotalMs: round2(total), + actualDurationMeanMs: round2(commits.length > 0 ? total / commits.length : 0), + actualDurationP50Ms: percentile(durations, 50), + actualDurationP95Ms: percentile(durations, 95), + actualDurationMaxMs: durations.at(-1) ?? 0, + }; +} + +function round2(value: number): number { + return Math.round(value * 100) / 100; +} + +function percentile(sorted: readonly number[], percentileValue: number): number { + if (sorted.length === 0) { + return 0; + } + return sorted[ + Math.min(sorted.length - 1, Math.floor((sorted.length - 1) * (percentileValue / 100))) + ]; +} diff --git a/packages/app/e2e/support/helpers/stream-smoothness.test.ts b/packages/app/e2e/support/helpers/stream-smoothness.test.ts new file mode 100644 index 0000000000..f57b848e88 --- /dev/null +++ b/packages/app/e2e/support/helpers/stream-smoothness.test.ts @@ -0,0 +1,31 @@ +import { describe, expect, it } from "vitest"; + +import { summarizeStreamSmoothness } from "./stream-smoothness"; + +describe("summarizeStreamSmoothness", () => { + it("reports canonical handover snaps separately from paced growth", () => { + const report = summarizeStreamSmoothness( + [ + { timestampMs: 0, growthChars: 0, firstPaintChars: 0 }, + { timestampMs: 10, growthChars: 4, firstPaintChars: 0 }, + { timestampMs: 20, growthChars: 100, firstPaintChars: 30 }, + { timestampMs: 30, growthChars: 4, firstPaintChars: 0 }, + { timestampMs: 40, growthChars: 0, firstPaintChars: 0 }, + ], + 40, + ); + + expect(report).toMatchObject({ + frames: 3, + growthFrames: 2, + charsPainted: 8, + maxCharsInOneFrame: 4, + firstPaintFrames: 1, + firstPaintChars: 30, + handoverGrowthFrames: 1, + handoverGrowthChars: 100, + maxHandoverGrowthCharsInOneFrame: 100, + updateGapP95Ms: 10, + }); + }); +}); diff --git a/packages/app/e2e/support/helpers/stream-smoothness.ts b/packages/app/e2e/support/helpers/stream-smoothness.ts new file mode 100644 index 0000000000..f2d5a577ac --- /dev/null +++ b/packages/app/e2e/support/helpers/stream-smoothness.ts @@ -0,0 +1,215 @@ +import type { Page } from "@playwright/test"; + +/** + * Frame-level sampling of the streaming assistant message, and the two numbers + * that describe how it reads: + * + * - `charsPerFrameCv` — coefficient of variation of characters painted per frame. + * This is the smoothness metric. Painting deltas as they land tracks the + * arrival lumps, so it is high; pacing the reveal flattens it. + * - `updateGapP95Ms` — how long the text can sit unchanged mid-stream. This is + * the stall metric, and it is what stops smoothness being bought with lag. + * + * Both are needed. A stream that never updates is perfectly smooth. + */ + +export interface StreamFrameSample { + timestampMs: number; + growthChars: number; + firstPaintChars: number; +} + +export interface StreamSmoothnessReport { + sampleWindowMs: number; + frames: number; + growthFrames: number; + charsPainted: number; + meanCharsPerFrame: number; + maxCharsInOneFrame: number; + charsPerFrameCv: number; + firstPaintFrames: number; + firstPaintChars: number; + maxFirstPaintCharsInOneFrame: number; + handoverGrowthFrames: number; + handoverGrowthChars: number; + maxHandoverGrowthCharsInOneFrame: number; + updateGapP50Ms: number; + updateGapP95Ms: number; + updateGapMaxMs: number; +} + +/** + * Record text growth on every mounted assistant message once per frame. + * + * Message occurrence identity separates paced growth from first paint. Completed markdown + * blocks move into canonical timeline rows while a turn streams, remounting text + * that was already visible. Counting the new node's full length as growth would + * report that handover as a large reveal even though no text appeared on screen. + * First paints stay in the report as a separate diagnostic. + * + * Runs entirely in the page so the sampling clock is the same clock the reveal + * runs on. Waits for the text to start moving first, so a slow model's + * time-to-first-token is not counted as a stall. + */ +export async function sampleStreamFrames( + page: Page, + options: { sampleWindowMs: number; startTimeoutMs?: number }, +): Promise { + const startTimeoutMs = options.startTimeoutMs ?? 60_000; + return await page.evaluate( + async ({ sampleWindowMs, startTimeoutMs: waitMs }) => { + const readLength = () => { + let total = 0; + for (const node of document.querySelectorAll('[data-testid="assistant-message"]')) { + total += node.textContent?.length ?? 0; + } + return total; + }; + + const waitStart = performance.now(); + const base = readLength(); + await new Promise((resolve) => { + const poll = () => { + if (readLength() > base + 20 || performance.now() - waitStart > waitMs) { + resolve(); + return; + } + requestAnimationFrame(poll); + }; + requestAnimationFrame(poll); + }); + + let previousLengths = new Map(); + for (const node of document.querySelectorAll('[data-testid="assistant-message"]')) { + const key = node.getAttribute("data-reveal-key"); + const length = Number(node.getAttribute("data-reveal-length")); + if (key !== null && Number.isFinite(length)) { + previousLengths.set(key, length); + } + } + + const samples: Array<{ + timestampMs: number; + growthChars: number; + firstPaintChars: number; + }> = []; + await new Promise((resolve) => { + const start = performance.now(); + const tick = (now: number) => { + const currentLengths = new Map(); + let growthChars = 0; + let firstPaintChars = 0; + for (const node of document.querySelectorAll('[data-testid="assistant-message"]')) { + const key = node.getAttribute("data-reveal-key"); + const currentLength = Number(node.getAttribute("data-reveal-length")); + if (key === null || !Number.isFinite(currentLength)) { + continue; + } + const previousLength = previousLengths.get(key); + if (previousLength === undefined || currentLength < previousLength) { + firstPaintChars += currentLength; + } else { + const nodeGrowth = currentLength - previousLength; + growthChars += nodeGrowth; + } + currentLengths.set(key, currentLength); + } + samples.push({ + timestampMs: now, + growthChars, + firstPaintChars, + }); + previousLengths = currentLengths; + if (now - start >= sampleWindowMs) { + resolve(); + return; + } + requestAnimationFrame(tick); + }; + requestAnimationFrame(tick); + }); + return samples; + }, + { sampleWindowMs: options.sampleWindowMs, startTimeoutMs }, + ); +} + +export function computePercentile(samples: readonly number[], percentile: number): number { + if (samples.length === 0) { + return 0; + } + const sorted = [...samples].sort((a, b) => a - b); + return sorted[Math.max(0, Math.ceil((percentile / 100) * sorted.length) - 1)]; +} + +function round2(value: number): number { + return Math.round(value * 100) / 100; +} + +/** + * Reduce frame samples to the report. Frames outside the streaming window are + * dropped: leading idle is time-to-first-token, and trailing idle means the turn + * ended mid-sample. First-paint frames are structural handovers. Existing blocks + * may also snap complete on those frames, so their growth is reported separately + * from the paced-growth CV. + */ +export function summarizeStreamSmoothness( + samples: readonly StreamFrameSample[], + sampleWindowMs: number, +): StreamSmoothnessReport { + const deltas: number[] = []; + const gapsMs: number[] = []; + let lastUpdateAtMs: number | null = null; + let firstGrowth = -1; + let lastGrowth = -1; + const handoverGrowths: number[] = []; + + for (let index = 0; index < samples.length; index += 1) { + const sample = samples[index]; + const isHandover = sample.firstPaintChars > 0; + const growth = isHandover ? 0 : sample.growthChars; + if (isHandover && sample.growthChars > 0) { + handoverGrowths.push(sample.growthChars); + lastUpdateAtMs = sample.timestampMs; + } + deltas.push(growth); + if (growth > 0) { + if (firstGrowth === -1) { + firstGrowth = deltas.length - 1; + } + lastGrowth = deltas.length - 1; + if (lastUpdateAtMs !== null) { + gapsMs.push(samples[index].timestampMs - lastUpdateAtMs); + } + lastUpdateAtMs = samples[index].timestampMs; + } + } + + const active = firstGrowth === -1 ? [] : deltas.slice(firstGrowth, lastGrowth + 1); + const charsPainted = active.reduce((sum, delta) => sum + delta, 0); + const mean = active.length > 0 ? charsPainted / active.length : 0; + const variance = + active.length > 0 + ? active.reduce((sum, delta) => sum + (delta - mean) ** 2, 0) / active.length + : 0; + const firstPaints = samples.map((sample) => sample.firstPaintChars).filter((chars) => chars > 0); + + return { + sampleWindowMs, + frames: active.length, + growthFrames: active.filter((delta) => delta > 0).length, + charsPainted, + meanCharsPerFrame: round2(mean), + maxCharsInOneFrame: active.length > 0 ? Math.max(...active) : 0, + charsPerFrameCv: mean === 0 ? Number.POSITIVE_INFINITY : round2(Math.sqrt(variance) / mean), + firstPaintFrames: firstPaints.length, + firstPaintChars: firstPaints.reduce((sum, chars) => sum + chars, 0), + maxFirstPaintCharsInOneFrame: firstPaints.length > 0 ? Math.max(...firstPaints) : 0, + handoverGrowthFrames: handoverGrowths.length, + handoverGrowthChars: handoverGrowths.reduce((sum, chars) => sum + chars, 0), + maxHandoverGrowthCharsInOneFrame: handoverGrowths.length > 0 ? Math.max(...handoverGrowths) : 0, + updateGapP50Ms: round2(computePercentile(gapsMs, 50)), + updateGapP95Ms: round2(computePercentile(gapsMs, 95)), + updateGapMaxMs: gapsMs.length > 0 ? round2(Math.max(...gapsMs)) : 0, + }; +} diff --git a/packages/app/package.json b/packages/app/package.json index 1bf3f369f8..319cce38bd 100644 --- a/packages/app/package.json +++ b/packages/app/package.json @@ -23,6 +23,7 @@ "test:browser": "vitest run --project browser", "test:e2e": "playwright test --project=browser", "test:e2e:diff-performance": "cross-env PASEO_DIFF_PERF_E2E=1 playwright test --project=browser e2e/browser/diff-performance.spec.ts", + "test:e2e:stream-smoothness": "cross-env PASEO_AGENT_STREAM_PERF_E2E=1 playwright test --project=browser e2e/browser/agent-stream-smoothness.spec.ts", "test:e2e:real": "cross-env E2E_FORK_PASEO_HOME_FROM=../../.dev/paseo-home playwright test --project=real-provider", "test:e2e:relay-deployment": "npm --prefix ../.. run build:server && npm --prefix ../.. run build:daemon-web-ui && cross-env E2E_FORK_PASEO_HOME_FROM=../../.dev/paseo-home playwright test --project=relay-deployment e2e/browser/relay-deployment-reconnect.real.spec.ts", "test:e2e:ui": "playwright test --ui", diff --git a/packages/app/src/agent-stream/text-reveal.test.ts b/packages/app/src/agent-stream/text-reveal.test.ts new file mode 100644 index 0000000000..b6d88f1108 --- /dev/null +++ b/packages/app/src/agent-stream/text-reveal.test.ts @@ -0,0 +1,277 @@ +import { describe, expect, it } from "vitest"; + +import { + advanceTextReveal, + beginTextReveal, + clampToSafeRevealBoundary, + completeTextReveal, + computeRevealStep, + nextTextRevealFrame, + isTextRevealSettled, + retargetTextReveal, + TEXT_REVEAL_FRAME_INTERVAL_MS, + TEXT_REVEAL_HORIZON_MS, + visibleRevealedText, +} from "./text-reveal"; + +describe("computeRevealStep", () => { + it("reveals nothing when there is no backlog", () => { + expect(computeRevealStep({ backlog: 0, elapsedMs: 16 })).toBe(0); + expect(computeRevealStep({ backlog: -5, elapsedMs: 16 })).toBe(0); + }); + + it("drains the backlog over the horizon", () => { + const step = computeRevealStep({ backlog: 300, elapsedMs: TEXT_REVEAL_HORIZON_MS / 2 }); + expect(step).toBe(150); + }); + + it("reveals the whole backlog once a frame covers the horizon", () => { + expect(computeRevealStep({ backlog: 42, elapsedMs: TEXT_REVEAL_HORIZON_MS })).toBe(42); + expect(computeRevealStep({ backlog: 42, elapsedMs: TEXT_REVEAL_HORIZON_MS * 10 })).toBe(42); + }); + + it("scales the rate with the backlog so a faster model reveals faster", () => { + const slow = computeRevealStep({ backlog: 20, elapsedMs: 16 }); + const fast = computeRevealStep({ backlog: 200, elapsedMs: 16 }); + expect(fast).toBeGreaterThan(slow); + }); + + it("always advances at least one character so the tail finishes", () => { + expect(computeRevealStep({ backlog: 1, elapsedMs: 1 })).toBe(1); + expect(computeRevealStep({ backlog: 2, elapsedMs: 0.01 })).toBe(1); + }); + + it("never overshoots the backlog", () => { + for (const backlog of [1, 3, 7, 50]) { + expect(computeRevealStep({ backlog, elapsedMs: 120 })).toBeLessThanOrEqual(backlog); + } + }); + + it("reveals nothing when no time has passed", () => { + expect(computeRevealStep({ backlog: 100, elapsedMs: 0 })).toBe(0); + expect(computeRevealStep({ backlog: 100, elapsedMs: -16 })).toBe(0); + }); + + it("converges within a bounded number of frames at 60Hz", () => { + let remaining = 500; + let frames = 0; + while (remaining > 0) { + remaining -= computeRevealStep({ backlog: remaining, elapsedMs: 16 }); + frames += 1; + expect(frames).toBeLessThan(120); + } + expect(remaining).toBe(0); + }); + + it("keeps per-frame steps smooth for a bursty arrival pattern", () => { + // A lumpy arrival sequence: the daemon's coalescing window carries wildly + // different character counts depending on model throughput. + const arrivals = [4, 180, 6, 240, 2, 90, 300, 8]; + const steps: number[] = []; + let backlog = 0; + + for (const arrival of arrivals) { + backlog += arrival; + // ~60ms of frames between coalesced flushes. + for (let frame = 0; frame < 4; frame += 1) { + const step = computeRevealStep({ backlog, elapsedMs: 16 }); + backlog -= step; + steps.push(step); + } + } + + // Painting arrivals directly would swing from 2 to 300 characters in one + // paint. Pacing keeps the largest single paint far below the largest arrival. + expect(Math.max(...steps)).toBeLessThan(Math.max(...arrivals) / 2); + expect(Math.min(...steps)).toBeGreaterThan(0); + }); +}); + +describe("clampToSafeRevealBoundary", () => { + it("leaves boundaries in plain text alone", () => { + expect(clampToSafeRevealBoundary("hello world", 5)).toBe(5); + }); + + it("clamps the ends", () => { + expect(clampToSafeRevealBoundary("hello", -3)).toBe(0); + expect(clampToSafeRevealBoundary("hello", 99)).toBe(5); + }); + + it("never splits a surrogate pair", () => { + // "a😀b" — the emoji occupies indices 1 and 2. + const text = "a😀b"; + expect(clampToSafeRevealBoundary(text, 2)).toBe(1); + expect(clampToSafeRevealBoundary(text, 3)).toBe(3); + }); + + it("does not strand a combining mark", () => { + // "e" + combining acute; cutting at 1 would show a bare "e" then reflow. + const text = `éx`; + expect(clampToSafeRevealBoundary(text, 1)).toBe(0); + expect(clampToSafeRevealBoundary(text, 2)).toBe(2); + }); + + it("does not split an emoji ZWJ sequence", () => { + // Family emoji: person ZWJ person ZWJ child. + const family = "\u{1F468}‍\u{1F469}‍\u{1F467}"; + const text = `${family}!`; + for (let index = 1; index < family.length; index += 1) { + expect(clampToSafeRevealBoundary(text, index)).toBe(0); + } + expect(clampToSafeRevealBoundary(text, family.length)).toBe(family.length); + }); + + it("does not split a skin tone modifier from its base", () => { + const text = "\u{1F44D}\u{1F3FD} nice"; + expect(clampToSafeRevealBoundary(text, 2)).toBe(0); + expect(clampToSafeRevealBoundary(text, 4)).toBe(4); + }); + + it("does not strand a variation selector", () => { + const text = "❤️ done"; + expect(clampToSafeRevealBoundary(text, 1)).toBe(0); + expect(clampToSafeRevealBoundary(text, 2)).toBe(2); + }); + + it("does not split extended grapheme clusters", () => { + for (const { text, boundaries } of [ + { text: "a🇺🇸b", boundaries: [0, 1, 5, 6] }, + { text: "a가b", boundaries: [0, 1, 3, 4] }, + { text: "aकःb", boundaries: [0, 1, 3, 4] }, + ]) { + for (let index = 1; index < text.length; index += 1) { + const expected = boundaries.toReversed().find((boundary) => boundary <= index); + expect(clampToSafeRevealBoundary(text, index)).toBe(expected); + } + } + }); +}); + +describe("nextTextRevealFrame", () => { + it("caps a high-refresh frame clock at 60Hz", () => { + expect(nextTextRevealFrame(null, 0)).toEqual({ + elapsedMs: TEXT_REVEAL_FRAME_INTERVAL_MS, + frameAtMs: 0, + }); + expect(nextTextRevealFrame(0, 8)).toBeNull(); + expect(nextTextRevealFrame(0, 17)).toEqual({ + elapsedMs: 17, + frameAtMs: expect.closeTo(TEXT_REVEAL_FRAME_INTERVAL_MS, 5), + }); + }); +}); + +describe("text reveal state", () => { + it("reveals the first text it sees in full", () => { + const state = beginTextReveal("already here"); + expect(visibleRevealedText(state)).toBe("already here"); + expect(isTextRevealSettled(state)).toBe(true); + }); + + it("paces growth instead of painting it whole", () => { + let state = beginTextReveal("a"); + state = retargetTextReveal(state, `a${"b".repeat(300)}`); + expect(visibleRevealedText(state)).toBe("a"); + + state = advanceTextReveal(state, 16); + const shown = visibleRevealedText(state); + expect(shown.length).toBeGreaterThan(1); + expect(shown.length).toBeLessThan(301); + }); + + it("catches up to the full text across frames", () => { + const full = `a${"b".repeat(200)}`; + let state = retargetTextReveal(beginTextReveal("a"), full); + for (let frame = 0; frame < 60 && !isTextRevealSettled(state); frame += 1) { + state = advanceTextReveal(state, 16); + } + expect(isTextRevealSettled(state)).toBe(true); + expect(visibleRevealedText(state)).toBe(full); + }); + + it("releases everything when the turn completes", () => { + const full = `a${"b".repeat(500)}`; + let state = advanceTextReveal(retargetTextReveal(beginTextReveal("a"), full), 16); + expect(visibleRevealedText(state)).not.toBe(full); + + state = completeTextReveal(state); + expect(visibleRevealedText(state)).toBe(full); + }); + + it("only ever paints a prefix of the real text", () => { + const full = "the quick brown fox jumps over the lazy dog"; + let state = retargetTextReveal(beginTextReveal("the "), full); + for (let frame = 0; frame < 20; frame += 1) { + expect(full.startsWith(visibleRevealedText(state))).toBe(true); + state = advanceTextReveal(state, 16); + } + }); + + it("clamps when the slot switches to a shorter message", () => { + const state = retargetTextReveal(beginTextReveal("a long streaming message"), "short"); + expect(visibleRevealedText(state)).toBe("short"); + }); + + it("never paints a lone surrogate while streaming emoji", () => { + const full = `progress ${"\u{1F680}".repeat(40)} done`; + let state = retargetTextReveal(beginTextReveal("progress "), full); + for (let frame = 0; frame < 200 && !isTextRevealSettled(state); frame += 1) { + const shown = visibleRevealedText(state); + // A stranded surrogate would make the painted prefix un-decodable. + expect(shown).toBe(Array.from(shown).join("")); + expect(full.startsWith(shown)).toBe(true); + state = advanceTextReveal(state, 16); + } + expect(visibleRevealedText(completeTextReveal(state))).toBe(full); + }); + + it("a long cluster cannot stall the reveal", () => { + // The rendered boundary backs up over this cluster, but the counter does not, + // so the reveal still finishes. + const full = `x${"́".repeat(40)}y`; + let state = retargetTextReveal(beginTextReveal("x"), full); + for (let frame = 0; frame < 200 && !isTextRevealSettled(state); frame += 1) { + state = advanceTextReveal(state, 16); + } + expect(isTextRevealSettled(state)).toBe(true); + expect(visibleRevealedText(state)).toBe(full); + }); +}); + +describe("trailing fragments from arrival boundaries", () => { + it("holds back half a flag while more text is coming", () => { + // The daemon's coalescing window can end a delta mid-cluster, so a caught-up + // reveal can be handed a single regional indicator. + const halfFlag = "flags: \u{1F1EC}"; + const state = beginTextReveal(halfFlag); + expect(visibleRevealedText(state, { streaming: true })).toBe("flags: "); + // Once the turn ends nothing more is coming, so paint what there is. + expect(visibleRevealedText(state, { streaming: false })).toBe(halfFlag); + }); + + it("paints the flag once its second half arrives", () => { + const whole = "flags: \u{1F1EC}\u{1F1E7}"; + expect(visibleRevealedText(beginTextReveal(whole), { streaming: true })).toBe(whole); + }); + + it("keeps an even run of regional indicators intact", () => { + const twoFlags = "\u{1F1EC}\u{1F1E7}\u{1F1EB}\u{1F1F7}"; + expect(visibleRevealedText(beginTextReveal(twoFlags), { streaming: true })).toBe(twoFlags); + }); + + it("holds back a dangling joiner", () => { + expect( + visibleRevealedText(beginTextReveal("family: \u{1F468}\u200D"), { streaming: true }), + ).toBe("family: \u{1F468}"); + }); + + it("holds back a dangling high surrogate", () => { + const loneHigh = `broken: ${String.fromCharCode(0xd83d)}`; + expect(visibleRevealedText(beginTextReveal(loneHigh), { streaming: true })).toBe("broken: "); + }); + + it("leaves ordinary trailing text alone", () => { + const prose = "just some prose"; + expect(visibleRevealedText(beginTextReveal(prose), { streaming: true })).toBe(prose); + }); +}); diff --git a/packages/app/src/agent-stream/text-reveal.ts b/packages/app/src/agent-stream/text-reveal.ts new file mode 100644 index 0000000000..5c965f543b --- /dev/null +++ b/packages/app/src/agent-stream/text-reveal.ts @@ -0,0 +1,240 @@ +/** + * Paced reveal for streaming assistant text. + * + * Deltas arrive lumpy: the daemon coalesces one message per 60ms window, carrying + * however many characters the model happened to produce in that window. Painting + * each delta as it lands makes the size of those lumps visible, which is what + * reads as jagged. So arrival changes the *target* and the reveal rate is derived + * from the backlog instead — a burst makes the text catch up faster, it doesn't + * make it jump. + * + * The store keeps the full text. Only the rendered slice is paced, so copy, + * selection, the chat outline, and scroll geometry all stay consistent with what + * is on screen. + * + * Everything here is pure. The only thing the React hook adds is a frame clock, + * which keeps the policy testable without a renderer. + */ + +// Backlog is drained over this horizon. Shorter feels more like the raw arrival +// pattern; longer adds lag the user can notice at the end of a turn. +export const TEXT_REVEAL_HORIZON_MS = 150; + +/** Reveal at most once per 60Hz frame, even on high-refresh displays. */ +export const TEXT_REVEAL_FRAME_INTERVAL_MS = 1000 / 60; + +// A frame's elapsed time is clamped to this before it is used, so a long stall +// (backgrounded tab, blocked main thread) doesn't produce a wild step from one +// enormous delta. +const MAX_ELAPSED_MS = 250; + +/** + * Characters to reveal on this frame. Proportional to the backlog, so the reveal + * accelerates when the model runs ahead and settles when it doesn't, with a + * one-character floor so the tail always finishes. + */ +export function computeRevealStep(input: { + backlog: number; + elapsedMs: number; + horizonMs?: number; +}): number { + const { backlog } = input; + if (backlog <= 0) { + return 0; + } + + const horizonMs = input.horizonMs ?? TEXT_REVEAL_HORIZON_MS; + if (horizonMs <= 0) { + return backlog; + } + + const elapsedMs = Math.min(Math.max(input.elapsedMs, 0), MAX_ELAPSED_MS); + if (elapsedMs <= 0) { + return 0; + } + if (elapsedMs >= horizonMs) { + return backlog; + } + + const step = Math.ceil((backlog * elapsedMs) / horizonMs); + return Math.min(backlog, Math.max(1, step)); +} + +const ZERO_WIDTH_JOINER = 0x200d; + +const graphemeSegmenter = + typeof Intl.Segmenter === "function" + ? new Intl.Segmenter(undefined, { granularity: "grapheme" }) + : null; + +/** + * Pacing needs a conformant grapheme segmenter. Older runtimes paint each + * arrival whole instead of risking a transient broken glyph. + */ +export function isTextRevealPacingSupported(): boolean { + return graphemeSegmenter !== null; +} + +/** + * Pull a cut index back to somewhere it is safe to slice. + * + * A raw `slice` at an arbitrary index can split a grapheme cluster, which makes + * one visible glyph briefly render as separate parts while the text is streaming. + * + * This only moves the *rendered* boundary. The reveal counter stays monotonic, so + * a long cluster can never stall the reveal — the next frame steps past it. + */ +export function clampToSafeRevealBoundary(text: string, index: number): number { + if (index <= 0) { + return 0; + } + if (index >= text.length) { + return text.length; + } + + const segment = graphemeSegmenter?.segment(text).containing(index); + return segment?.index ?? 0; +} + +export interface TextRevealFrame { + elapsedMs: number; + frameAtMs: number; +} + +/** + * Keep reveal commits at 60Hz while carrying timing remainder forward so a + * high-refresh display does not make the reveal render at hardware frame rate. + */ +export function nextTextRevealFrame( + previousFrameAtMs: number | null, + timestampMs: number, +): TextRevealFrame | null { + const elapsedMs = + previousFrameAtMs === null ? TEXT_REVEAL_FRAME_INTERVAL_MS : timestampMs - previousFrameAtMs; + if (elapsedMs < TEXT_REVEAL_FRAME_INTERVAL_MS) { + return null; + } + return { + elapsedMs, + frameAtMs: timestampMs - (elapsedMs % TEXT_REVEAL_FRAME_INTERVAL_MS), + }; +} + +export interface TextRevealState { + /** The full text as the store knows it. */ + readonly target: string; + /** How much of it has been released. Monotonic within a message. */ + readonly revealed: number; +} + +/** + * First sight of a text is revealed whole. Only growth is paced, which is what + * makes history hydration, timeline replay, a virtualized row remounting on + * scroll, and an already-finished message all render complete on first paint + * without a special case for each. + */ +export function beginTextReveal(text: string): TextRevealState { + return { target: text, revealed: text.length }; +} + +/** Point the reveal at newly arrived text without changing how much is shown. */ +export function retargetTextReveal(state: TextRevealState, text: string): TextRevealState { + if (state.target === text) { + return state; + } + // A shorter string means this slot is showing a different message than the one + // the reveal position belongs to. + return { target: text, revealed: Math.min(state.revealed, text.length) }; +} + +/** Release one frame's worth of characters. */ +export function advanceTextReveal( + state: TextRevealState, + elapsedMs: number, + horizonMs?: number, +): TextRevealState { + const step = computeRevealStep({ + backlog: state.target.length - state.revealed, + elapsedMs, + ...(horizonMs !== undefined ? { horizonMs } : {}), + }); + if (step <= 0) { + return state; + } + return { target: state.target, revealed: Math.min(state.target.length, state.revealed + step) }; +} + +/** Release everything, for when the turn ends and nothing may be held back. */ +export function completeTextReveal(state: TextRevealState): TextRevealState { + if (state.revealed >= state.target.length) { + return state; + } + return { target: state.target, revealed: state.target.length }; +} + +export function isTextRevealSettled(state: TextRevealState): boolean { + return state.revealed >= state.target.length; +} + +/** + * Hold back a trailing fragment that cannot stand on its own. + * + * The reveal boundary is not the only place a cluster gets cut. The daemon's + * coalescing window ends a delta wherever the window closes, so a caught-up + * reveal can be handed half a flag and paint it — measured on a mixed-script + * stream as frames ending in a single regional indicator. Only fragments that + * are certainly waiting on more input are withheld: a dangling high surrogate, a + * trailing joiner, or an odd regional indicator. + */ +function trimIncompleteTrailingCluster(text: string): string { + if (text.length === 0) { + return text; + } + + const lastUnit = text.charCodeAt(text.length - 1); + if (lastUnit >= 0xd800 && lastUnit <= 0xdbff) { + return text.slice(0, -1); + } + if (lastUnit === ZERO_WIDTH_JOINER) { + return text.slice(0, -1); + } + + const trailingRegionalIndicators = countTrailingRegionalIndicators(text); + if (trailingRegionalIndicators % 2 === 1) { + return text.slice(0, -2); + } + + return text; +} + +/** Number of regional indicators at the end of `text`. A flag is a pair. */ +function countTrailingRegionalIndicators(text: string): number { + let count = 0; + let cursor = text.length; + while (cursor >= 2) { + const codePoint = text.codePointAt(cursor - 2); + if (codePoint === undefined || codePoint < 0x1f1e6 || codePoint > 0x1f1ff) { + break; + } + count += 1; + cursor -= 2; + } + return count; +} + +/** + * What should actually be painted this frame. + * + * `streaming` says whether more text is still expected. While it is, a trailing + * fragment waits for the rest of its cluster; once the turn ends everything + * received is painted, because nothing more is coming. + */ +export function visibleRevealedText( + state: TextRevealState, + options?: { streaming?: boolean }, +): string { + if (state.revealed >= state.target.length) { + return options?.streaming ? trimIncompleteTrailingCluster(state.target) : state.target; + } + return state.target.slice(0, clampToSafeRevealBoundary(state.target, state.revealed)); +} diff --git a/packages/app/src/agent-stream/view.tsx b/packages/app/src/agent-stream/view.tsx index 4651710ace..e88b621762 100644 --- a/packages/app/src/agent-stream/view.tsx +++ b/packages/app/src/agent-stream/view.tsx @@ -51,6 +51,7 @@ import type { } from "@getpaseo/protocol/agent-types"; import type { AgentScreenAgent } from "@/hooks/use-agent-screen-state-machine"; import { useSessionStore } from "@/stores/session-store"; +import { useRevealedText } from "@/hooks/use-revealed-text"; import { useFileExplorerActions } from "@/hooks/use-file-explorer-actions"; import { useLoadOlderAgentHistory } from "@/hooks/use-load-older-agent-history"; import { useSettings } from "@/hooks/use-settings"; @@ -717,15 +718,13 @@ const AgentStreamViewComponent = forwardRef) => { return ( - ); }, @@ -1230,6 +1229,39 @@ interface ToolCallSlotProps extends Omit< onInlineDetailsExpandedChangeByItemId: (itemId: string, expanded: boolean) => void; } +interface ThoughtSlotProps { + itemId: string; + onInlineDetailsExpandedChangeByItemId: (itemId: string, expanded: boolean) => void; + text: string; + status: Extract["status"]; + isLastInSequence: boolean; + defaultExpanded: boolean; +} + +// Reasoning text is paced the same way assistant text is; see @/hooks/use-revealed-text. +function ThoughtSlot({ + itemId, + onInlineDetailsExpandedChangeByItemId, + text, + status, + isLastInSequence, + defaultExpanded, +}: ThoughtSlotProps) { + const revealedText = useRevealedText(text, status === "ready" ? "complete" : "streaming"); + return ( + + ); +} + function ToolCallSlot({ itemId, onInlineDetailsExpandedChangeByItemId, diff --git a/packages/app/src/components/message.tsx b/packages/app/src/components/message.tsx index ac7493b5b5..9b1b90528d 100644 --- a/packages/app/src/components/message.tsx +++ b/packages/app/src/components/message.tsx @@ -75,12 +75,14 @@ import { HighlightedCodeBlock } from "@/components/highlighted-code-block"; import { MarkdownFenceBlock } from "@/components/markdown/fence"; import type { MarkdownPhase } from "@/components/markdown/fence/types"; import { splitMarkdownBlocks } from "@/utils/split-markdown-blocks"; +import { useRevealedText } from "@/hooks/use-revealed-text"; import { colorMarkdownLinkChildren } from "@/components/markdown/link-children"; import { createAssistantMarkdownParser } from "@/utils/assistant-markdown-parser"; import { formatDuration, formatMessageTimestamp } from "@/utils/time"; import { writeMarkdownToRichClipboard } from "@/utils/rich-clipboard"; import { getDefaultMarkdownClipboardEnvironment } from "@/utils/rich-clipboard-default-environment"; import { setAssistantMarkdownBlockHeight } from "@/utils/assistant-message-height-estimate"; +import { isRenderProfileEnabled } from "@/utils/render-profiler"; import { getAgentAttachmentPillContent } from "@/attachments/attachment-pill-content"; import { PlanCard } from "./plan-card"; import { useToolCallSheet } from "./tool-call-sheet"; @@ -1459,6 +1461,9 @@ export const AssistantMessage = memo(function AssistantMessage({ phase, }: AssistantMessageProps) { const markdownParser = useMemo(createAssistantMarkdownParser, []); + // Paint a paced prefix while the turn is streaming so text arrives at a steady + // rate instead of in whatever lumps the daemon's coalescing window produced. + const revealedMessage = useRevealedText(message, phase); const fileLinkActions = useAssistantFileLinkActions(); const handleMarkdownLinkPress = useStableEvent((url: string) => { @@ -1897,7 +1902,7 @@ export const AssistantMessage = memo(function AssistantMessage({ }; }, [client, fileLinkActions, markdownParser, occurrenceKey, phase, serverId, workspaceRoot]); - const blocks = useMemo(() => splitMarkdownBlocks(message), [message]); + const blocks = useMemo(() => splitMarkdownBlocks(revealedMessage), [revealedMessage]); const keyedBlocks = useMemo( () => blocks.map((block, index) => ({ key: `block:${index}`, block })), [blocks], @@ -1913,9 +1918,16 @@ export const AssistantMessage = memo(function AssistantMessage({ ], [spacing], ); + const revealDataSet = useMemo( + () => + isRenderProfileEnabled() + ? { revealKey: occurrenceKey, revealLength: String(revealedMessage.length) } + : undefined, + [occurrenceKey, revealedMessage.length], + ); return ( - + {keyedBlocks.map(({ key, block }, index) => ( (beginTextReveal(text)); + const [, forceRender] = useState(0); + const frameRef = useRef(null); + const lastFrameAtRef = useRef(null); + + stateRef.current = retargetTextReveal(stateRef.current, text); + + useEffect(() => { + const settle = () => { + lastFrameAtRef.current = null; + const next = completeTextReveal(stateRef.current); + if (next !== stateRef.current) { + stateRef.current = next; + forceRender((tick) => tick + 1); + } + }; + + if (!pacingSupported || phase !== "streaming") { + settle(); + return; + } + if (isTextRevealSettled(stateRef.current)) { + lastFrameAtRef.current = null; + return; + } + if (typeof requestAnimationFrame !== "function") { + settle(); + return; + } + + const tick = (timestamp: number) => { + frameRef.current = null; + const frame = nextTextRevealFrame(lastFrameAtRef.current, timestamp); + if (!frame) { + frameRef.current = requestAnimationFrame(tick); + return; + } + lastFrameAtRef.current = frame.frameAtMs; + + const next = advanceTextReveal(stateRef.current, frame.elapsedMs); + if (next !== stateRef.current) { + stateRef.current = next; + forceRender((count) => count + 1); + } + if (!isTextRevealSettled(stateRef.current)) { + frameRef.current = requestAnimationFrame(tick); + } + }; + + frameRef.current = requestAnimationFrame(tick); + return () => { + if (frameRef.current !== null) { + cancelAnimationFrame(frameRef.current); + frameRef.current = null; + } + }; + }, [text, pacingSupported, phase]); + + return pacingSupported + ? visibleRevealedText(stateRef.current, { streaming: phase === "streaming" }) + : text; +} diff --git a/packages/app/src/timeline/session-stream-reducers.ts b/packages/app/src/timeline/session-stream-reducers.ts index 263d929741..70e005ee26 100644 --- a/packages/app/src/timeline/session-stream-reducers.ts +++ b/packages/app/src/timeline/session-stream-reducers.ts @@ -13,6 +13,8 @@ import { upsertUserMessageAcrossStream, } from "@/types/stream"; +// Ceiling for a pending commit. A frame callback normally gets there first; this +// is the fallback for when nothing is painting. const AGENT_STREAM_REDUCER_FLUSH_DELAY_MS = 16 * 3; // --------------------------------------------------------------------------- @@ -1833,12 +1835,52 @@ export interface CreateSessionAgentStreamReducerQueueInput { recoverTimelineGap: (agentId: string, cursor: { epoch: string; endSeq: number }) => void; } +interface ScheduledReducerFlush { + frameId: number | null; + timerId: ReturnType; +} + +const scheduledReducerFlushes = new Map(); +let nextScheduledReducerFlushId = 1; + +function clearScheduledReducerFlush(handle: ScheduledReducerFlush): void { + if (handle.frameId !== null) { + cancelAnimationFrame(handle.frameId); + } + clearTimeout(handle.timerId); +} + +// Commit deltas on a frame boundary so text lands in step with paint instead of on +// an arbitrary timer that drifts on and off the display beat. A frame callback +// never fires in a hidden tab, so a timer races it and wins when nothing is +// painting — the store has to keep advancing either way. function scheduleAgentStreamReducerFlush(callback: () => void): number { - return setTimeout(callback, AGENT_STREAM_REDUCER_FLUSH_DELAY_MS) as unknown as number; + const id = nextScheduledReducerFlushId; + nextScheduledReducerFlushId += 1; + + const run = () => { + const handle = scheduledReducerFlushes.get(id); + if (!handle) { + return; + } + scheduledReducerFlushes.delete(id); + clearScheduledReducerFlush(handle); + callback(); + }; + + const timerId = setTimeout(run, AGENT_STREAM_REDUCER_FLUSH_DELAY_MS); + const frameId = typeof requestAnimationFrame === "function" ? requestAnimationFrame(run) : null; + scheduledReducerFlushes.set(id, { frameId, timerId }); + return id; } function cancelAgentStreamReducerFlush(id: number) { - clearTimeout(id); + const handle = scheduledReducerFlushes.get(id); + if (!handle) { + return; + } + scheduledReducerFlushes.delete(id); + clearScheduledReducerFlush(handle); } export function createSessionAgentStreamReducerQueue( diff --git a/packages/app/src/utils/render-profiler.tsx b/packages/app/src/utils/render-profiler.tsx index 0144ba67ac..cefecffde2 100644 --- a/packages/app/src/utils/render-profiler.tsx +++ b/packages/app/src/utils/render-profiler.tsx @@ -13,6 +13,7 @@ declare global { var __PASEO_RENDER_PROFILE__: RenderProfileSample[] | undefined; var __PASEO_RENDER_PROFILE_REASONS__: Record> | undefined; var __PASEO_RESET_RENDER_PROFILE__: (() => void) | undefined; + var __PASEO_RENDER_PROFILE_ENABLED__: boolean | undefined; } function getSearchParam(name: string): string | null { @@ -27,8 +28,10 @@ function getSearchParam(name: string): string | null { return new URLSearchParams(search).get(name); } -function isRenderProfileEnabled(): boolean { - return getSearchParam("renderProfile") === "1"; +export function isRenderProfileEnabled(): boolean { + return ( + globalThis.__PASEO_RENDER_PROFILE_ENABLED__ === true || getSearchParam("renderProfile") === "1" + ); } const onRender: ProfilerOnRenderCallback = ( diff --git a/packages/server/src/server/agent/agent-manager-stream-coalescing.test.ts b/packages/server/src/server/agent/agent-manager-stream-coalescing.test.ts index 6d3be2f224..d3fb7c4b58 100644 --- a/packages/server/src/server/agent/agent-manager-stream-coalescing.test.ts +++ b/packages/server/src/server/agent/agent-manager-stream-coalescing.test.ts @@ -7,6 +7,7 @@ import { createTestLogger } from "../../test-utils/test-logger.js"; import { AgentManager, type AgentManagerEvent } from "./agent-manager.js"; import { AGENT_STREAM_COALESCE_DEFAULT_WINDOW_MS } from "./agent-stream-coalescer.js"; import type { AgentTimelineRow } from "./agent-timeline-store-types.js"; +import { projectTimelineRows } from "./timeline-projection.js"; import type { AgentCapabilityFlags, AgentClient, @@ -357,6 +358,13 @@ function getTimelineItems(rows: AgentTimelineRow[]): AgentTimelineItem[] { return rows.map((row) => row.item); } +// The leading-edge flush emits the first chunk of a burst as its own row. Clients +// read the projected timeline, where mergeAssistantChunks/mergeReasoningChunks +// join those contiguous rows back together, so history reads the same either way. +function getProjectedTimelineItems(rows: AgentTimelineRow[]): AgentTimelineItem[] { + return projectTimelineRows({ rows, mode: "projected" }).map((entry) => entry.item); +} + function expectContiguousRowSeqs(rows: AgentTimelineRow[], expected: number[]): void { expect(rows.map((row) => row.seq)).toEqual(expected); } @@ -505,29 +513,40 @@ describe("target coalesced behavior", () => { } await waitForSessionEventQueue(); + // Leading edge: the first chunk is already out, the other 999 are buffered. await vi.advanceTimersByTimeAsync(BEFORE_COALESCE_WINDOW_MS); - expect(await harness.manager.getTimelineRows(agentId)).toHaveLength(0); - expect(getTimelineStreamEvents(harness.events, agentId)).toHaveLength(0); + expect(getTimelineItems(await harness.manager.getTimelineRows(agentId))).toEqual([ + { type: "assistant_message", text: "x" }, + ]); + expect(getTimelineStreamEvents(harness.events, agentId)).toHaveLength(1); await vi.advanceTimersByTimeAsync(1); const rows = await harness.manager.getTimelineRows(agentId); const events = getTimelineStreamEvents(harness.events, agentId); - expect(rows).toHaveLength(1); - expect(events).toHaveLength(1); - expect(rows[0]?.item).toEqual({ - type: "assistant_message", - text: "x".repeat(1000), - }); + expect(rows).toHaveLength(2); + expect(events).toHaveLength(2); + expect(getTimelineItems(rows)).toEqual([ + { type: "assistant_message", text: "x" }, + { type: "assistant_message", text: "x".repeat(999) }, + ]); expect(events.map((event) => (event.type === "agent_stream" ? event.event : null))).toEqual([ { type: "timeline", provider: "codex", - item: { type: "assistant_message", text: "x".repeat(1000) }, + item: { type: "assistant_message", text: "x" }, + }, + { + type: "timeline", + provider: "codex", + item: { type: "assistant_message", text: "x".repeat(999) }, }, ]); - expectContiguousRowSeqs(rows, [1]); - expectContiguousLiveSeqs(events, [1]); + expectContiguousRowSeqs(rows, [1, 2]); + expectContiguousLiveSeqs(events, [1, 2]); + expect(getProjectedTimelineItems(rows)).toEqual([ + { type: "assistant_message", text: "x".repeat(1000) }, + ]); } finally { harness.cleanup(); } @@ -548,10 +567,16 @@ describe("target coalesced behavior", () => { const rows = await harness.manager.getTimelineRows(agentId); const events = getTimelineStreamEvents(harness.events, agentId); - expect(getTimelineItems(rows)).toEqual([{ type: "reasoning", text: "r".repeat(100) }]); - expect(events).toHaveLength(1); - expectContiguousRowSeqs(rows, [1]); - expectContiguousLiveSeqs(events, [1]); + expect(getTimelineItems(rows)).toEqual([ + { type: "reasoning", text: "r" }, + { type: "reasoning", text: "r".repeat(99) }, + ]); + expect(events).toHaveLength(2); + expectContiguousRowSeqs(rows, [1, 2]); + expectContiguousLiveSeqs(events, [1, 2]); + expect(getProjectedTimelineItems(rows)).toEqual([ + { type: "reasoning", text: "r".repeat(100) }, + ]); } finally { harness.cleanup(); } @@ -570,34 +595,47 @@ describe("target coalesced behavior", () => { session.pushEvent(timelineEvent(runningToolCall)); await waitForSessionEventQueue(); - expect(await harness.manager.getTimelineRows(agentId)).toHaveLength(0); - expect(getTimelineStreamEvents(harness.events, agentId)).toHaveLength(0); + // Only the leading chunk is out; the running tool call stays buffered with + // the rest of the text rather than forcing an early flush. + expect(getTimelineItems(await harness.manager.getTimelineRows(agentId))).toEqual([ + { type: "assistant_message", text: "a" }, + ]); + expect(getTimelineStreamEvents(harness.events, agentId)).toHaveLength(1); await vi.advanceTimersByTimeAsync(COALESCE_WINDOW_MS); expect(getTimelineItems(await harness.manager.getTimelineRows(agentId))).toEqual([ - { type: "assistant_message", text: "a".repeat(500) }, + { type: "assistant_message", text: "a" }, + { type: "assistant_message", text: "a".repeat(499) }, runningToolCall, ]); - expectContiguousLiveSeqs(getTimelineStreamEvents(harness.events, agentId), [1, 2]); + expectContiguousLiveSeqs(getTimelineStreamEvents(harness.events, agentId), [1, 2, 3]); for (let i = 0; i < 500; i++) { session.pushEvent(assistant("b")); } await waitForSessionEventQueue(); + // The "b" burst arrives inside the window that just flushed, so it gets no + // leading flush of its own and coalesces whole. await vi.advanceTimersByTimeAsync(BEFORE_COALESCE_WINDOW_MS); - expect(await harness.manager.getTimelineRows(agentId)).toHaveLength(2); + expect(await harness.manager.getTimelineRows(agentId)).toHaveLength(3); await vi.advanceTimersByTimeAsync(1); const rows = await harness.manager.getTimelineRows(agentId); const events = getTimelineStreamEvents(harness.events, agentId); expect(getTimelineItems(rows)).toEqual([ + { type: "assistant_message", text: "a" }, + { type: "assistant_message", text: "a".repeat(499) }, + runningToolCall, + { type: "assistant_message", text: "b".repeat(500) }, + ]); + expectContiguousRowSeqs(rows, [1, 2, 3, 4]); + expectContiguousLiveSeqs(events, [1, 2, 3, 4]); + expect(getProjectedTimelineItems(rows)).toEqual([ { type: "assistant_message", text: "a".repeat(500) }, runningToolCall, { type: "assistant_message", text: "b".repeat(500) }, ]); - expectContiguousRowSeqs(rows, [1, 2, 3]); - expectContiguousLiveSeqs(events, [1, 2, 3]); } finally { harness.cleanup(); } @@ -614,18 +652,27 @@ describe("target coalesced behavior", () => { } await waitForSessionEventQueue(); + // The leading edge flushes the first snapshot; the remaining 199 collapse to + // the latest one. Both carry the same callId, so the projection folds them + // into a single tool call row. await vi.advanceTimersByTimeAsync(BEFORE_COALESCE_WINDOW_MS); - expect(await harness.manager.getTimelineRows(agentId)).toHaveLength(0); - expect(getTimelineStreamEvents(harness.events, agentId)).toHaveLength(0); + expect(getTimelineItems(await harness.manager.getTimelineRows(agentId))).toEqual([ + toolCall({ output: "chunk-0" }), + ]); + expect(getTimelineStreamEvents(harness.events, agentId)).toHaveLength(1); await vi.advanceTimersByTimeAsync(1); const rows = await harness.manager.getTimelineRows(agentId); const events = getTimelineStreamEvents(harness.events, agentId); - expect(getTimelineItems(rows)).toEqual([toolCall({ output: "chunk-199" })]); - expect(events).toHaveLength(1); - expectContiguousRowSeqs(rows, [1]); - expectContiguousLiveSeqs(events, [1]); + expect(getTimelineItems(rows)).toEqual([ + toolCall({ output: "chunk-0" }), + toolCall({ output: "chunk-199" }), + ]); + expect(events).toHaveLength(2); + expectContiguousRowSeqs(rows, [1, 2]); + expectContiguousLiveSeqs(events, [1, 2]); + expect(getProjectedTimelineItems(rows)).toEqual([toolCall({ output: "chunk-199" })]); } finally { harness.cleanup(); } @@ -648,12 +695,17 @@ describe("target coalesced behavior", () => { const events = getTimelineStreamEvents(harness.events, agentId); expect(getTimelineItems(rows)).toEqual([ + toolCall({ callId: "tool-1", output: "one-a" }), + toolCall({ callId: "tool-2", output: "two-b" }), + toolCall({ callId: "tool-1", output: "one-b" }), + ]); + expect(events).toHaveLength(3); + expectContiguousRowSeqs(rows, [1, 2, 3]); + expectContiguousLiveSeqs(events, [1, 2, 3]); + expect(getProjectedTimelineItems(rows)).toEqual([ toolCall({ callId: "tool-1", output: "one-b" }), toolCall({ callId: "tool-2", output: "two-b" }), ]); - expect(events).toHaveLength(2); - expectContiguousRowSeqs(rows, [1, 2]); - expectContiguousLiveSeqs(events, [1, 2]); } finally { harness.cleanup(); } @@ -668,7 +720,10 @@ describe("target coalesced behavior", () => { session.pushEvent(assistant("before")); session.pushEvent(timelineEvent(toolCall({ output: "running" }))); await waitForSessionEventQueue(); - expect(await harness.manager.getTimelineRows(agentId)).toHaveLength(0); + // "before" led the burst; the running tool call is still buffered. + expect(getTimelineItems(await harness.manager.getTimelineRows(agentId))).toEqual([ + { type: "assistant_message", text: "before" }, + ]); session.pushEvent(timelineEvent(toolCall({ status: "completed", output: "done" }))); await waitForSessionEventQueue(); @@ -821,14 +876,26 @@ describe("target coalesced behavior", () => { const firstEvents = getTimelineStreamEvents(harness.events, first.agentId); const secondEvents = getTimelineStreamEvents(harness.events, second.agentId); - expect(getTimelineItems(firstRows)).toEqual([{ type: "assistant_message", text: "aa" }]); - expect(getTimelineItems(secondRows)).toEqual([{ type: "assistant_message", text: "bb" }]); - expect(firstEvents).toHaveLength(1); - expect(secondEvents).toHaveLength(1); - expectContiguousRowSeqs(firstRows, [1]); - expectContiguousRowSeqs(secondRows, [1]); - expectContiguousLiveSeqs(firstEvents, [1]); - expectContiguousLiveSeqs(secondEvents, [1]); + expect(getTimelineItems(firstRows)).toEqual([ + { type: "assistant_message", text: "a" }, + { type: "assistant_message", text: "a" }, + ]); + expect(getTimelineItems(secondRows)).toEqual([ + { type: "assistant_message", text: "b" }, + { type: "assistant_message", text: "b" }, + ]); + expect(firstEvents).toHaveLength(2); + expect(secondEvents).toHaveLength(2); + expectContiguousRowSeqs(firstRows, [1, 2]); + expectContiguousRowSeqs(secondRows, [1, 2]); + expectContiguousLiveSeqs(firstEvents, [1, 2]); + expectContiguousLiveSeqs(secondEvents, [1, 2]); + expect(getProjectedTimelineItems(firstRows)).toEqual([ + { type: "assistant_message", text: "aa" }, + ]); + expect(getProjectedTimelineItems(secondRows)).toEqual([ + { type: "assistant_message", text: "bb" }, + ]); rmSync(secondWorkdir, { recursive: true, force: true }); } finally { harness.cleanup(); @@ -923,10 +990,17 @@ describe("target coalesced behavior", () => { const rows = await harness.manager.getTimelineRows(agentId); const events = getTimelineStreamEvents(harness.events, agentId); - expect(getTimelineItems(rows)).toEqual([{ type: "assistant_message", text: " \n\tdone" }]); - expect(events).toHaveLength(1); - expectContiguousRowSeqs(rows, [1]); - expectContiguousLiveSeqs(events, [1]); + // The empty chunk is dropped before the coalescer, so " " leads the burst. + expect(getTimelineItems(rows)).toEqual([ + { type: "assistant_message", text: " " }, + { type: "assistant_message", text: "\n\tdone" }, + ]); + expect(events).toHaveLength(2); + expectContiguousRowSeqs(rows, [1, 2]); + expectContiguousLiveSeqs(events, [1, 2]); + expect(getProjectedTimelineItems(rows)).toEqual([ + { type: "assistant_message", text: " \n\tdone" }, + ]); } finally { harness.cleanup(); } @@ -1115,13 +1189,18 @@ describe("target coalesced behavior", () => { const timelineEvents = getTimelineStreamEvents(idempotencyHarness.events, agentId); expect(getTimelineItems(rows)).toEqual([ - { type: "assistant_message", text: "abc" }, + { type: "assistant_message", text: "a" }, + { type: "assistant_message", text: "bc" }, TOOL_CALL, TOOL_CALL, ]); - expect(timelineEvents).toHaveLength(3); - expectContiguousRowSeqs(rows, [1, 2, 3]); - expectContiguousLiveSeqs(timelineEvents, [1, 2, 3]); + expect(timelineEvents).toHaveLength(4); + expectContiguousRowSeqs(rows, [1, 2, 3, 4]); + expectContiguousLiveSeqs(timelineEvents, [1, 2, 3, 4]); + expect(getProjectedTimelineItems(rows)).toEqual([ + { type: "assistant_message", text: "abc" }, + TOOL_CALL, + ]); } finally { idempotencyHarness.cleanup(); } @@ -1150,9 +1229,8 @@ describe("target coalesced behavior", () => { session.pushEvent(assistant("A")); await waitForSessionEventQueue(); - await vi.advanceTimersByTimeAsync(COALESCE_WINDOW_MS); - await waitForSessionEventQueue(); - + // "A" flushes on the leading edge; the chunk the subscriber pushes during + // that flush must land in the next window, not corrupt this one. const rowsAfterFirstFlush = await reentryHarness.manager.getTimelineRows(agentId); expect(getTimelineItems(rowsAfterFirstFlush)).toEqual([ { type: "assistant_message", text: "A" }, @@ -1184,19 +1262,26 @@ describe("target coalesced behavior", () => { const { agentId, session } = await createManagedSession(harness); session.pushEvent(assistant("flush")); + session.pushEvent(assistant("-pending")); await waitForSessionEventQueue(); - expect(await harness.manager.getTimelineRows(agentId)).toHaveLength(0); + // "flush" left on the leading edge; "-pending" is what manager.flush() drains. + expect(getTimelineItems(await harness.manager.getTimelineRows(agentId))).toEqual([ + { type: "assistant_message", text: "flush" }, + ]); await harness.manager.flush(); const rows = await harness.manager.getTimelineRows(agentId); const events = getTimelineStreamEvents(harness.events, agentId); - expect(getTimelineItems(rows)).toEqual([{ type: "assistant_message", text: "flush" }]); - expect(events).toHaveLength(1); + expect(getTimelineItems(rows)).toEqual([ + { type: "assistant_message", text: "flush" }, + { type: "assistant_message", text: "-pending" }, + ]); + expect(events).toHaveLength(2); await vi.advanceTimersByTimeAsync(100); - expect(await harness.manager.getTimelineRows(agentId)).toHaveLength(1); - expect(getTimelineStreamEvents(harness.events, agentId)).toHaveLength(1); + expect(await harness.manager.getTimelineRows(agentId)).toHaveLength(2); + expect(getTimelineStreamEvents(harness.events, agentId)).toHaveLength(2); } finally { harness.cleanup(); } @@ -1257,7 +1342,16 @@ describe("target coalesced behavior", () => { type: "timeline", provider: "codex", turnId: "turn-1", - item: { type: "assistant_message", text: "hello world" }, + item: { type: "assistant_message", text: "hello " }, + }, + }); + await expect(stream.next()).resolves.toEqual({ + done: false, + value: { + type: "timeline", + provider: "codex", + turnId: "turn-1", + item: { type: "assistant_message", text: "world" }, }, }); await expect(stream.next()).resolves.toEqual({ @@ -1268,7 +1362,13 @@ describe("target coalesced behavior", () => { const rows = await harness.manager.getTimelineRows(agentId); const streamEvents = getStreamEvents(harness.events, agentId); - expect(getTimelineItems(rows)).toEqual([{ type: "assistant_message", text: "hello world" }]); + expect(getTimelineItems(rows)).toEqual([ + { type: "assistant_message", text: "hello " }, + { type: "assistant_message", text: "world" }, + ]); + expect(getProjectedTimelineItems(rows)).toEqual([ + { type: "assistant_message", text: "hello world" }, + ]); expect(streamEvents[0]).toMatchObject({ type: "agent_stream", event: { type: "turn_started", provider: "codex", turnId: "turn-1" }, @@ -1279,10 +1379,19 @@ describe("target coalesced behavior", () => { type: "timeline", provider: "codex", turnId: "turn-1", - item: { type: "assistant_message", text: "hello world" }, + item: { type: "assistant_message", text: "hello " }, }, }); expect(streamEvents[2]).toMatchObject({ + type: "agent_stream", + event: { + type: "timeline", + provider: "codex", + turnId: "turn-1", + item: { type: "assistant_message", text: "world" }, + }, + }); + expect(streamEvents[3]).toMatchObject({ type: "agent_stream", event: { type: "turn_completed", provider: "codex", turnId: "turn-1" }, }); @@ -1308,22 +1417,29 @@ describe("target coalesced behavior", () => { const rows = await harness.manager.getTimelineRows(agentId); expect(getTimelineItems(rows)).toEqual([ + { type: "assistant_message", text: "a1" }, + { type: "assistant_message", text: "a2" }, + { type: "reasoning", text: "r1r2" }, + { type: "assistant_message", text: "b1b2" }, + ]); + expectContiguousRowSeqs(rows, [1, 2, 3, 4]); + const timelineEvents = getTimelineStreamEvents(harness.events, agentId); + expect(timelineEvents).toHaveLength(4); + expectContiguousLiveSeqs(timelineEvents, [1, 2, 3, 4]); + expect(getProjectedTimelineItems(rows)).toEqual([ { type: "assistant_message", text: "a1a2" }, { type: "reasoning", text: "r1r2" }, { type: "assistant_message", text: "b1b2" }, ]); - expectContiguousRowSeqs(rows, [1, 2, 3]); - const timelineEvents = getTimelineStreamEvents(harness.events, agentId); - expect(timelineEvents).toHaveLength(3); - expectContiguousLiveSeqs(timelineEvents, [1, 2, 3]); await vi.advanceTimersByTimeAsync(100); expect(getTimelineItems(await harness.manager.getTimelineRows(agentId))).toEqual([ - { type: "assistant_message", text: "a1a2" }, + { type: "assistant_message", text: "a1" }, + { type: "assistant_message", text: "a2" }, { type: "reasoning", text: "r1r2" }, { type: "assistant_message", text: "b1b2" }, ]); - expect(getTimelineStreamEvents(harness.events, agentId)).toHaveLength(3); + expect(getTimelineStreamEvents(harness.events, agentId)).toHaveLength(4); } finally { harness.cleanup(); } diff --git a/packages/server/src/server/agent/agent-manager.test.ts b/packages/server/src/server/agent/agent-manager.test.ts index b61a7bcd36..07cd6ae72a 100644 --- a/packages/server/src/server/agent/agent-manager.test.ts +++ b/packages/server/src/server/agent/agent-manager.test.ts @@ -16,6 +16,7 @@ import { import { AgentStorage } from "./agent-storage.js"; import { InMemoryAgentTimelineStore } from "./agent-timeline-store.js"; import { toAgentPayload } from "./agent-projections.js"; +import { projectTimelineRows } from "./timeline-projection.js"; import { getOpenAgentTabLabel, PARENT_AGENT_ID_LABEL } from "@getpaseo/protocol/agent-labels"; import { formatSystemNotificationPrompt, startAgentRun } from "./agent-prompt.js"; import { ensureAgentLoaded, ensureUnarchivedAgentLoaded } from "./agent-loading.js"; @@ -4969,34 +4970,52 @@ test("coalesces assistant chunks and persists the canonical row", async () => { } } + // The coalescer flushes the first chunk on the leading edge, so "final " ships + // as its own row and "reply" follows on the trailing window. Clients read the + // projected timeline, which merges the two back into one assistant message. const assistantTimelineEvents = streamEvents.filter( (event) => event.itemType === "assistant_message", ); - expect(assistantTimelineEvents).toHaveLength(1); + expect(assistantTimelineEvents).toHaveLength(2); expect(assistantTimelineEvents[0]).toMatchObject({ eventType: "timeline", itemType: "assistant_message", - text: "final reply", + text: "final ", seq: 1, epoch: expect.any(String), }); + expect(assistantTimelineEvents[1]).toMatchObject({ + eventType: "timeline", + itemType: "assistant_message", + text: "reply", + seq: 2, + epoch: expect.any(String), + }); expect(manager.getTimeline(snapshot.id)).toEqual([ { type: "assistant_message", - text: "final reply", + text: "final ", + }, + { + type: "assistant_message", + text: "reply", }, ]); const fetched = await manager.fetchTimeline(snapshot.id, { direction: "tail", limit: 0, }); - expect(fetched.rows).toHaveLength(1); + expect(fetched.rows).toHaveLength(2); expect(assistantTimelineEvents[0]?.epoch).toBe(fetched.epoch); - expect(fetched.rows[0]?.item).toEqual({ - type: "assistant_message", - text: "final reply", - }); + expect(projectTimelineRows({ rows: fetched.rows, mode: "projected" }).map((e) => e.item)).toEqual( + [ + { + type: "assistant_message", + text: "final reply", + }, + ], + ); }); test("fetchTimeline supports older-history pagination with before seq", async () => { diff --git a/packages/server/src/server/agent/agent-stream-coalescer.test.ts b/packages/server/src/server/agent/agent-stream-coalescer.test.ts index 64b348d8ec..c48a5ec8e4 100644 --- a/packages/server/src/server/agent/agent-stream-coalescer.test.ts +++ b/packages/server/src/server/agent/agent-stream-coalescer.test.ts @@ -25,6 +25,24 @@ function createHarness(windowMs?: number) { return { coalescer, flushes }; } +// Consume the leading-edge flush for an agent and clear it from the record, so a +// test can assert trailing-window batching on its own. Leaves the coalescer +// inside the window, which is where the trailing timer governs. +function primeLeadingEdge( + coalescer: AgentStreamCoalescer, + flushes: AgentStreamCoalescerFlush[], + agentId = "agent-1", +): void { + coalescer.handle(agentId, assistant("prime")); + const index = flushes.findIndex( + (flush) => flush.agentId === agentId && flush.item.type === "assistant_message", + ); + if (index === -1) { + throw new Error(`expected a leading-edge flush for ${agentId}`); + } + flushes.splice(index, 1); +} + function timeline( item: Extract["item"], options?: { @@ -104,9 +122,23 @@ describe("AgentStreamCoalescer", () => { vi.useRealTimers(); }); - test("coalesces same-tick assistant chunks after the configured window", async () => { + test("flushes the first chunk of a burst on the leading edge", () => { const { coalescer, flushes } = createHarness(); + expect(coalescer.handle("agent-1", assistant("hel"))).toBe(true); + expect(flushes).toEqual([ + { + agentId: "agent-1", + item: { type: "assistant_message", text: "hel" }, + provider: "codex", + }, + ]); + }); + + test("coalesces the rest of a burst until the configured window elapses", async () => { + const { coalescer, flushes } = createHarness(); + primeLeadingEdge(coalescer, flushes); + expect(coalescer.handle("agent-1", assistant("hel"))).toBe(true); expect(coalescer.handle("agent-1", assistant("lo"))).toBe(true); @@ -124,10 +156,26 @@ describe("AgentStreamCoalescer", () => { ]); }); + test("leads again after an idle window", async () => { + const { coalescer, flushes } = createHarness(); + + coalescer.handle("agent-1", assistant("first")); + expect(flushes).toHaveLength(1); + + await vi.advanceTimersByTimeAsync(60); + coalescer.handle("agent-1", assistant("second")); + + expect(flushes.map((flush) => flush.item)).toEqual([ + { type: "assistant_message", text: "first" }, + { type: "assistant_message", text: "second" }, + ]); + }); + test("uses constructor windowMs instead of a hard-coded value", async () => { const { coalescer, flushes } = createHarness(10); expect(AGENT_STREAM_COALESCE_DEFAULT_WINDOW_MS).toBe(60); + primeLeadingEdge(coalescer, flushes); expect(coalescer.handle("agent-1", assistant("fast"))).toBe(true); await vi.advanceTimersByTimeAsync(9); @@ -321,6 +369,7 @@ describe("AgentStreamCoalescer", () => { test("preserves whitespace byte-exactly", async () => { const { coalescer, flushes } = createHarness(); + primeLeadingEdge(coalescer, flushes); coalescer.handle("agent-1", assistant(" ")); coalescer.handle("agent-1", assistant("\n")); @@ -342,6 +391,7 @@ describe("AgentStreamCoalescer", () => { test("reconstructs bytes exactly for fragmented text", async () => { const { coalescer, flushes } = createHarness(); + primeLeadingEdge(coalescer, flushes); const text = "Hello, 世界.\nPunctuation: ,.!?;: — but JS strings stay intact.\n"; const chunks = [ "Hello", @@ -369,6 +419,8 @@ describe("AgentStreamCoalescer", () => { test("isolates buffers per agent", async () => { const { coalescer, flushes } = createHarness(); + primeLeadingEdge(coalescer, flushes, "agent-1"); + primeLeadingEdge(coalescer, flushes, "agent-2"); coalescer.handle("agent-1", assistant("a")); coalescer.handle("agent-2", assistant("x")); @@ -445,6 +497,7 @@ describe("AgentStreamCoalescer", () => { }, }); + primeLeadingEdge(coalescer, flushes); coalescer.handle("agent-1", assistant("first")); coalescer.flushFor("agent-1"); @@ -522,6 +575,7 @@ describe("AgentStreamCoalescer", () => { }, }); + primeLeadingEdge(coalescer, flushes); coalescer.handle("agent-1", assistant("old")); const oldTimerCallback = scheduled[0]; coalescer.flushAndDiscard("agent-1"); @@ -546,6 +600,7 @@ describe("AgentStreamCoalescer", () => { test("preserves first chunk item shape and replaces only text on collapse", async () => { const { coalescer, flushes } = createHarness(); + primeLeadingEdge(coalescer, flushes); const firstItem = { type: "assistant_message" as const, text: "he", @@ -571,6 +626,7 @@ describe("AgentStreamCoalescer", () => { test("coalesces tool call updates by callId with latest snapshot winning", async () => { const { coalescer, flushes } = createHarness(); + primeLeadingEdge(coalescer, flushes); expect(coalescer.handle("agent-1", toolCall({ output: "first" }))).toBe(true); expect(coalescer.handle("agent-1", toolCall({ output: "second" }))).toBe(true); @@ -600,6 +656,7 @@ describe("AgentStreamCoalescer", () => { test("coalesces interleaved tool call updates independently by callId", async () => { const { coalescer, flushes } = createHarness(); + primeLeadingEdge(coalescer, flushes); coalescer.handle("agent-1", toolCall({ callId: "tool-1", output: "one-a" })); coalescer.handle("agent-1", toolCall({ callId: "tool-2", output: "two-a" })); @@ -640,6 +697,7 @@ describe("AgentStreamCoalescer", () => { test("terminal tool call statuses flush immediately", () => { const { coalescer, flushes } = createHarness(); + primeLeadingEdge(coalescer, flushes); coalescer.handle("agent-1", toolCall({ output: "running" })); expect(flushes).toEqual([]); @@ -667,6 +725,7 @@ describe("AgentStreamCoalescer", () => { test("preserves mixed text and tool call arrival order within a flush", async () => { const { coalescer, flushes } = createHarness(); + primeLeadingEdge(coalescer, flushes); coalescer.handle("agent-1", assistant("a")); coalescer.handle("agent-1", toolCall({ output: "running" })); diff --git a/packages/server/src/server/agent/agent-stream-coalescer.ts b/packages/server/src/server/agent/agent-stream-coalescer.ts index 46008fcb53..ca52a5479e 100644 --- a/packages/server/src/server/agent/agent-stream-coalescer.ts +++ b/packages/server/src/server/agent/agent-stream-coalescer.ts @@ -25,6 +25,7 @@ export interface AgentStreamCoalescerFlush { export interface AgentStreamCoalescerOptions { windowMs?: number; timers: AgentStreamCoalescerTimers; + now?: () => number; onFlush: (payload: AgentStreamCoalescerFlush) => void; } @@ -51,6 +52,7 @@ interface PendingAgentStreamBuffer { toolCallEntryIndexes: Map; timer: ReturnType | null; flushing: boolean; + lastFlushAt: number | null; } function isCoalescableTimelineEvent(event: AgentStreamEvent): event is CoalescableTimelineEvent { @@ -88,10 +90,12 @@ export class AgentStreamCoalescer { private readonly onFlush: (payload: AgentStreamCoalescerFlush) => void; private readonly timers: AgentStreamCoalescerTimers; private readonly windowMs: number; + private readonly now: () => number; constructor(options: AgentStreamCoalescerOptions) { this.windowMs = options.windowMs ?? AGENT_STREAM_COALESCE_DEFAULT_WINDOW_MS; this.timers = options.timers; + this.now = options.now ?? Date.now; this.onFlush = options.onFlush; } @@ -112,7 +116,17 @@ export class AgentStreamCoalescer { return true; } + // Leading edge: the first event after an idle window flushes synchronously so + // the first token of a turn isn't delayed a full window. Sustained bursts fall + // through to the trailing timer, which keeps the message rate at one per + // window. Same shape as TerminalOutputCoalescer. if (!buffer.timer) { + const elapsed = + buffer.lastFlushAt === null ? Number.POSITIVE_INFINITY : this.now() - buffer.lastFlushAt; + if (elapsed >= this.windowMs) { + this.flushBuffer(agentId); + return true; + } this.scheduleFlush(buffer); } @@ -150,6 +164,7 @@ export class AgentStreamCoalescer { toolCallEntryIndexes: new Map(), timer: null, flushing: false, + lastFlushAt: null, }; this.buffers.set(agentId, buffer); return buffer; @@ -221,6 +236,7 @@ export class AgentStreamCoalescer { buffer.entries = []; buffer.toolCallEntryIndexes.clear(); buffer.flushing = true; + buffer.lastFlushAt = this.now(); try { for (const entry of this.collapseEntries(entries)) { diff --git a/packages/server/src/server/agent/providers/mock-load-test-agent.test.ts b/packages/server/src/server/agent/providers/mock-load-test-agent.test.ts index 3e2374282b..27afed7b9a 100644 --- a/packages/server/src/server/agent/providers/mock-load-test-agent.test.ts +++ b/packages/server/src/server/agent/providers/mock-load-test-agent.test.ts @@ -424,6 +424,31 @@ describe("MockLoadTestAgentClient", () => { unsubscribe(); }); + test("uses one continuous assistant stream for bursty rendering measurements", async () => { + vi.useFakeTimers(); + const client = new MockLoadTestAgentClient(); + const session = await client.createSession({ + provider: "mock", + cwd: process.cwd(), + model: "bursty-stream", + }); + const events: AgentStreamEvent[] = []; + const unsubscribe = session.subscribe((event) => events.push(event)); + + await session.startTurn("Measure paced rendering."); + await vi.advanceTimersByTimeAsync(0); + + const timelineItems = events.flatMap((event) => + event.type === "timeline" ? [event.item] : [], + ); + expect(timelineItems.filter((item) => item.type === "assistant_message")).not.toHaveLength(0); + expect( + timelineItems.filter((item) => item.type === "reasoning" || item.type === "tool_call"), + ).toEqual([]); + await session.interrupt(); + unsubscribe(); + }); + test("emits a settled assistant Markdown image path selected by prompt", async () => { vi.useFakeTimers(); const client = new MockLoadTestAgentClient(); @@ -505,11 +530,13 @@ describe("MockLoadTestAgentClient", () => { .reduce((max, length) => Math.max(max, length), 0); expect(longestMessage).toBeGreaterThan(20); - // First message includes the cycle header. - expect(assistantMessages[0]).toMatchObject({ - type: "assistant_message", - text: expect.stringContaining("## Cycle 1"), - }); + // The cycle header is at the start of the stream. It can straddle the first + // two messages, because the coalescer flushes the leading token of a burst + // on its own before batching the rest. + const assistantText = assistantMessages + .map((item) => (item.type === "assistant_message" ? item.text : "")) + .join(""); + expect(assistantText).toContain("## Cycle 1"); // Tool calls land in expected order at least once. const runningTools = toolCalls diff --git a/packages/server/src/server/agent/providers/mock-load-test-agent.ts b/packages/server/src/server/agent/providers/mock-load-test-agent.ts index 9fbd6d8b62..1f1f35876c 100644 --- a/packages/server/src/server/agent/providers/mock-load-test-agent.ts +++ b/packages/server/src/server/agent/providers/mock-load-test-agent.ts @@ -115,6 +115,20 @@ const MODELS: AgentModelDefinition[] = [ intervalMs: 40, }, }, + { + provider: MOCK_LOAD_TEST_PROVIDER_ID, + id: "bursty-stream", + label: "Bursty stream", + description: + "Emits tokens in uneven bursts separated by idle gaps, reproducing the lumpy arrival pattern real models produce. Use this to measure streaming smoothness.", + metadata: { + durationMs: 60_000, + intervalMs: 0, + burstMinTokens: 1, + burstMaxTokens: 40, + burstGapMs: 90, + }, + }, { provider: MOCK_LOAD_TEST_PROVIDER_ID, id: "ten-second-stream", @@ -144,6 +158,28 @@ const MODELS: AgentModelDefinition[] = [ }, ]; +/** + * Bursty emission: instead of one token per interval, emit a run of tokens + * back-to-back and then idle. Real models arrive this way, and it is the arrival + * pattern the client's paced reveal exists to smooth out. Burst sizes come from a + * seeded generator so a run is reproducible. + */ +interface BurstProfile { + minTokens: number; + maxTokens: number; + gapMs: number; +} + +function nextBurstSize(profile: BurstProfile, sequence: number): number { + // Deterministic hash of the burst index; no Math.random so runs are repeatable. + let hash = (sequence + 1) * 2654435761; + hash ^= hash >>> 15; + hash = Math.imul(hash, 2246822519); + hash ^= hash >>> 13; + const span = profile.maxTokens - profile.minTokens + 1; + return profile.minTokens + ((hash >>> 0) % span); +} + interface ActiveTurn { turnId: string; assistantMessageId: string; @@ -158,6 +194,8 @@ interface ActiveTurn { queue: CycleEvent[]; emittedTokens: number; turnStarted: boolean; + burst: BurstProfile | null; + burstIndex: number; } type CycleEvent = @@ -277,15 +315,28 @@ function resolveModelProfile(modelId: string | null | undefined): { modelId: string; durationMs: number; intervalMs: number; + burst: BurstProfile | null; } { const model = MODELS.find((entry) => entry.id === modelId) ?? MODELS[0]; const metadata = model.metadata ?? {}; + const burstMinTokens = + typeof metadata.burstMinTokens === "number" ? metadata.burstMinTokens : null; + const burstMaxTokens = + typeof metadata.burstMaxTokens === "number" ? metadata.burstMaxTokens : null; return { modelId: model.id, durationMs: typeof metadata.durationMs === "number" ? metadata.durationMs : MOCK_LOAD_TEST_DURATION_MS, intervalMs: typeof metadata.intervalMs === "number" ? metadata.intervalMs : MOCK_LOAD_TEST_INTERVAL_MS, + burst: + burstMinTokens !== null && burstMaxTokens !== null + ? { + minTokens: Math.max(1, burstMinTokens), + maxTokens: Math.max(Math.max(1, burstMinTokens), burstMaxTokens), + gapMs: typeof metadata.burstGapMs === "number" ? metadata.burstGapMs : 90, + } + : null, }; } @@ -575,6 +626,12 @@ function buildCycleQueue(turnId: string, cycle: number): CycleEvent[] { return queue; } +function buildBurstyStreamQueue(cycle: number): CycleEvent[] { + return tokenize( + [buildIntroParagraph(cycle), buildMidParagraph(), buildClosingParagraph()].join("\n\n"), + ).map((text) => ({ kind: "assistant_token", text })); +} + function createToolCall(input: { callId: string; name: string; @@ -755,6 +812,8 @@ export class MockLoadTestAgentSession implements AgentSession { queue: [], emittedTokens: 0, turnStarted: false, + burst: profile.burst, + burstIndex: 0, }; this.activeTurn = turn; const largePayload = parseLargeAgentStreamPayloadPrompt(prompt); @@ -1425,17 +1484,24 @@ export class MockLoadTestAgentSession implements AgentSession { return; } - if (turn.queue.length === 0) { - turn.cycle += 1; - turn.queue = buildCycleQueue(turn.turnId, turn.cycle); - } + const eventsThisTick = turn.burst ? nextBurstSize(turn.burst, turn.burstIndex) : 1; + turn.burstIndex += 1; - const event = turn.queue.shift(); - if (event) { + for (let emitted = 0; emitted < eventsThisTick; emitted += 1) { + if (turn.queue.length === 0) { + turn.cycle += 1; + turn.queue = turn.burst + ? buildBurstyStreamQueue(turn.cycle) + : buildCycleQueue(turn.turnId, turn.cycle); + } + const event = turn.queue.shift(); + if (!event) { + break; + } this.dispatchCycleEvent(turn, event); } - this.schedule(turn, turn.intervalMs); + this.schedule(turn, turn.burst ? turn.burst.gapMs : turn.intervalMs); } private dispatchCycleEvent(turn: ActiveTurn, event: CycleEvent): void {