Something went wrong. Try again.
fork of prime agent with changes i want but are not upstreamable probqbly (and some fixes)
Something went wrong. Try again.
15 kB · 443 lines
TypeScript
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444import { existsSync, rmSync } from "node:fs";import { createRequire } from "node:module";import { dirname, join, resolve } from "node:path";import type { DatabaseSync, StatementSync } from "node:sqlite";import { realpathIfPresentSync } from "../utils/atomic-file.js";import { appendSessionEntryStructure, packSessionEntryStructure, type SessionEntryStructure,} from "./session-entry-structure.js";
/** * Persistent index of session-file summaries. The jsonl transcripts stay the * source of truth: every row is keyed by absolute path and validated against * the file identity (dev, ino, size, mtime) before it is trusted, so a stale or * corrupt database can only cost a rescan, never a wrong listing. * * `node:sqlite` is unavailable on some runtimes (bun) and node versions, and any * store failure is swallowed: the caller then behaves exactly like a cold scan. */
export const SESSION_CATALOG_DB_FILENAME = "session-catalog.db";// 3 adds the packed entry structure (byte ranges and links; payloads stay in the file).const SCHEMA_VERSION = 3;const BUSY_TIMEOUT_MS = 5_000;const ENV_SESSION_CATALOG_DB = "PRIME_AGENT_SESSION_CATALOG_DB";const TRANSIENT_SQLITE_ERROR = /busy|locked|interrupted|disk i\/o|no space|out of memory/i;const CORRUPT_SQLITE_ERROR = /not a database|malformed|encrypted|disk image/i;
export type SessionCatalogMode = "off" | "child" | "in-process";
/** * `off` disables the store entirely, `child` keeps the daemon catalog child * process for listings, `in-process` (default) answers listings in the daemon. */export function sessionCatalogMode(environment: NodeJS.ProcessEnv = process.env): SessionCatalogMode { const raw = environment[ENV_SESSION_CATALOG_DB]?.trim().toLowerCase(); if (raw === "0" || raw === "off" || raw === "false" || raw === "disabled") return "off"; if (raw === "child") return "child"; return "in-process";}
export function isSessionCatalogStoreEnabled(environment: NodeJS.ProcessEnv = process.env): boolean { return sessionCatalogMode(environment) !== "off";}
export interface SessionCatalogIdentity { dev: number; ino: number; size: number; mtimeMs: number;}
export interface SessionCatalogRow extends SessionCatalogIdentity { /** Absolute path of the session file; the row key. */ path: string; /** Absolute directory of the file, for pruning and directory-scoped queries. */ dir: string; id: string | null; modifiedMs: number; /** Serialized summary; null records a file that cannot be scanned at this identity. */ summaryJson: string | null; /** Serialized scan accumulator for an incremental resume; null when too large. */ resumeJson: string | null; indexedMs: number;}
/** Packed structure of a transcript: where each entry is and how entries link. */export interface SessionCatalogEntryStructure { /** Records covering transcript bytes `[0, throughOffset)`; null when a session's ids cannot be packed. */ blob: Buffer | null; /** Transcript byte offset the blob covers, matching the scan offset of the summary. */ throughOffset: number; /** Number of entries in the blob. */ count: number;}
export interface SessionCatalogStore { read(path: string): SessionCatalogRow | undefined; write(row: SessionCatalogRow): void; remove(path: string): void; /** Drop rows whose file is gone: rows in `directory` that are not in `present`. */ retain(directory: string, present: ReadonlySet<string>): void; /** * Attach the structure collected by a scan pass. Records replace the stored * blob when the pass started at 0 and append to it when the pass resumed at * the stored offset; any other offset means the row moved under us, so the * structure is dropped rather than mixed. */ writeEntryStructure( path: string, fromOffset: number, throughOffset: number, records: readonly SessionEntryStructure[], ): void; /** Mark a session's structure unusable, so readers parse the file instead of trusting a partial index. */ clearEntryStructure(path: string): void; readEntryStructure(path: string): SessionCatalogEntryStructure | undefined; close(): void;}
type DatabaseConstructor = new (path: string) => DatabaseSync;
const requireBuiltin = createRequire(import.meta.url);// node:sqlite is absent on bun and on node builds without it; bun:sqlite exposes// the same constructor surface under a different name.const SQLITE_MODULES = ["node:sqlite", "bun:sqlite"] as const;let sqliteCtor: DatabaseConstructor | undefined | null;
function resolveDatabaseConstructor(module: unknown): DatabaseConstructor | undefined { if (!module || typeof module !== "object") return undefined; const candidate = module as { DatabaseSync?: unknown; Database?: unknown }; const ctor = typeof candidate.DatabaseSync === "function" ? candidate.DatabaseSync : candidate.Database; return typeof ctor === "function" ? (ctor as DatabaseConstructor) : undefined;}
function loadSqlite(): DatabaseConstructor | undefined { if (sqliteCtor !== undefined) return sqliteCtor ?? undefined; for (const moduleId of SQLITE_MODULES) { try { const resolved = resolveDatabaseConstructor(requireBuiltin(moduleId)); if (resolved) { sqliteCtor = resolved; return resolved; } } catch { // Try the next runtime's sqlite module. } } sqliteCtor = null; return undefined;}
/** * Rows are keyed by canonical path: a session directory reached through a * symlink alias (macOS /tmp, mounted homes) must not index a second copy. */const canonicalDirCache = new Map<string, string>();
export function canonicalSessionCatalogDir(directory: string): string { const key = resolve(directory); const cached = canonicalDirCache.get(key); if (cached !== undefined) return cached; const canonical = realpathIfPresentSync(key) ?? key; canonicalDirCache.set(key, canonical); return canonical;}
function canonicalSessionCatalogPath(filePath: string): string { const resolved = resolve(filePath); const dir = dirname(resolved); const canonical = canonicalSessionCatalogDir(dir); return canonical === dir ? resolved : join(canonical, resolved.slice(dir.length + 1));}
function asNumber(value: unknown): number { if (typeof value === "number") return value; if (typeof value === "bigint") return Number(value); return 0;}
function asNullableString(value: unknown): string | null { return typeof value === "string" ? value : null;}
class SqliteSessionCatalogStore implements SessionCatalogStore { private failed = false; private readonly readStatement: StatementSync; private readonly writeStatement: StatementSync; private readonly removeStatement: StatementSync; private readonly dirStatement: StatementSync; private readonly entryStructureWriteStatement: StatementSync; private readonly entryStructureReadStatement: StatementSync;
constructor(private readonly db: DatabaseSync) { try { this.readStatement = db.prepare( "SELECT path, dev, ino, size, mtime_ms, id, modified_ms, summary_json, resume_json, indexed_ms FROM session_catalog WHERE path = ?", ); this.entryStructureReadStatement = db.prepare( "SELECT entries_blob, entries_count, entries_through_offset FROM session_catalog WHERE path = ?", ); this.entryStructureWriteStatement = db.prepare( "UPDATE session_catalog SET entries_blob = ?, entries_count = ?, entries_through_offset = ? WHERE path = ?", ); this.writeStatement = db.prepare( `INSERT INTO session_catalog (path, dir, dev, ino, size, mtime_ms, id, modified_ms, summary_json, resume_json, indexed_ms) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(path) DO UPDATE SET dir = excluded.dir, dev = excluded.dev, ino = excluded.ino, size = excluded.size, mtime_ms = excluded.mtime_ms, id = excluded.id, modified_ms = excluded.modified_ms, summary_json = excluded.summary_json, resume_json = excluded.resume_json, indexed_ms = excluded.indexed_ms`, ); this.removeStatement = db.prepare("DELETE FROM session_catalog WHERE path = ?"); this.dirStatement = db.prepare("SELECT path FROM session_catalog WHERE dir = ?"); } catch (error) { this.fail(); throw error; } }
read(path: string): SessionCatalogRow | undefined { if (this.failed) return undefined; try { const key = canonicalSessionCatalogPath(path); const row = this.readStatement.get(key); if (!row) return undefined; return { path: key, dir: canonicalSessionCatalogDir(dirname(key)), dev: asNumber(row.dev), ino: asNumber(row.ino), size: asNumber(row.size), mtimeMs: asNumber(row.mtime_ms), id: asNullableString(row.id), modifiedMs: asNumber(row.modified_ms), summaryJson: asNullableString(row.summary_json), resumeJson: asNullableString(row.resume_json), indexedMs: asNumber(row.indexed_ms), }; } catch (error) { this.handleError(error); return undefined; } }
write(row: SessionCatalogRow): void { if (this.failed) return; try { this.writeStatement.run( canonicalSessionCatalogPath(row.path), canonicalSessionCatalogDir(row.dir), row.dev, row.ino, row.size, row.mtimeMs, row.id, row.modifiedMs, row.summaryJson, row.resumeJson, row.indexedMs, ); } catch (error) { this.handleError(error); } }
remove(path: string): void { if (this.failed) return; try { this.removeStatement.run(canonicalSessionCatalogPath(path)); } catch (error) { this.handleError(error); } }
retain(directory: string, present: ReadonlySet<string>): void { if (this.failed) return; const stale: string[] = []; try { const keep = new Set([...present].map(canonicalSessionCatalogPath)); for (const row of this.dirStatement.all(canonicalSessionCatalogDir(directory))) { const path = asNullableString(row.path); if (path && !keep.has(path)) stale.push(path); } } catch (error) { this.handleError(error); return; } if (stale.length === 0) return; // One transaction: a mass deletion must not cost one commit per file. try { this.db.exec("BEGIN IMMEDIATE"); try { for (const path of stale) this.removeStatement.run(path); this.db.exec("COMMIT"); } catch (error) { this.db.exec("ROLLBACK"); throw error; } } catch (error) { this.handleError(error); } }
writeEntryStructure( path: string, fromOffset: number, throughOffset: number, records: readonly SessionEntryStructure[], ): void { if (this.failed || records.length === 0) return; const key = canonicalSessionCatalogPath(path); try { const stored = this.readEntryStructure(key); const resumed = fromOffset > 0 && stored?.blob && stored.throughOffset === fromOffset; const blob = resumed ? (appendSessionEntryStructure(stored.blob ?? undefined, fromOffset, records) ?? null) : (packSessionEntryStructure(records) ?? null); // A structure that cannot be represented, or a row that moved between the // scan and this write, is dropped instead of mixed: the next reader parses // the file rather than trusting a partial index. this.entryStructureWriteStatement.run( blob, blob ? (resumed ? (stored?.count ?? 0) + records.length : records.length) : 0, blob ? throughOffset : 0, key, ); } catch (error) { this.handleError(error); } }
clearEntryStructure(path: string): void { if (this.failed) return; try { this.entryStructureWriteStatement.run(null, 0, 0, canonicalSessionCatalogPath(path)); } catch (error) { this.handleError(error); } }
readEntryStructure(path: string): SessionCatalogEntryStructure | undefined { if (this.failed) return undefined; try { const row = this.entryStructureReadStatement.get(canonicalSessionCatalogPath(path)); if (!row) return undefined; return { blob: row.entries_blob instanceof Uint8Array ? Buffer.from(row.entries_blob) : null, throughOffset: asNumber(row.entries_through_offset), count: asNumber(row.entries_count), }; } catch (error) { this.handleError(error); return undefined; } }
close(): void { this.failed = true; try { this.db.close(); } catch { // A close failure leaves nothing to salvage: the file is only a cache. } }
/** A fatal store error degrades to scanning for the rest of the run. */ private fail(): void { if (this.failed) return; this.failed = true; try { this.db.close(); } catch { // Ignore: the connection is already unusable. } }
/** * Lock contention, a full disk, or an interrupted statement are temporary: * skip this operation but keep the connection so the catalog recovers. */ private handleError(error: unknown): void { const message = error instanceof Error ? error.message : String(error); if (TRANSIENT_SQLITE_ERROR.test(message)) return; this.fail(); }}
function applyPragmas(db: DatabaseSync): void { db.exec("PRAGMA journal_mode = WAL"); db.exec("PRAGMA synchronous = NORMAL"); db.exec(`PRAGMA busy_timeout = ${BUSY_TIMEOUT_MS}`);}
function createSchema(db: DatabaseSync): void { db.exec(`CREATE TABLE IF NOT EXISTS session_catalog ( path TEXT PRIMARY KEY, dir TEXT NOT NULL, dev INTEGER NOT NULL, ino INTEGER NOT NULL, size INTEGER NOT NULL, mtime_ms REAL NOT NULL, id TEXT, modified_ms REAL NOT NULL, summary_json TEXT, resume_json TEXT, entries_blob BLOB, entries_count INTEGER NOT NULL DEFAULT 0, entries_through_offset INTEGER NOT NULL DEFAULT 0, indexed_ms REAL NOT NULL ) STRICT`); db.exec("CREATE INDEX IF NOT EXISTS session_catalog_dir ON session_catalog (dir)"); db.exec("CREATE INDEX IF NOT EXISTS session_catalog_modified ON session_catalog (modified_ms DESC)"); db.exec(`PRAGMA user_version = ${SCHEMA_VERSION}`);}
function readSchemaVersion(db: DatabaseSync): number { const row = db.prepare("PRAGMA user_version").get(); return asNumber(row?.user_version);}
function openDatabase(directory: string, databaseCtor: DatabaseConstructor, recreate: boolean): DatabaseSync { const dbPath = join(directory, SESSION_CATALOG_DB_FILENAME); if (recreate) { for (const suffix of ["", "-wal", "-shm"]) { const path = `${dbPath}${suffix}`; if (existsSync(path)) rmSync(path, { force: true }); } } const db = new databaseCtor(dbPath); try { applyPragmas(db); if (readSchemaVersion(db) !== SCHEMA_VERSION) { db.exec("DROP TABLE IF EXISTS session_catalog"); db.exec("DROP TABLE IF EXISTS session_entries"); } createSchema(db); } catch (error) { db.close(); throw error; } return db;}
/** Opens the store for a session directory, or returns undefined when unavailable. */export function openSessionCatalogStore(directory: string): SessionCatalogStore | undefined { const databaseCtor = loadSqlite(); if (!databaseCtor) return undefined; const dir = resolve(directory); try { return new SqliteSessionCatalogStore(openDatabase(dir, databaseCtor, false)); } catch (error) { // Only a verified corrupt file may be replaced: a lock, a read-only // directory, or a missing path must never delete another process's db. const message = error instanceof Error ? error.message : String(error); if (!CORRUPT_SQLITE_ERROR.test(message)) return undefined; try { return new SqliteSessionCatalogStore(openDatabase(dir, databaseCtor, true)); } catch { return undefined; } }}