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.
2026-08-11 19:42:51 +02:00
parent 1c905a114e
commit b9d7d4b18e
4 changed files with 267 additions and 59 deletions
@@ -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);
+  });
+});
@@ -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<void> {
+  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<void> {
+  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<ReplicaCachePerfReport> {
+  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,
+    };
+  });
+}
@@ -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<string, string>();
+  writes = 0;

  async getItem(key: string): Promise<string | null> {
    return this.values.get(key) ?? null;
  }

  async setItem(key: string, value: string): Promise<void> {
+    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);
@@ -57,6 +57,13 @@ const StoredCacheSchema = z.object({
 type StoredAgent = z.infer<typeof StoredAgentSchema>;
 type StoredHost = z.infer<typeof StoredHostSchema>;

+interface ReplicaInput {
+  agent: Agent | undefined;
+  workspace: WorkspaceDescriptor | undefined;
+  project: ProjectDescriptor | undefined;
+  timelineItems: StreamItem[] | undefined;
+}
+
 export interface ReplicaCacheStorage {
  getItem: (key: string) => Promise<string | null>;
  setItem: (key: string, value: string) => Promise<void>;
@@ -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<string>();
  private readonly storedHosts = new Map<string, StoredHost>();
  private readonly lastFocusedAgentIds = new Map<string, string>();
-  private readonly capturedSessions = new Map<string, SessionState>();
+  private readonly capturedInputs = new Map<string, ReplicaInput>();
  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 } {