fix(sessiond): migrate legacy archives before startup

src/server/sessiond.ts

@@ -20,32 +20,39 @@ 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();
+await runSessionDaemonStartup({
-const workspaceActivity = new WorkspaceActivityService(eventHub);
+  logger: app.log,
-const auth = new AuthService();
+  createRuntime() {
-const spawnTargets = spawnSessionsEnabled(process.env, config)
+    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, {
+    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); });
+    auth.subscribe((change) => { sessions.applyAuthChange(change); });
-const terminals = new TerminalService(eventHub, workspaceActivity);
+    const terminals = new TerminalService(eventHub, workspaceActivity);
-registerWorkspaceActivityRoutes(app, workspaceActivity);
+    return { eventHub, workspaceActivity, auth, sessions, terminals };
-registerAuthRoutes(app, auth);
+  },
-registerSessionRoutes(app, sessions, eventHub);
+  registerRoutes({ eventHub, workspaceActivity, auth, sessions, terminals }) {
-registerTerminalRoutes(app, terminals);
+    registerWorkspaceActivityRoutes(app, workspaceActivity);
+    registerAuthRoutes(app, auth);
+    registerSessionRoutes(app, sessions, eventHub);
+    registerTerminalRoutes(app, terminals);
 
-app.get("/health", () => {
+    app.get("/health", () => {
       const runtime = getPiWebRuntimeComponent("sessiond", SESSIOND_RUNTIME_CAPABILITIES);
       return {
         ok: true,
@@ -59,12 +66,13 @@ app.get("/health", () => {
           available: runtime.available,
         },
       };
-});
+    });
 
-app.get("/runtime", () => getPiWebRuntimeComponent("sessiond", SESSIOND_RUNTIME_CAPABILITIES));
+    app.get("/runtime", () => getPiWebRuntimeComponent("sessiond", SESSIOND_RUNTIME_CAPABILITIES));
-
+  },
-let shuttingDown = false;
+  async listen({ auth, sessions, terminals }) {
-async function shutdown(signal: NodeJS.Signals): Promise<void> {
+    let shuttingDown = false;
+    async function shutdown(signal: NodeJS.Signals): Promise<void> {
       if (shuttingDown) return;
       shuttingDown = true;
       app.log.info({ signal }, "shutting down session daemon");
@@ -72,21 +80,23 @@ async function shutdown(signal: NodeJS.Signals): Promise<void> {
       auth.dispose();
       await sessions.dispose();
       await app.close();
-}
+    }
 
-process.once("SIGINT", (signal) => { void shutdown(signal); });
+    process.once("SIGINT", (signal) => { void shutdown(signal); });
-process.once("SIGTERM", (signal) => { void shutdown(signal); });
+    process.once("SIGTERM", (signal) => { void shutdown(signal); });
 
-const portValue = process.env["PI_WEB_SESSIOND_PORT"];
+    const portValue = process.env["PI_WEB_SESSIOND_PORT"];
-const port = portValue !== undefined && portValue !== "" ? Number(portValue) : undefined;
+    const port = portValue !== undefined && portValue !== "" ? Number(portValue) : undefined;
-const host = process.env["PI_WEB_SESSIOND_HOST"] ?? "127.0.0.1";
+    const host = process.env["PI_WEB_SESSIOND_HOST"] ?? "127.0.0.1";
 
-if (port !== undefined) {
+    if (port !== undefined) {
       await app.listen({ port, host });
-} else {
+    } 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 }));
-}
+    }
+  },
+});

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<LegacySessionArchiveMigrationResult>();
+    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<LegacySessionArchiveMigrationResult>({
+        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<LegacySessionArchiveMigrationResult>({
+        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<LegacySessionArchiveMigrationResult>({
+        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<LegacyArchiveFixture> {
+  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<T>() {
+  let resolve!: (value: T | PromiseLike<T>) => void;
+  const promise = new Promise<T>((resolvePromise) => {
+    resolve = resolvePromise;
+  });
+  return { promise, resolve };
+}

src/server/sessiond/sessionDaemonStartup.ts

@@ -0,0 +1,95 @@
+import {
+  migrateLegacySessionArchive,
+  type LegacySessionArchiveMigrationResult,
+} from "../sessions/sessionArchiveMigration.js";
+
+export interface SessionDaemonStartupLogger {
+  debug(details: Record<string, unknown>, message: string): void;
+  info(details: Record<string, unknown>, message: string): void;
+  warn(details: Record<string, unknown>, message: string): void;
+  error(details: Record<string, unknown>, message: string): void;
+}
+
+export interface SessionDaemonStartupSteps<Runtime> {
+  logger: SessionDaemonStartupLogger;
+  createRuntime(): Runtime;
+  registerRoutes(runtime: Runtime): void;
+  listen(runtime: Runtime): Promise<void>;
+  migrateArchive?: () => Promise<LegacySessionArchiveMigrationResult>;
+}
+
+/**
+ * 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<Runtime>(
+  steps: SessionDaemonStartupSteps<Runtime>,
+): Promise<Runtime> {
+  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",
+  );
+}