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 } {