import { Database } from "bun:sqlite"; import * as fs from "node:fs/promises"; import * as path from "node:path"; import { gunzipSync, gzipSync } from "node:zlib"; import { getAgentDir, getBlobsDir, getHistoryDbPath, getModelDbPath, getSessionsDir } from "@oh-my-pi/pi-utils"; import { Settings } from "../config/settings"; import { listSessionsReadOnly, type SessionInfo, type SessionStatus } from "../session/session-listing"; import { FileSessionStorage } from "../session/session-storage"; const HASH_RE = /^[a-f0-9]{64}$/; const BLOB_FILE_RE = /^([a-f0-9]{64})(?:\.[A-Za-z0-9][A-Za-z0-9._-]{0,31})?$/; const BLOB_REF_RE = /\bblob:sha256:([a-f0-9]{64})\b/gi; const JSONL_GLOB = new Bun.Glob("**/*.jsonl"); const JSONL_GZ_GLOB = new Bun.Glob("**/*.jsonl.gz"); const ACTIVE_STATUSES: ReadonlySet = new Set(["pending", "interrupted", "unknown"]); const DAY_MS = 86_400_000; const GC_WRITE_GRACE_MS = 5 * 60_000; const SESSION_SUFFIX = ".jsonl"; const COMPRESSED_SESSION_SUFFIX = ".jsonl.gz"; export interface GcCommandFlags { apply?: boolean; json?: boolean; agentDir?: string; blobs?: boolean; archive?: boolean; wal?: boolean; coldArchiveAfterDays?: number; retainNewestGlobal?: number; retainNewestPerCwd?: number; } export interface GcCommandArgs { flags: GcCommandFlags; } export interface BlobGcResult { referenced: number; candidates: number; wouldDelete: number; deleted: number; bytes: number; errors: string[]; } export interface ArchiveGcResult { scanned: number; skippedActive: number; keptNewestGlobal: number; keptNewestPerCwd: number; wouldArchive: number; archived: number; historyRowsDeleted: number; ftsRebuilt: boolean; errors: string[]; } export interface WalCheckpointResult { dbPath: string; walBytes: number; wouldCheckpoint: boolean; checkpointed: boolean; busy: number; log: number; checkpointedFrames: number; } export interface WalGcResult { databases: WalCheckpointResult[]; walBytes: number; wouldCheckpoint: boolean; checkpointed: boolean; } export interface GcResult { agentDir: string; apply: boolean; blobs?: BlobGcResult; archive?: ArchiveGcResult; wal?: WalGcResult; lockPath: string; } interface BlobCandidate { hash: string; paths: string[]; bytes: number; mtimeMs: number; } interface ArchiveCandidate { session: SessionInfo; relativePath: string; destinationPath: string; } interface ResolvedGcOptions { apply: boolean; json: boolean; agentDir: string; runBlobs: boolean; runArchive: boolean; runWal: boolean; coldArchiveAfterDays: number; retainNewestGlobal: number; retainNewestPerCwd: number; } interface SqliteRunResult { changes?: number | bigint; } interface WalCheckpointRow { busy?: number | bigint | null; log?: number | bigint | null; checkpointed?: number | bigint | null; } function numberSetting(value: number | undefined, fallback: number): number { if (value === undefined || !Number.isFinite(value)) return fallback; return Math.max(0, Math.floor(value)); } async function resolveOptions(flags: GcCommandFlags): Promise { const agentDir = path.resolve(flags.agentDir ?? getAgentDir()); const settings = await Settings.init({ agentDir }); const selected = flags.blobs === true || flags.archive === true || flags.wal === true; return { apply: flags.apply === true, json: flags.json === true, agentDir, runBlobs: selected ? flags.blobs === true : settings.get("gc.blobs"), runArchive: selected ? flags.archive === true : settings.get("gc.archive"), runWal: selected ? flags.wal === true : settings.get("gc.wal"), coldArchiveAfterDays: numberSetting(flags.coldArchiveAfterDays, settings.get("gc.coldArchiveAfterDays")), retainNewestGlobal: numberSetting(flags.retainNewestGlobal, settings.get("gc.retainNewestGlobal")), retainNewestPerCwd: numberSetting(flags.retainNewestPerCwd, settings.get("gc.retainNewestPerCwd")), }; } function getArchivedSessionsDir(agentDir: string): string { return path.join(path.dirname(getSessionsDir(agentDir)), "archive", "sessions"); } function errorMessage(error: unknown): string { return error instanceof Error ? error.message : String(error); } function codeOf(error: unknown): string | undefined { return typeof error === "object" && error !== null && "code" in error ? String((error as { code?: unknown }).code) : undefined; } async function pathExists(target: string): Promise { try { await fs.stat(target); return true; } catch (error) { if (codeOf(error) === "ENOENT") return false; throw error; } } async function statIfPresent(target: string) { try { return await fs.stat(target); } catch (error) { if (codeOf(error) === "ENOENT") return null; throw error; } } async function readTextIfPresent(file: string): Promise { try { if (file.endsWith(COMPRESSED_SESSION_SUFFIX)) { return new TextDecoder().decode(gunzipSync(await Bun.file(file).bytes())); } return await Bun.file(file).text(); } catch (error) { if (codeOf(error) === "ENOENT") return ""; throw error; } } async function collectJsonlFiles(root: string): Promise { try { const files = await Array.fromAsync(JSONL_GLOB.scan(root), name => path.join(root, name)); files.sort(); return files; } catch (error) { if (codeOf(error) === "ENOENT") return []; throw error; } } async function collectCompressedJsonlFiles(root: string): Promise { try { const files = await Array.fromAsync(JSONL_GZ_GLOB.scan(root), name => path.join(root, name)); files.sort(); return files; } catch (error) { if (codeOf(error) === "ENOENT") return []; throw error; } } async function collectReferencedBlobHashes(sessionRoots: string[]): Promise> { const hashes = new Set(); for (const root of sessionRoots) { const files = [...(await collectJsonlFiles(root)), ...(await collectCompressedJsonlFiles(root))]; for (const file of files) { const text = await readTextIfPresent(file); for (const match of text.matchAll(BLOB_REF_RE)) { const hash = match[1]?.toLowerCase(); if (hash && HASH_RE.test(hash)) hashes.add(hash); } } } return hashes; } async function collectBlobCandidates(blobDir: string): Promise { let entries: string[]; try { entries = await fs.readdir(blobDir); } catch (error) { if (codeOf(error) === "ENOENT") return []; throw error; } const byHash = new Map(); for (const entry of entries) { const match = entry.match(BLOB_FILE_RE); const hash = match?.[1]; if (!hash) continue; const file = path.join(blobDir, entry); const stat = await statIfPresent(file); if (!stat) continue; if (!stat.isFile()) continue; const candidate = byHash.get(hash) ?? { hash, paths: [], bytes: 0, mtimeMs: stat.mtimeMs }; candidate.paths.push(file); candidate.bytes += stat.size; candidate.mtimeMs = Math.max(candidate.mtimeMs, stat.mtimeMs); byHash.set(hash, candidate); } return [...byHash.values()].sort((a, b) => a.hash.localeCompare(b.hash)); } async function runBlobGc(options: ResolvedGcOptions, archiveSessionsRoot: string): Promise { const blobDir = getBlobsDir(options.agentDir); const sessionsRoot = getSessionsDir(options.agentDir); const referenced = await collectReferencedBlobHashes([sessionsRoot, archiveSessionsRoot]); const candidates = await collectBlobCandidates(blobDir); const result: BlobGcResult = { referenced: referenced.size, candidates: candidates.length, wouldDelete: 0, deleted: 0, bytes: 0, errors: [], }; const deleteBeforeMs = Date.now() - GC_WRITE_GRACE_MS; for (const candidate of candidates) { if (referenced.has(candidate.hash)) continue; if (candidate.mtimeMs > deleteBeforeMs) continue; result.wouldDelete += candidate.paths.length; result.bytes += candidate.bytes; if (!options.apply) continue; for (const file of candidate.paths) { try { await fs.unlink(file); result.deleted += 1; } catch (error) { if (codeOf(error) === "ENOENT") continue; result.errors.push(`${file}: ${errorMessage(error)}`); } } } return result; } async function listActiveSessions(sessionsRoot: string): Promise { let entries: Array<{ name: string; isDirectory(): boolean }>; try { entries = await fs.readdir(sessionsRoot, { withFileTypes: true }); } catch (error) { if (codeOf(error) === "ENOENT") return []; throw error; } const storage = new FileSessionStorage(); const sessions: SessionInfo[] = []; for (const entry of entries) { if (!entry.isDirectory()) continue; sessions.push(...(await listSessionsReadOnly(path.join(sessionsRoot, entry.name), storage))); } sessions.sort((a, b) => b.modified.getTime() - a.modified.getTime()); return sessions; } function archiveDestination( archiveRoot: string, sessionsRoot: string, session: SessionInfo, ): Omit | null { const sessionPath = session.path; const relativePath = path.relative(sessionsRoot, sessionPath); if (relativePath.startsWith("..") || path.isAbsolute(relativePath)) return null; if (!relativePath.endsWith(SESSION_SUFFIX)) return null; return { relativePath, destinationPath: path.join(archiveRoot, `${relativePath}.gz`), }; } function sessionCwdKey(sessionsRoot: string, session: SessionInfo): string { const relativePath = path.relative(sessionsRoot, session.path); if (relativePath.startsWith("..") || path.isAbsolute(relativePath)) return session.cwd || "."; const dirname = path.dirname(relativePath); return dirname === "." ? session.cwd || "." : dirname; } async function movePath(source: string, destination: string): Promise { await fs.mkdir(path.dirname(destination), { recursive: true }); try { await fs.rename(source, destination); return; } catch (error) { if (codeOf(error) !== "EXDEV") throw error; } const stat = await fs.stat(source); if (stat.isDirectory()) { await fs.cp(source, destination, { recursive: true }); await fs.rm(source, { recursive: true, force: true }); return; } await fs.copyFile(source, destination); await fs.unlink(source); } function sessionArtifactsPath(sessionPath: string): string { if (sessionPath.endsWith(COMPRESSED_SESSION_SUFFIX)) { return sessionPath.slice(0, -COMPRESSED_SESSION_SUFFIX.length); } return sessionPath.slice(0, -SESSION_SUFFIX.length); } async function gzipSessionFile(source: string, destination: string): Promise { await fs.mkdir(path.dirname(destination), { recursive: true }); const tempPath = `${destination}.${process.pid}.${Date.now()}.tmp`; let renamed = false; try { const compressed = gzipSync(await Bun.file(source).bytes(), { level: 9 }); await Bun.write(tempPath, compressed); await fs.rename(tempPath, destination); renamed = true; await fs.unlink(source); } catch (error) { await fs.rm(tempPath, { force: true }); if (renamed) await fs.rm(destination, { force: true }); throw error; } } async function restoreGzipSessionFile(source: string, destination: string): Promise { await fs.mkdir(path.dirname(destination), { recursive: true }); const decompressed = gunzipSync(await Bun.file(source).bytes()); await Bun.write(destination, decompressed); await fs.unlink(source); } async function moveSessionWithArtifacts(candidate: ArchiveCandidate): Promise { const sourceSession = candidate.session.path; const destSession = candidate.destinationPath; const legacyDestSession = destSession.endsWith(".gz") ? destSession.slice(0, -".gz".length) : `${destSession}.gz`; const sourceArtifacts = sessionArtifactsPath(sourceSession); const destArtifacts = sessionArtifactsPath(destSession); if (await pathExists(destSession)) throw new Error(`archive destination exists: ${destSession}`); if (await pathExists(legacyDestSession)) throw new Error(`archive destination exists: ${legacyDestSession}`); if ((await pathExists(sourceArtifacts)) && (await pathExists(destArtifacts))) { throw new Error(`archive artifacts destination exists: ${destArtifacts}`); } const moved: Array<{ source: string; destination: string; compressed?: boolean }> = []; try { await gzipSessionFile(sourceSession, destSession); moved.push({ source: sourceSession, destination: destSession, compressed: true }); if (await pathExists(sourceArtifacts)) { await movePath(sourceArtifacts, destArtifacts); moved.push({ source: sourceArtifacts, destination: destArtifacts }); } } catch (error) { for (const move of moved.reverse()) { try { if (move.compressed) { await restoreGzipSessionFile(move.destination, move.source); } else { await movePath(move.destination, move.source); } } catch { // Preserve the original failure; rollback failure is reported by the next scan. } } throw error; } } function sqliteNumber(value: number | bigint | null | undefined): number { if (typeof value === "bigint") return Number(value); if (typeof value === "number") return value; return 0; } function tableExists(db: Database, table: string): boolean { const row = db .prepare("SELECT 1 AS present FROM sqlite_master WHERE type IN ('table','view') AND name = ?") .get(table) as { present?: number } | null; return row?.present === 1; } function historyHasSessionId(db: Database): boolean { const rows = db.prepare("PRAGMA table_info(history)").all() as Array<{ name?: string | null }>; return rows.some(row => row.name === "session_id"); } function deleteHistoryRowsForSessions(dbPath: string, sessionIds: string[]): { deleted: number; ftsRebuilt: boolean } { if (sessionIds.length === 0) return { deleted: 0, ftsRebuilt: false }; const db = new Database(dbPath); try { db.run("PRAGMA busy_timeout = 5000"); if (!tableExists(db, "history")) return { deleted: 0, ftsRebuilt: false }; if (!historyHasSessionId(db)) return { deleted: 0, ftsRebuilt: false }; const hasFts = tableExists(db, "history_fts"); const deleteStmt = db.prepare("DELETE FROM history WHERE session_id = ?"); let deleted = 0; const tx = db.transaction((ids: string[]) => { for (const id of ids) { const result = deleteStmt.run(id) as SqliteRunResult; deleted += sqliteNumber(result.changes); } if (deleted > 0 && hasFts) db.run("INSERT INTO history_fts(history_fts) VALUES('rebuild')"); }); tx(sessionIds); return { deleted, ftsRebuilt: deleted > 0 && hasFts }; } finally { db.close(); } } async function runArchiveGc(options: ResolvedGcOptions, archiveRoot: string): Promise { const sessionsRoot = getSessionsDir(options.agentDir); const sessions = await listActiveSessions(sessionsRoot); const cutoffMs = Date.now() - options.coldArchiveAfterDays * DAY_MS; const result: ArchiveGcResult = { scanned: sessions.length, skippedActive: 0, keptNewestGlobal: 0, keptNewestPerCwd: 0, wouldArchive: 0, archived: 0, historyRowsDeleted: 0, ftsRebuilt: false, errors: [], }; const candidates: ArchiveCandidate[] = []; let inactiveSeen = 0; const inactiveSeenByCwd = new Map(); const archiveBeforeMs = Date.now() - GC_WRITE_GRACE_MS; for (const session of sessions) { if (session.status && ACTIVE_STATUSES.has(session.status)) { result.skippedActive += 1; continue; } if (session.modified.getTime() > archiveBeforeMs) { result.skippedActive += 1; continue; } const cwdKey = sessionCwdKey(sessionsRoot, session); const cwdSeen = inactiveSeenByCwd.get(cwdKey) ?? 0; const keepGlobal = inactiveSeen < options.retainNewestGlobal; const keepPerCwd = cwdSeen < options.retainNewestPerCwd; inactiveSeen += 1; inactiveSeenByCwd.set(cwdKey, cwdSeen + 1); if (keepGlobal) { result.keptNewestGlobal += 1; continue; } if (keepPerCwd) { result.keptNewestPerCwd += 1; continue; } if (options.coldArchiveAfterDays > 0 && session.modified.getTime() > cutoffMs) continue; const destination = archiveDestination(archiveRoot, sessionsRoot, session); if (!destination) continue; candidates.push({ ...destination, session }); } result.wouldArchive = candidates.length; if (!options.apply) return result; const archivedSessionIds: string[] = []; for (const candidate of candidates) { try { await moveSessionWithArtifacts(candidate); result.archived += 1; archivedSessionIds.push(candidate.session.id); } catch (error) { result.errors.push(`${candidate.session.path}: ${errorMessage(error)}`); } } const dbPath = getHistoryDbPath(options.agentDir); if (archivedSessionIds.length > 0 && (await pathExists(dbPath))) { const cleanup = deleteHistoryRowsForSessions(dbPath, archivedSessionIds); result.historyRowsDeleted = cleanup.deleted; result.ftsRebuilt = cleanup.ftsRebuilt; } return result; } async function checkpointWal(dbPath: string, apply: boolean): Promise { const walPath = `${dbPath}-wal`; let walBytes = 0; try { walBytes = (await fs.stat(walPath)).size; } catch (error) { if (codeOf(error) !== "ENOENT") throw error; } const result: WalCheckpointResult = { dbPath, walBytes, wouldCheckpoint: walBytes > 0, checkpointed: false, busy: 0, log: 0, checkpointedFrames: 0, }; if (!apply || !(await pathExists(dbPath))) return result; const db = new Database(dbPath); try { db.run("PRAGMA busy_timeout = 5000"); const row = db.prepare("PRAGMA wal_checkpoint(TRUNCATE)").get() as WalCheckpointRow | null; result.checkpointed = true; result.busy = sqliteNumber(row?.busy); result.log = sqliteNumber(row?.log); result.checkpointedFrames = sqliteNumber(row?.checkpointed); } finally { db.close(); } try { result.walBytes = (await fs.stat(walPath)).size; } catch (error) { if (codeOf(error) !== "ENOENT") throw error; result.walBytes = 0; } return result; } async function runWalGc(options: ResolvedGcOptions): Promise { const databases = await Promise.all( [getHistoryDbPath(options.agentDir), getModelDbPath(options.agentDir)].map(dbPath => checkpointWal(dbPath, options.apply), ), ); return { databases, walBytes: databases.reduce((total, db) => total + db.walBytes, 0), wouldCheckpoint: databases.some(db => db.wouldCheckpoint), checkpointed: databases.some(db => db.checkpointed), }; } async function withGcLock(agentDir: string, fn: (lockPath: string) => Promise): Promise { const lockPath = path.join(agentDir, "gc.lock"); await fs.mkdir(agentDir, { recursive: true }); let handle: fs.FileHandle; try { handle = await fs.open(lockPath, "wx"); } catch (error) { if (codeOf(error) === "EEXIST") throw new Error(`GC already running: ${lockPath}`); throw error; } let result: T | undefined; let runError: unknown; try { await handle.writeFile(`${process.pid}\n${new Date().toISOString()}\n`); result = await fn(lockPath); } catch (error) { runError = error; } let closeError: unknown; try { await handle.close(); } catch (error) { closeError = error; } let unlinkError: unknown; try { await fs.unlink(lockPath); } catch (error) { if (codeOf(error) !== "ENOENT") unlinkError = error; } if (runError) throw runError; if (closeError) throw closeError; if (unlinkError) throw unlinkError; return result as T; } function formatBytes(bytes: number): string { if (bytes < 1024) return `${bytes} B`; if (bytes < 1024 * 1024) return `${(bytes / 1024).toFixed(1)} KiB`; if (bytes < 1024 * 1024 * 1024) return `${(bytes / (1024 * 1024)).toFixed(1)} MiB`; return `${(bytes / (1024 * 1024 * 1024)).toFixed(1)} GiB`; } function renderText(result: GcResult): string { const lines = [`GC ${result.apply ? "applied" : "dry-run"} (${result.agentDir})`]; if (result.blobs) { lines.push( `blobs: ${result.blobs.deleted}/${result.blobs.wouldDelete} files, ${formatBytes(result.blobs.bytes)}, ${result.blobs.referenced} refs`, ); if (result.blobs.errors.length > 0) lines.push(`blob errors: ${result.blobs.errors.length}`); } if (result.archive) { lines.push( `sessions: ${result.archive.archived}/${result.archive.wouldArchive} archived, ${result.archive.historyRowsDeleted} history rows removed`, ); if (result.archive.skippedActive > 0) lines.push(`sessions skipped active: ${result.archive.skippedActive}`); if (result.archive.errors.length > 0) lines.push(`session errors: ${result.archive.errors.length}`); } if (result.wal) { const state = result.wal.checkpointed ? "checkpointed" : "checkpoint dry-run"; lines.push(`wal: ${state}, ${formatBytes(result.wal.walBytes)} across ${result.wal.databases.length} dbs`); } return `${lines.join("\n")}\n`; } export async function runGcCommand(args: GcCommandArgs): Promise { const options = await resolveOptions(args.flags); const archiveRoot = getArchivedSessionsDir(options.agentDir); const result = await withGcLock(options.agentDir, async lockPath => { const next: GcResult = { agentDir: options.agentDir, apply: options.apply, lockPath }; if (options.runBlobs) next.blobs = await runBlobGc(options, archiveRoot); if (options.runArchive) next.archive = await runArchiveGc(options, archiveRoot); if (options.runWal) next.wal = await runWalGc(options); return next; }); const output = options.json ? `${JSON.stringify(result, null, 2)}\n` : renderText(result); process.stdout.write(output); return result; }