diff --git a/docs/architecture.md b/docs/architecture.md index d35e079a89..6599fa673e 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -121,8 +121,9 @@ separate product/security decision. Agents, workspaces, and projects have indivi focused timeline tail and directory checkpoint each have one row per host. Writes serialize only rows whose in-memory identities changed. Restore validates every row before hydration; one invalid row clears the cache and starts cold. The 32 MiB byte budget is tracked incrementally from row -payload lengths and evicts whole hosts in least-recently-written order. Browser and Electron builds -use IndexedDB. Native builds use expo-sqlite. +payload lengths and evicts whole hosts in least-recently-written order. After restart, that order is +an approximation rebuilt from the row store's `readAll()` order. Browser and Electron builds use +IndexedDB. Native builds use expo-sqlite. The three directory entity types have independent monotonic sequences and share one daemon generation. The daemon retains only the latest projection per entity and bounded tombstones, not an diff --git a/packages/app/src/runtime/directory-sync/workspace-replica.test.ts b/packages/app/src/runtime/directory-sync/workspace-replica.test.ts index 9a1705bd53..cb6a9927c9 100644 --- a/packages/app/src/runtime/directory-sync/workspace-replica.test.ts +++ b/packages/app/src/runtime/directory-sync/workspace-replica.test.ts @@ -149,6 +149,52 @@ it("commits the authoritative snapshot before buffered project updates", () => { store.clearSession(serverId); }); +it("preserves unchanged project identity when another project changes", () => { + const serverId = "project-identity"; + const store = useSessionStore.getState(); + store.initializeSession(serverId, null as unknown as DaemonClient); + const replica = new WorkspaceDirectoryReplica(serverId); + const first = normalizeProjectDescriptor({ + projectId: "first", + projectDisplayName: "First", + projectRootPath: "/repo/first", + projectKind: "git", + }); + const second = normalizeProjectDescriptor({ + projectId: "second", + projectDisplayName: "Second", + projectRootPath: "/repo/second", + projectKind: "git", + }); + replica.commitSnapshot( + { + workspaces: new Map(), + projects: new Map([ + ["first", first], + ["second", second], + ]), + }, + [], + ); + const previousSecond = useSessionStore.getState().sessions[serverId]?.projects.get("second"); + + replica.commitSnapshot( + { + workspaces: new Map(), + projects: new Map([ + ["first", { ...first, projectDisplayName: "Updated" }], + ["second", { ...second }], + ]), + }, + [], + ); + + expect(useSessionStore.getState().sessions[serverId]?.projects.get("second")).toBe( + previousSecond, + ); + store.clearSession(serverId); +}); + it("does not invent a null-key project from a workspace update", () => { const serverId = "workspace-before-project-update"; const store = useSessionStore.getState(); diff --git a/packages/app/src/runtime/replica-cache/index.test.ts b/packages/app/src/runtime/replica-cache/index.test.ts index 96134fb803..12497e8a37 100644 --- a/packages/app/src/runtime/replica-cache/index.test.ts +++ b/packages/app/src/runtime/replica-cache/index.test.ts @@ -1,6 +1,8 @@ import { afterEach, describe, expect, it, vi } from "vitest"; +import type { DaemonClient } from "@getpaseo/client/internal/daemon-client"; import type { WorkspaceDescriptorPayload } from "@getpaseo/protocol/messages"; import { normalizeAgentSnapshot } from "@/utils/agent-snapshots"; +import { projectAgentSnapshot } from "@/utils/agent-snapshots"; import { normalizeProjectDescriptor, normalizeWorkspaceDescriptor, @@ -9,6 +11,7 @@ import { } from "@/stores/session-store"; import { createUserMessage, type StreamItem } from "@/types/stream"; import { ReplicaCache } from "."; +import { AgentDirectoryReplica } from "../directory-sync/agent-replica"; import type { ReplicaHostRows, ReplicaRow, ReplicaRowChanges, ReplicaRowStore } from "./row-store"; const SERVER_ID = "cached-host"; @@ -21,6 +24,7 @@ class MemoryStorage implements ReplicaRowStore { readonly renamedHosts: Array<{ oldServerId: string; newServerId: string }> = []; writes = 0; clears = 0; + applyResults: Promise[] = []; private key(row: Pick): string { return `${row.serverId}:${row.kind}:${row.id}`; @@ -40,6 +44,8 @@ class MemoryStorage implements ReplicaRowStore { async apply(changes: ReplicaRowChanges): Promise { this.writes += 1; + const result = this.applyResults.shift(); + if (result) await result; this.changes.push(changes); for (const key of changes.deletes) this.rows.delete(this.key(key)); for (const row of changes.upserts) this.rows.set(this.key(row), row); @@ -66,6 +72,24 @@ class MemoryStorage implements ReplicaRowStore { } } +function deferred(): { + promise: Promise; + reject: (error: Error) => void; +} { + let reject = (_error: Error): void => undefined; + const promise = new Promise((_resolve, rejectPromise) => { + reject = rejectPromise; + }); + return { promise, reject }; +} + +async function waitForWrites(storage: MemoryStorage, expected: number): Promise { + for (let attempt = 0; attempt < 20 && storage.writes < expected; attempt += 1) { + await Promise.resolve(); + } + expect(storage.writes).toBe(expected); +} + const NO_LEGACY_CLEANUP = { clearLegacyCache: async () => undefined }; function cache(storage: MemoryStorage, options: { maxBytes?: number } = {}): ReplicaCache { @@ -602,6 +626,97 @@ describe("ReplicaCache", () => { }); }); + it("persists the focused synced timeline produced by the session store", async () => { + const storage = new MemoryStorage(); + const replicaCache = cache(storage); + replicaCache.setHosts([SERVER_ID]); + const store = useSessionStore.getState(); + store.initializeSession(SERVER_ID, null as unknown as DaemonClient); + const focused = agent("agent-1"); + store.setAgents(SERVER_ID, new Map([[focused.id, focused]])); + store.setFocusedAgentId(SERVER_ID, focused.id); + store.setAgentStreamState(SERVER_ID, focused.id, { + tail: [message("synced-message", "Synced from daemon")], + }); + store.setAgentTimelineCursor( + SERVER_ID, + new Map([[focused.id, { epoch: "epoch-1", startSeq: 1, endSeq: 12 }]]), + ); + store.setAgentAuthoritativeHistoryApplied(SERVER_ID, focused.id, true); + + await replicaCache.flush(); + + expect(storage.changes).toHaveLength(1); + expect(storage.changes[0]?.upserts).toContainEqual( + expect.objectContaining({ + serverId: SERVER_ID, + kind: "timeline", + id: "singleton", + }), + ); + }); + + it("does not persist unchanged agents after a changes-mode reconciliation", async () => { + const storage = new MemoryStorage(); + const replicaCache = cache(storage); + replicaCache.setHosts([SERVER_ID]); + const store = useSessionStore.getState(); + store.initializeSession(SERVER_ID, null as unknown as DaemonClient); + const directory = new AgentDirectoryReplica(SERVER_ID, () => undefined); + const agents = [agent("agent-1"), agent("agent-2")]; + directory.commitSnapshot( + agents.map((value) => ({ + agent: projectAgentSnapshot(value), + project: value.projectPlacement!, + })), + [], + ); + await replicaCache.flush(); + storage.changes.length = 0; + + directory.commitChanges([], [], []); + await replicaCache.flush(); + + expect(storage.changes).toEqual([]); + }); + + it("persists only the changed agent after a changes-mode reconciliation", async () => { + const storage = new MemoryStorage(); + const replicaCache = cache(storage); + replicaCache.setHosts([SERVER_ID]); + const store = useSessionStore.getState(); + store.initializeSession(SERVER_ID, null as unknown as DaemonClient); + const directory = new AgentDirectoryReplica(SERVER_ID, () => undefined); + const agents = [agent("agent-1"), agent("agent-2")]; + directory.commitSnapshot( + agents.map((value) => ({ + agent: projectAgentSnapshot(value), + project: value.projectPlacement!, + })), + [], + ); + await replicaCache.flush(); + storage.changes.length = 0; + + directory.commitChanges( + [ + { + agent: { ...projectAgentSnapshot(agents[0]), title: "Updated" }, + project: agents[0].projectPlacement!, + }, + ], + [], + [], + ); + await replicaCache.flush(); + + expect(storage.changes).toHaveLength(1); + expect(storage.changes[0]).toMatchObject({ + upserts: [{ serverId: SERVER_ID, kind: "agent", id: "agent-1" }], + deletes: [], + }); + }); + it("persists a checkpoint without serializing host entity rows", async () => { const storage = new MemoryStorage(); const replicaCache = cache(storage); @@ -743,6 +858,153 @@ describe("ReplicaCache", () => { expect(Object.keys(useSessionStore.getState().sessions).sort()).toEqual(["host-a", "host-c"]); }); + it("never writes an oversized host and does not retry it on every flush", async () => { + const storage = new MemoryStorage(); + const replicaCache = cache(storage, { maxBytes: 1_000 }); + replicaCache.setHosts([SERVER_ID]); + seedTimeline(SERVER_ID, "oversized".repeat(500)); + + await replicaCache.flush(); + await replicaCache.flush(); + + expect(storage.changes).toEqual([]); + expect(storage.deletedHosts).toEqual([]); + }); + + it("lets a changed evicted host displace the oldest resident host", async () => { + const storage = new MemoryStorage(); + const replicaCache = cache(storage, { maxBytes: 7_000 }); + replicaCache.setHosts(LRU_SERVER_IDS.slice(0, 2)); + seedTimeline("host-a", "A".repeat(1_200)); + seedTimeline("host-b", "B".repeat(1_200)); + await replicaCache.flush(); + + seedTimeline("host-a", "A".repeat(1_201)); + await replicaCache.flush(); + replicaCache.setHosts(LRU_SERVER_IDS); + seedTimeline("host-c", "C".repeat(1_200)); + await replicaCache.flush(); + + seedTimeline("host-b", "B".repeat(1_201)); + await replicaCache.flush(); + + expect(new Set([...storage.rows.values()].map((row) => row.serverId))).toEqual( + new Set(["host-b", "host-c"]), + ); + }); + + it("persists an oversized host after its current input shrinks below the budget", async () => { + const storage = new MemoryStorage(); + const replicaCache = cache(storage, { maxBytes: 3_000 }); + replicaCache.setHosts([SERVER_ID]); + seedTimeline(SERVER_ID, "oversized".repeat(500)); + await replicaCache.flush(); + expect(storage.rows.size).toBe(0); + + seedTimeline(SERVER_ID, "small"); + await replicaCache.flush(); + + expect([...storage.rows.values()]).toEqual( + expect.arrayContaining([ + expect.objectContaining({ serverId: SERVER_ID, kind: "timeline", id: "singleton" }), + ]), + ); + }); + + it("keeps the newest row after two ordered write failures", async () => { + const storage = new MemoryStorage(); + const replicaCache = cache(storage); + replicaCache.setHosts([SERVER_ID]); + seedSession(); + await replicaCache.flush(); + storage.changes.length = 0; + const firstFailure = deferred(); + const secondFailure = deferred(); + storage.applyResults.push(firstFailure.promise, secondFailure.promise); + + useSessionStore.getState().setAgents(SERVER_ID, (agents) => { + const current = agents.get("agent-1"); + if (!current) throw new Error("Expected seeded agent"); + return new Map(agents).set(current.id, { ...current, title: "v1" }); + }); + const firstWrite = replicaCache.flush(); + await waitForWrites(storage, 2); + useSessionStore.getState().setAgents(SERVER_ID, (agents) => { + const current = agents.get("agent-1"); + if (!current) throw new Error("Expected seeded agent"); + return new Map(agents).set(current.id, { ...current, title: "v2" }); + }); + const secondWrite = replicaCache.flush(); + + firstFailure.reject(new Error("first failure")); + await waitForWrites(storage, 3); + secondFailure.reject(new Error("second failure")); + await Promise.all([firstWrite, secondWrite]); + await replicaCache.flush(); + + const persisted = [...storage.rows.values()].find( + (row) => row.serverId === SERVER_ID && row.kind === "agent" && row.id === "agent-1", + ); + expect(JSON.parse(persisted?.payload ?? "null").snapshot.title).toBe("v2"); + }); + + it("retries restored changes after a transient write failure", async () => { + vi.useFakeTimers(); + const storage = new MemoryStorage(); + const replicaCache = cache(storage); + replicaCache.setHosts([SERVER_ID]); + seedSession(); + await replicaCache.flush(); + const failure = deferred(); + storage.applyResults.push(failure.promise); + + useSessionStore.getState().setAgents(SERVER_ID, (agents) => { + const current = agents.get("agent-1"); + if (!current) throw new Error("Expected seeded agent"); + return new Map(agents).set(current.id, { ...current, title: "retried" }); + }); + const write = replicaCache.flush(); + await waitForWrites(storage, 2); + failure.reject(new Error("transient failure")); + await write; + + await vi.advanceTimersByTimeAsync(5_000); + await waitForWrites(storage, 3); + + const persisted = [...storage.rows.values()].find( + (row) => row.serverId === SERVER_ID && row.kind === "agent" && row.id === "agent-1", + ); + expect(JSON.parse(persisted?.payload ?? "null").snapshot.title).toBe("retried"); + }); + + it("does not restore a failed row write after its host was removed", async () => { + vi.useFakeTimers(); + const storage = new MemoryStorage(); + const replicaCache = cache(storage); + replicaCache.setHosts([SERVER_ID]); + seedSession(); + await replicaCache.flush(); + storage.changes.length = 0; + const failure = deferred(); + storage.applyResults.push(failure.promise); + + useSessionStore.getState().setAgents(SERVER_ID, (agents) => { + const current = agents.get("agent-1"); + if (!current) throw new Error("Expected seeded agent"); + return new Map(agents).set(current.id, { ...current, title: "removed" }); + }); + const write = replicaCache.flush(); + await waitForWrites(storage, 2); + replicaCache.setHosts([]); + failure.reject(new Error("failed after removal")); + await write; + await vi.advanceTimersByTimeAsync(5_000); + await replicaCache.flush(); + + expect([...storage.rows.values()].some((row) => row.serverId === SERVER_ID)).toBe(false); + expect(storage.changes).toEqual([]); + }); + it("clears the whole cache when one row payload is corrupt", async () => { const storage = new MemoryStorage(); storage.rows.set(`${SERVER_ID}:agent:agent-1`, { @@ -767,6 +1029,29 @@ describe("ReplicaCache", () => { expect(storage.clears).toBe(1); }); + it("clears every host when one host contains a corrupt row", async () => { + const storage = new MemoryStorage(); + storage.rows.set(`${SERVER_ID}:agent:agent-1`, { + serverId: SERVER_ID, + kind: "agent", + id: "agent-1", + payload: "{bad", + }); + storage.rows.set("second-host:checkpoint:singleton", { + serverId: "second-host", + kind: "checkpoint", + id: "singleton", + payload: JSON.stringify({ agents: { generation: "g", afterSeq: 1 } }), + }); + const replicaCache = cache(storage); + replicaCache.setHosts([SERVER_ID, "second-host"]); + + await replicaCache.restore(); + + expect(storage.rows.size).toBe(0); + expect(storage.clears).toBe(1); + }); + it("clears the whole cache when a valid JSON row has an unknown shape", async () => { const storage = new MemoryStorage(); storage.rows.set(`${SERVER_ID}:project:project-1`, { diff --git a/packages/app/src/runtime/replica-cache/index.ts b/packages/app/src/runtime/replica-cache/index.ts index bab3cf93f9..faa5c25081 100644 --- a/packages/app/src/runtime/replica-cache/index.ts +++ b/packages/app/src/runtime/replica-cache/index.ts @@ -728,7 +728,7 @@ export class ReplicaCache { private readonly storedRows = new Map>(); private readonly hostBytes = new Map(); private readonly hostWriteOrder = new Map(); - private readonly evictedServerIds = new Set(); + private readonly evictedHostBytes = new Map(); private readonly lastFocusedAgentIds = new Map(); private readonly capturedInputs = new Map(); private readonly directoryCheckpoints = new Map(); @@ -774,8 +774,8 @@ export class ReplicaCache { for (const serverId of this.checkpointInputs.keys()) { if (!next.has(serverId)) this.checkpointInputs.delete(serverId); } - for (const serverId of this.evictedServerIds) { - if (!next.has(serverId)) this.evictedServerIds.delete(serverId); + for (const serverId of this.evictedHostBytes.keys()) { + if (!next.has(serverId)) this.evictedHostBytes.delete(serverId); } } @@ -878,7 +878,11 @@ export class ReplicaCache { this.checkpointInputs.delete(oldServerId); this.checkpointInputs.set(newServerId, checkpoint); } - if (this.evictedServerIds.delete(oldServerId)) this.evictedServerIds.add(newServerId); + const evictedBytes = this.evictedHostBytes.get(oldServerId); + if (evictedBytes !== undefined) { + this.evictedHostBytes.delete(oldServerId); + this.evictedHostBytes.set(newServerId, evictedBytes); + } this.renamePendingHostChanges(oldServerId, newServerId); if (this.activeServerIds.delete(oldServerId)) this.activeServerIds.add(newServerId); this.queueOperation(() => this.rowStore.renameHost(oldServerId, newServerId)); @@ -890,13 +894,13 @@ export class ReplicaCache { writeDirectoryCheckpoint(serverId: string, checkpoint: unknown): void { if (this.checkpointInputs.get(serverId) === checkpoint) return; - if (this.evictedServerIds.has(serverId)) { - const session = useSessionStore.getState().sessions[serverId]; - if (!session) return; - this.captureHost(serverId, session); - } this.checkpointInputs.set(serverId, checkpoint); this.directoryCheckpoints.set(serverId, checkpoint); + if (this.evictedHostBytes.has(serverId)) { + const session = useSessionStore.getState().sessions[serverId]; + if (!session) return; + this.captureHost(serverId, session, true); + } const payload = JSON.stringify(checkpoint); if (payload === undefined) { this.queueDelete({ @@ -935,18 +939,21 @@ export class ReplicaCache { } const changed = this.captureSessions(); if (skipUnchanged && !changed && !this.hasPendingChanges()) return; - const changes = this.drainPendingChanges(); - if (changes.upserts.length === 0 && changes.deletes.length === 0) return; const write = this.writeQueue .catch(() => undefined) .then(async () => { + const changes = this.drainPendingChanges(); + if (changes.upserts.length === 0 && changes.deletes.length === 0) return; try { await this.prepareStore(); - await this.rowStore.apply(changes); - this.applyStoredChanges(changes); - await this.evictOverBudget(); + const boundedChanges = await this.fitChangesToBudget(changes); + if (boundedChanges.upserts.length > 0 || boundedChanges.deletes.length > 0) { + await this.rowStore.apply(boundedChanges); + this.applyStoredChanges(boundedChanges); + } } catch { this.restorePendingChanges(changes); + if (this.hasPendingChanges()) this.schedulePersist(); } return undefined; }); @@ -965,7 +972,7 @@ export class ReplicaCache { return changed; } - private captureHost(serverId: string, session: SessionState): boolean { + private captureHost(serverId: string, session: SessionState, force = false): boolean { if (session.focusedAgentId) { this.lastFocusedAgentIds.set(serverId, session.focusedAgentId); } @@ -984,11 +991,12 @@ export class ReplicaCache { } else if (hasTimelineHead) { input = { ...selected, timelineRange: null, timelineHasOlder: false }; } - const initialInput = this.evictedServerIds.has(serverId) ? undefined : previous; - const changes = diffReplicaInput(initialInput, input); + const changes = diffReplicaInput(previous, input); this.capturedInputs.set(serverId, input); - if (!hasReplicaInputChanges(changes)) return false; - this.queueReplicaChanges(serverId, input, changes); + const isEvicted = this.evictedHostBytes.has(serverId); + if (!hasReplicaInputChanges(changes) && !(force && isEvicted)) return false; + const persistedChanges = isEvicted ? diffReplicaInput(undefined, input) : changes; + this.queueReplicaChanges(serverId, input, persistedChanges); return true; } @@ -1036,7 +1044,7 @@ export class ReplicaCache { this.storedRows.clear(); this.hostBytes.clear(); this.hostWriteOrder.clear(); - this.evictedServerIds.clear(); + this.evictedHostBytes.clear(); this.directoryCheckpoints.clear(); this.checkpointInputs.clear(); this.capturedInputs.clear(); @@ -1143,11 +1151,26 @@ export class ReplicaCache { } private restorePendingChanges(changes: ReplicaRowChanges): void { - const newer = this.drainPendingChanges(); - for (const key of changes.deletes) this.queueDelete(key); - for (const row of changes.upserts) this.queueUpsert(row); - for (const key of newer.deletes) this.queueDelete(key); - for (const row of newer.upserts) this.queueUpsert(row); + for (const key of changes.deletes) { + const pendingKey = pendingRowKey(key); + if ( + this.activeServerIds.has(key.serverId) && + !this.pendingUpserts.has(pendingKey) && + !this.pendingDeletes.has(pendingKey) + ) { + this.queueDelete(key); + } + } + for (const row of changes.upserts) { + const pendingKey = pendingRowKey(row); + if ( + this.activeServerIds.has(row.serverId) && + !this.pendingUpserts.has(pendingKey) && + !this.pendingDeletes.has(pendingKey) + ) { + this.queueUpsert(row); + } + } } private applyStoredChanges(changes: ReplicaRowChanges): void { @@ -1168,7 +1191,7 @@ export class ReplicaCache { rows.set(rowKey(row), row); this.storedRows.set(row.serverId, rows); this.adjustHostBytes(row.serverId, payloadBytes(row.payload) - previousBytes); - this.evictedServerIds.delete(row.serverId); + this.evictedHostBytes.delete(row.serverId); touchedServerIds.add(row.serverId); } for (const serverId of touchedServerIds) { @@ -1194,14 +1217,64 @@ export class ReplicaCache { while (this.totalBytes > this.maxBytes) { const oldestServerId = this.hostWriteOrder.keys().next().value; if (oldestServerId === undefined) return; + const bytes = this.hostBytes.get(oldestServerId) ?? 0; await this.rowStore.deleteHost(oldestServerId); this.removeStoredHost(oldestServerId); - this.evictedServerIds.add(oldestServerId); + this.evictedHostBytes.set(oldestServerId, bytes); this.directoryCheckpoints.delete(oldestServerId); this.checkpointInputs.delete(oldestServerId); } } + private async fitChangesToBudget(changes: ReplicaRowChanges): Promise { + const touchedServerIds = new Set(); + for (const key of changes.deletes) touchedServerIds.add(key.serverId); + for (const row of changes.upserts) touchedServerIds.add(row.serverId); + + const projectedRows = new Map>(); + const projectedBytes = new Map(this.hostBytes); + for (const serverId of touchedServerIds) { + projectedRows.set(serverId, new Map(this.storedRows.get(serverId))); + } + for (const key of changes.deletes) { + projectedRows.get(key.serverId)?.delete(rowKey(key)); + } + for (const row of changes.upserts) { + projectedRows.get(row.serverId)?.set(rowKey(row), row); + } + for (const [serverId, rows] of projectedRows) { + projectedBytes.set( + serverId, + [...rows.values()].reduce((sum, row) => sum + payloadBytes(row.payload), 0), + ); + } + + const writeOrder = [...this.hostWriteOrder.keys()].filter( + (serverId) => !touchedServerIds.has(serverId), + ); + writeOrder.push(...touchedServerIds); + let projectedTotal = [...projectedBytes.values()].reduce((sum, bytes) => sum + bytes, 0); + const evicted = new Set(); + while (projectedTotal > this.maxBytes) { + const serverId = writeOrder.shift(); + if (serverId === undefined) break; + const bytes = projectedBytes.get(serverId) ?? 0; + projectedTotal -= bytes; + projectedBytes.delete(serverId); + evicted.add(serverId); + this.evictedHostBytes.set(serverId, bytes); + if (this.storedRows.has(serverId)) await this.rowStore.deleteHost(serverId); + this.removeStoredHost(serverId); + this.directoryCheckpoints.delete(serverId); + this.checkpointInputs.delete(serverId); + } + + return { + upserts: changes.upserts.filter((row) => !evicted.has(row.serverId)), + deletes: changes.deletes.filter((key) => !evicted.has(key.serverId)), + }; + } + private removeStoredHost(serverId: string): void { this.totalBytes -= this.hostBytes.get(serverId) ?? 0; this.storedRows.delete(serverId); diff --git a/packages/app/src/runtime/replica-cache/legacy-cleanup.web.test.ts b/packages/app/src/runtime/replica-cache/legacy-cleanup.web.test.ts index fb1fd7c37e..5adfbc6081 100644 --- a/packages/app/src/runtime/replica-cache/legacy-cleanup.web.test.ts +++ b/packages/app/src/runtime/replica-cache/legacy-cleanup.web.test.ts @@ -32,4 +32,17 @@ describe("clearLegacyReplicaCache", () => { reopened.database.close(); await clearLegacyReplicaCache(); }); + + it("reports a blocked deletion so a later launch can retry", async () => { + const open = await openLegacyDatabase(); + + await expect(clearLegacyReplicaCache()).rejects.toThrow("blocked"); + open.database.close(); + + await clearLegacyReplicaCache(); + const reopened = await openLegacyDatabase(); + expect(reopened.oldVersion).toBe(0); + reopened.database.close(); + await clearLegacyReplicaCache(); + }); }); diff --git a/packages/app/src/runtime/replica-cache/legacy-cleanup.web.ts b/packages/app/src/runtime/replica-cache/legacy-cleanup.web.ts index 5b15704535..fc8a9e4801 100644 --- a/packages/app/src/runtime/replica-cache/legacy-cleanup.web.ts +++ b/packages/app/src/runtime/replica-cache/legacy-cleanup.web.ts @@ -7,6 +7,8 @@ export function clearLegacyReplicaCache(): Promise { request.addEventListener("error", () => reject(request.error ?? new Error("Failed to delete legacy replica cache")), ); - request.addEventListener("blocked", () => resolve()); + request.addEventListener("blocked", () => + reject(new Error("Legacy replica cache deletion was blocked")), + ); }); } diff --git a/packages/app/src/stores/session-store.ts b/packages/app/src/stores/session-store.ts index fdae0941ed..d408525c15 100644 --- a/packages/app/src/stores/session-store.ts +++ b/packages/app/src/stores/session-store.ts @@ -205,21 +205,22 @@ function preserveWorkspaceDescriptorIdentity( return incoming; } -function preserveWorkspaceMapIdentity( - existing: Map, - incoming: Map, -): Map { +function preserveMapIdentity( + existing: Map, + incoming: Map, +): Map { if (existing === incoming) { return existing; } - const next = new Map(); + const next = new Map(); let changed = existing.size !== incoming.size; const existingEntries = existing.entries(); for (const [key, workspace] of incoming) { const existingWorkspace = existing.get(key); - const nextWorkspace = preserveWorkspaceDescriptorIdentity(workspace, existingWorkspace); + const nextWorkspace = + existingWorkspace && equal(existingWorkspace, workspace) ? existingWorkspace : workspace; next.set(key, nextWorkspace); const existingEntry = existingEntries.next().value; if (!existingEntry || existingEntry[0] !== key || existingEntry[1] !== nextWorkspace) { @@ -230,17 +231,6 @@ function preserveWorkspaceMapIdentity( return changed ? next : existing; } -function projectMapsEqual( - left: ReadonlyMap, - right: ReadonlyMap, -): boolean { - if (left.size !== right.size) return false; - for (const [projectId, project] of right) { - if (!equal(left.get(projectId), project)) return false; - } - return true; -} - export type ExplorerEntryKind = "file" | "directory"; export type ExplorerFileKind = "text" | "image" | "binary"; export type ExplorerEncoding = "utf-8" | "base64" | "none"; @@ -1659,10 +1649,7 @@ export const useSessionStore = create()( } const nextWorkspaces = typeof workspaces === "function" ? workspaces(session.workspaces) : workspaces; - const preservedWorkspaces = preserveWorkspaceMapIdentity( - session.workspaces, - nextWorkspaces, - ); + const preservedWorkspaces = preserveMapIdentity(session.workspaces, nextWorkspaces); if (session.workspaces === preservedWorkspaces) { return prev; } @@ -1681,10 +1668,15 @@ export const useSessionStore = create()( for (const project of projects) next.set(project.projectId, project); set((prev) => { const session = prev.sessions[serverId]; - if (!session || projectMapsEqual(session.projects, next)) return prev; + if (!session) return prev; + const preservedProjects = preserveMapIdentity(session.projects, next); + if (session.projects === preservedProjects) return prev; return { ...prev, - sessions: { ...prev.sessions, [serverId]: { ...session, projects: next } }, + sessions: { + ...prev.sessions, + [serverId]: { ...session, projects: preservedProjects }, + }, }; }); }, diff --git a/packages/app/src/utils/agent-directory-sync.ts b/packages/app/src/utils/agent-directory-sync.ts index 1e0f0f87c3..2d81158c81 100644 --- a/packages/app/src/utils/agent-directory-sync.ts +++ b/packages/app/src/utils/agent-directory-sync.ts @@ -1,3 +1,4 @@ +import equal from "fast-deep-equal"; import type { FetchAgentsEntry } from "@getpaseo/client/internal/daemon-client"; import { type Agent, useSessionStore } from "@/stores/session-store"; import { derivePendingPermissionKey, normalizeAgentSnapshot } from "@/utils/agent-snapshots"; @@ -187,13 +188,20 @@ export function replaceFetchedAgentDirectory(input: { } } - store.setAgents(input.serverId, fetchedAgents); - for (const agent of fetchedAgents.values()) { + const currentAgents = store.sessions[input.serverId]?.agents ?? new Map(); + const agents = new Map(); + for (const [agentId, fetchedAgent] of fetchedAgents) { + const current = currentAgents.get(agentId); + agents.set(agentId, current && equal(current, fetchedAgent) ? current : fetchedAgent); + } + + store.setAgents(input.serverId, agents); + for (const agent of agents.values()) { applyAgentTurnSnapshot(input.serverId, agent); } store.setAgentDetails(input.serverId, (prev) => { let next: Map | null = null; - for (const agentId of fetchedAgents.keys()) { + for (const agentId of agents.keys()) { if (!prev.has(agentId)) { continue; } @@ -204,12 +212,12 @@ export function replaceFetchedAgentDirectory(input: { }); const lastActivityByAgentId = new Map(); - for (const agent of fetchedAgents.values()) { + for (const agent of agents.values()) { lastActivityByAgentId.set(agent.id, agent.lastActivityAt); } store.setAgentLastActivityBatch(lastActivityByAgentId); store.setPendingPermissions(input.serverId, new Map(pendingPermissions)); store.setHasHydratedAgents(input.serverId, true); - return { agents: fetchedAgents }; + return { agents }; }