diff --git a/src/server/sessiond.ts b/src/server/sessiond.ts index f4cc36e..bcdd023 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), + 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("/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, + }, + }; + }); + + 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); }); + + 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 })); + } + }, }); -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, - }, - }; -}); - -app.get("/runtime", () => getPiWebRuntimeComponent("sessiond", SESSIOND_RUNTIME_CAPABILITIES)); - -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); }); - -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 })); -} diff --git a/src/server/sessiond/sessionDaemonStartup.test.ts b/src/server/sessiond/sessionDaemonStartup.test.ts new file mode 100644 index 0000000..6296131 --- /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 0000000..b0c645d --- /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", + ); +}