diff --git a/docs/architecture.md b/docs/architecture.md index b4ab8ef21..ce2189853 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -99,12 +99,20 @@ 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. - `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. + ### `packages/cli` — Command-line client Commander.js CLI with Docker-style commands. Common agent operations are also exposed at the top level (e.g. `paseo ls`, `paseo run`). diff --git a/packages/app/src/hooks/use-agent-screen-state-machine.test.ts b/packages/app/src/hooks/use-agent-screen-state-machine.test.ts index 5422bf63f..9b9b3e4f2 100644 --- a/packages/app/src/hooks/use-agent-screen-state-machine.test.ts +++ b/packages/app/src/hooks/use-agent-screen-state-machine.test.ts @@ -190,6 +190,25 @@ describe("deriveAgentScreenViewState", () => { expect(sync.ui).toBe("silent"); }); + it("keeps hydrated history visible while reconnect revalidation and visibility catch-up overlap", () => { + const memory = createBaseMemory({ + hasRenderedReady: true, + lastReadyAgent: createAgent("agent-1"), + }); + const input: AgentScreenMachineInput = { + ...createBaseInput(), + hasHydratedHistoryBefore: true, + needsAuthoritativeSync: true, + visibilityCatchUpStatus: "pending", + }; + + const result = deriveAgentScreenViewState({ input, memory }); + const ready = expectReadyState(result.state); + const sync = expectCatchingUpSync(ready); + + expect(sync.ui).toBe("silent"); + }); + it("covers already-hydrated history while a newly visible agent catches up", () => { const memory = createBaseMemory({ hasRenderedReady: true, diff --git a/packages/app/src/hooks/use-agent-screen-state-machine.ts b/packages/app/src/hooks/use-agent-screen-state-machine.ts index 941e87deb..daf8562f0 100644 --- a/packages/app/src/hooks/use-agent-screen-state-machine.ts +++ b/packages/app/src/hooks/use-agent-screen-state-machine.ts @@ -150,9 +150,11 @@ function resolveCatchingUpUi(args: { hasOptimisticCreateContinuity: boolean; isVisibilityCatchUpPending: boolean; hasHydratedHistoryBefore: boolean; + needsAuthoritativeSync: boolean; hadInitialSyncFailure: boolean; }): "overlay" | "silent" { if (args.hasOptimisticCreateContinuity) return "silent"; + if (args.hasHydratedHistoryBefore && args.needsAuthoritativeSync) return "silent"; if (args.isVisibilityCatchUpPending) return "overlay"; if (args.hasHydratedHistoryBefore) return "silent"; if (args.hadInitialSyncFailure) return "silent"; @@ -184,6 +186,7 @@ function resolveAgentScreenSync(args: { hasOptimisticCreateContinuity: hasOptimisticCreateContinuity(input), isVisibilityCatchUpPending: input.visibilityCatchUpStatus === "pending", hasHydratedHistoryBefore: input.hasHydratedHistoryBefore, + needsAuthoritativeSync: input.needsAuthoritativeSync, hadInitialSyncFailure, }), }; diff --git a/packages/app/src/panels/agent-panel.tsx b/packages/app/src/panels/agent-panel.tsx index 7613dd035..990527a7b 100644 --- a/packages/app/src/panels/agent-panel.tsx +++ b/packages/app/src/panels/agent-panel.tsx @@ -498,6 +498,13 @@ function AgentPanelContent({ const runtimeIsConnected = useHostRuntimeIsConnected(runtimeServerId); const runtimeConnectionStatus = useHostRuntimeConnectionStatus(runtimeServerId); const runtimeLastError = useHostRuntimeLastError(runtimeServerId); + const hasCachedAgent = useSessionStore((state) => { + if (!resolvedServerId || !resolvedAgentId) return false; + const session = state.sessions[resolvedServerId]; + return Boolean( + session?.agents.has(resolvedAgentId) || session?.agentDetails.has(resolvedAgentId), + ); + }); const connectionServerId = resolvedServerId ?? null; const daemon = connectionServerId @@ -512,7 +519,7 @@ function AgentPanelContent({ : runtimeConnectionStatus; const lastConnectionError = runtimeLastError; - if (!resolvedServerId || !runtimeClient) { + if (!resolvedServerId || (!runtimeClient && !hasCachedAgent)) { return ( >; + client: ReturnType; isConnected: boolean; connectionStatus: HostRuntimeConnectionStatus; onOpenWorkspaceFile?: (request: WorkspaceFileOpenRequest) => void; @@ -593,7 +600,7 @@ function AgentPanelBody({ } return; } - if (!isConnected || !hasSession) { + if (!client || !isConnected || !hasSession) { return; } if (lookupState.tag === "loading" || lookupState.tag === "not_found") { @@ -710,7 +717,7 @@ function ChatAgentContent({ serverId: string; agentId?: string; isPaneFocused: boolean; - client: NonNullable>; + client: ReturnType; isConnected: boolean; connectionStatus: HostRuntimeConnectionStatus; onOpenWorkspaceFile?: (request: WorkspaceFileOpenRequest) => void; @@ -949,7 +956,7 @@ function ChatAgentContent({ } return; } - if (!isPaneVisible || !isConnected || !hasSession) { + if (!client || !isPaneVisible || !isConnected || !hasSession) { return; } if ( diff --git a/packages/app/src/runtime/host-runtime.test.ts b/packages/app/src/runtime/host-runtime.test.ts index 4de541e34..156f554dc 100644 --- a/packages/app/src/runtime/host-runtime.test.ts +++ b/packages/app/src/runtime/host-runtime.test.ts @@ -20,6 +20,7 @@ import { type HostRuntimeControllerDeps, type HostRuntimeStorage, } from "./host-runtime"; +import { ReplicaCache } from "./replica-cache"; class FakeDaemonClient { private state: ConnectionState = { status: "idle" }; @@ -1364,6 +1365,74 @@ describe("HostRuntimeController", () => { }); describe("HostRuntimeStore", () => { + it("restores the display replica before declaring the host registry loaded", async () => { + const host = makeHost(); + const storage = createMemoryHostRuntimeStorage(); + await storage.setItem("@paseo:daemon-registry", JSON.stringify([host])); + await storage.setItem("@paseo:e2e", "1"); + + const cachedAgent = replicaAgent( + makeFetchAgentsEntry({ + id: "cached-agent", + cwd: "/repo/paseo", + updatedAt: "2026-07-18T08:00:00.000Z", + title: "Cached agent", + }).agent, + host.serverId, + ); + const session = useSessionStore.getState(); + session.initializeSession(host.serverId, null); + session.setAgents(host.serverId, new Map([[cachedAgent.id, cachedAgent]])); + session.setFocusedAgentId(host.serverId, cachedAgent.id); + session.setAgentStreamTail( + host.serverId, + new Map([ + [ + cachedAgent.id, + [ + { + kind: "assistant_message", + id: "cached-message", + text: "Already painted", + timestamp: new Date("2026-07-18T08:01:00.000Z"), + }, + ], + ], + ]), + ); + const cache = new ReplicaCache(storage); + cache.setHosts([host.serverId]); + await cache.flush(); + session.clearSession(host.serverId); + + const store = new HostRuntimeStore({ + storage, + deps: { + createClient: () => { + throw new Error("createClient should not be called"); + }, + connectToDaemon: async () => { + throw new Error("connectToDaemon should not be called"); + }, + getClientId: async () => "cid_test_runtime", + }, + }); + const registryLoaded = onceHostListMatches(store, () => store.isHostRegistryLoaded()); + store.boot(); + await registryLoaded; + + const restored = useSessionStore.getState().sessions[host.serverId]; + expect(restored?.agents.get(cachedAgent.id)?.title).toBe("Cached agent"); + expect(restored?.agentStreamTail.get(cachedAgent.id)?.[0]).toMatchObject({ + text: "Already painted", + timestamp: new Date("2026-07-18T08:01:00.000Z"), + }); + expect(restored?.hasHydratedAgents).toBe(false); + + store.syncHosts([]); + session.clearSession(host.serverId); + }); + it("marks the host registry loaded after boot reads storage", async () => { const previousOverride = process.env.EXPO_PUBLIC_LOCAL_DAEMON; process.env.EXPO_PUBLIC_LOCAL_DAEMON = "not-an-endpoint"; diff --git a/packages/app/src/runtime/host-runtime.ts b/packages/app/src/runtime/host-runtime.ts index 85f4634ec..59fdf9c09 100644 --- a/packages/app/src/runtime/host-runtime.ts +++ b/packages/app/src/runtime/host-runtime.ts @@ -55,6 +55,7 @@ import { } from "@/composer/attachments/submit"; import { encodeImages } from "@/utils/encode-images"; import { DirectorySync, type RefreshAgentDirectoryResult } from "@/runtime/directory-sync"; +import { ReplicaCache } from "@/runtime/replica-cache"; export type HostRuntimeConnectionStatus = "idle" | "connecting" | "online" | "offline" | "error"; export type HostRegistryStatus = "loading" | "ready"; @@ -1355,10 +1356,12 @@ export class HostRuntimeStore { private configuredOverrideBootstrapInFlight: Promise | null = null; private bootStarted = false; private storage: HostRuntimeStorage; + private replicaCache: ReplicaCache; constructor(input?: { deps?: HostRuntimeControllerDeps; storage?: HostRuntimeStorage }) { this.deps = input?.deps ?? createDefaultDeps(); this.storage = input?.storage ?? AsyncStorage; + this.replicaCache = new ReplicaCache(this.storage); } // --- Host registry --- @@ -1432,27 +1435,29 @@ export class HostRuntimeStore { private async loadFromStorage(): Promise { let shouldPersistHosts = false; + let profiles: HostProfile[] = []; try { const stored = await this.storage.getItem(REGISTRY_STORAGE_KEY); - if (!stored) { - return; + if (stored) { + const parsed = JSON.parse(stored) as unknown; + if (Array.isArray(parsed)) { + const normalizedProfiles = parsed + .map((entry) => normalizeStoredHostProfile(entry)) + .filter((entry): entry is HostProfile => entry !== null); + profiles = normalizedProfiles.filter((entry) => !isPlaceholderServerId(entry.serverId)); + if (profiles.length !== normalizedProfiles.length) { + shouldPersistHosts = true; + } + } } - const parsed = JSON.parse(stored) as unknown; - if (!Array.isArray(parsed)) { - return; - } - const normalizedProfiles = parsed - .map((entry) => normalizeStoredHostProfile(entry)) - .filter((entry): entry is HostProfile => entry !== null); - const profiles = normalizedProfiles.filter((entry) => !isPlaceholderServerId(entry.serverId)); this.hosts = profiles; + this.replicaCache.setHosts(profiles.map((profile) => profile.serverId)); + await this.replicaCache.restore(); this.syncHosts(profiles); - if (profiles.length !== normalizedProfiles.length) { - shouldPersistHosts = true; - } } catch (error) { console.error("[HostRuntime] Failed to load host registry from storage", error); } finally { + this.replicaCache.start(); this.hostRegistryStatus = "ready"; this.emitHostList(); if (shouldPersistHosts) { @@ -1582,6 +1587,7 @@ export class HostRuntimeStore { rekeyMap(this.lastConnectionStatusByServer, oldServerId, newServerId); rekeyMap(this.agentDirectoryBootstrapInFlight, oldServerId, newServerId); + this.replicaCache.reconcileServerId(oldServerId, newServerId); this.directorySyncByServer.get(oldServerId)?.dispose(); this.directorySyncByServer.delete(oldServerId); const directory = new DirectorySync(newServerId, { @@ -1885,6 +1891,7 @@ export class HostRuntimeStore { >; }, ): void { + this.replicaCache.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)) { diff --git a/packages/app/src/runtime/replica-cache/index.test.ts b/packages/app/src/runtime/replica-cache/index.test.ts new file mode 100644 index 000000000..69800f03a --- /dev/null +++ b/packages/app/src/runtime/replica-cache/index.test.ts @@ -0,0 +1,256 @@ +import { afterEach, describe, expect, it } from "vitest"; +import type { WorkspaceDescriptorPayload } from "@getpaseo/protocol/messages"; +import { normalizeAgentSnapshot } from "@/utils/agent-snapshots"; +import { + normalizeEmptyProjectDescriptor, + normalizeWorkspaceDescriptor, + useSessionStore, +} from "@/stores/session-store"; +import type { StreamItem } from "@/types/stream"; +import { ReplicaCache, type ReplicaCacheStorage } from "."; + +const SERVER_ID = "cached-host"; +const LRU_SERVER_IDS = ["host-a", "host-b", "host-c"] as const; + +class MemoryStorage implements ReplicaCacheStorage { + readonly values = new Map(); + + async getItem(key: string): Promise { + return this.values.get(key) ?? null; + } + + async setItem(key: string, value: string): Promise { + this.values.set(key, value); + } +} + +function workspace( + id = "workspace-1", + projectId = "project-1", + workspaceDirectory = "/repo/paseo", +): WorkspaceDescriptorPayload { + return { + id, + projectId, + projectDisplayName: "Paseo", + projectRootPath: workspaceDirectory, + workspaceDirectory, + projectKind: "git", + workspaceKind: "local_checkout", + name: "main", + status: "running", + statusEnteredAt: "2026-07-18T08:00:00.000Z", + activityAt: null, + archivingAt: null, + diffStat: null, + scripts: [], + }; +} + +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, + }, + currentModeId: null, + availableModes: [], + pendingPermissions: [], + persistence: null, + title: `Agent ${id}`, + labels: {}, + }, + SERVER_ID, + ); +} + +function message(id: string, text: string): StreamItem { + return { + kind: "assistant_message", + id, + text, + timestamp: new Date("2026-07-18T08:02:00.000Z"), + timelineCursor: { epoch: "epoch-1", seq: 12 }, + }; +} + +function seedSession(): void { + const store = useSessionStore.getState(); + store.initializeSession(SERVER_ID, null); + store.setAgents(SERVER_ID, new Map([["agent-1", agent("agent-1")]])); + store.setWorkspaces( + SERVER_ID, + new Map([["workspace-1", normalizeWorkspaceDescriptor(workspace())]]), + ); + store.setEmptyProjects(SERVER_ID, [ + normalizeEmptyProjectDescriptor({ + projectId: "empty-project", + projectDisplayName: "Empty project", + projectRootPath: "/repo/empty", + projectKind: "directory", + }), + ]); + store.setFocusedAgentId(SERVER_ID, "agent-1"); + store.setAgentStreamTail(SERVER_ID, new Map([["agent-1", [message("message-1", "Cached")]]])); + store.setAgentTimelineCursor( + SERVER_ID, + new Map([["agent-1", { epoch: "epoch-1", startSeq: 1, endSeq: 12 }]]), + ); + store.setAgentTimelineHasOlder(SERVER_ID, new Map([["agent-1", true]])); + store.setAgentAuthoritativeHistoryApplied(SERVER_ID, "agent-1", true); +} + +function seedTimeline(serverId: string, text: string): void { + const agentId = `agent-${serverId}`; + const workspaceId = `workspace-${serverId}`; + const workspaceDirectory = `/repo/${serverId}`; + const store = useSessionStore.getState(); + store.initializeSession(serverId, null); + store.setAgents(serverId, new Map([[agentId, agent(agentId, workspaceId, workspaceDirectory)]])); + store.setWorkspaces( + serverId, + new Map([ + [ + workspaceId, + normalizeWorkspaceDescriptor( + workspace(workspaceId, `project-${serverId}`, workspaceDirectory), + ), + ], + ]), + ); + store.setFocusedAgentId(serverId, agentId); + store.setAgentStreamTail(serverId, new Map([[agentId, [message(`message-${serverId}`, text)]]])); +} + +afterEach(() => { + const store = useSessionStore.getState(); + store.clearSession(SERVER_ID); + for (const serverId of LRU_SERVER_IDS) store.clearSession(serverId); +}); + +describe("ReplicaCache", () => { + it("restores a displayable stale replica without claiming remote hydration", async () => { + const storage = new MemoryStorage(); + const writer = new ReplicaCache(storage); + writer.setHosts([SERVER_ID]); + seedSession(); + await writer.flush(); + + useSessionStore.getState().clearSession(SERVER_ID); + + const reader = new ReplicaCache(storage); + reader.setHosts([SERVER_ID]); + await reader.restore(); + + const session = useSessionStore.getState().sessions[SERVER_ID]; + expect(session?.client).toBeNull(); + expect(session?.hasHydratedAgents).toBe(false); + expect(session?.hasHydratedWorkspaces).toBe(false); + expect(Array.from(session?.agents.keys() ?? [])).toEqual(["agent-1"]); + expect(Array.from(session?.workspaces.keys() ?? [])).toEqual(["workspace-1"]); + expect(Array.from(session?.emptyProjects.keys() ?? [])).toEqual([]); + expect(session?.agents.get("agent-1")?.updatedAt).toBeInstanceOf(Date); + expect(session?.workspaces.get("workspace-1")?.statusEnteredAt).toBeInstanceOf(Date); + expect(session?.agentStreamTail.get("agent-1")).toEqual([message("message-1", "Cached")]); + expect(session?.agentAuthoritativeHistoryApplied.get("agent-1")).toBe(true); + expect(session?.agentTimelineCursor.get("agent-1")).toEqual({ + epoch: "epoch-1", + startSeq: 1, + endSeq: 12, + }); + expect(session?.agentTimelineHasOlder.get("agent-1")).toBe(true); + }); + + it("persists only the focused agent view with a short timeline tail", async () => { + const storage = new MemoryStorage(); + const cache = new ReplicaCache(storage); + cache.setHosts([SERVER_ID]); + seedSession(); + + const store = useSessionStore.getState(); + store.setAgents(SERVER_ID, (agents) => + new Map(agents).set("agent-2", agent("agent-2", "workspace-2", "/repo/other")), + ); + store.setWorkspaces(SERVER_ID, (workspaces) => + new Map(workspaces).set( + "workspace-2", + normalizeWorkspaceDescriptor(workspace("workspace-2", "project-2", "/repo/other")), + ), + ); + const secondTimeline = Array.from({ length: 60 }, (_, index) => + message(`message-${index}`, `Second ${index}`), + ); + store.setAgentStreamTail( + SERVER_ID, + new Map([ + ["agent-1", [message("message-1", "First")]], + ["agent-2", secondTimeline], + ]), + ); + store.setFocusedAgentId(SERVER_ID, "agent-2"); + await cache.flush(); + + store.clearSession(SERVER_ID); + const reader = new ReplicaCache(storage); + reader.setHosts([SERVER_ID]); + await reader.restore(); + + 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?.emptyProjects.keys() ?? [])).toEqual([]); + expect(Array.from(timelines?.keys() ?? [])).toEqual(["agent-2"]); + expect(timelines?.get("agent-2")).toEqual(secondTimeline.slice(-50)); + }); + + 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 }); + cache.setHosts(LRU_SERVER_IDS.slice(0, 2)); + seedTimeline("host-a", "A".repeat(1_200)); + seedTimeline("host-b", "B".repeat(1_200)); + await cache.flush(); + + seedTimeline("host-a", "A".repeat(1_201)); + await cache.flush(); + + cache.setHosts(LRU_SERVER_IDS); + seedTimeline("host-c", "C".repeat(1_200)); + await cache.flush(); + + for (const serverId of LRU_SERVER_IDS) { + useSessionStore.getState().clearSession(serverId); + } + const reader = new ReplicaCache(storage, { maxBytes: 7_000 }); + reader.setHosts(LRU_SERVER_IDS); + await reader.restore(); + + expect(Object.keys(useSessionStore.getState().sessions).sort()).toEqual(["host-a", "host-c"]); + }); + + it("drops malformed or unknown cache versions", async () => { + const storage = new MemoryStorage(); + storage.values.set("@paseo:replica-cache", JSON.stringify({ version: 999, hosts: [] })); + const cache = new ReplicaCache(storage); + cache.setHosts([SERVER_ID]); + + await cache.restore(); + + expect(useSessionStore.getState().sessions[SERVER_ID]).toBeUndefined(); + }); +}); diff --git a/packages/app/src/runtime/replica-cache/index.ts b/packages/app/src/runtime/replica-cache/index.ts new file mode 100644 index 000000000..7f8582884 --- /dev/null +++ b/packages/app/src/runtime/replica-cache/index.ts @@ -0,0 +1,422 @@ +import { Buffer } from "buffer"; +import { z } from "zod"; +import { + AgentSnapshotPayloadSchema, + ProjectPlacementPayloadSchema, + WorkspaceDescriptorPayloadSchema, + WorkspaceProjectDescriptorPayloadSchema, + type AgentSnapshotPayload, + type WorkspaceDescriptorPayload, +} from "@getpaseo/protocol/messages"; +import { + normalizeEmptyProjectDescriptor, + normalizeWorkspaceDescriptor, + useSessionStore, + type Agent, + type SessionReplica, + type SessionState, + type WorkspaceDescriptor, +} from "@/stores/session-store"; +import type { StreamItem } from "@/types/stream"; +import { normalizeAgentSnapshot } from "@/utils/agent-snapshots"; + +const STORAGE_KEY = "@paseo:replica-cache"; +const CACHE_VERSION = 1; +const PERSIST_DELAY_MS = 750; +const MAX_TIMELINE_ITEMS = 50; +const MAX_CACHE_BYTES = 1024 * 1024; +const DATE_TAG = "__paseoDate"; + +const StoredAgentSchema = z.object({ + snapshot: AgentSnapshotPayloadSchema, + projectPlacement: ProjectPlacementPayloadSchema.nullable(), + lastActivityAt: z.string(), +}); + +const StoredTimelineSchema = z.object({ + agentId: z.string(), + items: z.unknown(), + cursor: z + .object({ + epoch: z.string(), + startSeq: z.number().int().nonnegative(), + endSeq: z.number().int().nonnegative(), + }) + .nullable(), + hasOlder: z.boolean(), +}); + +const StoredHostSchema = z.object({ + serverId: z.string(), + agents: z.array(StoredAgentSchema), + workspaces: z.array(WorkspaceDescriptorPayloadSchema), + emptyProjects: z.array(WorkspaceProjectDescriptorPayloadSchema), + timeline: StoredTimelineSchema.nullable(), +}); + +const StoredCacheSchema = z.object({ + version: z.literal(CACHE_VERSION), + hosts: z.array(StoredHostSchema), +}); + +type StoredAgent = z.infer; +type StoredHost = z.infer; + +export interface ReplicaCacheStorage { + getItem: (key: string) => Promise; + setItem: (key: string, value: string) => Promise; +} + +interface ReplicaCacheOptions { + maxBytes?: number; +} + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + +function hasString(value: Record, key: string): boolean { + return typeof value[key] === "string"; +} + +function isStreamItem(value: unknown): value is StreamItem { + if (!isRecord(value) || !hasString(value, "id") || !(value.timestamp instanceof Date)) { + return false; + } + switch (value.kind) { + case "user_message": + case "assistant_message": + return hasString(value, "text"); + case "thought": + return hasString(value, "text") && (value.status === "loading" || value.status === "ready"); + case "tool_call": + return isRecord(value.payload) && isRecord(value.payload.data); + case "todo_list": + return hasString(value, "provider") && Array.isArray(value.items); + case "activity_log": + return hasString(value, "message") && hasString(value, "activityType"); + case "compaction": + return value.status === "loading" || value.status === "completed"; + default: + return false; + } +} + +function encodeDates(value: unknown): unknown { + if (value instanceof Date) { + return { [DATE_TAG]: value.toISOString() }; + } + if (Array.isArray(value)) { + return value.map(encodeDates); + } + if (!isRecord(value)) { + return value; + } + return Object.fromEntries(Object.entries(value).map(([key, entry]) => [key, encodeDates(entry)])); +} + +function decodeDates(value: unknown): unknown { + if (Array.isArray(value)) { + return value.map(decodeDates); + } + if (!isRecord(value)) { + return value; + } + if (Object.keys(value).length === 1 && typeof value[DATE_TAG] === "string") { + const date = new Date(value[DATE_TAG]); + return Number.isNaN(date.getTime()) ? value : date; + } + return Object.fromEntries(Object.entries(value).map(([key, entry]) => [key, decodeDates(entry)])); +} + +function deserializeTimeline(stored: StoredHost["timeline"]): SessionReplica["timeline"] { + if (!stored) { + return null; + } + const decoded = decodeDates(stored.items); + if (!Array.isArray(decoded) || !decoded.every(isStreamItem)) { + return null; + } + return { + agentId: stored.agentId, + items: decoded, + cursor: stored.cursor, + hasOlder: stored.hasOlder, + }; +} + +function serializeAgent(agent: Agent): StoredAgent { + const snapshot: AgentSnapshotPayload = { + 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, + capabilities: agent.capabilities, + currentModeId: agent.currentModeId, + availableModes: agent.availableModes, + 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, + }; + return { + snapshot, + projectPlacement: agent.projectPlacement ?? null, + lastActivityAt: agent.lastActivityAt.toISOString(), + }; +} + +function deserializeAgent(serverId: string, stored: StoredAgent): Agent { + return { + ...normalizeAgentSnapshot(stored.snapshot, serverId), + lastActivityAt: new Date(stored.lastActivityAt), + projectPlacement: stored.projectPlacement, + }; +} + +function serializeWorkspace(workspace: WorkspaceDescriptor): WorkspaceDescriptorPayload { + return { + id: workspace.id, + projectId: workspace.projectId, + projectDisplayName: workspace.projectDisplayName, + projectCustomName: workspace.projectCustomName ?? null, + projectRootPath: workspace.projectRootPath, + workspaceDirectory: workspace.workspaceDirectory, + projectKind: workspace.projectKind, + workspaceKind: workspace.workspaceKind, + name: workspace.name, + title: workspace.title ?? null, + pinnedAt: workspace.pinnedAt ?? null, + status: workspace.status, + statusEnteredAt: workspace.statusEnteredAt?.toISOString() ?? null, + activityAt: null, + archivingAt: workspace.archivingAt, + diffStat: workspace.diffStat, + scripts: workspace.scripts, + gitRuntime: workspace.gitRuntime, + githubRuntime: workspace.githubRuntime, + forge: workspace.forge, + project: workspace.project, + }; +} + +function deserializeHost(stored: StoredHost): SessionReplica { + const agents = stored.agents.map((entry) => deserializeAgent(stored.serverId, entry)); + const workspaces = stored.workspaces.map(normalizeWorkspaceDescriptor); + const emptyProjects = stored.emptyProjects.map(normalizeEmptyProjectDescriptor); + return { + agents: new Map(agents.map((agent) => [agent.id, agent])), + workspaces: new Map(workspaces.map((workspace) => [workspace.id, workspace])), + emptyProjects: new Map(emptyProjects.map((project) => [project.projectId, project])), + timeline: deserializeTimeline(stored.timeline), + }; +} + +export class ReplicaCache { + private readonly activeServerIds = new Set(); + private readonly storedHosts = new Map(); + private readonly lastFocusedAgentIds = new Map(); + private readonly capturedSessions = new Map(); + private readonly maxBytes: number; + private needsPersist = false; + private unsubscribe: (() => void) | null = null; + private persistTimer: ReturnType | null = null; + private writeQueue: Promise = Promise.resolve(); + + constructor( + private readonly storage: ReplicaCacheStorage, + options: ReplicaCacheOptions = {}, + ) { + const emptyPayloadBytes = Buffer.byteLength( + JSON.stringify({ version: CACHE_VERSION, hosts: [] }), + "utf8", + ); + this.maxBytes = Math.max(options.maxBytes ?? MAX_CACHE_BYTES, emptyPayloadBytes); + } + + setHosts(serverIds: Iterable): void { + const next = new Set(serverIds); + this.activeServerIds.clear(); + for (const serverId of next) this.activeServerIds.add(serverId); + let removedStoredHost = false; + for (const serverId of this.storedHosts.keys()) { + if (!next.has(serverId)) { + this.storedHosts.delete(serverId); + removedStoredHost = true; + } + } + for (const serverId of this.lastFocusedAgentIds.keys()) { + if (!next.has(serverId)) this.lastFocusedAgentIds.delete(serverId); + } + for (const serverId of this.capturedSessions.keys()) { + if (!next.has(serverId)) this.capturedSessions.delete(serverId); + } + if (removedStoredHost) this.needsPersist = true; + if (this.unsubscribe && this.needsPersist) this.schedulePersist(); + } + + async restore(): Promise { + let raw: string | null; + try { + raw = await this.storage.getItem(STORAGE_KEY); + } catch { + return; + } + if (!raw) return; + let parsed: unknown; + try { + parsed = JSON.parse(raw); + } catch { + return; + } + const cache = StoredCacheSchema.safeParse(parsed); + if (!cache.success) return; + for (const host of cache.data.hosts) { + if (!this.activeServerIds.has(host.serverId)) { + this.needsPersist = true; + continue; + } + this.storedHosts.set(host.serverId, host); + if (host.timeline) this.lastFocusedAgentIds.set(host.serverId, host.timeline.agentId); + } + if (this.buildBoundedPayload().evicted) this.needsPersist = true; + for (const host of this.storedHosts.values()) { + useSessionStore.getState().restoreSessionReplica(host.serverId, deserializeHost(host)); + const session = useSessionStore.getState().sessions[host.serverId]; + if (session) this.capturedSessions.set(host.serverId, session); + } + } + + start(): void { + if (this.unsubscribe) return; + this.unsubscribe = useSessionStore.subscribe((state) => { + if (this.activeServerIds.size === 0) return; + for (const serverId of this.activeServerIds) { + const focusedAgentId = state.sessions[serverId]?.focusedAgentId; + if (focusedAgentId) this.lastFocusedAgentIds.set(serverId, focusedAgentId); + } + this.schedulePersist(); + }); + if (this.needsPersist) this.schedulePersist(); + } + + reconcileServerId(oldServerId: string, newServerId: string): void { + const stored = this.storedHosts.get(oldServerId); + if (stored) { + this.storedHosts.delete(oldServerId); + this.storedHosts.set(newServerId, { ...stored, serverId: newServerId }); + } + const focusedAgentId = this.lastFocusedAgentIds.get(oldServerId); + if (focusedAgentId) { + this.lastFocusedAgentIds.delete(oldServerId); + this.lastFocusedAgentIds.set(newServerId, focusedAgentId); + } + const capturedSession = this.capturedSessions.get(oldServerId); + if (capturedSession) { + this.capturedSessions.delete(oldServerId); + this.capturedSessions.set(newServerId, capturedSession); + } + if (this.activeServerIds.delete(oldServerId)) this.activeServerIds.add(newServerId); + this.needsPersist = true; + this.schedulePersist(); + } + + async flush(): Promise { + if (this.persistTimer) { + clearTimeout(this.persistTimer); + this.persistTimer = null; + } + this.captureSessions(); + const { payload } = this.buildBoundedPayload(); + this.needsPersist = false; + const write = this.writeQueue + .catch(() => undefined) + .then(() => this.storage.setItem(STORAGE_KEY, payload)); + this.writeQueue = write; + await write.catch(() => undefined); + } + + private captureSessions(): void { + const sessions = useSessionStore.getState().sessions; + for (const serverId of this.activeServerIds) { + const session = sessions[serverId]; + if (!session) continue; + if (this.capturedSessions.get(serverId) === session) continue; + this.capturedSessions.set(serverId, session); + if (session.focusedAgentId) { + this.lastFocusedAgentIds.set(serverId, session.focusedAgentId); + } + const focusedAgentId = this.lastFocusedAgentIds.get(serverId) ?? null; + const focusedAgent = focusedAgentId ? session.agents.get(focusedAgentId) : undefined; + const focusedWorkspace = focusedAgent + ? ((focusedAgent.workspaceId + ? session.workspaces.get(focusedAgent.workspaceId) + : undefined) ?? + Array.from(session.workspaces.values()).find( + (workspace) => workspace.workspaceDirectory === focusedAgent.cwd, + )) + : undefined; + const items = focusedAgentId ? session.agentStreamTail.get(focusedAgentId) : undefined; + const timeline = + focusedAgent && items + ? { + agentId: focusedAgent.id, + items: encodeDates(items.slice(-MAX_TIMELINE_ITEMS)), + cursor: session.agentTimelineCursor.get(focusedAgent.id) ?? null, + hasOlder: session.agentTimelineHasOlder.get(focusedAgent.id) ?? false, + } + : null; + const stored: StoredHost = { + serverId, + agents: focusedAgent ? [serializeAgent(focusedAgent)] : [], + workspaces: focusedWorkspace ? [serializeWorkspace(focusedWorkspace)] : [], + emptyProjects: [], + timeline, + }; + this.storedHosts.delete(serverId); + this.storedHosts.set(serverId, stored); + } + } + + private buildBoundedPayload(): { payload: string; evicted: boolean } { + let evicted = false; + let payload = this.serialize(); + while (Buffer.byteLength(payload, "utf8") > this.maxBytes && this.storedHosts.size > 0) { + const oldestServerId = this.storedHosts.keys().next().value; + if (oldestServerId === undefined) break; + this.storedHosts.delete(oldestServerId); + evicted = true; + payload = this.serialize(); + } + return { payload, evicted }; + } + + private serialize(): string { + return JSON.stringify({ + version: CACHE_VERSION, + hosts: Array.from(this.storedHosts.values()), + }); + } + + private schedulePersist(): void { + if (this.persistTimer) return; + this.persistTimer = setTimeout(() => { + this.persistTimer = null; + void this.flush(); + }, PERSIST_DELAY_MS); + } +} diff --git a/packages/app/src/stores/session-store.ts b/packages/app/src/stores/session-store.ts index baa5d26ba..d6e37a171 100644 --- a/packages/app/src/stores/session-store.ts +++ b/packages/app/src/stores/session-store.ts @@ -320,6 +320,20 @@ export interface AgentTimelineCursorState { endSeq: number; } +export interface SessionReplicaTimeline { + agentId: string; + items: StreamItem[]; + cursor: AgentTimelineCursorState | null; + hasOlder: boolean; +} + +export interface SessionReplica { + agents: Map; + workspaces: Map; + emptyProjects: Map; + timeline: SessionReplicaTimeline | null; +} + export type WorkspaceRestoreStatus = "restoring" | "failed" | "needs-host-upgrade"; // Per-session state @@ -398,7 +412,12 @@ interface SessionStoreState { // Action types interface SessionStoreActions { // Session management - initializeSession: (serverId: string, client: DaemonClient, clientGeneration?: number) => void; + initializeSession: ( + serverId: string, + client: DaemonClient | null, + clientGeneration?: number, + ) => void; + restoreSessionReplica: (serverId: string, replica: SessionReplica) => void; clearSession: (serverId: string) => void; getSession: (serverId: string) => SessionState | undefined; updateSessionClient: (serverId: string, client: DaemonClient, clientGeneration?: number) => void; @@ -549,7 +568,7 @@ const agentLastActivityCoalescer = createAgentLastActivityCoalescer(); // Helper to create initial session state function createInitialSessionState( serverId: string, - client: DaemonClient, + client: DaemonClient | null, clientGeneration = 0, ): SessionState { return { @@ -675,6 +694,51 @@ export const useSessionStore = create()( }); }, + restoreSessionReplica: (serverId, replica) => { + set((prev) => { + if (prev.sessions[serverId]) { + return prev; + } + const session = createInitialSessionState(serverId, null); + const timeline = replica.timeline; + const agentStreamTail = new Map(); + const agentTimelineCursor = new Map(); + const agentTimelineHasOlder = new Map(); + const agentAuthoritativeHistoryApplied = new Map(); + const agentHistorySyncGeneration = new Map(); + if (timeline) { + agentStreamTail.set(timeline.agentId, timeline.items); + agentTimelineHasOlder.set(timeline.agentId, timeline.hasOlder); + agentAuthoritativeHistoryApplied.set(timeline.agentId, true); + agentHistorySyncGeneration.set(timeline.agentId, session.historySyncGeneration); + if (timeline.cursor) agentTimelineCursor.set(timeline.agentId, timeline.cursor); + } + const agentLastActivity = new Map(prev.agentLastActivity); + for (const agent of replica.agents.values()) { + agentLastActivity.set(agent.id, agent.lastActivityAt); + } + return { + ...prev, + sessions: { + ...prev.sessions, + [serverId]: { + ...session, + agents: replica.agents, + workspaceAgentActivity: buildWorkspaceAgentActivityIndex(replica.agents), + workspaces: replica.workspaces, + emptyProjects: replica.emptyProjects, + agentStreamTail, + agentTimelineCursor, + agentTimelineHasOlder, + agentAuthoritativeHistoryApplied, + agentHistorySyncGeneration, + }, + }, + agentLastActivity, + }; + }); + }, + clearSession: (serverId) => { agentLastActivityCoalescer.flushNow(); set((prev) => {