Phase 3b: agent snapshot DB cutover + interface extraction

This commit is contained in:
Mohamed Boudra
2026-03-25 20:39:07 +07:00
parent 307a6e85f9
commit 3e9fd00501
17 changed files with 840 additions and 36 deletions
@@ -37,7 +37,7 @@ import {
import { toAgentPayload } from "./agent-projections.js";
import { curateAgentActivity } from "./activity-curator.js";
import { AGENT_PROVIDER_DEFINITIONS } from "./provider-registry.js";
import { AgentStorage } from "./agent-storage.js";
import type { AgentSnapshotStore } from "./agent-snapshot-store.js";
import {
appendTimelineItemIfAgentKnown,
emitLiveTimelineItemIfAgentKnown,
@@ -51,7 +51,7 @@ import { createAgentWorktree, runAsyncWorktreeBootstrap } from "../worktree-boot
export interface AgentManagementMcpOptions {
agentManager: AgentManager;
agentStorage: AgentStorage;
agentStorage: AgentSnapshotStore;
terminalManager?: TerminalManager | null;
paseoHome?: string;
logger: Logger;
@@ -178,7 +178,7 @@ function sanitizePermissionRequest(
}
async function resolveAgentTitle(
agentStorage: AgentStorage,
agentStorage: AgentSnapshotStore,
agentId: string,
logger: Logger,
): Promise<string | null> {
@@ -192,7 +192,7 @@ async function resolveAgentTitle(
}
async function serializeSnapshotWithMetadata(
agentStorage: AgentStorage,
agentStorage: AgentSnapshotStore,
snapshot: ManagedAgent,
logger: Logger,
) {
@@ -30,7 +30,8 @@ import type {
ListPersistedAgentsOptions,
PersistedAgentDescriptor,
} from "./agent-sdk-types.js";
import type { AgentStorage, StoredAgentRecord } from "./agent-storage.js";
import type { StoredAgentRecord } from "./agent-storage.js";
import type { AgentSnapshotStore } from "./agent-snapshot-store.js";
import {
InMemoryAgentTimelineStore,
type SeedAgentTimelineOptions,
@@ -74,7 +75,7 @@ export type ProviderAvailability = {
export type AgentManagerOptions = {
clients?: Partial<Record<AgentProvider, AgentClient>>;
idFactory?: () => string;
registry?: AgentStorage;
registry?: AgentSnapshotStore;
onAgentAttention?: AgentAttentionCallback;
logger: Logger;
};
@@ -303,7 +304,7 @@ export class AgentManager {
private readonly pendingForegroundRuns = new Map<string, PendingForegroundRun>();
private readonly subscribers = new Set<SubscriptionRecord>();
private readonly idFactory: () => string;
private readonly registry?: AgentStorage;
private readonly registry?: AgentSnapshotStore;
private readonly previousStatuses = new Map<string, AgentLifecycleStatus>();
private readonly backgroundTasks = new Set<Promise<void>>();
private onAgentAttention?: AgentAttentionCallback;
@@ -1794,7 +1795,7 @@ export class AgentManager {
await this.registry.applySnapshot(agent, options);
}
private requireRegistry(): AgentStorage {
private requireRegistry(): AgentSnapshotStore {
if (!this.registry) {
throw new Error("Agent storage unavailable");
}
@@ -0,0 +1,17 @@
import type { ManagedAgent } from "./agent-manager.js";
import type { StoredAgentRecord } from "./agent-storage.js";
export interface AgentSnapshotStore {
initialize(): Promise<void>;
list(): Promise<StoredAgentRecord[]>;
get(agentId: string): Promise<StoredAgentRecord | null>;
upsert(record: StoredAgentRecord): Promise<void>;
beginDelete(agentId: string): void;
remove(agentId: string): Promise<void>;
applySnapshot(
agent: ManagedAgent,
options?: { title?: string | null; internal?: boolean },
): Promise<void>;
setTitle(agentId: string, title: string): Promise<void>;
flush(): Promise<void>;
}
@@ -7,6 +7,7 @@ import type { Logger } from "pino";
import { AgentStatusSchema } from "../messages.js";
import { toStoredAgentRecord } from "./agent-projections.js";
import type { ManagedAgent } from "./agent-manager.js";
import type { AgentSnapshotStore } from "./agent-snapshot-store.js";
import type { AgentSessionConfig } from "./agent-sdk-types.js";
const SERIALIZABLE_CONFIG_SCHEMA = z
@@ -69,8 +70,11 @@ export type SerializableAgentConfig = Pick<
>;
export type StoredAgentRecord = z.infer<typeof STORED_AGENT_SCHEMA>;
export function parseStoredAgentRecord(value: unknown): StoredAgentRecord {
return STORED_AGENT_SCHEMA.parse(value);
}
export class AgentStorage {
export class AgentStorage implements AgentSnapshotStore {
private cache: Map<string, StoredAgentRecord> = new Map();
private pathById: Map<string, string> = new Map();
private pathsById: Map<string, Set<string>> = new Map();
@@ -300,7 +304,7 @@ export class AgentStorage {
try {
const content = await fs.readFile(filePath, "utf8");
const parsed = JSON.parse(content);
return STORED_AGENT_SCHEMA.parse(parsed);
return parseStoredAgentRecord(parsed);
} catch (error) {
this.logger.error({ err: error, filePath }, "Skipping invalid agent record");
return null;
@@ -6,11 +6,11 @@ import { tmpdir } from "node:os";
import { createTestLogger } from "../../test-utils/test-logger.js";
import { createAgentMcpServer } from "./mcp-server.js";
import type { AgentManager, ManagedAgent } from "./agent-manager.js";
import type { AgentStorage } from "./agent-storage.js";
import type { AgentSnapshotStore } from "./agent-snapshot-store.js";
type TestDeps = {
agentManager: AgentManager;
agentStorage: AgentStorage;
agentStorage: AgentSnapshotStore;
spies: {
agentManager: Record<string, any>;
agentStorage: Record<string, any>;
@@ -41,7 +41,7 @@ function createTestDeps(): TestDeps {
return {
agentManager: agentManagerSpies as unknown as AgentManager,
agentStorage: agentStorageSpies as unknown as AgentStorage,
agentStorage: agentStorageSpies as unknown as AgentSnapshotStore,
spies: {
agentManager: agentManagerSpies,
agentStorage: agentStorageSpies,
@@ -16,7 +16,7 @@ import {
import { toAgentPayload } from "./agent-projections.js";
import { curateAgentActivity } from "./activity-curator.js";
import { AGENT_PROVIDER_DEFINITIONS } from "./provider-registry.js";
import { AgentStorage } from "./agent-storage.js";
import type { AgentSnapshotStore } from "./agent-snapshot-store.js";
import {
appendTimelineItemIfAgentKnown,
emitLiveTimelineItemIfAgentKnown,
@@ -31,7 +31,7 @@ import { createAgentWorktree, runAsyncWorktreeBootstrap } from "../worktree-boot
export interface AgentMcpServerOptions {
agentManager: AgentManager;
agentStorage: AgentStorage;
agentStorage: AgentSnapshotStore;
terminalManager?: TerminalManager | null;
paseoHome?: string;
/**
@@ -240,7 +240,7 @@ function sanitizePermissionRequest(
}
async function resolveAgentTitle(
agentStorage: AgentStorage,
agentStorage: AgentSnapshotStore,
agentId: string,
logger: Logger,
): Promise<string | null> {
@@ -254,7 +254,7 @@ async function resolveAgentTitle(
}
async function serializeSnapshotWithMetadata(
agentStorage: AgentStorage,
agentStorage: AgentSnapshotStore,
snapshot: ManagedAgent,
logger: Logger,
) {
@@ -10,7 +10,7 @@ import { createPaseoDaemon, parseListenString, type PaseoDaemonConfig } from "./
import { createTestPaseoDaemon } from "./test-utils/paseo-daemon.js";
import { createTestAgentClients } from "./test-utils/fake-agent-client.js";
import { openPaseoDatabase } from "./db/pglite-database.js";
import { projects, workspaces } from "./db/schema.js";
import { agentSnapshots, projects, workspaces } from "./db/schema.js";
describe("paseo daemon bootstrap", () => {
afterEach(() => {
@@ -340,6 +340,14 @@ describe("paseo daemon bootstrap", () => {
const database = await openPaseoDatabase(path.join(config.paseoHome, "db"));
try {
expect(await database.db.select().from(agentSnapshots)).toEqual([
expect.objectContaining({
agentId: "agent-1",
cwd: "/tmp/db-only-project",
requiresAttention: false,
internal: false,
}),
]);
expect(await database.db.select().from(projects)).toHaveLength(1);
expect(await database.db.select().from(workspaces)).toHaveLength(1);
} finally {
+13 -5
View File
@@ -84,12 +84,14 @@ import type { LocalSpeechProviderConfig } from "./speech/providers/local/config.
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 type { AgentSnapshotStore } from "./agent/agent-snapshot-store.js";
import { createAgentMcpServer } from "./agent/mcp-server.js";
import { createAllClients, shutdownProviders } from "./agent/provider-registry.js";
import { bootstrapWorkspaceRegistries } from "./workspace-registry-bootstrap.js";
import { DbAgentSnapshotStore } from "./db/db-agent-snapshot-store.js";
import { DbProjectRegistry } from "./db/db-project-registry.js";
import { DbWorkspaceRegistry } from "./db/db-workspace-registry.js";
import { importLegacyAgentSnapshots } from "./db/legacy-agent-snapshot-import.js";
import { importLegacyProjectWorkspaceJson } from "./db/legacy-project-workspace-import.js";
import { openPaseoDatabase, type PaseoDatabaseHandle } from "./db/pglite-database.js";
import { createTerminalManager, type TerminalManager } from "../terminal/terminal-manager.js";
@@ -181,7 +183,7 @@ export type PaseoDaemonConfig = {
export interface PaseoDaemon {
config: PaseoDaemonConfig;
agentManager: AgentManager;
agentStorage: AgentStorage;
agentStorage: AgentSnapshotStore;
terminalManager: TerminalManager;
start(): Promise<void>;
stop(): Promise<void>;
@@ -359,7 +361,9 @@ export async function createPaseoDaemon(
const httpServer = createHTTPServer(app);
const agentStorage = new AgentStorage(config.agentStoragePath, logger);
database = await openPaseoDatabase(path.join(config.paseoHome, "db"));
logger.info({ elapsed: elapsed() }, "Paseo database opened");
const agentStorage = new DbAgentSnapshotStore(database.db);
const agentManager = new AgentManager({
clients: {
...createAllClients(logger, {
@@ -374,8 +378,6 @@ export async function createPaseoDaemon(
const terminalManager = createTerminalManager();
await agentStorage.initialize();
logger.info({ elapsed: elapsed() }, "Agent storage initialized");
database = await openPaseoDatabase(path.join(config.paseoHome, "db"));
logger.info({ elapsed: elapsed() }, "Paseo database opened");
const projectRegistry = new DbProjectRegistry(database.db);
const workspaceRegistry = new DbWorkspaceRegistry(database.db);
await importLegacyProjectWorkspaceJson({
@@ -384,6 +386,12 @@ export async function createPaseoDaemon(
logger,
});
logger.info({ elapsed: elapsed() }, "Legacy project/workspace import checked");
await importLegacyAgentSnapshots({
db: database.db,
paseoHome: config.paseoHome,
logger,
});
logger.info({ elapsed: elapsed() }, "Legacy agent snapshot import checked");
await bootstrapWorkspaceRegistries({
paseoHome: config.paseoHome,
agentStorage,
@@ -0,0 +1,280 @@
import os from "node:os";
import path from "node:path";
import { mkdtempSync, rmSync } from "node:fs";
import { afterEach, beforeEach, describe, expect, test } from "vitest";
import type { ManagedAgent } from "../agent/agent-manager.js";
import type {
AgentPermissionRequest,
AgentSession,
AgentSessionConfig,
} from "../agent/agent-sdk-types.js";
import type { StoredAgentRecord } from "../agent/agent-storage.js";
import { openPaseoDatabase, type PaseoDatabaseHandle } from "./pglite-database.js";
import { DbAgentSnapshotStore } from "./db-agent-snapshot-store.js";
import { agentSnapshots, projects, workspaces } from "./schema.js";
type ManagedAgentOverrides = Omit<
Partial<ManagedAgent>,
"config" | "pendingPermissions" | "session" | "activeForegroundTurnId"
> & {
config?: Partial<AgentSessionConfig>;
pendingPermissions?: Map<string, AgentPermissionRequest>;
session?: AgentSession | null;
activeForegroundTurnId?: string | null;
runtimeInfo?: ManagedAgent["runtimeInfo"];
attention?: ManagedAgent["attention"];
};
function createManagedAgent(overrides: ManagedAgentOverrides = {}): ManagedAgent {
const now = overrides.updatedAt ?? new Date("2026-03-01T00:00:00.000Z");
const provider = overrides.provider ?? "codex";
const cwd = overrides.cwd ?? "/tmp/project";
const lifecycle = overrides.lifecycle ?? "idle";
const configOverrides = overrides.config ?? {};
const config: AgentSessionConfig = {
provider,
cwd,
title: configOverrides.title,
modeId: configOverrides.modeId ?? "plan",
model: configOverrides.model ?? "gpt-5.1-codex-mini",
extra: configOverrides.extra ?? { codex: { approvalPolicy: "on-request" } },
systemPrompt: configOverrides.systemPrompt,
mcpServers: configOverrides.mcpServers,
};
const session = lifecycle === "closed" ? null : (overrides.session ?? ({} as AgentSession));
const activeForegroundTurnId =
overrides.activeForegroundTurnId ?? (lifecycle === "running" ? "turn-1" : null);
return {
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 ?? [],
attention: overrides.attention ?? { requiresAttention: false },
runtimeInfo: overrides.runtimeInfo ?? {
provider,
sessionId: overrides.sessionId ?? "session-123",
model: config.model ?? null,
modeId: config.modeId ?? null,
},
persistence: overrides.persistence ?? null,
historyPrimed: overrides.historyPrimed ?? true,
lastUserMessageAt: overrides.lastUserMessageAt ?? now,
lastUsage: overrides.lastUsage,
lastError: overrides.lastError,
internal: overrides.internal,
labels: overrides.labels ?? {},
pendingReplacement: false,
provisionalAssistantText: null,
};
}
function createStoredAgentRecord(overrides: Partial<StoredAgentRecord> = {}): StoredAgentRecord {
return {
id: "agent-1",
provider: "codex",
cwd: "/tmp/project",
createdAt: "2026-03-01T00:00:00.000Z",
updatedAt: "2026-03-01T00:00:00.000Z",
lastActivityAt: "2026-03-01T00:00:00.000Z",
lastUserMessageAt: null,
title: null,
labels: {},
lastStatus: "idle",
lastModeId: "plan",
config: {
modeId: "plan",
model: "gpt-5.1-codex-mini",
},
runtimeInfo: {
provider: "codex",
sessionId: "session-123",
model: "gpt-5.1-codex-mini",
modeId: "plan",
},
persistence: {
provider: "codex",
sessionId: "session-123",
},
requiresAttention: false,
attentionReason: null,
attentionTimestamp: null,
internal: false,
archivedAt: null,
...overrides,
};
}
describe("DbAgentSnapshotStore", () => {
let tmpDir: string;
let dataDir: string;
let database: PaseoDatabaseHandle;
let store: DbAgentSnapshotStore;
beforeEach(async () => {
tmpDir = mkdtempSync(path.join(os.tmpdir(), "db-agent-snapshot-store-"));
dataDir = path.join(tmpDir, "db");
database = await openPaseoDatabase(dataDir);
store = new DbAgentSnapshotStore(database.db);
await store.initialize();
});
afterEach(async () => {
await database.close();
rmSync(tmpDir, { recursive: true, force: true });
});
test("supports list/get/upsert/remove CRUD lifecycle", async () => {
await seedWorkspace(database, { workspaceId: "/tmp/project", projectId: "project-1" });
const record = createStoredAgentRecord();
expect(await store.list()).toEqual([]);
expect(await store.get(record.id)).toBeNull();
await store.upsert(record);
expect(await store.get(record.id)).toEqual(record);
expect(await store.list()).toEqual([record]);
expect(await database.db.select().from(agentSnapshots)).toEqual([
expect.objectContaining({
agentId: "agent-1",
workspaceId: "/tmp/project",
requiresAttention: false,
internal: false,
}),
]);
await store.remove(record.id);
expect(await store.get(record.id)).toBeNull();
expect(await store.list()).toEqual([]);
});
test("applySnapshot preserves title, createdAt, and archivedAt across updates", async () => {
await store.upsert(
createStoredAgentRecord({
id: "agent-apply",
title: "Pinned title",
createdAt: "2026-03-01T00:00:00.000Z",
archivedAt: "2026-03-05T00:00:00.000Z",
}),
);
await store.applySnapshot(
createManagedAgent({
id: "agent-apply",
createdAt: new Date("2026-03-10T00:00:00.000Z"),
updatedAt: new Date("2026-03-11T00:00:00.000Z"),
lifecycle: "running",
}),
);
expect(await store.get("agent-apply")).toEqual(
expect.objectContaining({
id: "agent-apply",
title: "Pinned title",
createdAt: "2026-03-01T00:00:00.000Z",
updatedAt: "2026-03-11T00:00:00.000Z",
archivedAt: "2026-03-05T00:00:00.000Z",
lastStatus: "running",
}),
);
});
test("setTitle throws for missing agents and updates existing agents", async () => {
await expect(store.setTitle("missing-agent", "Missing")).rejects.toThrow(
"Agent missing-agent not found",
);
await store.upsert(createStoredAgentRecord({ id: "agent-title", title: null }));
await store.setTitle("agent-title", "Renamed agent");
expect(await store.get("agent-title")).toEqual(
expect.objectContaining({
id: "agent-title",
title: "Renamed agent",
}),
);
});
test("beginDelete and flush are safe no-ops", async () => {
await store.upsert(createStoredAgentRecord({ id: "agent-noop" }));
expect(() => store.beginDelete("agent-noop")).not.toThrow();
await expect(store.flush()).resolves.toBeUndefined();
expect(await store.get("agent-noop")).toEqual(
expect.objectContaining({
id: "agent-noop",
}),
);
});
test("upsert is idempotent for the same agent ID", async () => {
await store.upsert(createStoredAgentRecord({ id: "agent-idempotent", title: "Initial" }));
await store.upsert(
createStoredAgentRecord({
id: "agent-idempotent",
title: "Updated",
updatedAt: "2026-03-02T00:00:00.000Z",
lastStatus: "running",
}),
);
expect(await store.list()).toEqual([
createStoredAgentRecord({
id: "agent-idempotent",
title: "Updated",
updatedAt: "2026-03-02T00:00:00.000Z",
lastStatus: "running",
}),
]);
expect(await database.db.select().from(agentSnapshots)).toHaveLength(1);
});
});
async function seedWorkspace(
database: PaseoDatabaseHandle,
options: { workspaceId: string; projectId: string },
): Promise<void> {
await database.db.insert(projects).values({
projectId: options.projectId,
rootPath: options.workspaceId,
kind: "git",
displayName: options.projectId,
createdAt: "2026-03-01T00:00:00.000Z",
updatedAt: "2026-03-01T00:00:00.000Z",
archivedAt: null,
});
await database.db.insert(workspaces).values({
workspaceId: options.workspaceId,
projectId: options.projectId,
cwd: options.workspaceId,
kind: "local_checkout",
displayName: "main",
createdAt: "2026-03-01T00:00:00.000Z",
updatedAt: "2026-03-01T00:00:00.000Z",
archivedAt: null,
});
}
@@ -0,0 +1,184 @@
import { asc, eq } from "drizzle-orm";
import type { ManagedAgent } from "../agent/agent-manager.js";
import type { AgentSnapshotStore } from "../agent/agent-snapshot-store.js";
import { toStoredAgentRecord } from "../agent/agent-projections.js";
import type { StoredAgentRecord } from "../agent/agent-storage.js";
import { normalizeWorkspaceId } from "../workspace-registry-model.js";
import type { PaseoDatabaseHandle } from "./pglite-database.js";
import { agentSnapshots, workspaces } from "./schema.js";
type AgentSnapshotRow = typeof agentSnapshots.$inferSelect;
type AgentSnapshotInsert = typeof agentSnapshots.$inferInsert;
export function toStoredAgentRecordFromRow(row: AgentSnapshotRow): StoredAgentRecord {
return {
id: row.agentId,
provider: row.provider,
cwd: row.cwd,
createdAt: row.createdAt,
updatedAt: row.updatedAt,
lastActivityAt: row.lastActivityAt ?? undefined,
lastUserMessageAt: row.lastUserMessageAt ?? null,
title: row.title ?? null,
labels: row.labels,
lastStatus: row.lastStatus as StoredAgentRecord["lastStatus"],
lastModeId: row.lastModeId ?? null,
config: row.config ?? null,
runtimeInfo: row.runtimeInfo ?? undefined,
persistence: row.persistence ?? null,
requiresAttention: row.requiresAttention,
attentionReason: (row.attentionReason ?? null) as StoredAgentRecord["attentionReason"],
attentionTimestamp: row.attentionTimestamp ?? null,
internal: row.internal,
archivedAt: row.archivedAt ?? null,
};
}
export function toAgentSnapshotRowValues(options: {
record: StoredAgentRecord;
workspaceId: string | null;
}): AgentSnapshotInsert {
const { record, workspaceId } = options;
return {
agentId: record.id,
provider: record.provider,
workspaceId,
cwd: record.cwd,
createdAt: record.createdAt,
updatedAt: record.updatedAt,
lastActivityAt: record.lastActivityAt ?? null,
lastUserMessageAt: record.lastUserMessageAt ?? null,
title: record.title ?? null,
labels: record.labels,
lastStatus: record.lastStatus,
lastModeId: record.lastModeId ?? null,
config: record.config ?? null,
runtimeInfo: record.runtimeInfo ?? null,
persistence: record.persistence ?? null,
requiresAttention: record.requiresAttention ?? false,
attentionReason: record.attentionReason ?? null,
attentionTimestamp: record.attentionTimestamp ?? null,
internal: record.internal ?? false,
archivedAt: record.archivedAt ?? null,
};
}
function toAgentSnapshotUpdateSet(values: AgentSnapshotInsert) {
return {
provider: values.provider,
workspaceId: values.workspaceId,
cwd: values.cwd,
createdAt: values.createdAt,
updatedAt: values.updatedAt,
lastActivityAt: values.lastActivityAt,
lastUserMessageAt: values.lastUserMessageAt,
title: values.title,
labels: values.labels,
lastStatus: values.lastStatus,
lastModeId: values.lastModeId,
config: values.config,
runtimeInfo: values.runtimeInfo,
persistence: values.persistence,
requiresAttention: values.requiresAttention,
attentionReason: values.attentionReason,
attentionTimestamp: values.attentionTimestamp,
internal: values.internal,
archivedAt: values.archivedAt,
} satisfies Omit<AgentSnapshotInsert, "agentId">;
}
export class DbAgentSnapshotStore implements AgentSnapshotStore {
private readonly db: PaseoDatabaseHandle["db"];
constructor(db: PaseoDatabaseHandle["db"]) {
this.db = db;
}
async initialize(): Promise<void> {
return Promise.resolve();
}
async list(): Promise<StoredAgentRecord[]> {
const rows = await this.db
.select()
.from(agentSnapshots)
.orderBy(asc(agentSnapshots.createdAt), asc(agentSnapshots.agentId));
return rows.map(toStoredAgentRecordFromRow);
}
async get(agentId: string): Promise<StoredAgentRecord | null> {
const rows = await this.db
.select()
.from(agentSnapshots)
.where(eq(agentSnapshots.agentId, agentId))
.limit(1);
const row = rows[0];
return row ? toStoredAgentRecordFromRow(row) : null;
}
async upsert(record: StoredAgentRecord): Promise<void> {
const values = toAgentSnapshotRowValues({
record,
workspaceId: await this.resolveWorkspaceId(record.cwd),
});
await this.db
.insert(agentSnapshots)
.values(values)
.onConflictDoUpdate({
target: agentSnapshots.agentId,
set: toAgentSnapshotUpdateSet(values),
});
}
beginDelete(_agentId: string): void {}
async remove(agentId: string): Promise<void> {
await this.db.delete(agentSnapshots).where(eq(agentSnapshots.agentId, agentId));
}
async applySnapshot(
agent: ManagedAgent,
options?: { title?: string | null; internal?: boolean },
): Promise<void> {
const existing = await this.get(agent.id);
const hasTitleOverride =
options !== undefined && Object.prototype.hasOwnProperty.call(options, "title");
const hasInternalOverride =
options !== undefined && Object.prototype.hasOwnProperty.call(options, "internal");
const record = toStoredAgentRecord(agent, {
title: hasTitleOverride ? (options?.title ?? null) : (existing?.title ?? null),
createdAt: existing?.createdAt,
internal: hasInternalOverride ? options?.internal : (agent.internal ?? existing?.internal),
});
if (existing && existing.archivedAt !== undefined) {
record.archivedAt = existing.archivedAt;
}
await this.upsert(record);
}
async setTitle(agentId: string, title: string): Promise<void> {
const record = await this.get(agentId);
if (!record) {
throw new Error(`Agent ${agentId} not found`);
}
await this.upsert({ ...record, title });
}
async flush(): Promise<void> {
return Promise.resolve();
}
private async resolveWorkspaceId(cwd: string): Promise<string | null> {
const workspaceId = normalizeWorkspaceId(cwd);
const rows = await this.db
.select({ workspaceId: workspaces.workspaceId })
.from(workspaces)
.where(eq(workspaces.workspaceId, workspaceId))
.limit(1);
return rows[0]?.workspaceId ?? null;
}
}
@@ -0,0 +1,173 @@
import os from "node:os";
import path from "node:path";
import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs";
import { afterEach, beforeEach, describe, expect, test } from "vitest";
import { createTestLogger } from "../../test-utils/test-logger.js";
import { openPaseoDatabase, type PaseoDatabaseHandle } from "./pglite-database.js";
import { importLegacyAgentSnapshots } from "./legacy-agent-snapshot-import.js";
import { agentSnapshots } from "./schema.js";
describe("importLegacyAgentSnapshots", () => {
let tmpDir: string;
let paseoHome: string;
let dbDir: string;
let database: PaseoDatabaseHandle;
beforeEach(async () => {
tmpDir = mkdtempSync(path.join(os.tmpdir(), "paseo-legacy-agent-import-"));
paseoHome = path.join(tmpDir, ".paseo");
dbDir = path.join(paseoHome, "db");
mkdirSync(paseoHome, { recursive: true });
database = await openPaseoDatabase(dbDir);
});
afterEach(async () => {
await database.close();
rmSync(tmpDir, { recursive: true, force: true });
});
test("imports agent JSON files when the DB is empty", async () => {
writeLegacyAgentJson({
paseoHome,
relativePath: "agents/agent-1.json",
payload: createLegacyAgentJson({
requiresAttention: undefined,
internal: undefined,
}),
});
const result = await importLegacyAgentSnapshots({
db: database.db,
paseoHome,
logger: createTestLogger(),
});
expect(result).toEqual({
status: "imported",
importedAgents: 1,
});
expect(await database.db.select().from(agentSnapshots)).toEqual([
expect.objectContaining({
agentId: "agent-1",
cwd: "/tmp/project",
requiresAttention: false,
internal: false,
}),
]);
});
test("skips import when the DB already has agent data", async () => {
await database.db.insert(agentSnapshots).values({
agentId: "existing-agent",
provider: "codex",
workspaceId: null,
cwd: "/tmp/existing-project",
createdAt: "2026-03-01T00:00:00.000Z",
updatedAt: "2026-03-01T00:00:00.000Z",
lastActivityAt: "2026-03-01T00:00:00.000Z",
lastUserMessageAt: null,
title: null,
labels: {},
lastStatus: "idle",
lastModeId: "plan",
config: null,
runtimeInfo: { provider: "codex", sessionId: "session-existing" },
persistence: null,
requiresAttention: false,
attentionReason: null,
attentionTimestamp: null,
internal: false,
archivedAt: null,
});
writeLegacyAgentJson({
paseoHome,
relativePath: "agents/legacy-agent.json",
payload: createLegacyAgentJson({ id: "legacy-agent" }),
});
const result = await importLegacyAgentSnapshots({
db: database.db,
paseoHome,
logger: createTestLogger(),
});
expect(result).toEqual({
status: "skipped",
reason: "database-not-empty",
});
expect(await database.db.select().from(agentSnapshots)).toHaveLength(1);
});
test("imports agent JSON files from nested project directories", async () => {
writeLegacyAgentJson({
paseoHome,
relativePath: "agents/agent-root.json",
payload: createLegacyAgentJson({ id: "agent-root", cwd: "/tmp/root-project" }),
});
writeLegacyAgentJson({
paseoHome,
relativePath: "agents/tmp-nested-project/agent-nested.json",
payload: createLegacyAgentJson({ id: "agent-nested", cwd: "/tmp/nested-project" }),
});
const result = await importLegacyAgentSnapshots({
db: database.db,
paseoHome,
logger: createTestLogger(),
});
expect(result).toEqual({
status: "imported",
importedAgents: 2,
});
expect(
(await database.db.select().from(agentSnapshots)).map((row) => row.agentId).sort(),
).toEqual(["agent-nested", "agent-root"]);
});
});
function createLegacyAgentJson(overrides: Record<string, unknown> = {}): Record<string, unknown> {
return {
id: "agent-1",
provider: "codex",
cwd: "/tmp/project",
createdAt: "2026-03-01T00:00:00.000Z",
updatedAt: "2026-03-02T00:00:00.000Z",
lastActivityAt: "2026-03-02T00:00:00.000Z",
lastUserMessageAt: null,
title: null,
labels: {},
lastStatus: "idle",
lastModeId: "plan",
config: {
model: "gpt-5.1-codex-mini",
modeId: "plan",
},
runtimeInfo: {
provider: "codex",
sessionId: "session-123",
model: "gpt-5.1-codex-mini",
modeId: "plan",
},
persistence: null,
attentionReason: null,
attentionTimestamp: null,
archivedAt: null,
...overrides,
};
}
function writeLegacyAgentJson(input: {
paseoHome: string;
relativePath: string;
payload: Record<string, unknown>;
}): void {
const absolutePath = path.join(input.paseoHome, input.relativePath);
mkdirSync(path.dirname(absolutePath), { recursive: true });
writeFileSync(absolutePath, JSON.stringify(input.payload, null, 2), {
encoding: "utf8",
flag: "w",
});
}
@@ -0,0 +1,128 @@
import path from "node:path";
import { promises as fs } from "node:fs";
import { count } from "drizzle-orm";
import type { Logger } from "pino";
import { parseStoredAgentRecord, type StoredAgentRecord } from "../agent/agent-storage.js";
import { normalizeWorkspaceId } from "../workspace-registry-model.js";
import type { PaseoDatabaseHandle } from "./pglite-database.js";
import { toAgentSnapshotRowValues } from "./db-agent-snapshot-store.js";
import { agentSnapshots, workspaces } from "./schema.js";
export type LegacyAgentSnapshotImportResult =
| {
status: "imported";
importedAgents: number;
}
| {
status: "skipped";
reason: "database-not-empty" | "no-legacy-files";
};
export async function importLegacyAgentSnapshots(options: {
db: PaseoDatabaseHandle["db"];
paseoHome: string;
logger: Logger;
}): Promise<LegacyAgentSnapshotImportResult> {
if (await hasAnyAgentSnapshotRows(options.db)) {
options.logger.info("Skipping legacy agent snapshot import because the DB is not empty");
return {
status: "skipped",
reason: "database-not-empty",
};
}
const records = await readLegacyAgentRecords(path.join(options.paseoHome, "agents"), options.logger);
if (records.length === 0) {
options.logger.info("Skipping legacy agent snapshot import because no legacy files exist");
return {
status: "skipped",
reason: "no-legacy-files",
};
}
await options.db.transaction(async (tx) => {
const workspaceRows = await tx.select({ workspaceId: workspaces.workspaceId }).from(workspaces);
const workspaceIds = new Set(workspaceRows.map((row) => row.workspaceId));
const rows = records.map((record) => {
const workspaceId = normalizeWorkspaceId(record.cwd);
return toAgentSnapshotRowValues({
record,
workspaceId: workspaceIds.has(workspaceId) ? workspaceId : null,
});
});
await tx.insert(agentSnapshots).values(rows);
});
options.logger.info(
{ importedAgents: records.length },
"Imported legacy agent snapshots into the database",
);
return {
status: "imported",
importedAgents: records.length,
};
}
async function readLegacyAgentRecords(baseDir: string, logger: Logger): Promise<StoredAgentRecord[]> {
let entries: Array<import("node:fs").Dirent> = [];
try {
entries = await fs.readdir(baseDir, { withFileTypes: true });
} catch (error) {
if ((error as NodeJS.ErrnoException).code === "ENOENT") {
return [];
}
throw error;
}
const recordsById = new Map<string, StoredAgentRecord>();
for (const entry of entries) {
if (entry.isFile() && entry.name.endsWith(".json")) {
const record = await readRecordFile(path.join(baseDir, entry.name), logger);
if (record) {
recordsById.set(record.id, record);
}
continue;
}
if (!entry.isDirectory()) {
continue;
}
let childEntries: Array<import("node:fs").Dirent> = [];
try {
childEntries = await fs.readdir(path.join(baseDir, entry.name), { withFileTypes: true });
} catch {
continue;
}
for (const childEntry of childEntries) {
if (!childEntry.isFile() || !childEntry.name.endsWith(".json")) {
continue;
}
const record = await readRecordFile(path.join(baseDir, entry.name, childEntry.name), logger);
if (record) {
recordsById.set(record.id, record);
}
}
}
return Array.from(recordsById.values());
}
async function readRecordFile(filePath: string, logger: Logger): Promise<StoredAgentRecord | null> {
try {
const raw = await fs.readFile(filePath, "utf8");
return parseStoredAgentRecord(JSON.parse(raw));
} catch (error) {
logger.error({ err: error, filePath }, "Skipping invalid legacy agent snapshot");
return null;
}
}
async function hasAnyAgentSnapshotRows(db: PaseoDatabaseHandle["db"]): Promise<boolean> {
const rows = await db.select({ count: count() }).from(agentSnapshots);
return (rows[0]?.count ?? 0) > 0;
}
@@ -3,7 +3,7 @@ import type pino from "pino";
import type { ManagedAgent } from "./agent/agent-manager.js";
import type { AgentManager } from "./agent/agent-manager.js";
import type { AgentPersistenceHandle, AgentSessionConfig } from "./agent/agent-sdk-types.js";
import type { AgentStorage } from "./agent/agent-storage.js";
import type { AgentSnapshotStore } from "./agent/agent-snapshot-store.js";
import {
buildConfigOverrides,
buildSessionConfig,
@@ -22,7 +22,7 @@ export type ProviderHistoryCompatibilityServiceOptions = {
| "reloadAgentSession"
| "resumeAgentFromPersistence"
>;
agentStorage: Pick<AgentStorage, "get">;
agentStorage: Pick<AgentSnapshotStore, "get">;
logger: pino.Logger;
};
+4 -3
View File
@@ -93,7 +93,8 @@ import type {
AgentProvider,
AgentPersistenceHandle,
} from "./agent/agent-sdk-types.js";
import { AgentStorage, type StoredAgentRecord } from "./agent/agent-storage.js";
import type { StoredAgentRecord } from "./agent/agent-storage.js";
import type { AgentSnapshotStore } from "./agent/agent-snapshot-store.js";
import { isValidAgentProvider, AGENT_PROVIDER_IDS } from "./agent/provider-manifest.js";
import {
buildProjectPlacementForCwd,
@@ -388,7 +389,7 @@ export type SessionOptions = {
pushTokenStore: PushTokenStore;
paseoHome: string;
agentManager: AgentManager;
agentStorage: AgentStorage;
agentStorage: AgentSnapshotStore;
projectRegistry: ProjectRegistry;
workspaceRegistry: WorkspaceRegistry;
workspaceReconciliationService?: WorkspaceReconciliationService;
@@ -549,7 +550,7 @@ export class Session {
private agentMcpClient: Awaited<ReturnType<typeof experimental_createMCPClient>> | null = null;
private agentTools: ToolSet | null = null;
private agentManager: AgentManager;
private readonly agentStorage: AgentStorage;
private readonly agentStorage: AgentSnapshotStore;
private readonly projectRegistry: ProjectRegistry;
private readonly workspaceRegistry: WorkspaceRegistry;
private readonly workspaceReconciliationService: WorkspaceReconciliationService;
@@ -4,7 +4,7 @@ import type { Transport } from "@modelcontextprotocol/sdk/shared/transport.js";
import { join } from "path";
import { hostname as getHostname } from "node:os";
import type { AgentManager } from "./agent/agent-manager.js";
import type { AgentStorage } from "./agent/agent-storage.js";
import type { AgentSnapshotStore } from "./agent/agent-snapshot-store.js";
import type { DownloadTokenStore } from "./file-download/token-store.js";
import type { TerminalManager } from "../terminal/terminal-manager.js";
import type pino from "pino";
@@ -223,7 +223,7 @@ export class VoiceAssistantWebSocketServer {
private readonly serverId: string;
private readonly daemonVersion: string;
private readonly agentManager: AgentManager;
private readonly agentStorage: AgentStorage;
private readonly agentStorage: AgentSnapshotStore;
private readonly projectRegistry: ProjectRegistry;
private readonly workspaceRegistry: WorkspaceRegistry;
private readonly downloadTokenStore: DownloadTokenStore;
@@ -281,7 +281,7 @@ export class VoiceAssistantWebSocketServer {
logger: pino.Logger,
serverId: string,
agentManager: AgentManager,
agentStorage: AgentStorage,
agentStorage: AgentSnapshotStore,
downloadTokenStore: DownloadTokenStore,
paseoHome: string,
createAgentMcpTransport: AgentMcpTransportFactory,
@@ -1,7 +1,7 @@
import { stat } from "node:fs/promises";
import type { ProjectPlacementPayload } from "./messages.js";
import type { AgentStorage } from "./agent/agent-storage.js";
import type { AgentSnapshotStore } from "./agent/agent-snapshot-store.js";
import {
detectStaleWorkspaces,
deriveProjectKind,
@@ -26,7 +26,7 @@ type Awaitable<T> = T | Promise<T>;
export type WorkspaceReconciliationServiceOptions = {
projectRegistry: ProjectRegistry;
workspaceRegistry: WorkspaceRegistry;
agentStorage: Pick<AgentStorage, "list">;
agentStorage: Pick<AgentSnapshotStore, "list">;
buildProjectPlacement: (cwd: string) => Promise<ProjectPlacementPayload>;
syncWorkspaceGitWatchTarget?: (
cwd: string,
@@ -45,7 +45,7 @@ export type ReconcileWorkspaceRecordResult = {
export class WorkspaceReconciliationService {
private readonly projectRegistry: ProjectRegistry;
private readonly workspaceRegistry: WorkspaceRegistry;
private readonly agentStorage: Pick<AgentStorage, "list">;
private readonly agentStorage: Pick<AgentSnapshotStore, "list">;
private readonly buildProjectPlacement: (cwd: string) => Promise<ProjectPlacementPayload>;
private readonly syncWorkspaceGitWatchTarget: (
cwd: string,
@@ -3,7 +3,7 @@ import path from "node:path";
import type { Logger } from "pino";
import type { StoredAgentRecord } from "./agent/agent-storage.js";
import type { AgentStorage } from "./agent/agent-storage.js";
import type { AgentSnapshotStore } from "./agent/agent-snapshot-store.js";
import {
buildProjectPlacementForCwd,
deriveProjectKind,
@@ -49,7 +49,7 @@ function resolveAgentUpdatedAt(record: StoredAgentRecord): string {
export async function bootstrapWorkspaceRegistries(options: {
paseoHome: string;
agentStorage: AgentStorage;
agentStorage: Pick<AgentSnapshotStore, "list">;
projectRegistry: ProjectRegistry;
workspaceRegistry: WorkspaceRegistry;
logger: Logger;