diff --git a/.changeset/honor-archive-data-dir.md b/.changeset/honor-archive-data-dir.md new file mode 100644 index 00000000..f50f6c5b --- /dev/null +++ b/.changeset/honor-archive-data-dir.md @@ -0,0 +1,7 @@ +--- +"@jmfederico/pi-web": patch +--- + +Store session archive metadata and archived session files under `PI_WEB_DATA_DIR` when configured, and automatically migrate a legacy archive on the first eligible session-daemon startup after upgrading. + +Migration runs only when `PI_WEB_DATA_DIR` explicitly selects a different root, the legacy index and every referenced file form a complete valid archive, and the destination archive is pristine. PI WEB copies and verifies files across filesystem boundaries, rewrites their `archivePath` values, atomically commits the destination index, and only then removes legacy archive state. Ambiguous, invalid, partial, or coexisting layouts are left untouched instead of being merged or overwritten; active Pi session files are never moved. diff --git a/docs/config.md b/docs/config.md index cc99e673..8eb92820 100644 --- a/docs/config.md +++ b/docs/config.md @@ -115,7 +115,7 @@ Rows with JSON key `—` are runtime-only environment variables, not config-file | Project config version | `version` | — | Project | Project-local only; must be `1` when present | Next project-config read | | **Runtime-only environment variables** | | | | | | | Global config file path | — | `PI_WEB_CONFIG` (`XDG_CONFIG_HOME` affects the default path) | Process/env | Selects the global config file; not a project config | Restart services/processes after changing env | -| Managed data directory | — | `PI_WEB_DATA_DIR` | Process/env | Not supported locally | Restart services before changing; moves managed state location | +| Managed data directory | — | `PI_WEB_DATA_DIR` | Process/env | Not supported locally | Restart web/API and session daemon | | Session daemon socket | — | `PI_WEB_SESSIOND_SOCKET` | Web/API + session daemon env | Not supported locally | Restart daemon and web/API; both must match | | Session daemon TCP port | — | `PI_WEB_SESSIOND_PORT` | Session daemon env | Not supported locally | Restart session daemon; set `PI_WEB_SESSIOND_URL` for web/API too | | Session daemon TCP host | — | `PI_WEB_SESSIOND_HOST` | Session daemon env | Not supported locally | Restart session daemon | @@ -128,6 +128,12 @@ Rows with JSON key `—` are runtime-only environment variables, not config-file ## Key details +### Managed data directory + +`PI_WEB_DATA_DIR` sets the root for PI WEB-managed runtime state and defaults to `~/.pi-web`. Unless a more specific path override is configured, PI WEB stores its project and machine registries, locally discovered plugins, default session-daemon socket, and session archives beneath this root. + +This setting does not change the PI WEB config file selected by `PI_WEB_CONFIG` or Pi-owned state such as the active session files selected by `PI_CODING_AGENT_SESSION_DIR`. + ### External path access `pathAccess.allowedPaths` grants PI WEB's file explorer and absolute `@` path completions access to specific filesystem roots outside the current workspace. diff --git a/src/server/sessiond.ts b/src/server/sessiond.ts index f4cc36eb..bcdd0238 100644 --- a/src/server/sessiond.ts +++ b/src/server/sessiond.ts @@ -20,73 +20,83 @@ import { registerTerminalRoutes } from "./terminals/terminalRoutes.js"; import { getPiWebRuntimeComponent } from "./piWebStatus.js"; import { SESSIOND_RUNTIME_CAPABILITIES } from "../shared/capabilities.js"; import { effectivePiWebConfig, maxUploadBytes, spawnSessionsEnabled, subsessionsEnabled } from "../config.js"; +import { runSessionDaemonStartup } from "./sessiond/sessionDaemonStartup.js"; const { config } = effectivePiWebConfig(); const app = Fastify({ logger: true, bodyLimit: maxUploadBytes(process.env, config) }); await app.register(fastifyWebsocket); -const eventHub = new SessionEventHub(); -const workspaceActivity = new WorkspaceActivityService(eventHub); -const auth = new AuthService(); -const spawnTargets = spawnSessionsEnabled(process.env, config) - ? new ProjectScopedSpawnTargetResolver({ projects: new ProjectService(new ProjectStore()), workspaces: new WorkspaceService() }) - : undefined; -const sessions = new PiSessionService(eventHub, { - modelRegistry: auth.modelRegistry, - workspaceActivity, +await runSessionDaemonStartup({ logger: app.log, - ...(spawnTargets === undefined ? {} : { spawnTargets }), - subsessionsEnabled: spawnTargets !== undefined && subsessionsEnabled(process.env, config), -}); -auth.subscribe((change) => { sessions.applyAuthChange(change); }); -const terminals = new TerminalService(eventHub, workspaceActivity); -registerWorkspaceActivityRoutes(app, workspaceActivity); -registerAuthRoutes(app, auth); -registerSessionRoutes(app, sessions, eventHub); -registerTerminalRoutes(app, terminals); - -app.get("/health", () => { - const runtime = getPiWebRuntimeComponent("sessiond", SESSIOND_RUNTIME_CAPABILITIES); - return { - ok: true, - activeSessions: sessions.activeCount(), - checkedAt: new Date().toISOString(), - version: { - component: runtime.component, - label: runtime.label, - ...(runtime.runtimeVersion === undefined ? {} : { runtimeVersion: runtime.runtimeVersion }), - stale: false, - available: runtime.available, - }, - }; -}); + createRuntime() { + const eventHub = new SessionEventHub(); + const workspaceActivity = new WorkspaceActivityService(eventHub); + const auth = new AuthService(); + const spawnTargets = spawnSessionsEnabled(process.env, config) + ? new ProjectScopedSpawnTargetResolver({ projects: new ProjectService(new ProjectStore()), workspaces: new WorkspaceService() }) + : undefined; + const sessions = new PiSessionService(eventHub, { + modelRegistry: auth.modelRegistry, + workspaceActivity, + logger: app.log, + ...(spawnTargets === undefined ? {} : { spawnTargets }), + subsessionsEnabled: spawnTargets !== undefined && subsessionsEnabled(process.env, config), + }); + auth.subscribe((change) => { sessions.applyAuthChange(change); }); + const terminals = new TerminalService(eventHub, workspaceActivity); + return { eventHub, workspaceActivity, auth, sessions, terminals }; + }, + registerRoutes({ eventHub, workspaceActivity, auth, sessions, terminals }) { + registerWorkspaceActivityRoutes(app, workspaceActivity); + registerAuthRoutes(app, auth); + registerSessionRoutes(app, sessions, eventHub); + registerTerminalRoutes(app, terminals); -app.get("/runtime", () => getPiWebRuntimeComponent("sessiond", SESSIOND_RUNTIME_CAPABILITIES)); + app.get("/health", () => { + const runtime = getPiWebRuntimeComponent("sessiond", SESSIOND_RUNTIME_CAPABILITIES); + return { + ok: true, + activeSessions: sessions.activeCount(), + checkedAt: new Date().toISOString(), + version: { + component: runtime.component, + label: runtime.label, + ...(runtime.runtimeVersion === undefined ? {} : { runtimeVersion: runtime.runtimeVersion }), + stale: false, + available: runtime.available, + }, + }; + }); -let shuttingDown = false; -async function shutdown(signal: NodeJS.Signals): Promise { - if (shuttingDown) return; - shuttingDown = true; - app.log.info({ signal }, "shutting down session daemon"); - terminals.dispose(); - auth.dispose(); - await sessions.dispose(); - await app.close(); -} + app.get("/runtime", () => getPiWebRuntimeComponent("sessiond", SESSIOND_RUNTIME_CAPABILITIES)); + }, + async listen({ auth, sessions, terminals }) { + let shuttingDown = false; + async function shutdown(signal: NodeJS.Signals): Promise { + if (shuttingDown) return; + shuttingDown = true; + app.log.info({ signal }, "shutting down session daemon"); + terminals.dispose(); + auth.dispose(); + await sessions.dispose(); + await app.close(); + } -process.once("SIGINT", (signal) => { void shutdown(signal); }); -process.once("SIGTERM", (signal) => { void shutdown(signal); }); + process.once("SIGINT", (signal) => { void shutdown(signal); }); + process.once("SIGTERM", (signal) => { void shutdown(signal); }); -const portValue = process.env["PI_WEB_SESSIOND_PORT"]; -const port = portValue !== undefined && portValue !== "" ? Number(portValue) : undefined; -const host = process.env["PI_WEB_SESSIOND_HOST"] ?? "127.0.0.1"; + const portValue = process.env["PI_WEB_SESSIOND_PORT"]; + const port = portValue !== undefined && portValue !== "" ? Number(portValue) : undefined; + const host = process.env["PI_WEB_SESSIOND_HOST"] ?? "127.0.0.1"; -if (port !== undefined) { - await app.listen({ port, host }); -} else { - const path = sessiondSocketPath(); - await mkdir(dirname(path), { recursive: true }); - await rm(path, { force: true }); - await app.listen({ path }); - process.on("exit", () => void rm(path, { force: true })); -} + if (port !== undefined) { + await app.listen({ port, host }); + } else { + const path = sessiondSocketPath(); + await mkdir(dirname(path), { recursive: true }); + await rm(path, { force: true }); + await app.listen({ path }); + process.on("exit", () => void rm(path, { force: true })); + } + }, +}); diff --git a/src/server/sessiond/sessionDaemonStartup.test.ts b/src/server/sessiond/sessionDaemonStartup.test.ts new file mode 100644 index 00000000..62961318 --- /dev/null +++ b/src/server/sessiond/sessionDaemonStartup.test.ts @@ -0,0 +1,261 @@ +import { existsSync, readFileSync } from "node:fs"; +import { mkdtemp, mkdir, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { + migrateLegacySessionArchive, + type LegacySessionArchiveMigrationOptions, + type LegacySessionArchiveMigrationResult, +} from "../sessions/sessionArchiveMigration.js"; +import { runSessionDaemonStartup } from "./sessionDaemonStartup.js"; + +const tempRoots: string[] = []; + +describe("session daemon archive migration startup", () => { + afterEach(async () => { + await Promise.all(tempRoots.splice(0).map((path) => rm(path, { recursive: true, force: true }))); + }); + + it("finishes an eligible migration before constructing PiSessionService, registering archive routes, or listening", async () => { + const fixture = await createLegacyArchiveFixture(); + const logger = createLogger(); + const boundaries: string[] = []; + const runtime = { ready: true }; + + const startedRuntime = await runSessionDaemonStartup({ + logger, + migrateArchive: () => migrateLegacySessionArchive(fixture.options), + createRuntime() { + boundaries.push("construct-session-service"); + expectMigrationComplete(fixture); + return runtime; + }, + registerRoutes(createdRuntime) { + boundaries.push("register-archive-routes"); + expect(createdRuntime).toBe(runtime); + expectMigrationComplete(fixture); + }, + listen(createdRuntime) { + boundaries.push("listen"); + expect(createdRuntime).toBe(runtime); + expectMigrationComplete(fixture); + return Promise.resolve(); + }, + }); + + expect(startedRuntime).toBe(runtime); + expect(boundaries).toEqual([ + "construct-session-service", + "register-archive-routes", + "listen", + ]); + expect(logger.info).toHaveBeenCalledWith( + { archiveFileCount: 1 }, + "migrated legacy session archive to the configured PI_WEB_DATA_DIR", + ); + }); + + it("does not initialize archive consumers while a mutation-free eligibility check is pending", async () => { + const logger = createLogger(); + const migration = deferred(); + const boundaries: string[] = []; + + const startup = runSessionDaemonStartup({ + logger, + migrateArchive: () => migration.promise, + createRuntime() { + boundaries.push("construct-session-service"); + return { ready: true }; + }, + registerRoutes() { + boundaries.push("register-archive-routes"); + }, + listen() { + boundaries.push("listen"); + return Promise.resolve(); + }, + }); + + expect(boundaries).toEqual([]); + migration.resolve({ status: "skipped", reason: "legacy-index-missing" }); + await startup; + + expect(boundaries).toEqual([ + "construct-session-service", + "register-archive-routes", + "listen", + ]); + expect(logger.debug).toHaveBeenCalledWith( + { reason: "legacy-index-missing" }, + "legacy session archive migration is not eligible; continuing session daemon startup", + ); + }); + + it("warns and continues normal startup when eligibility inspection is inconclusive", async () => { + const logger = createLogger(); + const inspectionError = Object.assign(new Error("permission denied"), { code: "EACCES" }); + const createRuntime = vi.fn(() => ({ ready: true })); + const registerRoutes = vi.fn(); + const listen = vi.fn(() => Promise.resolve()); + + await runSessionDaemonStartup({ + logger, + migrateArchive: () => Promise.resolve({ + status: "skipped", + reason: "inspection-failed", + error: inspectionError, + }), + createRuntime, + registerRoutes, + listen, + }); + + expect(logger.warn).toHaveBeenCalledWith( + { err: inspectionError, reason: "inspection-failed" }, + "could not inspect legacy session archive migration eligibility; continuing session daemon startup without migration", + ); + expect(createRuntime).toHaveBeenCalledOnce(); + expect(registerRoutes).toHaveBeenCalledOnce(); + expect(listen).toHaveBeenCalledOnce(); + }); + + it("logs an eligible migration failure and stops before archive consumers can mutate destination state", async () => { + const logger = createLogger(); + const migrationError = new Error("copy failed"); + const rollbackError = new Error("rollback failed"); + const createRuntime = vi.fn(() => ({ ready: true })); + const registerRoutes = vi.fn(); + const listen = vi.fn(() => Promise.resolve()); + + await expect(runSessionDaemonStartup({ + logger, + migrateArchive: () => Promise.resolve({ + status: "failed", + phase: "publish-files", + error: migrationError, + rollbackErrors: [rollbackError], + }), + createRuntime, + registerRoutes, + listen, + })).rejects.toThrow("Legacy session archive migration failed during publish-files; session daemon startup stopped"); + + expect(logger.error).toHaveBeenCalledWith( + { + err: migrationError, + phase: "publish-files", + rollbackErrorCount: 1, + rollbackErrors: [rollbackError], + }, + "legacy session archive migration failed before commit and rollback was incomplete; stopping session daemon startup", + ); + expect(createRuntime).not.toHaveBeenCalled(); + expect(registerRoutes).not.toHaveBeenCalled(); + expect(listen).not.toHaveBeenCalled(); + }); + + it("warns but starts from the committed destination when legacy cleanup is incomplete", async () => { + const logger = createLogger(); + const cleanupError = new Error("legacy index could not be removed"); + const createRuntime = vi.fn(() => ({ ready: true })); + const registerRoutes = vi.fn(); + const listen = vi.fn(() => Promise.resolve()); + + await runSessionDaemonStartup({ + logger, + migrateArchive: () => Promise.resolve({ + status: "migrated", + archiveFileCount: 2, + cleanup: "incomplete", + cleanupErrors: [cleanupError], + }), + createRuntime, + registerRoutes, + listen, + }); + + expect(logger.warn).toHaveBeenCalledWith( + { + archiveFileCount: 2, + cleanupErrorCount: 1, + cleanupErrors: [cleanupError], + }, + "legacy session archive migration committed but cleanup was incomplete; continuing with the migrated destination archive", + ); + expect(createRuntime).toHaveBeenCalledOnce(); + expect(registerRoutes).toHaveBeenCalledOnce(); + expect(listen).toHaveBeenCalledOnce(); + }); +}); + +interface LegacyArchiveFixture { + legacyIndexPath: string; + legacyFilePath: string; + destinationIndexPath: string; + destinationFilePath: string; + options: LegacySessionArchiveMigrationOptions; +} + +async function createLegacyArchiveFixture(): Promise { + const root = await mkdtemp(join(tmpdir(), "pi-web-sessiond-startup-")); + tempRoots.push(root); + const homeDir = join(root, "home"); + const legacyRoot = join(homeDir, ".pi-web"); + const legacyArchiveDir = join(legacyRoot, "archived-sessions"); + const legacyIndexPath = join(legacyRoot, "archived-sessions.json"); + const legacyFilePath = join(legacyArchiveDir, "legacy-session.jsonl"); + const destinationRoot = join(root, "managed-state"); + const destinationIndexPath = join(destinationRoot, "archived-sessions.json"); + const destinationFilePath = join(destinationRoot, "archived-sessions", "legacy-session.jsonl"); + + await mkdir(legacyArchiveDir, { recursive: true }); + await writeFile(legacyFilePath, "legacy session\n", "utf8"); + await writeFile(legacyIndexPath, `${JSON.stringify({ + sessions: [{ + sessionId: "legacy-session", + cwd: join(root, "workspace"), + archivedAt: "2026-01-01T00:00:00.000Z", + archivePath: legacyFilePath, + }], + }, null, 2)}\n`, "utf8"); + + return { + legacyIndexPath, + legacyFilePath, + destinationIndexPath, + destinationFilePath, + options: { + env: { PI_WEB_DATA_DIR: destinationRoot }, + cwd: root, + homeDir, + createAttemptId: () => "sessiond-startup-test", + }, + }; +} + +function expectMigrationComplete(fixture: LegacyArchiveFixture): void { + expect(existsSync(fixture.legacyIndexPath)).toBe(false); + expect(existsSync(fixture.legacyFilePath)).toBe(false); + expect(readFileSync(fixture.destinationFilePath, "utf8")).toBe("legacy session\n"); + expect(JSON.parse(readFileSync(fixture.destinationIndexPath, "utf8"))).toMatchObject({ + sessions: [{ archivePath: fixture.destinationFilePath }], + }); +} + +function createLogger() { + return { + debug: vi.fn(), + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + }; +} + +function deferred() { + let resolve!: (value: T | PromiseLike) => void; + const promise = new Promise((resolvePromise) => { + resolve = resolvePromise; + }); + return { promise, resolve }; +} diff --git a/src/server/sessiond/sessionDaemonStartup.ts b/src/server/sessiond/sessionDaemonStartup.ts new file mode 100644 index 00000000..b0c645d5 --- /dev/null +++ b/src/server/sessiond/sessionDaemonStartup.ts @@ -0,0 +1,95 @@ +import { + migrateLegacySessionArchive, + type LegacySessionArchiveMigrationResult, +} from "../sessions/sessionArchiveMigration.js"; + +export interface SessionDaemonStartupLogger { + debug(details: Record, message: string): void; + info(details: Record, message: string): void; + warn(details: Record, message: string): void; + error(details: Record, message: string): void; +} + +export interface SessionDaemonStartupSteps { + logger: SessionDaemonStartupLogger; + createRuntime(): Runtime; + registerRoutes(runtime: Runtime): void; + listen(runtime: Runtime): Promise; + migrateArchive?: () => Promise; +} + +/** + * Keeps archive migration ahead of every archive-state consumer in sessiond. + * A failed eligible migration stops startup so runtime writes cannot make a + * clean retry ambiguous; mutation-free eligibility skips still start normally. + */ +export async function runSessionDaemonStartup( + steps: SessionDaemonStartupSteps, +): Promise { + const result = await (steps.migrateArchive ?? migrateLegacySessionArchive)(); + reportMigrationResult(result, steps.logger); + + if (result.status === "failed") { + throw new Error( + `Legacy session archive migration failed during ${result.phase}; session daemon startup stopped`, + { cause: result.error }, + ); + } + + const runtime = steps.createRuntime(); + steps.registerRoutes(runtime); + await steps.listen(runtime); + return runtime; +} + +function reportMigrationResult( + result: LegacySessionArchiveMigrationResult, + logger: SessionDaemonStartupLogger, +): void { + if (result.status === "skipped") { + if (result.reason === "inspection-failed") { + logger.warn( + { err: result.error, reason: result.reason }, + "could not inspect legacy session archive migration eligibility; continuing session daemon startup without migration", + ); + } else { + logger.debug( + { reason: result.reason }, + "legacy session archive migration is not eligible; continuing session daemon startup", + ); + } + return; + } + + if (result.status === "failed") { + logger.error( + { + err: result.error, + phase: result.phase, + rollbackErrorCount: result.rollbackErrors.length, + rollbackErrors: result.rollbackErrors, + }, + result.rollbackErrors.length === 0 + ? "legacy session archive migration failed before commit; stopping session daemon startup" + : "legacy session archive migration failed before commit and rollback was incomplete; stopping session daemon startup", + ); + return; + } + + if (result.cleanup === "incomplete") { + logger.warn( + { + archiveFileCount: result.archiveFileCount, + cleanupErrorCount: result.cleanupErrors.length, + cleanupErrors: result.cleanupErrors, + }, + "legacy session archive migration committed but cleanup was incomplete; continuing with the migrated destination archive", + ); + return; + } + + logger.info( + { archiveFileCount: result.archiveFileCount }, + "migrated legacy session archive to the configured PI_WEB_DATA_DIR", + ); +} diff --git a/src/server/sessions/sessionArchiveMigration.test.ts b/src/server/sessions/sessionArchiveMigration.test.ts new file mode 100644 index 00000000..876534c4 --- /dev/null +++ b/src/server/sessions/sessionArchiveMigration.test.ts @@ -0,0 +1,555 @@ +import { constants } from "node:fs"; +import { + access, + appendFile, + copyFile, + link as createHardLink, + lstat, + mkdtemp, + mkdir, + readFile, + readdir, + realpath, + rm, + unlink, + writeFile, +} from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { dirname, join, resolve, sep } from "node:path"; +import { afterEach, describe, expect, it } from "vitest"; +import { + inspectLegacySessionArchiveMigration, + migrateLegacySessionArchive, + type LegacySessionArchiveMigrationOptions, +} from "./sessionArchiveMigration.js"; + +const tempRoots: string[] = []; + +describe("legacy session archive migration preflight", () => { + afterEach(async () => { + await Promise.all(tempRoots.splice(0).map((path) => rm(path, { recursive: true, force: true }))); + }); + + it("proves the ordinary legacy layout eligible without creating destination state", async () => { + const fixture = await createLegacyArchiveFixture(); + + await expect(inspectLegacySessionArchiveMigration(fixture.options)).resolves.toEqual({ + status: "eligible", + legacyIndexPath: fixture.legacyIndexPath, + destinationIndexPath: fixture.destinationIndexPath, + archiveFileCount: 1, + }); + expect(await exists(fixture.destinationRoot)).toBe(false); + await expect(readFile(fixture.legacyFilePath, "utf8")).resolves.toBe("legacy session\n"); + }); + + it("skips when PI_WEB_DATA_DIR is not explicitly configured", async () => { + const fixture = await createLegacyArchiveFixture(); + + await expect(migrateLegacySessionArchive({ ...fixture.options, env: {} })).resolves.toEqual({ + status: "skipped", + reason: "data-dir-not-configured", + }); + await expectLegacyStateUntouched(fixture); + expect(await exists(fixture.destinationRoot)).toBe(false); + }); + + it("skips when the configured data root resolves to the legacy root", async () => { + const fixture = await createLegacyArchiveFixture(); + + await expect(migrateLegacySessionArchive({ + ...fixture.options, + env: { PI_WEB_DATA_DIR: fixture.legacyRoot }, + })).resolves.toEqual({ status: "skipped", reason: "data-dir-not-distinct" }); + await expectLegacyStateUntouched(fixture); + }); + + it("skips when no legacy index exists without adopting loose legacy files", async () => { + const fixture = await createLegacyArchiveFixture(); + await unlink(fixture.legacyIndexPath); + + await expect(migrateLegacySessionArchive(fixture.options)).resolves.toEqual({ + status: "skipped", + reason: "legacy-index-missing", + }); + await expect(readFile(fixture.legacyFilePath, "utf8")).resolves.toBe("legacy session\n"); + await expect(readFile(fixture.activeFilePath, "utf8")).resolves.toBe("active session\n"); + expect(await exists(fixture.destinationRoot)).toBe(false); + }); + + it("skips malformed legacy indexes without creating destination state", async () => { + const fixture = await createLegacyArchiveFixture(); + await writeFile(fixture.legacyIndexPath, "not json\n", "utf8"); + + await expect(migrateLegacySessionArchive(fixture.options)).resolves.toEqual({ + status: "skipped", + reason: "legacy-index-invalid", + }); + await expect(readFile(fixture.legacyIndexPath, "utf8")).resolves.toBe("not json\n"); + expect(await exists(fixture.destinationRoot)).toBe(false); + }); + + it("treats inconclusive filesystem inspection as a mutation-free skip", async () => { + const fixture = await createLegacyArchiveFixture(); + const inspectionError = Object.assign(new Error("inspection denied"), { code: "EACCES" }); + + const result = await migrateLegacySessionArchive({ + ...fixture.options, + fileSystem: { + lstat: (path) => path === fixture.destinationIndexPath + ? Promise.reject(inspectionError) + : lstat(path), + }, + }); + + expect(result).toMatchObject({ + status: "skipped", + reason: "inspection-failed", + error: inspectionError, + }); + await expectLegacyStateUntouched(fixture); + expect(await exists(fixture.destinationRoot)).toBe(false); + }); + + it("does not merge with or overwrite an existing destination index", async () => { + const fixture = await createLegacyArchiveFixture(); + await mkdir(fixture.destinationRoot, { recursive: true }); + await writeFile(fixture.destinationIndexPath, "destination owner\n", "utf8"); + + await expect(migrateLegacySessionArchive(fixture.options)).resolves.toEqual({ + status: "skipped", + reason: "destination-index-exists", + }); + await expect(readFile(fixture.destinationIndexPath, "utf8")).resolves.toBe("destination owner\n"); + await expectLegacyStateUntouched(fixture); + }); + + it("leaves an interrupted pre-commit file untouched instead of adopting a non-empty destination archive", async () => { + const fixture = await createLegacyArchiveFixture({ createDestinationArchive: true }); + await writeFile(fixture.destinationFilePath, "partial destination copy\n", "utf8"); + + await expect(migrateLegacySessionArchive(fixture.options)).resolves.toEqual({ + status: "skipped", + reason: "destination-archive-not-empty-or-invalid", + }); + await expect(readFile(fixture.destinationFilePath, "utf8")).resolves.toBe("partial destination copy\n"); + await expectLegacyStateUntouched(fixture); + }); + + it("skips archive paths that are not direct regular files in the legacy archive directory", async () => { + const fixture = await createLegacyArchiveFixture(); + const outsidePath = join(fixture.root, "outside.jsonl"); + await writeFile(outsidePath, "outside\n", "utf8"); + const firstRecord = requiredRecord(fixture.document.sessions, 0); + const changedDocument = { + ...fixture.document, + sessions: [{ ...firstRecord, archivePath: outsidePath }, ...fixture.document.sessions.slice(1)], + }; + await writeArchiveIndex(fixture.legacyIndexPath, changedDocument); + + await expect(migrateLegacySessionArchive(fixture.options)).resolves.toEqual({ + status: "skipped", + reason: "legacy-archive-layout-invalid", + }); + await expect(readFile(outsidePath, "utf8")).resolves.toBe("outside\n"); + await expect(readFile(fixture.legacyFilePath, "utf8")).resolves.toBe("legacy session\n"); + expect(await exists(fixture.destinationRoot)).toBe(false); + }); + + it("uses Windows case-insensitive path semantics without weakening Linux containment", async () => { + const fixture = await createLegacyArchiveFixture(); + const casedArchivePath = resolve(fixture.legacyFilePath.toUpperCase()); + const firstRecord = requiredRecord(fixture.document.sessions, 0); + await writeArchiveIndex(fixture.legacyIndexPath, { + ...fixture.document, + sessions: [ + { ...firstRecord, archivePath: casedArchivePath }, + ...fixture.document.sessions.slice(1), + ], + }); + const mapCaseVariant = (path: string): string => path.toLowerCase() === casedArchivePath.toLowerCase() + ? fixture.legacyFilePath + : path; + const fileSystem = { + lstat: (path: string) => lstat(mapCaseVariant(path)), + realpath: (path: string) => realpath(mapCaseVariant(path)), + }; + + await expect(inspectLegacySessionArchiveMigration({ + ...fixture.options, + platform: "linux", + fileSystem, + })).resolves.toEqual({ status: "skipped", reason: "legacy-archive-layout-invalid" }); + await expect(inspectLegacySessionArchiveMigration({ + ...fixture.options, + platform: "win32", + fileSystem, + })).resolves.toEqual({ + status: "eligible", + legacyIndexPath: fixture.legacyIndexPath, + destinationIndexPath: fixture.destinationIndexPath, + archiveFileCount: 1, + }); + }); + + it("skips legacy directories containing unindexed entries", async () => { + const fixture = await createLegacyArchiveFixture(); + const unexpectedPath = join(fixture.legacyArchiveDir, "unexpected.tmp"); + await writeFile(unexpectedPath, "unexpected\n", "utf8"); + + await expect(migrateLegacySessionArchive(fixture.options)).resolves.toEqual({ + status: "skipped", + reason: "legacy-archive-layout-invalid", + }); + await expect(readFile(unexpectedPath, "utf8")).resolves.toBe("unexpected\n"); + await expectLegacyStateUntouched(fixture); + expect(await exists(fixture.destinationRoot)).toBe(false); + }); + + it("skips duplicate session IDs", async () => { + const fixture = await createLegacyArchiveFixture(); + const firstRecord = requiredRecord(fixture.document.sessions, 0); + const secondRecord = requiredRecord(fixture.document.sessions, 1); + await writeArchiveIndex(fixture.legacyIndexPath, { + ...fixture.document, + sessions: [firstRecord, { ...secondRecord, sessionId: firstRecord["sessionId"] }], + }); + + await expect(migrateLegacySessionArchive(fixture.options)).resolves.toEqual({ + status: "skipped", + reason: "archive-record-conflict", + }); + await expect(readFile(fixture.legacyFilePath, "utf8")).resolves.toBe("legacy session\n"); + expect(await exists(fixture.destinationRoot)).toBe(false); + }); + + it("skips duplicate source and destination file mappings", async () => { + const fixture = await createLegacyArchiveFixture(); + const firstRecord = requiredRecord(fixture.document.sessions, 0); + await writeArchiveIndex(fixture.legacyIndexPath, { + ...fixture.document, + sessions: [firstRecord, { ...firstRecord, sessionId: "second-file-record" }], + }); + + await expect(migrateLegacySessionArchive(fixture.options)).resolves.toEqual({ + status: "skipped", + reason: "archive-record-conflict", + }); + await expectLegacyStateUntouched({ + ...fixture, + sourceIndexContents: await readFile(fixture.legacyIndexPath, "utf8"), + }); + expect(await exists(fixture.destinationRoot)).toBe(false); + }); +}); + +describe("legacy session archive migration execution", () => { + afterEach(async () => { + await Promise.all(tempRoots.splice(0).map((path) => rm(path, { recursive: true, force: true }))); + }); + + it("copies across the source boundary, atomically publishes the rewritten index, then removes legacy state", async () => { + const fixture = await createLegacyArchiveFixture({ createDestinationArchive: true }); + const copies: { source: string; destination: string; mode: number }[] = []; + const links: { source: string; destination: string }[] = []; + + await expect(migrateLegacySessionArchive({ + ...fixture.options, + fileSystem: { + copyFile: async (source, destination, mode) => { + copies.push({ source, destination, mode }); + await copyFile(source, destination, mode); + }, + link: async (source, destination) => { + links.push({ source, destination }); + await createHardLink(source, destination); + }, + unlink: async (path) => { + if (path === fixture.legacyFilePath || path === fixture.legacyIndexPath) { + expect(await exists(fixture.destinationIndexPath)).toBe(true); + } + await unlink(path); + }, + }, + })).resolves.toEqual({ status: "migrated", archiveFileCount: 1, cleanup: "complete" }); + + expect(copies).toHaveLength(2); + expect(copies[0]).toMatchObject({ source: fixture.legacyFilePath, mode: constants.COPYFILE_EXCL }); + expect(copies[0]?.destination).toContain(".archived-sessions-migration-test-attempt"); + expect(copies[1]).toMatchObject({ destination: fixture.destinationFilePath, mode: constants.COPYFILE_EXCL }); + expect(links).toEqual([{ + source: join(fixture.destinationRoot, ".archived-sessions-migration-test-attempt", "archived-sessions.json"), + destination: fixture.destinationIndexPath, + }]); + await expect(readFile(fixture.destinationFilePath, "utf8")).resolves.toBe("legacy session\n"); + + const migratedDocument: unknown = JSON.parse(await readFile(fixture.destinationIndexPath, "utf8")); + const firstRecord = requiredRecord(fixture.document.sessions, 0); + expect(migratedDocument).toEqual({ + ...fixture.document, + sessions: [ + { ...firstRecord, archivePath: fixture.destinationFilePath }, + requiredRecord(fixture.document.sessions, 1), + ], + }); + expect(await exists(fixture.legacyIndexPath)).toBe(false); + expect(await exists(fixture.legacyFilePath)).toBe(false); + expect(await exists(fixture.legacyArchiveDir)).toBe(false); + await expect(readFile(fixture.activeFilePath, "utf8")).resolves.toBe("active session\n"); + expect(new Set(await readdir(fixture.destinationRoot))).toEqual(new Set([ + "archived-sessions", + "archived-sessions.json", + ])); + }); + + it("retries safely when an interrupted staging-only attempt left an unowned sibling tree", async () => { + const fixture = await createLegacyArchiveFixture(); + const abandonedFile = join( + fixture.destinationRoot, + ".archived-sessions-migration-interrupted-attempt", + "files", + "abandoned.jsonl", + ); + await mkdir(dirname(abandonedFile), { recursive: true }); + await writeFile(abandonedFile, "unowned staging data\n", "utf8"); + + await expect(migrateLegacySessionArchive(fixture.options)).resolves.toEqual({ + status: "migrated", + archiveFileCount: 1, + cleanup: "complete", + }); + + await expect(readFile(abandonedFile, "utf8")).resolves.toBe("unowned staging data\n"); + await expect(readFile(fixture.destinationFilePath, "utf8")).resolves.toBe("legacy session\n"); + expect(await exists(fixture.destinationIndexPath)).toBe(true); + expect(await exists(fixture.legacyIndexPath)).toBe(false); + }); + + it("rolls back destination artifacts and preserves all legacy state when staged-copy verification fails", async () => { + const fixture = await createLegacyArchiveFixture(); + + const result = await migrateLegacySessionArchive({ + ...fixture.options, + fileSystem: { + copyFile: async (source, destination, mode) => { + await copyFile(source, destination, mode); + if (source === fixture.legacyFilePath) await appendFile(destination, "corrupt", "utf8"); + }, + }, + }); + + expect(result).toMatchObject({ status: "failed", phase: "stage", rollbackErrors: [] }); + await expectLegacyStateUntouched(fixture); + expect(await exists(fixture.destinationIndexPath)).toBe(false); + await expect(readdir(fixture.destinationRoot)).resolves.toEqual([]); + }); + + it("revalidates source files before commit and rolls back if one changes during migration", async () => { + const fixture = await createLegacyArchiveFixture(); + + const result = await migrateLegacySessionArchive({ + ...fixture.options, + fileSystem: { + copyFile: async (source, destination, mode) => { + await copyFile(source, destination, mode); + if (destination === fixture.destinationFilePath) { + await appendFile(fixture.legacyFilePath, "changed during migration\n", "utf8"); + } + }, + }, + }); + + expect(result).toMatchObject({ status: "failed", phase: "commit-index", rollbackErrors: [] }); + await expect(readFile(fixture.legacyIndexPath, "utf8")).resolves.toBe(fixture.sourceIndexContents); + await expect(readFile(fixture.legacyFilePath, "utf8")).resolves.toBe( + "legacy session\nchanged during migration\n", + ); + await expect(readFile(fixture.activeFilePath, "utf8")).resolves.toBe("active session\n"); + expect(await exists(fixture.destinationIndexPath)).toBe(false); + expect(await exists(fixture.destinationArchiveDir)).toBe(false); + }); + + it("rolls back published files but never source state when atomic index publication fails", async () => { + const fixture = await createLegacyArchiveFixture(); + const publicationError = Object.assign(new Error("link failed"), { code: "EIO" }); + + const result = await migrateLegacySessionArchive({ + ...fixture.options, + fileSystem: { + link: () => Promise.reject(publicationError), + }, + }); + + expect(result).toMatchObject({ + status: "failed", + phase: "commit-index", + error: publicationError, + rollbackErrors: [], + }); + await expectLegacyStateUntouched(fixture); + expect(await exists(fixture.destinationIndexPath)).toBe(false); + expect(await exists(fixture.destinationArchiveDir)).toBe(false); + await expect(readdir(fixture.destinationRoot)).resolves.toEqual([]); + }); + + it("does not overwrite a destination index that appears at the atomic commit boundary", async () => { + const fixture = await createLegacyArchiveFixture(); + const publicationError = Object.assign(new Error("destination index won the race"), { code: "EEXIST" }); + + const result = await migrateLegacySessionArchive({ + ...fixture.options, + fileSystem: { + link: async (_source, destination) => { + await writeFile(destination, "destination owner\n", { encoding: "utf8", flag: "wx" }); + throw publicationError; + }, + }, + }); + + expect(result).toMatchObject({ + status: "failed", + phase: "commit-index", + error: publicationError, + rollbackErrors: [], + }); + await expect(readFile(fixture.destinationIndexPath, "utf8")).resolves.toBe("destination owner\n"); + expect(await exists(fixture.destinationArchiveDir)).toBe(false); + await expectLegacyStateUntouched(fixture); + }); + + it("keeps the committed destination authoritative when legacy cleanup fails", async () => { + const fixture = await createLegacyArchiveFixture(); + const cleanupError = Object.assign(new Error("source cleanup failed"), { code: "EACCES" }); + + const result = await migrateLegacySessionArchive({ + ...fixture.options, + fileSystem: { + unlink: async (path) => { + if (path === fixture.legacyFilePath) throw cleanupError; + await unlink(path); + }, + }, + }); + + expect(result).toMatchObject({ + status: "migrated", + cleanup: "incomplete", + cleanupErrors: [cleanupError], + }); + await expect(readFile(fixture.destinationFilePath, "utf8")).resolves.toBe("legacy session\n"); + expect(await exists(fixture.destinationIndexPath)).toBe(true); + await expectLegacyStateUntouched(fixture); + }); +}); + +interface LegacyArchiveFixture { + root: string; + legacyRoot: string; + legacyIndexPath: string; + legacyArchiveDir: string; + legacyFilePath: string; + activeFilePath: string; + destinationRoot: string; + destinationIndexPath: string; + destinationArchiveDir: string; + destinationFilePath: string; + sourceIndexContents: string; + document: { marker: string; sessions: Record[] }; + options: LegacySessionArchiveMigrationOptions; +} + +async function createLegacyArchiveFixture( + setup: { createDestinationArchive?: boolean } = {}, +): Promise { + const root = await mkdtemp(join(tmpdir(), "pi-web-archive-migration-")); + tempRoots.push(root); + const homeDir = join(root, "home"); + const legacyRoot = join(homeDir, ".pi-web"); + const legacyIndexPath = join(legacyRoot, "archived-sessions.json"); + const legacyArchiveDir = join(legacyRoot, "archived-sessions"); + const legacyFilePath = join(legacyArchiveDir, "2026-01-01_file-session.jsonl"); + const activeFilePath = join(root, "active", "2026-01-01_file-session.jsonl"); + const destinationRoot = join(root, "managed-state"); + const destinationIndexPath = join(destinationRoot, "archived-sessions.json"); + const destinationArchiveDir = join(destinationRoot, "archived-sessions"); + const destinationFilePath = join(destinationArchiveDir, "2026-01-01_file-session.jsonl"); + const document = { + marker: "preserve root metadata", + sessions: [ + { + sessionId: "file-session", + cwd: `${join(root, "workspace")}${sep}..${sep}workspace`, + archivedAt: "2026-01-01T00:02:00.000Z", + originalPath: activeFilePath, + archivePath: legacyFilePath, + created: "2026-01-01T00:00:00.000Z", + modified: "2026-01-01T00:01:00.000Z", + messageCount: 2, + firstMessage: "hello", + name: "Legacy file session", + customMetadata: { preserved: true }, + }, + { + sessionId: "metadata-only", + cwd: "/workspace", + archivedAt: "2026-01-02T00:00:00.000Z", + name: "Metadata only", + }, + ], + }; + + await mkdir(legacyArchiveDir, { recursive: true }); + await mkdir(join(root, "active"), { recursive: true }); + await writeFile(legacyFilePath, "legacy session\n", "utf8"); + await writeFile(activeFilePath, "active session\n", "utf8"); + const sourceIndexContents = await writeArchiveIndex(legacyIndexPath, document); + if (setup.createDestinationArchive === true) await mkdir(destinationArchiveDir, { recursive: true }); + + return { + root, + legacyRoot, + legacyIndexPath, + legacyArchiveDir, + legacyFilePath, + activeFilePath, + destinationRoot, + destinationIndexPath, + destinationArchiveDir, + destinationFilePath, + sourceIndexContents, + document, + options: { + env: { PI_WEB_DATA_DIR: destinationRoot }, + cwd: root, + homeDir, + createAttemptId: () => "test-attempt", + }, + }; +} + +async function writeArchiveIndex(path: string, document: unknown): Promise { + const contents = `${JSON.stringify(document, null, 2)}\n`; + await writeFile(path, contents, "utf8"); + return contents; +} + +async function expectLegacyStateUntouched(fixture: LegacyArchiveFixture): Promise { + await expect(readFile(fixture.legacyIndexPath, "utf8")).resolves.toBe(fixture.sourceIndexContents); + await expect(readFile(fixture.legacyFilePath, "utf8")).resolves.toBe("legacy session\n"); + await expect(readFile(fixture.activeFilePath, "utf8")).resolves.toBe("active session\n"); +} + +function requiredRecord(records: Record[], index: number): Record { + const record = records[index]; + if (record === undefined) throw new Error(`Missing fixture record ${index.toString()}`); + return record; +} + +async function exists(path: string): Promise { + try { + await access(path, constants.F_OK); + return true; + } catch { + return false; + } +} diff --git a/src/server/sessions/sessionArchiveMigration.ts b/src/server/sessions/sessionArchiveMigration.ts new file mode 100644 index 00000000..4d43c5c7 --- /dev/null +++ b/src/server/sessions/sessionArchiveMigration.ts @@ -0,0 +1,781 @@ +import { randomUUID } from "node:crypto"; +import { constants, type Dirent, type Stats } from "node:fs"; +import { + copyFile, + link, + lstat, + mkdir, + open, + readFile, + readdir, + realpath, + rm, + rmdir, + unlink, + writeFile, +} from "node:fs/promises"; +import { homedir } from "node:os"; +import { basename, dirname, isAbsolute, join, relative, resolve } from "node:path"; +import { piWebDataDir } from "../../config.js"; +import { parseSessionArchiveFile, type ArchivedSessionRecord } from "./sessionArchiveStore.js"; + +export type LegacySessionArchiveMigrationSkipReason = + | "data-dir-not-configured" + | "data-dir-not-distinct" + | "legacy-index-missing" + | "legacy-index-invalid" + | "destination-index-exists" + | "destination-archive-not-empty-or-invalid" + | "legacy-archive-layout-invalid" + | "archive-record-conflict" + | "inspection-failed"; + +export interface LegacySessionArchiveMigrationSkipped { + status: "skipped"; + reason: LegacySessionArchiveMigrationSkipReason; + error?: unknown; +} + +export type LegacySessionArchiveMigrationPreflight = + | LegacySessionArchiveMigrationSkipped + | { + status: "eligible"; + legacyIndexPath: string; + destinationIndexPath: string; + archiveFileCount: number; + }; + +export type LegacySessionArchiveMigrationPhase = "stage" | "publish-files" | "commit-index"; + +export type LegacySessionArchiveMigrationResult = + | LegacySessionArchiveMigrationSkipped + | { + status: "failed"; + phase: LegacySessionArchiveMigrationPhase; + error: unknown; + rollbackErrors: unknown[]; + } + | { + status: "migrated"; + archiveFileCount: number; + cleanup: "complete"; + } + | { + status: "migrated"; + archiveFileCount: number; + cleanup: "incomplete"; + cleanupErrors: unknown[]; + }; + +export interface SessionArchiveMigrationReadHandle { + read(buffer: Buffer, offset: number, length: number, position: null): Promise<{ bytesRead: number }>; + close(): Promise; +} + +export interface SessionArchiveMigrationFileSystem { + lstat(path: string): Promise; + realpath(path: string): Promise; + readFile(path: string): Promise; + readdir(path: string): Promise; + mkdir(path: string, options?: { recursive?: boolean }): Promise; + copyFile(source: string, destination: string, mode: number): Promise; + writeFile(path: string, contents: string): Promise; + link(source: string, destination: string): Promise; + unlink(path: string): Promise; + rmdir(path: string): Promise; + rmOwnedTree(path: string): Promise; + open(path: string): Promise; +} + +export interface LegacySessionArchiveMigrationOptions { + env?: NodeJS.ProcessEnv; + cwd?: string; + homeDir?: string; + platform?: NodeJS.Platform; + createAttemptId?: () => string; + fileSystem?: Partial; +} + +interface PlannedArchiveFile { + sourcePath: string; + destinationPath: string; + fileName: string; +} + +interface MigrationPlan { + legacyRoot: string; + canonicalLegacyRoot: string; + legacyIndexPath: string; + legacyArchiveDir: string; + canonicalLegacyArchiveDir: string; + legacyArchiveDirExists: boolean; + destinationRoot: string; + canonicalDestinationRoot: string; + destinationIndexPath: string; + destinationArchiveDir: string; + canonicalDestinationArchiveDir: string; + destinationArchiveDirExists: boolean; + sourceIndexContents: string; + destinationIndexContents: string; + files: PlannedArchiveFile[]; + platform: NodeJS.Platform; +} + +interface InternalEligiblePreflight { + status: "eligible"; + plan: MigrationPlan; +} + +type InternalPreflight = LegacySessionArchiveMigrationSkipped | InternalEligiblePreflight; + +const defaultFileSystem: SessionArchiveMigrationFileSystem = { + lstat: async (path) => lstat(path), + realpath: async (path) => realpath(path), + readFile: async (path) => readFile(path, "utf8"), + readdir: async (path) => readdir(path, { withFileTypes: true }), + mkdir: async (path, options) => { + await mkdir(path, options); + }, + copyFile: async (source, destination, mode) => copyFile(source, destination, mode), + writeFile: async (path, contents) => { + await writeFile(path, contents, { encoding: "utf8", flag: "wx" }); + }, + link: async (source, destination) => link(source, destination), + unlink: async (path) => unlink(path), + rmdir: async (path) => rmdir(path), + rmOwnedTree: async (path) => { + await rm(path, { recursive: true, force: true }); + }, + open: async (path) => open(path, "r"), +}; + +export async function inspectLegacySessionArchiveMigration( + options: LegacySessionArchiveMigrationOptions = {}, +): Promise { + const preflight = await buildMigrationPreflight(options, migrationFileSystem(options)); + if (preflight.status === "skipped") return preflight; + return { + status: "eligible", + legacyIndexPath: preflight.plan.legacyIndexPath, + destinationIndexPath: preflight.plan.destinationIndexPath, + archiveFileCount: preflight.plan.files.length, + }; +} + +export async function migrateLegacySessionArchive( + options: LegacySessionArchiveMigrationOptions = {}, +): Promise { + const fileSystem = migrationFileSystem(options); + const preflight = await buildMigrationPreflight(options, fileSystem); + if (preflight.status === "skipped") return preflight; + + const plan = preflight.plan; + let phase: LegacySessionArchiveMigrationPhase = "stage"; + let stagingRoot: string | undefined; + let stagingCreated = false; + let destinationArchiveCreated = false; + const ownedDestinationFiles: string[] = []; + + try { + await fileSystem.mkdir(plan.destinationRoot, { recursive: true }); + const currentDestinationRoot = await canonicalizeAllowMissing(plan.destinationRoot, fileSystem); + if (!pathsEqual(currentDestinationRoot, plan.canonicalDestinationRoot, plan.platform)) { + throw new Error("Destination data root changed after migration preflight"); + } + await assertDestinationStateUnchanged(plan, fileSystem); + + const attemptId = safeAttemptId((options.createAttemptId ?? randomUUID)()); + // Keep staging beside, not inside, the authoritative archive directory so an + // interrupted staging copy cannot be mistaken for destination archive data. + stagingRoot = join(plan.destinationRoot, `.archived-sessions-migration-${attemptId}`); + await fileSystem.mkdir(stagingRoot); + stagingCreated = true; + const stagingFilesDir = join(stagingRoot, "files"); + await fileSystem.mkdir(stagingFilesDir); + + const stagedFiles = new Map(); + for (const file of plan.files) { + const stagedPath = join(stagingFilesDir, file.fileName); + await fileSystem.copyFile(file.sourcePath, stagedPath, constants.COPYFILE_EXCL); + if (!await filesEqual(file.sourcePath, stagedPath, fileSystem)) { + throw new Error(`Staged archive file verification failed: ${file.fileName}`); + } + stagedFiles.set(file.destinationPath, stagedPath); + } + + const stagedIndexPath = join(stagingRoot, "archived-sessions.json"); + await fileSystem.writeFile(stagedIndexPath, plan.destinationIndexContents); + if (await fileSystem.readFile(stagedIndexPath) !== plan.destinationIndexContents) { + throw new Error("Staged archive index verification failed"); + } + + phase = "publish-files"; + await assertDestinationStateUnchanged(plan, fileSystem); + if (plan.files.length > 0 && !plan.destinationArchiveDirExists) { + await fileSystem.mkdir(plan.destinationArchiveDir); + destinationArchiveCreated = true; + } + + for (const file of plan.files) { + const stagedPath = stagedFiles.get(file.destinationPath); + if (stagedPath === undefined) throw new Error(`Missing staged archive file: ${file.fileName}`); + try { + await fileSystem.copyFile(stagedPath, file.destinationPath, constants.COPYFILE_EXCL); + ownedDestinationFiles.push(file.destinationPath); + } catch (error: unknown) { + if (!isNodeErrorWithCode(error, "EEXIST")) ownedDestinationFiles.push(file.destinationPath); + throw error; + } + if (!await filesEqual(stagedPath, file.destinationPath, fileSystem)) { + throw new Error(`Published archive file verification failed: ${file.fileName}`); + } + } + + phase = "commit-index"; + await assertPlanReadyToCommit(plan, stagedIndexPath, fileSystem); + // A same-directory hard link atomically publishes the already-verified index + // and, unlike rename on POSIX, fails rather than replacing an existing index. + await fileSystem.link(stagedIndexPath, plan.destinationIndexPath); + } catch (error: unknown) { + const rollbackErrors = await rollbackUncommittedDestination({ + stagingRoot, + stagingCreated, + destinationArchiveDir: plan.destinationArchiveDir, + destinationArchiveCreated, + ownedDestinationFiles, + }, fileSystem); + return { status: "failed", phase, error, rollbackErrors }; + } + + const cleanupErrors: unknown[] = []; + try { + await fileSystem.rmOwnedTree(stagingRoot); + } catch (error: unknown) { + cleanupErrors.push(error); + } + cleanupErrors.push(...await removeCommittedLegacyState(plan, fileSystem)); + + return cleanupErrors.length === 0 + ? { status: "migrated", archiveFileCount: plan.files.length, cleanup: "complete" } + : { status: "migrated", archiveFileCount: plan.files.length, cleanup: "incomplete", cleanupErrors }; +} + +async function buildMigrationPreflight( + options: LegacySessionArchiveMigrationOptions, + fileSystem: SessionArchiveMigrationFileSystem, +): Promise { + const env = options.env ?? process.env; + const configuredDataDir = env["PI_WEB_DATA_DIR"]; + if (configuredDataDir === undefined || configuredDataDir.trim() === "") { + return migrationSkipped("data-dir-not-configured"); + } + + const cwd = options.cwd ?? process.cwd(); + const platform = options.platform ?? process.platform; + const legacyRoot = resolve(options.homeDir ?? homedir(), ".pi-web"); + const destinationRoot = piWebDataDir(env, cwd); + const legacyIndexPath = join(legacyRoot, "archived-sessions.json"); + const legacyArchiveDir = join(legacyRoot, "archived-sessions"); + const destinationIndexPath = join(destinationRoot, "archived-sessions.json"); + const destinationArchiveDir = join(destinationRoot, "archived-sessions"); + + try { + const canonicalLegacyRoot = await canonicalizeAllowMissing(legacyRoot, fileSystem); + const canonicalDestinationRoot = await canonicalizeAllowMissing(destinationRoot, fileSystem); + if (pathsEqual(canonicalLegacyRoot, canonicalDestinationRoot, platform)) { + return migrationSkipped("data-dir-not-distinct"); + } + + const legacyIndexStats = await lstatIfExists(legacyIndexPath, fileSystem); + if (legacyIndexStats === undefined) return migrationSkipped("legacy-index-missing"); + if (!legacyIndexStats.isFile() || legacyIndexStats.isSymbolicLink()) { + return migrationSkipped("legacy-index-invalid"); + } + + const sourceIndexContents = await fileSystem.readFile(legacyIndexPath); + const parsedDocument = parseArchiveDocument(sourceIndexContents); + if (parsedDocument === undefined) return migrationSkipped("legacy-index-invalid"); + + if (await lstatIfExists(destinationIndexPath, fileSystem) !== undefined) { + return migrationSkipped("destination-index-exists"); + } + + const destinationArchiveStats = await lstatIfExists(destinationArchiveDir, fileSystem); + let destinationArchiveDirExists = false; + if (destinationArchiveStats !== undefined) { + if (!destinationArchiveStats.isDirectory() || destinationArchiveStats.isSymbolicLink()) { + return migrationSkipped("destination-archive-not-empty-or-invalid"); + } + if ((await fileSystem.readdir(destinationArchiveDir)).length !== 0) { + return migrationSkipped("destination-archive-not-empty-or-invalid"); + } + destinationArchiveDirExists = true; + } + + const legacyArchiveStats = await lstatIfExists(legacyArchiveDir, fileSystem); + if (legacyArchiveStats !== undefined && (!legacyArchiveStats.isDirectory() || legacyArchiveStats.isSymbolicLink())) { + return migrationSkipped("legacy-archive-layout-invalid"); + } + const legacyArchiveDirExists = legacyArchiveStats !== undefined; + const canonicalLegacyArchiveDir = legacyArchiveDirExists + ? await fileSystem.realpath(legacyArchiveDir) + : await canonicalizeAllowMissing(legacyArchiveDir, fileSystem); + const canonicalDestinationArchiveDir = destinationArchiveDirExists + ? await fileSystem.realpath(destinationArchiveDir) + : join(canonicalDestinationRoot, "archived-sessions"); + if (pathsOverlap(canonicalLegacyArchiveDir, canonicalDestinationArchiveDir, platform)) { + return migrationSkipped("data-dir-not-distinct"); + } + + const plannedFiles = await planArchiveFiles({ + sessions: parsedDocument.sessions, + legacyArchiveDir, + canonicalLegacyArchiveDir, + legacyArchiveDirExists, + destinationArchiveDir, + canonicalDestinationArchiveDir, + platform, + }, fileSystem); + if (plannedFiles.status === "skipped") return plannedFiles; + + if (!await legacyDirectoryMatchesPlan(legacyArchiveDir, legacyArchiveDirExists, plannedFiles.files, platform, fileSystem)) { + return migrationSkipped("legacy-archive-layout-invalid"); + } + + const rewrittenSessions = parsedDocument.rawSessions.map((record, index) => { + const destinationPath = plannedFiles.destinationPathsByRecord[index]; + return destinationPath === undefined ? record : { ...record, archivePath: destinationPath }; + }); + const destinationIndexContents = `${JSON.stringify({ ...parsedDocument.rawDocument, sessions: rewrittenSessions }, null, 2)}\n`; + + return { + status: "eligible", + plan: { + legacyRoot, + canonicalLegacyRoot, + legacyIndexPath, + legacyArchiveDir, + canonicalLegacyArchiveDir, + legacyArchiveDirExists, + destinationRoot, + canonicalDestinationRoot, + destinationIndexPath, + destinationArchiveDir, + canonicalDestinationArchiveDir, + destinationArchiveDirExists, + sourceIndexContents, + destinationIndexContents, + files: plannedFiles.files, + platform, + }, + }; + } catch (error: unknown) { + return migrationSkipped("inspection-failed", error); + } +} + +interface ParsedArchiveDocument { + rawDocument: Record; + rawSessions: Record[]; + sessions: ArchivedSessionRecord[]; +} + +function parseArchiveDocument(contents: string): ParsedArchiveDocument | undefined { + try { + const value: unknown = JSON.parse(contents); + if (!isRecord(value)) return undefined; + const rawSessions = recordArray(value["sessions"]); + if (rawSessions === undefined) return undefined; + const archive = parseSessionArchiveFile(value); + return { rawDocument: value, rawSessions, sessions: archive.sessions }; + } catch { + return undefined; + } +} + +async function planArchiveFiles( + input: { + sessions: ArchivedSessionRecord[]; + legacyArchiveDir: string; + canonicalLegacyArchiveDir: string; + legacyArchiveDirExists: boolean; + destinationArchiveDir: string; + canonicalDestinationArchiveDir: string; + platform: NodeJS.Platform; + }, + fileSystem: SessionArchiveMigrationFileSystem, +): Promise { + const sessionIds = new Set(); + const sourcePaths = new Set(); + const destinationPaths = new Set(); + const destinationPathsByRecord: (string | undefined)[] = Array.from({ length: input.sessions.length }); + const files: PlannedArchiveFile[] = []; + + for (const [index, session] of input.sessions.entries()) { + if (session.sessionId.trim() === "" || sessionIds.has(session.sessionId)) { + return migrationSkipped("archive-record-conflict"); + } + sessionIds.add(session.sessionId); + + if (session.archivePath === undefined) continue; + if (!input.legacyArchiveDirExists || !isAbsolute(session.archivePath)) { + return migrationSkipped("legacy-archive-layout-invalid"); + } + + const sourcePath = resolve(session.archivePath); + if (!pathsEqual(dirname(sourcePath), input.legacyArchiveDir, input.platform)) { + return migrationSkipped("legacy-archive-layout-invalid"); + } + const sourceStats = await lstatIfExists(sourcePath, fileSystem); + if (sourceStats === undefined || !sourceStats.isFile() || sourceStats.isSymbolicLink()) { + return migrationSkipped("legacy-archive-layout-invalid"); + } + const canonicalSourcePath = await fileSystem.realpath(sourcePath); + if (!pathsEqual(dirname(canonicalSourcePath), input.canonicalLegacyArchiveDir, input.platform)) { + return migrationSkipped("legacy-archive-layout-invalid"); + } + + const fileName = basename(sourcePath); + const destinationPath = join(input.destinationArchiveDir, fileName); + const canonicalDestinationPath = join(input.canonicalDestinationArchiveDir, fileName); + const sourceKey = pathKey(canonicalSourcePath, input.platform); + const destinationKey = collisionKey(canonicalDestinationPath); + if (sourcePaths.has(sourceKey) || destinationPaths.has(destinationKey)) { + return migrationSkipped("archive-record-conflict"); + } + sourcePaths.add(sourceKey); + destinationPaths.add(destinationKey); + destinationPathsByRecord[index] = destinationPath; + files.push({ sourcePath, destinationPath, fileName }); + } + + return { status: "planned", files, destinationPathsByRecord }; +} + +async function legacyDirectoryMatchesPlan( + legacyArchiveDir: string, + legacyArchiveDirExists: boolean, + files: PlannedArchiveFile[], + platform: NodeJS.Platform, + fileSystem: SessionArchiveMigrationFileSystem, +): Promise { + const archiveStats = await lstatIfExists(legacyArchiveDir, fileSystem); + if (!legacyArchiveDirExists) return archiveStats === undefined && files.length === 0; + if (archiveStats === undefined || !archiveStats.isDirectory() || archiveStats.isSymbolicLink()) return false; + const entries = await fileSystem.readdir(legacyArchiveDir); + if (entries.some((entry) => !entry.isFile() || entry.isSymbolicLink())) return false; + const actualNames = new Set(entries.map((entry) => pathNameKey(entry.name, platform))); + const expectedNames = new Set(files.map((file) => pathNameKey(file.fileName, platform))); + if (actualNames.size !== entries.length || expectedNames.size !== files.length || actualNames.size !== expectedNames.size) return false; + return [...expectedNames].every((name) => actualNames.has(name)); +} + +async function assertDestinationStateUnchanged( + plan: MigrationPlan, + fileSystem: SessionArchiveMigrationFileSystem, +): Promise { + if (await lstatIfExists(plan.destinationIndexPath, fileSystem) !== undefined) { + throw new Error("Destination archive index appeared after migration preflight"); + } + const archiveStats = await lstatIfExists(plan.destinationArchiveDir, fileSystem); + if (!plan.destinationArchiveDirExists) { + if (archiveStats !== undefined) throw new Error("Destination archive directory appeared after migration preflight"); + return; + } + if (archiveStats === undefined || !archiveStats.isDirectory() || archiveStats.isSymbolicLink()) { + throw new Error("Destination archive directory changed after migration preflight"); + } + if ((await fileSystem.readdir(plan.destinationArchiveDir)).length !== 0) { + throw new Error("Destination archive directory is no longer empty"); + } +} + +async function assertPlanReadyToCommit( + plan: MigrationPlan, + stagedIndexPath: string, + fileSystem: SessionArchiveMigrationFileSystem, +): Promise { + const currentLegacyRoot = await canonicalizeAllowMissing(plan.legacyRoot, fileSystem); + const currentDestinationRoot = await canonicalizeAllowMissing(plan.destinationRoot, fileSystem); + if (!pathsEqual(currentLegacyRoot, plan.canonicalLegacyRoot, plan.platform) + || !pathsEqual(currentDestinationRoot, plan.canonicalDestinationRoot, plan.platform)) { + throw new Error("Archive data root changed during migration"); + } + const legacyIndexStats = await lstatIfExists(plan.legacyIndexPath, fileSystem); + if (legacyIndexStats === undefined || !legacyIndexStats.isFile() || legacyIndexStats.isSymbolicLink()) { + throw new Error("Legacy archive index changed during migration"); + } + if (await fileSystem.readFile(plan.legacyIndexPath) !== plan.sourceIndexContents) { + throw new Error("Legacy archive index changed during migration"); + } + if (!await legacyDirectoryMatchesPlan(plan.legacyArchiveDir, plan.legacyArchiveDirExists, plan.files, plan.platform, fileSystem)) { + throw new Error("Legacy archive directory changed during migration"); + } + if (plan.legacyArchiveDirExists + && !pathsEqual(await fileSystem.realpath(plan.legacyArchiveDir), plan.canonicalLegacyArchiveDir, plan.platform)) { + throw new Error("Legacy archive directory changed during migration"); + } + if (await lstatIfExists(plan.destinationIndexPath, fileSystem) !== undefined) { + throw new Error("Destination archive index appeared during migration"); + } + if (await fileSystem.readFile(stagedIndexPath) !== plan.destinationIndexContents) { + throw new Error("Staged archive index changed during migration"); + } + + if (plan.files.length === 0) { + const destinationStats = await lstatIfExists(plan.destinationArchiveDir, fileSystem); + if (plan.destinationArchiveDirExists) { + if (destinationStats === undefined || !destinationStats.isDirectory() || destinationStats.isSymbolicLink() + || !pathsEqual(await fileSystem.realpath(plan.destinationArchiveDir), plan.canonicalDestinationArchiveDir, plan.platform) + || (await fileSystem.readdir(plan.destinationArchiveDir)).length !== 0) { + throw new Error("Destination archive directory changed during migration"); + } + } else if (destinationStats !== undefined) { + throw new Error("Destination archive directory appeared during migration"); + } + return; + } + + const destinationStats = await lstatIfExists(plan.destinationArchiveDir, fileSystem); + if (destinationStats === undefined || !destinationStats.isDirectory() || destinationStats.isSymbolicLink()) { + throw new Error("Destination archive directory is not a real directory"); + } + if (!pathsEqual(await fileSystem.realpath(plan.destinationArchiveDir), plan.canonicalDestinationArchiveDir, plan.platform)) { + throw new Error("Destination archive directory changed during migration"); + } + const destinationEntries = await fileSystem.readdir(plan.destinationArchiveDir); + if (destinationEntries.some((entry) => !entry.isFile() || entry.isSymbolicLink())) { + throw new Error("Destination archive directory contains an unexpected entry"); + } + const actualNames = new Set(destinationEntries.map((entry) => pathNameKey(entry.name, plan.platform))); + const expectedNames = new Set(plan.files.map((file) => pathNameKey(file.fileName, plan.platform))); + if (actualNames.size !== destinationEntries.length || actualNames.size !== expectedNames.size + || ![...expectedNames].every((name) => actualNames.has(name))) { + throw new Error("Destination archive directory does not match the migration plan"); + } + + for (const file of plan.files) { + const sourceStats = await lstatIfExists(file.sourcePath, fileSystem); + const destinationFileStats = await lstatIfExists(file.destinationPath, fileSystem); + if (sourceStats === undefined || !sourceStats.isFile() || sourceStats.isSymbolicLink() + || destinationFileStats === undefined || !destinationFileStats.isFile() || destinationFileStats.isSymbolicLink()) { + throw new Error(`Archive file changed during migration: ${file.fileName}`); + } + if (!await filesEqual(file.sourcePath, file.destinationPath, fileSystem)) { + throw new Error(`Archive file verification failed before commit: ${file.fileName}`); + } + } +} + +async function rollbackUncommittedDestination( + state: { + stagingRoot: string | undefined; + stagingCreated: boolean; + destinationArchiveDir: string; + destinationArchiveCreated: boolean; + ownedDestinationFiles: string[]; + }, + fileSystem: SessionArchiveMigrationFileSystem, +): Promise { + const errors: unknown[] = []; + for (const path of [...state.ownedDestinationFiles].reverse()) { + try { + await removeFileIfPresent(path, fileSystem); + } catch (error: unknown) { + errors.push(error); + } + } + if (state.destinationArchiveCreated) { + try { + await removeDirectoryIfPresent(state.destinationArchiveDir, fileSystem); + } catch (error: unknown) { + errors.push(error); + } + } + if (state.stagingCreated && state.stagingRoot !== undefined) { + try { + await fileSystem.rmOwnedTree(state.stagingRoot); + } catch (error: unknown) { + errors.push(error); + } + } + return errors; +} + +async function removeCommittedLegacyState( + plan: MigrationPlan, + fileSystem: SessionArchiveMigrationFileSystem, +): Promise { + const errors: unknown[] = []; + // Remove the index last. If any file/directory cleanup fails, the complete + // legacy index remains as a recovery marker while the destination is valid. + for (const file of plan.files) { + try { + await removeFileIfPresent(file.sourcePath, fileSystem); + } catch (error: unknown) { + errors.push(error); + return errors; + } + } + if (plan.legacyArchiveDirExists) { + try { + await removeDirectoryIfPresent(plan.legacyArchiveDir, fileSystem); + } catch (error: unknown) { + errors.push(error); + return errors; + } + } + try { + await removeFileIfPresent(plan.legacyIndexPath, fileSystem); + } catch (error: unknown) { + errors.push(error); + } + return errors; +} + +async function filesEqual( + firstPath: string, + secondPath: string, + fileSystem: SessionArchiveMigrationFileSystem, +): Promise { + const first = await fileSystem.open(firstPath); + try { + const second = await fileSystem.open(secondPath); + try { + const firstBuffer = Buffer.allocUnsafe(64 * 1024); + const secondBuffer = Buffer.allocUnsafe(64 * 1024); + for (;;) { + const [firstBytes, secondBytes] = await Promise.all([ + readChunk(first, firstBuffer), + readChunk(second, secondBuffer), + ]); + if (firstBytes !== secondBytes) return false; + if (firstBytes === 0) return true; + if (!firstBuffer.subarray(0, firstBytes).equals(secondBuffer.subarray(0, secondBytes))) return false; + } + } finally { + await second.close(); + } + } finally { + await first.close(); + } +} + +async function readChunk(handle: SessionArchiveMigrationReadHandle, buffer: Buffer): Promise { + let total = 0; + while (total < buffer.length) { + const { bytesRead } = await handle.read(buffer, total, buffer.length - total, null); + if (bytesRead === 0) break; + total += bytesRead; + } + return total; +} + +async function canonicalizeAllowMissing(path: string, fileSystem: SessionArchiveMigrationFileSystem): Promise { + let cursor = resolve(path); + const missingParts: string[] = []; + for (;;) { + try { + const canonical = await fileSystem.realpath(cursor); + return resolve(canonical, ...missingParts); + } catch (error: unknown) { + if (!isNodeErrorWithCode(error, "ENOENT")) throw error; + if (await lstatIfExists(cursor, fileSystem) !== undefined) { + throw new Error(`Cannot resolve existing path: ${cursor}`, { cause: error }); + } + const parent = dirname(cursor); + if (parent === cursor) throw error; + missingParts.unshift(basename(cursor)); + cursor = parent; + } + } +} + +async function lstatIfExists(path: string, fileSystem: SessionArchiveMigrationFileSystem): Promise { + try { + return await fileSystem.lstat(path); + } catch (error: unknown) { + if (isNodeErrorWithCode(error, "ENOENT")) return undefined; + throw error; + } +} + +async function removeFileIfPresent(path: string, fileSystem: SessionArchiveMigrationFileSystem): Promise { + try { + await fileSystem.unlink(path); + } catch (error: unknown) { + if (!isNodeErrorWithCode(error, "ENOENT")) throw error; + } +} + +async function removeDirectoryIfPresent(path: string, fileSystem: SessionArchiveMigrationFileSystem): Promise { + try { + await fileSystem.rmdir(path); + } catch (error: unknown) { + if (!isNodeErrorWithCode(error, "ENOENT")) throw error; + } +} + +function migrationFileSystem(options: LegacySessionArchiveMigrationOptions): SessionArchiveMigrationFileSystem { + return { ...defaultFileSystem, ...options.fileSystem }; +} + +function migrationSkipped( + reason: LegacySessionArchiveMigrationSkipReason, + error?: unknown, +): LegacySessionArchiveMigrationSkipped { + return error === undefined ? { status: "skipped", reason } : { status: "skipped", reason, error }; +} + +function pathsOverlap(first: string, second: string, platform: NodeJS.Platform): boolean { + return pathContains(first, second, platform) || pathContains(second, first, platform); +} + +function pathContains(parent: string, candidate: string, platform: NodeJS.Platform): boolean { + const parentPath = pathKey(parent, platform); + const candidatePath = pathKey(candidate, platform); + const pathFromParent = relative(parentPath, candidatePath); + const firstSegment = pathFromParent.split(/[\\/]/, 1)[0]; + return pathFromParent === "" || (firstSegment !== ".." && !isAbsolute(pathFromParent)); +} + +function pathsEqual(first: string, second: string, platform: NodeJS.Platform): boolean { + return pathKey(first, platform) === pathKey(second, platform); +} + +function pathKey(path: string, platform: NodeJS.Platform): string { + const normalized = resolve(path); + return platform === "win32" ? normalized.toLowerCase() : normalized; +} + +function pathNameKey(name: string, platform: NodeJS.Platform): string { + return platform === "win32" ? name.toLowerCase() : name; +} + +function collisionKey(path: string): string { + // Conservatively reject case/normalization variants even when the source + // platform happens to allow them; the destination filesystem may not. + return resolve(path).normalize("NFC").toLowerCase(); +} + +function safeAttemptId(value: string): string { + return value.replace(/[^a-zA-Z0-9._-]/g, "_") || "attempt"; +} + +function recordArray(value: unknown): Record[] | undefined { + if (!Array.isArray(value)) return undefined; + const records: Record[] = []; + for (const item of value) { + if (!isRecord(item)) return undefined; + records.push(item); + } + return records; +} + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + +function isNodeErrorWithCode(error: unknown, code: string): error is NodeJS.ErrnoException { + return error instanceof Error && "code" in error && error.code === code; +} diff --git a/src/server/sessions/sessionArchiveStore.test.ts b/src/server/sessions/sessionArchiveStore.test.ts index c11f579b..83bdc8c0 100644 --- a/src/server/sessions/sessionArchiveStore.test.ts +++ b/src/server/sessions/sessionArchiveStore.test.ts @@ -1,17 +1,55 @@ import { constants } from "node:fs"; import { access, mkdtemp, mkdir, readFile, rm, writeFile } from "node:fs/promises"; import { join, resolve } from "node:path"; -import { tmpdir } from "node:os"; -import { afterEach, describe, expect, it } from "vitest"; -import { SessionArchiveStore } from "./sessionArchiveStore.js"; +import { homedir, tmpdir } from "node:os"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { defaultSessionArchiveFilePath, SessionArchiveStore } from "./sessionArchiveStore.js"; const tempRoots: string[] = []; +describe("defaultSessionArchiveFilePath", () => { + it("uses PI_WEB_DATA_DIR when configured", () => { + expect(defaultSessionArchiveFilePath({ PI_WEB_DATA_DIR: "managed-state" }, "/tmp/pi-web")).toBe(resolve("/tmp/pi-web", "managed-state", "archived-sessions.json")); + }); + + it("preserves the ~/.pi-web default when PI_WEB_DATA_DIR is unset", () => { + expect(defaultSessionArchiveFilePath({}, "/tmp/pi-web")).toBe(join(homedir(), ".pi-web", "archived-sessions.json")); + }); +}); + describe("SessionArchiveStore", () => { afterEach(async () => { + vi.unstubAllEnvs(); await Promise.all(tempRoots.splice(0).map((path) => rm(path, { recursive: true, force: true }))); }); + it("stores its default index and archived files under PI_WEB_DATA_DIR", async () => { + const root = await mkdtemp(join(tmpdir(), "pi-web-archive-data-dir-")); + tempRoots.push(root); + const dataDir = join(root, "managed-state"); + const activeDir = join(root, "active"); + await mkdir(activeDir, { recursive: true }); + const sourcePath = join(activeDir, "2026-01-01_managed.jsonl"); + await writeFile(sourcePath, "session contents\n", "utf8"); + vi.stubEnv("PI_WEB_DATA_DIR", dataDir); + + const store = new SessionArchiveStore(); + const record = await store.archive({ + sessionId: "managed", + cwd: "/workspace", + path: sourcePath, + created: "2026-01-01T00:00:00.000Z", + modified: "2026-01-01T00:01:00.000Z", + messageCount: 1, + firstMessage: "hello", + }); + + const archivePath = join(dataDir, "archived-sessions", "2026-01-01_managed.jsonl"); + expect(record.archivePath).toBe(archivePath); + await expect(readFile(archivePath, "utf8")).resolves.toBe("session contents\n"); + await expect(readFile(join(dataDir, "archived-sessions.json"), "utf8")).resolves.toContain('"sessionId": "managed"'); + }); + it("moves archived session files out of the active session directory and restores them", async () => { const root = await mkdtemp(join(tmpdir(), "pi-web-archive-")); tempRoots.push(root); diff --git a/src/server/sessions/sessionArchiveStore.ts b/src/server/sessions/sessionArchiveStore.ts index cefbb0fd..dd80a4e1 100644 --- a/src/server/sessions/sessionArchiveStore.ts +++ b/src/server/sessions/sessionArchiveStore.ts @@ -2,7 +2,7 @@ import { randomUUID } from "node:crypto"; import { constants } from "node:fs"; import { access, copyFile, mkdir, readFile, rename, unlink, writeFile } from "node:fs/promises"; import { basename, dirname, join } from "node:path"; -import { homedir } from "node:os"; +import { piWebDataDir } from "../../config.js"; import { canonicalizeStoredCwd } from "../workingDirectory.js"; export interface ArchiveSessionInput { @@ -31,15 +31,19 @@ export interface ArchivedSessionRecord { parentSessionPath?: string; } -interface ArchiveFile { +export interface SessionArchiveFile { sessions: ArchivedSessionRecord[]; } +export function defaultSessionArchiveFilePath(env: NodeJS.ProcessEnv = process.env, cwd = process.cwd()): string { + return join(piWebDataDir(env, cwd), "archived-sessions.json"); +} + export class SessionArchiveStore { private operationQueue: Promise = Promise.resolve(); constructor( - private readonly filePath = join(homedir(), ".pi-web", "archived-sessions.json"), + private readonly filePath = defaultSessionArchiveFilePath(), private readonly archiveDir = join(dirname(filePath), "archived-sessions"), ) {} @@ -149,17 +153,17 @@ export class SessionArchiveStore { } } - private async read(): Promise { + private async read(): Promise { try { const value: unknown = JSON.parse(await readFile(this.filePath, "utf8")); - return parseArchiveFile(value); + return parseSessionArchiveFile(value); } catch (error: unknown) { if (isNodeErrorWithCode(error, "ENOENT")) return { sessions: [] }; throw error; } } - private async write(data: ArchiveFile): Promise { + private async write(data: SessionArchiveFile): Promise { await mkdir(dirname(this.filePath), { recursive: true }); const tempPath = join(dirname(this.filePath), `.${basename(this.filePath)}.${String(process.pid)}.${Date.now().toString()}.${randomUUID()}.tmp`); try { @@ -227,7 +231,7 @@ async function pathExists(path: string): Promise { } } -function parseArchiveFile(value: unknown): ArchiveFile { +export function parseSessionArchiveFile(value: unknown): SessionArchiveFile { if (!isRecord(value) || !Array.isArray(value["sessions"])) throw new Error("Invalid archive file"); return { sessions: value["sessions"].map(parseArchivedSessionRecord) }; }