From b9d7d4b18eab5f2bf2390aeefb0cc224022874ee Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Tue, 11 Aug 2026 19:42:51 +0200 Subject: [PATCH] Reduce storage writes during agent streaming (#3232) * perf(app): skip transient replica cache writes The replica cache subscribed to the entire session store, so live stream-head updates forced repeated JSON serialization and browser storage writes even though the head is not persisted. Select the cache's persisted inputs once and schedule writes only when that projection changes. * test(app): move perf cleanup into fixture Keep the replica-cache performance scenario as a direct setup, stream, measure, and assert flow while the spec-local Playwright fixture owns deterministic mock-agent cleanup. --- .../browser/replica-cache-performance.spec.ts | 65 ++++++++ .../e2e/support/helpers/replica-cache-perf.ts | 80 ++++++++++ .../src/runtime/replica-cache/index.test.ts | 31 +++- .../app/src/runtime/replica-cache/index.ts | 150 +++++++++++------- 4 files changed, 267 insertions(+), 59 deletions(-) create mode 100644 packages/app/e2e/browser/replica-cache-performance.spec.ts create mode 100644 packages/app/e2e/support/helpers/replica-cache-perf.ts diff --git a/packages/app/e2e/browser/replica-cache-performance.spec.ts b/packages/app/e2e/browser/replica-cache-performance.spec.ts new file mode 100644 index 000000000..11ac76bd7 --- /dev/null +++ b/packages/app/e2e/browser/replica-cache-performance.spec.ts @@ -0,0 +1,65 @@ +import { expect, test as base } from "../support/fixtures"; +import { expectAgentIdle } from "../support/helpers/agent-stream"; +import { expectComposerVisible, submitMessage } from "../support/helpers/composer"; +import { + openAgentRoute, + seedMockAgentWorkspace, + type MockAgentWorkspace, +} from "../support/helpers/mock-agent"; +import { + beginReplicaCacheMeasurement, + observeReplicaCacheWrites, + readReplicaCacheMeasurement, +} from "../support/helpers/replica-cache-perf"; + +const RUN_REPLICA_CACHE_PERF = process.env.PASEO_REPLICA_CACHE_PERF_E2E === "1"; +const MAX_STREAM_WRITES = 12; +const MAX_SERIALIZED_CHARS = 190_000; + +const test = base.extend<{ replicaCacheAgent: MockAgentWorkspace }>({ + replicaCacheAgent: async ({ page: _page }, provide) => { + const agent = await seedMockAgentWorkspace({ + repoPrefix: "replica-cache-perf-", + title: "Replica cache performance", + model: "ten-second-stream", + }); + try { + await provide(agent); + } finally { + await agent.cleanup(); + } + }, +}); +const replicaCachePerfDescribe = RUN_REPLICA_CACHE_PERF ? test.describe : test.describe.skip; + +replicaCachePerfDescribe("Replica cache performance", () => { + test("streaming writes only persisted replica changes", async ({ + page, + replicaCacheAgent: agent, + }, testInfo) => { + test.setTimeout(60_000); + await observeReplicaCacheWrites(page); + await openAgentRoute(page, agent); + await expectComposerVisible(page); + await beginReplicaCacheMeasurement(page); + + await submitMessage(page, "Measure replica persistence during a sustained stream."); + await agent.client.waitForFinish(agent.agentId, 20_000); + await expectAgentIdle(page); + await page.waitForTimeout(1_000); + + const report = await readReplicaCacheMeasurement(page); + await testInfo.attach("replica-cache-performance", { + body: JSON.stringify(report, null, 2), + contentType: "application/json", + }); + console.log(`[perf] Replica cache: ${JSON.stringify(report)}`); + + expect( + report.writes, + "streaming should not continuously persist transient state", + ).toBeLessThanOrEqual(MAX_STREAM_WRITES); + expect(report.redundantWrites).toBe(0); + expect(report.serializedChars).toBeLessThanOrEqual(MAX_SERIALIZED_CHARS); + }); +}); diff --git a/packages/app/e2e/support/helpers/replica-cache-perf.ts b/packages/app/e2e/support/helpers/replica-cache-perf.ts new file mode 100644 index 000000000..834a69475 --- /dev/null +++ b/packages/app/e2e/support/helpers/replica-cache-perf.ts @@ -0,0 +1,80 @@ +import type { Page } from "@playwright/test"; + +const REPLICA_CACHE_STORAGE_KEY = "@paseo:replica-cache"; + +export interface ReplicaCachePerfReport { + elapsedMs: number; + redundantWrites: number; + serializedChars: number; + storageDurationMs: number; + writes: number; +} + +interface ReplicaCachePerfState { + lastValue: string | null; + resetAt: number; + redundantWrites: number; + serializedChars: number; + storageDurationMs: number; + writes: number; +} + +declare global { + interface Window { + __replicaCachePerf?: ReplicaCachePerfState; + } +} + +export async function observeReplicaCacheWrites(page: Page): Promise { + await page.addInitScript((storageKey) => { + const originalSetItem = Storage.prototype.setItem; + window.__replicaCachePerf = { + lastValue: localStorage.getItem(storageKey), + resetAt: performance.now(), + redundantWrites: 0, + serializedChars: 0, + storageDurationMs: 0, + writes: 0, + }; + Storage.prototype.setItem = function measuredSetItem(key: string, value: string): void { + const startedAt = performance.now(); + originalSetItem.call(this, key, value); + if (key !== storageKey) return; + const state = window.__replicaCachePerf; + if (!state) return; + if (state.lastValue === value) state.redundantWrites += 1; + state.lastValue = value; + state.writes += 1; + state.serializedChars += value.length; + state.storageDurationMs += performance.now() - startedAt; + }; + }, REPLICA_CACHE_STORAGE_KEY); +} + +export async function beginReplicaCacheMeasurement(page: Page): Promise { + await page.evaluate(() => { + const lastValue = window.__replicaCachePerf?.lastValue ?? null; + window.__replicaCachePerf = { + lastValue, + resetAt: performance.now(), + redundantWrites: 0, + serializedChars: 0, + storageDurationMs: 0, + writes: 0, + }; + }); +} + +export async function readReplicaCacheMeasurement(page: Page): Promise { + return page.evaluate(() => { + const state = window.__replicaCachePerf; + if (!state) throw new Error("Replica cache performance observer is not installed"); + return { + elapsedMs: performance.now() - state.resetAt, + redundantWrites: state.redundantWrites, + serializedChars: state.serializedChars, + storageDurationMs: state.storageDurationMs, + writes: state.writes, + }; + }); +} diff --git a/packages/app/src/runtime/replica-cache/index.test.ts b/packages/app/src/runtime/replica-cache/index.test.ts index c634ff250..eee7f7aff 100644 --- a/packages/app/src/runtime/replica-cache/index.test.ts +++ b/packages/app/src/runtime/replica-cache/index.test.ts @@ -1,4 +1,4 @@ -import { afterEach, describe, expect, it } from "vitest"; +import { afterEach, describe, expect, it, vi } from "vitest"; import type { WorkspaceDescriptorPayload } from "@getpaseo/protocol/messages"; import { normalizeAgentSnapshot } from "@/utils/agent-snapshots"; import { @@ -15,12 +15,14 @@ const LRU_SERVER_IDS = ["host-a", "host-b", "host-c"] as const; class MemoryStorage implements ReplicaCacheStorage { readonly values = new Map(); + writes = 0; async getItem(key: string): Promise { return this.values.get(key) ?? null; } async setItem(key: string, value: string): Promise { + this.writes += 1; this.values.set(key, value); } } @@ -154,12 +156,39 @@ function seedTimeline(serverId: string, text: string): void { } afterEach(() => { + vi.useRealTimers(); const store = useSessionStore.getState(); store.clearSession(SERVER_ID); for (const serverId of LRU_SERVER_IDS) store.clearSession(serverId); }); describe("ReplicaCache", () => { + it("persists focused replica changes without writing transient stream head updates", async () => { + vi.useFakeTimers(); + const storage = new MemoryStorage(); + const cache = new ReplicaCache(storage); + cache.setHosts([SERVER_ID]); + seedSession(); + await cache.flush(); + cache.start(); + const writesBeforeStream = storage.writes; + + useSessionStore + .getState() + .setAgentStreamHead(SERVER_ID, new Map([["agent-1", [message("live", "Streaming")]]])); + await vi.advanceTimersByTimeAsync(1_000); + + expect(storage.writes).toBe(writesBeforeStream); + + useSessionStore + .getState() + .setAgentStreamTail(SERVER_ID, new Map([["agent-1", [message("saved", "Committed")]]])); + await vi.advanceTimersByTimeAsync(1_000); + + expect(storage.writes).toBe(writesBeforeStream + 1); + cache.setHosts([]); + }); + it("restores a displayable stale replica without claiming remote hydration", async () => { const storage = new MemoryStorage(); const writer = new ReplicaCache(storage); diff --git a/packages/app/src/runtime/replica-cache/index.ts b/packages/app/src/runtime/replica-cache/index.ts index 67c45a547..c2649c69c 100644 --- a/packages/app/src/runtime/replica-cache/index.ts +++ b/packages/app/src/runtime/replica-cache/index.ts @@ -57,6 +57,13 @@ const StoredCacheSchema = z.object({ type StoredAgent = z.infer; type StoredHost = z.infer; +interface ReplicaInput { + agent: Agent | undefined; + workspace: WorkspaceDescriptor | undefined; + project: ProjectDescriptor | undefined; + timelineItems: StreamItem[] | undefined; +} + export interface ReplicaCacheStorage { getItem: (key: string) => Promise; setItem: (key: string, value: string) => Promise; @@ -220,6 +227,54 @@ function serializeProject(project: ProjectDescriptor) { }; } +function replicaInputsEqual(left: ReplicaInput, right: ReplicaInput): boolean { + return ( + left.agent === right.agent && + left.workspace === right.workspace && + left.project === right.project && + left.timelineItems === right.timelineItems + ); +} + +function selectReplicaInput(session: SessionState, agentId: string | null): ReplicaInput { + const agent = agentId ? session.agents.get(agentId) : undefined; + const workspace = agent + ? ((agent.workspaceId ? session.workspaces.get(agent.workspaceId) : undefined) ?? + Array.from(session.workspaces.values()).find( + (candidate) => candidate.workspaceDirectory === agent.cwd, + )) + : undefined; + const timeline = agentId + ? selectAgentTimelineState(session, agentId) + : { status: "cold" as const }; + return { + agent, + workspace, + project: workspace ? session.projects.get(workspace.projectId) : undefined, + timelineItems: timeline.status === "cold" ? undefined : timeline.items, + }; +} + +function serializeHost(serverId: string, input: ReplicaInput): StoredHost { + const items = input.timelineItems?.filter( + (item) => item.kind !== "user_message" || !isUnreconciledLocalUserMessage(item), + ); + return { + serverId, + agents: input.agent ? [serializeAgent(input.agent)] : [], + workspaces: input.workspace ? [serializeWorkspace(input.workspace)] : [], + projects: input.project ? [serializeProject(input.project)] : [], + emptyProjects: [], + timeline: + input.agent && items + ? { + agentId: input.agent.id, + items: encodeDates(items.slice(-MAX_TIMELINE_ITEMS)), + } + : null, + }; +} + function deserializeHost(stored: StoredHost): SessionReplica { const agents = stored.agents.map((entry) => deserializeAgent(stored.serverId, entry)); const workspaces = stored.workspaces.map(normalizeWorkspaceDescriptor); @@ -254,7 +309,7 @@ export class ReplicaCache { private readonly activeServerIds = new Set(); private readonly storedHosts = new Map(); private readonly lastFocusedAgentIds = new Map(); - private readonly capturedSessions = new Map(); + private readonly capturedInputs = new Map(); private readonly maxBytes: number; private needsPersist = false; private unsubscribe: (() => void) | null = null; @@ -286,8 +341,8 @@ export class ReplicaCache { for (const serverId of this.lastFocusedAgentIds.keys()) { if (!next.has(serverId)) this.lastFocusedAgentIds.delete(serverId); } - for (const serverId of this.capturedSessions.keys()) { - if (!next.has(serverId)) this.capturedSessions.delete(serverId); + for (const serverId of this.capturedInputs.keys()) { + if (!next.has(serverId)) this.capturedInputs.delete(serverId); } if (removedStoredHost) this.needsPersist = true; if (this.unsubscribe && this.needsPersist) this.schedulePersist(); @@ -321,21 +376,28 @@ export class ReplicaCache { for (const host of this.storedHosts.values()) { useSessionStore.getState().restoreSessionReplica(host.serverId, deserializeHost(host)); const session = useSessionStore.getState().sessions[host.serverId]; - if (session) this.capturedSessions.set(host.serverId, session); + if (session) { + this.capturedInputs.set( + host.serverId, + selectReplicaInput(session, this.lastFocusedAgentIds.get(host.serverId) ?? null), + ); + } } } start(): void { if (this.unsubscribe) return; + const changedBeforeSubscription = this.captureSessions(); this.unsubscribe = useSessionStore.subscribe((state) => { if (this.activeServerIds.size === 0) return; + let changed = false; for (const serverId of this.activeServerIds) { - const focusedAgentId = state.sessions[serverId]?.focusedAgentId; - if (focusedAgentId) this.lastFocusedAgentIds.set(serverId, focusedAgentId); + const session = state.sessions[serverId]; + if (session && this.captureHost(serverId, session)) changed = true; } - this.schedulePersist(); + if (changed) this.schedulePersist(); }); - if (this.needsPersist) this.schedulePersist(); + if (changedBeforeSubscription || this.needsPersist) this.schedulePersist(); } reconcileServerId(oldServerId: string, newServerId: string): void { @@ -349,10 +411,10 @@ export class ReplicaCache { this.lastFocusedAgentIds.delete(oldServerId); this.lastFocusedAgentIds.set(newServerId, focusedAgentId); } - const capturedSession = this.capturedSessions.get(oldServerId); - if (capturedSession) { - this.capturedSessions.delete(oldServerId); - this.capturedSessions.set(newServerId, capturedSession); + const capturedInput = this.capturedInputs.get(oldServerId); + if (capturedInput) { + this.capturedInputs.delete(oldServerId); + this.capturedInputs.set(newServerId, capturedInput); } if (this.activeServerIds.delete(oldServerId)) this.activeServerIds.add(newServerId); this.needsPersist = true; @@ -374,57 +436,29 @@ export class ReplicaCache { await write.catch(() => undefined); } - private captureSessions(): void { + private captureSessions(): boolean { const sessions = useSessionStore.getState().sessions; + let changed = false; for (const serverId of this.activeServerIds) { const session = sessions[serverId]; if (!session) continue; - if (this.capturedSessions.get(serverId) === session) continue; - this.capturedSessions.set(serverId, session); - if (session.focusedAgentId) { - this.lastFocusedAgentIds.set(serverId, session.focusedAgentId); - } - const focusedAgentId = this.lastFocusedAgentIds.get(serverId) ?? null; - const focusedAgent = focusedAgentId ? session.agents.get(focusedAgentId) : undefined; - const focusedWorkspace = focusedAgent - ? ((focusedAgent.workspaceId - ? session.workspaces.get(focusedAgent.workspaceId) - : undefined) ?? - Array.from(session.workspaces.values()).find( - (workspace) => workspace.workspaceDirectory === focusedAgent.cwd, - )) - : undefined; - const timelineState = focusedAgentId - ? selectAgentTimelineState(session, focusedAgentId) - : { status: "cold" as const }; - const items = - timelineState.status === "cold" - ? undefined - : timelineState.items.filter( - (item) => item.kind !== "user_message" || !isUnreconciledLocalUserMessage(item), - ); - const timeline = - focusedAgent && items - ? { - agentId: focusedAgent.id, - items: encodeDates(items.slice(-MAX_TIMELINE_ITEMS)), - } - : null; - const stored: StoredHost = { - serverId, - agents: focusedAgent ? [serializeAgent(focusedAgent)] : [], - workspaces: focusedWorkspace ? [serializeWorkspace(focusedWorkspace)] : [], - projects: focusedWorkspace - ? [session.projects.get(focusedWorkspace.projectId)].flatMap((project) => - project ? [serializeProject(project)] : [], - ) - : [], - emptyProjects: [], - timeline, - }; - this.storedHosts.delete(serverId); - this.storedHosts.set(serverId, stored); + if (this.captureHost(serverId, session)) changed = true; } + return changed; + } + + private captureHost(serverId: string, session: SessionState): boolean { + if (session.focusedAgentId) { + this.lastFocusedAgentIds.set(serverId, session.focusedAgentId); + } + const input = selectReplicaInput(session, this.lastFocusedAgentIds.get(serverId) ?? null); + const previous = this.capturedInputs.get(serverId); + if (previous && replicaInputsEqual(previous, input)) return false; + + this.capturedInputs.set(serverId, input); + this.storedHosts.delete(serverId); + this.storedHosts.set(serverId, serializeHost(serverId, input)); + return true; } private buildBoundedPayload(): { payload: string; evicted: boolean } {