mirror of
https://github.com/getpaseo/paseo.git
synced 2026-10-30 05:02:58 +00:00
Phase 5: remove legacy storage paths and dead compatibility code
This commit is contained in:
+8
-12
@@ -13,7 +13,7 @@ import {
|
||||
|
||||
const pendingAgentBootstrapLoads = new Map<string, Promise<ManagedAgent>>();
|
||||
|
||||
export type ProviderHistoryCompatibilityServiceOptions = {
|
||||
export type AgentLoadingServiceOptions = {
|
||||
agentManager: Pick<
|
||||
AgentManager,
|
||||
| "createAgent"
|
||||
@@ -25,17 +25,16 @@ export type ProviderHistoryCompatibilityServiceOptions = {
|
||||
logger: pino.Logger;
|
||||
};
|
||||
|
||||
// Compatibility-only bridge for runtime paths that still need coordinated cold
|
||||
// load, explicit resume, or refresh behavior.
|
||||
export class ProviderHistoryCompatibilityService {
|
||||
private readonly agentManager: ProviderHistoryCompatibilityServiceOptions["agentManager"];
|
||||
private readonly agentStorage: ProviderHistoryCompatibilityServiceOptions["agentStorage"];
|
||||
// Coordinates cold loads, explicit resumes, and refreshes for persisted agents.
|
||||
export class AgentLoadingService {
|
||||
private readonly agentManager: AgentLoadingServiceOptions["agentManager"];
|
||||
private readonly agentStorage: AgentLoadingServiceOptions["agentStorage"];
|
||||
private readonly logger: pino.Logger;
|
||||
|
||||
constructor(options: ProviderHistoryCompatibilityServiceOptions) {
|
||||
constructor(options: AgentLoadingServiceOptions) {
|
||||
this.agentManager = options.agentManager;
|
||||
this.agentStorage = options.agentStorage;
|
||||
this.logger = options.logger.child({ component: "provider-history-compatibility" });
|
||||
this.logger = options.logger.child({ component: "agent-loading" });
|
||||
}
|
||||
|
||||
async ensureAgentLoaded(options: { agentId: string }): Promise<ManagedAgent> {
|
||||
@@ -66,10 +65,7 @@ export class ProviderHistoryCompatibilityService {
|
||||
handle: AgentPersistenceHandle;
|
||||
overrides?: Partial<AgentSessionConfig>;
|
||||
}): Promise<ManagedAgent> {
|
||||
return this.agentManager.resumeAgentFromPersistence(
|
||||
options.handle,
|
||||
options.overrides,
|
||||
);
|
||||
return this.agentManager.resumeAgentFromPersistence(options.handle, options.overrides);
|
||||
}
|
||||
|
||||
async refreshAgent(options: { agentId: string }): Promise<ManagedAgent> {
|
||||
@@ -1397,12 +1397,17 @@ export class AgentManager {
|
||||
return iterator.done ? null : iterator.value;
|
||||
}
|
||||
|
||||
/**
|
||||
* Test-only compatibility hook for managers constructed without a durable
|
||||
* timeline store. Production loads committed history from the durable store
|
||||
* during session registration instead of replaying provider history here.
|
||||
*/
|
||||
async hydrateTimelineFromProvider(agentId: string): Promise<void> {
|
||||
const agent = this.requireAgent(agentId);
|
||||
if (this.durableTimelineStore) {
|
||||
return;
|
||||
}
|
||||
await this.hydrateTimeline(agent);
|
||||
await this.hydrateTimelineFromLegacyProviderHistory(agent);
|
||||
}
|
||||
|
||||
async deleteCommittedTimeline(agentId: string): Promise<void> {
|
||||
@@ -1885,7 +1890,9 @@ export class AgentManager {
|
||||
}
|
||||
}
|
||||
|
||||
private async hydrateTimeline(agent: ActiveManagedAgent): Promise<void> {
|
||||
private async hydrateTimelineFromLegacyProviderHistory(
|
||||
agent: ActiveManagedAgent,
|
||||
): Promise<void> {
|
||||
if (agent.historyPrimed) {
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -2,16 +2,13 @@ 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>;
|
||||
}
|
||||
|
||||
@@ -379,8 +379,6 @@ export async function createPaseoDaemon(
|
||||
});
|
||||
|
||||
const terminalManager = createTerminalManager();
|
||||
await agentStorage.initialize();
|
||||
logger.info({ elapsed: elapsed() }, "Agent storage initialized");
|
||||
const projectRegistry = new DbProjectRegistry(database.db);
|
||||
const workspaceRegistry = new DbWorkspaceRegistry(database.db);
|
||||
await importLegacyProjectWorkspaceJson({
|
||||
@@ -719,7 +717,6 @@ export async function createPaseoDaemon(
|
||||
const stop = async () => {
|
||||
await closeAllAgents(logger, agentManager);
|
||||
await agentManager.flush().catch(() => undefined);
|
||||
await agentStorage.flush().catch(() => undefined);
|
||||
await shutdownProviders(logger, {
|
||||
runtimeSettings: config.agentProviderSettings,
|
||||
});
|
||||
|
||||
@@ -137,7 +137,6 @@ describe("DbAgentSnapshotStore", () => {
|
||||
dataDir = path.join(tmpDir, "db");
|
||||
database = await openPaseoDatabase(dataDir);
|
||||
store = new DbAgentSnapshotStore(database.db);
|
||||
await store.initialize();
|
||||
});
|
||||
|
||||
afterEach(async () => {
|
||||
@@ -219,18 +218,6 @@ describe("DbAgentSnapshotStore", () => {
|
||||
);
|
||||
});
|
||||
|
||||
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(
|
||||
|
||||
@@ -95,10 +95,6 @@ export class DbAgentSnapshotStore implements AgentSnapshotStore {
|
||||
this.db = db;
|
||||
}
|
||||
|
||||
async initialize(): Promise<void> {
|
||||
return Promise.resolve();
|
||||
}
|
||||
|
||||
async list(): Promise<StoredAgentRecord[]> {
|
||||
const rows = await this.db
|
||||
.select()
|
||||
@@ -132,8 +128,6 @@ export class DbAgentSnapshotStore implements AgentSnapshotStore {
|
||||
});
|
||||
}
|
||||
|
||||
beginDelete(_agentId: string): void {}
|
||||
|
||||
async remove(agentId: string): Promise<void> {
|
||||
await this.db.delete(agentSnapshots).where(eq(agentSnapshots.agentId, agentId));
|
||||
}
|
||||
@@ -168,10 +162,6 @@ export class DbAgentSnapshotStore implements AgentSnapshotStore {
|
||||
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
|
||||
|
||||
@@ -1,15 +1,15 @@
|
||||
import { readFileSync } from "node:fs";
|
||||
import { describe, expect, test } from "vitest";
|
||||
|
||||
describe("provider history compatibility boundary", () => {
|
||||
describe("agent loading boundary", () => {
|
||||
test("session runtime code does not directly hydrate provider history", () => {
|
||||
const sessionSource = readFileSync(new URL("./session.ts", import.meta.url), "utf8");
|
||||
const compatibilitySource = readFileSync(
|
||||
new URL("./provider-history-compatibility-service.ts", import.meta.url),
|
||||
const agentLoadingSource = readFileSync(
|
||||
new URL("./agent-loading-service.ts", import.meta.url),
|
||||
"utf8",
|
||||
);
|
||||
|
||||
expect(sessionSource).not.toMatch(/hydrateTimelineFromProvider\s*\(/);
|
||||
expect(compatibilitySource).not.toMatch(/hydrateTimelineFromProvider\s*\(/);
|
||||
expect(agentLoadingSource).not.toMatch(/hydrateTimelineFromProvider\s*\(/);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -4,11 +4,11 @@ import path from "node:path";
|
||||
import pino from "pino";
|
||||
import { describe, expect, test, vi } from "vitest";
|
||||
|
||||
import { AgentLoadingService } from "./agent-loading-service.js";
|
||||
import { AgentManager } from "./agent/agent-manager.js";
|
||||
import { AgentStorage, type StoredAgentRecord } from "./agent/agent-storage.js";
|
||||
import { DbAgentTimelineStore } from "./db/db-agent-timeline-store.js";
|
||||
import { openPaseoDatabase } from "./db/pglite-database.js";
|
||||
import { ProviderHistoryCompatibilityService } from "./provider-history-compatibility-service.js";
|
||||
import { createTestAgentClients } from "./test-utils/fake-agent-client.js";
|
||||
|
||||
function createStoredAgentRecord(overrides?: Partial<StoredAgentRecord>): StoredAgentRecord {
|
||||
@@ -56,7 +56,7 @@ function createCompatibilitySnapshot(overrides?: Partial<Record<string, unknown>
|
||||
};
|
||||
}
|
||||
|
||||
describe("ProviderHistoryCompatibilityService", () => {
|
||||
describe("AgentLoadingService", () => {
|
||||
test("ensureAgentLoaded seeds the live timeline from durable rows for an unloaded persisted agent", async () => {
|
||||
const workspaceRoot = mkdtempSync(path.join(os.tmpdir(), "provider-history-compat-load-"));
|
||||
const logger = pino({ level: "silent" });
|
||||
@@ -71,7 +71,7 @@ describe("ProviderHistoryCompatibilityService", () => {
|
||||
logger,
|
||||
idFactory: () => "00000000-0000-4000-8000-000000000301",
|
||||
});
|
||||
const service = new ProviderHistoryCompatibilityService({
|
||||
const service = new AgentLoadingService({
|
||||
agentManager: manager as any,
|
||||
agentStorage: storage as any,
|
||||
logger,
|
||||
@@ -118,7 +118,7 @@ describe("ProviderHistoryCompatibilityService", () => {
|
||||
logger,
|
||||
idFactory: () => "00000000-0000-4000-8000-000000000302",
|
||||
});
|
||||
const service = new ProviderHistoryCompatibilityService({
|
||||
const service = new AgentLoadingService({
|
||||
agentManager: manager as any,
|
||||
agentStorage: storage as any,
|
||||
logger,
|
||||
@@ -169,7 +169,7 @@ describe("ProviderHistoryCompatibilityService", () => {
|
||||
info: vi.fn(),
|
||||
warn: vi.fn(),
|
||||
};
|
||||
const service = new ProviderHistoryCompatibilityService({
|
||||
const service = new AgentLoadingService({
|
||||
agentManager: agentManager as any,
|
||||
agentStorage: agentStorage as any,
|
||||
logger: logger as any,
|
||||
@@ -200,7 +200,7 @@ describe("ProviderHistoryCompatibilityService", () => {
|
||||
info: vi.fn(),
|
||||
warn: vi.fn(),
|
||||
};
|
||||
const service = new ProviderHistoryCompatibilityService({
|
||||
const service = new AgentLoadingService({
|
||||
agentManager: agentManager as any,
|
||||
agentStorage: {
|
||||
get: async () => null,
|
||||
@@ -248,7 +248,7 @@ describe("ProviderHistoryCompatibilityService", () => {
|
||||
info: vi.fn(),
|
||||
warn: vi.fn(),
|
||||
};
|
||||
const service = new ProviderHistoryCompatibilityService({
|
||||
const service = new AgentLoadingService({
|
||||
agentManager: agentManager as any,
|
||||
agentStorage: {
|
||||
get: async () => null,
|
||||
@@ -278,7 +278,7 @@ describe("ProviderHistoryCompatibilityService", () => {
|
||||
info: vi.fn(),
|
||||
warn: vi.fn(),
|
||||
};
|
||||
const service = new ProviderHistoryCompatibilityService({
|
||||
const service = new AgentLoadingService({
|
||||
agentManager: agentManager as any,
|
||||
agentStorage: {
|
||||
get: async () => null,
|
||||
@@ -313,7 +313,7 @@ describe("ProviderHistoryCompatibilityService", () => {
|
||||
info: vi.fn(),
|
||||
warn: vi.fn(),
|
||||
};
|
||||
const service = new ProviderHistoryCompatibilityService({
|
||||
const service = new AgentLoadingService({
|
||||
agentManager: agentManager as any,
|
||||
agentStorage: {
|
||||
get: vi.fn(async () => record),
|
||||
@@ -352,7 +352,7 @@ describe("ProviderHistoryCompatibilityService", () => {
|
||||
});
|
||||
|
||||
test("refreshAgent preserves the unloaded no-persistence error", async () => {
|
||||
const service = new ProviderHistoryCompatibilityService({
|
||||
const service = new AgentLoadingService({
|
||||
agentManager: {
|
||||
getAgent: vi.fn(() => null),
|
||||
resumeAgentFromPersistence: vi.fn(),
|
||||
|
||||
+10
-14
@@ -39,7 +39,7 @@ function createCompatibilitySnapshot(overrides?: Partial<Record<string, unknown>
|
||||
}
|
||||
|
||||
function createSessionForOwnershipTests(options?: {
|
||||
providerHistoryCompatibilityService?: {
|
||||
agentLoadingService?: {
|
||||
ensureAgentLoaded?: (options: { agentId: string }) => Promise<any>;
|
||||
resumeAgent?: (options: {
|
||||
handle: { provider: string; sessionId: string };
|
||||
@@ -66,17 +66,13 @@ function createSessionForOwnershipTests(options?: {
|
||||
listAgents: () => [],
|
||||
getAgent: vi.fn(() => options?.loadedAgent ?? null),
|
||||
createAgent: vi.fn(async () => {
|
||||
throw new Error(
|
||||
"Session should delegate unloaded bootstrap to ProviderHistoryCompatibilityService",
|
||||
);
|
||||
throw new Error("Session should delegate unloaded bootstrap to AgentLoadingService");
|
||||
}),
|
||||
resumeAgentFromPersistence: vi.fn(async () => {
|
||||
throw new Error(
|
||||
"Session should delegate persistence resume to ProviderHistoryCompatibilityService",
|
||||
);
|
||||
throw new Error("Session should delegate persistence resume to AgentLoadingService");
|
||||
}),
|
||||
reloadAgentSession: vi.fn(async () => {
|
||||
throw new Error("Session should delegate refresh reload to ProviderHistoryCompatibilityService");
|
||||
throw new Error("Session should delegate refresh reload to AgentLoadingService");
|
||||
}),
|
||||
hydrateTimelineFromProvider: vi.fn(async () => {
|
||||
throw new Error("Session should not call hydrateTimelineFromProvider directly");
|
||||
@@ -127,7 +123,7 @@ function createSessionForOwnershipTests(options?: {
|
||||
stt: null,
|
||||
tts: null,
|
||||
terminalManager: null,
|
||||
providerHistoryCompatibilityService: options?.providerHistoryCompatibilityService,
|
||||
agentLoadingService: options?.agentLoadingService,
|
||||
} as any) as any;
|
||||
|
||||
return { session, emitted, agentManager };
|
||||
@@ -145,7 +141,7 @@ describe("provider history compatibility ownership", () => {
|
||||
timestamp: new Date("2026-03-24T00:00:01.000Z"),
|
||||
},
|
||||
],
|
||||
providerHistoryCompatibilityService: {
|
||||
agentLoadingService: {
|
||||
ensureAgentLoaded,
|
||||
},
|
||||
});
|
||||
@@ -178,7 +174,7 @@ describe("provider history compatibility ownership", () => {
|
||||
const ensureAgentLoaded = vi.fn(async () => createCompatibilitySnapshot());
|
||||
const { session, agentManager, emitted } = createSessionForOwnershipTests({
|
||||
storedRecord: createStoredAgentRecord(),
|
||||
providerHistoryCompatibilityService: {
|
||||
agentLoadingService: {
|
||||
ensureAgentLoaded,
|
||||
},
|
||||
});
|
||||
@@ -220,7 +216,7 @@ describe("provider history compatibility ownership", () => {
|
||||
test("resume_agent_request delegates persistence bootstrap through the compatibility seam", async () => {
|
||||
const resumeAgent = vi.fn(async () => createCompatibilitySnapshot());
|
||||
const { session, emitted } = createSessionForOwnershipTests({
|
||||
providerHistoryCompatibilityService: {
|
||||
agentLoadingService: {
|
||||
resumeAgent,
|
||||
},
|
||||
});
|
||||
@@ -272,7 +268,7 @@ describe("provider history compatibility ownership", () => {
|
||||
);
|
||||
const { session, emitted } = createSessionForOwnershipTests({
|
||||
loadedAgent: createCompatibilitySnapshot(),
|
||||
providerHistoryCompatibilityService: {
|
||||
agentLoadingService: {
|
||||
refreshAgent,
|
||||
},
|
||||
});
|
||||
@@ -304,7 +300,7 @@ describe("provider history compatibility ownership", () => {
|
||||
const refreshAgent = vi.fn(async () => createCompatibilitySnapshot());
|
||||
const { session, emitted } = createSessionForOwnershipTests({
|
||||
storedRecord: createStoredAgentRecord(),
|
||||
providerHistoryCompatibilityService: {
|
||||
agentLoadingService: {
|
||||
refreshAgent,
|
||||
},
|
||||
});
|
||||
|
||||
@@ -108,7 +108,7 @@ import type {
|
||||
WorkspaceRegistry,
|
||||
} from "./workspace-registry.js";
|
||||
import { WorkspaceReconciliationService } from "./workspace-reconciliation-service.js";
|
||||
import { ProviderHistoryCompatibilityService } from "./provider-history-compatibility-service.js";
|
||||
import { AgentLoadingService } from "./agent-loading-service.js";
|
||||
import {
|
||||
buildVoiceAgentMcpServerConfig,
|
||||
buildVoiceModeSystemPrompt,
|
||||
@@ -181,6 +181,16 @@ const TERMINAL_STREAM_HIGH_WATER_BYTES = 256 * 1024;
|
||||
const TERMINAL_STREAM_LOW_WATER_BYTES = 16 * 1024;
|
||||
const MAX_TERMINAL_STREAM_SLOTS = 256;
|
||||
|
||||
type DeleteFencedAgentSnapshotStore = AgentSnapshotStore & {
|
||||
beginDelete(agentId: string): void;
|
||||
};
|
||||
|
||||
function beginAgentDeleteIfSupported(agentStorage: AgentSnapshotStore, agentId: string): void {
|
||||
if ("beginDelete" in agentStorage && typeof agentStorage.beginDelete === "function") {
|
||||
(agentStorage as DeleteFencedAgentSnapshotStore).beginDelete(agentId);
|
||||
}
|
||||
}
|
||||
|
||||
function deriveInitialAgentTitle(prompt: string): string | null {
|
||||
const firstContentLine = prompt
|
||||
.split(/\r?\n/)
|
||||
@@ -393,7 +403,7 @@ export type SessionOptions = {
|
||||
projectRegistry: ProjectRegistry;
|
||||
workspaceRegistry: WorkspaceRegistry;
|
||||
workspaceReconciliationService?: WorkspaceReconciliationService;
|
||||
providerHistoryCompatibilityService?: ProviderHistoryCompatibilityService;
|
||||
agentLoadingService?: AgentLoadingService;
|
||||
createAgentMcpTransport: AgentMcpTransportFactory;
|
||||
stt: Resolvable<SpeechToTextProvider | null>;
|
||||
tts: Resolvable<TextToSpeechProvider | null>;
|
||||
@@ -554,7 +564,7 @@ export class Session {
|
||||
private readonly projectRegistry: ProjectRegistry;
|
||||
private readonly workspaceRegistry: WorkspaceRegistry;
|
||||
private readonly workspaceReconciliationService: WorkspaceReconciliationService;
|
||||
private readonly providerHistoryCompatibilityService: ProviderHistoryCompatibilityService;
|
||||
private readonly agentLoadingService: AgentLoadingService;
|
||||
private readonly createAgentMcpTransport: AgentMcpTransportFactory;
|
||||
private readonly downloadTokenStore: DownloadTokenStore;
|
||||
private readonly pushTokenStore: PushTokenStore;
|
||||
@@ -616,7 +626,7 @@ export class Session {
|
||||
projectRegistry,
|
||||
workspaceRegistry,
|
||||
workspaceReconciliationService,
|
||||
providerHistoryCompatibilityService,
|
||||
agentLoadingService,
|
||||
createAgentMcpTransport,
|
||||
stt,
|
||||
tts,
|
||||
@@ -655,9 +665,9 @@ export class Session {
|
||||
this.syncWorkspaceGitWatchTarget(cwd, options),
|
||||
removeWorkspaceGitWatchTarget: async (cwd) => this.removeWorkspaceGitWatchTarget(cwd),
|
||||
});
|
||||
this.providerHistoryCompatibilityService =
|
||||
providerHistoryCompatibilityService ??
|
||||
new ProviderHistoryCompatibilityService({
|
||||
this.agentLoadingService =
|
||||
agentLoadingService ??
|
||||
new AgentLoadingService({
|
||||
agentManager: this.agentManager,
|
||||
agentStorage: this.agentStorage,
|
||||
logger: this.sessionLogger,
|
||||
@@ -1082,7 +1092,7 @@ export class Session {
|
||||
}
|
||||
|
||||
private async ensureAgentLoaded(agentId: string): Promise<ManagedAgent> {
|
||||
return this.providerHistoryCompatibilityService.ensureAgentLoaded({ agentId });
|
||||
return this.agentLoadingService.ensureAgentLoaded({ agentId });
|
||||
}
|
||||
|
||||
private matchesAgentFilter(options: {
|
||||
@@ -1675,8 +1685,8 @@ export class Session {
|
||||
(await this.agentStorage.get(agentId))?.cwd ??
|
||||
null;
|
||||
|
||||
// Prevent the persistence hook from re-creating the record while we close/delete.
|
||||
this.agentStorage.beginDelete(agentId);
|
||||
// File-backed storage still needs an early delete fence before closeAgent().
|
||||
beginAgentDeleteIfSupported(this.agentStorage, agentId);
|
||||
|
||||
try {
|
||||
await this.agentManager.closeAgent(agentId);
|
||||
@@ -2482,7 +2492,7 @@ export class Session {
|
||||
);
|
||||
try {
|
||||
await this.unarchiveAgentByHandle(handle);
|
||||
const snapshot = await this.providerHistoryCompatibilityService.resumeAgent({
|
||||
const snapshot = await this.agentLoadingService.resumeAgent({
|
||||
handle,
|
||||
overrides,
|
||||
});
|
||||
@@ -2530,7 +2540,7 @@ export class Session {
|
||||
if (this.agentManager.getAgent(agentId)) {
|
||||
await this.interruptAgentIfRunning(agentId);
|
||||
}
|
||||
const snapshot = await this.providerHistoryCompatibilityService.refreshAgent({ agentId });
|
||||
const snapshot = await this.agentLoadingService.refreshAgent({ agentId });
|
||||
await this.forwardAgentUpdate(snapshot);
|
||||
const timelineSize = this.agentManager.getTimeline(agentId).length;
|
||||
if (requestId) {
|
||||
|
||||
@@ -22,13 +22,11 @@ describe("snapshot mutation ownership boundary", () => {
|
||||
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);
|
||||
|
||||
|
||||
@@ -6,7 +6,10 @@ import { afterEach, beforeEach, describe, expect, test } from "vitest";
|
||||
|
||||
import { createTestLogger } from "../test-utils/test-logger.js";
|
||||
import { AgentStorage } from "./agent/agent-storage.js";
|
||||
import { FileBackedProjectRegistry, FileBackedWorkspaceRegistry } from "./workspace-registry.js";
|
||||
import {
|
||||
FileBackedProjectRegistry,
|
||||
FileBackedWorkspaceRegistry,
|
||||
} from "./workspace-registry.test-helpers.js";
|
||||
import { bootstrapWorkspaceRegistries } from "./workspace-registry-bootstrap.js";
|
||||
import { openPaseoDatabase } from "./db/pglite-database.js";
|
||||
import { DbProjectRegistry } from "./db/db-project-registry.js";
|
||||
|
||||
@@ -0,0 +1,162 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { promises as fs } from "node:fs";
|
||||
import path from "node:path";
|
||||
|
||||
import type { Logger } from "pino";
|
||||
|
||||
import {
|
||||
parsePersistedProjectRecords,
|
||||
parsePersistedWorkspaceRecords,
|
||||
type PersistedProjectRecord,
|
||||
type PersistedWorkspaceRecord,
|
||||
type ProjectRegistry,
|
||||
type WorkspaceRegistry,
|
||||
} from "./workspace-registry.js";
|
||||
|
||||
type RegistryRecord = PersistedProjectRecord | PersistedWorkspaceRecord;
|
||||
|
||||
class FileBackedRegistry<TRecord extends RegistryRecord> {
|
||||
private readonly filePath: string;
|
||||
private readonly logger: Logger;
|
||||
private readonly parseRecord: (record: unknown) => TRecord;
|
||||
private readonly parseRecords: (input: unknown) => TRecord[];
|
||||
private readonly getId: (record: TRecord) => string;
|
||||
private loaded = false;
|
||||
private readonly cache = new Map<string, TRecord>();
|
||||
private persistQueue: Promise<void> = Promise.resolve();
|
||||
|
||||
constructor(options: {
|
||||
filePath: string;
|
||||
logger: Logger;
|
||||
parseRecords: (input: unknown) => TRecord[];
|
||||
getId: (record: TRecord) => string;
|
||||
component: string;
|
||||
}) {
|
||||
this.filePath = options.filePath;
|
||||
this.parseRecords = options.parseRecords;
|
||||
this.parseRecord = (record) => options.parseRecords([record])[0]!;
|
||||
this.getId = options.getId;
|
||||
this.logger = options.logger.child({
|
||||
module: "workspace-registry",
|
||||
component: options.component,
|
||||
});
|
||||
}
|
||||
|
||||
async initialize(): Promise<void> {
|
||||
await this.load();
|
||||
}
|
||||
|
||||
async existsOnDisk(): Promise<boolean> {
|
||||
try {
|
||||
await fs.access(this.filePath);
|
||||
return true;
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
async list(): Promise<TRecord[]> {
|
||||
await this.load();
|
||||
return Array.from(this.cache.values());
|
||||
}
|
||||
|
||||
async get(id: string): Promise<TRecord | null> {
|
||||
await this.load();
|
||||
return this.cache.get(id) ?? null;
|
||||
}
|
||||
|
||||
async upsert(record: TRecord): Promise<void> {
|
||||
await this.load();
|
||||
const parsed = this.parseRecord(record);
|
||||
this.cache.set(this.getId(parsed), parsed);
|
||||
await this.enqueuePersist();
|
||||
}
|
||||
|
||||
async archive(id: string, archivedAt: string): Promise<void> {
|
||||
await this.load();
|
||||
const existing = this.cache.get(id);
|
||||
if (!existing) {
|
||||
return;
|
||||
}
|
||||
const next = this.parseRecord({
|
||||
...existing,
|
||||
updatedAt: archivedAt,
|
||||
archivedAt,
|
||||
});
|
||||
this.cache.set(id, next);
|
||||
await this.enqueuePersist();
|
||||
}
|
||||
|
||||
async remove(id: string): Promise<void> {
|
||||
await this.load();
|
||||
if (!this.cache.delete(id)) {
|
||||
return;
|
||||
}
|
||||
await this.enqueuePersist();
|
||||
}
|
||||
|
||||
private async load(): Promise<void> {
|
||||
if (this.loaded) {
|
||||
return;
|
||||
}
|
||||
|
||||
this.cache.clear();
|
||||
try {
|
||||
const raw = await fs.readFile(this.filePath, "utf8");
|
||||
const parsed = this.parseRecords(JSON.parse(raw));
|
||||
for (const record of parsed) {
|
||||
this.cache.set(this.getId(record), record);
|
||||
}
|
||||
} catch (error) {
|
||||
const code = (error as NodeJS.ErrnoException).code;
|
||||
if (code !== "ENOENT") {
|
||||
this.logger.error({ err: error, filePath: this.filePath }, "Failed to load registry file");
|
||||
}
|
||||
}
|
||||
this.loaded = true;
|
||||
}
|
||||
|
||||
private async persist(): Promise<void> {
|
||||
const records = Array.from(this.cache.values());
|
||||
await fs.mkdir(path.dirname(this.filePath), { recursive: true });
|
||||
const tempPath = `${this.filePath}.${process.pid}.${Date.now()}.${randomUUID()}.tmp`;
|
||||
await fs.writeFile(tempPath, JSON.stringify(records, null, 2), "utf8");
|
||||
await fs.rename(tempPath, this.filePath);
|
||||
}
|
||||
|
||||
private async enqueuePersist(): Promise<void> {
|
||||
const nextPersist = this.persistQueue.then(() => this.persist());
|
||||
this.persistQueue = nextPersist.catch(() => {});
|
||||
await nextPersist;
|
||||
}
|
||||
}
|
||||
|
||||
export class FileBackedProjectRegistry
|
||||
extends FileBackedRegistry<PersistedProjectRecord>
|
||||
implements ProjectRegistry
|
||||
{
|
||||
constructor(filePath: string, logger: Logger) {
|
||||
super({
|
||||
filePath,
|
||||
logger,
|
||||
parseRecords: parsePersistedProjectRecords,
|
||||
getId: (record) => record.projectId,
|
||||
component: "projects",
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
export class FileBackedWorkspaceRegistry
|
||||
extends FileBackedRegistry<PersistedWorkspaceRecord>
|
||||
implements WorkspaceRegistry
|
||||
{
|
||||
constructor(filePath: string, logger: Logger) {
|
||||
super({
|
||||
filePath,
|
||||
logger,
|
||||
parseRecords: parsePersistedWorkspaceRecords,
|
||||
getId: (record) => record.workspaceId,
|
||||
component: "workspaces",
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -6,10 +6,12 @@ import { beforeEach, afterEach, describe, expect, test } from "vitest";
|
||||
|
||||
import { createTestLogger } from "../test-utils/test-logger.js";
|
||||
import {
|
||||
createPersistedProjectRecord,
|
||||
createPersistedWorkspaceRecord,
|
||||
FileBackedProjectRegistry,
|
||||
FileBackedWorkspaceRegistry,
|
||||
} from "./workspace-registry.test-helpers.js";
|
||||
import {
|
||||
createPersistedProjectRecord,
|
||||
createPersistedWorkspaceRecord,
|
||||
} from "./workspace-registry.js";
|
||||
|
||||
describe("workspace registries", () => {
|
||||
|
||||
@@ -1,8 +1,3 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { promises as fs } from "node:fs";
|
||||
import path from "node:path";
|
||||
|
||||
import type { Logger } from "pino";
|
||||
import { z } from "zod";
|
||||
|
||||
import type { PersistedProjectKind, PersistedWorkspaceKind } from "./workspace-registry-model.js";
|
||||
@@ -59,152 +54,6 @@ export interface WorkspaceRegistry {
|
||||
remove(workspaceId: string): Promise<void>;
|
||||
}
|
||||
|
||||
type RegistryRecord = PersistedProjectRecord | PersistedWorkspaceRecord;
|
||||
|
||||
class FileBackedRegistry<TRecord extends RegistryRecord> {
|
||||
private readonly filePath: string;
|
||||
private readonly logger: Logger;
|
||||
private readonly schema: z.ZodSchema<TRecord>;
|
||||
private readonly getId: (record: TRecord) => string;
|
||||
private loaded = false;
|
||||
private readonly cache = new Map<string, TRecord>();
|
||||
private persistQueue: Promise<void> = Promise.resolve();
|
||||
|
||||
constructor(options: {
|
||||
filePath: string;
|
||||
logger: Logger;
|
||||
schema: z.ZodSchema<TRecord>;
|
||||
getId: (record: TRecord) => string;
|
||||
component: string;
|
||||
}) {
|
||||
this.filePath = options.filePath;
|
||||
this.schema = options.schema;
|
||||
this.getId = options.getId;
|
||||
this.logger = options.logger.child({
|
||||
module: "workspace-registry",
|
||||
component: options.component,
|
||||
});
|
||||
}
|
||||
|
||||
async initialize(): Promise<void> {
|
||||
await this.load();
|
||||
}
|
||||
|
||||
async existsOnDisk(): Promise<boolean> {
|
||||
try {
|
||||
await fs.access(this.filePath);
|
||||
return true;
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
async list(): Promise<TRecord[]> {
|
||||
await this.load();
|
||||
return Array.from(this.cache.values());
|
||||
}
|
||||
|
||||
async get(id: string): Promise<TRecord | null> {
|
||||
await this.load();
|
||||
return this.cache.get(id) ?? null;
|
||||
}
|
||||
|
||||
async upsert(record: TRecord): Promise<void> {
|
||||
await this.load();
|
||||
const parsed = this.schema.parse(record);
|
||||
this.cache.set(this.getId(parsed), parsed);
|
||||
await this.enqueuePersist();
|
||||
}
|
||||
|
||||
async archive(id: string, archivedAt: string): Promise<void> {
|
||||
await this.load();
|
||||
const existing = this.cache.get(id);
|
||||
if (!existing) {
|
||||
return;
|
||||
}
|
||||
const next = this.schema.parse({
|
||||
...existing,
|
||||
updatedAt: archivedAt,
|
||||
archivedAt,
|
||||
});
|
||||
this.cache.set(id, next);
|
||||
await this.enqueuePersist();
|
||||
}
|
||||
|
||||
async remove(id: string): Promise<void> {
|
||||
await this.load();
|
||||
if (!this.cache.delete(id)) {
|
||||
return;
|
||||
}
|
||||
await this.enqueuePersist();
|
||||
}
|
||||
|
||||
private async load(): Promise<void> {
|
||||
if (this.loaded) {
|
||||
return;
|
||||
}
|
||||
|
||||
this.cache.clear();
|
||||
try {
|
||||
const raw = await fs.readFile(this.filePath, "utf8");
|
||||
const parsed = z.array(this.schema).parse(JSON.parse(raw));
|
||||
for (const record of parsed) {
|
||||
this.cache.set(this.getId(record), record);
|
||||
}
|
||||
} catch (error) {
|
||||
const code = (error as NodeJS.ErrnoException).code;
|
||||
if (code !== "ENOENT") {
|
||||
this.logger.error({ err: error, filePath: this.filePath }, "Failed to load registry file");
|
||||
}
|
||||
}
|
||||
this.loaded = true;
|
||||
}
|
||||
|
||||
private async persist(): Promise<void> {
|
||||
const records = Array.from(this.cache.values());
|
||||
await fs.mkdir(path.dirname(this.filePath), { recursive: true });
|
||||
const tempPath = `${this.filePath}.${process.pid}.${Date.now()}.${randomUUID()}.tmp`;
|
||||
await fs.writeFile(tempPath, JSON.stringify(records, null, 2), "utf8");
|
||||
await fs.rename(tempPath, this.filePath);
|
||||
}
|
||||
|
||||
private async enqueuePersist(): Promise<void> {
|
||||
const nextPersist = this.persistQueue.then(() => this.persist());
|
||||
this.persistQueue = nextPersist.catch(() => {});
|
||||
await nextPersist;
|
||||
}
|
||||
}
|
||||
|
||||
export class FileBackedProjectRegistry
|
||||
extends FileBackedRegistry<PersistedProjectRecord>
|
||||
implements ProjectRegistry
|
||||
{
|
||||
constructor(filePath: string, logger: Logger) {
|
||||
super({
|
||||
filePath,
|
||||
logger,
|
||||
schema: PersistedProjectRecordSchema,
|
||||
getId: (record) => record.projectId,
|
||||
component: "projects",
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
export class FileBackedWorkspaceRegistry
|
||||
extends FileBackedRegistry<PersistedWorkspaceRecord>
|
||||
implements WorkspaceRegistry
|
||||
{
|
||||
constructor(filePath: string, logger: Logger) {
|
||||
super({
|
||||
filePath,
|
||||
logger,
|
||||
schema: PersistedWorkspaceRecordSchema,
|
||||
getId: (record) => record.workspaceId,
|
||||
component: "workspaces",
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
export function createPersistedProjectRecord(input: {
|
||||
projectId: string;
|
||||
rootPath: string;
|
||||
|
||||
Reference in New Issue
Block a user