Phase 1a: collapse snapshot mutation ownership

This commit is contained in:
Mohamed Boudra
2026-03-25 20:39:07 +07:00
parent c202bd5ff7
commit 6eb6be77df
8 changed files with 491 additions and 241 deletions
@@ -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 () => {
+125 -10
View File
@@ -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<string, string>): Promise<void> {
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<StoredAgentRecord> {
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<boolean> {
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<void> {
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<string, string>;
},
): Promise<void> {
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<void> {
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;
}
}
-8
View File
@@ -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,
@@ -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-"));
@@ -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<ManagedAgent>,
"config" | "pendingPermissions" | "session" | "activeForegroundTurnId"
> & {
config?: Partial<AgentSessionConfig>;
pendingPermissions?: Map<string, AgentPermissionRequest>;
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<string, AgentPermissionRequest>(),
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>): StoredAgentRecord {
const now = new Date().toISOString();
@@ -100,43 +23,6 @@ function createRecord(overrides?: Partial<StoredAgentRecord>): 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)",
@@ -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<string, unknown>): LoggerLike;
error(...args: any[]): void;
};
function getLogger(logger: LoggerLike): LoggerLike {
return logger.child({ module: "persistence" });
}
type AgentStoragePersistence = Pick<AgentStorage, "applySnapshot" | "list">;
type AgentManagerStateSource = Pick<AgentManager, "subscribe">;
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<AgentSessionConfig> {
return {
cwd: record.cwd,
+7 -71
View File
@@ -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<boolean> {
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<void> {
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",
@@ -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();
});
});