Restore host directories instantly before reconnect (#3259)

* feat(sync): restore host directories before reconnect

Persist complete project, workspace, and active-agent replicas so every registered host can paint immediately, including offline hosts. Reconcile those replicas through monotonic per-entity cursors on the existing directory RPCs without retaining an event journal.

* fix(sync): preserve active turn identity during catch-up

Own Agent-to-wire projection beside the existing snapshot normalizer so replica persistence does not reach through directory-sync internals. This also keeps identified active turns intact when unchanged agents are folded into an incremental directory response.

* test(e2e): inspect the real IndexedDB replica

Keep cache setup, reload, and hydration coverage on the production browser storage backend after the replica moved off localStorage. Measure writes at the actual IndexedDB put boundary as well.

* fix warm reconnect cache paths

* feat(sync): persist project icons and deepen directory sync

* fix(sync): preserve project update order

* fix(cache): preserve validated replica persistence

* fix(sync): retain cached agents during catch-up
This commit is contained in:
Mohamed Boudra
2026-08-13 23:10:44 +02:00
committed by GitHub
parent 599846650c
commit 373a98c64d
51 changed files with 2428 additions and 400 deletions
+16 -7
View File
@@ -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
+1 -1
View File
@@ -139,7 +139,7 @@ export default {
},
plugins: [
"expo-router",
[withAndroidAsyncStorageSize, 10],
[withAndroidAsyncStorageSize, 64],
...buildProfile.cameraPlugins,
[
"expo-splash-screen",
@@ -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<void>
async function rewriteCachedMessageAsLegacyRow(page: Page, prompt: string): Promise<void> {
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<Record<string, unknown>> } | 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<Record<string, unknown>> } | 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<void> {
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<Record<string, unknown>> } | 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<void> {
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<Record<string, unknown>> } | 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();
}
@@ -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<void> {
await page.addInitScript((storageKey) => {
const originalSetItem = Storage.prototype.setItem;
window.__replicaCachePerf = {
lastValue: localStorage.getItem(storageKey),
resetAt: performance.now(),
redundantWrites: 0,
serializedChars: 0,
storageDurationMs: 0,
writes: 0,
};
Storage.prototype.setItem = function measuredSetItem(key: string, value: string): void {
const startedAt = performance.now();
originalSetItem.call(this, key, value);
if (key !== storageKey) return;
const state = window.__replicaCachePerf;
if (!state) return;
if (state.lastValue === value) state.redundantWrites += 1;
state.lastValue = value;
state.writes += 1;
state.serializedChars += value.length;
state.storageDurationMs += performance.now() - startedAt;
};
}, REPLICA_CACHE_STORAGE_KEY);
await observeReplicaCacheStorageWrites(page);
await page.addInitScript(() => {
window.__replicaCachePerf = { resetAt: performance.now() };
});
}
export async function beginReplicaCacheMeasurement(page: Page): Promise<void> {
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<ReplicaCachePerfReport> {
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 };
}
@@ -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<Record<string, unknown>>;
} | 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<string | null> {
const database = await new Promise<IDBDatabase>((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<ReplicaCacheRecord | null> {
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<void> {
await page.evaluate(
async ({ databaseName, storeName, storageKey, raw }) => {
const database = await new Promise<IDBDatabase>((resolve, reject) => {
const request = indexedDB.open(databaseName, 1);
request.addEventListener("success", () => resolve(request.result));
request.addEventListener("error", () => reject(request.error));
});
await new Promise<void>((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<void> {
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<void> {
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<ReplicaCacheWriteObserverState> {
return page.evaluate(() => {
const state = window.__replicaCacheWriteObserver;
if (!state) throw new Error("Replica cache write observer is not installed");
return state;
});
}
@@ -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<void> {
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();
@@ -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" },
]);
});
});
@@ -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();
},
@@ -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(
() =>
@@ -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<string, string>();
async getItem(key: string): Promise<string | null> {
return this.values.get(key) ?? null;
}
async setItem(key: string, value: string): Promise<void> {
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);
});
});
+217
View File
@@ -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<string | null>;
setItem(key: string, value: string): Promise<void>;
}
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<string, unknown>;
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<string, unknown>;
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<string, StoredIcon>();
private activeServerIds = new Set<string>();
private persistTimer: ReturnType<typeof setTimeout> | null = null;
private writeQueue: Promise<void> = Promise.resolve();
constructor(private readonly storage: ProjectIconCacheStorage) {}
setHosts(serverIds: Iterable<string>): 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<void> {
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<void> {
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<ProjectIconTarget, "projectId" | "iconWorkingDir">,
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 };
}
+4
View File
@@ -20,4 +20,8 @@ describe("project icon lookup", () => {
cwd: "/projects/paseo",
});
});
it("waits while custom-icon capability is unknown", () => {
expect(resolveProjectIconLookup(target, null)).toBeNull();
});
});
+12 -34
View File
@@ -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<ProjectIconTarget, "projectId" | "iconWorkingDir">,
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<string, ProjectIconTarget>();
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,
};
}),
});
@@ -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) {
@@ -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();
@@ -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<string, Agent> {
const previous = useSessionStore.getState().sessions[this.serverId]?.agents ?? new Map();
const merged = new Map<string, FetchAgentsEntry>();
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) => {
@@ -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<ReturnType<DaemonClient["fetchWorkspaces"]>>;
@@ -9,8 +9,11 @@ type ProjectListResult = Awaited<ReturnType<DaemonClient["listProjects"]>>;
class FakeDirectoryClient {
fetchAgentsCalls = 0;
lastAgentOptions: unknown;
fetchWorkspacesCalls = 0;
listProjectsCalls = 0;
lastProjectOptions: unknown;
projectResult: ProjectListResult | null = null;
private pendingWorkspaceFetch: Promise<WorkspaceFetchResult> | null = null;
private readonly handlers = new Map<
SessionOutboundMessage["type"],
@@ -42,8 +45,9 @@ class FakeDirectoryClient {
return complete;
}
async fetchAgents(): Promise<Awaited<ReturnType<DaemonClient["fetchAgents"]>>> {
async fetchAgents(options?: unknown): Promise<Awaited<ReturnType<DaemonClient["fetchAgents"]>>> {
this.fetchAgentsCalls += 1;
this.lastAgentOptions = options;
return {
requestId: "agents",
entries: [],
@@ -66,8 +70,10 @@ class FakeDirectoryClient {
};
}
async listProjects(): Promise<ProjectListResult> {
async listProjects(options?: unknown): Promise<ProjectListResult> {
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);
+196 -29
View File
@@ -36,10 +36,41 @@ const AGENT_SORT: NonNullable<FetchAgentsOptions["sort"]> = [
{ 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<void> {
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<WorkspaceDirectorySnapshot, WorkspaceDirectoryDelta>,
initialSubscribe: boolean,
supportsDirectorySync: boolean,
): Promise<void> {
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<string, Agent> {
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<WorkspaceDirectorySnapshot, WorkspaceDirectoryDelta>,
supportsDirectorySync: boolean,
): Promise<void> {
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<WorkspaceDirectorySnapshot, WorkspaceDirectoryDelta>,
): 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<ReturnType<DaemonClient["fetchAgents"]>>["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<string, unknown>)[entity];
if (!cursor || typeof cursor !== "object" || Array.isArray(cursor)) continue;
const { generation, afterSeq } = cursor as Record<string, unknown>;
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 &&
@@ -21,6 +21,9 @@ type ProjectDirectoryDelta = Extract<SessionOutboundMessage, { type: "project.up
export interface WorkspaceDirectorySnapshot {
workspaces: Map<string, WorkspaceDescriptor>;
projects: Map<string, ProjectDescriptor>;
syncCursors?: Partial<
Record<"projects" | "workspaces", { generation: string; afterSeq: number }>
>;
}
function applyProjectDelta(
+19
View File
@@ -51,3 +51,22 @@ export function useHostFeatureMap(
[flags, serverIds],
);
}
export function useHostFeatureAvailabilityMap(
serverIds: readonly string[],
feature: HostFeatureName,
): ReadonlyMap<string, boolean | null> {
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],
);
}
+28 -30
View File
@@ -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);
@@ -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 });
+90 -28
View File
@@ -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<typeof StoredWorkspaceSchema>;
type StoredProject = z.infer<typeof StoredProjectSchema>;
interface ReplicaInput {
agent: Agent | undefined;
workspace: WorkspaceDescriptor | undefined;
project: ProjectDescriptor | undefined;
agents: ReadonlyMap<string, Agent>;
workspaces: ReadonlyMap<string, WorkspaceDescriptor>;
projects: ReadonlyMap<string, ProjectDescriptor>;
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<void> {
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;
}
@@ -0,0 +1,4 @@
import AsyncStorage from "@react-native-async-storage/async-storage";
import type { ReplicaCacheStorage } from "./index";
export const replicaCacheStorage: ReplicaCacheStorage = AsyncStorage;
@@ -0,0 +1,70 @@
import type { ReplicaCacheStorage } from "./index";
const DATABASE_NAME = "paseo-replica-cache";
const STORE_NAME = "key-value";
let databasePromise: Promise<IDBDatabase> | null = null;
function openDatabase(): Promise<IDBDatabase> {
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<string | null> {
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<void> {
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<void> {
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 };
@@ -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,
@@ -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",
@@ -16,6 +16,7 @@ export interface SessionsSnapshot {
string,
{
hasHydratedWorkspaces?: boolean;
hasWorkspaceDirectorySnapshot?: boolean;
workspaces: Map<string, WorkspaceDescriptor>;
projects?: Map<string, ProjectDescriptor>;
}
@@ -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[],
+5
View File
@@ -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<SessionStore>()(
workspaceAgentActivity: buildWorkspaceAgentActivityIndex(replica.agents),
workspaces: replica.workspaces,
projects: replica.projects,
hasWorkspaceDirectorySnapshot: true,
agentStreamTail,
agentTasks,
},
+13 -1
View File
@@ -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<Omit<AgentSnapshotPayload, "labels">> & {
@@ -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(
+43
View File
@@ -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<AgentSnapshotPayload, "activeTurn"> {
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);
@@ -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,
};
}
+9 -1
View File
@@ -677,6 +677,10 @@ export type ProjectListPayload = Extract<
SessionOutboundMessage,
{ type: "project.list.response" }
>["payload"];
type ProjectListRequest = Extract<SessionInboundMessage, { type: "project.list.request" }>;
export type ProjectListOptions = Omit<ProjectListRequest, "type" | "requestId"> & {
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<ProjectListPayload> {
async listProjects(options?: string | ProjectListOptions): Promise<ProjectListPayload> {
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,
+23
View File
@@ -5,6 +5,7 @@ import {
parseServerInfoStatusPayload,
SessionInboundMessageSchema,
SessionOutboundMessageSchema,
WorkspaceProjectDescriptorPayloadSchema,
} from "./messages.js";
function workspaceDescriptor(overrides: Record<string, unknown> = {}) {
@@ -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(
+61 -2
View File
@@ -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(),
}),
});
@@ -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",
@@ -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,
});
});
});
@@ -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<SessionOutboundMessage, { type: "project.update" }>;
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<WorkspaceProjectDescriptorPayload>({
getId: (project) => project.projectId,
});
private readonly workspaces = new VersionedCollection<WorkspaceDescriptorPayload>({
getId: (workspace) => workspace.id,
});
private readonly agents = new VersionedCollection<AgentDirectoryEntry>({
getId: (entry) => entry.agent.id,
});
constructor(generation = randomUUID()) {
this.generation = generation;
}
synchronizeProjects(
snapshot: Iterable<WorkspaceProjectDescriptorPayload>,
cursor: DirectorySyncCursor,
): Pick<ProjectListResponse, "projects" | "sync"> {
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<WorkspaceDescriptorPayload>,
cursor: DirectorySyncCursor,
): Pick<FetchWorkspacesResponse, "entries" | "emptyProjects" | "pageInfo" | "sync"> {
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<AgentDirectoryEntry>,
cursor: DirectorySyncCursor,
): Pick<FetchAgentsResponse, "entries" | "pageInfo" | "sync"> {
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<T extends object>(
payload: T,
value: WorkspaceDescriptorPayload | null,
id: string,
includeSequence: boolean,
): T & Partial<DirectoryVersion> {
if (value) {
this.workspaces.replace(value);
} else {
this.workspaces.remove(id);
}
return this.withVersion(payload, includeSequence, this.version(this.workspaces, id));
}
sequenceAgentUpdate<T extends object>(
payload: T,
value: AgentDirectoryEntry | null,
id: string,
includeSequence: boolean,
): T & Partial<DirectoryVersion> {
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<unknown>) {
return {
generation: this.generation,
headSeq: read.headSeq,
mode: read.mode,
...(read.reason ? { reason: read.reason } : {}),
removals: read.removals,
};
}
private withVersion<T extends object>(
payload: T,
includeSequence: boolean,
version: DirectoryVersion | null,
): T & Partial<DirectoryVersion> {
return version && includeSequence ? { ...payload, ...version } : payload;
}
private read<T>(
collection: VersionedCollection<T>,
cursor: DirectorySyncCursor,
): CollectionRead<T> {
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<T>(collection: VersionedCollection<T>, id: string): DirectoryVersion | null {
const seq = collection.sequenceFor(id);
return seq === null ? null : { generation: this.generation, seq };
}
}
@@ -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: [],
});
});
});
@@ -0,0 +1,141 @@
export interface VersionedValue<T> {
seq: number;
value: T;
}
export interface VersionedRemoval {
id: string;
seq: number;
}
export interface CollectionRead<T> {
mode: "changes" | "snapshot";
reason?: "no_cursor" | "generation_changed" | "cursor_expired";
headSeq: number;
values: VersionedValue<T>[];
removals: VersionedRemoval[];
}
export type PublishedChange<T> =
| { kind: "upsert"; seq: number; value: T }
| { kind: "remove"; seq: number; id: string };
interface StoredValue<T> extends VersionedValue<T> {
fingerprint: string;
}
interface VersionedCollectionOptions<T> {
getId(value: T): string;
fingerprint?(value: T): string;
maxTombstones?: number;
}
/** Owns latest-state sequencing for one directory entity type. */
export class VersionedCollection<T> {
private readonly valuesById = new Map<string, StoredValue<T>>();
private readonly tombstonesById = new Map<string, number>();
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<T>) {
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<PublishedChange<T>, { 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<PublishedChange<T>, { 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<T>): void {
const retainedIds = new Set<string>();
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<T> {
return {
mode: "snapshot",
reason,
headSeq: this.headSeq,
values: this.sortedValues(),
removals: [],
};
}
readAfter(afterSeq: number): CollectionRead<T> {
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<T>[] {
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);
}
}
}
+4 -2
View File
@@ -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",
},
+119 -29
View File
@@ -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<string, WorkspaceUpdatePayload>;
@@ -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<void> = Promise.resolve();
private projectUpdateQueue: Promise<void> = Promise.resolve();
private isCleanedUp = false;
private viewedTimelineAgentIds = new Set<string>();
private readonly viewedTimelineAgentIdsBySource = new Map<object, Set<string>>();
@@ -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<string, Promise<void>>();
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<void> {
const queued = this.projectUpdateQueue
.catch(() => undefined)
.then(() => this.publishProjectUpdate(update));
this.projectUpdateQueue = queued;
return queued;
}
private async publishProjectUpdate(update: ProjectUpdate): Promise<void> {
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<WorkspaceProjectDescriptorPayload> {
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<string, number>();
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<void> {
private async handleProjectListRequest(
request: Extract<SessionInboundMessage, { type: "project.list.request" }>,
): Promise<void> {
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<SessionInboundMessage, { type: "fetch_agents_request" }>,
) {
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<SessionInboundMessage, { type: "fetch_workspaces_request" }>,
) {
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,
},
});
@@ -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",
@@ -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,
});
@@ -16,6 +16,7 @@ type AgentUpdatesFilter = NonNullable<
interface AgentUpdatesSubscriptionState {
subscriptionId: string;
syncEnabled?: boolean;
filter?: AgentUpdatesFilter;
isBootstrapping: boolean;
pendingUpdatesByAgentId: Map<string, AgentUpdatePayload>;
@@ -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<string, number> },
@@ -57,6 +62,13 @@ export interface AgentUpdatesServiceDeps {
isProviderVisibleToClient(provider: string): boolean;
buildProjectPlacementForWorkspaceId(workspaceId: string): Promise<ProjectPlacementPayload | null>;
emitWorkspaceUpdateForWorkspaceId(workspaceId: string): Promise<void>;
sequenceAgentUpdate<T extends AgentUpdatePayload>(
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<string, Promise<void>>();
const sequence = <T extends AgentUpdatePayload>(
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<void> {
return enqueueAgentUpdate(agentId, () => {
if (subscription) {
bufferOrEmit(subscription, { kind: "remove", agentId });
bufferOrEmit(
subscription,
sequence(subscription, { kind: "remove", agentId }, null, null, agentId),
);
}
});
}
@@ -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", () => {
@@ -153,6 +153,7 @@ export class CheckoutSession {
private readonly worktreesRoot: string | undefined;
private readonly logger: pino.Logger;
private readonly diffSubscriptions = new Map<string, () => void>();
private readonly statusUpdateFingerprints = new Map<string, string>();
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();
}
}
+10 -1
View File
@@ -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.
+55 -5
View File
@@ -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<string, unknown> | 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();
@@ -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") });
@@ -57,12 +57,63 @@ export async function readProjectIcon(input: {
paseoHome: string;
project: PersistedProjectRecord;
}): Promise<ProjectIcon | null> {
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<string, ProjectIconSnapshot>();
constructor(private readonly paseoHome: string) {}
async snapshot(project: PersistedProjectRecord): Promise<ProjectIconSnapshot> {
const snapshot = await readProjectIconSnapshot({ paseoHome: this.paseoHome, project });
this.snapshots.set(project.projectId, snapshot);
return snapshot;
}
async read(project: PersistedProjectRecord): Promise<ProjectIcon | null> {
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<ProjectIconSnapshot> {
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: {