diff --git a/docs/architecture.md b/docs/architecture.md index 4dcf7a676..2946da8b4 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -69,6 +69,7 @@ not retain non-Git directories. | `server/bootstrap.ts` | Daemon initialization: HTTP server, WS server, agent manager, storage, relay | | `server/websocket-server.ts` | WebSocket connection management, hello handshake, binary frame routing | | `server/session.ts` | Per-client session state, timeline subscriptions, terminal operations | +| `server/directory-sync/` | Daemon-global latest-state sequences for projects, workspaces, and agents | | `server/agent/agent-manager.ts` | Agent lifecycle state machine, timeline tracking, subscriber management | | `server/agent/agent-storage.ts` | File-backed JSON persistence at `$PASEO_HOME/agents/` | | `server/agent/tools/` | Transport-neutral catalog for workspaces, agents, permissions, and automation | @@ -97,19 +98,27 @@ Cross-platform React Native app that connects to one or more daemons. - Expo Router navigation (`/h/[serverId]/workspace/[workspaceId]`, `/h/[serverId]/agent/[agentId]`, etc.). The `workspaceId` URL segment is an opaque workspace id, not a directly meaningful filesystem path. - `HostRuntimeController` manages saved host connections, reconnection, and per-host runtime state -- `runtime/replica-cache` keeps a non-authoritative per-host display replica in AsyncStorage: only the last focused agent, its workspace, and a short timeline tail. It restores before navigation becomes ready, leaves remote hydration flags false, and is atomically replaced by the normal snapshot-plus-delta synchronization path. +- `runtime/replica-cache` keeps the complete project, workspace, and active-agent directory plus one short focused timeline tail in AsyncStorage. It restores before navigation becomes ready and leaves remote hydration flags false. +- `runtime/directory-sync` owns directory reconciliation. On reconnect it passes the persisted per-entity cursor through `project.list`, `fetch_workspaces`, and `fetch_agents`; the daemon returns each entity's latest projection when its sequence is newer, plus tombstones. - `SessionContext` wraps the daemon client for the active session - Composer UI and submit/draft behavior live in `packages/app/src/composer/`; screens and panels should integrate it from there instead of dropping composer internals into `components/`, `hooks/`, or `screens/workspace/` - Timeline reducers in `timeline/session-stream-reducers.ts` handle compaction, gap detection, sequence-based deduplication - Timeline sync correctness is documented in [docs/timeline-sync.md](timeline-sync.md): live streams are for immediacy, `fetch_agent_timeline_request` is authoritative, and catch-up is paged but complete. - Voice features: dictation (STT) and voice agent (realtime) -The replica cache exists only to paint stale data immediately while the host connects. It does not -own mutations, infer deletions, or replace daemon reconciliation. Pending permission requests are -not restored from it. AsyncStorage is not encrypted, so the cached timeline tail may contain source -code, prompts, and tool output; encrypted-at-rest storage is a separate product/security decision. -Its serialized payload has a 1 MiB byte budget and evicts whole host snapshots in least-recently- -written order; a single oversized host is omitted rather than partially restored. +The replica cache paints stale data immediately while the host connects. Directory cursors are +reconciliation checkpoints; cached entities remain non-authoritative until the daemon answers. +Pending permission requests are not restored from it. AsyncStorage is not encrypted, so the cached +timeline tail may contain source code, prompts, and tool output; encrypted-at-rest storage is a +separate product/security decision. Its serialized payload has a 32 MiB byte budget and evicts whole +host snapshots in least-recently-written order; a single oversized host is omitted rather than +partially restored. Browser and Electron builds store it in IndexedDB. Native builds use +AsyncStorage, and Android reserves 64 MiB for that database. + +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 +event log. A missing, expired, or previous-generation cursor receives a full snapshot. Projects are +independent records; a project with no workspaces does not need a workspace placeholder. ### `packages/cli` — Command-line client diff --git a/packages/app/app.config.js b/packages/app/app.config.js index fef9d0e32..db7120882 100644 --- a/packages/app/app.config.js +++ b/packages/app/app.config.js @@ -139,7 +139,7 @@ export default { }, plugins: [ "expo-router", - [withAndroidAsyncStorageSize, 10], + [withAndroidAsyncStorageSize, 64], ...buildProfile.cameraPlugins, [ "expo-splash-screen", diff --git a/packages/app/e2e/browser/rewind-menu.ui-contract.spec.ts b/packages/app/e2e/browser/rewind-menu.ui-contract.spec.ts index e620b4d0f..ab1ce4e19 100644 --- a/packages/app/e2e/browser/rewind-menu.ui-contract.spec.ts +++ b/packages/app/e2e/browser/rewind-menu.ui-contract.spec.ts @@ -10,6 +10,7 @@ import { fillComposerDraft, submitMessage, } from "../support/helpers/composer"; +import { readReplicaCache, writeReplicaCache } from "../support/helpers/replica-cache-storage"; // UI plumbing contract against the dev mock provider. Real-provider behavior is tested in `daemon-e2e/*-rewind.real.e2e.test.ts`. @@ -27,39 +28,28 @@ async function expectUserMessageVisible(page: Page, text: string): Promise async function rewriteCachedMessageAsLegacyRow(page: Page, prompt: string): Promise { await expect - .poll(() => - page.evaluate((messageText) => { - const raw = localStorage.getItem("@paseo:replica-cache"); - if (!raw) return false; - const cache = JSON.parse(raw) as { - hosts?: Array<{ timeline?: { items?: Array> } | null }>; - }; - for (const host of cache.hosts ?? []) { - for (const item of host.timeline?.items ?? []) { - if (item.kind === "user_message" && item.text === messageText && item.messageId) { - return true; - } + .poll(async () => { + const cache = await readReplicaCache(page); + if (!cache) return false; + for (const host of cache.hosts ?? []) { + for (const item of host.timeline?.items ?? []) { + if (item.kind === "user_message" && item.text === prompt && item.messageId) { + return true; } } - return false; - }, prompt), - ) + } + return false; + }) .toBe(true); - await page.evaluate((messageText) => { - const key = "@paseo:replica-cache"; - const raw = localStorage.getItem(key); - if (!raw) throw new Error("Replica cache was not persisted"); - const cache = JSON.parse(raw) as { - hosts?: Array<{ timeline?: { items?: Array> } | null }>; - }; - const cachedMessage = cache.hosts - ?.flatMap((host) => host.timeline?.items ?? []) - .find((item) => item.kind === "user_message" && item.text === messageText); - if (!cachedMessage) throw new Error("Cached user message was not found"); - delete cachedMessage.messageId; - localStorage.setItem(key, JSON.stringify(cache)); - }, prompt); + const cache = await readReplicaCache(page); + if (!cache) throw new Error("Replica cache was not persisted"); + const cachedMessage = cache.hosts + ?.flatMap((host) => host.timeline?.items ?? []) + .find((item) => item.kind === "user_message" && item.text === prompt); + if (!cachedMessage) throw new Error("Cached user message was not found"); + delete cachedMessage.messageId; + await writeReplicaCache(page, cache); } async function waitForCurrentSubmissionExcludedFromCache( @@ -67,46 +57,34 @@ async function waitForCurrentSubmissionExcludedFromCache( prompt: string, ): Promise { await expect - .poll(() => - page.evaluate((messageText) => { - const raw = localStorage.getItem("@paseo:replica-cache"); - if (!raw) return false; - const cache = JSON.parse(raw) as { - hosts?: Array<{ timeline?: { items?: Array> } | null }>; - }; - return !cache.hosts - ?.flatMap((host) => host.timeline?.items ?? []) - .some( - (item) => - item.kind === "user_message" && - item.text === messageText && - typeof item.clientMessageId === "string" && - item.messageId === undefined, - ); - }, prompt), - ) + .poll(async () => { + const cache = await readReplicaCache(page); + if (!cache) return false; + return !cache.hosts + ?.flatMap((host) => host.timeline?.items ?? []) + .some( + (item) => + item.kind === "user_message" && + item.text === prompt && + typeof item.clientMessageId === "string" && + item.messageId === undefined, + ); + }) .toBe(true); } async function waitForCachedMessageWithoutProviderId(page: Page, prompt: string): Promise { await expect - .poll(() => - page.evaluate((messageText) => { - const raw = localStorage.getItem("@paseo:replica-cache"); - if (!raw) return false; - const cache = JSON.parse(raw) as { - hosts?: Array<{ timeline?: { items?: Array> } | null }>; - }; - return cache.hosts - ?.flatMap((host) => host.timeline?.items ?? []) - .some( - (item) => - item.kind === "user_message" && - item.text === messageText && - item.messageId === undefined, - ); - }, prompt), - ) + .poll(async () => { + const cache = await readReplicaCache(page); + if (!cache) return false; + return cache.hosts + ?.flatMap((host) => host.timeline?.items ?? []) + .some( + (item) => + item.kind === "user_message" && item.text === prompt && item.messageId === undefined, + ); + }) .toBe(true); } @@ -148,16 +126,13 @@ test.describe("Rewind sheet", () => { title: "Rewind cache upgrade e2e", initialPrompt: prompt, }); - let heldTimelineRequest = false; try { await openAgentRoute(page, session); await expectUserMessageVisible(page, prompt); + await gate.drop(); await rewriteCachedMessageAsLegacyRow(page, prompt); - gate.holdNextClientRequest("fetch_agent_timeline_request"); await page.reload(); - await gate.waitForHeldClientRequest(); - heldTimelineRequest = true; const restoredMessage = userMessage(page, prompt); await expect(restoredMessage).toBeVisible(); @@ -165,7 +140,6 @@ test.describe("Rewind sheet", () => { await expect(restoredMessage.getByTestId("rewind-menu-trigger")).toHaveCount(0); await waitForCachedMessageWithoutProviderId(page, prompt); } finally { - if (heldTimelineRequest) gate.releaseHeldClientRequest(); gate.restore(); await session.cleanup(); } diff --git a/packages/app/e2e/support/helpers/replica-cache-perf.ts b/packages/app/e2e/support/helpers/replica-cache-perf.ts index 834a69475..5b48b02b9 100644 --- a/packages/app/e2e/support/helpers/replica-cache-perf.ts +++ b/packages/app/e2e/support/helpers/replica-cache-perf.ts @@ -1,6 +1,9 @@ import type { Page } from "@playwright/test"; - -const REPLICA_CACHE_STORAGE_KEY = "@paseo:replica-cache"; +import { + observeReplicaCacheStorageWrites, + readReplicaCacheStorageWriteObserver, + resetReplicaCacheStorageWriteObserver, +} from "./replica-cache-storage"; export interface ReplicaCachePerfReport { elapsedMs: number; @@ -11,12 +14,7 @@ export interface ReplicaCachePerfReport { } interface ReplicaCachePerfState { - lastValue: string | null; resetAt: number; - redundantWrites: number; - serializedChars: number; - storageDurationMs: number; - writes: number; } declare global { @@ -26,55 +24,25 @@ declare global { } export async function observeReplicaCacheWrites(page: Page): Promise { - 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); + await observeReplicaCacheStorageWrites(page); + await page.addInitScript(() => { + window.__replicaCachePerf = { resetAt: performance.now() }; + }); } export async function beginReplicaCacheMeasurement(page: Page): Promise { + await resetReplicaCacheStorageWriteObserver(page); await page.evaluate(() => { - const lastValue = window.__replicaCachePerf?.lastValue ?? null; - window.__replicaCachePerf = { - lastValue, - resetAt: performance.now(), - redundantWrites: 0, - serializedChars: 0, - storageDurationMs: 0, - writes: 0, - }; + window.__replicaCachePerf = { resetAt: performance.now() }; }); } export async function readReplicaCacheMeasurement(page: Page): Promise { - return page.evaluate(() => { + const storage = await readReplicaCacheStorageWriteObserver(page); + const elapsedMs = await 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, - }; + return performance.now() - state.resetAt; }); + return { elapsedMs, ...storage }; } diff --git a/packages/app/e2e/support/helpers/replica-cache-storage.ts b/packages/app/e2e/support/helpers/replica-cache-storage.ts new file mode 100644 index 000000000..8d42be348 --- /dev/null +++ b/packages/app/e2e/support/helpers/replica-cache-storage.ts @@ -0,0 +1,150 @@ +import type { Page } from "@playwright/test"; + +const DATABASE_NAME = "paseo-replica-cache"; +const STORE_NAME = "key-value"; +const STORAGE_KEY = "@paseo:replica-cache"; + +export interface ReplicaCacheRecord { + hosts?: Array<{ + timeline?: { + agentId?: string; + items?: Array>; + } | null; + }>; +} + +interface ReplicaCacheWriteObserverState { + lastValue: string | null; + redundantWrites: number; + serializedChars: number; + storageDurationMs: number; + writes: number; +} + +declare global { + interface Window { + __replicaCacheWriteObserver?: ReplicaCacheWriteObserverState; + } +} + +async function readStoredValue(input: { + databaseName: string; + storeName: string; + storageKey: string; +}): Promise { + const database = await new Promise((resolve, reject) => { + const request = indexedDB.open(input.databaseName, 1); + request.addEventListener("success", () => resolve(request.result)); + request.addEventListener("error", () => reject(request.error)); + }); + return new Promise((resolve, reject) => { + const request = database + .transaction(input.storeName, "readonly") + .objectStore(input.storeName) + .get(input.storageKey); + request.addEventListener("success", () => + resolve(typeof request.result === "string" ? request.result : null), + ); + request.addEventListener("error", () => reject(request.error)); + }); +} + +export async function readReplicaCache(page: Page): Promise { + const raw = await page.evaluate(readStoredValue, { + databaseName: DATABASE_NAME, + storeName: STORE_NAME, + storageKey: STORAGE_KEY, + }); + return raw ? (JSON.parse(raw) as ReplicaCacheRecord) : null; +} + +export async function writeReplicaCache(page: Page, value: ReplicaCacheRecord): Promise { + await page.evaluate( + async ({ databaseName, storeName, storageKey, raw }) => { + const database = await new Promise((resolve, reject) => { + const request = indexedDB.open(databaseName, 1); + request.addEventListener("success", () => resolve(request.result)); + request.addEventListener("error", () => reject(request.error)); + }); + await new Promise((resolve, reject) => { + const transaction = database.transaction(storeName, "readwrite"); + transaction.objectStore(storeName).put(raw, storageKey); + transaction.addEventListener("complete", () => resolve()); + transaction.addEventListener("error", () => reject(transaction.error)); + transaction.addEventListener("abort", () => reject(transaction.error)); + }); + }, + { + databaseName: DATABASE_NAME, + storeName: STORE_NAME, + storageKey: STORAGE_KEY, + raw: JSON.stringify(value), + }, + ); +} + +export async function observeReplicaCacheStorageWrites(page: Page): Promise { + await page.addInitScript( + ({ databaseName, storeName, storageKey }) => { + window.__replicaCacheWriteObserver = { + lastValue: null, + redundantWrites: 0, + serializedChars: 0, + storageDurationMs: 0, + writes: 0, + }; + 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 || + this.name !== storeName || + key !== storageKey || + typeof value !== "string" + ) { + return request; + } + const startedAt = performance.now(); + request.addEventListener("success", () => { + const state = window.__replicaCacheWriteObserver; + 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; + }); + return request; + }; + }, + { databaseName: DATABASE_NAME, storeName: STORE_NAME, storageKey: STORAGE_KEY }, + ); +} + +export async function resetReplicaCacheStorageWriteObserver(page: Page): Promise { + const lastValue = await page.evaluate(readStoredValue, { + databaseName: DATABASE_NAME, + storeName: STORE_NAME, + storageKey: STORAGE_KEY, + }); + await page.evaluate((value) => { + window.__replicaCacheWriteObserver = { + lastValue: value, + redundantWrites: 0, + serializedChars: 0, + storageDurationMs: 0, + writes: 0, + }; + }, lastValue); +} + +export async function readReplicaCacheStorageWriteObserver( + page: Page, +): Promise { + return page.evaluate(() => { + const state = window.__replicaCacheWriteObserver; + if (!state) throw new Error("Replica cache write observer is not installed"); + return state; + }); +} diff --git a/packages/app/e2e/support/helpers/timeline-pagination.ts b/packages/app/e2e/support/helpers/timeline-pagination.ts index 8acdcac3b..f7c972295 100644 --- a/packages/app/e2e/support/helpers/timeline-pagination.ts +++ b/packages/app/e2e/support/helpers/timeline-pagination.ts @@ -1,5 +1,6 @@ import { expect, type Page } from "@playwright/test"; import { buildAgentRoute, seedMockAgentWorkspace, type MockAgentWorkspace } from "./mock-agent"; +import { readReplicaCache } from "./replica-cache-storage"; import { delayAgentBootstrapTailResponse, delayAgentOlderTimelineResponse, @@ -284,22 +285,13 @@ export async function reloadAgentTimelineFromPersistedReplica( agent: LongTimelineAgent, ): Promise { await expect - .poll(() => - page.evaluate((agentId) => { - const raw = localStorage.getItem("@paseo:replica-cache"); - if (!raw) return false; - const cache = JSON.parse(raw) as { - hosts?: Array<{ - timeline?: { - agentId?: string; - items?: unknown[]; - } | null; - }>; - }; - const timeline = cache.hosts?.find((host) => host.timeline?.agentId === agentId)?.timeline; - return timeline?.items?.length === 50; - }, agent.agentId), - ) + .poll(async () => { + const cache = await readReplicaCache(page); + const timeline = cache?.hosts?.find( + (host) => host.timeline?.agentId === agent.agentId, + )?.timeline; + return timeline?.items?.length === 50; + }) .toBe(true); await page.reload(); diff --git a/packages/app/src/android-async-storage-config.test.ts b/packages/app/src/android-async-storage-config.test.ts index 03091afbb..fa49bbf83 100644 --- a/packages/app/src/android-async-storage-config.test.ts +++ b/packages/app/src/android-async-storage-config.test.ts @@ -3,7 +3,7 @@ import path from "node:path"; import { describe, expect, it } from "vitest"; describe("Android AsyncStorage configuration", () => { - it("writes a 10 MiB database limit into Gradle properties", () => { + it("writes a 64 MiB database limit into Gradle properties", () => { const appDirectory = path.resolve(import.meta.dirname, ".."); const output = execFileSync( process.execPath, @@ -11,7 +11,7 @@ describe("Android AsyncStorage configuration", () => { "-e", ` const plugin = require("./plugins/with-android-async-storage-size"); - const config = plugin({ name: "Paseo", slug: "paseo" }, 10); + const config = plugin({ name: "Paseo", slug: "paseo" }, 64); const mod = config.mods.android.gradleProperties; mod({ ...config, modResults: [], modRequest: {}, modRawConfig: config }) .then((result) => process.stdout.write(JSON.stringify(result.modResults))); @@ -21,7 +21,7 @@ describe("Android AsyncStorage configuration", () => { ); expect(JSON.parse(output)).toEqual([ - { type: "property", key: "AsyncStorage_db_size_in_MB", value: "10" }, + { type: "property", key: "AsyncStorage_db_size_in_MB", value: "64" }, ]); }); }); diff --git a/packages/app/src/components/project-edit-sheet.tsx b/packages/app/src/components/project-edit-sheet.tsx index 65801dee2..a2790cb43 100644 --- a/packages/app/src/components/project-edit-sheet.tsx +++ b/packages/app/src/components/project-edit-sheet.tsx @@ -40,7 +40,6 @@ export interface ProjectEditSheetProps { export function ProjectEditSheet({ visible, onClose, - serverId, projectId, projectViewKey, client, @@ -60,7 +59,6 @@ export function ProjectEditSheet({ submitProjectEdit({ client, projectId, submission }), onSuccess: () => { queryClient.invalidateQueries({ queryKey: ["projects"] }); - queryClient.invalidateQueries({ queryKey: ["projectIcon", serverId] }); toast.show(t("settings.project.edit.savedToast"), { variant: "success" }); onClose(); }, diff --git a/packages/app/src/hooks/use-sidebar-workspaces-list.ts b/packages/app/src/hooks/use-sidebar-workspaces-list.ts index 38e9d84ff..279d838ea 100644 --- a/packages/app/src/hooks/use-sidebar-workspaces-list.ts +++ b/packages/app/src/hooks/use-sidebar-workspaces-list.ts @@ -2,7 +2,10 @@ import { useCallback, useEffect, useMemo } from "react"; import { useStoreWithEqualityFn } from "zustand/traditional"; import { useCreateFlowStore } from "@/stores/create-flow-store"; import { useSessionStore } from "@/stores/session-store"; -import { useHydratedWorkspaceServerIds } from "@/stores/session-store-hooks"; +import { + useHydratedWorkspaceServerIds, + useWorkspaceDirectoryServerIds, +} from "@/stores/session-store-hooks"; import { workspaceEqualityFns } from "@/stores/session-store-hooks/selectors"; import { useHostProjects } from "@/projects/host-projects"; import { getHostRuntimeStore, useHostRegistryLoaded, useHosts } from "@/runtime/host-runtime"; @@ -132,8 +135,9 @@ export function useSidebarWorkspacesList(options?: { const persistedProjectOrder = useSidebarOrderStore((state) => state.projectOrder ?? EMPTY_ORDER); const hydratedServerIds = useHydratedWorkspaceServerIds(serverIds); + const directoryServerIds = useWorkspaceDirectoryServerIds(serverIds); - const hostProjects = useHostProjects(hydratedServerIds); + const hostProjects = useHostProjects(directoryServerIds); const sidebarModel = useMemo( () => diff --git a/packages/app/src/projects/icon-cache.test.ts b/packages/app/src/projects/icon-cache.test.ts new file mode 100644 index 000000000..92df4c409 --- /dev/null +++ b/packages/app/src/projects/icon-cache.test.ts @@ -0,0 +1,68 @@ +import { describe, expect, it } from "vitest"; +import type { DaemonClient } from "@getpaseo/client/internal/daemon-client"; +import { ProjectIconCache, type ProjectIconCacheStorage } from "./icon-cache"; + +class MemoryStorage implements ProjectIconCacheStorage { + readonly values = new Map(); + async getItem(key: string): Promise { + return this.values.get(key) ?? null; + } + async setItem(key: string, value: string): Promise { + this.values.set(key, value); + } +} + +const target = { + serverId: "host-a", + projectId: "project-a", + iconWorkingDir: "/project-a", + iconRevision: "revision-a", +}; +const icon = { data: "aWNvbg==", mimeType: "image/png" }; + +function clientWith(iconResult: typeof icon | null) { + return { + getProjectIcon: async () => ({ requestId: "icon", icon: iconResult }), + } as unknown as DaemonClient; +} + +describe("ProjectIconCache", () => { + it("resolves, persists, and hydrates an icon before a host connects", async () => { + const storage = new MemoryStorage(); + const writer = new ProjectIconCache(storage); + writer.setHosts([target.serverId]); + await writer.query(target, true, () => clientWith(icon), true).queryFn(); + await writer.flush(); + + const reader = new ProjectIconCache(storage); + reader.setHosts([target.serverId]); + await reader.restore(); + + expect(reader.query(target, true, () => null, false).initialData).toEqual(icon); + }); + + it("persists an explicit no-icon result", async () => { + const storage = new MemoryStorage(); + const writer = new ProjectIconCache(storage); + writer.setHosts([target.serverId]); + await writer.query(target, true, () => clientWith(null), true).queryFn(); + await writer.flush(); + + const reader = new ProjectIconCache(storage); + reader.setHosts([target.serverId]); + await reader.restore(); + expect(reader.query(target, true, () => null, false)).toHaveProperty("initialData", null); + }); + + it("misses only the project whose revision changed", async () => { + const cache = new ProjectIconCache(new MemoryStorage()); + const other = { ...target, projectId: "project-b" }; + await cache.query(target, true, () => clientWith(icon), true).queryFn(); + await cache.query(other, true, () => clientWith(null), true).queryFn(); + + expect( + cache.query({ ...target, iconRevision: "revision-b" }, true, () => null, false), + ).not.toHaveProperty("initialData"); + expect(cache.query(other, true, () => null, false)).toHaveProperty("initialData", null); + }); +}); diff --git a/packages/app/src/projects/icon-cache.ts b/packages/app/src/projects/icon-cache.ts new file mode 100644 index 000000000..93761708c --- /dev/null +++ b/packages/app/src/projects/icon-cache.ts @@ -0,0 +1,217 @@ +import type { ProjectIcon } from "@getpaseo/protocol/messages"; +import type { DaemonClient } from "@getpaseo/client/internal/daemon-client"; +import { replicaCacheStorage } from "@/runtime/replica-cache/storage"; + +const STORAGE_KEY = "@paseo:project-icon-cache"; +const CACHE_VERSION = 1; +const PERSIST_DELAY_MS = 250; +const MAX_ENTRIES = 512; + +export interface ProjectIconCacheStorage { + getItem(key: string): Promise; + setItem(key: string, value: string): Promise; +} + +interface ProjectIconIdentity { + serverId: string; + projectId: string; + revision: string; +} + +type ProjectIconCacheRead = { hit: true; icon: ProjectIcon | null } | { hit: false }; + +interface ProjectIconTarget { + serverId: string; + projectId: string; + iconWorkingDir: string; + customIconRevision?: string | null; + iconRevision?: string; +} + +interface StoredIcon extends ProjectIconIdentity { + icon: ProjectIcon | null; +} + +function keyOf(identity: ProjectIconIdentity): string { + return JSON.stringify([identity.serverId, identity.projectId, identity.revision]); +} + +function parseStoredIcon(value: unknown): StoredIcon | null { + if (!value || typeof value !== "object") return null; + const entry = value as Record; + if ( + typeof entry.serverId !== "string" || + typeof entry.projectId !== "string" || + typeof entry.revision !== "string" + ) { + return null; + } + if (entry.icon === null) { + return { + serverId: entry.serverId, + projectId: entry.projectId, + revision: entry.revision, + icon: null, + }; + } + if (!entry.icon || typeof entry.icon !== "object") return null; + const icon = entry.icon as Record; + if (typeof icon.data !== "string" || typeof icon.mimeType !== "string") return null; + return { + serverId: entry.serverId, + projectId: entry.projectId, + revision: entry.revision, + icon: { data: icon.data, mimeType: icon.mimeType }, + }; +} + +/** Owns project-icon persistence, revision matching, and negative results. */ +export class ProjectIconCache { + private readonly entries = new Map(); + private activeServerIds = new Set(); + private persistTimer: ReturnType | null = null; + private writeQueue: Promise = Promise.resolve(); + + constructor(private readonly storage: ProjectIconCacheStorage) {} + + setHosts(serverIds: Iterable): void { + this.activeServerIds = new Set(serverIds); + let changed = false; + for (const [key, entry] of this.entries) { + if (!this.activeServerIds.has(entry.serverId)) { + this.entries.delete(key); + changed = true; + } + } + if (changed) this.schedulePersist(); + } + + async restore(): Promise { + let raw: string | null; + try { + raw = await this.storage.getItem(STORAGE_KEY); + } catch { + return; + } + if (!raw) return; + try { + const parsed = JSON.parse(raw) as { version?: unknown; entries?: unknown }; + if (parsed.version !== CACHE_VERSION || !Array.isArray(parsed.entries)) return; + for (const value of parsed.entries) { + const entry = parseStoredIcon(value); + if (entry && this.activeServerIds.has(entry.serverId)) { + this.entries.set(keyOf(entry), entry); + } + } + } catch { + // A corrupt cache is disposable. + } + } + + private read(identity: ProjectIconIdentity): ProjectIconCacheRead { + const entry = this.entries.get(keyOf(identity)); + return entry ? { hit: true, icon: entry.icon } : { hit: false }; + } + + private write(identity: ProjectIconIdentity, icon: ProjectIcon | null): void { + const key = keyOf(identity); + this.entries.delete(key); + this.entries.set(key, { ...identity, icon }); + while (this.entries.size > MAX_ENTRIES) { + const oldest = this.entries.keys().next().value; + if (oldest === undefined) break; + this.entries.delete(oldest); + } + this.schedulePersist(); + } + + query( + target: ProjectIconTarget, + supportsCustomIcons: boolean | null, + getClient: () => DaemonClient | null, + connected: boolean, + ) { + // COMPAT(projectIconCache): daemons before v0.2.7 do not expose an effective revision, so + // their icon results remain memory-only. + const revision = target.iconRevision ?? target.customIconRevision ?? "automatic"; + const identity: ProjectIconIdentity | null = target.iconRevision + ? { serverId: target.serverId, projectId: target.projectId, revision: target.iconRevision } + : null; + const cached = identity ? this.read(identity) : { hit: false as const }; + const lookup = resolveLookup(target, supportsCustomIcons); + let queryKey: readonly string[]; + if (!lookup) { + queryKey = ["projectIcon", target.serverId, "pending", target.projectId]; + } else if (lookup.kind === "project") { + queryKey = ["projectIcon", target.serverId, target.projectId, revision]; + } else { + queryKey = ["projectIcon", target.serverId, "legacy", lookup.cwd]; + } + return { + queryKey, + queryFn: async () => { + if (!lookup) return null; + const client = getClient(); + if (!client) return null; + const result = + lookup.kind === "project" + ? await client.getProjectIcon(lookup.projectId) + : await client.requestProjectIcon(lookup.cwd); + if (result.error) throw new Error(result.error); + if (identity) this.write(identity, result.icon); + return result.icon; + }, + enabled: Boolean(lookup && getClient() && connected), + staleTime: Infinity, + gcTime: 1000 * 60 * 60, + refetchOnMount: false as const, + refetchOnWindowFocus: false as const, + refetchOnReconnect: false as const, + ...(cached.hit ? { initialData: cached.icon } : {}), + }; + } + + reconcileServerId(oldServerId: string, newServerId: string): void { + for (const [key, entry] of this.entries) { + if (entry.serverId !== oldServerId) continue; + this.entries.delete(key); + const reconciled = { ...entry, serverId: newServerId }; + this.entries.set(keyOf(reconciled), reconciled); + } + if (this.activeServerIds.delete(oldServerId)) this.activeServerIds.add(newServerId); + this.schedulePersist(); + } + + async flush(): Promise { + if (this.persistTimer) { + clearTimeout(this.persistTimer); + this.persistTimer = null; + } + const payload = JSON.stringify({ version: CACHE_VERSION, entries: [...this.entries.values()] }); + const write = this.writeQueue + .catch(() => undefined) + .then(() => this.storage.setItem(STORAGE_KEY, payload)); + this.writeQueue = write; + await write.catch(() => undefined); + } + + private schedulePersist(): void { + if (this.persistTimer) return; + this.persistTimer = setTimeout(() => { + this.persistTimer = null; + void this.flush(); + }, PERSIST_DELAY_MS); + } +} + +export const projectIconCache = new ProjectIconCache(replicaCacheStorage); + +function resolveLookup( + target: Pick, + supportsCustomIcons: boolean | null, +): { kind: "project"; projectId: string } | { kind: "legacy"; cwd: string } | null { + if (supportsCustomIcons === null) return null; + return supportsCustomIcons + ? { kind: "project", projectId: target.projectId } + : { kind: "legacy", cwd: target.iconWorkingDir }; +} diff --git a/packages/app/src/projects/icons.test.ts b/packages/app/src/projects/icons.test.ts index 48b99df4a..e13b468cb 100644 --- a/packages/app/src/projects/icons.test.ts +++ b/packages/app/src/projects/icons.test.ts @@ -20,4 +20,8 @@ describe("project icon lookup", () => { cwd: "/projects/paseo", }); }); + + it("waits while custom-icon capability is unknown", () => { + expect(resolveProjectIconLookup(target, null)).toBeNull(); + }); }); diff --git a/packages/app/src/projects/icons.ts b/packages/app/src/projects/icons.ts index f94fdc0f4..92bef909a 100644 --- a/packages/app/src/projects/icons.ts +++ b/packages/app/src/projects/icons.ts @@ -2,7 +2,8 @@ import { useMemo, useRef } from "react"; import { useTranslation } from "react-i18next"; import { useQueries, useQuery } from "@tanstack/react-query"; import type { ProjectIcon } from "@getpaseo/protocol/messages"; -import { useHostFeatureMap } from "@/runtime/host-features"; +import { useHostFeatureAvailabilityMap } from "@/runtime/host-features"; +import { projectIconCache } from "@/projects/icon-cache"; import { getHostRuntimeStore, isHostRuntimeConnected, @@ -16,6 +17,7 @@ interface ProjectIconTarget { projectId: string; iconWorkingDir: string; customIconRevision?: string | null; + iconRevision?: string; } /** @@ -24,8 +26,9 @@ interface ProjectIconTarget { */ export function resolveProjectIconLookup( target: Pick, - supportsCustomIcons: boolean, -): { kind: "project"; projectId: string } | { kind: "legacy"; cwd: string } { + supportsCustomIcons: boolean | null, +): { kind: "project"; projectId: string } | { kind: "legacy"; cwd: string } | null { + if (supportsCustomIcons === null) return null; return supportsCustomIcons ? { kind: "project", projectId: target.projectId } : { kind: "legacy", cwd: target.iconWorkingDir }; @@ -35,10 +38,6 @@ function legacyIconQueryKey(serverId: string, cwd: string) { return ["projectIcon", serverId, "legacy", cwd] as const; } -function iconQueryKey(serverId: string, projectId: string, revision: string) { - return ["projectIcon", serverId, projectId, revision] as const; -} - function iconDataUri(icon: ProjectIcon | null): string | null { if (!icon) return null; return `data:${icon.mimeType};base64,${icon.data}`; @@ -88,7 +87,7 @@ export function useProjectIcons(input: { () => [...new Set(input.projects.map((project) => project.serverId))], [input.projects], ); - const supportsCustomIcons = useHostFeatureMap(serverIds, "projectCustomIcon"); + const supportsCustomIcons = useHostFeatureAvailabilityMap(serverIds, "projectCustomIcon"); const requests = useMemo(() => { const unique = new Map(); for (const project of input.projects) { @@ -100,35 +99,14 @@ export function useProjectIcons(input: { const queries = useQueries({ queries: requests.map((request) => { - const revision = request.customIconRevision ?? "automatic"; - const lookup = resolveProjectIconLookup( - request, - supportsCustomIcons.get(request.serverId) === true, - ); return { - queryKey: - lookup.kind === "project" - ? iconQueryKey(request.serverId, lookup.projectId, revision) - : legacyIconQueryKey(request.serverId, lookup.cwd), - queryFn: async () => { - const client = getHostRuntimeStore().getClient(request.serverId); - if (!client) return null; - const result = - lookup.kind === "project" - ? await client.getProjectIcon(lookup.projectId) - : await client.requestProjectIcon(lookup.cwd); - return result.icon; - }, - select: iconDataUri, - enabled: Boolean( - getHostRuntimeStore().getClient(request.serverId) && + ...projectIconCache.query( + request, + supportsCustomIcons.get(request.serverId) ?? null, + () => getHostRuntimeStore().getClient(request.serverId), isHostRuntimeConnected(getHostRuntimeStore().getSnapshot(request.serverId)), ), - staleTime: Infinity, - gcTime: 1000 * 60 * 60, - refetchOnMount: false, - refetchOnWindowFocus: false, - refetchOnReconnect: false, + select: iconDataUri, }; }), }); diff --git a/packages/app/src/projects/workspace-structure.ts b/packages/app/src/projects/workspace-structure.ts index 6a70f21fb..939e1381d 100644 --- a/packages/app/src/projects/workspace-structure.ts +++ b/packages/app/src/projects/workspace-structure.ts @@ -7,6 +7,7 @@ export interface WorkspaceStructureHostPlacement { iconWorkingDir: string; worktreeSupport: "supported" | "unsupported" | "unknown"; customIconRevision?: string | null; + iconRevision?: string; } export interface WorkspaceStructureProject { @@ -160,6 +161,7 @@ function addProjectToView(input: { iconWorkingDir: project.projectRootPath, worktreeSupport: project.projectKind === "git" ? "supported" : "unsupported", customIconRevision: project.projectCustomIconRevision, + iconRevision: project.projectIconRevision, }; const draft = byProject.get(viewKey); if (!draft) { diff --git a/packages/app/src/runtime/directory-sync/agent-replica.test.ts b/packages/app/src/runtime/directory-sync/agent-replica.test.ts index 4456ecc0a..d798282aa 100644 --- a/packages/app/src/runtime/directory-sync/agent-replica.test.ts +++ b/packages/app/src/runtime/directory-sync/agent-replica.test.ts @@ -51,6 +51,31 @@ function entry(agent: AgentSnapshotPayload): FetchAgentsEntry { } describe("AgentDirectoryReplica", () => { + it("preserves an unchanged running agent's turn identity during catch-up", () => { + const serverId = "agent-replica-catch-up"; + const store = useSessionStore.getState(); + store.initializeSession(serverId, null as unknown as DaemonClient); + const replica = new AgentDirectoryReplica(serverId, () => undefined); + replica.commitSnapshot( + [ + entry({ + ...payload("running"), + status: "running", + activeTurn: { turnId: "turn-1", startedAt: "2026-07-17T00:01:00.000Z" }, + }), + ], + [], + ); + + replica.commitChanges([], [], []); + + expect(useSessionStore.getState().sessions[serverId]?.agents.get("agent")?.activeTurn).toEqual({ + turnId: "turn-1", + startedAt: new Date("2026-07-17T00:01:00.000Z"), + }); + store.clearSession(serverId); + }); + it("keeps membership authoritative across remove, stale timeline, and re-add", () => { const serverId = "agent-replica"; const store = useSessionStore.getState(); diff --git a/packages/app/src/runtime/directory-sync/agent-replica.ts b/packages/app/src/runtime/directory-sync/agent-replica.ts index 62de843cd..7f90998e5 100644 --- a/packages/app/src/runtime/directory-sync/agent-replica.ts +++ b/packages/app/src/runtime/directory-sync/agent-replica.ts @@ -3,7 +3,7 @@ import type { AgentSnapshotPayload } from "@getpaseo/protocol/messages"; import { clearArchiveAgentPending } from "@/hooks/use-archive-agent"; import { queryClient } from "@/data/query-client"; import { useSessionStore, type Agent } from "@/stores/session-store"; -import { normalizeAgentSnapshot } from "@/utils/agent-snapshots"; +import { normalizeAgentSnapshot, projectAgentSnapshot } from "@/utils/agent-snapshots"; import { applyAgentDirectoryDelta, type AgentDirectoryDelta, @@ -15,6 +15,12 @@ import { import { reconcileAgentDirectory } from "@/utils/agent-directory-reconciliation"; import { applyLegacyDaemonWorkspaceOwnership } from "@/workspace/legacy-daemon-workspaces"; +function projectAgentDirectoryEntry(agent: Agent): FetchAgentsEntry | null { + return agent.projectPlacement + ? { agent: projectAgentSnapshot(agent), project: agent.projectPlacement } + : null; +} + export interface AgentLifecycleToken { readonly agentId: string; readonly version: number; @@ -97,6 +103,25 @@ export class AgentDirectoryReplica { return agents; } + commitChanges( + entries: FetchAgentsEntry[], + removals: readonly { id: string }[], + deltas: readonly AgentDirectoryDelta[], + ): Map { + const previous = useSessionStore.getState().sessions[this.serverId]?.agents ?? new Map(); + const merged = new Map(); + for (const agent of previous.values()) { + const entry = projectAgentDirectoryEntry(agent); + if (entry) merged.set(agent.id, entry); + } + for (const entry of entries) merged.set(entry.agent.id, entry); + const removalsAsDeltas: AgentDirectoryDelta[] = removals.map(({ id }) => ({ + kind: "remove", + agentId: id, + })); + return this.commitSnapshot(Array.from(merged.values()), [...removalsAsDeltas, ...deltas]); + } + archive(agentId: string, archivedAt: string): void { this.advance(agentId); useSessionStore.getState().setAgents(this.serverId, (current) => { diff --git a/packages/app/src/runtime/directory-sync/index.test.ts b/packages/app/src/runtime/directory-sync/index.test.ts index fc9368fae..9725170d9 100644 --- a/packages/app/src/runtime/directory-sync/index.test.ts +++ b/packages/app/src/runtime/directory-sync/index.test.ts @@ -1,7 +1,7 @@ import { afterEach, describe, expect, it } from "vitest"; import type { DaemonClient } from "@getpaseo/client/internal/daemon-client"; import type { SessionOutboundMessage } from "@getpaseo/protocol/messages"; -import { useSessionStore } from "@/stores/session-store"; +import { normalizeProjectDescriptor, useSessionStore } from "@/stores/session-store"; import { DirectoryRefreshSupersededError, DirectorySync } from "./index"; type WorkspaceFetchResult = Awaited>; @@ -9,8 +9,11 @@ type ProjectListResult = Awaited>; class FakeDirectoryClient { fetchAgentsCalls = 0; + lastAgentOptions: unknown; fetchWorkspacesCalls = 0; listProjectsCalls = 0; + lastProjectOptions: unknown; + projectResult: ProjectListResult | null = null; private pendingWorkspaceFetch: Promise | null = null; private readonly handlers = new Map< SessionOutboundMessage["type"], @@ -42,8 +45,9 @@ class FakeDirectoryClient { return complete; } - async fetchAgents(): Promise>> { + async fetchAgents(options?: unknown): Promise>> { this.fetchAgentsCalls += 1; + this.lastAgentOptions = options; return { requestId: "agents", entries: [], @@ -66,8 +70,10 @@ class FakeDirectoryClient { }; } - async listProjects(): Promise { + async listProjects(options?: unknown): Promise { this.listProjectsCalls += 1; + this.lastProjectOptions = options; + if (this.projectResult) return this.projectResult; return { requestId: "projects", projects: [ @@ -111,6 +117,52 @@ afterEach(() => { }); describe("DirectorySync session readiness", () => { + it("uses the saved agent sequence while bounding the bootstrap page", async () => { + const serverId = "agent-list-sequence-page"; + serverIds.add(serverId); + const client = new FakeDirectoryClient(); + const directory = new DirectorySync( + serverId, + { + onAgentStoppedRunning: () => undefined, + markAgentLoading: () => undefined, + markAgentReady: () => undefined, + markAgentError: () => undefined, + }, + { + readDirectoryCheckpoint: () => ({ agents: { generation: "generation", afterSeq: 12 } }), + writeDirectoryCheckpoint: () => undefined, + }, + ); + directory.connectionChanged({ + client: client as unknown as DaemonClient, + status: "online", + source: { clientGeneration: 1, connectionEpoch: 1 }, + }); + const store = useSessionStore.getState(); + store.initializeSession(serverId, client as unknown as DaemonClient, 1); + store.updateSessionServerInfo(serverId, { + serverId, + hostname: null, + version: "test", + features: { directorySync: true, workspaceMultiplicity: true }, + }); + + await directory.refreshAgents({ + subscribe: { subscriptionId: `app:${serverId}` }, + page: { limit: 200 }, + }); + + expect(client.lastAgentOptions).toEqual({ + scope: "active", + sort: [{ key: "updated_at", direction: "desc" }], + subscribe: { subscriptionId: `app:${serverId}` }, + page: { limit: 200 }, + sync: { generation: "generation", afterSeq: 12 }, + }); + directory.dispose(); + }); + it("waits for workspace capability metadata before choosing the workspace protocol", async () => { const serverId = "workspace-metadata"; const { client, directory } = createDirectory(serverId); @@ -159,6 +211,82 @@ describe("DirectorySync session readiness", () => { directory.dispose(); }); + it("merges project changes from the existing list RPC and advances its cursor", async () => { + const serverId = "project-list-sequence"; + serverIds.add(serverId); + const client = new FakeDirectoryClient(); + const writes: unknown[] = []; + const directory = new DirectorySync( + serverId, + { + onAgentStoppedRunning: () => undefined, + markAgentLoading: () => undefined, + markAgentReady: () => undefined, + markAgentError: () => undefined, + }, + { + readDirectoryCheckpoint: () => ({ projects: { generation: "generation", afterSeq: 4 } }), + writeDirectoryCheckpoint: (_serverId, checkpoint) => writes.push(checkpoint), + }, + ); + directory.connectionChanged({ + client: client as unknown as DaemonClient, + status: "online", + source: { clientGeneration: 1, connectionEpoch: 1 }, + }); + const store = useSessionStore.getState(); + store.initializeSession(serverId, client as unknown as DaemonClient, 1); + store.setProjects(serverId, [ + normalizeProjectDescriptor({ + projectId: "project-1", + projectDisplayName: "Old name", + projectRootPath: "/repo/one", + projectKind: "git", + }), + normalizeProjectDescriptor({ + projectId: "project-2", + projectDisplayName: "Removed", + projectRootPath: "/repo/two", + projectKind: "git", + }), + ]); + store.updateSessionServerInfo(serverId, { + serverId, + hostname: null, + version: "test", + features: { workspaceMultiplicity: true, projectList: true, directorySync: true }, + }); + client.projectResult = { + requestId: "projects", + projects: [ + { + projectId: "project-1", + projectDisplayName: "New name", + projectRootPath: "/repo/one", + projectKind: "git", + syncSeq: 5, + }, + ], + sync: { + generation: "generation", + headSeq: 6, + mode: "changes", + removals: [{ id: "project-2", seq: 6 }], + }, + }; + + await directory.refreshWorkspaces(); + + expect(client.lastProjectOptions).toEqual({ + sync: { generation: "generation", afterSeq: 4 }, + }); + const projects = useSessionStore.getState().sessions[serverId]?.projects; + expect(Array.from(projects?.keys() ?? [])).toEqual(["project-1"]); + expect(projects?.get("project-1")?.projectDisplayName).toBe("New name"); + expect(writes).toContainEqual({ projects: { generation: "generation", afterSeq: 6 } }); + directory.dispose(); + }); + it("rejects a session wait on disconnect so the reconnect can refresh", async () => { const serverId = "session-wait-reconnect"; const { client, directory } = createDirectory(serverId); diff --git a/packages/app/src/runtime/directory-sync/index.ts b/packages/app/src/runtime/directory-sync/index.ts index bfb2f93a6..ac80a81bd 100644 --- a/packages/app/src/runtime/directory-sync/index.ts +++ b/packages/app/src/runtime/directory-sync/index.ts @@ -36,10 +36,41 @@ const AGENT_SORT: NonNullable = [ { key: "updated_at", direction: "desc" }, ]; +function resolveAgentNextPage(pageInfo: AgentPageInfo): { + hasMore: boolean; + nextCursor: string | null; +} { + return { + hasMore: pageInfo.hasMore ?? pageInfo.hasMoreAfter ?? false, + nextCursor: pageInfo.nextCursor ?? pageInfo.afterCursor ?? null, + }; +} + interface AgentSnapshot { entries: FetchAgentsEntry[]; subscriptionId: string | null; legacy: boolean; + syncMode?: "snapshot" | "changes"; + syncCursor?: DirectoryCursor; + syncRemovals: Array<{ id: string; seq: number }>; +} + +interface AgentPageInfo { + hasMore?: boolean; + hasMoreAfter?: boolean; + nextCursor?: string | null; + afterCursor?: string | null; +} + +interface DirectoryCursor { + generation: string; + afterSeq: number; +} + +interface DirectoryCursors { + projects?: DirectoryCursor; + workspaces?: DirectoryCursor; + agents?: DirectoryCursor; } export interface DirectoryConnection { @@ -48,6 +79,11 @@ export interface DirectoryConnection { source: DirectorySourceToken; } +export interface DirectoryCheckpointStorage { + readDirectoryCheckpoint(serverId: string): unknown; + writeDirectoryCheckpoint(serverId: string, checkpoint: unknown): void; +} + export interface RefreshAgentDirectoryInput { filter?: FetchAgentsOptions["filter"]; subscribe?: FetchAgentsOptions["subscribe"]; @@ -86,6 +122,7 @@ export class DirectorySync { markAgentReady: () => void; markAgentError: (error: string) => void; }, + private readonly checkpoints?: DirectoryCheckpointStorage, ) { this.agents = new AgentDirectoryReplica(serverId, callbacks.onAgentStoppedRunning); this.workspaces = new WorkspaceDirectoryReplica(serverId); @@ -116,19 +153,26 @@ export class DirectorySync { const subscriptions = [ client.on("agent_update", (message) => { if (message.type !== "agent_update" || !this.isCurrent(client, source)) return; - if (!this.agentTransactions.record(source, message.payload)) + const recorded = this.agentTransactions.record(source, message.payload); + if (!recorded) { this.agents.applyDelta(message.payload); + this.noteLiveCursor("agents", message.payload); + } }), client.on("workspace_update", (message) => { if (message.type !== "workspace_update" || !this.isCurrent(client, source)) return; - if (!this.workspaceTransactions.record(source, message.payload)) { + const recorded = this.workspaceTransactions.record(source, message.payload); + if (!recorded) { this.workspaces.applyDelta(message.payload); + this.noteLiveCursor("workspaces", message.payload); } }), client.on("project.update", (message) => { if (message.type !== "project.update" || !this.isCurrent(client, source)) return; - if (!this.workspaceTransactions.record(source, message.payload)) { + const recorded = this.workspaceTransactions.record(source, message.payload); + if (!recorded) { this.workspaces.applyDelta(message.payload); + this.noteLiveCursor("projects", message.payload); } }), client.on("agent_deleted", (message) => { @@ -174,6 +218,7 @@ export class DirectorySync { entries: [], subscriptionId: null, legacy: false, + syncRemovals: [], })); this.callbacks.markAgentLoading(); try { @@ -222,7 +267,8 @@ export class DirectorySync { : delta, ) : completion.deltas; - const agents = this.agents.commitSnapshot(completion.snapshot.entries, deltas); + const agents = this.commitAgentSnapshot(completion.snapshot, deltas); + this.persistAgentCursors(completion.snapshot, completion.deltas); this.callbacks.markAgentReady(); return { agents, subscriptionId: completion.snapshot.subscriptionId }; } catch (error) { @@ -238,8 +284,9 @@ export class DirectorySync { async refreshWorkspaces(input?: { subscribe?: boolean }): Promise { const { client, source } = this.requireOnline(); const transaction = this.workspaceTransactions.begin(source, () => ({ - workspaces: new Map(), - projects: new Map(), + workspaces: new Map(useSessionStore.getState().sessions[this.serverId]?.workspaces), + projects: new Map(useSessionStore.getState().sessions[this.serverId]?.projects), + syncCursors: {}, })); try { await this.waitForSessionMetadata(client, source); @@ -250,24 +297,19 @@ export class DirectorySync { return; } const supportsProjectList = serverInfo.features?.projectList === true; + const supportsDirectorySync = serverInfo.features?.directorySync === true; if (supportsProjectList) { - const payload = await client.listProjects(); - this.assertWorkspaceTransactionCurrent(client, source, transaction); - for (const entry of payload.projects) { - const project = normalizeProjectDescriptor(entry); - transaction.snapshot.projects.set(project.projectId, project); - } + await this.fetchProjectSnapshot(client, source, transaction, supportsDirectorySync); } - await this.fetchWorkspaceSnapshot(client, source, transaction, input?.subscribe === true); + await this.fetchWorkspaceSnapshot( + client, + source, + transaction, + input?.subscribe === true, + supportsDirectorySync, + ); if (!supportsProjectList) this.buildLegacyProjectSnapshot(transaction.snapshot); - if (!this.isCurrent(client, source) || !this.hasMatchingSession(client, source)) { - throw new DirectoryRefreshSupersededError("workspace completion no longer current"); - } - const completion = this.workspaceTransactions.complete(transaction); - if (completion.kind === "stale") { - throw new DirectoryRefreshSupersededError("workspace completion was superseded"); - } - this.workspaces.commitSnapshot(completion.snapshot, completion.deltas); + this.completeWorkspaceRefresh(client, source, transaction); } catch (error) { const deltas = this.workspaceTransactions.fail(transaction); if (deltas) for (const delta of deltas) this.workspaces.applyDelta(delta); @@ -280,6 +322,7 @@ export class DirectorySync { source: DirectorySourceToken, transaction: DirectoryTransaction, initialSubscribe: boolean, + supportsDirectorySync: boolean, ): Promise { let cursor: string | null = null; let subscribe = initialSubscribe; @@ -288,12 +331,23 @@ export class DirectorySync { sort: [{ key: "activity_at", direction: "desc" }], ...(subscribe ? { subscribe: {} } : {}), page: cursor ? { limit: PAGE_LIMIT, cursor } : { limit: PAGE_LIMIT }, + ...(supportsDirectorySync ? { sync: this.readCursors().workspaces ?? {} } : {}), }); this.assertWorkspaceTransactionCurrent(client, source, transaction); + if (payload.sync?.mode === "snapshot") transaction.snapshot.workspaces.clear(); for (const entry of payload.entries) { const workspace = normalizeWorkspaceDescriptor(entry); transaction.snapshot.workspaces.set(workspace.id, workspace); } + for (const removal of payload.sync?.removals ?? []) { + transaction.snapshot.workspaces.delete(removal.id); + } + if (payload.sync) { + (transaction.snapshot.syncCursors ??= {}).workspaces = { + generation: payload.sync.generation, + afterSeq: payload.sync.headSeq, + }; + } for (const entry of payload.emptyProjects ?? []) { const project = normalizeProjectDescriptor(entry); transaction.snapshot.projects.set(project.projectId, project); @@ -332,18 +386,15 @@ export class DirectorySync { sort: AGENT_SORT, ...(subscribe ? { subscribe } : {}), page: cursor ? { limit, cursor } : { limit }, + ...(!input.filter && cursor === null && this.supportsDirectorySync() + ? { sync: this.readCursors().agents ?? {} } + : {}), }); this.assertAgentTransactionCurrent(client, source, transaction); transaction.snapshot.entries.push(...payload.entries); transaction.snapshot.subscriptionId ??= payload.subscriptionId ?? null; - const pageInfo = payload.pageInfo as { - hasMore?: boolean; - hasMoreAfter?: boolean; - nextCursor?: string | null; - afterCursor?: string | null; - }; - const hasMore = pageInfo.hasMore ?? pageInfo.hasMoreAfter ?? false; - const nextCursor = pageInfo.nextCursor ?? pageInfo.afterCursor ?? null; + this.recordAgentSync(transaction.snapshot, payload.sync); + const { hasMore, nextCursor } = resolveAgentNextPage(payload.pageInfo as AgentPageInfo); if (!hasMore || !nextCursor) break; cursor = nextCursor; subscribe = undefined; @@ -377,6 +428,122 @@ export class DirectorySync { return { client: this.connection.client, source: this.connection.source }; } + private supportsDirectorySync(): boolean { + return ( + useSessionStore.getState().sessions[this.serverId]?.serverInfo?.features?.directorySync === + true + ); + } + + private commitAgentSnapshot( + snapshot: AgentSnapshot, + deltas: readonly AgentDirectoryDelta[], + ): Map { + return snapshot.syncMode === "changes" + ? this.agents.commitChanges(snapshot.entries, snapshot.syncRemovals, deltas) + : this.agents.commitSnapshot(snapshot.entries, deltas); + } + + private persistAgentCursors( + snapshot: AgentSnapshot, + deltas: readonly AgentDirectoryDelta[], + ): void { + if (snapshot.syncCursor) this.writeCursor("agents", snapshot.syncCursor); + for (const delta of deltas) this.noteLiveCursor("agents", delta); + } + + private async fetchProjectSnapshot( + client: DaemonClient, + source: DirectorySourceToken, + transaction: DirectoryTransaction, + supportsDirectorySync: boolean, + ): Promise { + const payload = await client.listProjects( + supportsDirectorySync ? { sync: this.readCursors().projects ?? {} } : undefined, + ); + this.assertWorkspaceTransactionCurrent(client, source, transaction); + if (payload.sync?.mode === "snapshot") transaction.snapshot.projects.clear(); + for (const entry of payload.projects) { + const project = normalizeProjectDescriptor(entry); + transaction.snapshot.projects.set(project.projectId, project); + } + for (const removal of payload.sync?.removals ?? []) { + transaction.snapshot.projects.delete(removal.id); + } + if (payload.sync) { + (transaction.snapshot.syncCursors ??= {}).projects = { + generation: payload.sync.generation, + afterSeq: payload.sync.headSeq, + }; + } + } + + private completeWorkspaceRefresh( + client: DaemonClient, + source: DirectorySourceToken, + transaction: DirectoryTransaction, + ): void { + if (!this.isCurrent(client, source) || !this.hasMatchingSession(client, source)) { + throw new DirectoryRefreshSupersededError("workspace completion no longer current"); + } + const completion = this.workspaceTransactions.complete(transaction); + if (completion.kind === "stale") { + throw new DirectoryRefreshSupersededError("workspace completion was superseded"); + } + this.workspaces.commitSnapshot(completion.snapshot, completion.deltas); + for (const [entity, cursor] of Object.entries(completion.snapshot.syncCursors ?? {})) { + if (cursor) this.writeCursor(entity as "projects" | "workspaces", cursor); + } + for (const delta of completion.deltas) { + const entity = "projectId" in delta || "project" in delta ? "projects" : "workspaces"; + this.noteLiveCursor(entity, delta); + } + } + + private recordAgentSync( + snapshot: AgentSnapshot, + sync: Awaited>["sync"], + ): void { + if (!sync) return; + snapshot.syncMode = sync.mode; + snapshot.syncRemovals.push(...sync.removals); + snapshot.syncCursor = { generation: sync.generation, afterSeq: sync.headSeq }; + } + + private noteLiveCursor( + entity: keyof DirectoryCursors, + payload: { generation?: string; seq?: number }, + ): void { + if (!payload.generation || payload.seq === undefined) return; + this.writeCursor(entity, { + generation: payload.generation, + afterSeq: payload.seq, + }); + } + + private readCursors(): DirectoryCursors { + const value = this.checkpoints?.readDirectoryCheckpoint(this.serverId); + if (!value || typeof value !== "object" || Array.isArray(value)) return {}; + const cursors: DirectoryCursors = {}; + for (const entity of ["projects", "workspaces", "agents"] as const) { + const cursor = (value as Record)[entity]; + if (!cursor || typeof cursor !== "object" || Array.isArray(cursor)) continue; + const { generation, afterSeq } = cursor as Record; + if (typeof generation === "string" && typeof afterSeq === "number" && afterSeq >= 0) { + cursors[entity] = { generation, afterSeq }; + } + } + return cursors; + } + + private writeCursor(entity: keyof DirectoryCursors, cursor: DirectoryCursor): void { + if (!this.checkpoints) return; + const current = this.readCursors(); + const previous = current[entity]; + if (previous?.generation === cursor.generation && previous.afterSeq >= cursor.afterSeq) return; + this.checkpoints.writeDirectoryCheckpoint(this.serverId, { ...current, [entity]: cursor }); + } + private isCurrent(client: DaemonClient, source: DirectorySourceToken): boolean { return ( this.connection.client === client && diff --git a/packages/app/src/runtime/directory-sync/workspace-replica.ts b/packages/app/src/runtime/directory-sync/workspace-replica.ts index 0d940724b..b64b5b14c 100644 --- a/packages/app/src/runtime/directory-sync/workspace-replica.ts +++ b/packages/app/src/runtime/directory-sync/workspace-replica.ts @@ -21,6 +21,9 @@ type ProjectDirectoryDelta = Extract; projects: Map; + syncCursors?: Partial< + Record<"projects" | "workspaces", { generation: string; afterSeq: number }> + >; } function applyProjectDelta( diff --git a/packages/app/src/runtime/host-features.ts b/packages/app/src/runtime/host-features.ts index e513c3364..37453fecb 100644 --- a/packages/app/src/runtime/host-features.ts +++ b/packages/app/src/runtime/host-features.ts @@ -51,3 +51,22 @@ export function useHostFeatureMap( [flags, serverIds], ); } + +export function useHostFeatureAvailabilityMap( + serverIds: readonly string[], + feature: HostFeatureName, +): ReadonlyMap { + const flags = useSessionStore( + useShallow((state) => + serverIds.map((serverId) => { + const serverInfo = state.sessions[serverId]?.serverInfo; + return serverInfo ? hostSupportsFeature(serverInfo, feature) : null; + }), + ), + ); + + return useMemo( + () => new Map(serverIds.map((serverId, index) => [serverId, flags[index]] as const)), + [flags, serverIds], + ); +} diff --git a/packages/app/src/runtime/host-runtime.ts b/packages/app/src/runtime/host-runtime.ts index eff24ee32..86ee71c73 100644 --- a/packages/app/src/runtime/host-runtime.ts +++ b/packages/app/src/runtime/host-runtime.ts @@ -60,6 +60,8 @@ import { resolveComposerAttachmentSubmitFormat } from "@/composer/attachments/su import { encodeImages } from "@/utils/encode-images"; import { DirectorySync, type RefreshAgentDirectoryResult } from "@/runtime/directory-sync"; import { ReplicaCache } from "@/runtime/replica-cache"; +import { replicaCacheStorage } from "@/runtime/replica-cache/storage"; +import { projectIconCache } from "@/projects/icon-cache"; import { nativePerformanceTrace } from "@/performance/native-trace"; import { revokePushNotifications } from "@/push-notifications"; import { createAppWebSocketFactory } from "./websocket-factory"; @@ -1374,7 +1376,7 @@ export class HostRuntimeStore { }) { this.deps = input?.deps ?? createDefaultDeps(); this.storage = input?.storage ?? AsyncStorage; - this.replicaCache = new ReplicaCache(this.storage); + this.replicaCache = new ReplicaCache(input?.storage ?? replicaCacheStorage); this.revokePushNotifications = input?.revokePushNotifications ?? revokePushNotifications; } @@ -1473,7 +1475,8 @@ export class HostRuntimeStore { } this.hosts = profiles; this.replicaCache.setHosts(profiles.map((profile) => profile.serverId)); - await this.replicaCache.restore(); + projectIconCache.setHosts(profiles.map((profile) => profile.serverId)); + await Promise.all([this.replicaCache.restore(), projectIconCache.restore()]); this.syncHosts(profiles); } catch (error) { console.error("[HostRuntime] Failed to load host registry from storage", error); @@ -1610,14 +1613,19 @@ export class HostRuntimeStore { rekeyMap(this.connectionStatusStartedAtByServer, oldServerId, newServerId); rekeyMap(this.directoryBootstrapInFlight, oldServerId, newServerId); this.replicaCache.reconcileServerId(oldServerId, newServerId); + projectIconCache.reconcileServerId(oldServerId, newServerId); this.directorySyncByServer.get(oldServerId)?.dispose(); this.directorySyncByServer.delete(oldServerId); - const directory = new DirectorySync(newServerId, { - onAgentStoppedRunning: (agentId) => this.drainQueuedAgentMessage(newServerId, agentId), - markAgentLoading: () => controller.markAgentDirectorySyncLoading(), - markAgentReady: () => controller.markAgentDirectorySyncReady(), - markAgentError: (error) => controller.markAgentDirectorySyncError(error), - }); + const directory = new DirectorySync( + newServerId, + { + onAgentStoppedRunning: (agentId) => this.drainQueuedAgentMessage(newServerId, agentId), + markAgentLoading: () => controller.markAgentDirectorySyncLoading(), + markAgentReady: () => controller.markAgentDirectorySyncReady(), + markAgentError: (error) => controller.markAgentDirectorySyncError(error), + }, + this.replicaCache, + ); this.directorySyncByServer.set(newServerId, directory); controller.adoptReconciledServerId(newServerId); const snapshot = controller.getSnapshot(); @@ -1970,6 +1978,7 @@ export class HostRuntimeStore { }, ): void { this.replicaCache.setHosts(hosts.map((host) => host.serverId)); + projectIconCache.setHosts(hosts.map((host) => host.serverId)); const nextIds = new Set(hosts.map((host) => host.serverId)); for (const [serverId, controller] of this.controllers) { if (nextIds.has(serverId)) { @@ -2006,12 +2015,17 @@ export class HostRuntimeStore { this.controllers.set(host.serverId, controller); this.directorySyncByServer.set( host.serverId, - new DirectorySync(host.serverId, { - onAgentStoppedRunning: (agentId) => this.drainQueuedAgentMessage(host.serverId, agentId), - markAgentLoading: () => controller.markAgentDirectorySyncLoading(), - markAgentReady: () => controller.markAgentDirectorySyncReady(), - markAgentError: (error) => controller.markAgentDirectorySyncError(error), - }), + new DirectorySync( + host.serverId, + { + onAgentStoppedRunning: (agentId) => + this.drainQueuedAgentMessage(host.serverId, agentId), + markAgentLoading: () => controller.markAgentDirectorySyncLoading(), + markAgentReady: () => controller.markAgentDirectorySyncReady(), + markAgentError: (error) => controller.markAgentDirectorySyncError(error), + }, + this.replicaCache, + ), ); const initialSnapshot = controller.getSnapshot(); this.lastConnectionStatusByServer.set(host.serverId, initialSnapshot.connectionStatus); @@ -2294,22 +2308,6 @@ export class HostRuntimeStore { } } - markAgentDirectorySyncLoading(serverId: string): void { - this.controllers.get(serverId)?.markAgentDirectorySyncLoading(); - } - - markAgentDirectorySyncReady(serverId: string): void { - this.controllers.get(serverId)?.markAgentDirectorySyncReady(); - } - - markAgentDirectorySyncError(serverId: string, error: string): void { - this.controllers.get(serverId)?.markAgentDirectorySyncError(error); - } - - markAgentDirectorySyncIdle(serverId: string): void { - this.controllers.get(serverId)?.markAgentDirectorySyncIdle(); - } - private emit(serverId: string): void { this.version += 1; const listeners = this.serverListeners.get(serverId); diff --git a/packages/app/src/runtime/replica-cache/index.test.ts b/packages/app/src/runtime/replica-cache/index.test.ts index 6066aed51..ee008f2ab 100644 --- a/packages/app/src/runtime/replica-cache/index.test.ts +++ b/packages/app/src/runtime/replica-cache/index.test.ts @@ -55,34 +55,50 @@ function workspace( } function agent(id: string, workspaceId = "workspace-1", cwd = "/repo/paseo") { - return normalizeAgentSnapshot( - { - id, - provider: "codex", - cwd, - workspaceId, - model: null, - createdAt: "2026-07-18T08:00:00.000Z", - updatedAt: "2026-07-18T08:01:00.000Z", - lastUserMessageAt: "2026-07-18T08:01:00.000Z", - status: "idle", - capabilities: { - supportsStreaming: true, - supportsSessionPersistence: true, - supportsDynamicModes: true, - supportsMcpServers: true, - supportsReasoningStream: true, - supportsToolInvocations: true, + return { + ...normalizeAgentSnapshot( + { + id, + provider: "codex", + cwd, + workspaceId, + model: null, + createdAt: "2026-07-18T08:00:00.000Z", + updatedAt: "2026-07-18T08:01:00.000Z", + lastUserMessageAt: "2026-07-18T08:01:00.000Z", + status: "idle", + capabilities: { + supportsStreaming: true, + supportsSessionPersistence: true, + supportsDynamicModes: true, + supportsMcpServers: true, + supportsReasoningStream: true, + supportsToolInvocations: true, + }, + currentModeId: null, + availableModes: [], + pendingPermissions: [], + persistence: null, + title: `Agent ${id}`, + labels: {}, + }, + SERVER_ID, + ), + projectPlacement: { + projectKey: cwd, + projectName: cwd.split("/").at(-1) ?? cwd, + workspaceName: workspaceId, + checkout: { + cwd, + isGit: false as const, + currentBranch: null, + remoteUrl: null, + worktreeRoot: null, + isPaseoOwnedWorktree: false as const, + mainRepoRoot: null, }, - currentModeId: null, - availableModes: [], - pendingPermissions: [], - persistence: null, - title: `Agent ${id}`, - labels: {}, }, - SERVER_ID, - ); + }; } function message(id: string, text: string): StreamItem { @@ -212,10 +228,12 @@ describe("ReplicaCache", () => { expect(session.client).toBeNull(); expect(session.hasHydratedAgents).toBe(false); expect(session.hasHydratedWorkspaces).toBe(false); + expect(session.hasWorkspaceDirectorySnapshot).toBe(true); expect(Array.from(session.agents.keys())).toEqual(["agent-1"]); expect(Array.from(session.workspaces.keys())).toEqual(["workspace-1"]); - expect(Array.from(session.projects.keys())).toEqual(["project-1"]); + expect(Array.from(session.projects.keys())).toEqual(["project-1", "empty-project"]); expect(session.agents.get("agent-1")?.updatedAt).toBeInstanceOf(Date); + expect(session.agents.get("agent-1")?.projectPlacement?.checkout.cwd).toBe("/repo/paseo"); expect(session.workspaces.get("workspace-1")?.statusEnteredAt).toBeInstanceOf(Date); expect(session.workspaces.get("workspace-1")?.worktreeSlug).toBe("owned-worktree"); expect(session.agentStreamTail.get("agent-1")).toEqual([message("message-1", "Cached")]); @@ -230,7 +248,7 @@ describe("ReplicaCache", () => { }); }); - it("persists only the focused agent view with a short timeline tail", async () => { + it("persists the complete directory with only the focused timeline tail", async () => { const storage = new MemoryStorage(); const cache = new ReplicaCache(storage); cache.setHosts([SERVER_ID]); @@ -266,9 +284,13 @@ describe("ReplicaCache", () => { const session = useSessionStore.getState().sessions[SERVER_ID]; const timelines = session?.agentStreamTail; - expect(Array.from(session?.agents.keys() ?? [])).toEqual(["agent-2"]); - expect(Array.from(session?.workspaces.keys() ?? [])).toEqual(["workspace-2"]); - expect(Array.from(session?.projects.keys() ?? [])).toEqual(["project-2"]); + expect(Array.from(session?.agents.keys() ?? [])).toEqual(["agent-1", "agent-2"]); + expect(Array.from(session?.workspaces.keys() ?? [])).toEqual(["workspace-1", "workspace-2"]); + expect(Array.from(session?.projects.keys() ?? [])).toEqual([ + "project-1", + "project-2", + "empty-project", + ]); expect(Array.from(timelines?.keys() ?? [])).toEqual(["agent-2"]); expect(timelines?.get("agent-2")).toEqual(secondTimeline.slice(-50)); @@ -310,6 +332,53 @@ describe("ReplicaCache", () => { ]); }); + it("persists monotonic directory cursors with the complete host replica", async () => { + const storage = new MemoryStorage(); + const cache = new ReplicaCache(storage); + cache.setHosts([SERVER_ID]); + seedSession(); + await cache.flush(); + + cache.writeDirectoryCheckpoint(SERVER_ID, { + agents: { generation: "daemon-generation", afterSeq: 7 }, + }); + 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("agent-1", { ...current, title: "Updated agent" }); + }); + await cache.flush(); + + const reader = new ReplicaCache(storage); + reader.setHosts([SERVER_ID]); + await reader.restore(); + expect(reader.readDirectoryCheckpoint(SERVER_ID)).toEqual({ + agents: { generation: "daemon-generation", afterSeq: 7 }, + }); + }); + + it("restores every registered host directory before any host reconnects", async () => { + const storage = new MemoryStorage(); + const writer = new ReplicaCache(storage); + writer.setHosts(LRU_SERVER_IDS); + for (const serverId of LRU_SERVER_IDS) seedTimeline(serverId, `cached-${serverId}`); + await writer.flush(); + for (const serverId of LRU_SERVER_IDS) useSessionStore.getState().clearSession(serverId); + + const reader = new ReplicaCache(storage); + reader.setHosts(LRU_SERVER_IDS); + await reader.restore(); + + for (const serverId of LRU_SERVER_IDS) { + const session = useSessionStore.getState().sessions[serverId]; + expect(Array.from(session?.agents.keys() ?? [])).toEqual([`agent-${serverId}`]); + expect(Array.from(session?.workspaces.keys() ?? [])).toEqual([`workspace-${serverId}`]); + expect(session?.hasHydratedAgents).toBe(false); + expect(session?.hasHydratedWorkspaces).toBe(false); + expect(session?.hasWorkspaceDirectorySnapshot).toBe(true); + } + }); + it("evicts the least recently written host when the cache exceeds its byte budget", async () => { const storage = new MemoryStorage(); const cache = new ReplicaCache(storage, { maxBytes: 7_000 }); diff --git a/packages/app/src/runtime/replica-cache/index.ts b/packages/app/src/runtime/replica-cache/index.ts index 7459b526e..35ddf09fe 100644 --- a/packages/app/src/runtime/replica-cache/index.ts +++ b/packages/app/src/runtime/replica-cache/index.ts @@ -8,7 +8,6 @@ import { selectAgentTimelineState, useSessionStore, type Agent, - type AgentTimelineState, type SessionReplica, type SessionState, type ProjectDescriptor, @@ -21,7 +20,7 @@ const STORAGE_KEY = "@paseo:replica-cache"; const CACHE_VERSION = 5; const PERSIST_DELAY_MS = 750; const MAX_TIMELINE_ITEMS = 50; -const MAX_CACHE_BYTES = 1024 * 1024; +const MAX_CACHE_BYTES = 32 * 1024 * 1024; const IsoDateSchema = z.iso.datetime(); const TimelinePositionSchema = z.strictObject({ epoch: z.string(), @@ -107,6 +106,43 @@ const AgentCapabilitiesSchema = z.strictObject({ supportsRewindBoth: z.boolean().optional(), }); +const StoredProjectCheckoutSchema = z.union([ + z.strictObject({ + cwd: z.string(), + isGit: z.literal(false), + currentBranch: z.null(), + remoteUrl: z.null(), + worktreeRoot: z.null(), + isPaseoOwnedWorktree: z.literal(false), + mainRepoRoot: z.null(), + }), + z.strictObject({ + cwd: z.string(), + isGit: z.literal(true), + currentBranch: z.string().nullable(), + remoteUrl: z.string().nullable(), + worktreeRoot: z.string(), + isPaseoOwnedWorktree: z.literal(false), + mainRepoRoot: z.string().nullable(), + }), + z.strictObject({ + cwd: z.string(), + isGit: z.literal(true), + currentBranch: z.string().nullable(), + remoteUrl: z.string().nullable(), + worktreeRoot: z.string(), + isPaseoOwnedWorktree: z.literal(true), + mainRepoRoot: z.string(), + }), +]); + +const StoredProjectPlacementSchema = z.strictObject({ + projectKey: z.string(), + projectName: z.string(), + workspaceName: z.string().nullable().optional(), + checkout: StoredProjectCheckoutSchema, +}); + const StoredAgentSnapshotSchema = z.strictObject({ id: z.string(), provider: AgentProviderSchema, @@ -141,7 +177,7 @@ const StoredAgentSnapshotSchema = z.strictObject({ const StoredAgentSchema = z.strictObject({ snapshot: StoredAgentSnapshotSchema, - projectPlacement: z.null(), + projectPlacement: StoredProjectPlacementSchema.nullable(), lastActivityAt: IsoDateSchema, }); @@ -201,6 +237,8 @@ const StoredProjectSchema = z.strictObject({ projectKey: z.string().optional(), projectDisplayName: z.string(), projectCustomName: z.string().nullable(), + projectCustomIconRevision: z.string().nullable(), + projectIconRevision: z.string().optional(), projectRootPath: z.string(), projectKind: z.enum(["git", "non_git", "directory"]), }); @@ -217,6 +255,7 @@ const StoredHostSchema = z.strictObject({ projects: z.array(StoredProjectSchema), emptyProjects: z.array(StoredProjectSchema), timeline: StoredTimelineSchema.nullable(), + directorySync: z.unknown().optional(), }); const StoredCacheSchema = z.strictObject({ @@ -231,9 +270,10 @@ type StoredWorkspace = z.infer; type StoredProject = z.infer; interface ReplicaInput { - agent: Agent | undefined; - workspace: WorkspaceDescriptor | undefined; - project: ProjectDescriptor | undefined; + agents: ReadonlyMap; + workspaces: ReadonlyMap; + projects: ReadonlyMap; + focusedAgent: Agent | undefined; timelineItems: StreamItem[] | undefined; } @@ -367,6 +407,10 @@ function deserializeTimelineItem(item: StoredTimelineItem): StreamItem { } } +function serializeProjectPlacement(agent: Agent): StoredAgent["projectPlacement"] { + return agent.projectPlacement ?? null; +} + function serializeAgent(agent: Agent): StoredAgent { const snapshot = { id: agent.id, @@ -421,7 +465,7 @@ function serializeAgent(agent: Agent): StoredAgent { }; return { snapshot, - projectPlacement: null, + projectPlacement: serializeProjectPlacement(agent), lastActivityAt: agent.lastActivityAt.toISOString(), }; } @@ -478,6 +522,8 @@ function serializeProject(project: ProjectDescriptor): StoredProject { ...(project.projectKey ? { projectKey: project.projectKey } : {}), projectDisplayName: project.projectDisplayName, projectCustomName: project.projectCustomName, + projectCustomIconRevision: project.projectCustomIconRevision ?? null, + projectIconRevision: project.projectIconRevision, projectRootPath: project.projectRootPath, projectKind: project.projectKind, }; @@ -485,50 +531,47 @@ function serializeProject(project: ProjectDescriptor): StoredProject { function replicaInputsEqual(left: ReplicaInput, right: ReplicaInput): boolean { return ( - left.agent === right.agent && - left.workspace === right.workspace && - left.project === right.project && + left.agents === right.agents && + left.workspaces === right.workspaces && + left.projects === right.projects && + left.focusedAgent === right.focusedAgent && 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: AgentTimelineState = agentId + const timeline = agentId ? selectAgentTimelineState(session, agentId) - : { status: "cold" }; + : { status: "cold" as const }; return { - agent, - workspace, - project: workspace ? session.projects.get(workspace.projectId) : undefined, + agents: session.agents, + workspaces: session.workspaces, + projects: session.projects, + focusedAgent: agent, timelineItems: timeline.status === "cold" ? undefined : timeline.items, }; } -function serializeHost(serverId: string, input: ReplicaInput): StoredHost { +function serializeHost(serverId: string, input: ReplicaInput, directorySync?: unknown): StoredHost { const items = input.timelineItems ?.filter((item) => item.kind !== "user_message" || !isUnreconciledLocalUserMessage(item)) .map(serializeTimelineItem) .filter((item) => item !== null); return { serverId, - agents: input.agent ? [serializeAgent(input.agent)] : [], - workspaces: input.workspace ? [serializeWorkspace(input.workspace)] : [], - projects: input.project ? [serializeProject(input.project)] : [], + agents: Array.from(input.agents.values(), serializeAgent), + workspaces: Array.from(input.workspaces.values(), serializeWorkspace), + projects: Array.from(input.projects.values(), serializeProject), emptyProjects: [], timeline: - input.agent && items + input.focusedAgent && items ? { - agentId: input.agent.id, + agentId: input.focusedAgent.id, items: items.slice(-MAX_TIMELINE_ITEMS), } : null, + ...(directorySync ? { directorySync } : {}), }; } @@ -688,6 +731,24 @@ export class ReplicaCache { this.schedulePersist(); } + readDirectoryCheckpoint(serverId: string): unknown { + return this.storedHosts.get(serverId)?.directorySync; + } + + writeDirectoryCheckpoint(serverId: string, checkpoint: unknown): void { + let stored = this.storedHosts.get(serverId); + if (!stored) { + const session = useSessionStore.getState().sessions[serverId]; + if (!session) return; + const input = selectReplicaInput(session, this.lastFocusedAgentIds.get(serverId) ?? null); + this.capturedInputs.set(serverId, input); + stored = serializeHost(serverId, input); + } + this.storedHosts.delete(serverId); + this.storedHosts.set(serverId, { ...stored, directorySync: checkpoint }); + this.schedulePersist(); + } + async flush(): Promise { if (this.persistTimer) { clearTimeout(this.persistTimer); @@ -727,8 +788,9 @@ export class ReplicaCache { if (previous && replicaInputsEqual(previous, input)) return false; this.capturedInputs.set(serverId, input); + const directorySync = this.storedHosts.get(serverId)?.directorySync; this.storedHosts.delete(serverId); - this.storedHosts.set(serverId, serializeHost(serverId, input)); + this.storedHosts.set(serverId, serializeHost(serverId, input, directorySync)); return true; } diff --git a/packages/app/src/runtime/replica-cache/storage.ts b/packages/app/src/runtime/replica-cache/storage.ts new file mode 100644 index 000000000..e4bc220e2 --- /dev/null +++ b/packages/app/src/runtime/replica-cache/storage.ts @@ -0,0 +1,4 @@ +import AsyncStorage from "@react-native-async-storage/async-storage"; +import type { ReplicaCacheStorage } from "./index"; + +export const replicaCacheStorage: ReplicaCacheStorage = AsyncStorage; diff --git a/packages/app/src/runtime/replica-cache/storage.web.ts b/packages/app/src/runtime/replica-cache/storage.web.ts new file mode 100644 index 000000000..ef2ef6896 --- /dev/null +++ b/packages/app/src/runtime/replica-cache/storage.web.ts @@ -0,0 +1,70 @@ +import type { ReplicaCacheStorage } from "./index"; + +const DATABASE_NAME = "paseo-replica-cache"; +const STORE_NAME = "key-value"; + +let databasePromise: Promise | null = null; + +function openDatabase(): Promise { + databasePromise ??= new Promise((resolve, reject) => { + const request = indexedDB.open(DATABASE_NAME, 1); + request.addEventListener("upgradeneeded", () => { + if (!request.result.objectStoreNames.contains(STORE_NAME)) { + request.result.createObjectStore(STORE_NAME); + } + }); + request.addEventListener("success", () => resolve(request.result)); + request.addEventListener("error", () => + reject(request.error ?? new Error("Failed to open replica cache")), + ); + request.addEventListener("blocked", () => + reject(new Error("Replica cache database upgrade was blocked")), + ); + }); + return databasePromise; +} + +async function getItem(key: string): Promise { + const database = await openDatabase(); + return new Promise((resolve, reject) => { + const request = database.transaction(STORE_NAME, "readonly").objectStore(STORE_NAME).get(key); + request.addEventListener("success", () => + resolve(typeof request.result === "string" ? request.result : null), + ); + request.addEventListener("error", () => + reject(request.error ?? new Error("Failed to read replica cache")), + ); + }); +} + +async function setItem(key: string, value: string): Promise { + const database = await openDatabase(); + return new Promise((resolve, reject) => { + const transaction = database.transaction(STORE_NAME, "readwrite"); + transaction.objectStore(STORE_NAME).put(value, key); + transaction.addEventListener("complete", () => resolve()); + transaction.addEventListener("error", () => + reject(transaction.error ?? new Error("Failed to write replica cache")), + ); + transaction.addEventListener("abort", () => + reject(transaction.error ?? new Error("Replica cache write was aborted")), + ); + }); +} + +async function removeItem(key: string): Promise { + const database = await openDatabase(); + return new Promise((resolve, reject) => { + const transaction = database.transaction(STORE_NAME, "readwrite"); + transaction.objectStore(STORE_NAME).delete(key); + transaction.addEventListener("complete", () => resolve()); + transaction.addEventListener("error", () => + reject(transaction.error ?? new Error("Failed to remove replica cache")), + ); + transaction.addEventListener("abort", () => + reject(transaction.error ?? new Error("Replica cache removal was aborted")), + ); + }); +} + +export const replicaCacheStorage: ReplicaCacheStorage = { getItem, setItem, removeItem }; diff --git a/packages/app/src/stores/session-store-hooks/index.ts b/packages/app/src/stores/session-store-hooks/index.ts index f2021a490..7d01fc9a9 100644 --- a/packages/app/src/stores/session-store-hooks/index.ts +++ b/packages/app/src/stores/session-store-hooks/index.ts @@ -5,6 +5,7 @@ import { composeWorkspaceStructure, selectHasHydratedWorkspaces, selectHydratedWorkspaceServerIds, + selectWorkspaceDirectoryServerIds, selectHasWorkspaces, selectProjectOrder, selectRecommendedProjectPaths, @@ -79,6 +80,14 @@ export function useHydratedWorkspaceServerIds(serverIds: string[]): string[] { ); } +export function useWorkspaceDirectoryServerIds(serverIds: string[]): string[] { + return useStoreWithEqualityFn( + useSessionStore, + (state) => selectWorkspaceDirectoryServerIds(state, serverIds), + workspaceEqualityFns.deep, + ); +} + export function useWorkspaceDirectory( serverId: string | null, workspaceId: string | null, diff --git a/packages/app/src/stores/session-store-hooks/selectors.test.ts b/packages/app/src/stores/session-store-hooks/selectors.test.ts index 96f21de53..2c0ec9b46 100644 --- a/packages/app/src/stores/session-store-hooks/selectors.test.ts +++ b/packages/app/src/stores/session-store-hooks/selectors.test.ts @@ -6,6 +6,7 @@ import { composeWorkspaceStructure, selectHasWorkspaces, selectHydratedWorkspaceServerIds, + selectWorkspaceDirectoryServerIds, selectProjectOrder, selectRecommendedProjectPaths, selectWorkspace, @@ -131,20 +132,32 @@ afterEach(() => { }); describe("workspace replica authority", () => { - it("keeps a cached workspace addressable without publishing it as an authoritative directory", () => { + it("publishes a restored directory while keeping remote hydration false", () => { const cachedWorkspace = createWorkspace({ id: "cached-workspace" }); - initializeWorkspaces([cachedWorkspace]); + useSessionStore.getState().restoreSessionReplica(SERVER_ID, { + agents: new Map(), + workspaces: new Map([[cachedWorkspace.id, cachedWorkspace]]), + projects: new Map([ + [cachedWorkspace.projectId, projectDescriptorFromTestWorkspace(cachedWorkspace)], + ]), + timeline: null, + }); + const cachedSession = useSessionStore.getState().sessions[SERVER_ID]; + if (!cachedSession) throw new Error("expected initialized session"); - const cachedServerIds = selectHydratedWorkspaceServerIds(useSessionStore.getState(), [ + const cachedServerIds = selectWorkspaceDirectoryServerIds(useSessionStore.getState(), [ SERVER_ID, ]); expect(selectWorkspace(useSessionStore.getState(), SERVER_ID, cachedWorkspace.id)).toBe( cachedWorkspace, ); - expect(selectWorkspaceStructureProjects(useSessionStore.getState(), cachedServerIds)).toEqual( - [], - ); + expect(cachedSession.hasHydratedWorkspaces).toBe(false); + expect( + selectWorkspaceStructureProjects(useSessionStore.getState(), cachedServerIds).map( + (project) => project.workspaceKeys, + ), + ).toEqual([[`${SERVER_ID}:${cachedWorkspace.id}`]]); const authoritativeWorkspace = createWorkspace({ id: "authoritative-workspace", diff --git a/packages/app/src/stores/session-store-hooks/selectors.ts b/packages/app/src/stores/session-store-hooks/selectors.ts index f911e30d4..409a66055 100644 --- a/packages/app/src/stores/session-store-hooks/selectors.ts +++ b/packages/app/src/stores/session-store-hooks/selectors.ts @@ -16,6 +16,7 @@ export interface SessionsSnapshot { string, { hasHydratedWorkspaces?: boolean; + hasWorkspaceDirectorySnapshot?: boolean; workspaces: Map; projects?: Map; } @@ -138,6 +139,18 @@ export function selectHydratedWorkspaceServerIds( return serverIds.filter((serverId) => state.sessions[serverId]?.hasHydratedWorkspaces === true); } +export function selectWorkspaceDirectoryServerIds( + state: SessionsSnapshot, + serverIds: readonly string[], +): string[] { + return serverIds.filter((serverId) => { + const session = state.sessions[serverId]; + return ( + session?.hasHydratedWorkspaces === true || session?.hasWorkspaceDirectorySnapshot === true + ); + }); +} + export function selectWorkspaceStructureProjects( state: SessionsSnapshot, serverIds: readonly string[], diff --git a/packages/app/src/stores/session-store.ts b/packages/app/src/stores/session-store.ts index 758b02c28..6045ce5f4 100644 --- a/packages/app/src/stores/session-store.ts +++ b/packages/app/src/stores/session-store.ts @@ -172,6 +172,7 @@ export interface ProjectDescriptor { projectDisplayName: string; projectCustomName: string | null; projectCustomIconRevision?: string | null; + projectIconRevision?: string; projectRootPath: string; projectKind: WorkspaceDescriptorPayload["projectKind"]; } @@ -185,6 +186,7 @@ export function normalizeProjectDescriptor( projectDisplayName: payload.projectDisplayName, projectCustomName: payload.projectCustomName ?? null, projectCustomIconRevision: payload.projectCustomIconRevision ?? null, + projectIconRevision: payload.projectIconRevision, projectRootPath: payload.projectRootPath, projectKind: payload.projectKind, }; @@ -391,6 +393,7 @@ export interface SessionState { // Hydration status hasHydratedAgents: boolean; hasHydratedWorkspaces: boolean; + hasWorkspaceDirectorySnapshot: boolean; // Audio state isPlayingAudio: boolean; @@ -653,6 +656,7 @@ function createInitialSessionState( serverInfo: null, hasHydratedAgents: false, hasHydratedWorkspaces: false, + hasWorkspaceDirectorySnapshot: false, isPlayingAudio: false, focusedAgentId: null, focusedTerminalId: null, @@ -799,6 +803,7 @@ export const useSessionStore = create()( workspaceAgentActivity: buildWorkspaceAgentActivityIndex(replica.agents), workspaces: replica.workspaces, projects: replica.projects, + hasWorkspaceDirectorySnapshot: true, agentStreamTail, agentTasks, }, diff --git a/packages/app/src/utils/agent-snapshots.test.ts b/packages/app/src/utils/agent-snapshots.test.ts index 966b17eb3..334e36d9f 100644 --- a/packages/app/src/utils/agent-snapshots.test.ts +++ b/packages/app/src/utils/agent-snapshots.test.ts @@ -1,7 +1,7 @@ import { describe, expect, it } from "vitest"; import type { AgentSnapshotPayload } from "@getpaseo/protocol/messages"; import { PARENT_AGENT_ID_LABEL } from "@getpaseo/protocol/agent-labels"; -import { normalizeAgentSnapshot } from "./agent-snapshots"; +import { normalizeAgentSnapshot, projectAgentSnapshot } from "./agent-snapshots"; function createSnapshot( input: Partial> & { @@ -36,6 +36,18 @@ function createSnapshot( } describe("normalizeAgentSnapshot", () => { + it("round-trips identified active turns through the canonical snapshot boundary", () => { + const snapshot = createSnapshot({ + status: "running", + activeTurn: { turnId: "turn-1", startedAt: "2026-07-31T12:00:00.000Z" }, + }); + + expect(projectAgentSnapshot(normalizeAgentSnapshot(snapshot, "server-1"))).toMatchObject({ + status: "running", + activeTurn: snapshot.activeTurn, + }); + }); + it("normalizes identified and legacy active turns at the snapshot boundary", () => { const startedAt = "2026-07-31T12:00:00.000Z"; expect( diff --git a/packages/app/src/utils/agent-snapshots.ts b/packages/app/src/utils/agent-snapshots.ts index aeaef3ca4..a035a054a 100644 --- a/packages/app/src/utils/agent-snapshots.ts +++ b/packages/app/src/utils/agent-snapshots.ts @@ -2,6 +2,7 @@ import type { AgentSnapshotPayload } from "@getpaseo/protocol/messages"; import type { AgentPermissionRequest } from "@getpaseo/protocol/agent-types"; import { getParentAgentIdFromLabels } from "@getpaseo/protocol/agent-labels"; import type { ActiveTurnIdentity } from "@/timeline/turn-liveness"; +import type { Agent } from "@/stores/session-store"; function normalizeActiveTurn( snapshot: AgentSnapshotPayload, @@ -17,6 +18,17 @@ function normalizeActiveTurn( return snapshot.status === "running" ? { turnId: null, startedAt: lastUserMessageAt } : null; } +function projectActiveTurn(agent: Agent): Pick { + if (agent.activeTurn === null) return { activeTurn: null }; + if (agent.activeTurn.turnId === null) return {}; + return { + activeTurn: { + turnId: agent.activeTurn.turnId, + startedAt: agent.activeTurn.startedAt?.toISOString() ?? null, + }, + }; +} + export function derivePendingPermissionKey( agentId: string, request: AgentPermissionRequest, @@ -31,6 +43,37 @@ export function derivePendingPermissionKey( return `${agentId}:${fallbackId}`; } +export function projectAgentSnapshot(agent: Agent): AgentSnapshotPayload { + return { + id: agent.id, + provider: agent.provider, + cwd: agent.cwd, + ...(agent.workspaceId ? { workspaceId: agent.workspaceId } : {}), + model: agent.model, + ...(agent.features ? { features: agent.features } : {}), + thinkingOptionId: agent.thinkingOptionId ?? null, + createdAt: agent.createdAt.toISOString(), + updatedAt: agent.updatedAt.toISOString(), + lastUserMessageAt: agent.lastUserMessageAt?.toISOString() ?? null, + status: agent.status, + ...projectActiveTurn(agent), + capabilities: agent.capabilities, + currentModeId: agent.currentModeId, + availableModes: agent.availableModes, + pendingPermissions: agent.pendingPermissions, + persistence: agent.persistence, + ...(agent.runtimeInfo ? { runtimeInfo: agent.runtimeInfo } : {}), + ...(agent.lastUsage ? { lastUsage: agent.lastUsage } : {}), + ...(agent.lastError ? { lastError: agent.lastError } : {}), + title: agent.title, + labels: agent.labels, + requiresAttention: agent.requiresAttention ?? false, + attentionReason: agent.attentionReason ?? null, + attentionTimestamp: agent.attentionTimestamp?.toISOString() ?? null, + archivedAt: agent.archivedAt?.toISOString() ?? null, + }; +} + export function normalizeAgentSnapshot(snapshot: AgentSnapshotPayload, serverId: string) { const createdAt = new Date(snapshot.createdAt); const updatedAt = new Date(snapshot.updatedAt); diff --git a/packages/app/src/utils/sidebar-project-row-model.ts b/packages/app/src/utils/sidebar-project-row-model.ts index 76e42097a..4a9bf0a8b 100644 --- a/packages/app/src/utils/sidebar-project-row-model.ts +++ b/packages/app/src/utils/sidebar-project-row-model.ts @@ -5,6 +5,7 @@ export interface SidebarProjectHostTarget { projectId: string; iconWorkingDir: string; customIconRevision?: string | null; + iconRevision?: string; } export type SidebarProjectTrailingAction = @@ -25,6 +26,7 @@ function hostTarget(input: { projectId: string; iconWorkingDir: string; customIconRevision?: string | null; + iconRevision?: string; }): SidebarProjectHostTarget | null { const iconWorkingDir = input.iconWorkingDir.trim(); if (!input.serverId || !iconWorkingDir) { @@ -35,6 +37,7 @@ function hostTarget(input: { projectId: input.projectId, iconWorkingDir, customIconRevision: input.customIconRevision, + iconRevision: input.iconRevision, }; } diff --git a/packages/client/src/daemon-client.ts b/packages/client/src/daemon-client.ts index 6df09fb27..cf1053d12 100644 --- a/packages/client/src/daemon-client.ts +++ b/packages/client/src/daemon-client.ts @@ -677,6 +677,10 @@ export type ProjectListPayload = Extract< SessionOutboundMessage, { type: "project.list.response" } >["payload"]; +type ProjectListRequest = Extract; +export type ProjectListOptions = Omit & { + requestId?: string; +}; export interface CreateScheduleOptions { prompt: string; name?: string | null; @@ -2004,6 +2008,7 @@ export class DaemonClient { ...(options?.sort ? { sort: options.sort } : {}), ...(options?.page ? { page: options.page } : {}), ...(options?.subscribe ? { subscribe: options.subscribe } : {}), + ...(options?.sync ? { sync: options.sync } : {}), }); return this.sendRequest({ requestId: resolvedRequestId, @@ -2085,6 +2090,7 @@ export class DaemonClient { ...(options?.sort ? { sort: options.sort } : {}), ...(options?.page ? { page: options.page } : {}), ...(options?.subscribe ? { subscribe: options.subscribe } : {}), + ...(options?.sync ? { sync: options.sync } : {}), }); return this.sendRequest({ requestId: resolvedRequestId, @@ -2102,11 +2108,13 @@ export class DaemonClient { }); } - async listProjects(requestId?: string): Promise { + async listProjects(options?: string | ProjectListOptions): Promise { + const requestId = typeof options === "string" ? options : options?.requestId; const resolvedRequestId = this.createRequestId(requestId); const message = SessionInboundMessageSchema.parse({ type: "project.list.request", requestId: resolvedRequestId, + ...(typeof options === "object" && options.sync ? { sync: options.sync } : {}), }); return this.sendRequest({ requestId: resolvedRequestId, diff --git a/packages/protocol/src/messages.test.ts b/packages/protocol/src/messages.test.ts index 9e6f628b1..015164f70 100644 --- a/packages/protocol/src/messages.test.ts +++ b/packages/protocol/src/messages.test.ts @@ -5,6 +5,7 @@ import { parseServerInfoStatusPayload, SessionInboundMessageSchema, SessionOutboundMessageSchema, + WorkspaceProjectDescriptorPayloadSchema, } from "./messages.js"; function workspaceDescriptor(overrides: Record = {}) { @@ -53,6 +54,28 @@ describe("project icon message security", () => { }); }); +describe("project icon revision compatibility", () => { + const project = { + projectId: "project-1", + projectDisplayName: "Project", + projectRootPath: "/repo/project", + projectKind: "git" as const, + }; + + test("accepts an old project snapshot without an effective icon revision", () => { + expect(WorkspaceProjectDescriptorPayloadSchema.parse(project)).toEqual(project); + }); + + test("accepts an effective icon revision on a new project snapshot", () => { + expect( + WorkspaceProjectDescriptorPayloadSchema.parse({ + ...project, + projectIconRevision: "automatic:none:v1", + }), + ).toEqual({ ...project, projectIconRevision: "automatic:none:v1" }); + }); +}); + describe("workspace descriptor message compatibility", () => { test("old-shaped fetch_workspaces_response without project still parses", () => { const parsed = SessionOutboundMessageSchema.parse( diff --git a/packages/protocol/src/messages.ts b/packages/protocol/src/messages.ts index 1243b9adc..70f554431 100644 --- a/packages/protocol/src/messages.ts +++ b/packages/protocol/src/messages.ts @@ -1104,6 +1104,26 @@ export const SendAgentMessageSchema = z.object({ // Agent RPCs (requestId-correlated) // ============================================================================ +const DirectorySyncRequestSchema = z.object({ + generation: z.string().optional(), + afterSeq: z.number().int().nonnegative().optional(), +}); + +const DirectorySyncRemovalSchema = z.object({ + id: z.string(), + seq: z.number().int().positive(), +}); + +const DirectorySyncMetadataSchema = z.object({ + generation: z.string(), + headSeq: z.number().int().nonnegative(), + mode: z.enum(["snapshot", "changes"]), + reason: z + .enum(["no_cursor", "generation_changed", "cursor_expired", "changes_too_large"]) + .optional(), + removals: z.array(DirectorySyncRemovalSchema), +}); + export const FetchAgentsRequestMessageSchema = z.object({ type: z.literal("fetch_agents_request"), requestId: z.string(), @@ -1128,6 +1148,8 @@ export const FetchAgentsRequestMessageSchema = z.object({ subscriptionId: z.string().optional(), }) .optional(), + // COMPAT(directorySync): added in v0.3.x, remove optional after 2027-02-12. + sync: DirectorySyncRequestSchema.optional(), }); const WorkspaceStateBucketSchema = z.enum([ @@ -1168,11 +1190,15 @@ export const FetchWorkspacesRequestMessageSchema = z.object({ subscriptionId: z.string().optional(), }) .optional(), + // COMPAT(directorySync): added in v0.3.x, remove optional after 2027-02-12. + sync: DirectorySyncRequestSchema.optional(), }); export const ProjectListRequestMessageSchema = z.object({ type: z.literal("project.list.request"), requestId: z.string(), + // COMPAT(directorySync): added in v0.3.x, remove optional after 2027-02-12. + sync: DirectorySyncRequestSchema.optional(), }); export const FetchAgentHistoryRequestMessageSchema = z.object({ @@ -3038,6 +3064,8 @@ export const ServerInfoStatusPayloadSchema = z providersSnapshot: z.boolean().optional(), // COMPAT(providersSnapshotCwd): added in v0.3.2, remove gate after 2027-02-10. providersSnapshotCwd: z.boolean().optional(), + // COMPAT(directorySync): added in v0.3.x, remove gate after 2027-02-12. + directorySync: z.boolean().optional(), // COMPAT(checkoutForgeSetAutoMerge): added in v0.1.106, remove old // checkoutGithubSetAutoMerge fallback after 2026-12-28. checkoutForgeSetAutoMerge: z.boolean().optional(), @@ -3457,6 +3485,8 @@ export const WorkspaceDescriptorPayloadSchema = z // Old daemons omit it; absent means the client falls back to GitHub. forge: z.string().optional(), project: ProjectPlacementPayloadSchema.optional(), + // COMPAT(directorySync): sequence of this latest directory projection. + syncSeq: z.number().int().positive().optional(), }) .transform((workspace) => ({ ...workspace, @@ -3470,10 +3500,14 @@ export const AgentUpdateMessageSchema = z.object({ kind: z.literal("upsert"), agent: AgentSnapshotPayloadSchema, project: ProjectPlacementPayloadSchema.nullable().optional(), + generation: z.string().optional(), + seq: z.number().int().positive().optional(), }), z.object({ kind: z.literal("remove"), agentId: z.string(), + generation: z.string().optional(), + seq: z.number().int().positive().optional(), }), ]), }); @@ -3531,6 +3565,8 @@ const AgentDirectoryResponseEntrySchema = z.object({ // anyway; sending it keeps the client from re-deriving a second opinion that // could disagree with the ranking it is explaining. searchMatches: z.array(AgentSearchMatchSchema).optional(), + // COMPAT(directorySync): sequence of this latest directory projection. + syncSeq: z.number().int().positive().optional(), }); const AgentDirectoryPageInfoSchema = z.object({ @@ -3546,6 +3582,7 @@ export const FetchAgentsResponseMessageSchema = z.object({ subscriptionId: z.string().nullable().optional(), entries: z.array(AgentDirectoryResponseEntrySchema), pageInfo: AgentDirectoryPageInfoSchema, + sync: DirectorySyncMetadataSchema.optional(), }), }); @@ -3583,8 +3620,14 @@ export const WorkspaceProjectDescriptorPayloadSchema = z.object({ projectCustomName: z.string().nullable().optional(), // COMPAT(projectCustomIcon): added in v0.2.0, remove after 2027-01-20. projectCustomIconRevision: z.string().nullable().optional(), + // Fingerprints the effective icon, including automatic discovery and the + // absence of an icon. Clients may persist icon results against this value. + // COMPAT(projectIconCache): added in v0.2.7, remove optional after 2027-02-12. + projectIconRevision: z.string().optional(), projectRootPath: z.string(), projectKind: z.enum(["git", "non_git", "directory"]), + // COMPAT(directorySync): sequence of this latest directory projection. + syncSeq: z.number().int().positive().optional(), }); export const FetchWorkspacesResponseMessageSchema = z.object({ @@ -3602,6 +3645,7 @@ export const FetchWorkspacesResponseMessageSchema = z.object({ prevCursor: z.string().nullable(), hasMore: z.boolean(), }), + sync: DirectorySyncMetadataSchema.optional(), }), }); @@ -3611,6 +3655,8 @@ export const WorkspaceUpdateMessageSchema = z.object({ z.object({ kind: z.literal("upsert"), workspace: WorkspaceDescriptorPayloadSchema, + generation: z.string().optional(), + seq: z.number().int().positive().optional(), }), z.object({ kind: z.literal("remove"), @@ -3625,6 +3671,8 @@ export const WorkspaceUpdateMessageSchema = z.object({ // Project removal is represented on the existing workspace update channel // so old clients can still parse the message and ignore the extra field. removedProjectId: z.string().optional(), + generation: z.string().optional(), + seq: z.number().int().positive().optional(), }), ]), }); @@ -3632,8 +3680,18 @@ export const WorkspaceUpdateMessageSchema = z.object({ export const ProjectUpdateMessageSchema = z.object({ type: z.literal("project.update"), payload: z.discriminatedUnion("kind", [ - z.object({ kind: z.literal("upsert"), project: WorkspaceProjectDescriptorPayloadSchema }), - z.object({ kind: z.literal("remove"), projectId: z.string() }), + z.object({ + kind: z.literal("upsert"), + project: WorkspaceProjectDescriptorPayloadSchema, + generation: z.string().optional(), + seq: z.number().int().positive().optional(), + }), + z.object({ + kind: z.literal("remove"), + projectId: z.string(), + generation: z.string().optional(), + seq: z.number().int().positive().optional(), + }), ]), }); @@ -3642,6 +3700,7 @@ export const ProjectListResponseMessageSchema = z.object({ payload: z.object({ requestId: z.string(), projects: z.array(WorkspaceProjectDescriptorPayloadSchema), + sync: DirectorySyncMetadataSchema.optional(), }), }); diff --git a/packages/protocol/src/messages.workspaces.test.ts b/packages/protocol/src/messages.workspaces.test.ts index 03a0bd432..bdf985152 100644 --- a/packages/protocol/src/messages.workspaces.test.ts +++ b/packages/protocol/src/messages.workspaces.test.ts @@ -93,6 +93,59 @@ describe("workspace message schemas", () => { expect(activeScoped.scope).toBe("active"); }); + test("parses optional sequenced directory requests and responses", () => { + expect( + SessionInboundMessageSchema.parse({ + type: "project.list.request", + requestId: "projects-sync", + sync: { generation: "daemon-generation", afterSeq: 7 }, + }), + ).toMatchObject({ sync: { generation: "daemon-generation", afterSeq: 7 } }); + expect( + SessionInboundMessageSchema.parse({ + type: "fetch_workspaces_request", + requestId: "workspaces-sync", + sync: { generation: "daemon-generation", afterSeq: 11 }, + }), + ).toMatchObject({ sync: { generation: "daemon-generation", afterSeq: 11 } }); + expect( + SessionInboundMessageSchema.parse({ + type: "fetch_agents_request", + requestId: "agents-sync", + scope: "active", + sync: { generation: "daemon-generation", afterSeq: 13 }, + }), + ).toMatchObject({ sync: { generation: "daemon-generation", afterSeq: 13 } }); + + const response = SessionOutboundMessageSchema.parse({ + type: "project.list.response", + payload: { + requestId: "projects-sync", + projects: [ + { + projectId: "project-1", + projectDisplayName: "Project", + projectRootPath: "/repo", + projectKind: "git", + syncSeq: 12, + }, + ], + sync: { + generation: "daemon-generation", + headSeq: 12, + mode: "changes", + removals: [{ id: "project-removed", seq: 10 }], + }, + }, + }); + expect(response).toMatchObject({ + payload: { + projects: [{ projectId: "project-1", syncSeq: 12 }], + sync: { headSeq: 12, removals: [{ id: "project-removed", seq: 10 }] }, + }, + }); + }); + test("parses agent_update without project placement", () => { const result = SessionOutboundMessageSchema.safeParse({ type: "agent_update", diff --git a/packages/server/src/server/directory-sync/index.test.ts b/packages/server/src/server/directory-sync/index.test.ts new file mode 100644 index 000000000..0d41d449a --- /dev/null +++ b/packages/server/src/server/directory-sync/index.test.ts @@ -0,0 +1,92 @@ +import { describe, expect, test } from "vitest"; +import { DirectorySyncService } from "./index.js"; + +function project(projectId: string, name: string) { + return { + projectId, + projectDisplayName: name, + projectRootPath: `/${projectId}`, + projectKind: "git" as const, + }; +} + +describe("DirectorySyncService", () => { + test("owns sequencing and returns wire-ready latest project changes", () => { + const service = new DirectorySyncService("generation"); + expect(service.synchronizeProjects([project("one", "One")], {}).sync).toMatchObject({ + mode: "snapshot", + headSeq: 1, + }); + + service.sequenceProjectUpdate({ kind: "upsert", project: project("one", "Renamed") }, true); + service.sequenceProjectUpdate({ kind: "upsert", project: project("one", "Latest") }, true); + expect( + service.synchronizeProjects([project("one", "Latest")], { + generation: "generation", + afterSeq: 1, + }), + ).toEqual({ + projects: [{ ...project("one", "Latest"), syncSeq: 3 }], + sync: { generation: "generation", mode: "changes", headSeq: 3, removals: [] }, + }); + }); + + test("sequences a hard project removal once", () => { + const service = new DirectorySyncService("generation"); + service.synchronizeProjects([project("one", "One")], {}); + const update = { kind: "remove" as const, projectId: "one" }; + expect(service.sequenceProjectUpdate(update, true)).toEqual({ + ...update, + generation: "generation", + seq: 2, + }); + expect(service.sequenceProjectUpdate(update, true)).toEqual({ + ...update, + generation: "generation", + seq: 2, + }); + expect( + service.synchronizeProjects([], { generation: "generation", afterSeq: 1 }).sync.removals, + ).toEqual([{ id: "one", seq: 2 }]); + }); + + test("keeps one daemon-wide version for repeated observations", () => { + const service = new DirectorySyncService("generation"); + const update = { kind: "upsert" as const, project: project("one", "One") }; + const sequenced = { ...update, generation: "generation", seq: 1 }; + expect(service.sequenceProjectUpdate(update, true)).toEqual(sequenced); + expect(service.sequenceProjectUpdate(update, true)).toEqual(sequenced); + }); + + test("sequences only the project whose effective icon changed", () => { + const service = new DirectorySyncService("generation"); + const one = { ...project("one", "One"), projectIconRevision: "icon-a" }; + const two = { ...project("two", "Two"), projectIconRevision: "icon-a" }; + service.synchronizeProjects([one, two], {}); + const changed = { ...one, projectIconRevision: "icon-b" }; + service.sequenceProjectUpdate({ kind: "upsert", project: changed }, true); + + expect( + service.synchronizeProjects([changed, two], { + generation: "generation", + afterSeq: 2, + }), + ).toEqual({ + projects: [{ ...changed, syncSeq: 3 }], + sync: { generation: "generation", mode: "changes", headSeq: 3, removals: [] }, + }); + }); + + test("falls back to a snapshot when the daemon generation changes", () => { + const service = new DirectorySyncService("new-generation"); + const read = service.synchronizeProjects([project("one", "One")], { + generation: "old-generation", + afterSeq: 10, + }); + expect(read.sync).toMatchObject({ + mode: "snapshot", + reason: "generation_changed", + headSeq: 1, + }); + }); +}); diff --git a/packages/server/src/server/directory-sync/index.ts b/packages/server/src/server/directory-sync/index.ts new file mode 100644 index 000000000..1b08360cb --- /dev/null +++ b/packages/server/src/server/directory-sync/index.ts @@ -0,0 +1,174 @@ +import { randomUUID } from "node:crypto"; +import type { + AgentSnapshotPayload, + ProjectPlacementPayload, + SessionOutboundMessage, + WorkspaceDescriptorPayload, + WorkspaceProjectDescriptorPayload, +} from "@getpaseo/protocol/messages"; +import { type CollectionRead, VersionedCollection } from "./internal/versioned-collection.js"; + +interface AgentDirectoryEntry { + agent: AgentSnapshotPayload; + project: ProjectPlacementPayload; +} + +interface DirectorySyncCursor { + generation?: string; + afterSeq?: number; +} + +interface DirectoryVersion { + generation: string; + seq: number; +} + +type ProjectUpdate = Extract; +type ProjectListResponse = Extract< + SessionOutboundMessage, + { type: "project.list.response" } +>["payload"]; +type FetchWorkspacesResponse = Extract< + SessionOutboundMessage, + { type: "fetch_workspaces_response" } +>["payload"]; +type FetchAgentsResponse = Extract< + SessionOutboundMessage, + { type: "fetch_agents_response" } +>["payload"]; + +/** Daemon-global latest-state sequence owner for the active app directory. */ +export class DirectorySyncService { + private readonly generation: string; + private readonly projects = new VersionedCollection({ + getId: (project) => project.projectId, + }); + private readonly workspaces = new VersionedCollection({ + getId: (workspace) => workspace.id, + }); + private readonly agents = new VersionedCollection({ + getId: (entry) => entry.agent.id, + }); + + constructor(generation = randomUUID()) { + this.generation = generation; + } + + synchronizeProjects( + snapshot: Iterable, + cursor: DirectorySyncCursor, + ): Pick { + this.projects.replaceAll(snapshot); + const read = this.read(this.projects, cursor); + return { + projects: read.values.map(({ seq, value }) => ({ ...value, syncSeq: seq })), + sync: this.metadata(read), + }; + } + + synchronizeWorkspaces( + snapshot: Iterable, + cursor: DirectorySyncCursor, + ): Pick { + this.workspaces.replaceAll(snapshot); + const read = this.read(this.workspaces, cursor); + return { + entries: read.values.map(({ seq, value }) => ({ ...value, syncSeq: seq })), + emptyProjects: [], + pageInfo: { nextCursor: null, prevCursor: null, hasMore: false }, + sync: this.metadata(read), + }; + } + + synchronizeAgents( + snapshot: Iterable, + cursor: DirectorySyncCursor, + ): Pick { + this.agents.replaceAll(snapshot); + const read = this.read(this.agents, cursor); + return { + entries: read.values.map(({ seq, value }) => ({ ...value, syncSeq: seq })), + pageInfo: { nextCursor: null, prevCursor: null, hasMore: false }, + sync: this.metadata(read), + }; + } + + sequenceProjectUpdate( + update: ProjectUpdate["payload"], + includeSequence: boolean, + ): ProjectUpdate["payload"] { + if (update.kind === "upsert") { + this.projects.replace(update.project); + } else { + this.projects.remove(update.projectId); + } + const id = update.kind === "upsert" ? update.project.projectId : update.projectId; + return this.withVersion(update, includeSequence, this.version(this.projects, id)); + } + + sequenceWorkspaceUpdate( + payload: T, + value: WorkspaceDescriptorPayload | null, + id: string, + includeSequence: boolean, + ): T & Partial { + if (value) { + this.workspaces.replace(value); + } else { + this.workspaces.remove(id); + } + return this.withVersion(payload, includeSequence, this.version(this.workspaces, id)); + } + + sequenceAgentUpdate( + payload: T, + value: AgentDirectoryEntry | null, + id: string, + includeSequence: boolean, + ): T & Partial { + if (value) { + this.agents.replace(value); + } else { + this.agents.remove(id); + } + return this.withVersion(payload, includeSequence, this.version(this.agents, id)); + } + + private metadata(read: CollectionRead) { + return { + generation: this.generation, + headSeq: read.headSeq, + mode: read.mode, + ...(read.reason ? { reason: read.reason } : {}), + removals: read.removals, + }; + } + + private withVersion( + payload: T, + includeSequence: boolean, + version: DirectoryVersion | null, + ): T & Partial { + return version && includeSequence ? { ...payload, ...version } : payload; + } + + private read( + collection: VersionedCollection, + cursor: DirectorySyncCursor, + ): CollectionRead { + if (cursor.generation !== this.generation) { + return collection.readSnapshot( + cursor.generation === undefined ? "no_cursor" : "generation_changed", + ); + } + if (cursor.afterSeq === undefined) { + return collection.readSnapshot("no_cursor"); + } + return collection.readAfter(cursor.afterSeq); + } + + private version(collection: VersionedCollection, id: string): DirectoryVersion | null { + const seq = collection.sequenceFor(id); + return seq === null ? null : { generation: this.generation, seq }; + } +} diff --git a/packages/server/src/server/directory-sync/internal/versioned-collection.test.ts b/packages/server/src/server/directory-sync/internal/versioned-collection.test.ts new file mode 100644 index 000000000..f74cdd787 --- /dev/null +++ b/packages/server/src/server/directory-sync/internal/versioned-collection.test.ts @@ -0,0 +1,77 @@ +import { describe, expect, test } from "vitest"; +import { VersionedCollection } from "./versioned-collection.js"; + +describe("VersionedCollection", () => { + test("returns only each entity's latest visible snapshot after a sequence", () => { + const collection = new VersionedCollection<{ id: string; title: string }>({ + getId: (value) => value.id, + }); + collection.replace({ id: "one", title: "first" }); + const cursor = collection.headSeq; + collection.replace({ id: "one", title: "second" }); + collection.replace({ id: "one", title: "latest" }); + collection.replace({ id: "two", title: "other" }); + + expect(collection.readAfter(cursor)).toEqual({ + mode: "changes", + headSeq: 4, + values: [ + { seq: 3, value: { id: "one", title: "latest" } }, + { seq: 4, value: { id: "two", title: "other" } }, + ], + removals: [], + }); + }); + + test("does not advance for an unchanged projection or repeated removal", () => { + const collection = new VersionedCollection<{ id: string; title: string }>({ + getId: (value) => value.id, + }); + expect(collection.replace({ id: "one", title: "same" })?.seq).toBe(1); + expect(collection.replace({ id: "one", title: "same" })).toBeNull(); + expect(collection.remove("one")?.seq).toBe(2); + expect(collection.remove("one")).toBeNull(); + expect(collection.headSeq).toBe(2); + }); + + test("returns tombstones and lets a later value replace them", () => { + const collection = new VersionedCollection<{ id: string; title: string }>({ + getId: (value) => value.id, + }); + collection.replace({ id: "one", title: "before" }); + collection.remove("one"); + expect(collection.readAfter(1)).toEqual({ + mode: "changes", + headSeq: 2, + values: [], + removals: [{ id: "one", seq: 2 }], + }); + + collection.replace({ id: "one", title: "after" }); + expect(collection.readAfter(2)).toEqual({ + mode: "changes", + headSeq: 3, + values: [{ seq: 3, value: { id: "one", title: "after" } }], + removals: [], + }); + }); + + test("falls back to a latest snapshot when a required tombstone expired", () => { + const collection = new VersionedCollection<{ id: string }>({ + getId: (value) => value.id, + maxTombstones: 1, + }); + collection.replace({ id: "one" }); + collection.replace({ id: "two" }); + collection.remove("one"); + collection.remove("two"); + + expect(collection.readAfter(2)).toEqual({ + mode: "snapshot", + reason: "cursor_expired", + headSeq: 4, + values: [], + removals: [], + }); + }); +}); diff --git a/packages/server/src/server/directory-sync/internal/versioned-collection.ts b/packages/server/src/server/directory-sync/internal/versioned-collection.ts new file mode 100644 index 000000000..e25bc958f --- /dev/null +++ b/packages/server/src/server/directory-sync/internal/versioned-collection.ts @@ -0,0 +1,141 @@ +export interface VersionedValue { + seq: number; + value: T; +} + +export interface VersionedRemoval { + id: string; + seq: number; +} + +export interface CollectionRead { + mode: "changes" | "snapshot"; + reason?: "no_cursor" | "generation_changed" | "cursor_expired"; + headSeq: number; + values: VersionedValue[]; + removals: VersionedRemoval[]; +} + +export type PublishedChange = + | { kind: "upsert"; seq: number; value: T } + | { kind: "remove"; seq: number; id: string }; + +interface StoredValue extends VersionedValue { + fingerprint: string; +} + +interface VersionedCollectionOptions { + getId(value: T): string; + fingerprint?(value: T): string; + maxTombstones?: number; +} + +/** Owns latest-state sequencing for one directory entity type. */ +export class VersionedCollection { + private readonly valuesById = new Map>(); + private readonly tombstonesById = new Map(); + private readonly getId: (value: T) => string; + private readonly fingerprint: (value: T) => string; + private readonly maxTombstones: number; + private expiredThroughSeq = 0; + private nextSeq = 0; + + constructor(options: VersionedCollectionOptions) { + this.getId = options.getId; + this.fingerprint = options.fingerprint ?? ((value) => JSON.stringify(value)); + this.maxTombstones = options.maxTombstones ?? 5_000; + } + + get headSeq(): number { + return this.nextSeq; + } + + sequenceFor(id: string): number | null { + return this.valuesById.get(id)?.seq ?? this.tombstonesById.get(id) ?? null; + } + + replace(value: T): Extract, { kind: "upsert" }> | null { + const id = this.getId(value); + const fingerprint = this.fingerprint(value); + const previous = this.valuesById.get(id); + if (previous?.fingerprint === fingerprint) return null; + + const seq = this.advance(); + this.valuesById.set(id, { seq, fingerprint, value }); + this.tombstonesById.delete(id); + return { kind: "upsert", seq, value }; + } + + remove(id: string): Extract, { kind: "remove" }> | null { + if (!this.valuesById.has(id) && this.tombstonesById.has(id)) return null; + if (!this.valuesById.delete(id) && !this.tombstonesById.has(id)) return null; + + const seq = this.advance(); + this.tombstonesById.delete(id); + this.tombstonesById.set(id, seq); + this.trimTombstones(); + return { kind: "remove", seq, id }; + } + + replaceAll(values: Iterable): void { + const retainedIds = new Set(); + for (const value of values) { + const id = this.getId(value); + retainedIds.add(id); + this.replace(value); + } + for (const id of this.valuesById.keys()) { + if (!retainedIds.has(id)) this.remove(id); + } + } + + readSnapshot( + reason: "no_cursor" | "generation_changed" | "cursor_expired" = "no_cursor", + ): CollectionRead { + return { + mode: "snapshot", + reason, + headSeq: this.headSeq, + values: this.sortedValues(), + removals: [], + }; + } + + readAfter(afterSeq: number): CollectionRead { + if (afterSeq < this.expiredThroughSeq || afterSeq > this.headSeq) { + return this.readSnapshot("cursor_expired"); + } + return { + mode: "changes", + headSeq: this.headSeq, + values: this.sortedValues().filter((entry) => entry.seq > afterSeq), + removals: Array.from(this.tombstonesById, ([id, seq]) => ({ id, seq })) + .filter((entry) => entry.seq > afterSeq) + .sort((left, right) => left.seq - right.seq || left.id.localeCompare(right.id)), + }; + } + + private advance(): number { + this.nextSeq += 1; + return this.nextSeq; + } + + private sortedValues(): VersionedValue[] { + return Array.from(this.valuesById.values()) + .map(({ seq, value }) => ({ seq, value })) + .sort( + (left, right) => + left.seq - right.seq || this.getId(left.value).localeCompare(this.getId(right.value)), + ); + } + + private trimTombstones(): void { + while (this.tombstonesById.size > this.maxTombstones) { + const oldest = this.tombstonesById.entries().next().value; + if (!oldest) return; + const [id, seq] = oldest; + this.tombstonesById.delete(id); + this.expiredThroughSeq = Math.max(this.expiredThroughSeq, seq); + } + } +} diff --git a/packages/server/src/server/session.test.ts b/packages/server/src/server/session.test.ts index 81312a757..663e147ef 100644 --- a/packages/server/src/server/session.test.ts +++ b/packages/server/src/server/session.test.ts @@ -710,6 +710,7 @@ describe("project command-center RPCs", () => { projectDisplayName: "new-project", projectCustomName: null, projectCustomIconRevision: null, + projectIconRevision: "automatic:none:v1", projectRootPath: directoryPath, projectKind: "non_git", }, @@ -4967,7 +4968,7 @@ test("keeps selective delivery scoped per socket when a retained session also ha ]); }); -test("sends project updates only to capable sockets in a retained session", () => { +test("sends project updates only to capable sockets in a retained session", async () => { const messages: SessionOutboundMessage[] = []; const targetedMessages: Array<{ source: object; message: SessionOutboundMessage }> = []; const session = createSessionForTest({ messages, targetedMessages }); @@ -4976,7 +4977,7 @@ test("sends project updates only to capable sockets in a retained session", () = session.updateClientCapabilities(null, legacySocket); session.updateClientCapabilities({ [CLIENT_CAPS.projectUpdates]: true }, capableSocket); - session.emitProjectUpdate({ + await session.emitProjectUpdate({ kind: "upsert", project: createPersistedProjectRecord({ projectId: "project-capable-socket", @@ -5039,6 +5040,7 @@ test("project.list returns every active project descriptor", async () => { projectDisplayName: "acme/app", projectCustomName: null, projectCustomIconRevision: null, + projectIconRevision: "automatic:none:v1", projectRootPath: "/tmp/project-active", projectKind: "git", }, diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 1fb8228a8..1710a7072 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -70,6 +70,7 @@ import { CLIENT_SHUTDOWN_RPC_REASON, normalizeClientRestartRpcReason, } from "./lifecycle-reasons.js"; +import { DirectorySyncService } from "./directory-sync/index.js"; import { AgentManager, AgentRunCancellationError } from "./agent/agent-manager.js"; import { buildTimelinePromptIndex } from "./agent/timeline-prompt-index.js"; @@ -144,7 +145,7 @@ import { import { wrapSpokenInput } from "./voice-config.js"; import { isVoicePermissionAllowed } from "./voice-permission-policy.js"; import { - readProjectIcon, + ProjectIconReader, removeProjectCustomIcon, setProjectCustomIcon, } from "../utils/project-custom-icon.js"; @@ -394,6 +395,7 @@ type WorkspaceUpdatePayload = Extract< >["payload"]; interface WorkspaceUpdatesSubscriptionState { subscriptionId: string; + syncEnabled?: boolean; filter?: WorkspaceUpdatesFilter; isBootstrapping: boolean; pendingUpdatesByWorkspaceId: Map; @@ -446,6 +448,7 @@ export interface SessionOptions { agentStorage: AgentStorage; projectRegistry: ProjectRegistry; workspaceRegistry: WorkspaceRegistry; + directorySync?: DirectorySyncService; filesystem?: SessionFileSystem; scheduleService: ScheduleService; checkoutDiffManager: CheckoutDiffManager; @@ -579,6 +582,10 @@ interface WorkspaceUpdateOptions { optimisticStatus?: WorkspaceDescriptorPayload["status"]; } +function resolveDirectorySync(service: DirectorySyncService | undefined): DirectorySyncService { + return service ?? new DirectorySyncService(); +} + function describeRegistryTransition(record: ArchivedRecordSnapshot | null): RegistryTransition { if (!record) { return "created"; @@ -612,12 +619,14 @@ export class Session { | null; private readonly sessionLogger: pino.Logger; private readonly paseoHome: string; + private readonly projectIcons: ProjectIconReader; private readonly worktreesRoot: string | undefined; private agentManager: AgentManager; private readonly agentStorage: AgentStorage; private readonly projectRegistry: ProjectRegistry; private readonly workspaceRegistry: WorkspaceRegistry; + private readonly directorySync: DirectorySyncService; private readonly filesystem: SessionFileSystem; private readonly github: ForgeService; private readonly renameCurrentBranch: typeof renameCurrentBranchDefault; @@ -632,6 +641,7 @@ export class Session { private unsubscribeProjectMutations: (() => void) | null = null; private unsubscribeWorkspaceMutations: (() => void) | null = null; private registryMutationQueue: Promise = Promise.resolve(); + private projectUpdateQueue: Promise = Promise.resolve(); private isCleanedUp = false; private viewedTimelineAgentIds = new Set(); private readonly viewedTimelineAgentIdsBySource = new Map>(); @@ -640,6 +650,7 @@ export class Session { private unsubscribeTerminalWorkspaceContributionEvents: (() => void) | null = null; private readonly agentUpdates: AgentUpdatesService; private workspaceUpdatesSubscription: WorkspaceUpdatesSubscriptionState | null = null; + private projectSyncEnabled = false; private readonly workspaceUpdateTails = new Map>(); private clientActivity: { deviceType: "web" | "mobile"; @@ -699,6 +710,7 @@ export class Session { agentStorage, projectRegistry, workspaceRegistry, + directorySync, filesystem, scheduleService, checkoutDiffManager, @@ -744,6 +756,7 @@ export class Session { this.onWorkspaceRecovered = onWorkspaceRecovered ?? null; this.pushNotifications = pushNotifications; this.paseoHome = paseoHome; + this.projectIcons = new ProjectIconReader(paseoHome); this.worktreesRoot = worktreesRoot; this.sessionLogger = logger.child({ module: "session", @@ -764,6 +777,7 @@ export class Session { this.agentStorage = agentStorage; this.projectRegistry = projectRegistry; this.workspaceRegistry = workspaceRegistry; + this.directorySync = resolveDirectorySync(directorySync); this.filesystem = filesystem ?? nodeSessionFileSystem; this.github = github ?? createGitHubService(); this.renameCurrentBranch = renameCurrentBranch ?? renameCurrentBranchDefault; @@ -922,6 +936,13 @@ export class Session { this.buildProjectPlacementForWorkspaceId(workspaceId), emitWorkspaceUpdateForWorkspaceId: (workspaceId) => this.emitWorkspaceUpdateForWorkspaceId(workspaceId), + sequenceAgentUpdate: (payload, agent, project, agentId, includeSequence) => + this.directorySync.sequenceAgentUpdate( + payload, + agent && project ? { agent, project } : null, + agentId, + includeSequence, + ), logger: this.sessionLogger, }); this.createAgentLifecycleDispatch = new CreateAgentLifecycleDispatch({ @@ -1141,13 +1162,22 @@ export class Session { ); } - emitProjectUpdate(update: ProjectUpdate): void { + emitProjectUpdate(update: ProjectUpdate): Promise { + const queued = this.projectUpdateQueue + .catch(() => undefined) + .then(() => this.publishProjectUpdate(update)); + this.projectUpdateQueue = queued; + return queued; + } + + private async publishProjectUpdate(update: ProjectUpdate): Promise { + const projectedPayload = + update.kind === "upsert" + ? { kind: "upsert" as const, project: await this.buildProjectDescriptor(update.project) } + : update; const message: SessionOutboundMessage = { type: "project.update", - payload: - update.kind === "upsert" - ? { kind: "upsert", project: this.buildProjectDescriptor(update.project) } - : update, + payload: this.directorySync.sequenceProjectUpdate(projectedPayload, this.projectSyncEnabled), }; if (this.clientCapabilitiesBySource.size === 0 || !this.onMessageToSource) { if (this.supports(CLIENT_CAPS.projectUpdates)) this.emit(message); @@ -2121,7 +2151,7 @@ export class Session { case "fetch_workspaces_request": return this.handleFetchWorkspacesRequest(msg); case "project.list.request": - return this.handleProjectListRequest(msg.requestId); + return this.handleProjectListRequest(msg); case "paseo_worktree_list_request": return this.handlePaseoWorktreeListRequest(msg); case "paseo_worktree_archive_request": @@ -2692,7 +2722,7 @@ export class Session { // Emit a project.update so clients that track the project as an empty // project (no workspaces yet) receive the resolved name immediately. - this.emitProjectUpdate({ kind: "upsert", project: updated }); + await this.emitProjectUpdate({ kind: "upsert", project: updated }); // Re-emit descriptors for every workspace under this project so the new // resolved name lands in the UI immediately. @@ -2746,7 +2776,7 @@ export class Session { type: "project.icon.set.response", payload: { requestId, projectId, accepted: true, error: null }, }); - this.emitProjectUpdate({ kind: "upsert", project: updated }); + await this.emitProjectUpdate({ kind: "upsert", project: updated }); const affectedWorkspaceIds = (await this.workspaceRegistry.list()) .filter((workspace) => workspace.projectId === projectId) @@ -2772,7 +2802,7 @@ export class Session { const project = await this.projectRegistry.get(projectId); if (!project) throw new Error("Project not found"); - const icon = await readProjectIcon({ paseoHome: this.paseoHome, project }); + const icon = await this.projectIcons.read(project); this.emit({ type: "project.icon.get.response", payload: { projectId, icon, error: null, requestId }, @@ -4684,15 +4714,17 @@ export class Session { } } - private buildProjectDescriptor( + private async buildProjectDescriptor( project: PersistedProjectRecord, - ): WorkspaceProjectDescriptorPayload { + ): Promise { + const icon = await this.projectIcons.snapshot(project); return { projectId: project.projectId, ...(project.projectKey ? { projectKey: project.projectKey } : {}), projectDisplayName: resolveProjectDisplayName(project), projectCustomName: project.customName ?? null, projectCustomIconRevision: project.customIconRevision ?? null, + projectIconRevision: icon.revision, projectRootPath: project.rootPath, projectKind: project.kind, }; @@ -4890,7 +4922,6 @@ export class Session { continue; } this.workspaceGitObserver.recordDescriptorState(workspaceId, nextWorkspace); - if (!nextWorkspace) { if (this.shouldSkipWorkspaceRemoval(lastEmitted, options?.removedProjectId)) { continue; @@ -4899,17 +4930,28 @@ export class Session { return; } subscription.lastEmittedByWorkspaceId.delete(workspaceId); + const removePayload = await this.buildWorkspaceRemoveUpdatePayload( + workspaceId, + options?.removedProjectId, + ); this.bufferOrEmitWorkspaceUpdate( subscription, - await this.buildWorkspaceRemoveUpdatePayload(workspaceId, options?.removedProjectId), + this.directorySync.sequenceWorkspaceUpdate( + removePayload, + workspace ?? null, + workspaceId, + subscription.syncEnabled === true, + ), ); continue; } - const nextPayload: WorkspaceUpdatePayload = { - kind: "upsert", - workspace: nextWorkspace, - }; + const nextPayload: WorkspaceUpdatePayload = this.directorySync.sequenceWorkspaceUpdate( + { kind: "upsert", workspace: nextWorkspace }, + workspace ?? null, + workspaceId, + subscription.syncEnabled === true, + ); if ( lastEmitted && @@ -5013,10 +5055,13 @@ export class Session { this.agentUpdates.beginSubscription({ subscriptionId, filter: request.filter, + syncEnabled: Boolean(request.sync), }); } - const payload = await this.listFetchAgentsEntries(request); + const payload = request.sync + ? await this.readAgentDirectorySync(request) + : await this.listFetchAgentsEntries(request); const snapshotUpdatedAtByAgentId = new Map(); for (const entry of payload.entries) { const parsedUpdatedAt = Date.parse(entry.agent.updatedAt); @@ -5147,6 +5192,7 @@ export class Session { if (subscriptionId) { this.workspaceUpdatesSubscription = { subscriptionId, + syncEnabled: Boolean(request.sync), filter: request.filter, isBootstrapping: true, pendingUpdatesByWorkspaceId: new Map(), @@ -5155,7 +5201,9 @@ export class Session { }; } - const payload = await this.listFetchWorkspacesEntries(request); + const payload = request.sync + ? await this.readWorkspaceDirectorySync(request) + : await this.listFetchWorkspacesEntries(request); this.workspaceGitObserver.syncObservers(payload.entries); this.sessionLogger.debug( { @@ -5205,21 +5253,32 @@ export class Session { } } - private async handleProjectListRequest(requestId: string): Promise { + private async handleProjectListRequest( + request: Extract, + ): Promise { try { - const projects = (await this.projectRegistry.list()) - .filter((project) => !project.archivedAt) - .map((project) => this.buildProjectDescriptor(project)); + const projects = await Promise.all( + (await this.projectRegistry.list()) + .filter((project) => !project.archivedAt) + .map((project) => this.buildProjectDescriptor(project)), + ); + const synchronized = request.sync + ? this.directorySync.synchronizeProjects(projects, request.sync) + : null; + if (synchronized) this.projectSyncEnabled = true; this.emit({ type: "project.list.response", - payload: { requestId, projects }, + payload: { + requestId: request.requestId, + ...(synchronized ?? { projects }), + }, }); } catch (error) { this.sessionLogger.error({ err: error }, "Failed to handle project.list.request"); this.emit({ type: "rpc_error", payload: { - requestId, + requestId: request.requestId, requestType: "project.list.request", error: error instanceof Error ? error.message : "Failed to list projects", code: "project_list_failed", @@ -5228,6 +5287,37 @@ export class Session { } } + private async readAgentDirectorySync( + request: Extract, + ) { + if (request.scope !== "active" || request.filter) { + throw new SessionRequestError( + "invalid_request", + "Sequenced agent directory reads require scope=active and no filter.", + ); + } + const snapshot = await this.listFetchAgentsEntries({ + ...request, + page: { limit: Number.MAX_SAFE_INTEGER }, + }); + return this.directorySync.synchronizeAgents(snapshot.entries, request.sync ?? {}); + } + + private async readWorkspaceDirectorySync( + request: Extract, + ) { + if (request.filter) { + throw new SessionRequestError( + "invalid_request", + "Sequenced workspace directory reads do not support filters.", + ); + } + return this.directorySync.synchronizeWorkspaces( + await this.workspaceDirectory.listDescriptors(), + request.sync ?? {}, + ); + } + // Build the bootstrap snapshot used by `flushBootstrappedWorkspaceUpdates` // to decide which pending updates to drop. Captures the status, // statusEnteredAt, and activityAt (parsed to ms) for each workspace entry @@ -5580,7 +5670,7 @@ export class Session { type: "project.add.response", payload: { requestId: request.requestId, - project: this.buildProjectDescriptor(project), + project: await this.buildProjectDescriptor(project), error: null, }, }); @@ -5614,7 +5704,7 @@ export class Session { payload: { requestId: request.requestId, directoryPath: result.directoryPath, - project: this.buildProjectDescriptor(result.project), + project: await this.buildProjectDescriptor(result.project), error: null, errorCode: null, }, @@ -5779,7 +5869,7 @@ export class Session { requestId: request.requestId, repo: repo.displayName, checkoutPath, - project: this.buildProjectDescriptor(project), + project: await this.buildProjectDescriptor(project), error: null, }, }); diff --git a/packages/server/src/server/session.workspaces.test.ts b/packages/server/src/server/session.workspaces.test.ts index 572680f9a..d72e0b799 100644 --- a/packages/server/src/server/session.workspaces.test.ts +++ b/packages/server/src/server/session.workspaces.test.ts @@ -736,6 +736,55 @@ function createSessionForWorkspaceTests( return session; } +test("project.list.request catches up by sequence on the existing RPC", async () => { + const emitted: SessionOutboundMessage[] = []; + let project = createPersistedProjectRecord({ + projectId: "project-sequenced", + rootPath: "/repo/sequenced", + kind: "git", + displayName: "Before", + createdAt: "2026-08-12T08:00:00.000Z", + updatedAt: "2026-08-12T08:00:00.000Z", + }); + const session = createSessionForWorkspaceTests({ + onMessage: (message) => emitted.push(message), + projectRegistry: { + initialize: async () => {}, + existsOnDisk: async () => true, + list: async () => [project], + get: async () => project, + getOrCreateActiveByRoot: async () => project, + upsert: async () => {}, + archive: async () => {}, + remove: async () => {}, + }, + }); + + await session.handleMessage({ + type: "project.list.request", + requestId: "initial", + sync: {}, + }); + const initial = findByType(emitted, "project.list.response")?.payload; + expect(initial?.sync).toMatchObject({ mode: "snapshot", headSeq: 1 }); + + project = { ...project, displayName: "After", updatedAt: "2026-08-12T08:01:00.000Z" }; + await session.handleMessage({ + type: "project.list.request", + requestId: "catch-up", + sync: { + generation: initial?.sync?.generation, + afterSeq: initial?.sync?.headSeq, + }, + }); + const responses = filterByType(emitted, "project.list.response"); + expect(responses.at(-1)?.payload).toMatchObject({ + requestId: "catch-up", + projects: [{ projectId: "project-sequenced", projectDisplayName: "After", syncSeq: 2 }], + sync: { mode: "changes", headSeq: 2, removals: [] }, + }); +}); + test("agent updates preserve queued live transitions across stored metadata reads", async () => { const running = makeManagedAgent({ id: "agent-coherent", @@ -7754,15 +7803,17 @@ const ICON_PNG_1X1 = Buffer.from([ test("project.icon.set.request publishes a custom icon that project.icon.get serves back", async () => { const emitted: SessionOutboundMessage[] = []; + const tempDir = realpathSync(mkdtempSync(path.join(tmpdir(), "session-project-icon-test-"))); const session = asTestSession( - createSessionForWorkspaceTests({ onMessage: (message) => emitted.push(message) }), + createSessionForWorkspaceTests({ + paseoHome: path.join(tempDir, "paseo-home"), + onMessage: (message) => emitted.push(message), + }), ); session.updateClientCapabilities({ [CLIENT_CAPS.projectUpdates]: true }); - const tempDir = realpathSync(mkdtempSync(path.join(tmpdir(), "session-project-icon-test-"))); const projectRoot = path.join(tempDir, "project-without-icons"); mkdirSync(projectRoot, { recursive: true }); - session.paseoHome = path.join(tempDir, "paseo-home"); const project = createPersistedProjectRecord({ projectId: "prj_icon", diff --git a/packages/server/src/server/session/agent-updates/agent-updates-service.test.ts b/packages/server/src/server/session/agent-updates/agent-updates-service.test.ts index 29da3106f..66d58042b 100644 --- a/packages/server/src/server/session/agent-updates/agent-updates-service.test.ts +++ b/packages/server/src/server/session/agent-updates/agent-updates-service.test.ts @@ -8,6 +8,7 @@ import type { } from "../../messages.js"; import type { ManagedAgent } from "../../agent/agent-manager.js"; import type { StoredAgentRecord } from "../../agent/agent-storage.js"; +import { DirectorySyncService } from "../../directory-sync/index.js"; // No mocks — every dependency is an injected in-memory fake. The agent payloads // are supplied through the fake builders, so each test fully controls the @@ -100,6 +101,7 @@ function buildHarness() { let providerVisible: (provider: string) => boolean = () => true; let buildAgentPayloadError: Error | null = null; let enrichProjectedPayload = false; + const directorySync = new DirectorySyncService("test-generation"); const service = createAgentUpdatesService({ emit: (message) => emitted.push(message), @@ -133,6 +135,13 @@ function buildHarness() { emitWorkspaceUpdateForWorkspaceId: async (workspaceId) => { workspaceUpdates.push(workspaceId); }, + sequenceAgentUpdate: (payload, agent, project, agentId, includeSequence) => + directorySync.sequenceAgentUpdate( + payload, + agent && project ? { agent, project } : null, + agentId, + includeSequence, + ), logger: { error: (...args: unknown[]) => loggedErrors.push(args) } as unknown as pino.Logger, }); diff --git a/packages/server/src/server/session/agent-updates/agent-updates-service.ts b/packages/server/src/server/session/agent-updates/agent-updates-service.ts index ac4f3c8d7..763275a39 100644 --- a/packages/server/src/server/session/agent-updates/agent-updates-service.ts +++ b/packages/server/src/server/session/agent-updates/agent-updates-service.ts @@ -16,6 +16,7 @@ type AgentUpdatesFilter = NonNullable< interface AgentUpdatesSubscriptionState { subscriptionId: string; + syncEnabled?: boolean; filter?: AgentUpdatesFilter; isBootstrapping: boolean; pendingUpdatesByAgentId: Map; @@ -37,7 +38,11 @@ interface AgentUpdatesSubscriptionState { * stay consistent. */ export interface AgentUpdatesService { - beginSubscription(input: { subscriptionId: string; filter?: AgentUpdatesFilter }): void; + beginSubscription(input: { + subscriptionId: string; + filter?: AgentUpdatesFilter; + syncEnabled?: boolean; + }): void; flushBootstrapped( subscriptionId: string, options?: { snapshotUpdatedAtByAgentId?: Map }, @@ -57,6 +62,13 @@ export interface AgentUpdatesServiceDeps { isProviderVisibleToClient(provider: string): boolean; buildProjectPlacementForWorkspaceId(workspaceId: string): Promise; emitWorkspaceUpdateForWorkspaceId(workspaceId: string): Promise; + sequenceAgentUpdate( + payload: T, + agent: AgentSnapshotPayload | null, + project: ProjectPlacementPayload | null, + agentId: string, + includeSequence: boolean, + ): T; logger: pino.Logger; } @@ -151,6 +163,13 @@ function agentUpdateTargetId(update: AgentUpdatePayload): string { export function createAgentUpdatesService(deps: AgentUpdatesServiceDeps): AgentUpdatesService { let subscription: AgentUpdatesSubscriptionState | null = null; const liveAgentUpdateTails = new Map>(); + const sequence = ( + sub: AgentUpdatesSubscriptionState, + payload: T, + agent: AgentSnapshotPayload | null, + project: ProjectPlacementPayload | null, + agentId: string, + ) => deps.sequenceAgentUpdate(payload, agent, project, agentId, sub.syncEnabled === true); function bufferOrEmit(sub: AgentUpdatesSubscriptionState, payload: AgentUpdatePayload): void { if (payload.kind === "upsert" && !deps.isProviderVisibleToClient(payload.agent.provider)) { @@ -167,9 +186,14 @@ export function createAgentUpdatesService(deps: AgentUpdatesServiceDeps): AgentU }); } - function beginSubscription(input: { subscriptionId: string; filter?: AgentUpdatesFilter }): void { + function beginSubscription(input: { + subscriptionId: string; + filter?: AgentUpdatesFilter; + syncEnabled?: boolean; + }): void { subscription = { subscriptionId: input.subscriptionId, + syncEnabled: input.syncEnabled, filter: input.filter, isBootstrapping: true, pendingUpdatesByAgentId: new Map(), @@ -230,10 +254,10 @@ export function createAgentUpdatesService(deps: AgentUpdatesServiceDeps): AgentU ? await deps.buildProjectPlacementForWorkspaceId(payload.workspaceId) : null; if (!project) { - bufferOrEmit(sub, { - kind: "remove", - agentId: payload.id, - }); + bufferOrEmit( + sub, + sequence(sub, { kind: "remove", agentId: payload.id }, null, null, payload.id), + ); return payload; } @@ -244,16 +268,22 @@ export function createAgentUpdatesService(deps: AgentUpdatesServiceDeps): AgentU }); bufferOrEmit( sub, - matches - ? { - kind: "upsert", - agent: payload, - project, - } - : { - kind: "remove", - agentId: payload.id, - }, + sequence( + sub, + matches + ? { + kind: "upsert", + agent: payload, + project, + } + : { + kind: "remove", + agentId: payload.id, + }, + payload, + project, + payload.id, + ), ); return payload; } @@ -267,10 +297,10 @@ export function createAgentUpdatesService(deps: AgentUpdatesServiceDeps): AgentU ? await deps.buildProjectPlacementForWorkspaceId(payload.workspaceId) : null; if (!project) { - bufferOrEmit(sub, { - kind: "remove", - agentId: payload.id, - }); + bufferOrEmit( + sub, + sequence(sub, { kind: "remove", agentId: payload.id }, null, null, payload.id), + ); } else { const matches = matchesAgentUpdatesFilter({ agent: payload, @@ -279,16 +309,34 @@ export function createAgentUpdatesService(deps: AgentUpdatesServiceDeps): AgentU }); if (matches) { - bufferOrEmit(sub, { - kind: "upsert", - agent: payload, - project, - }); + bufferOrEmit( + sub, + sequence( + sub, + { + kind: "upsert", + agent: payload, + project, + }, + payload, + project, + payload.id, + ), + ); } else { - bufferOrEmit(sub, { - kind: "remove", - agentId: payload.id, - }); + bufferOrEmit( + sub, + sequence( + sub, + { + kind: "remove", + agentId: payload.id, + }, + payload, + project, + payload.id, + ), + ); } } } @@ -326,7 +374,10 @@ export function createAgentUpdatesService(deps: AgentUpdatesServiceDeps): AgentU function removeAgent(agentId: string): Promise { return enqueueAgentUpdate(agentId, () => { if (subscription) { - bufferOrEmit(subscription, { kind: "remove", agentId }); + bufferOrEmit( + subscription, + sequence(subscription, { kind: "remove", agentId }, null, null, agentId), + ); } }); } diff --git a/packages/server/src/server/session/checkout/checkout-session.test.ts b/packages/server/src/server/session/checkout/checkout-session.test.ts index 94e80cc32..d29418a9a 100644 --- a/packages/server/src/server/session/checkout/checkout-session.test.ts +++ b/packages/server/src/server/session/checkout/checkout-session.test.ts @@ -645,6 +645,16 @@ describe("CheckoutSession", () => { }, ]); }); + + it("does not emit the same checkout status twice", () => { + const { checkout, emitted } = makeCheckoutSession(); + const snapshot = createGitSnapshot("/repo", "main"); + + checkout.emitStatusUpdate("/repo", snapshot); + checkout.emitStatusUpdate("/repo", snapshot); + + expect(emitted).toHaveLength(1); + }); }); describe("switch branch", () => { diff --git a/packages/server/src/server/session/checkout/checkout-session.ts b/packages/server/src/server/session/checkout/checkout-session.ts index e349f8edd..7545b9f51 100644 --- a/packages/server/src/server/session/checkout/checkout-session.ts +++ b/packages/server/src/server/session/checkout/checkout-session.ts @@ -153,6 +153,7 @@ export class CheckoutSession { private readonly worktreesRoot: string | undefined; private readonly logger: pino.Logger; private readonly diffSubscriptions = new Map void>(); + private readonly statusUpdateFingerprints = new Map(); constructor(options: CheckoutSessionOptions) { this.host = options.host; @@ -480,20 +481,24 @@ export class CheckoutSession { emitStatusUpdate(cwd: string, snapshot: WorkspaceGitRuntimeSnapshot): void { try { const requestId = `subscription:${cwd}`; + const payload = { + ...buildCheckoutStatusPayloadFromSnapshot({ + cwd, + requestId, + snapshot, + }), + prStatus: buildCheckoutPrStatusPayloadFromSnapshot({ + cwd, + requestId, + snapshot, + }), + }; + const fingerprint = JSON.stringify(payload); + if (this.statusUpdateFingerprints.get(cwd) === fingerprint) return; + this.statusUpdateFingerprints.set(cwd, fingerprint); this.host.emit({ type: "checkout_status_update", - payload: { - ...buildCheckoutStatusPayloadFromSnapshot({ - cwd, - requestId, - snapshot, - }), - prStatus: buildCheckoutPrStatusPayloadFromSnapshot({ - cwd, - requestId, - snapshot, - }), - }, + payload, }); } catch (error) { this.logger.warn({ err: error, cwd }, "Failed to emit workspace checkout status update"); @@ -1411,6 +1416,7 @@ export class CheckoutSession { unsubscribe(); } this.diffSubscriptions.clear(); + this.statusUpdateFingerprints.clear(); } } diff --git a/packages/server/src/server/websocket-server.ts b/packages/server/src/server/websocket-server.ts index d14681b53..bc2c347a8 100644 --- a/packages/server/src/server/websocket-server.ts +++ b/packages/server/src/server/websocket-server.ts @@ -91,6 +91,7 @@ import { } from "@getpaseo/protocol/browser-automation/capabilities"; import type { BrowserToolsBroker } from "./browser-tools/broker.js"; import type { DaemonRuntimeConfig } from "./session/daemon/daemon-session.js"; +import { DirectorySyncService } from "./directory-sync/index.js"; import { APPLICATION_SOCKET_LEASE_CHECK_INTERVAL_MS, ApplicationSocketLease, @@ -577,6 +578,7 @@ export class VoiceAssistantWebSocketServer { private acceptingConnections = true; private readonly advertiseDaemonStatusRpc: boolean; private readonly advertiseRelayConfig: boolean; + private readonly directorySync = new DirectorySyncService(); constructor( server: HTTPServer, @@ -900,7 +902,11 @@ export class VoiceAssistantWebSocketServer { } public publishProjectUpdate(update: ProjectUpdate): void { - for (const session of this.listTrustedSessions()) session.emitProjectUpdate(update); + for (const session of this.listTrustedSessions()) { + void session + .emitProjectUpdate(update) + .catch((error) => this.logger.warn({ err: error }, "Failed to publish project update")); + } } public publishSpeechReadiness(readiness: SpeechReadinessSnapshot | null): void { @@ -1344,6 +1350,7 @@ export class VoiceAssistantWebSocketServer { agentStorage: this.agentStorage, projectRegistry: this.projectRegistry, workspaceRegistry: this.workspaceRegistry, + directorySync: this.directorySync, scheduleService: this.scheduleService, checkoutDiffManager: this.checkoutDiffManager, github: this.github, @@ -1529,6 +1536,8 @@ export class VoiceAssistantWebSocketServer { desktopManaged: this.daemonRuntimeConfig?.desktopManaged === true, ...(this.serverCapabilities ? { capabilities: this.serverCapabilities } : {}), features: { + // COMPAT(directorySync): added in v0.3.x, remove gate after 2027-02-12. + directorySync: true, // COMPAT(providersSnapshot): keep optional until all clients rely on snapshot flow. providersSnapshot: true, // COMPAT(providersSnapshotCwd): added in v0.3.2, remove gate after 2027-02-10. diff --git a/packages/server/src/server/wire-compat.test.ts b/packages/server/src/server/wire-compat.test.ts index 623e2be2c..89c441982 100644 --- a/packages/server/src/server/wire-compat.test.ts +++ b/packages/server/src/server/wire-compat.test.ts @@ -11,6 +11,7 @@ import { type SessionOutboundMessage, } from "@getpaseo/protocol/messages"; import { Session, type SessionOptions } from "./session.js"; +import { DirectorySyncService } from "./directory-sync/index.js"; import { createProviderSnapshotManagerStub } from "./test-utils/session-stubs.js"; import type { AgentTimelineRow } from "./agent/agent-manager.js"; import { InMemoryAgentTimelineStore } from "./agent/agent-timeline-store.js"; @@ -180,6 +181,7 @@ class InMemoryWorktreeWorkflow { function createSessionForWireCompatTest(options?: { clientCapabilities?: Record | null; + directorySync?: DirectorySyncService; messages?: SessionOutboundMessage[]; rows?: AgentTimelineRow[]; }): Session { @@ -218,6 +220,7 @@ function createSessionForWireCompatTest(options?: { projectRegistry: new EmptyProjectRegistry() as unknown as SessionOptions["projectRegistry"], workspaceRegistry: new EmptyWorkspaceRegistry() as unknown as SessionOptions["workspaceRegistry"], + directorySync: options?.directorySync, scheduleService: {} as SessionOptions["scheduleService"], checkoutDiffManager: { scheduleRefreshForCwd() {}, @@ -304,7 +307,7 @@ async function emitTimelineResponse(options?: { } describe("wire compatibility", () => { - test("sends project updates only to clients that declare support", () => { + test("sends project updates only to clients that declare support", async () => { const project = createPersistedProjectRecord({ projectId: "project-1", rootPath: "/tmp/project", @@ -322,10 +325,12 @@ describe("wire compatibility", () => { messages: capableMessages, }); - legacy.emitProjectUpdate({ kind: "upsert", project }); - legacy.emitProjectUpdate({ kind: "remove", projectId: project.projectId }); - capable.emitProjectUpdate({ kind: "upsert", project }); - capable.emitProjectUpdate({ kind: "remove", projectId: project.projectId }); + await Promise.all([ + legacy.emitProjectUpdate({ kind: "upsert", project }), + legacy.emitProjectUpdate({ kind: "remove", projectId: project.projectId }), + capable.emitProjectUpdate({ kind: "upsert", project }), + capable.emitProjectUpdate({ kind: "remove", projectId: project.projectId }), + ]); expect(legacyMessages).toEqual([]); expect(capableMessages.map((message) => SessionOutboundMessageSchema.parse(message))).toEqual([ @@ -338,6 +343,7 @@ describe("wire compatibility", () => { projectDisplayName: "Favorite project", projectCustomName: "Favorite project", projectCustomIconRevision: null, + projectIconRevision: "automatic:none:v1", projectRootPath: "/tmp/project", projectKind: "git", }, @@ -350,6 +356,50 @@ describe("wire compatibility", () => { ]); }); + test("publishes rapid project mutations in order before incremental reconciliation", async () => { + const directorySync = new DirectorySyncService("generation"); + const initial = directorySync.synchronizeProjects([], {}); + const messages: SessionOutboundMessage[] = []; + const session = createSessionForWireCompatTest({ + clientCapabilities: { [CLIENT_CAPS.projectUpdates]: true }, + directorySync, + messages, + }); + const project = createPersistedProjectRecord({ + projectId: "project-ordered", + rootPath: "/tmp/project-ordered", + kind: "git", + displayName: "Ordered project", + createdAt: "2026-07-15T00:00:00.000Z", + updatedAt: "2026-07-15T00:00:00.000Z", + }); + + await Promise.all([ + session.emitProjectUpdate({ kind: "upsert", project }), + session.emitProjectUpdate({ kind: "remove", projectId: project.projectId }), + ]); + + expect( + messages.flatMap((message) => + message.type === "project.update" ? [message.payload.kind] : [], + ), + ).toEqual(["upsert", "remove"]); + expect( + directorySync.synchronizeProjects([], { + generation: initial.sync.generation, + afterSeq: initial.sync.headSeq, + }), + ).toEqual({ + projects: [], + sync: { + generation: "generation", + mode: "changes", + headSeq: 2, + removals: [{ id: "project-ordered", seq: 2 }], + }, + }); + }); + test("downgrades reasoning_merge for clients that do not declare the capability", async () => { const response = await emitTimelineResponse(); diff --git a/packages/server/src/utils/project-custom-icon.test.ts b/packages/server/src/utils/project-custom-icon.test.ts index fa1a81ba8..dc9568d1d 100644 --- a/packages/server/src/utils/project-custom-icon.test.ts +++ b/packages/server/src/utils/project-custom-icon.test.ts @@ -10,7 +10,9 @@ import { type ProjectRegistry, } from "../server/workspace-registry.js"; import { + ProjectIconReader, readProjectIcon, + readProjectIconSnapshot, removeProjectCustomIcon, setProjectCustomIcon, } from "./project-custom-icon.js"; @@ -52,6 +54,7 @@ async function project() { return record; }, } as unknown as ProjectRegistry; + const reader = new ProjectIconReader(paseoHome); return { paseoHome, @@ -59,6 +62,9 @@ async function project() { set: (source: ProjectIconSource) => setProjectCustomIcon({ paseoHome, projectId: "project-a", source, projects }), read: () => readProjectIcon({ paseoHome, project: record }), + snapshot: () => readProjectIconSnapshot({ paseoHome, project: record }), + advertisedSnapshot: () => reader.snapshot(record), + readAdvertised: () => reader.read(record), revision: () => record.customIconRevision, remove: () => removeProjectCustomIcon({ paseoHome, projectId: "project-a" }), }; @@ -88,6 +94,40 @@ describe("project custom icon", () => { await expect(target.read()).resolves.toEqual(PNG_1X1_ICON); }); + it("fingerprints automatic icons and explicit no-icon results", async () => { + const target = await project(); + + const missing = await target.snapshot(); + expect(missing).toEqual({ icon: null, revision: "automatic:none:v1" }); + + await mkdir(join(target.rootPath, "public")); + await writeFile(join(target.rootPath, "public", "favicon.png"), PNG_1X1); + const first = await target.snapshot(); + await writeFile( + join(target.rootPath, "public", "favicon.png"), + Buffer.concat([PNG_1X1, Buffer.from([1])]), + ); + const changed = await target.snapshot(); + + expect(first.icon).toEqual(PNG_1X1_ICON); + expect(first.revision).toMatch(/^automatic:[a-f0-9]{64}$/); + expect(changed.revision).not.toBe(first.revision); + }); + + it("serves the bytes that match the advertised revision", async () => { + const target = await project(); + await mkdir(join(target.rootPath, "public")); + await writeFile(join(target.rootPath, "public", "favicon.png"), PNG_1X1); + const advertised = await target.advertisedSnapshot(); + + await writeFile( + join(target.rootPath, "public", "favicon.png"), + Buffer.concat([PNG_1X1, Buffer.from([1])]), + ); + + await expect(target.readAdvertised()).resolves.toEqual(advertised.icon); + }); + it("drops the stored image when the project goes back to automatic", async () => { const target = await project(); await target.set({ type: "upload", data: PNG_1X1.toString("base64") }); diff --git a/packages/server/src/utils/project-custom-icon.ts b/packages/server/src/utils/project-custom-icon.ts index 412ca831c..7d5aba2fa 100644 --- a/packages/server/src/utils/project-custom-icon.ts +++ b/packages/server/src/utils/project-custom-icon.ts @@ -57,12 +57,63 @@ export async function readProjectIcon(input: { paseoHome: string; project: PersistedProjectRecord; }): Promise { - if (!input.project.customIconRevision) return getProjectIcon(input.project.rootPath); - try { - return validateIcon(await readFile(cachePath(input.paseoHome, input.project.projectId))); - } catch { - return null; + return (await readProjectIconSnapshot(input)).icon; +} + +export interface ProjectIconSnapshot { + icon: ProjectIcon | null; + revision: string; +} + +/** Keeps the icon bytes served by a session aligned with its advertised revision. */ +export class ProjectIconReader { + private readonly snapshots = new Map(); + + constructor(private readonly paseoHome: string) {} + + async snapshot(project: PersistedProjectRecord): Promise { + const snapshot = await readProjectIconSnapshot({ paseoHome: this.paseoHome, project }); + this.snapshots.set(project.projectId, snapshot); + return snapshot; } + + async read(project: PersistedProjectRecord): Promise { + return (this.snapshots.get(project.projectId) ?? (await this.snapshot(project))).icon; + } +} + +/** Reads the effective icon and gives that exact result a stable identity. */ +export async function readProjectIconSnapshot(input: { + paseoHome: string; + project: PersistedProjectRecord; +}): Promise { + if (input.project.customIconRevision) { + let icon: ProjectIcon | null = null; + try { + icon = validateIcon(await readFile(cachePath(input.paseoHome, input.project.projectId))); + } catch { + // A missing/corrupt custom icon is itself a cacheable result until the + // registry revision changes. + } + return { + icon, + revision: icon + ? iconRevision("custom", icon) + : `custom:none:${input.project.customIconRevision}`, + }; + } + const icon = await getProjectIcon(input.project.rootPath); + if (!icon) return { icon: null, revision: "automatic:none:v1" }; + return { icon, revision: iconRevision("automatic", icon) }; +} + +function iconRevision(source: "automatic" | "custom", icon: ProjectIcon): string { + const fingerprint = createHash("sha256") + .update(icon.mimeType) + .update("\0") + .update(icon.data) + .digest("hex"); + return `${source}:${fingerprint}`; } export async function removeProjectCustomIcon(input: {