From ec0ca131200724967ed1b09b4286822c304dc834 Mon Sep 17 00:00:00 2001 From: Federico Jaramillo Martinez Date: Sun, 12 Jul 2026 22:52:59 +0200 Subject: [PATCH 1/7] fix: store session archives under PI_WEB_DATA_DIR --- .changeset/honor-archive-data-dir.md | 5 +++ docs/config.md | 13 +++++- .../sessions/sessionArchiveStore.test.ts | 44 +++++++++++++++++-- src/server/sessions/sessionArchiveStore.ts | 8 +++- 4 files changed, 64 insertions(+), 6 deletions(-) create mode 100644 .changeset/honor-archive-data-dir.md diff --git a/.changeset/honor-archive-data-dir.md b/.changeset/honor-archive-data-dir.md new file mode 100644 index 00000000..705e87b0 --- /dev/null +++ b/.changeset/honor-archive-data-dir.md @@ -0,0 +1,5 @@ +--- +"@jmfederico/pi-web": patch +--- + +Store session archive metadata and archived session files under `PI_WEB_DATA_DIR` when configured. diff --git a/docs/config.md b/docs/config.md index 1d8dfee3..b06bb9b0 100644 --- a/docs/config.md +++ b/docs/config.md @@ -109,7 +109,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 before changing; relocates managed state, including session archives | | 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 | @@ -122,6 +122,17 @@ Rows with JSON key `—` are runtime-only environment variables, not config-file ## Key details +### Managed data directory + +`PI_WEB_DATA_DIR` selects the machine-global root for PI WEB-managed state. It defaults to `~/.pi-web`. Session archival uses this layout: + +- `$PI_WEB_DATA_DIR/archived-sessions.json` for the archive index. +- `$PI_WEB_DATA_DIR/archived-sessions/` for archived session files. + +The session daemon owns both archive locations. Restart the session daemon after setting or changing `PI_WEB_DATA_DIR`; restart the web/API process as well so its other managed-state stores use the same root. + +PI WEB does not automatically migrate existing managed state. Archives created by earlier releases while a custom `PI_WEB_DATA_DIR` was already set may remain under `~/.pi-web` and will not appear automatically after upgrading. Stop the session daemon and back up both locations before reconciling them manually. The archive index stores absolute `archivePath` values, so update those values if archived files are moved to a different path. + ### 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/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..53085cdd 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 { @@ -35,11 +35,15 @@ interface ArchiveFile { 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"), ) {} From 619e2938578873491498253ec1c27b2a0c7cf05e Mon Sep 17 00:00:00 2001 From: Federico Jaramillo Martinez Date: Sun, 12 Jul 2026 23:10:10 +0200 Subject: [PATCH 2/7] docs: move archive upgrade note to changeset --- .changeset/honor-archive-data-dir.md | 2 ++ docs/config.md | 2 +- 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/.changeset/honor-archive-data-dir.md b/.changeset/honor-archive-data-dir.md index 705e87b0..df074400 100644 --- a/.changeset/honor-archive-data-dir.md +++ b/.changeset/honor-archive-data-dir.md @@ -3,3 +3,5 @@ --- Store session archive metadata and archived session files under `PI_WEB_DATA_DIR` when configured. + +Previously, session archives always used `~/.pi-web`, even when `PI_WEB_DATA_DIR` selected another managed-state root. Existing archives created with a custom `PI_WEB_DATA_DIR` remain in `~/.pi-web` and are not migrated automatically. Before upgrading, stop the session daemon and back up both locations before reconciling them manually. Because the archive index stores absolute `archivePath` values, update those values when moving archived files. diff --git a/docs/config.md b/docs/config.md index b06bb9b0..37e4840b 100644 --- a/docs/config.md +++ b/docs/config.md @@ -131,7 +131,7 @@ Rows with JSON key `—` are runtime-only environment variables, not config-file The session daemon owns both archive locations. Restart the session daemon after setting or changing `PI_WEB_DATA_DIR`; restart the web/API process as well so its other managed-state stores use the same root. -PI WEB does not automatically migrate existing managed state. Archives created by earlier releases while a custom `PI_WEB_DATA_DIR` was already set may remain under `~/.pi-web` and will not appear automatically after upgrading. Stop the session daemon and back up both locations before reconciling them manually. The archive index stores absolute `archivePath` values, so update those values if archived files are moved to a different path. +PI WEB does not move existing managed state when `PI_WEB_DATA_DIR` changes. Stop the session daemon and back up the source and destination before moving archive data manually. The archive index stores absolute `archivePath` values, so update those values when archived files move to a different path. ### External path access From a0ea2bbb45a917a37819011779ac8c5754c5ea75 Mon Sep 17 00:00:00 2001 From: Federico Jaramillo Martinez Date: Sun, 12 Jul 2026 23:20:02 +0200 Subject: [PATCH 3/7] docs: keep managed data guidance general --- docs/config.md | 13 +------------ 1 file changed, 1 insertion(+), 12 deletions(-) diff --git a/docs/config.md b/docs/config.md index 37e4840b..e171afe2 100644 --- a/docs/config.md +++ b/docs/config.md @@ -109,7 +109,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 web/API and session daemon before changing; relocates managed state, including session archives | +| 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 | @@ -122,17 +122,6 @@ Rows with JSON key `—` are runtime-only environment variables, not config-file ## Key details -### Managed data directory - -`PI_WEB_DATA_DIR` selects the machine-global root for PI WEB-managed state. It defaults to `~/.pi-web`. Session archival uses this layout: - -- `$PI_WEB_DATA_DIR/archived-sessions.json` for the archive index. -- `$PI_WEB_DATA_DIR/archived-sessions/` for archived session files. - -The session daemon owns both archive locations. Restart the session daemon after setting or changing `PI_WEB_DATA_DIR`; restart the web/API process as well so its other managed-state stores use the same root. - -PI WEB does not move existing managed state when `PI_WEB_DATA_DIR` changes. Stop the session daemon and back up the source and destination before moving archive data manually. The archive index stores absolute `archivePath` values, so update those values when archived files move to a different path. - ### External path access `pathAccess.allowedPaths` grants PI WEB's file explorer and absolute `@` path completions access to specific filesystem roots outside the current workspace. From 7878cbe57b4814c1f988d379b6a2d9a22c950b70 Mon Sep 17 00:00:00 2001 From: Federico Jaramillo Martinez Date: Mon, 13 Jul 2026 14:26:32 +0200 Subject: [PATCH 4/7] fix(sessions): add safe legacy archive migration core --- .../sessions/sessionArchiveMigration.test.ts | 430 ++++++++++ .../sessions/sessionArchiveMigration.ts | 781 ++++++++++++++++++ src/server/sessions/sessionArchiveStore.ts | 10 +- 3 files changed, 1216 insertions(+), 5 deletions(-) create mode 100644 src/server/sessions/sessionArchiveMigration.test.ts create mode 100644 src/server/sessions/sessionArchiveMigration.ts diff --git a/src/server/sessions/sessionArchiveMigration.test.ts b/src/server/sessions/sessionArchiveMigration.test.ts new file mode 100644 index 00000000..6f1b6cf4 --- /dev/null +++ b/src/server/sessions/sessionArchiveMigration.test.ts @@ -0,0 +1,430 @@ +import { constants } from "node:fs"; +import { + access, + appendFile, + copyFile, + lstat, + mkdtemp, + mkdir, + readFile, + readdir, + rm, + unlink, + writeFile, +} from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join, 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("skips a non-empty destination archive directory without changing either side", async () => { + const fixture = await createLegacyArchiveFixture({ createDestinationArchive: true }); + const destinationEntry = join(fixture.destinationArchiveDir, "already-here.jsonl"); + await writeFile(destinationEntry, "destination owner\n", "utf8"); + + await expect(migrateLegacySessionArchive(fixture.options)).resolves.toEqual({ + status: "skipped", + reason: "destination-archive-not-empty-or-invalid", + }); + await expect(readFile(destinationEntry, "utf8")).resolves.toBe("destination owner\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("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("stages and verifies copies, atomically publishes the rewritten index, then removes legacy state", async () => { + const fixture = await createLegacyArchiveFixture({ createDestinationArchive: true }); + const copies: { source: string; destination: string; mode: number }[] = []; + + await expect(migrateLegacySessionArchive({ + ...fixture.options, + fileSystem: { + copyFile: async (source, destination, mode) => { + copies.push({ source, destination, mode }); + await copyFile(source, destination, mode); + }, + }, + })).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 }); + 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("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("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("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.ts b/src/server/sessions/sessionArchiveStore.ts index 53085cdd..dd80a4e1 100644 --- a/src/server/sessions/sessionArchiveStore.ts +++ b/src/server/sessions/sessionArchiveStore.ts @@ -31,7 +31,7 @@ export interface ArchivedSessionRecord { parentSessionPath?: string; } -interface ArchiveFile { +export interface SessionArchiveFile { sessions: ArchivedSessionRecord[]; } @@ -153,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 { @@ -231,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) }; } From 31c98454eca98276c3895e240cc0ec551783d01b Mon Sep 17 00:00:00 2001 From: Federico Jaramillo Martinez Date: Mon, 13 Jul 2026 14:37:05 +0200 Subject: [PATCH 5/7] fix(sessiond): migrate legacy archives before startup --- src/server/sessiond.ts | 128 +++++---- .../sessiond/sessionDaemonStartup.test.ts | 261 ++++++++++++++++++ src/server/sessiond/sessionDaemonStartup.ts | 95 +++++++ 3 files changed, 425 insertions(+), 59 deletions(-) create mode 100644 src/server/sessiond/sessionDaemonStartup.test.ts create mode 100644 src/server/sessiond/sessionDaemonStartup.ts 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", + ); +} From 050b053d237e8a5b0edfa27e651c7effc12412aa Mon Sep 17 00:00:00 2001 From: Federico Jaramillo Martinez Date: Mon, 13 Jul 2026 14:45:14 +0200 Subject: [PATCH 6/7] test(sessiond): harden archive migration safety --- .changeset/honor-archive-data-dir.md | 4 +- docs/config.md | 20 +++ .../sessions/sessionArchiveMigration.test.ts | 137 +++++++++++++++++- 3 files changed, 153 insertions(+), 8 deletions(-) diff --git a/.changeset/honor-archive-data-dir.md b/.changeset/honor-archive-data-dir.md index df074400..f50f6c5b 100644 --- a/.changeset/honor-archive-data-dir.md +++ b/.changeset/honor-archive-data-dir.md @@ -2,6 +2,6 @@ "@jmfederico/pi-web": patch --- -Store session archive metadata and archived session files under `PI_WEB_DATA_DIR` when configured. +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. -Previously, session archives always used `~/.pi-web`, even when `PI_WEB_DATA_DIR` selected another managed-state root. Existing archives created with a custom `PI_WEB_DATA_DIR` remain in `~/.pi-web` and are not migrated automatically. Before upgrading, stop the session daemon and back up both locations before reconciling them manually. Because the archive index stores absolute `archivePath` values, update those values when moving archived files. +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 e171afe2..2f0e398a 100644 --- a/docs/config.md +++ b/docs/config.md @@ -122,6 +122,26 @@ Rows with JSON key `—` are runtime-only environment variables, not config-file ## Key details +### Managed data directory and session archive migration + +`PI_WEB_DATA_DIR` selects the root for PI WEB-managed state, including the session archive index and archived session files. It does not move active Pi session files from `PI_CODING_AGENT_SESSION_DIR`. + +After an upgrade from a version that stored session archives only in `~/.pi-web`, the session daemon performs a one-time migration on the first eligible startup, before it creates the session runtime, registers routes, or starts listening. Automatic migration is deliberately strict. It runs only when all of these conditions are proven: + +1. `PI_WEB_DATA_DIR` is explicitly set to a non-empty value and resolves to a root other than `~/.pi-web`. +2. `~/.pi-web/archived-sessions.json` is a regular file that parses with the current archive schema. +3. `$PI_WEB_DATA_DIR/archived-sessions.json` does not exist. +4. `$PI_WEB_DATA_DIR/archived-sessions/` is absent or is a real, empty directory. +5. Every defined `archivePath` points directly into the real `~/.pi-web/archived-sessions/` directory, names an existing regular file, and maps uniquely to the destination. Metadata-only records without an `archivePath` are preserved. +6. The legacy archive directory is absent or empty when no files are referenced; otherwise it contains exactly the regular files referenced by the index, with no extra files, symlinks, nested directories, or temporary entries. +7. Session IDs are non-empty and unique, source paths are unique under the host's path semantics, and destination mappings have no case/normalization collisions or other ambiguity. + +An eligible migration copies and byte-verifies the archived files, so it works when the legacy and configured roots are on different filesystems. It rewrites only the migrated `archivePath` values, preserves the remaining archive metadata, verifies the complete destination, and atomically publishes the destination index. Only after that commit does it remove the legacy archived files, archive directory, and index. + +If any preflight condition is false or filesystem inspection is inconclusive, migration makes no archive filesystem changes and the session daemon starts against the configured destination. PI WEB does not infer intent, merge indexes, adopt partial files, or overwrite either side. If you expected a migration but it was skipped, stop the daemon, back up both roots, and inspect them before reconciling the state manually; do not delete either side based only on the skip. + +A copy, verification, or commit failure before the destination index is published keeps the legacy index authoritative, rolls back artifacts owned by that attempt where possible, and stops session-daemon startup. Correct the reported filesystem problem and remove only artifacts you have verified belong to the failed attempt before restarting. If cleanup fails after the destination index is committed, the daemon warns and starts from the verified destination; treat that destination as authoritative and inspect any legacy or staging leftovers manually rather than merging them back. + ### 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/sessions/sessionArchiveMigration.test.ts b/src/server/sessions/sessionArchiveMigration.test.ts index 6f1b6cf4..876534c4 100644 --- a/src/server/sessions/sessionArchiveMigration.test.ts +++ b/src/server/sessions/sessionArchiveMigration.test.ts @@ -3,17 +3,19 @@ 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 { join, sep } from "node:path"; +import { dirname, join, resolve, sep } from "node:path"; import { afterEach, describe, expect, it } from "vitest"; import { inspectLegacySessionArchiveMigration, @@ -122,16 +124,15 @@ describe("legacy session archive migration preflight", () => { await expectLegacyStateUntouched(fixture); }); - it("skips a non-empty destination archive directory without changing either side", async () => { + it("leaves an interrupted pre-commit file untouched instead of adopting a non-empty destination archive", async () => { const fixture = await createLegacyArchiveFixture({ createDestinationArchive: true }); - const destinationEntry = join(fixture.destinationArchiveDir, "already-here.jsonl"); - await writeFile(destinationEntry, "destination owner\n", "utf8"); + 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(destinationEntry, "utf8")).resolves.toBe("destination owner\n"); + await expect(readFile(fixture.destinationFilePath, "utf8")).resolves.toBe("partial destination copy\n"); await expectLegacyStateUntouched(fixture); }); @@ -155,6 +156,42 @@ describe("legacy session archive migration preflight", () => { 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"); @@ -211,9 +248,10 @@ describe("legacy session archive migration execution", () => { await Promise.all(tempRoots.splice(0).map((path) => rm(path, { recursive: true, force: true }))); }); - it("stages and verifies copies, atomically publishes the rewritten index, then removes legacy state", async () => { + 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, @@ -222,6 +260,16 @@ describe("legacy session archive migration execution", () => { 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" }); @@ -229,6 +277,10 @@ describe("legacy session archive migration execution", () => { 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")); @@ -250,6 +302,29 @@ describe("legacy session archive migration execution", () => { ])); }); + 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(); @@ -269,6 +344,31 @@ describe("legacy session archive migration execution", () => { 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" }); @@ -292,6 +392,31 @@ describe("legacy session archive migration execution", () => { 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" }); From 1cad424b07afb5e14da2abeb505f0870ac51ccd6 Mon Sep 17 00:00:00 2001 From: Federico Jaramillo Martinez Date: Mon, 13 Jul 2026 19:55:09 +0200 Subject: [PATCH 7/7] docs: keep managed data reference timeless --- docs/config.md | 20 +++----------------- 1 file changed, 3 insertions(+), 17 deletions(-) diff --git a/docs/config.md b/docs/config.md index 2f0e398a..fd4f3438 100644 --- a/docs/config.md +++ b/docs/config.md @@ -122,25 +122,11 @@ Rows with JSON key `—` are runtime-only environment variables, not config-file ## Key details -### Managed data directory and session archive migration +### Managed data directory -`PI_WEB_DATA_DIR` selects the root for PI WEB-managed state, including the session archive index and archived session files. It does not move active Pi session files from `PI_CODING_AGENT_SESSION_DIR`. +`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. -After an upgrade from a version that stored session archives only in `~/.pi-web`, the session daemon performs a one-time migration on the first eligible startup, before it creates the session runtime, registers routes, or starts listening. Automatic migration is deliberately strict. It runs only when all of these conditions are proven: - -1. `PI_WEB_DATA_DIR` is explicitly set to a non-empty value and resolves to a root other than `~/.pi-web`. -2. `~/.pi-web/archived-sessions.json` is a regular file that parses with the current archive schema. -3. `$PI_WEB_DATA_DIR/archived-sessions.json` does not exist. -4. `$PI_WEB_DATA_DIR/archived-sessions/` is absent or is a real, empty directory. -5. Every defined `archivePath` points directly into the real `~/.pi-web/archived-sessions/` directory, names an existing regular file, and maps uniquely to the destination. Metadata-only records without an `archivePath` are preserved. -6. The legacy archive directory is absent or empty when no files are referenced; otherwise it contains exactly the regular files referenced by the index, with no extra files, symlinks, nested directories, or temporary entries. -7. Session IDs are non-empty and unique, source paths are unique under the host's path semantics, and destination mappings have no case/normalization collisions or other ambiguity. - -An eligible migration copies and byte-verifies the archived files, so it works when the legacy and configured roots are on different filesystems. It rewrites only the migrated `archivePath` values, preserves the remaining archive metadata, verifies the complete destination, and atomically publishes the destination index. Only after that commit does it remove the legacy archived files, archive directory, and index. - -If any preflight condition is false or filesystem inspection is inconclusive, migration makes no archive filesystem changes and the session daemon starts against the configured destination. PI WEB does not infer intent, merge indexes, adopt partial files, or overwrite either side. If you expected a migration but it was skipped, stop the daemon, back up both roots, and inspect them before reconciling the state manually; do not delete either side based only on the skip. - -A copy, verification, or commit failure before the destination index is published keeps the legacy index authoritative, rolls back artifacts owned by that attempt where possible, and stops session-daemon startup. Correct the reported filesystem problem and remove only artifacts you have verified belong to the failed attempt before restarting. If cleanup fails after the destination index is committed, the daemon warns and starts from the verified destination; treat that destination as authoritative and inspect any legacy or staging leftovers manually rather than merging them back. +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