From ce821afd27506d682c929424b4591f9f9a670ed8 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Thu, 2 Apr 2026 15:28:45 +0700 Subject: [PATCH] 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 --- .../server/src/server/agent/agent-manager.ts | 18 +- packages/server/src/server/bootstrap.ts | 34 ++-- .../src/server/db/db-project-registry.ts | 14 +- .../src/server/db/db-workspace-registry.ts | 14 +- .../server/db/legacy-agent-snapshot-import.ts | 75 +++++++-- .../db/legacy-import-real-data.adhoc.test.ts | 158 ++++++++++++++++++ .../db/legacy-project-workspace-import.ts | 41 +++-- 7 files changed, 302 insertions(+), 52 deletions(-) create mode 100644 packages/server/src/server/db/legacy-import-real-data.adhoc.test.ts diff --git a/packages/server/src/server/agent/agent-manager.ts b/packages/server/src/server/agent/agent-manager.ts index 7698dda02..d25f1cf0d 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -575,6 +575,10 @@ export class AgentManager { options?: AgentTimelineFetchOptions, ): Promise { 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 { 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( diff --git a/packages/server/src/server/bootstrap.ts b/packages/server/src/server/bootstrap.ts index 208021b41..2f8f9752e 100644 --- a/packages/server/src/server/bootstrap.ts +++ b/packages/server/src/server/bootstrap.ts @@ -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({ diff --git a/packages/server/src/server/db/db-project-registry.ts b/packages/server/src/server/db/db-project-registry.ts index 9c623edb7..c7fad1b5d 100644 --- a/packages/server/src/server/db/db-project-registry.ts +++ b/packages/server/src/server/db/db-project-registry.ts @@ -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, }, diff --git a/packages/server/src/server/db/db-workspace-registry.ts b/packages/server/src/server/db/db-workspace-registry.ts index 0ebb9099f..f5e2ea7a9 100644 --- a/packages/server/src/server/db/db-workspace-registry.ts +++ b/packages/server/src/server/db/db-workspace-registry.ts @@ -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, }, diff --git a/packages/server/src/server/db/legacy-agent-snapshot-import.ts b/packages/server/src/server/db/legacy-agent-snapshot-import.ts index 986715ec6..5a9538128 100644 --- a/packages/server/src/server/db/legacy-agent-snapshot-import.ts +++ b/packages/server/src/server/db/legacy-agent-snapshot-import.ts @@ -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 { throw error; } } + +function detectGitInfoForCwd( + cwd: string, +): { toplevel: string; metadata: ReturnType } | 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; + } +} diff --git a/packages/server/src/server/db/legacy-import-real-data.adhoc.test.ts b/packages/server/src/server/db/legacy-import-real-data.adhoc.test.ts new file mode 100644 index 000000000..8ff8a0e91 --- /dev/null +++ b/packages/server/src/server/db/legacy-import-real-data.adhoc.test.ts @@ -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(); + 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(); + } + }); +}); diff --git a/packages/server/src/server/db/legacy-project-workspace-import.ts b/packages/server/src/server/db/legacy-project-workspace-import.ts index 4010b63b3..8b9b55f86 100644 --- a/packages/server/src/server/db/legacy-project-workspace-import.ts +++ b/packages/server/src/server/db/legacy-project-workspace-import.ts @@ -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(); + 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(); - 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(); 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, }; }