diff --git a/packages/app/e2e/browser/replica-cache-measurement.spec.ts b/packages/app/e2e/browser/replica-cache-measurement.spec.ts new file mode 100644 index 0000000000..cc77f2260c --- /dev/null +++ b/packages/app/e2e/browser/replica-cache-measurement.spec.ts @@ -0,0 +1,50 @@ +import { mkdir, writeFile } from "node:fs/promises"; +import path from "node:path"; +import { expect, test } from "@playwright/test"; +import { + beginReplicaCacheObservation, + finishReplicaCacheObservation, + installReplicaCacheMeasurement, + measureReplicaCacheRestore, + type ReplicaCacheMeasurementReport, +} from "../support/helpers/replica-cache-measurement"; + +const RUN_MEASUREMENT = process.env.PASEO_REPLICA_CACHE_MEASUREMENT === "1"; +const TARGET_URL = process.env.PASEO_REPLICA_CACHE_MEASUREMENT_URL; +const REPORT_PATH = process.env.PASEO_REPLICA_CACHE_MEASUREMENT_REPORT; +const OBSERVATION_MS = Number(process.env.PASEO_REPLICA_CACHE_MEASUREMENT_MS ?? 60_000); +const HYDRATED_SELECTOR = + process.env.PASEO_REPLICA_CACHE_HYDRATED_SELECTOR ?? + '[data-testid="agent-chat-scroll"], [data-testid^="sidebar-workspace-row-"]'; + +const measurementDescribe = RUN_MEASUREMENT ? test.describe : test.describe.skip; + +measurementDescribe("Replica cache live measurement", () => { + test("observes persistence and cached restore without mutating the daemon", async ({ + page, + }, testInfo) => { + test.setTimeout(OBSERVATION_MS + 90_000); + expect(TARGET_URL, "PASEO_REPLICA_CACHE_MEASUREMENT_URL must be a full app URL").toBeTruthy(); + expect(REPORT_PATH, "PASEO_REPLICA_CACHE_MEASUREMENT_REPORT must be set").toBeTruthy(); + + await installReplicaCacheMeasurement(page); + await page.goto(TARGET_URL!, { waitUntil: "domcontentloaded" }); + await page.waitForLoadState("networkidle").catch(() => undefined); + + const observation = await beginReplicaCacheObservation(page); + await page.waitForTimeout(OBSERVATION_MS); + const observed = await finishReplicaCacheObservation(page, observation, TARGET_URL!); + const restoreMs = await measureReplicaCacheRestore(page, TARGET_URL!, HYDRATED_SELECTOR); + const report: ReplicaCacheMeasurementReport = { ...observed, restoreMs }; + + await mkdir(path.dirname(REPORT_PATH!), { recursive: true }); + await writeFile(REPORT_PATH!, `${JSON.stringify(report, null, 2)}\n`); + await testInfo.attach("replica-cache-measurement", { + body: JSON.stringify(report, null, 2), + contentType: "application/json", + }); + console.log(`[measurement] Replica cache: ${JSON.stringify(report)}`); + + expect(report.observationMs).toBeGreaterThanOrEqual(OBSERVATION_MS); + }); +}); diff --git a/packages/app/e2e/support/global-setup.ts b/packages/app/e2e/support/global-setup.ts index a7c087c635..a7e25fb50b 100644 --- a/packages/app/e2e/support/global-setup.ts +++ b/packages/app/e2e/support/global-setup.ts @@ -179,6 +179,12 @@ async function loadHarnessEnvironment(repoRoot: string): Promise { } export default async function globalSetup() { + if (process.env.PASEO_REPLICA_CACHE_MEASUREMENT === "1") { + if (!process.env.PASEO_REPLICA_CACHE_MEASUREMENT_URL) { + throw new Error("PASEO_REPLICA_CACHE_MEASUREMENT_URL must be set for live measurement"); + } + return; + } const repoRoot = path.resolve(__dirname, "../../../.."); await loadHarnessEnvironment(repoRoot); diff --git a/packages/app/e2e/support/helpers/replica-cache-measurement.ts b/packages/app/e2e/support/helpers/replica-cache-measurement.ts new file mode 100644 index 0000000000..2042a0af0c --- /dev/null +++ b/packages/app/e2e/support/helpers/replica-cache-measurement.ts @@ -0,0 +1,307 @@ +import type { CDPSession, Page } from "@playwright/test"; + +const REPLICA_CACHE_DATABASE = "paseo-replica-cache"; +const PERSIST_FRAME_NAMES = [ + "captureSessions", + "captureHost", + "buildBoundedPayload", + "serialize", + "safeParse", + "stringify", + "setItem", + "measuredPut", +] as const; +const PERSIST_STACK_ANCHORS = new Set([ + "captureSessions", + "captureHost", + "buildBoundedPayload", + "serialize", + "persist", + "flushPending", + "measuredPut", +]); + +export interface ReplicaCacheMeasurementReport { + targetUrl: string; + observationMs: number; + persistCpuMs: number; + persistCpuFrames: Record; + flushes: number; + bytesWritten: { + total: number; + perFlush: number[]; + }; + redundantWrites: number; + storageDurationMs: number; + longTasks: { + count: number; + totalDurationMs: number; + maximumDurationMs: number; + durationsMs: number[]; + }; + restoreMs: number | null; +} + +interface BrowserMeasurementState { + flushes: Array<{ bytes: number; durationMs: number }>; + longTasks: number[]; + redundantWrites: number; +} + +interface ProfileNode { + id: number; + callFrame: { functionName: string; url: string }; + parent?: number; +} + +interface SamplingProfile { + nodes: ProfileNode[]; + samples?: number[]; + timeDeltas?: number[]; +} + +declare global { + interface Window { + __replicaCacheMeasurement?: BrowserMeasurementState; + __replicaCacheRestoreReadStartedAt?: number; + } +} + +export async function installReplicaCacheMeasurement(page: Page): Promise { + await page.addInitScript( + ({ databaseName }) => { + const encoder = new TextEncoder(); + const transactionWrites = new WeakMap< + IDBTransaction, + Array< + | { type: "put"; key: string; value: string; bytes: number; startedAt: number } + | { type: "delete"; key: string; bytes: 0; startedAt: number } + > + >(); + const lastValues = new Map(); + + const resolveStorageKey = ( + store: IDBObjectStore, + value: unknown, + explicitKey: IDBValidKey | undefined, + ): unknown => { + if (explicitKey !== undefined) return explicitKey; + if (value === null || typeof value !== "object") return null; + if (Array.isArray(store.keyPath)) { + return store.keyPath.map((part) => Reflect.get(value, part)); + } + return typeof store.keyPath === "string" ? Reflect.get(value, store.keyPath) : null; + }; + + window.__replicaCacheMeasurement = { + flushes: [], + longTasks: [], + redundantWrites: 0, + }; + + const observeTransaction = (transaction: IDBTransaction) => { + let writes = transactionWrites.get(transaction); + if (writes) return writes; + writes = []; + transactionWrites.set(transaction, writes); + transaction.addEventListener("complete", () => { + const state = window.__replicaCacheMeasurement; + if (!state || writes!.length === 0) return; + let bytes = 0; + let startedAt = Number.POSITIVE_INFINITY; + for (const write of writes!) { + bytes += write.bytes; + startedAt = Math.min(startedAt, write.startedAt); + if (write.type === "put") { + if (lastValues.get(write.key) === write.value) state.redundantWrites += 1; + lastValues.set(write.key, write.value); + } else { + lastValues.delete(write.key); + } + } + state.flushes.push({ bytes, durationMs: performance.now() - startedAt }); + }); + return writes; + }; + + const originalPut = IDBObjectStore.prototype.put; + IDBObjectStore.prototype.put = function measuredPut(value: unknown, key?: IDBValidKey) { + const request = + key === undefined ? originalPut.call(this, value) : originalPut.call(this, value, key); + if (this.transaction.db.name !== databaseName) return request; + const serialized = typeof value === "string" ? value : JSON.stringify(value); + const storageKey = JSON.stringify(resolveStorageKey(this, value, key)); + observeTransaction(this.transaction).push({ + type: "put", + key: `${this.name}:${storageKey}`, + value: serialized, + bytes: encoder.encode(serialized).byteLength, + startedAt: performance.now(), + }); + return request; + }; + const originalDelete = IDBObjectStore.prototype.delete; + IDBObjectStore.prototype.delete = function measuredDelete(query: IDBValidKey | IDBKeyRange) { + const request = originalDelete.call(this, query); + if (this.transaction.db.name !== databaseName) return request; + observeTransaction(this.transaction).push({ + type: "delete", + key: `${this.name}:${JSON.stringify(query)}`, + bytes: 0, + startedAt: performance.now(), + }); + return request; + }; + + const markRestoreRead = (store: IDBObjectStore) => { + if ( + store.transaction.db.name === databaseName && + store.transaction.mode === "readonly" && + window.__replicaCacheRestoreReadStartedAt === undefined + ) { + window.__replicaCacheRestoreReadStartedAt = performance.now(); + } + }; + const originalGet = IDBObjectStore.prototype.get; + IDBObjectStore.prototype.get = function measuredGet(query: IDBValidKey | IDBKeyRange) { + markRestoreRead(this); + return originalGet.call(this, query); + }; + const originalGetAll = IDBObjectStore.prototype.getAll; + IDBObjectStore.prototype.getAll = function measuredGetAll( + query?: IDBValidKey | IDBKeyRange | null, + count?: number, + ) { + markRestoreRead(this); + return originalGetAll.call(this, query, count); + }; + const originalOpenCursor = IDBObjectStore.prototype.openCursor; + IDBObjectStore.prototype.openCursor = function measuredOpenCursor( + query?: IDBValidKey | IDBKeyRange | null, + direction?: IDBCursorDirection, + ) { + markRestoreRead(this); + return originalOpenCursor.call(this, query, direction); + }; + + if (typeof PerformanceObserver !== "undefined") { + const observer = new PerformanceObserver((entries) => { + const state = window.__replicaCacheMeasurement; + if (!state) return; + for (const entry of entries.getEntries()) state.longTasks.push(entry.duration); + }); + try { + observer.observe({ type: "longtask", buffered: true }); + } catch { + // Chromium exposes Long Tasks; leave an empty list on runtimes that do not. + } + } + }, + { databaseName: REPLICA_CACHE_DATABASE }, + ); +} + +export async function beginReplicaCacheObservation( + page: Page, +): Promise<{ cdp: CDPSession; startedAt: number }> { + await page.evaluate(() => { + const state = window.__replicaCacheMeasurement; + if (!state) throw new Error("Replica cache measurement is not installed"); + state.flushes = []; + state.longTasks = []; + state.redundantWrites = 0; + }); + const cdp = await page.context().newCDPSession(page); + await cdp.send("Profiler.enable"); + await cdp.send("Profiler.setSamplingInterval", { interval: 100 }); + await cdp.send("Profiler.start"); + return { cdp, startedAt: Date.now() }; +} + +export async function finishReplicaCacheObservation( + page: Page, + observation: { cdp: CDPSession; startedAt: number }, + targetUrl: string, +): Promise> { + const result = (await observation.cdp.send("Profiler.stop")) as { profile: SamplingProfile }; + await observation.cdp.detach(); + const browser = await page.evaluate(() => { + const state = window.__replicaCacheMeasurement; + if (!state) throw new Error("Replica cache measurement is not installed"); + return state; + }); + const cpu = summarizePersistCpu(result.profile); + const bytesPerFlush = browser.flushes.map((flush) => flush.bytes); + const longTaskDurations = browser.longTasks; + return { + targetUrl, + observationMs: Date.now() - observation.startedAt, + persistCpuMs: cpu.totalMs, + persistCpuFrames: cpu.byFrame, + flushes: browser.flushes.length, + bytesWritten: { + total: bytesPerFlush.reduce((sum, bytes) => sum + bytes, 0), + perFlush: bytesPerFlush, + }, + redundantWrites: browser.redundantWrites, + storageDurationMs: browser.flushes.reduce((sum, flush) => sum + flush.durationMs, 0), + longTasks: { + count: longTaskDurations.length, + totalDurationMs: longTaskDurations.reduce((sum, duration) => sum + duration, 0), + maximumDurationMs: Math.max(0, ...longTaskDurations), + durationsMs: longTaskDurations, + }, + }; +} + +export async function measureReplicaCacheRestore( + page: Page, + targetUrl: string, + hydratedSelector: string, +): Promise { + await page.routeWebSocket(/\/ws(?:\?|$)/, async (socket) => { + await socket.close({ code: 1000, reason: "Hold reconnect until cached replica paints" }); + }); + await page.goto(targetUrl, { waitUntil: "domcontentloaded" }); + const restoredAt = await page.waitForFunction( + (selector) => { + const startedAt = window.__replicaCacheRestoreReadStartedAt; + if (startedAt === undefined) return null; + const element = document.querySelector(selector); + if (!element || !(element instanceof HTMLElement) || element.offsetParent === null) + return null; + return performance.now(); + }, + hydratedSelector, + { timeout: 30_000 }, + ); + const restoredAtMs = await restoredAt.jsonValue(); + const startedAtMs = await page.evaluate(() => window.__replicaCacheRestoreReadStartedAt ?? null); + return typeof restoredAtMs === "number" && startedAtMs !== null + ? restoredAtMs - startedAtMs + : null; +} + +function summarizePersistCpu(profile: SamplingProfile): { + totalMs: number; + byFrame: Record; +} { + const nodes = new Map(profile.nodes.map((node) => [node.id, node])); + const byFrame: Record = {}; + let totalMicroseconds = 0; + for (const [index, sampleId] of (profile.samples ?? []).entries()) { + const delta = profile.timeDeltas?.[index] ?? 0; + const stackFunctions = new Set(); + let node = nodes.get(sampleId); + while (node) { + stackFunctions.add(node.callFrame.functionName); + node = node.parent === undefined ? undefined : nodes.get(node.parent); + } + if (![...stackFunctions].some((name) => PERSIST_STACK_ANCHORS.has(name))) continue; + const matched = PERSIST_FRAME_NAMES.filter((name) => stackFunctions.has(name)); + if (matched.length === 0) continue; + totalMicroseconds += delta; + for (const pattern of matched) byFrame[pattern] = (byFrame[pattern] ?? 0) + delta / 1_000; + } + return { totalMs: totalMicroseconds / 1_000, byFrame }; +} diff --git a/packages/app/src/runtime/replica-cache/index.ts b/packages/app/src/runtime/replica-cache/index.ts index a22ef07abc..ba758efc5d 100644 --- a/packages/app/src/runtime/replica-cache/index.ts +++ b/packages/app/src/runtime/replica-cache/index.ts @@ -19,7 +19,11 @@ import { type ProjectDescriptor, type WorkspaceDescriptor, } from "@/stores/session-store"; -import { isUnreconciledLocalUserMessage, type StreamItem } from "@/types/stream"; +import { + isUnreconciledLocalUserMessage, + type AgentToolCallData, + type StreamItem, +} from "@/types/stream"; import { normalizeAgentSnapshot } from "@/utils/agent-snapshots"; const STORAGE_KEY = "@paseo:replica-cache"; @@ -293,6 +297,7 @@ const StoredCacheSchema = z.strictObject({ type StoredAgent = z.infer; type StoredHost = z.infer; type StoredTimelineItem = z.infer; +type StoredToolCall = Extract["item"]; type StoredWorkspace = z.infer; type StoredProject = z.infer; @@ -337,6 +342,24 @@ function timelineBase(item: StreamItem) { }; } +function serializeAgentToolCall(data: AgentToolCallData): StoredToolCall { + const base = { + type: "tool_call" as const, + callId: data.callId, + name: data.name, + detail: data.detail, + ...(data.metadata ? { metadata: data.metadata } : {}), + }; + switch (data.status) { + case "running": + case "completed": + case "canceled": + return { ...base, status: data.status, error: null }; + case "failed": + return { ...base, status: data.status, error: data.error }; + } +} + function serializeTimelineItem(item: StreamItem): StoredTimelineItem | null { const base = timelineBase(item); switch (item.kind) { @@ -384,21 +407,11 @@ function serializeTimelineItem(item: StreamItem): StoredTimelineItem | null { }; case "tool_call": if (item.payload.source !== "agent") return null; - const storedTool = AgentTimelineItemPayloadSchema.safeParse({ - type: "tool_call", - callId: item.payload.data.callId, - name: item.payload.data.name, - status: item.payload.data.status, - error: item.payload.data.error, - detail: item.payload.data.detail, - ...(item.payload.data.metadata ? { metadata: item.payload.data.metadata } : {}), - }); - if (!storedTool.success || storedTool.data.type !== "tool_call") return null; return { ...base, kind: item.kind, provider: item.payload.data.provider, - item: storedTool.data, + item: serializeAgentToolCall(item.payload.data), }; } } @@ -788,11 +801,6 @@ export class ReplicaCache { if (host.timeline) this.lastFocusedAgentIds.set(host.serverId, host.timeline.agentId); } const bounded = this.buildBoundedPayload(); - if (!bounded) { - await this.clearInvalidCache(); - this.storedHosts.clear(); - return; - } if (bounded.evicted) this.needsPersist = true; for (const host of this.storedHosts.values()) { useSessionStore.getState().restoreSessionReplica(host.serverId, deserializeHost(host)); @@ -884,10 +892,6 @@ export class ReplicaCache { if (skipUnchanged && !changed && !this.needsPersist) return; const bounded = this.buildBoundedPayload(); this.needsPersist = false; - if (!bounded) { - await this.clearInvalidCache(); - return; - } const write = this.writeQueue .catch(() => undefined) .then(() => this.storage.setItem(STORAGE_KEY, bounded.payload)); @@ -934,27 +938,24 @@ export class ReplicaCache { return true; } - private buildBoundedPayload(): { payload: string; evicted: boolean } | null { + private buildBoundedPayload(): { payload: string; evicted: boolean } { let evicted = false; let payload = this.serialize(); - if (payload === null) return null; while (Buffer.byteLength(payload, "utf8") > this.maxBytes && this.storedHosts.size > 0) { const oldestServerId = this.storedHosts.keys().next().value; if (oldestServerId === undefined) break; this.storedHosts.delete(oldestServerId); evicted = true; payload = this.serialize(); - if (payload === null) return null; } return { payload, evicted }; } - private serialize(): string | null { - const cache = StoredCacheSchema.safeParse({ + private serialize(): string { + return JSON.stringify({ version: CACHE_VERSION, hosts: Array.from(this.storedHosts.values()), }); - return cache.success ? JSON.stringify(cache.data) : null; } private async clearInvalidCache(): Promise {