perf(app): move replica cache validation to restore

Add an observational live measurement harness for the row-store migration baseline. Trust already-shaped cache values on persist while retaining full schema validation at the restore boundary.

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);
+  });
+});

packages/app/e2e/support/global-setup.ts

@@ -179,6 +179,12 @@ async function loadHarnessEnvironment(repoRoot: string): Promise<void> {
 }
 
 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);
 

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<string, number>;
+  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<void> {
+  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<string, string>();
+
+      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<Omit<ReplicaCacheMeasurementReport, "restoreMs">> {
+  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<number | null> {
+  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<string, number>;
+} {
+  const nodes = new Map(profile.nodes.map((node) => [node.id, node]));
+  const byFrame: Record<string, number> = {};
+  let totalMicroseconds = 0;
+  for (const [index, sampleId] of (profile.samples ?? []).entries()) {
+    const delta = profile.timeDeltas?.[index] ?? 0;
+    const stackFunctions = new Set<string>();
+    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 };
+}

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<typeof StoredAgentSchema>;
 type StoredHost = z.infer<typeof StoredHostSchema>;
 type StoredTimelineItem = z.infer<typeof StoredTimelineItemSchema>;
+type StoredToolCall = Extract<StoredTimelineItem, { kind: "tool_call" }>["item"];
 type StoredWorkspace = z.infer<typeof StoredWorkspaceSchema>;
 type StoredProject = z.infer<typeof StoredProjectSchema>;
 
@@ -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<void> {