From 9c028a7353a774ca7852a285a45e2494b1a21319 Mon Sep 17 00:00:00 2001 From: Federico Jaramillo Martinez Date: Mon, 18 May 2026 21:58:40 +0200 Subject: [PATCH] feat: move archived sessions out of active storage --- .changeset/archive-sessions-move-files.md | 5 + src/server/sessions/piSessionService.test.ts | 29 +++ src/server/sessions/piSessionService.ts | 166 +++++++++++--- .../sessions/sessionArchiveStore.test.ts | 56 +++++ src/server/sessions/sessionArchiveStore.ts | 206 ++++++++++++++++-- 5 files changed, 419 insertions(+), 43 deletions(-) create mode 100644 .changeset/archive-sessions-move-files.md create mode 100644 src/server/sessions/sessionArchiveStore.test.ts diff --git a/.changeset/archive-sessions-move-files.md b/.changeset/archive-sessions-move-files.md new file mode 100644 index 0000000..6b9767d --- /dev/null +++ b/.changeset/archive-sessions-move-files.md @@ -0,0 +1,5 @@ +--- +"@jmfederico/pi-web": patch +--- + +Move archived session files out of active Pi session directories so normal session lists no longer scan archived histories. diff --git a/src/server/sessions/piSessionService.test.ts b/src/server/sessions/piSessionService.test.ts index 0e4f6a1..75cffa9 100644 --- a/src/server/sessions/piSessionService.test.ts +++ b/src/server/sessions/piSessionService.test.ts @@ -152,6 +152,7 @@ describe("PiSessionService", () => { const service = new PiSessionService(new CapturingSessionEventHub(), { archiveStore: { list: () => Promise.resolve([{ sessionId: "archived", cwd: "/workspace", archivedAt: "2026-01-01T00:00:00.000Z" }]), + get: () => Promise.resolve(undefined), archive: () => Promise.resolve({ sessionId: "archived", cwd: "/workspace", archivedAt: "2026-01-01T00:00:00.000Z" }), restore: () => Promise.resolve(), isArchived: () => Promise.resolve(false), @@ -177,6 +178,34 @@ describe("PiSessionService", () => { await service.dispose(); }); + it("lists archived records that have been moved out of the active session directory", async () => { + const service = new PiSessionService(new CapturingSessionEventHub(), { + archiveStore: { + list: () => Promise.resolve([{ sessionId: "archived", cwd: "/workspace", archivedAt: "2026-01-02T00:00:00.000Z", originalPath: "/sessions/archived.jsonl", archivePath: "/archive/archived.jsonl", created: "2026-01-01T00:00:00.000Z", modified: "2026-01-01T00:01:00.000Z", messageCount: 2, firstMessage: "bye" }]), + get: () => Promise.resolve(undefined), + archive: () => { throw new Error("archive should not be called for moved records"); }, + restore: () => Promise.resolve(), + isArchived: () => Promise.resolve(false), + }, + sessionManager: { + create: () => fakeSessionManager(), + list: () => Promise.resolve([{ ...sessionRecord("active"), messageCount: 1, firstMessage: "hello", allMessagesText: "hello" }]), + listAll: () => Promise.resolve([]), + open: () => fakeSessionManager(), + }, + heartbeatIntervalMs: 60_000, + }); + + const sessions = await service.list("/workspace"); + + expect(sessions).toHaveLength(2); + expect(sessions[0]).toMatchObject({ id: "active" }); + expect(sessions[0]?.archived).toBeUndefined(); + expect(sessions[1]).toMatchObject({ id: "archived", path: "/sessions/archived.jsonl", archived: true, archivedAt: "2026-01-02T00:00:00.000Z" }); + + await service.dispose(); + }); + it("sends prompts to an injected runtime without touching the SDK runtime", async () => { const fake = fakeRuntime("prompt-session"); const service = new PiSessionService(new CapturingSessionEventHub(), { diff --git a/src/server/sessions/piSessionService.ts b/src/server/sessions/piSessionService.ts index 73bf1ab..63ae653 100644 --- a/src/server/sessions/piSessionService.ts +++ b/src/server/sessions/piSessionService.ts @@ -15,7 +15,7 @@ import { pageMessagesAtSafeBoundary } from "./messagePaging.js"; import type { SessionEventHub } from "../realtime/sessionEventHub.js"; import { BUILTIN_COMMANDS } from "./builtinCommands.js"; import { SessionCommandService } from "./sessionCommandService.js"; -import { SessionArchiveStore } from "./sessionArchiveStore.js"; +import { SessionArchiveStore, type ArchivedSessionRecord, type ArchiveSessionInput } from "./sessionArchiveStore.js"; import type { ActiveSession } from "./sessionRuntimeStore.js"; import type { AuthChange } from "./authService.js"; import { fallbackSessionName, generateShortSessionName } from "./sessionNameGenerator.js"; @@ -28,7 +28,19 @@ function authLossWarningKey(sessionId: string, provider: string, modelId: string return `${sessionId}:${provider}/${modelId}`; } -type SessionArchiveRepository = Pick; +type SessionArchiveRepository = Pick; +interface PiSessionListEntry { + id: string; + path: string; + cwd: string; + created: Date; + modified: Date; + messageCount: number; + firstMessage: string; + allMessagesText: string; + name?: string; + parentSessionPath?: string; +} type AgentModel = Model; type ModelRegistryInstance = ReturnType; @@ -40,9 +52,9 @@ export interface PiSessionManager { } export interface PiSessionManagerGateway { - list(cwd: string): Promise<{ id: string; path: string; cwd: string; created: Date; modified: Date; messageCount: number; firstMessage: string; allMessagesText: string; name?: string; parentSessionPath?: string }[]>; + list(cwd: string): Promise; create(cwd: string): PiSessionManager; - listAll(): Promise<{ id: string; path: string; cwd: string; created: Date; modified: Date; messageCount: number; firstMessage: string; allMessagesText: string }[]>; + listAll(): Promise; open(path: string): PiSessionManager; } @@ -181,22 +193,19 @@ export class PiSessionService { async list(cwd: string): Promise { const [sessions, archivedRecords] = await Promise.all([this.sessionManager.list(cwd), this.archiveStore.list()]); - const archivedById = new Map(archivedRecords.filter((record) => record.cwd === cwd).map((record) => [record.sessionId, record])); - return sessions.map((s) => { - const archived = archivedById.get(s.id); - return { - id: s.id, - path: s.path, - cwd: s.cwd, - ...(s.name === undefined ? {} : { name: s.name }), - created: s.created.toISOString(), - modified: s.modified.toISOString(), - messageCount: s.messageCount, - firstMessage: s.firstMessage, - ...(s.parentSessionPath === undefined ? {} : { parentSessionPath: s.parentSessionPath }), - ...(archived === undefined ? {} : { archived: true, archivedAt: archived.archivedAt }), - }; - }); + const sessionsById = new Map(sessions.map((session) => [session.id, session])); + const archivedForCwd = await Promise.all( + archivedRecords + .filter((record) => record.cwd === cwd) + .map((record) => this.ensureArchivedSessionMoved(record, sessionsById.get(record.sessionId))), + ); + const archivedById = new Map(archivedForCwd.map((record) => [record.sessionId, record])); + const activeSessions = sessions.filter((session) => !archivedById.has(session.id)).map(clientSessionFromListEntry); + const archivedSessions = archivedForCwd + .sort(compareArchivedRecords) + .map((record) => clientSessionFromArchivedRecord(record, sessionsById.get(record.sessionId))) + .filter(isDefined); + return [...activeSessions, ...archivedSessions]; } async start(cwd: string): Promise { @@ -364,11 +373,13 @@ export class PiSessionService { async archive(sessionId: string): Promise { const session = await this.getOrOpen(sessionId); if (session.isStreaming || session.isCompacting || session.isBashRunning || session.pendingMessageCount > 0) throw new Error("Stop current session activity before archiving"); - await this.archiveStore.archive(sessionId, session.sessionManager.getCwd()); - this.stop(sessionId); + const archiveInput = await this.archiveInputForSession(session); + await this.closeActive(session.sessionId); + await this.archiveStore.archive(archiveInput); } async restore(sessionId: string): Promise { + await this.closeActive(sessionId); await this.archiveStore.restore(sessionId); } @@ -389,14 +400,51 @@ export class PiSessionService { } stop(sessionId: string): void { + void this.closeActive(sessionId).catch(() => { + // Best-effort shutdown; callers that need errors await closeActive directly. + }); + } + + private async ensureArchivedSessionMoved(record: ArchivedSessionRecord, session: PiSessionListEntry | undefined): Promise { + if (session === undefined || this.active.has(record.sessionId)) return record; + try { + return await this.archiveStore.archive(archiveInputFromListEntry(session)); + } catch { + return record; + } + } + + private async archiveInputForSession(session: PiAgentSession): Promise { + const cwd = session.sessionManager.getCwd(); + const sessionFile = session.sessionFile; + if (sessionFile === undefined || sessionFile === "") throw new Error("Session is not persisted"); + const listed = (await this.sessionManager.list(cwd)).find((candidate) => candidate.id === session.sessionId); + if (listed !== undefined) return archiveInputFromListEntry(listed); + return { + sessionId: session.sessionId, + cwd, + path: sessionFile, + created: new Date().toISOString(), + modified: new Date().toISOString(), + messageCount: session.messages.length, + firstMessage: "", + ...(session.sessionName === undefined ? {} : { name: session.sessionName }), + }; + } + + private async closeActive(sessionId: string): Promise { const active = this.active.get(sessionId); if (!active) return; - clearSessionQueue(active.runtime.session); - active.unsubscribe(); - void active.runtime.session.abort().finally(() => active.runtime.dispose()); this.active.delete(sessionId); this.activities.delete(sessionId); this.clearAuthLossWarningsForSession(sessionId); + clearSessionQueue(active.runtime.session); + active.unsubscribe(); + try { + await active.runtime.session.abort(); + } finally { + await active.runtime.dispose(); + } } private async assertWritable(sessionId: string): Promise { @@ -411,6 +459,9 @@ export class PiSessionService { const active = this.active.get(sessionId); if (active) return active; + const archived = await this.archiveStore.get(sessionId); + if (archived?.archivePath !== undefined) return this.create(this.sessionManager.open(archived.archivePath), archived.cwd); + const match = (await this.sessionManager.listAll()).find((s) => s.id === sessionId || s.id.startsWith(sessionId)); if (!match) throw new Error("Session not found"); return this.create(this.sessionManager.open(match.path), match.cwd); @@ -601,6 +652,71 @@ function modelToClientModel(model: PiAgentSession["model"]): ClientSessionModel }; } +function clientSessionFromListEntry(session: PiSessionListEntry): ClientSession { + return { + id: session.id, + path: session.path, + cwd: session.cwd, + ...(session.name === undefined ? {} : { name: session.name }), + created: session.created.toISOString(), + modified: session.modified.toISOString(), + messageCount: session.messageCount, + firstMessage: session.firstMessage, + ...(session.parentSessionPath === undefined ? {} : { parentSessionPath: session.parentSessionPath }), + }; +} + +function archiveInputFromListEntry(session: PiSessionListEntry): ArchiveSessionInput { + return { + sessionId: session.id, + cwd: session.cwd, + path: session.path, + created: session.created.toISOString(), + modified: session.modified.toISOString(), + messageCount: session.messageCount, + firstMessage: session.firstMessage, + ...(session.name === undefined ? {} : { name: session.name }), + ...(session.parentSessionPath === undefined ? {} : { parentSessionPath: session.parentSessionPath }), + }; +} + +function clientSessionFromArchivedRecord(record: ArchivedSessionRecord, fallback: PiSessionListEntry | undefined): ClientSession | undefined { + const path = record.originalPath ?? fallback?.path; + const created = record.created ?? fallback?.created.toISOString(); + const modified = record.modified ?? fallback?.modified.toISOString(); + const messageCount = record.messageCount ?? fallback?.messageCount; + const firstMessage = record.firstMessage ?? fallback?.firstMessage; + if (path === undefined || created === undefined || modified === undefined || messageCount === undefined || firstMessage === undefined) return undefined; + const name = record.name ?? fallback?.name; + const parentSessionPath = record.parentSessionPath ?? fallback?.parentSessionPath; + return { + id: record.sessionId, + path, + cwd: record.cwd, + ...(name === undefined ? {} : { name }), + created, + modified, + messageCount, + firstMessage, + ...(parentSessionPath === undefined ? {} : { parentSessionPath }), + archived: true, + archivedAt: record.archivedAt, + }; +} + +function compareArchivedRecords(a: ArchivedSessionRecord, b: ArchivedSessionRecord): number { + return archivedTimestamp(b) - archivedTimestamp(a); +} + +function archivedTimestamp(record: ArchivedSessionRecord): number { + const time = Date.parse(record.archivedAt); + return Number.isNaN(time) ? 0 : time; +} + +function isDefined(value: T | undefined): value is T { + return value !== undefined; +} + async function clearParentSession(sessionFile: string): Promise { const content = await readFile(sessionFile, "utf8"); const newlineIndex = content.indexOf("\n"); diff --git a/src/server/sessions/sessionArchiveStore.test.ts b/src/server/sessions/sessionArchiveStore.test.ts new file mode 100644 index 0000000..f228b8c --- /dev/null +++ b/src/server/sessions/sessionArchiveStore.test.ts @@ -0,0 +1,56 @@ +import { constants } from "node:fs"; +import { access, mkdtemp, mkdir, readFile, rm, writeFile } from "node:fs/promises"; +import { join } from "node:path"; +import { tmpdir } from "node:os"; +import { afterEach, describe, expect, it } from "vitest"; +import { SessionArchiveStore } from "./sessionArchiveStore.js"; + +const tempRoots: string[] = []; + +describe("SessionArchiveStore", () => { + afterEach(async () => { + await Promise.all(tempRoots.splice(0).map((path) => rm(path, { recursive: true, force: true }))); + }); + + it("moves archived session files out of the active session directory and restores them", async () => { + const root = await mkdtemp(join(tmpdir(), "pi-web-archive-")); + tempRoots.push(root); + const activeDir = join(root, "active"); + await mkdir(activeDir, { recursive: true }); + const sourcePath = join(activeDir, "2026-01-01_s1.jsonl"); + await writeFile(sourcePath, "session contents\n", "utf8"); + + const store = new SessionArchiveStore(join(root, "archived-sessions.json"), join(root, "archived-files")); + const record = await store.archive({ + sessionId: "s1", + cwd: "/workspace", + path: sourcePath, + created: "2026-01-01T00:00:00.000Z", + modified: "2026-01-01T00:01:00.000Z", + messageCount: 2, + firstMessage: "hello", + }); + + expect(await exists(sourcePath)).toBe(false); + expect(record.originalPath).toBe(sourcePath); + expect(record.archivePath).toBeDefined(); + if (record.archivePath === undefined) throw new Error("Expected archive path"); + expect(await readFile(record.archivePath, "utf8")).toBe("session contents\n"); + await expect(store.list()).resolves.toMatchObject([{ sessionId: "s1", originalPath: sourcePath, archivePath: record.archivePath, messageCount: 2 }]); + + await store.restore("s1"); + + expect(await readFile(sourcePath, "utf8")).toBe("session contents\n"); + expect(await exists(record.archivePath)).toBe(false); + await expect(store.list()).resolves.toEqual([]); + }); +}); + +async function exists(path: string): Promise { + try { + await access(path, constants.F_OK); + return true; + } catch { + return false; + } +} diff --git a/src/server/sessions/sessionArchiveStore.ts b/src/server/sessions/sessionArchiveStore.ts index f6c2da4..7c55a50 100644 --- a/src/server/sessions/sessionArchiveStore.ts +++ b/src/server/sessions/sessionArchiveStore.ts @@ -1,11 +1,33 @@ -import { mkdir, readFile, writeFile } from "node:fs/promises"; -import { dirname, join } from "node:path"; +import { randomUUID } from "node:crypto"; +import { constants } from "node:fs"; +import { access, copyFile, mkdir, readFile, rename, unlink, writeFile } from "node:fs/promises"; +import { basename, dirname, join } from "node:path"; import { homedir } from "node:os"; +export interface ArchiveSessionInput { + sessionId: string; + cwd: string; + path: string; + created: string; + modified: string; + messageCount: number; + firstMessage: string; + name?: string; + parentSessionPath?: string; +} + export interface ArchivedSessionRecord { sessionId: string; cwd: string; archivedAt: string; + originalPath?: string; + archivePath?: string; + created?: string; + modified?: string; + messageCount?: number; + firstMessage?: string; + name?: string; + parentSessionPath?: string; } interface ArchiveFile { @@ -13,31 +35,79 @@ interface ArchiveFile { } export class SessionArchiveStore { - constructor(private readonly filePath = join(homedir(), ".pi-web", "archived-sessions.json")) {} + private operationQueue: Promise = Promise.resolve(); + + constructor( + private readonly filePath = join(homedir(), ".pi-web", "archived-sessions.json"), + private readonly archiveDir = join(dirname(filePath), "archived-sessions"), + ) {} async list(): Promise { return (await this.read()).sessions; } - async archive(sessionId: string, cwd: string): Promise { - const data = await this.read(); - const existing = data.sessions.find((session) => session.sessionId === sessionId); - if (existing !== undefined) return existing; - const record = { sessionId, cwd, archivedAt: new Date().toISOString() }; - data.sessions.push(record); - await this.write(data); - return record; + async get(sessionId: string): Promise { + const sessions = (await this.read()).sessions; + return sessions.find((session) => session.sessionId === sessionId) ?? sessions.find((session) => session.sessionId.startsWith(sessionId)); + } + + async archive(session: ArchiveSessionInput): Promise { + return this.exclusive(async () => { + const data = await this.read(); + const existingIndex = data.sessions.findIndex((record) => record.sessionId === session.sessionId); + const existing = existingIndex === -1 ? undefined : data.sessions[existingIndex]; + const archivePath = existing?.archivePath ?? this.archivePathFor(session); + const record = archiveRecordFromInput(session, { + archivedAt: existing?.archivedAt ?? new Date().toISOString(), + originalPath: existing?.originalPath ?? session.path, + archivePath, + }); + + await copySessionFileToArchive(session.path, archivePath); + + if (existingIndex === -1) data.sessions.push(record); + else data.sessions[existingIndex] = record; + await this.write(data); + await removeActiveSessionFile(session.path, archivePath); + return record; + }); } async restore(sessionId: string): Promise { - const data = await this.read(); - const sessions = data.sessions.filter((session) => session.sessionId !== sessionId); - if (sessions.length === data.sessions.length) return; - await this.write({ sessions }); + await this.exclusive(async () => { + const data = await this.read(); + const record = data.sessions.find((session) => session.sessionId === sessionId); + if (record === undefined) return; + + if (record.archivePath !== undefined && record.originalPath !== undefined) { + await restoreSessionFile(record.archivePath, record.originalPath); + } + + const sessions = data.sessions.filter((session) => session.sessionId !== sessionId); + await this.write({ sessions }); + }); } async isArchived(sessionId: string): Promise { - return (await this.list()).some((session) => session.sessionId === sessionId); + return (await this.get(sessionId)) !== undefined; + } + + private archivePathFor(session: ArchiveSessionInput): string { + const sourceName = basename(session.path); + const fileName = sourceName === "" ? `${safeFileName(session.sessionId)}.jsonl` : sourceName; + return join(this.archiveDir, fileName); + } + + private async exclusive(operation: () => Promise): Promise { + const previous = this.operationQueue; + let release = (): void => undefined; + this.operationQueue = new Promise((resolve) => { release = resolve; }); + await previous.catch(() => undefined); + try { + return await operation(); + } finally { + release(); + } } private async read(): Promise { @@ -52,7 +122,69 @@ export class SessionArchiveStore { private async write(data: ArchiveFile): Promise { await mkdir(dirname(this.filePath), { recursive: true }); - await writeFile(this.filePath, `${JSON.stringify(data, null, 2)}\n`, "utf8"); + const tempPath = join(dirname(this.filePath), `.${basename(this.filePath)}.${String(process.pid)}.${Date.now().toString()}.${randomUUID()}.tmp`); + try { + await writeFile(tempPath, `${JSON.stringify(data, null, 2)}\n`, "utf8"); + await rename(tempPath, this.filePath); + } catch (error: unknown) { + await unlink(tempPath).catch(() => undefined); + throw error; + } + } +} + +function archiveRecordFromInput(session: ArchiveSessionInput, archive: { archivedAt: string; originalPath: string; archivePath: string }): ArchivedSessionRecord { + return { + sessionId: session.sessionId, + cwd: session.cwd, + archivedAt: archive.archivedAt, + originalPath: archive.originalPath, + archivePath: archive.archivePath, + created: session.created, + modified: session.modified, + messageCount: session.messageCount, + firstMessage: session.firstMessage, + ...(session.name === undefined ? {} : { name: session.name }), + ...(session.parentSessionPath === undefined ? {} : { parentSessionPath: session.parentSessionPath }), + }; +} + +async function copySessionFileToArchive(source: string, archivePath: string): Promise { + if (source === archivePath) return; + await mkdir(dirname(archivePath), { recursive: true }); + if (await pathExists(archivePath)) return; + await copyFile(source, archivePath); +} + +async function removeActiveSessionFile(source: string, archivePath: string): Promise { + if (source === archivePath) return; + if (await pathExists(source)) await unlink(source); +} + +async function restoreSessionFile(archivePath: string, originalPath: string): Promise { + if (archivePath === originalPath) return; + if (await pathExists(originalPath)) throw new Error(`Cannot restore archived session because a session already exists at ${originalPath}`); + await mkdir(dirname(originalPath), { recursive: true }); + await moveFile(archivePath, originalPath); +} + +async function moveFile(source: string, destination: string): Promise { + try { + await rename(source, destination); + } catch (error: unknown) { + if (!isNodeErrorWithCode(error, "EXDEV")) throw error; + await copyFile(source, destination); + await unlink(source); + } +} + +async function pathExists(path: string): Promise { + try { + await access(path, constants.F_OK); + return true; + } catch (error: unknown) { + if (isNodeErrorWithCode(error, "ENOENT")) return false; + throw error; } } @@ -67,7 +199,45 @@ function parseArchivedSessionRecord(value: unknown): ArchivedSessionRecord { const cwd = value["cwd"]; const archivedAt = value["archivedAt"]; if (typeof sessionId !== "string" || typeof cwd !== "string" || typeof archivedAt !== "string") throw new Error("Invalid archived session record"); - return { sessionId, cwd, archivedAt }; + const originalPath = optionalString(value, "originalPath"); + const archivePath = optionalString(value, "archivePath"); + const created = optionalString(value, "created"); + const modified = optionalString(value, "modified"); + const messageCount = optionalNumber(value, "messageCount"); + const firstMessage = optionalString(value, "firstMessage"); + const name = optionalString(value, "name"); + const parentSessionPath = optionalString(value, "parentSessionPath"); + return { + sessionId, + cwd, + archivedAt, + ...(originalPath === undefined ? {} : { originalPath }), + ...(archivePath === undefined ? {} : { archivePath }), + ...(created === undefined ? {} : { created }), + ...(modified === undefined ? {} : { modified }), + ...(messageCount === undefined ? {} : { messageCount }), + ...(firstMessage === undefined ? {} : { firstMessage }), + ...(name === undefined ? {} : { name }), + ...(parentSessionPath === undefined ? {} : { parentSessionPath }), + }; +} + +function optionalString(record: Record, key: string): string | undefined { + const value = record[key]; + if (value === undefined) return undefined; + if (typeof value !== "string") throw new Error("Invalid archived session record"); + return value; +} + +function optionalNumber(record: Record, key: string): number | undefined { + const value = record[key]; + if (value === undefined) return undefined; + if (typeof value !== "number") throw new Error("Invalid archived session record"); + return value; +} + +function safeFileName(value: string): string { + return value.replace(/[^a-zA-Z0-9._-]/g, "_") || "session"; } function isRecord(value: unknown): value is Record {