Archived
fix(sessiond): remove legacy session archive migration
The migration moved ~/.pi-web/archived-sessions* into a custom PI_WEB_DATA_DIR at session daemon startup. Remove it outright: data directories are independent, and anyone who sets a custom data directory can copy archived-sessions.json and archived-sessions/ manually while the daemon is stopped. Without the migration, runSessionDaemonStartup's only remaining contract was sequencing create/register/listen, so inline those steps into sessiond.ts and drop the wrapper module and its tests.
This commit is contained in:
Federico Jaramillo Martinez
+121
-117
@@ -27,7 +27,6 @@ import { getPiWebRuntimeComponent } from "./piWebStatus.js";
|
||||
import { SESSIOND_RUNTIME_CAPABILITIES } from "../shared/capabilities.js";
|
||||
import { agentSessionDirEnvKeys, effectivePiWebConfig, maxUploadBytes, offlineModeEnabled } from "../config.js";
|
||||
import { createActiveAgentProfileDescriptor } from "../sessiond/activeAgentProfile.js";
|
||||
import { runSessionDaemonStartup } from "./sessiond/sessionDaemonStartup.js";
|
||||
import { sessionServiceDependencies } from "./sessiond/sessionServiceDependencies.js";
|
||||
|
||||
const daemonEnvironment: NodeJS.ProcessEnv = Object.freeze({ ...process.env });
|
||||
@@ -40,125 +39,130 @@ const activeAgentProfile = createActiveAgentProfileDescriptor({
|
||||
const app = Fastify({ logger: true, bodyLimit: maxUploadBytes(daemonEnvironment, config) });
|
||||
await app.register(fastifyWebsocket);
|
||||
|
||||
await runSessionDaemonStartup({
|
||||
logger: app.log,
|
||||
async createRuntime() {
|
||||
const eventHub = new SessionEventHub();
|
||||
const notificationStore = new SessionNotificationStore();
|
||||
const unreadStore = new SessionUnreadStore({
|
||||
persistence: new FileSessionUnreadPersistence(),
|
||||
onPersistenceError(operation, error) {
|
||||
app.log.error({ err: error, operation }, "session unread persistence failed");
|
||||
},
|
||||
});
|
||||
await unreadStore.load();
|
||||
const workspaceActivity = new WorkspaceActivityService(eventHub);
|
||||
const auth = await AuthService.create({ agentDir: activeAgentProfile.dir, logger: app.log });
|
||||
// Capture providers registered by global extensions while the runtime is
|
||||
// still mutable, then freeze every later extension-provider mutation before
|
||||
// any real session can load project resources.
|
||||
await bootstrapAndFreezeGlobalExtensionProviders(auth.runtime, activeAgentProfile.dir, app.log);
|
||||
// The shared model runtime is constructed offline so request paths never
|
||||
// wait on provider-catalog fetches; this is the single bounded network
|
||||
// refresher, and auth changes (login/logout) ask it for a prompt run. It
|
||||
// stays fully inert when the operator asked for offline behavior.
|
||||
const catalogRefresher = new ModelCatalogRefresher({
|
||||
runtime: auth.runtime,
|
||||
logger: app.log,
|
||||
offline: offlineModeEnabled(daemonEnvironment),
|
||||
});
|
||||
catalogRefresher.start();
|
||||
auth.subscribe(() => { catalogRefresher.requestRefresh(); });
|
||||
// Cross-workspace session relationships are reported regardless of whether
|
||||
// agents may spawn sessions: children can predate a config change, and the
|
||||
// session tree should stay honest about them either way.
|
||||
const projectWorkspaceDeps = { projects: new ProjectService(new ProjectStore()), workspaces: new WorkspaceService() };
|
||||
const projectWorkspaces = new RegisteredProjectWorkspaceCwds(projectWorkspaceDeps);
|
||||
const spawnTargets = config.spawnSessions ? new ProjectScopedSpawnTargetResolver(projectWorkspaceDeps) : undefined;
|
||||
const sessions = new PiSessionService(eventHub, sessionServiceDependencies({
|
||||
modelRuntime: auth.runtime,
|
||||
const runtime = await createSessionDaemonRuntime();
|
||||
registerSessionDaemonRoutes(runtime);
|
||||
await listenSessionDaemon(runtime);
|
||||
|
||||
type SessionDaemonRuntime = Awaited<ReturnType<typeof createSessionDaemonRuntime>>;
|
||||
|
||||
async function createSessionDaemonRuntime() {
|
||||
const eventHub = new SessionEventHub();
|
||||
const notificationStore = new SessionNotificationStore();
|
||||
const unreadStore = new SessionUnreadStore({
|
||||
persistence: new FileSessionUnreadPersistence(),
|
||||
onPersistenceError(operation, error) {
|
||||
app.log.error({ err: error, operation }, "session unread persistence failed");
|
||||
},
|
||||
});
|
||||
await unreadStore.load();
|
||||
const workspaceActivity = new WorkspaceActivityService(eventHub);
|
||||
const auth = await AuthService.create({ agentDir: activeAgentProfile.dir, logger: app.log });
|
||||
// Capture providers registered by global extensions while the runtime is
|
||||
// still mutable, then freeze every later extension-provider mutation before
|
||||
// any real session can load project resources.
|
||||
await bootstrapAndFreezeGlobalExtensionProviders(auth.runtime, activeAgentProfile.dir, app.log);
|
||||
// The shared model runtime is constructed offline so request paths never
|
||||
// wait on provider-catalog fetches; this is the single bounded network
|
||||
// refresher, and auth changes (login/logout) ask it for a prompt run. It
|
||||
// stays fully inert when the operator asked for offline behavior.
|
||||
const catalogRefresher = new ModelCatalogRefresher({
|
||||
runtime: auth.runtime,
|
||||
logger: app.log,
|
||||
offline: offlineModeEnabled(daemonEnvironment),
|
||||
});
|
||||
catalogRefresher.start();
|
||||
auth.subscribe(() => { catalogRefresher.requestRefresh(); });
|
||||
// Cross-workspace session relationships are reported regardless of whether
|
||||
// agents may spawn sessions: children can predate a config change, and the
|
||||
// session tree should stay honest about them either way.
|
||||
const projectWorkspaceDeps = { projects: new ProjectService(new ProjectStore()), workspaces: new WorkspaceService() };
|
||||
const projectWorkspaces = new RegisteredProjectWorkspaceCwds(projectWorkspaceDeps);
|
||||
const spawnTargets = config.spawnSessions ? new ProjectScopedSpawnTargetResolver(projectWorkspaceDeps) : undefined;
|
||||
const sessions = new PiSessionService(eventHub, sessionServiceDependencies({
|
||||
modelRuntime: auth.runtime,
|
||||
agentDir: activeAgentProfile.dir,
|
||||
workspaceActivity,
|
||||
logger: app.log,
|
||||
...(spawnTargets === undefined ? {} : { spawnTargets }),
|
||||
projectWorkspaces,
|
||||
subsessionsEnabled: config.subsessions,
|
||||
askUserEnabled: config.askUser,
|
||||
notificationStore,
|
||||
unreadStore,
|
||||
catalogRefreshStatus: catalogRefresher,
|
||||
sessionManager: createPiSessionManagerGateway({
|
||||
agentDir: activeAgentProfile.dir,
|
||||
workspaceActivity,
|
||||
logger: app.log,
|
||||
...(spawnTargets === undefined ? {} : { spawnTargets }),
|
||||
projectWorkspaces,
|
||||
subsessionsEnabled: config.subsessions,
|
||||
askUserEnabled: config.askUser,
|
||||
notificationStore,
|
||||
unreadStore,
|
||||
catalogRefreshStatus: catalogRefresher,
|
||||
sessionManager: createPiSessionManagerGateway({
|
||||
agentDir: activeAgentProfile.dir,
|
||||
env: daemonEnvironment,
|
||||
sessionDirEnvKeys: activeAgentProfile.sessionDirEnvKeys,
|
||||
}),
|
||||
}));
|
||||
auth.subscribe((change) => { sessions.applyAuthChange(change); });
|
||||
const terminals = new TerminalService(eventHub, workspaceActivity);
|
||||
const runtimeComponent = Object.freeze({
|
||||
...getPiWebRuntimeComponent("sessiond", SESSIOND_RUNTIME_CAPABILITIES),
|
||||
activeAgentProfile,
|
||||
});
|
||||
return { eventHub, workspaceActivity, auth, sessions, terminals, unreadStore, activeAgentProfile, runtimeComponent, catalogRefresher };
|
||||
},
|
||||
registerRoutes({ eventHub, workspaceActivity, auth, sessions, terminals, runtimeComponent }) {
|
||||
registerWorkspaceActivityRoutes(app, workspaceActivity);
|
||||
registerAuthRoutes(app, auth);
|
||||
registerSessionRoutes(app, sessions, eventHub);
|
||||
registerTerminalRoutes(app, terminals);
|
||||
env: daemonEnvironment,
|
||||
sessionDirEnvKeys: activeAgentProfile.sessionDirEnvKeys,
|
||||
}),
|
||||
}));
|
||||
auth.subscribe((change) => { sessions.applyAuthChange(change); });
|
||||
const terminals = new TerminalService(eventHub, workspaceActivity);
|
||||
const runtimeComponent = Object.freeze({
|
||||
...getPiWebRuntimeComponent("sessiond", SESSIOND_RUNTIME_CAPABILITIES),
|
||||
activeAgentProfile,
|
||||
});
|
||||
return { eventHub, workspaceActivity, auth, sessions, terminals, unreadStore, activeAgentProfile, runtimeComponent, catalogRefresher };
|
||||
}
|
||||
|
||||
app.get("/health", () => ({
|
||||
ok: true,
|
||||
activeSessions: sessions.activeCount(),
|
||||
checkedAt: new Date().toISOString(),
|
||||
version: {
|
||||
component: runtimeComponent.component,
|
||||
label: runtimeComponent.label,
|
||||
...(runtimeComponent.runtimeVersion === undefined ? {} : { runtimeVersion: runtimeComponent.runtimeVersion }),
|
||||
stale: false,
|
||||
available: runtimeComponent.available,
|
||||
},
|
||||
}));
|
||||
function registerSessionDaemonRoutes({ eventHub, workspaceActivity, auth, sessions, terminals, runtimeComponent }: SessionDaemonRuntime): void {
|
||||
registerWorkspaceActivityRoutes(app, workspaceActivity);
|
||||
registerAuthRoutes(app, auth);
|
||||
registerSessionRoutes(app, sessions, eventHub);
|
||||
registerTerminalRoutes(app, terminals);
|
||||
|
||||
app.get("/runtime", () => runtimeComponent);
|
||||
},
|
||||
async listen({ auth, sessions, terminals, unreadStore, catalogRefresher }) {
|
||||
let shuttingDown = false;
|
||||
async function shutdown(signal: NodeJS.Signals): Promise<void> {
|
||||
if (shuttingDown) return;
|
||||
shuttingDown = true;
|
||||
app.log.info({ signal }, "shutting down session daemon");
|
||||
const attempt = async (operation: string, run: () => void | Promise<void>): Promise<void> => {
|
||||
try {
|
||||
await run();
|
||||
} catch (error: unknown) {
|
||||
process.exitCode = 1;
|
||||
app.log.error({ err: error, operation }, "session daemon shutdown operation failed");
|
||||
}
|
||||
};
|
||||
await attempt("dispose terminals", () => { terminals.dispose(); });
|
||||
await attempt("dispose catalog refresher", () => { catalogRefresher.dispose(); });
|
||||
await attempt("dispose auth", () => { auth.dispose(); });
|
||||
await attempt("dispose sessions", () => sessions.dispose());
|
||||
await attempt("flush session unread state", () => unreadStore.flush());
|
||||
await attempt("close server", () => app.close());
|
||||
}
|
||||
app.get("/health", () => ({
|
||||
ok: true,
|
||||
activeSessions: sessions.activeCount(),
|
||||
checkedAt: new Date().toISOString(),
|
||||
version: {
|
||||
component: runtimeComponent.component,
|
||||
label: runtimeComponent.label,
|
||||
...(runtimeComponent.runtimeVersion === undefined ? {} : { runtimeVersion: runtimeComponent.runtimeVersion }),
|
||||
stale: false,
|
||||
available: runtimeComponent.available,
|
||||
},
|
||||
}));
|
||||
|
||||
process.once("SIGINT", (signal) => { void shutdown(signal); });
|
||||
process.once("SIGTERM", (signal) => { void shutdown(signal); });
|
||||
app.get("/runtime", () => runtimeComponent);
|
||||
}
|
||||
|
||||
const portValue = daemonEnvironment["PI_WEB_SESSIOND_PORT"];
|
||||
const port = portValue !== undefined && portValue !== "" ? Number(portValue) : undefined;
|
||||
const host = daemonEnvironment["PI_WEB_SESSIOND_HOST"] ?? "127.0.0.1";
|
||||
async function listenSessionDaemon({ auth, sessions, terminals, unreadStore, catalogRefresher }: SessionDaemonRuntime): 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");
|
||||
const attempt = async (operation: string, run: () => void | Promise<void>): Promise<void> => {
|
||||
try {
|
||||
await run();
|
||||
} catch (error: unknown) {
|
||||
process.exitCode = 1;
|
||||
app.log.error({ err: error, operation }, "session daemon shutdown operation failed");
|
||||
}
|
||||
};
|
||||
await attempt("dispose terminals", () => { terminals.dispose(); });
|
||||
await attempt("dispose catalog refresher", () => { catalogRefresher.dispose(); });
|
||||
await attempt("dispose auth", () => { auth.dispose(); });
|
||||
await attempt("dispose sessions", () => sessions.dispose());
|
||||
await attempt("flush session unread state", () => unreadStore.flush());
|
||||
await attempt("close server", () => app.close());
|
||||
}
|
||||
|
||||
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 }));
|
||||
}
|
||||
},
|
||||
});
|
||||
process.once("SIGINT", (signal) => { void shutdown(signal); });
|
||||
process.once("SIGTERM", (signal) => { void shutdown(signal); });
|
||||
|
||||
const portValue = daemonEnvironment["PI_WEB_SESSIOND_PORT"];
|
||||
const port = portValue !== undefined && portValue !== "" ? Number(portValue) : undefined;
|
||||
const host = daemonEnvironment["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 }));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,261 +0,0 @@
|
||||
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 };
|
||||
}
|
||||
@@ -1,95 +0,0 @@
|
||||
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 | Promise<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 = await 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",
|
||||
);
|
||||
}
|
||||
@@ -1,555 +0,0 @@
|
||||
import { constants } from "node:fs";
|
||||
import {
|
||||
access,
|
||||
appendFile,
|
||||
copyFile,
|
||||
link as createHardLink,
|
||||
lstat,
|
||||
mkdtemp,
|
||||
mkdir,
|
||||
readFile,
|
||||
readdir,
|
||||
realpath,
|
||||
rm,
|
||||
unlink,
|
||||
writeFile,
|
||||
} from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { dirname, join, resolve, sep } from "node:path";
|
||||
import { afterEach, describe, expect, it } from "vitest";
|
||||
import {
|
||||
inspectLegacySessionArchiveMigration,
|
||||
migrateLegacySessionArchive,
|
||||
type LegacySessionArchiveMigrationOptions,
|
||||
} from "./sessionArchiveMigration.js";
|
||||
|
||||
const tempRoots: string[] = [];
|
||||
|
||||
describe("legacy session archive migration preflight", () => {
|
||||
afterEach(async () => {
|
||||
await Promise.all(tempRoots.splice(0).map((path) => rm(path, { recursive: true, force: true })));
|
||||
});
|
||||
|
||||
it("proves the ordinary legacy layout eligible without creating destination state", async () => {
|
||||
const fixture = await createLegacyArchiveFixture();
|
||||
|
||||
await expect(inspectLegacySessionArchiveMigration(fixture.options)).resolves.toEqual({
|
||||
status: "eligible",
|
||||
legacyIndexPath: fixture.legacyIndexPath,
|
||||
destinationIndexPath: fixture.destinationIndexPath,
|
||||
archiveFileCount: 1,
|
||||
});
|
||||
expect(await exists(fixture.destinationRoot)).toBe(false);
|
||||
await expect(readFile(fixture.legacyFilePath, "utf8")).resolves.toBe("legacy session\n");
|
||||
});
|
||||
|
||||
it("skips when PI_WEB_DATA_DIR is not explicitly configured", async () => {
|
||||
const fixture = await createLegacyArchiveFixture();
|
||||
|
||||
await expect(migrateLegacySessionArchive({ ...fixture.options, env: {} })).resolves.toEqual({
|
||||
status: "skipped",
|
||||
reason: "data-dir-not-configured",
|
||||
});
|
||||
await expectLegacyStateUntouched(fixture);
|
||||
expect(await exists(fixture.destinationRoot)).toBe(false);
|
||||
});
|
||||
|
||||
it("skips when the configured data root resolves to the legacy root", async () => {
|
||||
const fixture = await createLegacyArchiveFixture();
|
||||
|
||||
await expect(migrateLegacySessionArchive({
|
||||
...fixture.options,
|
||||
env: { PI_WEB_DATA_DIR: fixture.legacyRoot },
|
||||
})).resolves.toEqual({ status: "skipped", reason: "data-dir-not-distinct" });
|
||||
await expectLegacyStateUntouched(fixture);
|
||||
});
|
||||
|
||||
it("skips when no legacy index exists without adopting loose legacy files", async () => {
|
||||
const fixture = await createLegacyArchiveFixture();
|
||||
await unlink(fixture.legacyIndexPath);
|
||||
|
||||
await expect(migrateLegacySessionArchive(fixture.options)).resolves.toEqual({
|
||||
status: "skipped",
|
||||
reason: "legacy-index-missing",
|
||||
});
|
||||
await expect(readFile(fixture.legacyFilePath, "utf8")).resolves.toBe("legacy session\n");
|
||||
await expect(readFile(fixture.activeFilePath, "utf8")).resolves.toBe("active session\n");
|
||||
expect(await exists(fixture.destinationRoot)).toBe(false);
|
||||
});
|
||||
|
||||
it("skips malformed legacy indexes without creating destination state", async () => {
|
||||
const fixture = await createLegacyArchiveFixture();
|
||||
await writeFile(fixture.legacyIndexPath, "not json\n", "utf8");
|
||||
|
||||
await expect(migrateLegacySessionArchive(fixture.options)).resolves.toEqual({
|
||||
status: "skipped",
|
||||
reason: "legacy-index-invalid",
|
||||
});
|
||||
await expect(readFile(fixture.legacyIndexPath, "utf8")).resolves.toBe("not json\n");
|
||||
expect(await exists(fixture.destinationRoot)).toBe(false);
|
||||
});
|
||||
|
||||
it("treats inconclusive filesystem inspection as a mutation-free skip", async () => {
|
||||
const fixture = await createLegacyArchiveFixture();
|
||||
const inspectionError = Object.assign(new Error("inspection denied"), { code: "EACCES" });
|
||||
|
||||
const result = await migrateLegacySessionArchive({
|
||||
...fixture.options,
|
||||
fileSystem: {
|
||||
lstat: (path) => path === fixture.destinationIndexPath
|
||||
? Promise.reject(inspectionError)
|
||||
: lstat(path),
|
||||
},
|
||||
});
|
||||
|
||||
expect(result).toMatchObject({
|
||||
status: "skipped",
|
||||
reason: "inspection-failed",
|
||||
error: inspectionError,
|
||||
});
|
||||
await expectLegacyStateUntouched(fixture);
|
||||
expect(await exists(fixture.destinationRoot)).toBe(false);
|
||||
});
|
||||
|
||||
it("does not merge with or overwrite an existing destination index", async () => {
|
||||
const fixture = await createLegacyArchiveFixture();
|
||||
await mkdir(fixture.destinationRoot, { recursive: true });
|
||||
await writeFile(fixture.destinationIndexPath, "destination owner\n", "utf8");
|
||||
|
||||
await expect(migrateLegacySessionArchive(fixture.options)).resolves.toEqual({
|
||||
status: "skipped",
|
||||
reason: "destination-index-exists",
|
||||
});
|
||||
await expect(readFile(fixture.destinationIndexPath, "utf8")).resolves.toBe("destination owner\n");
|
||||
await expectLegacyStateUntouched(fixture);
|
||||
});
|
||||
|
||||
it("leaves an interrupted pre-commit file untouched instead of adopting a non-empty destination archive", async () => {
|
||||
const fixture = await createLegacyArchiveFixture({ createDestinationArchive: true });
|
||||
await writeFile(fixture.destinationFilePath, "partial destination copy\n", "utf8");
|
||||
|
||||
await expect(migrateLegacySessionArchive(fixture.options)).resolves.toEqual({
|
||||
status: "skipped",
|
||||
reason: "destination-archive-not-empty-or-invalid",
|
||||
});
|
||||
await expect(readFile(fixture.destinationFilePath, "utf8")).resolves.toBe("partial destination copy\n");
|
||||
await expectLegacyStateUntouched(fixture);
|
||||
});
|
||||
|
||||
it("skips archive paths that are not direct regular files in the legacy archive directory", async () => {
|
||||
const fixture = await createLegacyArchiveFixture();
|
||||
const outsidePath = join(fixture.root, "outside.jsonl");
|
||||
await writeFile(outsidePath, "outside\n", "utf8");
|
||||
const firstRecord = requiredRecord(fixture.document.sessions, 0);
|
||||
const changedDocument = {
|
||||
...fixture.document,
|
||||
sessions: [{ ...firstRecord, archivePath: outsidePath }, ...fixture.document.sessions.slice(1)],
|
||||
};
|
||||
await writeArchiveIndex(fixture.legacyIndexPath, changedDocument);
|
||||
|
||||
await expect(migrateLegacySessionArchive(fixture.options)).resolves.toEqual({
|
||||
status: "skipped",
|
||||
reason: "legacy-archive-layout-invalid",
|
||||
});
|
||||
await expect(readFile(outsidePath, "utf8")).resolves.toBe("outside\n");
|
||||
await expect(readFile(fixture.legacyFilePath, "utf8")).resolves.toBe("legacy session\n");
|
||||
expect(await exists(fixture.destinationRoot)).toBe(false);
|
||||
});
|
||||
|
||||
it("uses Windows case-insensitive path semantics without weakening Linux containment", async () => {
|
||||
const fixture = await createLegacyArchiveFixture();
|
||||
const casedArchivePath = resolve(fixture.legacyFilePath.toUpperCase());
|
||||
const firstRecord = requiredRecord(fixture.document.sessions, 0);
|
||||
await writeArchiveIndex(fixture.legacyIndexPath, {
|
||||
...fixture.document,
|
||||
sessions: [
|
||||
{ ...firstRecord, archivePath: casedArchivePath },
|
||||
...fixture.document.sessions.slice(1),
|
||||
],
|
||||
});
|
||||
const mapCaseVariant = (path: string): string => path.toLowerCase() === casedArchivePath.toLowerCase()
|
||||
? fixture.legacyFilePath
|
||||
: path;
|
||||
const fileSystem = {
|
||||
lstat: (path: string) => lstat(mapCaseVariant(path)),
|
||||
realpath: (path: string) => realpath(mapCaseVariant(path)),
|
||||
};
|
||||
|
||||
await expect(inspectLegacySessionArchiveMigration({
|
||||
...fixture.options,
|
||||
platform: "linux",
|
||||
fileSystem,
|
||||
})).resolves.toEqual({ status: "skipped", reason: "legacy-archive-layout-invalid" });
|
||||
await expect(inspectLegacySessionArchiveMigration({
|
||||
...fixture.options,
|
||||
platform: "win32",
|
||||
fileSystem,
|
||||
})).resolves.toEqual({
|
||||
status: "eligible",
|
||||
legacyIndexPath: fixture.legacyIndexPath,
|
||||
destinationIndexPath: fixture.destinationIndexPath,
|
||||
archiveFileCount: 1,
|
||||
});
|
||||
});
|
||||
|
||||
it("skips legacy directories containing unindexed entries", async () => {
|
||||
const fixture = await createLegacyArchiveFixture();
|
||||
const unexpectedPath = join(fixture.legacyArchiveDir, "unexpected.tmp");
|
||||
await writeFile(unexpectedPath, "unexpected\n", "utf8");
|
||||
|
||||
await expect(migrateLegacySessionArchive(fixture.options)).resolves.toEqual({
|
||||
status: "skipped",
|
||||
reason: "legacy-archive-layout-invalid",
|
||||
});
|
||||
await expect(readFile(unexpectedPath, "utf8")).resolves.toBe("unexpected\n");
|
||||
await expectLegacyStateUntouched(fixture);
|
||||
expect(await exists(fixture.destinationRoot)).toBe(false);
|
||||
});
|
||||
|
||||
it("skips duplicate session IDs", async () => {
|
||||
const fixture = await createLegacyArchiveFixture();
|
||||
const firstRecord = requiredRecord(fixture.document.sessions, 0);
|
||||
const secondRecord = requiredRecord(fixture.document.sessions, 1);
|
||||
await writeArchiveIndex(fixture.legacyIndexPath, {
|
||||
...fixture.document,
|
||||
sessions: [firstRecord, { ...secondRecord, sessionId: firstRecord["sessionId"] }],
|
||||
});
|
||||
|
||||
await expect(migrateLegacySessionArchive(fixture.options)).resolves.toEqual({
|
||||
status: "skipped",
|
||||
reason: "archive-record-conflict",
|
||||
});
|
||||
await expect(readFile(fixture.legacyFilePath, "utf8")).resolves.toBe("legacy session\n");
|
||||
expect(await exists(fixture.destinationRoot)).toBe(false);
|
||||
});
|
||||
|
||||
it("skips duplicate source and destination file mappings", async () => {
|
||||
const fixture = await createLegacyArchiveFixture();
|
||||
const firstRecord = requiredRecord(fixture.document.sessions, 0);
|
||||
await writeArchiveIndex(fixture.legacyIndexPath, {
|
||||
...fixture.document,
|
||||
sessions: [firstRecord, { ...firstRecord, sessionId: "second-file-record" }],
|
||||
});
|
||||
|
||||
await expect(migrateLegacySessionArchive(fixture.options)).resolves.toEqual({
|
||||
status: "skipped",
|
||||
reason: "archive-record-conflict",
|
||||
});
|
||||
await expectLegacyStateUntouched({
|
||||
...fixture,
|
||||
sourceIndexContents: await readFile(fixture.legacyIndexPath, "utf8"),
|
||||
});
|
||||
expect(await exists(fixture.destinationRoot)).toBe(false);
|
||||
});
|
||||
});
|
||||
|
||||
describe("legacy session archive migration execution", () => {
|
||||
afterEach(async () => {
|
||||
await Promise.all(tempRoots.splice(0).map((path) => rm(path, { recursive: true, force: true })));
|
||||
});
|
||||
|
||||
it("copies across the source boundary, atomically publishes the rewritten index, then removes legacy state", async () => {
|
||||
const fixture = await createLegacyArchiveFixture({ createDestinationArchive: true });
|
||||
const copies: { source: string; destination: string; mode: number }[] = [];
|
||||
const links: { source: string; destination: string }[] = [];
|
||||
|
||||
await expect(migrateLegacySessionArchive({
|
||||
...fixture.options,
|
||||
fileSystem: {
|
||||
copyFile: async (source, destination, mode) => {
|
||||
copies.push({ source, destination, mode });
|
||||
await copyFile(source, destination, mode);
|
||||
},
|
||||
link: async (source, destination) => {
|
||||
links.push({ source, destination });
|
||||
await createHardLink(source, destination);
|
||||
},
|
||||
unlink: async (path) => {
|
||||
if (path === fixture.legacyFilePath || path === fixture.legacyIndexPath) {
|
||||
expect(await exists(fixture.destinationIndexPath)).toBe(true);
|
||||
}
|
||||
await unlink(path);
|
||||
},
|
||||
},
|
||||
})).resolves.toEqual({ status: "migrated", archiveFileCount: 1, cleanup: "complete" });
|
||||
|
||||
expect(copies).toHaveLength(2);
|
||||
expect(copies[0]).toMatchObject({ source: fixture.legacyFilePath, mode: constants.COPYFILE_EXCL });
|
||||
expect(copies[0]?.destination).toContain(".archived-sessions-migration-test-attempt");
|
||||
expect(copies[1]).toMatchObject({ destination: fixture.destinationFilePath, mode: constants.COPYFILE_EXCL });
|
||||
expect(links).toEqual([{
|
||||
source: join(fixture.destinationRoot, ".archived-sessions-migration-test-attempt", "archived-sessions.json"),
|
||||
destination: fixture.destinationIndexPath,
|
||||
}]);
|
||||
await expect(readFile(fixture.destinationFilePath, "utf8")).resolves.toBe("legacy session\n");
|
||||
|
||||
const migratedDocument: unknown = JSON.parse(await readFile(fixture.destinationIndexPath, "utf8"));
|
||||
const firstRecord = requiredRecord(fixture.document.sessions, 0);
|
||||
expect(migratedDocument).toEqual({
|
||||
...fixture.document,
|
||||
sessions: [
|
||||
{ ...firstRecord, archivePath: fixture.destinationFilePath },
|
||||
requiredRecord(fixture.document.sessions, 1),
|
||||
],
|
||||
});
|
||||
expect(await exists(fixture.legacyIndexPath)).toBe(false);
|
||||
expect(await exists(fixture.legacyFilePath)).toBe(false);
|
||||
expect(await exists(fixture.legacyArchiveDir)).toBe(false);
|
||||
await expect(readFile(fixture.activeFilePath, "utf8")).resolves.toBe("active session\n");
|
||||
expect(new Set(await readdir(fixture.destinationRoot))).toEqual(new Set([
|
||||
"archived-sessions",
|
||||
"archived-sessions.json",
|
||||
]));
|
||||
});
|
||||
|
||||
it("retries safely when an interrupted staging-only attempt left an unowned sibling tree", async () => {
|
||||
const fixture = await createLegacyArchiveFixture();
|
||||
const abandonedFile = join(
|
||||
fixture.destinationRoot,
|
||||
".archived-sessions-migration-interrupted-attempt",
|
||||
"files",
|
||||
"abandoned.jsonl",
|
||||
);
|
||||
await mkdir(dirname(abandonedFile), { recursive: true });
|
||||
await writeFile(abandonedFile, "unowned staging data\n", "utf8");
|
||||
|
||||
await expect(migrateLegacySessionArchive(fixture.options)).resolves.toEqual({
|
||||
status: "migrated",
|
||||
archiveFileCount: 1,
|
||||
cleanup: "complete",
|
||||
});
|
||||
|
||||
await expect(readFile(abandonedFile, "utf8")).resolves.toBe("unowned staging data\n");
|
||||
await expect(readFile(fixture.destinationFilePath, "utf8")).resolves.toBe("legacy session\n");
|
||||
expect(await exists(fixture.destinationIndexPath)).toBe(true);
|
||||
expect(await exists(fixture.legacyIndexPath)).toBe(false);
|
||||
});
|
||||
|
||||
it("rolls back destination artifacts and preserves all legacy state when staged-copy verification fails", async () => {
|
||||
const fixture = await createLegacyArchiveFixture();
|
||||
|
||||
const result = await migrateLegacySessionArchive({
|
||||
...fixture.options,
|
||||
fileSystem: {
|
||||
copyFile: async (source, destination, mode) => {
|
||||
await copyFile(source, destination, mode);
|
||||
if (source === fixture.legacyFilePath) await appendFile(destination, "corrupt", "utf8");
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
expect(result).toMatchObject({ status: "failed", phase: "stage", rollbackErrors: [] });
|
||||
await expectLegacyStateUntouched(fixture);
|
||||
expect(await exists(fixture.destinationIndexPath)).toBe(false);
|
||||
await expect(readdir(fixture.destinationRoot)).resolves.toEqual([]);
|
||||
});
|
||||
|
||||
it("revalidates source files before commit and rolls back if one changes during migration", async () => {
|
||||
const fixture = await createLegacyArchiveFixture();
|
||||
|
||||
const result = await migrateLegacySessionArchive({
|
||||
...fixture.options,
|
||||
fileSystem: {
|
||||
copyFile: async (source, destination, mode) => {
|
||||
await copyFile(source, destination, mode);
|
||||
if (destination === fixture.destinationFilePath) {
|
||||
await appendFile(fixture.legacyFilePath, "changed during migration\n", "utf8");
|
||||
}
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
expect(result).toMatchObject({ status: "failed", phase: "commit-index", rollbackErrors: [] });
|
||||
await expect(readFile(fixture.legacyIndexPath, "utf8")).resolves.toBe(fixture.sourceIndexContents);
|
||||
await expect(readFile(fixture.legacyFilePath, "utf8")).resolves.toBe(
|
||||
"legacy session\nchanged during migration\n",
|
||||
);
|
||||
await expect(readFile(fixture.activeFilePath, "utf8")).resolves.toBe("active session\n");
|
||||
expect(await exists(fixture.destinationIndexPath)).toBe(false);
|
||||
expect(await exists(fixture.destinationArchiveDir)).toBe(false);
|
||||
});
|
||||
|
||||
it("rolls back published files but never source state when atomic index publication fails", async () => {
|
||||
const fixture = await createLegacyArchiveFixture();
|
||||
const publicationError = Object.assign(new Error("link failed"), { code: "EIO" });
|
||||
|
||||
const result = await migrateLegacySessionArchive({
|
||||
...fixture.options,
|
||||
fileSystem: {
|
||||
link: () => Promise.reject(publicationError),
|
||||
},
|
||||
});
|
||||
|
||||
expect(result).toMatchObject({
|
||||
status: "failed",
|
||||
phase: "commit-index",
|
||||
error: publicationError,
|
||||
rollbackErrors: [],
|
||||
});
|
||||
await expectLegacyStateUntouched(fixture);
|
||||
expect(await exists(fixture.destinationIndexPath)).toBe(false);
|
||||
expect(await exists(fixture.destinationArchiveDir)).toBe(false);
|
||||
await expect(readdir(fixture.destinationRoot)).resolves.toEqual([]);
|
||||
});
|
||||
|
||||
it("does not overwrite a destination index that appears at the atomic commit boundary", async () => {
|
||||
const fixture = await createLegacyArchiveFixture();
|
||||
const publicationError = Object.assign(new Error("destination index won the race"), { code: "EEXIST" });
|
||||
|
||||
const result = await migrateLegacySessionArchive({
|
||||
...fixture.options,
|
||||
fileSystem: {
|
||||
link: async (_source, destination) => {
|
||||
await writeFile(destination, "destination owner\n", { encoding: "utf8", flag: "wx" });
|
||||
throw publicationError;
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
expect(result).toMatchObject({
|
||||
status: "failed",
|
||||
phase: "commit-index",
|
||||
error: publicationError,
|
||||
rollbackErrors: [],
|
||||
});
|
||||
await expect(readFile(fixture.destinationIndexPath, "utf8")).resolves.toBe("destination owner\n");
|
||||
expect(await exists(fixture.destinationArchiveDir)).toBe(false);
|
||||
await expectLegacyStateUntouched(fixture);
|
||||
});
|
||||
|
||||
it("keeps the committed destination authoritative when legacy cleanup fails", async () => {
|
||||
const fixture = await createLegacyArchiveFixture();
|
||||
const cleanupError = Object.assign(new Error("source cleanup failed"), { code: "EACCES" });
|
||||
|
||||
const result = await migrateLegacySessionArchive({
|
||||
...fixture.options,
|
||||
fileSystem: {
|
||||
unlink: async (path) => {
|
||||
if (path === fixture.legacyFilePath) throw cleanupError;
|
||||
await unlink(path);
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
expect(result).toMatchObject({
|
||||
status: "migrated",
|
||||
cleanup: "incomplete",
|
||||
cleanupErrors: [cleanupError],
|
||||
});
|
||||
await expect(readFile(fixture.destinationFilePath, "utf8")).resolves.toBe("legacy session\n");
|
||||
expect(await exists(fixture.destinationIndexPath)).toBe(true);
|
||||
await expectLegacyStateUntouched(fixture);
|
||||
});
|
||||
});
|
||||
|
||||
interface LegacyArchiveFixture {
|
||||
root: string;
|
||||
legacyRoot: string;
|
||||
legacyIndexPath: string;
|
||||
legacyArchiveDir: string;
|
||||
legacyFilePath: string;
|
||||
activeFilePath: string;
|
||||
destinationRoot: string;
|
||||
destinationIndexPath: string;
|
||||
destinationArchiveDir: string;
|
||||
destinationFilePath: string;
|
||||
sourceIndexContents: string;
|
||||
document: { marker: string; sessions: Record<string, unknown>[] };
|
||||
options: LegacySessionArchiveMigrationOptions;
|
||||
}
|
||||
|
||||
async function createLegacyArchiveFixture(
|
||||
setup: { createDestinationArchive?: boolean } = {},
|
||||
): Promise<LegacyArchiveFixture> {
|
||||
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<string> {
|
||||
const contents = `${JSON.stringify(document, null, 2)}\n`;
|
||||
await writeFile(path, contents, "utf8");
|
||||
return contents;
|
||||
}
|
||||
|
||||
async function expectLegacyStateUntouched(fixture: LegacyArchiveFixture): Promise<void> {
|
||||
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<string, unknown>[], index: number): Record<string, unknown> {
|
||||
const record = records[index];
|
||||
if (record === undefined) throw new Error(`Missing fixture record ${index.toString()}`);
|
||||
return record;
|
||||
}
|
||||
|
||||
async function exists(path: string): Promise<boolean> {
|
||||
try {
|
||||
await access(path, constants.F_OK);
|
||||
return true;
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
@@ -1,781 +0,0 @@
|
||||
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<void>;
|
||||
}
|
||||
|
||||
export interface SessionArchiveMigrationFileSystem {
|
||||
lstat(path: string): Promise<Stats>;
|
||||
realpath(path: string): Promise<string>;
|
||||
readFile(path: string): Promise<string>;
|
||||
readdir(path: string): Promise<Dirent[]>;
|
||||
mkdir(path: string, options?: { recursive?: boolean }): Promise<void>;
|
||||
copyFile(source: string, destination: string, mode: number): Promise<void>;
|
||||
writeFile(path: string, contents: string): Promise<void>;
|
||||
link(source: string, destination: string): Promise<void>;
|
||||
unlink(path: string): Promise<void>;
|
||||
rmdir(path: string): Promise<void>;
|
||||
rmOwnedTree(path: string): Promise<void>;
|
||||
open(path: string): Promise<SessionArchiveMigrationReadHandle>;
|
||||
}
|
||||
|
||||
export interface LegacySessionArchiveMigrationOptions {
|
||||
env?: NodeJS.ProcessEnv;
|
||||
cwd?: string;
|
||||
homeDir?: string;
|
||||
platform?: NodeJS.Platform;
|
||||
createAttemptId?: () => string;
|
||||
fileSystem?: Partial<SessionArchiveMigrationFileSystem>;
|
||||
}
|
||||
|
||||
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<LegacySessionArchiveMigrationPreflight> {
|
||||
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<LegacySessionArchiveMigrationResult> {
|
||||
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<string, string>();
|
||||
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<InternalPreflight> {
|
||||
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<string, unknown>;
|
||||
rawSessions: Record<string, unknown>[];
|
||||
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<LegacySessionArchiveMigrationSkipped | { status: "planned"; files: PlannedArchiveFile[]; destinationPathsByRecord: (string | undefined)[] }> {
|
||||
const sessionIds = new Set<string>();
|
||||
const sourcePaths = new Set<string>();
|
||||
const destinationPaths = new Set<string>();
|
||||
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<boolean> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<unknown[]> {
|
||||
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<unknown[]> {
|
||||
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<boolean> {
|
||||
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<number> {
|
||||
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<string> {
|
||||
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<Stats | undefined> {
|
||||
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<void> {
|
||||
try {
|
||||
await fileSystem.unlink(path);
|
||||
} catch (error: unknown) {
|
||||
if (!isNodeErrorWithCode(error, "ENOENT")) throw error;
|
||||
}
|
||||
}
|
||||
|
||||
async function removeDirectoryIfPresent(path: string, fileSystem: SessionArchiveMigrationFileSystem): Promise<void> {
|
||||
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<string, unknown>[] | undefined {
|
||||
if (!Array.isArray(value)) return undefined;
|
||||
const records: Record<string, unknown>[] = [];
|
||||
for (const item of value) {
|
||||
if (!isRecord(item)) return undefined;
|
||||
records.push(item);
|
||||
}
|
||||
return records;
|
||||
}
|
||||
|
||||
function isRecord(value: unknown): value is Record<string, unknown> {
|
||||
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;
|
||||
}
|
||||
Reference in New Issue
Block a user