fix: harden migration/reconciliation and fix legacy import project grouping

- DbProjectRegistry/DbWorkspaceRegistry: use ON CONFLICT(directory) instead
  of ON CONFLICT(id) so inserts and upserts handle duplicate directories
  gracefully instead of crashing
- Bootstrap: wrap legacy imports in try/catch (non-fatal), move
  reconciliation start after imports so first pass cleans up stale data
- Legacy project/workspace import: deduplicate by rootPath (prefer git
  over non_git), derive gitRemote from legacy projectId field
- Legacy agent snapshot import: detect git metadata and group agents by
  remote/toplevel instead of creating one project per cwd, clamp
  running/initializing statuses to closed
- Timeline hydration: set historyPrimed based on whether durable store
  has rows (not just whether it exists), remove hard gate so imported
  agents get their timelines populated lazily from provider history
- Durable timeline append: use bulkInsert with pre-assigned seq instead
  of appendCommitted which recalculates seq, fixing UNIQUE constraint
  failures during concurrent hydration writes
This commit is contained in:
Mohamed Boudra
2026-04-05 22:26:45 +07:00
parent cc6b0f80be
commit ce821afd27
7 changed files with 302 additions and 52 deletions
@@ -575,6 +575,10 @@ export class AgentManager {
options?: AgentTimelineFetchOptions,
): Promise<AgentTimelineFetchResult> {
this.requireAgent(id);
const agent = this.agents.get(id);
if (agent && !agent.historyPrimed) {
await this.hydrateTimelineFromLegacyProviderHistory(agent);
}
if (this.durableTimelineStore) {
return await this.durableTimelineStore.fetchCommitted(id, options);
}
@@ -1509,15 +1513,12 @@ export class AgentManager {
}
/**
* 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.
* Hydrates the timeline from provider history if the agent's durable
* timeline is empty (e.g., imported agents that have provider history
* on disk but no persisted timeline rows). No-ops if already hydrated.
*/
async hydrateTimelineFromProvider(agentId: string): Promise<void> {
const agent = this.requireSessionAgent(agentId);
if (this.durableTimelineStore) {
return;
}
await this.hydrateTimelineFromLegacyProviderHistory(agent);
}
@@ -1820,6 +1821,7 @@ export class AgentManager {
const durableTimelineSeed = shouldSeedFromDurable
? await this.loadCommittedTimelineSeed(resolvedAgentId, now)
: null;
const durableTimelineHasRows = durableTimelineSeed != null && (durableTimelineSeed.nextSeq ?? 1) > 1;
const timelineSeed = explicitTimelineSeed ?? durableTimelineSeed;
if (timelineSeed || !this.timelineStore.has(resolvedAgentId)) {
this.timelineStore.initialize(resolvedAgentId, timelineSeed ?? { timestamp: now.toISOString() });
@@ -1848,7 +1850,7 @@ export class AgentManager {
unsubscribeSession: null,
provisionalAssistantText: options?.provisionalAssistantText ?? null,
persistence: attachPersistenceCwd(session.describePersistence(), config.cwd),
historyPrimed: options?.historyPrimed ?? shouldSeedFromDurable,
historyPrimed: options?.historyPrimed ?? durableTimelineHasRows,
lastUserMessageAt: options?.lastUserMessageAt ?? null,
lastUsage: options?.lastUsage,
lastError: options?.lastError,
@@ -2576,7 +2578,7 @@ export class AgentManager {
return;
}
const task = this.durableTimelineStore
.appendCommitted(agentId, row.item, { timestamp: row.timestamp })
.bulkInsert(agentId, [row])
.then(() => undefined)
.catch((err) => {
this.logger.error(
+21 -13
View File
@@ -403,6 +403,27 @@ export async function createPaseoDaemon(
const projectRegistry = new DbProjectRegistry(database.db);
const workspaceRegistry = new DbWorkspaceRegistry(database.db);
try {
await importLegacyProjectWorkspaceJson({
db: database.db,
paseoHome: config.paseoHome,
logger,
});
logger.info({ elapsed: elapsed() }, "Legacy project/workspace import checked");
} catch (err) {
logger.error({ err }, "Legacy project/workspace import failed (non-fatal)");
}
try {
await importLegacyAgentSnapshots({
db: database.db,
paseoHome: config.paseoHome,
logger,
});
logger.info({ elapsed: elapsed() }, "Legacy agent snapshot import checked");
} catch (err) {
logger.error({ err }, "Legacy agent snapshot import failed (non-fatal)");
}
const reconciliationService = new WorkspaceReconciliationService({
projectRegistry,
workspaceRegistry,
@@ -410,19 +431,6 @@ export async function createPaseoDaemon(
});
reconciliationService.start();
logger.info({ elapsed: elapsed() }, "Workspace reconciliation service started");
await importLegacyProjectWorkspaceJson({
db: database.db,
paseoHome: config.paseoHome,
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 chatService.initialize();
logger.info({ elapsed: elapsed() }, "Chat service initialized");
const checkoutDiffManager = new CheckoutDiffManager({
@@ -42,6 +42,16 @@ export class DbProjectRegistry implements ProjectRegistry {
const [row] = await this.db
.insert(projects)
.values(record)
.onConflictDoUpdate({
target: projects.directory,
set: {
kind: record.kind,
displayName: record.displayName,
gitRemote: record.gitRemote,
updatedAt: record.updatedAt,
archivedAt: record.archivedAt,
},
})
.returning({ id: projects.id });
return row!.id;
}
@@ -52,13 +62,11 @@ export class DbProjectRegistry implements ProjectRegistry {
.insert(projects)
.values(nextRecord)
.onConflictDoUpdate({
target: projects.id,
target: projects.directory,
set: {
directory: nextRecord.directory,
kind: nextRecord.kind,
displayName: nextRecord.displayName,
gitRemote: nextRecord.gitRemote,
createdAt: nextRecord.createdAt,
updatedAt: nextRecord.updatedAt,
archivedAt: nextRecord.archivedAt,
},
@@ -46,6 +46,16 @@ export class DbWorkspaceRegistry implements WorkspaceRegistry {
const [row] = await this.db
.insert(workspaces)
.values(record)
.onConflictDoUpdate({
target: workspaces.directory,
set: {
projectId: record.projectId,
kind: record.kind,
displayName: record.displayName,
updatedAt: record.updatedAt,
archivedAt: record.archivedAt,
},
})
.returning({ id: workspaces.id });
return row!.id;
}
@@ -56,13 +66,11 @@ export class DbWorkspaceRegistry implements WorkspaceRegistry {
.insert(workspaces)
.values(nextRecord)
.onConflictDoUpdate({
target: workspaces.id,
target: workspaces.directory,
set: {
projectId: nextRecord.projectId,
directory: nextRecord.directory,
kind: nextRecord.kind,
displayName: nextRecord.displayName,
createdAt: nextRecord.createdAt,
updatedAt: nextRecord.updatedAt,
archivedAt: nextRecord.archivedAt,
},
@@ -1,10 +1,13 @@
import path from "node:path";
import { execSync } from "node:child_process";
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 { detectWorkspaceGitMetadata } from "../workspace-git-metadata.js";
import { READ_ONLY_GIT_ENV } from "../checkout-git-utils.js";
import { normalizeWorkspaceId } from "../workspace-registry-model.js";
import type { PaseoDatabaseHandle } from "./sqlite-database.js";
import { toAgentSnapshotRowValues } from "./db-agent-snapshot-store.js";
@@ -76,10 +79,15 @@ export async function importLegacyAgentSnapshots(options: {
workspaceRows.map((row) => [row.directory, row.id] as const),
);
const projectRows = tx
.select({ id: projects.id, directory: projects.directory })
.select({ id: projects.id, directory: projects.directory, gitRemote: projects.gitRemote })
.from(projects)
.all();
const projectIdsByDirectory = new Map(projectRows.map((row) => [row.directory, row.id] as const));
const projectIdsByRemote = new Map(
projectRows
.filter((row): row is typeof row & { gitRemote: string } => row.gitRemote !== null)
.map((row) => [row.gitRemote, row.id] as const),
);
for (const record of records) {
const normalizedDirectory = normalizeWorkspaceId(record.cwd);
if (workspaceIdsByDirectory.has(normalizedDirectory)) {
@@ -87,17 +95,30 @@ export async function importLegacyAgentSnapshots(options: {
}
const timestamp = record.updatedAt ?? record.createdAt;
const displayName =
normalizedDirectory.split(/[\\/]/).filter(Boolean).at(-1) ?? normalizedDirectory;
let projectId = projectIdsByDirectory.get(normalizedDirectory);
const gitInfo = detectGitInfoForCwd(record.cwd);
const resolvedDirectory = gitInfo?.toplevel
? normalizeWorkspaceId(gitInfo.toplevel)
: normalizedDirectory;
const projectDisplayName = gitInfo?.metadata.projectDisplayName
?? resolvedDirectory.split(/[\\/]/).filter(Boolean).at(-1)
?? resolvedDirectory;
const projectKind = gitInfo?.metadata.projectKind ?? "directory";
const gitRemote = gitInfo?.metadata.gitRemote ?? null;
const workspaceKind = gitInfo?.metadata.isWorktree ? "worktree" : "checkout";
const workspaceDisplayName = gitInfo?.metadata.workspaceDisplayName
?? normalizedDirectory.split(/[\\/]/).filter(Boolean).at(-1)
?? normalizedDirectory;
let projectId = projectIdsByDirectory.get(resolvedDirectory)
?? (gitRemote !== null ? projectIdsByRemote.get(gitRemote) : undefined);
if (projectId === undefined) {
const projectRow = tx
.insert(projects)
.values({
directory: normalizedDirectory,
displayName,
kind: "directory",
gitRemote: null,
directory: resolvedDirectory,
displayName: projectDisplayName,
kind: projectKind,
gitRemote,
createdAt: record.createdAt,
updatedAt: timestamp,
archivedAt: null,
@@ -105,7 +126,10 @@ export async function importLegacyAgentSnapshots(options: {
.returning({ id: projects.id })
.get();
projectId = projectRow!.id;
projectIdsByDirectory.set(normalizedDirectory, projectId);
projectIdsByDirectory.set(resolvedDirectory, projectId);
if (gitRemote !== null) {
projectIdsByRemote.set(gitRemote, projectId);
}
}
const workspaceRow = tx
@@ -113,8 +137,8 @@ export async function importLegacyAgentSnapshots(options: {
.values({
projectId,
directory: normalizedDirectory,
displayName,
kind: "checkout",
displayName: workspaceDisplayName,
kind: workspaceKind,
createdAt: record.createdAt,
updatedAt: timestamp,
archivedAt: null,
@@ -125,7 +149,13 @@ export async function importLegacyAgentSnapshots(options: {
}
const rows = records.flatMap((record) => {
const workspaceId = workspaceIdsByDirectory.get(normalizeWorkspaceId(record.cwd));
return workspaceId === undefined ? [] : [toAgentSnapshotRowValues({ record, workspaceId })];
if (workspaceId === undefined) {
return [];
}
const clampedRecord = (record.lastStatus === "running" || record.lastStatus === "initializing")
? { ...record, lastStatus: "closed" as const }
: record;
return [toAgentSnapshotRowValues({ record: clampedRecord, workspaceId })];
});
const totalBatches = Math.ceil(rows.length / MAX_AGENT_SNAPSHOT_ROWS_PER_INSERT);
for (let startIndex = 0; startIndex < rows.length; startIndex += MAX_AGENT_SNAPSHOT_ROWS_PER_INSERT) {
@@ -245,3 +275,24 @@ async function pathExists(targetPath: string): Promise<boolean> {
throw error;
}
}
function detectGitInfoForCwd(
cwd: string,
): { toplevel: string; metadata: ReturnType<typeof detectWorkspaceGitMetadata> } | null {
try {
const toplevel = execSync("git rev-parse --show-toplevel", {
cwd,
env: READ_ONLY_GIT_ENV,
encoding: "utf8",
stdio: ["ignore", "pipe", "ignore"],
}).trim();
if (!toplevel) {
return null;
}
const directoryName = toplevel.split(/[\\/]/).filter(Boolean).at(-1) ?? toplevel;
const metadata = detectWorkspaceGitMetadata(cwd, directoryName);
return { toplevel, metadata };
} catch {
return null;
}
}
@@ -0,0 +1,158 @@
/**
* Adhoc test: imports real legacy data from ~/.paseo into a fresh SQLite DB
* and asserts that projects/workspaces are properly grouped.
*
* Run with: npx vitest run packages/server/src/server/db/legacy-import-real-data.adhoc.test.ts
*/
import os from "node:os";
import path from "node:path";
import { mkdirSync, mkdtempSync, rmSync } from "node:fs";
import { afterEach, beforeEach, describe, expect, test } from "vitest";
import { createTestLogger } from "../../test-utils/test-logger.js";
import { openPaseoDatabase, type PaseoDatabaseHandle } from "./sqlite-database.js";
import { importLegacyProjectWorkspaceJson } from "./legacy-project-workspace-import.js";
import { importLegacyAgentSnapshots } from "./legacy-agent-snapshot-import.js";
import { projects, workspaces, agentSnapshots } from "./schema.js";
import { eq } from "drizzle-orm";
const REAL_PASEO_HOME = path.join(os.homedir(), ".paseo");
describe("legacy import from real ~/.paseo data", () => {
let tmpDir: string;
let dbDir: string;
let database: PaseoDatabaseHandle;
beforeEach(async () => {
tmpDir = mkdtempSync(path.join(os.tmpdir(), "paseo-real-import-"));
dbDir = path.join(tmpDir, "db");
mkdirSync(dbDir, { recursive: true });
database = await openPaseoDatabase(dbDir);
});
afterEach(async () => {
await database?.close();
rmSync(tmpDir, { recursive: true, force: true });
});
test("imports real data and groups projects correctly", async () => {
const logger = createTestLogger();
// Phase 1: Import legacy project/workspace JSON (has proper grouping)
const pwResult = await importLegacyProjectWorkspaceJson({
db: database.db,
paseoHome: REAL_PASEO_HOME,
logger,
});
console.log("Project/workspace import result:", pwResult);
// Phase 2: Import legacy agent snapshots
const agentResult = await importLegacyAgentSnapshots({
db: database.db,
paseoHome: REAL_PASEO_HOME,
logger,
});
console.log("Agent snapshot import result:", agentResult);
// --- Assertions ---
const allProjects = await database.db.select().from(projects);
const allWorkspaces = await database.db.select().from(workspaces);
const allAgents = await database.db.select().from(agentSnapshots);
console.log(`Total projects: ${allProjects.length}`);
console.log(`Total workspaces: ${allWorkspaces.length}`);
console.log(`Total agents: ${allAgents.length}`);
// 1. There should be fewer projects than workspaces (workspaces group under projects)
expect(allProjects.length).toBeLessThan(allWorkspaces.length);
// 2. There should be exactly ONE project per unique git remote
const projectsByRemote = new Map<string, typeof allProjects>();
for (const project of allProjects) {
if (project.gitRemote) {
const existing = projectsByRemote.get(project.gitRemote) ?? [];
existing.push(project);
projectsByRemote.set(project.gitRemote, existing);
}
}
const duplicateRemotes: string[] = [];
for (const [remote, projectList] of projectsByRemote) {
if (projectList.length > 1) {
duplicateRemotes.push(remote);
console.log(`DUPLICATE: ${remote} has ${projectList.length} projects:`);
for (const p of projectList) {
console.log(` id=${p.id} directory=${p.directory}`);
}
}
}
expect(duplicateRemotes).toEqual([]);
// 3. Specifically: getpaseo/paseo should be ONE project
const paseoProjects = allProjects.filter(
(p) => p.gitRemote === "git@github.com:getpaseo/paseo.git",
);
expect(paseoProjects).toHaveLength(1);
const paseoProject = paseoProjects[0]!;
// 4. All paseo workspaces (worktrees + main checkout + subdirs) should be under that one project
const paseoWorkspaces = allWorkspaces.filter(
(w) => w.projectId === paseoProject.id,
);
console.log(`Paseo project id=${paseoProject.id}, directory=${paseoProject.directory}`);
console.log(`Paseo workspaces: ${paseoWorkspaces.length}`);
// The old data had ~51 paseo workspaces
expect(paseoWorkspaces.length).toBeGreaterThanOrEqual(10);
// 5. Subdirectory workspaces (packages/server, packages/app) should be under the same project
const subdirWorkspaces = paseoWorkspaces.filter((w) =>
w.directory.includes("/packages/"),
);
console.log(`Paseo subdirectory workspaces: ${subdirWorkspaces.length}`);
for (const w of subdirWorkspaces) {
console.log(` ${w.directory} (projectId=${w.projectId})`);
expect(w.projectId).toBe(paseoProject.id);
}
// 6. Worktree workspaces should be under the same project
const worktreeWorkspaces = paseoWorkspaces.filter((w) =>
w.directory.includes("/.paseo/worktrees/"),
);
console.log(`Paseo worktree workspaces: ${worktreeWorkspaces.length}`);
expect(worktreeWorkspaces.length).toBeGreaterThan(0);
// 7. No git project should be a subdirectory of another project with the SAME git remote
// (e.g., /dev/paseo/packages/server should not be its own project if /dev/paseo exists)
const activeGitProjects = allProjects.filter((p) => !p.archivedAt && p.gitRemote);
const subdirProjects: string[] = [];
for (const project of activeGitProjects) {
for (const other of activeGitProjects) {
if (
project.id !== other.id &&
project.gitRemote === other.gitRemote &&
project.directory.startsWith(other.directory + "/")
) {
subdirProjects.push(
`${project.directory} (id=${project.id}) is under ${other.directory} (id=${other.id}), both remote=${project.gitRemote}`,
);
}
}
}
if (subdirProjects.length > 0) {
console.log("Subdirectory git projects with same remote (should be empty):");
for (const s of subdirProjects) {
console.log(` ${s}`);
}
}
expect(subdirProjects).toEqual([]);
// 8. All agent snapshots should reference valid workspaces
for (const agent of allAgents) {
const workspace = allWorkspaces.find((w) => w.id === agent.workspaceId);
expect(workspace).toBeDefined();
}
});
});
@@ -9,6 +9,16 @@ import { z } from "zod";
import type { PaseoDatabaseHandle } from "./sqlite-database.js";
import { projects, workspaces } from "./schema.js";
const LEGACY_REMOTE_PREFIX = "remote:";
function deriveGitRemoteFromLegacyProjectId(projectId: string): string | null {
if (!projectId.startsWith(LEGACY_REMOTE_PREFIX)) {
return null;
}
const hostAndPath = projectId.slice(LEGACY_REMOTE_PREFIX.length);
return `git@${hostAndPath.replace("/", ":")}.git`;
}
// Legacy JSON schemas — these match the old pre-migration format
const LegacyProjectSchema = z.object({
projectId: z.string(),
@@ -91,24 +101,26 @@ export async function importLegacyProjectWorkspaceJson(options: {
};
}
const dedupedProjects = [...new Map(projectRows.map((project) => [project.rootPath, project])).values()];
const dedupedWorkspaces = [...new Map(workspaceRows.map((workspace) => [workspace.cwd, workspace])).values()];
options.logger.info(
{ projects: dedupedProjects.length, workspaces: dedupedWorkspaces.length },
"Starting legacy project/workspace import",
);
// Deduplicate legacy projects by rootPath — prefer git over non_git
const deduplicatedProjects = new Map<string, (typeof projectRows)[number]>();
for (const legacy of projectRows) {
const existing = deduplicatedProjects.get(legacy.rootPath);
if (!existing || (legacy.kind === "git" && existing.kind !== "git")) {
deduplicatedProjects.set(legacy.rootPath, legacy);
}
}
options.db.transaction((tx) => {
// Insert projects, mapping old format to new schema
const projectDirectoryToId = new Map<string, number>();
for (const legacy of dedupedProjects) {
for (const legacy of deduplicatedProjects.values()) {
const row = tx
.insert(projects)
.values({
directory: legacy.rootPath,
displayName: legacy.displayName,
kind: legacy.kind === "non_git" ? "directory" : legacy.kind,
gitRemote: deriveGitRemoteFromLegacyProjectId(legacy.projectId),
createdAt: legacy.createdAt,
updatedAt: legacy.updatedAt,
archivedAt: legacy.archivedAt,
@@ -119,6 +131,7 @@ export async function importLegacyProjectWorkspaceJson(options: {
}
// Build a map from legacy projectId -> new integer id
// Uses original projectRows so all duplicate projectIds resolve to the same new id
const legacyProjectIdToNewId = new Map<string, number>();
for (const legacy of projectRows) {
const newId = projectDirectoryToId.get(legacy.rootPath);
@@ -128,7 +141,7 @@ export async function importLegacyProjectWorkspaceJson(options: {
}
// Insert workspaces, resolving project FK
for (const legacy of dedupedWorkspaces) {
for (const legacy of workspaceRows) {
const projectId = legacyProjectIdToNewId.get(legacy.projectId);
if (projectId === undefined) {
throw new Error(`Legacy workspace ${legacy.workspaceId} references unknown project ${legacy.projectId}`);
@@ -151,18 +164,20 @@ export async function importLegacyProjectWorkspaceJson(options: {
}
});
const importedProjects = deduplicatedProjects.size;
options.logger.info(
{
importedProjects: dedupedProjects.length,
importedWorkspaces: dedupedWorkspaces.length,
importedProjects,
importedWorkspaces: workspaceRows.length,
},
"Imported legacy project/workspace JSON into the database",
);
return {
status: "imported",
importedProjects: dedupedProjects.length,
importedWorkspaces: dedupedWorkspaces.length,
importedProjects,
importedWorkspaces: workspaceRows.length,
};
}