fix(app): prevent archived workspaces returning from cache (#3975)

Durable replica reads could race accepted directory replacements and restore rows that had already been removed in memory. Serialize reads behind pending writes and discard reads when the host changes in flight; local archives also suppress targeted cache restoration until removal is acknowledged.

packages/app/e2e/browser/workspace-archive-cache.spec.ts

@@ -0,0 +1,36 @@
+import { expect, test } from "../support/fixtures";
+import { gotoAppShell } from "../support/helpers/app";
+import { waitForWorkspaceInReplicaCache } from "../support/helpers/replica-cache-storage";
+import { seedWorkspace, type SeededWorkspace } from "../support/helpers/seed-client";
+import {
+  expectWorkspaceAbsentFromSidebar,
+  selectWorkspaceInSidebar,
+} from "../support/helpers/sidebar";
+import { waitForSidebarHydration } from "../support/helpers/workspace-ui";
+
+async function archiveWorkspaceOutsideTheApp(workspace: SeededWorkspace): Promise<void> {
+  const result = await workspace.client.archiveWorkspace(workspace.workspaceId);
+  expect(result.error).toBeNull();
+}
+
+test.describe("Workspace archive cache coherence", () => {
+  test("an archived selected workspace cannot return from the durable cache", async ({ page }) => {
+    const workspace = await seedWorkspace({ repoPrefix: "archive-cache-" });
+
+    try {
+      await gotoAppShell(page);
+      await waitForSidebarHydration(page);
+      await selectWorkspaceInSidebar(page, workspace.workspaceId);
+      await waitForWorkspaceInReplicaCache(page, workspace.workspaceId);
+
+      await archiveWorkspaceOutsideTheApp(workspace);
+
+      await expectWorkspaceAbsentFromSidebar(page, workspace.workspaceId);
+      await expect(page.getByText("Workspace unavailable", { exact: true })).toBeVisible({
+        timeout: 30_000,
+      });
+    } finally {
+      await workspace.cleanup();
+    }
+  });
+});

packages/app/e2e/support/helpers/replica-cache-storage.ts

@@ -1,4 +1,4 @@
-import type { Page } from "@playwright/test";
+import { expect, type Page } from "@playwright/test";
 
 const DATABASE_NAME = "paseo-replica-row-store";
 const STORE_NAME = "rows";
@@ -116,6 +116,23 @@ export async function readReplicaCache(page: Page): Promise<ReplicaCacheRecord |
   return rows.length > 0 ? assembleCache(rows) : null;
 }
 
+export async function waitForWorkspaceInReplicaCache(
+  page: Page,
+  workspaceId: string,
+): Promise<void> {
+  await expect
+    .poll(
+      async () => {
+        const cache = await readReplicaCache(page);
+        return cache?.hosts?.some((host) =>
+          host.workspaces.some((workspace) => workspace.id === workspaceId),
+        );
+      },
+      { timeout: 15_000 },
+    )
+    .toBe(true);
+}
+
 export async function writeReplicaCache(page: Page, value: ReplicaCacheRecord): Promise<void> {
   const rows: ReplicaRowRecord[] = [];
   for (const host of value.hosts ?? []) {

packages/app/e2e/support/helpers/seed-client.ts

@@ -69,6 +69,7 @@ export interface SeedDaemonClient {
     workspace: SeedWorkspaceDescriptor | null;
     error: string | null;
   }>;
+  archiveWorkspace(workspaceId: string): Promise<{ error: string | null }>;
   /**
    * Force the daemon to recompute its git snapshot and diff for a checkout,
    * mirroring the UI's manual refresh. Tests use this to make an out-of-band

packages/app/src/runtime/directory-sync/workspace-replica.test.ts

@@ -6,6 +6,10 @@ import {
   normalizeWorkspaceDescriptor,
   useSessionStore,
 } from "@/stores/session-store";
+import {
+  clearWorkspaceArchivePending,
+  markWorkspaceArchivePending,
+} from "@/contexts/session-workspace-upserts";
 import { WorkspaceDirectoryReplica } from "./workspace-replica";
 
 function workspace(id: string, projectId = "project"): WorkspaceDescriptorPayload {
@@ -208,3 +212,21 @@ it("does not invent a null-key project from a workspace update", () => {
   expect(session?.projects.has("fresh-project")).toBe(false);
   store.clearSession(serverId);
 });
+
+it("does not restore a targeted cached workspace while its archive is pending", () => {
+  const serverId = "cached-workspace-during-archive";
+  const workspaceId = "archived-workspace";
+  const store = useSessionStore.getState();
+  store.initializeSession(serverId, null as unknown as DaemonClient);
+  const replica = new WorkspaceDirectoryReplica(serverId);
+  markWorkspaceArchivePending({ serverId, workspaceId });
+
+  try {
+    replica.commitCachedWorkspace(normalizeWorkspaceDescriptor(workspace(workspaceId)), undefined);
+
+    expect(useSessionStore.getState().sessions[serverId]?.workspaces.has(workspaceId)).toBe(false);
+  } finally {
+    clearWorkspaceArchivePending({ serverId, workspaceId });
+    store.clearSession(serverId);
+  }
+});

packages/app/src/runtime/directory-sync/workspace-replica.ts

@@ -96,6 +96,7 @@ export class WorkspaceDirectoryReplica {
     workspace: WorkspaceDescriptor,
     project: ProjectDescriptor | undefined,
   ): void {
+    if (shouldSuppressWorkspaceForLocalArchive({ serverId: this.serverId, workspace })) return;
     const snapshot = this.read();
     snapshot.workspaces.set(workspace.id, workspace);
     if (project) snapshot.projects.set(project.projectId, project);

packages/app/src/runtime/replica-cache/index.test.ts

@@ -12,6 +12,14 @@ import type { ReplicaHostRows, ReplicaRow, ReplicaRowChanges, ReplicaRowStore }
 
 const SERVER_ID = "cached-host";
 
+function deferred(): { promise: Promise<void>; resolve(): void } {
+  let resolve!: () => void;
+  const promise = new Promise<void>((resolvePromise) => {
+    resolve = resolvePromise;
+  });
+  return { promise, resolve };
+}
+
 class MemoryStorage implements ReplicaRowStore {
   readonly rows = new Map<string, ReplicaRow>();
   readonly changes: ReplicaRowChanges[] = [];
@@ -23,6 +31,8 @@ class MemoryStorage implements ReplicaRowStore {
   writes = 0;
   cleanups = 0;
   nextWriteFailure: Error | null = null;
+  readGate: Promise<void> | null = null;
+  onRead: (() => void) | null = null;
 
   private key(row: Pick<ReplicaRow, "serverId" | "kind" | "id">): string {
     return `${row.serverId}:${row.kind}:${row.id}`;
@@ -36,6 +46,8 @@ class MemoryStorage implements ReplicaRowStore {
     ids?: readonly string[],
   ): Promise<ReplicaRow[]> {
     this.reads.push({ serverId, kinds, ...(ids ? { ids } : {}) });
+    this.onRead?.();
+    await this.readGate;
     const acceptedKinds = new Set(kinds);
     const acceptedIds = ids ? new Set(ids) : null;
     return [...this.rows.values()].filter(
@@ -217,6 +229,72 @@ describe("ReplicaCache", () => {
     expect(restoredTimeline).toEqual(timeline());
   });
 
+  it("never reads directory rows older than an accepted deferred deletion", async () => {
+    const storage = new MemoryStorage();
+    const cache = createCache(storage);
+    cache.commitDirectory(SERVER_ID, directory());
+    await cache.flush();
+
+    cache.commitDirectory(SERVER_ID, {
+      agents: new Map(),
+      workspaces: new Map(),
+      projects: new Map(),
+    });
+
+    expect(await cache.readAgent(SERVER_ID, "agent-1")).toBeUndefined();
+    expect(await cache.readWorkspace(SERVER_ID, "workspace-1")).toBeUndefined();
+    expect((await cache.readDirectory(SERVER_ID)).projects.size).toBe(0);
+  });
+
+  it("fails closed when an accepted deletion cannot be persisted before a read", async () => {
+    const storage = new MemoryStorage();
+    const cache = createCache(storage);
+    cache.commitDirectory(SERVER_ID, directory());
+    await cache.flush();
+    storage.nextWriteFailure = new Error("disk busy");
+
+    cache.commitDirectory(SERVER_ID, {
+      agents: new Map(),
+      workspaces: new Map(),
+      projects: new Map(),
+    });
+
+    expect(await cache.readWorkspace(SERVER_ID, "workspace-1")).toBeUndefined();
+  });
+
+  it("discards a durable read when the host changes while it is in flight", async () => {
+    const storage = new MemoryStorage();
+    const cache = createCache(storage);
+    cache.commitDirectory(SERVER_ID, directory());
+    await cache.flush();
+    const started = deferred();
+    const release = deferred();
+    storage.onRead = started.resolve;
+    storage.readGate = release.promise;
+
+    const reading = cache.readAgent(SERVER_ID, "agent-1");
+    await started.promise;
+    cache.commitDirectory(SERVER_ID, {
+      agents: new Map(),
+      workspaces: new Map(),
+      projects: new Map(),
+    });
+    release.resolve();
+
+    expect(await reading).toBeUndefined();
+  });
+
+  it("never reads a timeline older than an accepted deferred replacement", async () => {
+    const storage = new MemoryStorage();
+    const cache = createCache(storage);
+    cache.commitTimeline(SERVER_ID, "agent-1", timeline("Old"));
+    await cache.flush();
+
+    cache.commitTimeline(SERVER_ID, "agent-1", timeline("New"));
+
+    expect((await cache.readTimeline(SERVER_ID, "agent-1"))?.items).toEqual([timelineItem("New")]);
+  });
+
   it("round-trips plugin timeline items", async () => {
     const storage = new MemoryStorage();
     const writer = createCache(storage);

packages/app/src/runtime/replica-cache/index.ts

@@ -784,6 +784,7 @@ function directoryEntityForRow(row: ReplicaRow): keyof DirectoryCheckpoint | und
 
 export class ReplicaCache {
   private readonly activeServerIds = new Set<string>();
+  private readonly hostRevisions = new Map<string, number>();
   private readonly storedRows = new Map<string, Map<string, ReplicaRow>>();
   private readonly hostBytes = new Map<string, number>();
   private readonly hostWriteOrder = new Map<string, true>();
@@ -897,9 +898,13 @@ export class ReplicaCache {
     ids?: readonly string[],
   ): Promise<ReplicaRow[]> {
     if (!this.activeServerIds.has(serverId)) return [];
+    const revision = this.hostRevisions.get(serverId) ?? 0;
     try {
       await this.prepareStore();
-      return await this.rowStore.read(serverId, kinds, ids);
+      await this.flush();
+      if (!this.canReadHostRevision(serverId, revision)) return [];
+      const rows = await this.rowStore.read(serverId, kinds, ids);
+      return this.canReadHostRevision(serverId, revision) ? rows : [];
     } catch {
       return [];
     }
@@ -985,6 +990,7 @@ export class ReplicaCache {
     },
   ): void {
     if (!this.activeServerIds.has(serverId)) return;
+    this.advanceHostRevision(serverId);
     this.clearPendingDirectoryChanges(serverId);
     const desiredRows: ReplicaRow[] = [];
     for (const agent of directory.agents.values()) {
@@ -1013,6 +1019,7 @@ export class ReplicaCache {
   commitTimeline(serverId: string, agentId: string, timeline: CachedTimeline): void {
     if (!this.activeServerIds.has(serverId)) return;
     if (timeline.agentId !== agentId) throw new Error("Timeline cache key does not match payload");
+    this.advanceHostRevision(serverId);
     const stored = serializeTimeline(timeline);
     if (stored) this.queueEntityUpsert(serverId, "timeline", agentId, stored);
     else this.queueEntityDelete(serverId, "timeline", agentId);
@@ -1022,6 +1029,8 @@ export class ReplicaCache {
   setHosts(serverIds: Iterable<string>): void {
     const next = new Set(serverIds);
     const removed = [...this.activeServerIds].filter((serverId) => !next.has(serverId));
+    const added = [...next].filter((serverId) => !this.activeServerIds.has(serverId));
+    for (const serverId of [...removed, ...added]) this.advanceHostRevision(serverId);
     this.activeServerIds.clear();
     for (const serverId of next) this.activeServerIds.add(serverId);
     for (const serverId of removed) {
@@ -1032,6 +1041,8 @@ export class ReplicaCache {
   }
 
   reconcileServerId(oldServerId: string, newServerId: string): void {
+    this.advanceHostRevision(oldServerId);
+    this.advanceHostRevision(newServerId);
     const rows = this.storedRows.get(oldServerId);
     if (rows) {
       const newRows = this.storedRows.get(newServerId) ?? new Map<string, ReplicaRow>();
@@ -1142,6 +1153,29 @@ export class ReplicaCache {
     );
   }
 
+  private canReadHostRevision(serverId: string, revision: number): boolean {
+    return (
+      this.activeServerIds.has(serverId) &&
+      (this.hostRevisions.get(serverId) ?? 0) === revision &&
+      !this.hasPendingHostChanges(serverId)
+    );
+  }
+
+  private hasPendingHostChanges(serverId: string): boolean {
+    if (this.pendingDirectoryReplacements.has(serverId)) return true;
+    for (const row of this.pendingUpserts.values()) {
+      if (row.serverId === serverId) return true;
+    }
+    for (const row of this.pendingDeletes.values()) {
+      if (row.serverId === serverId) return true;
+    }
+    return false;
+  }
+
+  private advanceHostRevision(serverId: string): void {
+    this.hostRevisions.set(serverId, (this.hostRevisions.get(serverId) ?? 0) + 1);
+  }
+
   private drainPendingChanges(): PendingReplicaChanges {
     const changes = {
       upserts: [...this.pendingUpserts.values()],