From 6eb6be77dfd3612a171089cf5a4adae4f8d1e40a Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Wed, 25 Mar 2026 01:59:37 +0700 Subject: [PATCH] Phase 1a: collapse snapshot mutation ownership --- .../src/server/agent/agent-manager.test.ts | 160 ++++++++++++++++++ .../server/src/server/agent/agent-manager.ts | 135 +++++++++++++-- packages/server/src/server/bootstrap.ts | 8 - .../src/server/daemon-client.e2e.test.ts | 38 +++++ .../src/server/persistence-hooks.test.ts | 118 +------------ .../server/src/server/persistence-hooks.ts | 37 +--- packages/server/src/server/session.ts | 78 +-------- .../snapshot-mutation-ownership.test.ts | 158 +++++++++++++++++ 8 files changed, 491 insertions(+), 241 deletions(-) create mode 100644 packages/server/src/server/snapshot-mutation-ownership.test.ts diff --git a/packages/server/src/server/agent/agent-manager.test.ts b/packages/server/src/server/agent/agent-manager.test.ts index 5d2e78a471..d049d43996 100644 --- a/packages/server/src/server/agent/agent-manager.test.ts +++ b/packages/server/src/server/agent/agent-manager.test.ts @@ -742,6 +742,144 @@ describe("AgentManager", () => { expect(live!.updatedAt.getTime()).toBeGreaterThan(Date.parse(before!.updatedAt)); }); + test("persists live mode, model, and thinking changes without an external snapshot subscriber", async () => { + const workdir = mkdtempSync(join(tmpdir(), "agent-manager-live-persist-")); + const storagePath = join(workdir, "agents"); + const storage = new AgentStorage(storagePath, logger); + const manager = new AgentManager({ + clients: { + codex: new TestAgentClient(), + }, + registry: storage, + logger, + idFactory: () => "00000000-0000-4000-8000-000000000132", + }); + + const snapshot = await manager.createAgent({ + provider: "codex", + cwd: workdir, + modeId: "plan", + model: "gpt-5.2-codex", + thinkingOptionId: "low", + }); + + await manager.setAgentMode(snapshot.id, "build"); + await manager.setAgentModel(snapshot.id, "gpt-5.4"); + await manager.setAgentThinkingOption(snapshot.id, "high"); + await manager.flush(); + + const persisted = await storage.get(snapshot.id); + expect(persisted).not.toBeNull(); + expect(persisted?.lastModeId).toBe("build"); + expect(persisted?.config?.model).toBe("gpt-5.4"); + expect(persisted?.config?.thinkingOptionId).toBe("high"); + expect(persisted?.runtimeInfo?.modeId).toBe("build"); + expect(persisted?.runtimeInfo?.model).toBe("gpt-5.4"); + }); + + test("setLabels merges and persists labels", async () => { + const workdir = mkdtempSync(join(tmpdir(), "agent-manager-set-labels-")); + const storagePath = join(workdir, "agents"); + const storage = new AgentStorage(storagePath, logger); + const manager = new AgentManager({ + clients: { + codex: new TestAgentClient(), + }, + registry: storage, + logger, + idFactory: () => "00000000-0000-4000-8000-000000000133", + }); + + const snapshot = await manager.createAgent({ + provider: "codex", + cwd: workdir, + title: "Label test", + }); + + await manager.setLabels(snapshot.id, { surface: "mobile" }); + await manager.setLabels(snapshot.id, { phase: "1a" }); + + const persisted = await storage.get(snapshot.id); + expect(persisted?.labels).toEqual({ + surface: "mobile", + phase: "1a", + }); + }); + + test("runAgent persists finished attention and idle status without an external snapshot subscriber", async () => { + const workdir = mkdtempSync(join(tmpdir(), "agent-manager-finished-attention-")); + const storagePath = join(workdir, "agents"); + const storage = new AgentStorage(storagePath, logger); + const manager = new AgentManager({ + clients: { + codex: new TestAgentClient(), + }, + registry: storage, + logger, + idFactory: () => "00000000-0000-4000-8000-000000000134", + }); + + const snapshot = await manager.createAgent({ + provider: "codex", + cwd: workdir, + title: "Finished attention test", + }); + + await manager.runAgent(snapshot.id, "say hello"); + await manager.flush(); + + const persisted = await storage.get(snapshot.id); + expect(persisted?.lastStatus).toBe("idle"); + expect(persisted?.requiresAttention).toBe(true); + expect(persisted?.attentionReason).toBe("finished"); + expect(persisted?.attentionTimestamp).toEqual(expect.any(String)); + }); + + test("archiveSnapshot clears persisted attention and normalizes running status", async () => { + const workdir = mkdtempSync(join(tmpdir(), "agent-manager-archive-attention-")); + const storagePath = join(workdir, "agents"); + const storage = new AgentStorage(storagePath, logger); + const manager = new AgentManager({ + clients: { + codex: new TestAgentClient(), + }, + registry: storage, + logger, + idFactory: () => "00000000-0000-4000-8000-000000000135", + }); + + const snapshot = await manager.createAgent({ + provider: "codex", + cwd: workdir, + title: "Archive attention test", + }); + + const live = manager.getAgent(snapshot.id); + expect(live).not.toBeNull(); + live!.lifecycle = "running"; + live!.attention = { + requiresAttention: true, + attentionReason: "finished", + attentionTimestamp: new Date("2025-01-02T00:00:00.000Z"), + }; + + const archivedAt = "2025-01-03T00:00:00.000Z"; + const archivedRecord = await manager.archiveSnapshot(snapshot.id, archivedAt); + + expect(archivedRecord.archivedAt).toBe(archivedAt); + expect(archivedRecord.lastStatus).toBe("idle"); + expect(archivedRecord.requiresAttention).toBe(false); + expect(archivedRecord.attentionReason).toBeNull(); + expect(archivedRecord.attentionTimestamp).toBeNull(); + + const persisted = await storage.get(snapshot.id); + expect(persisted?.archivedAt).toBe(archivedAt); + expect(persisted?.lastStatus).toBe("idle"); + expect(persisted?.requiresAttention).toBe(false); + expect(persisted?.attentionReason).toBeNull(); + expect(persisted?.attentionTimestamp).toBeNull(); + }); + test("reloadAgentSession cancels active run and resumes existing session once thread_started is observed", async () => { const workdir = mkdtempSync(join(tmpdir(), "agent-manager-reload-active-")); const storagePath = join(workdir, "agents"); @@ -2541,6 +2679,7 @@ describe("AgentManager", () => { }); await expect(manager.runAgent(agent.id, "fail once")).rejects.toThrow("boom-1"); + await manager.flush(); const afterFirstFailure = manager.getAgent(agent.id); expect(afterFirstFailure?.lifecycle).toBe("error"); @@ -2550,14 +2689,26 @@ describe("AgentManager", () => { attentionReason: "error", }); + const persistedAfterFirstFailure = await storage.get(agent.id); + expect(persistedAfterFirstFailure?.lastStatus).toBe("error"); + expect(persistedAfterFirstFailure?.requiresAttention).toBe(true); + expect(persistedAfterFirstFailure?.attentionReason).toBe("error"); + await manager.clearAgentAttention(agent.id); manager.notifyAgentState(agent.id); + await manager.flush(); const afterClear = manager.getAgent(agent.id); expect(afterClear?.lifecycle).toBe("error"); expect(afterClear?.attention).toEqual({ requiresAttention: false }); + const persistedAfterClear = await storage.get(agent.id); + expect(persistedAfterClear?.lastStatus).toBe("error"); + expect(persistedAfterClear?.requiresAttention).toBe(false); + expect(persistedAfterClear?.attentionReason).toBeNull(); + await expect(manager.runAgent(agent.id, "fail again")).rejects.toThrow("boom-2"); + await manager.flush(); const afterSecondFailure = manager.getAgent(agent.id); expect(afterSecondFailure?.lifecycle).toBe("error"); @@ -2566,6 +2717,11 @@ describe("AgentManager", () => { attentionReason: "error", }); expect(attentionReasons).toEqual(["error", "error"]); + + const persistedAfterSecondFailure = await storage.get(agent.id); + expect(persistedAfterSecondFailure?.lastStatus).toBe("error"); + expect(persistedAfterSecondFailure?.requiresAttention).toBe(true); + expect(persistedAfterSecondFailure?.attentionReason).toBe("error"); }); test("turn_failed emits a system error assistant timeline message and keeps error lifecycle", async () => { @@ -2929,6 +3085,10 @@ describe("AgentManager", () => { // The manager should have updated currentModeId to reflect this const updatedAgent = manager.getAgent(snapshot.id); expect(updatedAgent?.currentModeId).toBe("acceptEdits"); + + await manager.flush(); + const persisted = await storage.get(snapshot.id); + expect(persisted?.lastModeId).toBe("acceptEdits"); }); test("close during in-flight stream does not clear persistence sessionId", async () => { diff --git a/packages/server/src/server/agent/agent-manager.ts b/packages/server/src/server/agent/agent-manager.ts index e8b99dd8f2..05afa54ae3 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -30,7 +30,7 @@ import type { ListPersistedAgentsOptions, PersistedAgentDescriptor, } from "./agent-sdk-types.js"; -import type { AgentStorage } from "./agent-storage.js"; +import type { AgentStorage, StoredAgentRecord } from "./agent-storage.js"; import { AGENT_PROVIDER_IDS } from "./provider-manifest.js"; export { AGENT_LIFECYCLE_STATUSES, type AgentLifecycleStatus }; @@ -797,6 +797,12 @@ export class AgentManager { } agent.config.thinkingOptionId = normalizedThinkingOptionId ?? undefined; + if (agent.runtimeInfo) { + agent.runtimeInfo = { + ...agent.runtimeInfo, + thinkingOptionId: normalizedThinkingOptionId, + }; + } this.touchUpdatedAt(agent); this.emitState(agent); } @@ -809,15 +815,15 @@ export class AgentManager { } this.touchUpdatedAt(agent); await this.persistSnapshot(agent, { title: normalizedTitle }); - this.emitState(agent); + this.emitState(agent, { persist: false }); } async setLabels(agentId: string, labels: Record): Promise { const agent = this.requireAgent(agentId); agent.labels = { ...agent.labels, ...labels }; - await this.persistSnapshot(agent); this.touchUpdatedAt(agent); - this.emitState(agent); + await this.persistSnapshot(agent); + this.emitState(agent, { persist: false }); } notifyAgentState(agentId: string): void { @@ -834,10 +840,105 @@ export class AgentManager { if (agent.attention.requiresAttention) { agent.attention = { requiresAttention: false }; await this.persistSnapshot(agent); - this.emitState(agent); + this.emitState(agent, { persist: false }); } } + async archiveSnapshot(agentId: string, archivedAt: string): Promise { + const registry = this.requireRegistry(); + const liveAgent = this.getAgent(agentId); + if (liveAgent) { + await this.persistSnapshot(liveAgent, { + internal: liveAgent.internal, + }); + } + + const record = await registry.get(agentId); + if (!record) { + throw new Error(`Agent not found: ${agentId}`); + } + + const normalizedStatus = + record.lastStatus === "running" || record.lastStatus === "initializing" + ? "idle" + : record.lastStatus; + + const nextRecord: StoredAgentRecord = { + ...record, + archivedAt, + lastStatus: normalizedStatus, + requiresAttention: false, + attentionReason: null, + attentionTimestamp: null, + }; + await registry.upsert(nextRecord); + return nextRecord; + } + + async unarchiveSnapshot(agentId: string): Promise { + const registry = this.requireRegistry(); + const record = await registry.get(agentId); + if (!record || !record.archivedAt) { + return false; + } + + await registry.upsert({ + ...record, + archivedAt: null, + }); + + if (this.getAgent(agentId)) { + this.notifyAgentState(agentId); + } + return true; + } + + async unarchiveSnapshotByHandle(handle: AgentPersistenceHandle): Promise { + const registry = this.requireRegistry(); + const records = await registry.list(); + const matched = records.find( + (record) => + record.persistence?.provider === handle.provider && + record.persistence?.sessionId === handle.sessionId, + ); + if (!matched) { + return; + } + + await this.unarchiveSnapshot(matched.id); + } + + async updateAgentMetadata( + agentId: string, + updates: { + title?: string; + labels?: Record; + }, + ): Promise { + const liveAgent = this.getAgent(agentId); + if (liveAgent) { + if (updates.title) { + await this.setTitle(agentId, updates.title); + } + if (updates.labels) { + await this.setLabels(agentId, updates.labels); + } + return; + } + + const registry = this.requireRegistry(); + const existing = await registry.get(agentId); + if (!existing) { + throw new Error(`Agent not found: ${agentId}`); + } + + await registry.upsert({ + ...existing, + ...(updates.title ? { title: updates.title } : {}), + ...(updates.labels ? { labels: { ...existing.labels, ...updates.labels } } : {}), + }); + } + async runAgent( agentId: string, prompt: AgentPromptInput, @@ -1263,6 +1364,12 @@ export class AgentManager { // (e.g., plan approval changes mode from "plan" to "acceptEdits") try { agent.currentModeId = await agent.session.getCurrentMode(); + if (agent.runtimeInfo) { + agent.runtimeInfo = { + ...agent.runtimeInfo, + modeId: agent.currentModeId, + }; + } } catch { // Ignore errors from getCurrentMode - mode tracking is best effort } @@ -1652,12 +1759,12 @@ export class AgentManager { await this.persistSnapshot(managed, { title: initialPersistedTitle, }); - this.emitState(managed); + this.emitState(managed, { persist: false }); await this.refreshSessionState(managed); managed.lifecycle = "idle"; await this.persistSnapshot(managed); - this.emitState(managed); + this.emitState(managed, { persist: false }); this.subscribeToSession(managed); return { ...managed }; } @@ -1807,6 +1914,13 @@ export class AgentManager { await this.registry.applySnapshot(agent, options); } + private requireRegistry(): AgentStorage { + if (!this.registry) { + throw new Error("Agent storage unavailable"); + } + return this.registry; + } + private async refreshSessionState(agent: ActiveManagedAgent): Promise { try { const modes = await agent.session.getAvailableModes(); @@ -2251,9 +2365,12 @@ export class AgentManager { return row; } - private emitState(agent: ManagedAgent): void { + private emitState(agent: ManagedAgent, options?: { persist?: boolean }): void { // Keep attention as an edge-triggered unread signal, not a level signal. this.checkAndSetAttention(agent); + if (options?.persist !== false) { + this.enqueueBackgroundPersist(agent); + } this.dispatch({ type: "agent_state", @@ -2286,7 +2403,6 @@ export class AgentManager { attentionTimestamp: new Date(), }; this.broadcastAgentAttention(agent, "finished"); - this.enqueueBackgroundPersist(agent); return; } @@ -2298,7 +2414,6 @@ export class AgentManager { attentionTimestamp: new Date(), }; this.broadcastAgentAttention(agent, "error"); - this.enqueueBackgroundPersist(agent); return; } } diff --git a/packages/server/src/server/bootstrap.ts b/packages/server/src/server/bootstrap.ts index 2485db8d01..92a67bb693 100644 --- a/packages/server/src/server/bootstrap.ts +++ b/packages/server/src/server/bootstrap.ts @@ -85,7 +85,6 @@ import type { RequestedSpeechProviders } from "./speech/speech-types.js"; import { initializeSpeechRuntime } from "./speech/speech-runtime.js"; import { AgentManager } from "./agent/agent-manager.js"; import { AgentStorage } from "./agent/agent-storage.js"; -import { attachAgentStoragePersistence } from "./persistence-hooks.js"; import { createAgentMcpServer } from "./agent/mcp-server.js"; import { createAllClients, shutdownProviders } from "./agent/provider-registry.js"; import { bootstrapWorkspaceRegistries } from "./workspace-registry-bootstrap.js"; @@ -377,12 +376,6 @@ export async function createPaseoDaemon( }); const terminalManager = createTerminalManager(); - - const detachAgentStoragePersistence = attachAgentStoragePersistence( - logger, - agentManager, - agentStorage, - ); await agentStorage.initialize(); logger.info({ elapsed: elapsed() }, "Agent storage initialized"); await bootstrapWorkspaceRegistries({ @@ -709,7 +702,6 @@ export async function createPaseoDaemon( const stop = async () => { await closeAllAgents(logger, agentManager); await agentManager.flush().catch(() => undefined); - detachAgentStoragePersistence(); await agentStorage.flush().catch(() => undefined); await shutdownProviders(logger, { runtimeSettings: config.agentProviderSettings, diff --git a/packages/server/src/server/daemon-client.e2e.test.ts b/packages/server/src/server/daemon-client.e2e.test.ts index 7ab5383717..ec55000914 100644 --- a/packages/server/src/server/daemon-client.e2e.test.ts +++ b/packages/server/src/server/daemon-client.e2e.test.ts @@ -224,6 +224,8 @@ describe("daemon client E2E", () => { expect(archivedResult).not.toBeNull(); expect(archivedResult?.agent.archivedAt).toBeTruthy(); expect(archivedResult?.agent.status).not.toBe("running"); + expect(archivedResult?.agent.requiresAttention).toBe(false); + expect(archivedResult?.agent.attentionReason).toBeNull(); expect(archivedResult?.project).not.toBeNull(); expect(archivedResult?.project?.checkout.cwd).toBe(cwd); @@ -309,6 +311,42 @@ describe("daemon client E2E", () => { } }, 180000); + test("update_agent persists unloaded title and labels across auto-unarchive", async () => { + const cwd = tmpCwd(); + try { + const created = await ctx.client.createAgent({ + config: { + ...getFullAccessConfig("codex"), + cwd, + }, + }); + + await ctx.client.archiveAgent(created.id); + await ctx.client.updateAgent(created.id, { + name: "Pinned Title", + labels: { lane: "phase-1a" }, + }); + + const archived = await ctx.client.fetchAgent(created.id); + expect(archived).not.toBeNull(); + expect(archived?.agent.archivedAt).toBeTruthy(); + expect(archived?.agent.title).toBe("Pinned Title"); + expect(archived?.agent.labels).toMatchObject({ lane: "phase-1a" }); + + await ctx.client.sendMessage(created.id, "Say hello and nothing else"); + const finalState = await ctx.client.waitForFinish(created.id, 120000); + expect(finalState.status).toBe("idle"); + + const unarchived = await ctx.client.fetchAgent(created.id); + expect(unarchived).not.toBeNull(); + expect(unarchived?.agent.archivedAt).toBeNull(); + expect(unarchived?.agent.title).toBe("Pinned Title"); + expect(unarchived?.agent.labels).toMatchObject({ lane: "phase-1a" }); + } finally { + rmSync(cwd, { recursive: true, force: true }); + } + }, 180000); + test("returns home-scoped directory suggestions", async () => { const insideHomeDir = mkdtempSync(path.join(homedir(), "paseo-dir-suggestion-")); const outsideHomeDir = mkdtempSync(path.join(tmpdir(), "paseo-dir-suggestion-outside-")); diff --git a/packages/server/src/server/persistence-hooks.test.ts b/packages/server/src/server/persistence-hooks.test.ts index bedcb57348..18e465c010 100644 --- a/packages/server/src/server/persistence-hooks.test.ts +++ b/packages/server/src/server/persistence-hooks.test.ts @@ -1,83 +1,6 @@ -import { describe, expect, test, vi } from "vitest"; - -import type { ManagedAgent } from "./agent/agent-manager.js"; +import { describe, expect, test } from "vitest"; import type { StoredAgentRecord } from "./agent/agent-storage.js"; -import { - attachAgentStoragePersistence, - buildConfigOverrides, - buildSessionConfig, -} from "./persistence-hooks.js"; -import type { - AgentPermissionRequest, - AgentSession, - AgentSessionConfig, -} from "./agent/agent-sdk-types.js"; - -const testLogger = { - child: () => testLogger, - error: vi.fn(), -} as any; - -type ManagedAgentOverrides = Omit< - Partial, - "config" | "pendingPermissions" | "session" | "activeForegroundTurnId" -> & { - config?: Partial; - pendingPermissions?: Map; - session?: AgentSession | null; - activeForegroundTurnId?: string | null; -}; - -function createManagedAgent(overrides: ManagedAgentOverrides = {}): ManagedAgent { - const now = overrides.updatedAt ?? new Date("2025-01-01T00:00:00.000Z"); - const provider = overrides.provider ?? "claude"; - const cwd = overrides.cwd ?? "/tmp/project"; - const lifecycle = overrides.lifecycle ?? "idle"; - const configOverrides = overrides.config ?? {}; - const config: AgentSessionConfig = { - provider, - cwd, - modeId: configOverrides.modeId ?? "plan", - model: configOverrides.model ?? "claude-3.5-sonnet", - extra: configOverrides.extra ?? { claude: { tone: "focused" } }, - }; - const session = lifecycle === "closed" ? null : (overrides.session ?? ({} as AgentSession)); - const activeForegroundTurnId = - overrides.activeForegroundTurnId ?? (lifecycle === "running" ? "test-turn-id" : null); - - const agent: ManagedAgent = { - id: overrides.id ?? "agent-1", - provider, - cwd, - session, - capabilities: overrides.capabilities ?? { - supportsStreaming: true, - supportsSessionPersistence: true, - supportsDynamicModes: true, - supportsMcpServers: true, - supportsReasoningStream: true, - supportsToolInvocations: true, - }, - config, - lifecycle, - createdAt: overrides.createdAt ?? now, - updatedAt: overrides.updatedAt ?? now, - availableModes: overrides.availableModes ?? [], - currentModeId: overrides.currentModeId ?? config.modeId ?? null, - pendingPermissions: overrides.pendingPermissions ?? new Map(), - activeForegroundTurnId, - foregroundTurnWaiters: new Set(), - unsubscribeSession: null, - timeline: overrides.timeline ?? [], - persistence: overrides.persistence ?? null, - historyPrimed: overrides.historyPrimed ?? true, - lastUserMessageAt: overrides.lastUserMessageAt ?? now, - lastUsage: overrides.lastUsage, - lastError: overrides.lastError, - }; - - return agent; -} +import { buildConfigOverrides, buildSessionConfig } from "./persistence-hooks.js"; function createRecord(overrides?: Partial): StoredAgentRecord { const now = new Date().toISOString(); @@ -100,43 +23,6 @@ function createRecord(overrides?: Partial): StoredAgentRecord } describe("persistence hooks", () => { - test("attachAgentStoragePersistence forwards agent snapshots", async () => { - const applySnapshot = vi.fn().mockResolvedValue(undefined); - let subscriber: (event: any) => void = () => { - throw new Error("Agent manager subscriber was not registered"); - }; - const agentManager = { - subscribe: vi.fn((callback: (event: any) => void) => { - subscriber = callback; - return () => { - subscriber = () => { - throw new Error("Agent manager subscriber was not registered"); - }; - }; - }), - }; - attachAgentStoragePersistence( - testLogger, - agentManager as any, - { - applySnapshot, - list: vi.fn(), - } as any, - ); - - expect(agentManager.subscribe).toHaveBeenCalledTimes(1); - const agent = createManagedAgent(); - subscriber({ type: "agent_state", agent }); - expect(applySnapshot).toHaveBeenCalledWith(agent); - - subscriber({ - type: "agent_stream", - agentId: agent.id, - event: { type: "timeline", item: { type: "assistant_message", text: "hi" } }, - }); - expect(applySnapshot).toHaveBeenCalledTimes(1); - }); - test("buildConfigOverrides carries systemPrompt and mcpServers", () => { const record = createRecord({ title: "Voice agent (current)", diff --git a/packages/server/src/server/persistence-hooks.ts b/packages/server/src/server/persistence-hooks.ts index 9a60af11cf..2e958181a5 100644 --- a/packages/server/src/server/persistence-hooks.ts +++ b/packages/server/src/server/persistence-hooks.ts @@ -1,45 +1,10 @@ -import type { AgentManager } from "./agent/agent-manager.js"; import type { AgentProvider, AgentSessionConfig } from "./agent/agent-sdk-types.js"; -import type { AgentStorage, StoredAgentRecord } from "./agent/agent-storage.js"; - -type LoggerLike = { - child(bindings: Record): LoggerLike; - error(...args: any[]): void; -}; - -function getLogger(logger: LoggerLike): LoggerLike { - return logger.child({ module: "persistence" }); -} - -type AgentStoragePersistence = Pick; -type AgentManagerStateSource = Pick; +import type { StoredAgentRecord } from "./agent/agent-storage.js"; function isKnownProvider(provider: string): provider is AgentProvider { return provider === "claude" || provider === "codex" || provider === "opencode"; } -/** - * Attach AgentStorage persistence to an AgentManager instance so every - * agent_state snapshot is flushed to disk. - */ -export function attachAgentStoragePersistence( - logger: LoggerLike, - agentManager: AgentManagerStateSource, - storage: AgentStoragePersistence, -): () => void { - const log = getLogger(logger); - const unsubscribe = agentManager.subscribe((event) => { - if (event.type !== "agent_state") { - return; - } - void storage.applySnapshot(event.agent).catch((error) => { - log.error({ err: error, agentId: event.agent.id }, "Failed to persist agent snapshot"); - }); - }); - - return unsubscribe; -} - export function buildConfigOverrides(record: StoredAgentRecord): Partial { return { cwd: record.cwd, diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index df7b9097e1..3c5f1f04ed 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -1940,37 +1940,7 @@ export class Session { } const archivedAt = new Date().toISOString(); - const existing = await this.agentStorage.get(agentId); - let archivedRecord: StoredAgentRecord | null = existing; - if (!archivedRecord) { - const liveAgent = this.agentManager.getAgent(agentId); - if (!liveAgent) { - throw new Error(`Agent not found: ${agentId}`); - } - - await this.agentStorage.applySnapshot(liveAgent, { - internal: liveAgent.internal, - }); - archivedRecord = await this.agentStorage.get(agentId); - if (!archivedRecord) { - throw new Error(`Agent not found in storage after snapshot: ${agentId}`); - } - } - - const normalizedStatus = - archivedRecord.lastStatus === "running" || archivedRecord.lastStatus === "initializing" - ? "idle" - : archivedRecord.lastStatus; - - const nextRecord: StoredAgentRecord = { - ...archivedRecord, - archivedAt, - lastStatus: normalizedStatus, - requiresAttention: false, - attentionReason: null, - attentionTimestamp: null, - }; - await this.agentStorage.upsert(nextRecord); + const nextRecord = await this.agentManager.archiveSnapshot(agentId, archivedAt); // Unload the agent from memory — the storage record is the source of truth now. // This tears down the provider session and drops the hydrated timeline, @@ -1987,29 +1957,11 @@ export class Session { } private async unarchiveAgentState(agentId: string): Promise { - const record = await this.agentStorage.get(agentId); - if (!record || !record.archivedAt) { - return false; - } - await this.agentStorage.upsert({ - ...record, - archivedAt: null, - }); - this.agentManager.notifyAgentState(agentId); - return true; + return this.agentManager.unarchiveSnapshot(agentId); } private async unarchiveAgentByHandle(handle: AgentPersistenceHandle): Promise { - const records = await this.agentStorage.list(); - const matched = records.find( - (record) => - record.persistence?.provider === handle.provider && - record.persistence?.sessionId === handle.sessionId, - ); - if (!matched) { - return; - } - await this.unarchiveAgentState(matched.id); + await this.agentManager.unarchiveSnapshotByHandle(handle); } private async handleUpdateAgentRequest( @@ -2045,26 +1997,10 @@ export class Session { } try { - const liveAgent = this.agentManager.getAgent(agentId); - if (liveAgent) { - if (normalizedName) { - await this.agentManager.setTitle(agentId, normalizedName); - } - if (normalizedLabels) { - await this.agentManager.setLabels(agentId, normalizedLabels); - } - } else { - const existing = await this.agentStorage.get(agentId); - if (!existing) { - throw new Error(`Agent not found: ${agentId}`); - } - - await this.agentStorage.upsert({ - ...existing, - ...(normalizedName ? { title: normalizedName } : {}), - ...(normalizedLabels ? { labels: { ...existing.labels, ...normalizedLabels } } : {}), - }); - } + await this.agentManager.updateAgentMetadata(agentId, { + ...(normalizedName ? { title: normalizedName } : {}), + ...(normalizedLabels ? { labels: normalizedLabels } : {}), + }); this.emit({ type: "update_agent_response", diff --git a/packages/server/src/server/snapshot-mutation-ownership.test.ts b/packages/server/src/server/snapshot-mutation-ownership.test.ts new file mode 100644 index 0000000000..5c4fdcbff8 --- /dev/null +++ b/packages/server/src/server/snapshot-mutation-ownership.test.ts @@ -0,0 +1,158 @@ +import os from "node:os"; +import path from "node:path"; +import { mkdtempSync, rmSync } from "node:fs"; +import { afterEach, describe, expect, test, vi } from "vitest"; + +import { Session } from "./session.js"; +import { createTestPaseoDaemon } from "./test-utils/paseo-daemon.js"; + +describe("snapshot mutation ownership boundary", () => { + afterEach(() => { + vi.restoreAllMocks(); + }); + + test("daemon live mutations write one durable snapshot through the manager-owned path", async () => { + const daemonHandle = await createTestPaseoDaemon(); + const cwd = mkdtempSync(path.join(os.tmpdir(), "snapshot-owner-live-")); + + try { + const snapshot = await daemonHandle.daemon.agentManager.createAgent({ + provider: "codex", + cwd, + model: "gpt-5.2-codex", + }); + await daemonHandle.daemon.agentManager.flush(); + await daemonHandle.daemon.agentStorage.flush(); + + const applySnapshotSpy = vi.spyOn(daemonHandle.daemon.agentStorage, "applySnapshot"); + + await daemonHandle.daemon.agentManager.setAgentModel(snapshot.id, "gpt-5.4"); + await daemonHandle.daemon.agentManager.flush(); + await daemonHandle.daemon.agentStorage.flush(); + + expect(applySnapshotSpy).toHaveBeenCalledTimes(1); + + const persisted = await daemonHandle.daemon.agentStorage.get(snapshot.id); + expect(persisted?.config?.model).toBe("gpt-5.4"); + } finally { + rmSync(cwd, { recursive: true, force: true }); + await daemonHandle.close(); + } + }); + + test("session runtime flows delegate snapshot mutations to agent manager without direct storage writes", async () => { + const onMessage = vi.fn(); + const archiveSnapshot = vi.fn(async (_agentId: string, archivedAt: string) => ({ + id: "agent-1", + provider: "codex", + cwd: "/tmp/project", + createdAt: "2026-03-24T00:00:00.000Z", + updatedAt: archivedAt, + title: null, + labels: {}, + lastStatus: "idle" as const, + config: null, + persistence: null, + archivedAt, + requiresAttention: false, + attentionReason: null, + attentionTimestamp: null, + })); + const unarchiveSnapshot = vi.fn(async () => true); + const unarchiveSnapshotByHandle = vi.fn(async () => undefined); + const updateAgentMetadata = vi.fn(async () => undefined); + const directStorageWrite = vi.fn(async () => { + throw new Error("Session should not write snapshots directly"); + }); + + const logger = { + child: () => logger, + trace: vi.fn(), + debug: vi.fn(), + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + }; + + const session = new Session({ + clientId: "test-client", + onMessage, + logger: logger as any, + downloadTokenStore: {} as any, + pushTokenStore: {} as any, + paseoHome: "/tmp/paseo-test", + agentManager: { + subscribe: () => () => {}, + listAgents: () => [], + getAgent: () => null, + archiveSnapshot, + unarchiveSnapshot, + unarchiveSnapshotByHandle, + updateAgentMetadata, + } as any, + agentStorage: { + list: async () => [], + get: async () => null, + applySnapshot: directStorageWrite, + upsert: directStorageWrite, + } as any, + projectRegistry: { + initialize: async () => {}, + existsOnDisk: async () => true, + list: async () => [], + get: async () => null, + upsert: async () => {}, + archive: async () => {}, + remove: async () => {}, + } as any, + workspaceRegistry: { + initialize: async () => {}, + existsOnDisk: async () => true, + list: async () => [], + get: async () => null, + upsert: async () => {}, + archive: async () => {}, + remove: async () => {}, + } as any, + createAgentMcpTransport: async () => { + throw new Error("not used"); + }, + stt: null, + tts: null, + terminalManager: null, + }) as any; + + const archiveResult = await session.archiveAgentState("agent-1"); + expect(archiveSnapshot).toHaveBeenCalledTimes(1); + expect(archiveResult.archivedAt).toBeTruthy(); + + await session.unarchiveAgentState("agent-1"); + expect(unarchiveSnapshot).toHaveBeenCalledWith("agent-1"); + + const handle = { provider: "codex", sessionId: "session-1" }; + await session.unarchiveAgentByHandle(handle); + expect(unarchiveSnapshotByHandle).toHaveBeenCalledWith(handle); + + await session.handleUpdateAgentRequest( + "agent-1", + "Renamed agent", + { lane: "phase-1a" }, + "req-1", + ); + expect(updateAgentMetadata).toHaveBeenCalledWith("agent-1", { + title: "Renamed agent", + labels: { lane: "phase-1a" }, + }); + expect(onMessage).toHaveBeenCalledWith({ + type: "update_agent_response", + payload: { + requestId: "req-1", + agentId: "agent-1", + accepted: true, + error: null, + }, + }); + + expect(directStorageWrite).not.toHaveBeenCalled(); + }); +});