mirror of
https://github.com/badlogic/pi-mono.git
synced 2026-11-06 23:44:36 +00:00
feat(agent): establish durable harness type contracts
This commit is contained in:
Mario Zechner
parent
40a3d8556a
commit
5aeb06bb11
78 changed files with 1211 additions and 12174 deletions
|
|
@ -1,101 +0,0 @@
|
|||
import { SessionError } from "@earendil-works/pi-agent-core";
|
||||
import { uuidv7 } from "@earendil-works/pi-ai";
|
||||
import { sql } from "./sql.ts";
|
||||
import {
|
||||
copyBranchEntriesThroughSeq,
|
||||
deleteBranchEntries,
|
||||
insertBranchEntriesForPath,
|
||||
insertBranchEntry,
|
||||
readBranchContainingEntry,
|
||||
} from "./storage/branch-entries.ts";
|
||||
import { deleteBranchTips, insertBranchTip, readBranchTipBranchId, updateBranchTip } from "./storage/branch-tips.ts";
|
||||
import type { SqliteDatabase } from "./types.ts";
|
||||
|
||||
export function deleteBranchCache(db: SqliteDatabase, sessionId: string) {
|
||||
deleteBranchTips(db, sessionId);
|
||||
deleteBranchEntries(db, sessionId);
|
||||
}
|
||||
|
||||
export function rebuildBranchCache(db: SqliteDatabase, sessionId: string) {
|
||||
const tips = sql`SELECT leaf.id
|
||||
FROM entries AS leaf
|
||||
WHERE leaf.session_id = ${sessionId}
|
||||
AND NOT EXISTS (
|
||||
SELECT 1 FROM entries AS child WHERE child.session_id = leaf.session_id AND child.parent_id = leaf.id
|
||||
)
|
||||
ORDER BY leaf.seq`.all<{ id: string }>(db);
|
||||
deleteBranchCache(db, sessionId);
|
||||
for (const tip of tips) buildCachedBranch(db, sessionId, tip.id);
|
||||
}
|
||||
|
||||
export function buildCachedBranch(db: SqliteDatabase, sessionId: string, leafId: string) {
|
||||
sql`SAVEPOINT build_branch_cache`.exec(db);
|
||||
try {
|
||||
const branchId = uuidv7();
|
||||
insertBranchEntriesForPath(db, sessionId, branchId, leafId);
|
||||
insertBranchTip(db, sessionId, leafId, branchId);
|
||||
sql`RELEASE SAVEPOINT build_branch_cache`.exec(db);
|
||||
} catch (error) {
|
||||
try {
|
||||
sql`ROLLBACK TO SAVEPOINT build_branch_cache`.exec(db);
|
||||
sql`RELEASE SAVEPOINT build_branch_cache`.exec(db);
|
||||
} catch {
|
||||
// Preserve the original build failure.
|
||||
}
|
||||
if (error instanceof SessionError) throw error;
|
||||
throw new SessionError(
|
||||
"storage",
|
||||
`Failed to build SQLite branch cache at entry ${leafId}`,
|
||||
error instanceof Error ? error : undefined,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
function extendBranch(
|
||||
db: SqliteDatabase,
|
||||
sessionId: string,
|
||||
branchId: string,
|
||||
parentId: string,
|
||||
entryId: string,
|
||||
entrySeq: number,
|
||||
entryType: string,
|
||||
customType: string | null,
|
||||
) {
|
||||
insertBranchEntry(db, sessionId, branchId, entryId, entrySeq, entryType, customType);
|
||||
if (!updateBranchTip(db, sessionId, branchId, parentId, entryId)) {
|
||||
throw new SessionError("invalid_entry", `Branch tip ${parentId} changed during append`);
|
||||
}
|
||||
}
|
||||
|
||||
export function appendEntryToBranchCache(
|
||||
db: SqliteDatabase,
|
||||
sessionId: string,
|
||||
entryId: string,
|
||||
entrySeq: number,
|
||||
entryType: string,
|
||||
customType: string | null,
|
||||
parentId: string | null,
|
||||
) {
|
||||
if (parentId === null) {
|
||||
const branchId = uuidv7();
|
||||
insertBranchEntry(db, sessionId, branchId, entryId, entrySeq, entryType, customType);
|
||||
insertBranchTip(db, sessionId, entryId, branchId);
|
||||
return;
|
||||
}
|
||||
|
||||
const tipBranchId = readBranchTipBranchId(db, sessionId, parentId);
|
||||
if (tipBranchId !== undefined) {
|
||||
extendBranch(db, sessionId, tipBranchId, parentId, entryId, entrySeq, entryType, customType);
|
||||
return;
|
||||
}
|
||||
|
||||
const source = readBranchContainingEntry(db, sessionId, parentId);
|
||||
if (!source) {
|
||||
throw new SessionError("invalid_entry", `Branch cache has no branch containing parent entry ${parentId}`);
|
||||
}
|
||||
|
||||
const branchId = uuidv7();
|
||||
copyBranchEntriesThroughSeq(db, sessionId, branchId, source.branchId, source.entrySeq);
|
||||
insertBranchEntry(db, sessionId, branchId, entryId, entrySeq, entryType, customType);
|
||||
insertBranchTip(db, sessionId, entryId, branchId);
|
||||
}
|
||||
|
|
@ -1,18 +1,7 @@
|
|||
export * from "./migrations.ts";
|
||||
export {
|
||||
SqliteSessionRepository,
|
||||
type SqliteSessionRepositoryOptions,
|
||||
type SqliteWriterLeaseOptions,
|
||||
} from "./repo.ts";
|
||||
export * from "./search-backend.ts";
|
||||
export * from "./sql.ts";
|
||||
export type {
|
||||
SqliteDatabase,
|
||||
SqliteDatabaseFactory,
|
||||
SqliteRunResult,
|
||||
SqliteSessionCreateOptions,
|
||||
SqliteSessionListOptions,
|
||||
SqliteSessionMetadata,
|
||||
SqliteSessionRepositoryEnv,
|
||||
SqliteStatement,
|
||||
} from "./types.ts";
|
||||
|
|
|
|||
|
|
@ -1,49 +0,0 @@
|
|||
import { readFile } from "node:fs/promises";
|
||||
import { fileURLToPath } from "node:url";
|
||||
import { sql } from "./sql.ts";
|
||||
import type { SqliteDatabase } from "./types.ts";
|
||||
|
||||
export interface SqliteMigration {
|
||||
id: string;
|
||||
order: number;
|
||||
sql: string;
|
||||
}
|
||||
|
||||
async function loadMigrationSql(relativePath: string): Promise<string> {
|
||||
return readFile(fileURLToPath(new URL(relativePath, import.meta.url)), "utf8");
|
||||
}
|
||||
|
||||
export async function loadMigrations(): Promise<SqliteMigration[]> {
|
||||
return [
|
||||
{
|
||||
id: "001_initial.sql",
|
||||
order: 1,
|
||||
sql: await loadMigrationSql("./migrations/001_initial.sql"),
|
||||
},
|
||||
];
|
||||
}
|
||||
|
||||
function ensureMigrationsTable(db: SqliteDatabase): void {
|
||||
sql`
|
||||
CREATE TABLE IF NOT EXISTS migrations (
|
||||
id TEXT PRIMARY KEY,
|
||||
applied_at TEXT NOT NULL
|
||||
);
|
||||
`.exec(db);
|
||||
}
|
||||
|
||||
export async function applyMigrations(db: SqliteDatabase): Promise<void> {
|
||||
ensureMigrationsTable(db);
|
||||
const migrations = await loadMigrations();
|
||||
const appliedRows = sql`SELECT id FROM migrations ORDER BY applied_at, id`.all<{ id: string }>(db);
|
||||
const applied = new Set(appliedRows.map((row) => row.id));
|
||||
|
||||
for (const migration of migrations) {
|
||||
if (applied.has(migration.id)) continue;
|
||||
db.transaction(() => {
|
||||
db.exec(migration.sql);
|
||||
sql`INSERT INTO migrations (id, applied_at) VALUES (${migration.id}, ${new Date().toISOString()})`.run(db);
|
||||
});
|
||||
applied.add(migration.id);
|
||||
}
|
||||
}
|
||||
|
|
@ -1,122 +0,0 @@
|
|||
CREATE TABLE IF NOT EXISTS sessions (
|
||||
id TEXT PRIMARY KEY,
|
||||
created_at INTEGER NOT NULL,
|
||||
cwd TEXT NOT NULL,
|
||||
parent_session_id TEXT NULL,
|
||||
metadata TEXT NULL
|
||||
) WITHOUT ROWID;
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_sessions_created_at ON sessions(created_at DESC);
|
||||
CREATE INDEX IF NOT EXISTS idx_sessions_cwd_created_at ON sessions(cwd, created_at DESC);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS entries (
|
||||
session_id TEXT NOT NULL,
|
||||
seq INTEGER NOT NULL,
|
||||
id TEXT NOT NULL,
|
||||
parent_id TEXT NULL,
|
||||
type TEXT NOT NULL,
|
||||
timestamp INTEGER NOT NULL,
|
||||
payload TEXT NOT NULL,
|
||||
PRIMARY KEY (session_id, id),
|
||||
UNIQUE (session_id, seq)
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_entries_session_parent ON entries(session_id, parent_id);
|
||||
CREATE INDEX IF NOT EXISTS idx_entries_session_type_seq ON entries(session_id, type, seq);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS session_sequences (
|
||||
session_id TEXT PRIMARY KEY,
|
||||
next_seq INTEGER NOT NULL
|
||||
) WITHOUT ROWID;
|
||||
|
||||
CREATE TABLE IF NOT EXISTS session_stats (
|
||||
session_id TEXT PRIMARY KEY,
|
||||
message_count INTEGER NOT NULL,
|
||||
cached_tokens REAL NOT NULL,
|
||||
uncached_tokens REAL NOT NULL,
|
||||
total_tokens REAL NOT NULL,
|
||||
cost_total REAL NOT NULL
|
||||
) WITHOUT ROWID;
|
||||
|
||||
-- Derived branch cache. Parent links in entries remain canonical; this cache
|
||||
-- exists only to make branch scans cheap.
|
||||
CREATE TABLE IF NOT EXISTS branch_entries (
|
||||
session_id TEXT NOT NULL,
|
||||
branch_id TEXT NOT NULL,
|
||||
entry_id TEXT NOT NULL,
|
||||
entry_seq INTEGER NOT NULL,
|
||||
entry_type TEXT NULL,
|
||||
custom_type TEXT NULL,
|
||||
PRIMARY KEY (session_id, branch_id, entry_id)
|
||||
) WITHOUT ROWID;
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_branch_entries_session_branch_seq ON branch_entries(session_id, branch_id, entry_seq);
|
||||
CREATE INDEX IF NOT EXISTS idx_branch_entries_session_entry ON branch_entries(session_id, entry_id, branch_id, entry_seq);
|
||||
CREATE INDEX IF NOT EXISTS idx_branch_entries_session_branch_type_seq ON branch_entries(session_id, branch_id, entry_type, entry_seq);
|
||||
CREATE INDEX IF NOT EXISTS idx_branch_entries_session_branch_custom_seq ON branch_entries(session_id, branch_id, custom_type, entry_seq);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS lanes (
|
||||
session_id TEXT NOT NULL,
|
||||
lane TEXT NOT NULL,
|
||||
leaf_id TEXT NULL,
|
||||
open_operation_id TEXT NULL,
|
||||
PRIMARY KEY (session_id, lane)
|
||||
) WITHOUT ROWID;
|
||||
|
||||
CREATE TABLE IF NOT EXISTS records (
|
||||
session_id TEXT NOT NULL,
|
||||
seq INTEGER NOT NULL,
|
||||
id TEXT NOT NULL,
|
||||
lane TEXT NOT NULL,
|
||||
run_id TEXT NULL,
|
||||
type TEXT NOT NULL,
|
||||
op_kind TEXT NULL,
|
||||
timestamp INTEGER NOT NULL,
|
||||
payload TEXT NOT NULL,
|
||||
PRIMARY KEY (session_id, id),
|
||||
UNIQUE (session_id, seq)
|
||||
) WITHOUT ROWID;
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_records_session_lane_seq ON records(session_id, lane, seq);
|
||||
CREATE INDEX IF NOT EXISTS idx_records_session_type_seq ON records(session_id, type, seq);
|
||||
CREATE INDEX IF NOT EXISTS idx_records_session_type_op_kind_seq ON records(session_id, type, op_kind, seq);
|
||||
CREATE INDEX IF NOT EXISTS idx_records_session_lane_type_seq ON records(session_id, lane, type, seq);
|
||||
CREATE INDEX IF NOT EXISTS idx_records_session_lane_type_op_kind_seq ON records(session_id, lane, type, op_kind, seq);
|
||||
CREATE INDEX IF NOT EXISTS idx_records_session_run_id_seq ON records(session_id, run_id, seq);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS lane_moves (
|
||||
session_id TEXT NOT NULL,
|
||||
seq INTEGER NOT NULL,
|
||||
lane TEXT NOT NULL,
|
||||
leaf_id TEXT NULL,
|
||||
PRIMARY KEY (session_id, seq)
|
||||
) WITHOUT ROWID;
|
||||
|
||||
|
||||
CREATE TABLE IF NOT EXISTS facts (
|
||||
session_id TEXT NOT NULL,
|
||||
seq INTEGER NOT NULL,
|
||||
kind TEXT NOT NULL,
|
||||
key TEXT NULL,
|
||||
value TEXT NULL,
|
||||
PRIMARY KEY (session_id, seq)
|
||||
) WITHOUT ROWID;
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_facts_session_kind_key_seq ON facts(session_id, kind, key, seq);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS branch_tips (
|
||||
session_id TEXT NOT NULL,
|
||||
branch_id TEXT NOT NULL,
|
||||
tip_id TEXT NOT NULL,
|
||||
PRIMARY KEY (session_id, tip_id),
|
||||
UNIQUE (session_id, branch_id)
|
||||
) WITHOUT ROWID;
|
||||
|
||||
-- Per-session writer claim. The fence prevents an expired owner from writing
|
||||
-- after a new owner takes over the session.
|
||||
CREATE TABLE IF NOT EXISTS writer_leases (
|
||||
session_id TEXT PRIMARY KEY,
|
||||
owner_id TEXT NOT NULL,
|
||||
fence INTEGER NOT NULL,
|
||||
expires_at_ms INTEGER NOT NULL
|
||||
) WITHOUT ROWID;
|
||||
|
|
@ -1,953 +0,0 @@
|
|||
import type { FileError, Result } from "@earendil-works/pi-agent-core";
|
||||
import {
|
||||
type BranchBounds,
|
||||
type Entry,
|
||||
type EntryQuery,
|
||||
type ForkOptions,
|
||||
type LaneRecord,
|
||||
type LogItem,
|
||||
type LogOptions,
|
||||
type NewRecord,
|
||||
type OperationStartedRecord,
|
||||
type ProvisionedEntry,
|
||||
type RecordQuery,
|
||||
Session,
|
||||
SessionError,
|
||||
type SessionRepo as SessionRepository,
|
||||
type SessionStats,
|
||||
type SessionStorage,
|
||||
} from "@earendil-works/pi-agent-core";
|
||||
import { uuidv7 } from "@earendil-works/pi-ai";
|
||||
import { appendEntryToBranchCache, buildCachedBranch, deleteBranchCache, rebuildBranchCache } from "./branch-cache.ts";
|
||||
import { applyMigrations } from "./migrations.ts";
|
||||
import { sql } from "./sql.ts";
|
||||
import { type CachedBranchEntryRow, queryCachedBranchRows, readCachedBranch } from "./storage/branch-entries.ts";
|
||||
import { readBranchTipIds } from "./storage/branch-tips.ts";
|
||||
import {
|
||||
deleteEntryRows,
|
||||
type EntryRow,
|
||||
entryPayload,
|
||||
idExistsInEntries,
|
||||
insertEntryRow,
|
||||
readEntryRow,
|
||||
readEntryRows,
|
||||
} from "./storage/entries.ts";
|
||||
import { appendFact, deleteFactRows, readFactRows, readLatestFact, readLatestLabelFacts } from "./storage/facts.ts";
|
||||
import {
|
||||
createInitialLane,
|
||||
deleteLaneRows,
|
||||
finishLaneOperation,
|
||||
createLane as insertLane,
|
||||
readLane,
|
||||
readLaneHead,
|
||||
readLaneMoveRows,
|
||||
readLanes,
|
||||
setLaneLeaf,
|
||||
startLaneOperation,
|
||||
moveLane as updateLane,
|
||||
} from "./storage/lanes.ts";
|
||||
import {
|
||||
appendRecordRow,
|
||||
deleteRecordRows,
|
||||
idExistsInRecords,
|
||||
readOpenOperationRows,
|
||||
readRecordRows,
|
||||
} from "./storage/records.ts";
|
||||
import {
|
||||
advanceSequence,
|
||||
createSequence,
|
||||
deleteSequence,
|
||||
getNextSequence,
|
||||
setNextSequence,
|
||||
} from "./storage/session-sequences.ts";
|
||||
import {
|
||||
addUsageToStats,
|
||||
createStats,
|
||||
deleteStats,
|
||||
incrementMessageCount,
|
||||
readStats,
|
||||
} from "./storage/session-stats.ts";
|
||||
import {
|
||||
decodeSessionMetadata,
|
||||
deleteSessionRow,
|
||||
insertSessionRow,
|
||||
readSessionRow,
|
||||
readSessionRows,
|
||||
type SessionRow,
|
||||
sessionExists,
|
||||
} from "./storage/sessions.ts";
|
||||
import {
|
||||
acquireWriterLease,
|
||||
deleteWriterLease,
|
||||
releaseWriterLease,
|
||||
renewWriterLease,
|
||||
type WriterLease,
|
||||
} from "./storage/writer-leases.ts";
|
||||
import type {
|
||||
SqliteDatabase,
|
||||
SqliteDatabaseFactory,
|
||||
SqliteSessionCreateOptions,
|
||||
SqliteSessionListOptions,
|
||||
SqliteSessionMetadata,
|
||||
SqliteSessionRepositoryEnv,
|
||||
} from "./types.ts";
|
||||
|
||||
export interface SqliteWriterLeaseOptions {
|
||||
/** Time without a successful heartbeat before another writer may take over. Default: 30 seconds. */
|
||||
ttlMs?: number;
|
||||
/** Idle heartbeat cadence. Default: 10 seconds. Must be less than ttlMs. */
|
||||
heartbeatIntervalMs?: number;
|
||||
}
|
||||
|
||||
export interface SqliteSessionRepositoryOptions {
|
||||
env: SqliteSessionRepositoryEnv;
|
||||
sqlite: SqliteDatabaseFactory;
|
||||
databasePath: string;
|
||||
writerLease?: SqliteWriterLeaseOptions;
|
||||
}
|
||||
|
||||
interface ResolvedWriterLeaseOptions {
|
||||
ttlMs: number;
|
||||
heartbeatIntervalMs: number;
|
||||
}
|
||||
|
||||
function resolveWriterLeaseOptions(options: SqliteWriterLeaseOptions | undefined): ResolvedWriterLeaseOptions {
|
||||
const ttlMs = options?.ttlMs ?? 30_000;
|
||||
const heartbeatIntervalMs = options?.heartbeatIntervalMs ?? 10_000;
|
||||
if (!Number.isSafeInteger(ttlMs) || ttlMs <= 0) throw new RangeError("writerLease.ttlMs must be positive");
|
||||
if (!Number.isSafeInteger(heartbeatIntervalMs) || heartbeatIntervalMs <= 0 || heartbeatIntervalMs >= ttlMs) {
|
||||
throw new RangeError("writerLease.heartbeatIntervalMs must be positive and less than ttlMs");
|
||||
}
|
||||
return { ttlMs, heartbeatIntervalMs };
|
||||
}
|
||||
|
||||
function activeWriterError(sessionId: string): SessionError {
|
||||
return new SessionError("storage", `SQLite session ${sessionId} already has an active writer`);
|
||||
}
|
||||
|
||||
function lostWriterError(sessionId: string): SessionError {
|
||||
return new SessionError("storage", `SQLite session ${sessionId} writer lease was lost`);
|
||||
}
|
||||
|
||||
function claimWriterLease(db: SqliteDatabase, sessionId: string, options: ResolvedWriterLeaseOptions): WriterLease {
|
||||
const now = Date.now();
|
||||
const lease = acquireWriterLease(db, sessionId, uuidv7(), now, now + options.ttlMs);
|
||||
if (!lease) throw activeWriterError(sessionId);
|
||||
return lease;
|
||||
}
|
||||
|
||||
class SerialOperationQueue {
|
||||
private tail: Promise<void> = Promise.resolve();
|
||||
|
||||
enqueue<T>(operation: () => Promise<T> | T): Promise<T> {
|
||||
const result = this.tail.then(operation);
|
||||
this.tail = result.then(
|
||||
() => undefined,
|
||||
() => undefined,
|
||||
);
|
||||
return result;
|
||||
}
|
||||
|
||||
async drain(): Promise<void> {
|
||||
await this.tail;
|
||||
}
|
||||
}
|
||||
|
||||
function resultOrThrow<T>(result: Result<T, FileError>, message: string): T {
|
||||
if (!result.ok) {
|
||||
const code = result.error.code === "not_found" ? "not_found" : "storage";
|
||||
throw new SessionError(code, `${message}: ${result.error.message}`, result.error);
|
||||
}
|
||||
return result.value;
|
||||
}
|
||||
|
||||
function getParentPath(path: string): string {
|
||||
const normalized = path.replace(/[\\/]+$/, "");
|
||||
const lastSlash = Math.max(normalized.lastIndexOf("/"), normalized.lastIndexOf("\\"));
|
||||
if (lastSlash < 0) return ".";
|
||||
if (lastSlash === 0) return normalized.slice(0, 1);
|
||||
return normalized.slice(0, lastSlash);
|
||||
}
|
||||
|
||||
function configureSqliteDatabase(db: SqliteDatabase): void {
|
||||
sql`PRAGMA journal_mode=WAL`.exec(db);
|
||||
sql`PRAGMA synchronous=FULL`.exec(db);
|
||||
sql`PRAGMA busy_timeout=5000`.exec(db);
|
||||
}
|
||||
|
||||
function entryRowFromCached(row: CachedBranchEntryRow): EntryRow {
|
||||
return { ...row, seq: row.entry_seq, type: row.type as Entry["type"] };
|
||||
}
|
||||
|
||||
function readObjectPayload(row: EntryRow): Record<string, unknown> {
|
||||
const payload = JSON.parse(row.payload) as unknown;
|
||||
if (typeof payload !== "object" || payload === null || Array.isArray(payload)) {
|
||||
throw new Error("Payload is not an object");
|
||||
}
|
||||
return payload as Record<string, unknown>;
|
||||
}
|
||||
|
||||
function decodeEntry(row: EntryRow): Entry {
|
||||
try {
|
||||
const payload = readObjectPayload(row);
|
||||
const base = { id: row.id, seq: row.seq, parentId: row.parent_id, timestamp: row.timestamp };
|
||||
switch (row.type) {
|
||||
case "message":
|
||||
if (typeof payload.message !== "object" || payload.message === null) throw new Error("Missing message");
|
||||
return {
|
||||
...base,
|
||||
type: "message",
|
||||
message: payload.message as Extract<Entry, { type: "message" }>["message"],
|
||||
...(payload.terminate === true ? { terminate: true as const } : {}),
|
||||
};
|
||||
case "model_change":
|
||||
if (typeof payload.provider !== "string" || typeof payload.modelId !== "string") {
|
||||
throw new Error("Invalid model_change payload");
|
||||
}
|
||||
return { ...base, type: "model_change", provider: payload.provider, modelId: payload.modelId };
|
||||
case "thinking_level_change":
|
||||
if (typeof payload.thinkingLevel !== "string") throw new Error("Invalid thinking_level_change payload");
|
||||
return { ...base, type: "thinking_level_change", thinkingLevel: payload.thinkingLevel };
|
||||
case "active_tools_change":
|
||||
if (!Array.isArray(payload.activeToolNames)) throw new Error("Invalid active_tools_change payload");
|
||||
if (payload.activeToolNames.some((value) => typeof value !== "string")) {
|
||||
throw new Error("Invalid active_tools_change payload");
|
||||
}
|
||||
return { ...base, type: "active_tools_change", activeToolNames: payload.activeToolNames };
|
||||
case "compaction":
|
||||
if (
|
||||
typeof payload.summary !== "string" ||
|
||||
!Array.isArray(payload.retainedTail) ||
|
||||
typeof payload.tokensBefore !== "number"
|
||||
) {
|
||||
throw new Error("Invalid compaction payload");
|
||||
}
|
||||
return {
|
||||
...base,
|
||||
type: "compaction",
|
||||
summary: payload.summary,
|
||||
retainedTail: payload.retainedTail as Extract<Entry, { type: "compaction" }>["retainedTail"],
|
||||
tokensBefore: payload.tokensBefore,
|
||||
...(Object.hasOwn(payload, "details") ? { details: payload.details } : {}),
|
||||
...(Object.hasOwn(payload, "usage")
|
||||
? { usage: payload.usage as Extract<Entry, { type: "compaction" }>["usage"] }
|
||||
: {}),
|
||||
};
|
||||
case "branch_summary":
|
||||
if (typeof payload.fromId !== "string" || typeof payload.summary !== "string") {
|
||||
throw new Error("Invalid branch_summary payload");
|
||||
}
|
||||
return {
|
||||
...base,
|
||||
type: "branch_summary",
|
||||
fromId: payload.fromId,
|
||||
summary: payload.summary,
|
||||
...(Object.hasOwn(payload, "details") ? { details: payload.details } : {}),
|
||||
...(Object.hasOwn(payload, "usage")
|
||||
? { usage: payload.usage as Extract<Entry, { type: "branch_summary" }>["usage"] }
|
||||
: {}),
|
||||
};
|
||||
case "custom":
|
||||
if (typeof payload.customType !== "string") throw new Error("Invalid custom payload");
|
||||
return {
|
||||
...base,
|
||||
type: "custom",
|
||||
customType: payload.customType,
|
||||
...(Object.hasOwn(payload, "data") ? { data: payload.data } : {}),
|
||||
};
|
||||
}
|
||||
} catch (error) {
|
||||
throw new SessionError(
|
||||
"invalid_entry",
|
||||
`Invalid SQLite session entry ${row.id}: failed to decode entry ${row.id}`,
|
||||
error instanceof Error ? error : undefined,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
function recordRunId(record: NewRecord): string | undefined {
|
||||
return record.type === "operation_started" ? record.id : "runId" in record ? record.runId : undefined;
|
||||
}
|
||||
|
||||
function recordOpKind(record: NewRecord): string | undefined {
|
||||
return record.type === "operation_started" ? record.intent.kind : undefined;
|
||||
}
|
||||
|
||||
function decodeRecord(row: { seq: number; timestamp: number; payload: string }): LaneRecord {
|
||||
try {
|
||||
return {
|
||||
...(JSON.parse(row.payload) as object),
|
||||
seq: row.seq,
|
||||
timestamp: row.timestamp,
|
||||
} as LaneRecord;
|
||||
} catch (error) {
|
||||
throw new SessionError(
|
||||
"storage",
|
||||
`Invalid SQLite session record at sequence ${row.seq}: failed to decode payload`,
|
||||
error instanceof Error ? error : undefined,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
function validateCachedBranchRows(rows: readonly CachedBranchEntryRow[], query: BranchBounds & EntryQuery): void {
|
||||
if (rows.length === 0 || query.type !== undefined || query.customType !== undefined) return;
|
||||
const path = [...rows].sort((left, right) => left.entry_seq - right.entry_seq);
|
||||
const shouldIncludeRoot =
|
||||
query.stopAtId === undefined &&
|
||||
query.stopAtType === undefined &&
|
||||
query.cursor === undefined &&
|
||||
(query.order === "oldestFirst" || query.limit === undefined);
|
||||
if (shouldIncludeRoot && path[0]?.parent_id !== null) {
|
||||
throw new SessionError("invalid_entry", `Entry ${path[0]?.parent_id} not found`);
|
||||
}
|
||||
for (let index = 1; index < path.length; index++) {
|
||||
const previous = path[index - 1]!;
|
||||
const current = path[index]!;
|
||||
if (current.parent_id !== previous.id) {
|
||||
throw new SessionError("invalid_entry", `Entry ${current.parent_id} not found`);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function matchesEntryQuery(entry: Entry, query: EntryQuery): boolean {
|
||||
return (
|
||||
(query.type === undefined || entry.type === query.type) &&
|
||||
(query.customType === undefined || (entry.type === "custom" && entry.customType === query.customType)) &&
|
||||
(query.cursor === undefined ||
|
||||
(query.order === "oldestFirst" ? entry.seq > query.cursor.afterSeq : entry.seq < query.cursor.afterSeq))
|
||||
);
|
||||
}
|
||||
|
||||
function assertUnusedId(db: SqliteDatabase, sessionId: string, id: string): void {
|
||||
if (idExistsInEntries(db, sessionId, id) || idExistsInRecords(db, sessionId, id)) {
|
||||
throw new SessionError("already_exists", `ID already exists: ${id}`);
|
||||
}
|
||||
}
|
||||
|
||||
function requireSessionRow(db: SqliteDatabase, sessionId: string): SessionRow {
|
||||
const row = readSessionRow(db, sessionId);
|
||||
if (!row) throw new SessionError("not_found", `Session not found: ${sessionId}`);
|
||||
return row;
|
||||
}
|
||||
|
||||
class SqliteSessionStorage implements SessionStorage<SqliteSessionMetadata> {
|
||||
private readonly db: SqliteDatabase;
|
||||
private readonly metadata: SqliteSessionMetadata;
|
||||
private readonly lease: WriterLease;
|
||||
private readonly leaseOptions: ResolvedWriterLeaseOptions;
|
||||
private readonly onRelease: () => void;
|
||||
private readonly operations = new SerialOperationQueue();
|
||||
private heartbeatTimer: ReturnType<typeof setTimeout> | undefined;
|
||||
private leaseError: SessionError | undefined;
|
||||
private closing = false;
|
||||
private releasePromise: Promise<void> | undefined;
|
||||
|
||||
constructor(
|
||||
db: SqliteDatabase,
|
||||
metadata: SqliteSessionMetadata,
|
||||
lease: WriterLease,
|
||||
leaseOptions: ResolvedWriterLeaseOptions,
|
||||
onRelease: () => void,
|
||||
) {
|
||||
this.db = db;
|
||||
this.metadata = metadata;
|
||||
this.lease = lease;
|
||||
this.leaseOptions = leaseOptions;
|
||||
this.onRelease = onRelease;
|
||||
this.scheduleHeartbeat();
|
||||
}
|
||||
|
||||
async release(): Promise<void> {
|
||||
this.releasePromise ??= this.finishRelease();
|
||||
await this.releasePromise;
|
||||
}
|
||||
|
||||
private async finishRelease(): Promise<void> {
|
||||
this.closing = true;
|
||||
if (this.heartbeatTimer !== undefined) clearTimeout(this.heartbeatTimer);
|
||||
try {
|
||||
await this.operations.enqueue(() =>
|
||||
this.db.transaction(() => releaseWriterLease(this.db, this.metadata.id, this.lease)),
|
||||
);
|
||||
} finally {
|
||||
this.onRelease();
|
||||
}
|
||||
}
|
||||
|
||||
private enqueueWrite<T>(operation: () => T): Promise<T> {
|
||||
if (this.closing)
|
||||
return Promise.reject(new SessionError("storage", `SQLite session ${this.metadata.id} is closed`));
|
||||
return this.operations.enqueue(() => {
|
||||
if (this.leaseError) throw this.leaseError;
|
||||
return this.db.transaction(() => {
|
||||
const now = Date.now();
|
||||
if (!renewWriterLease(this.db, this.metadata.id, this.lease, now, now + this.leaseOptions.ttlMs)) {
|
||||
this.leaseError = lostWriterError(this.metadata.id);
|
||||
if (this.heartbeatTimer !== undefined) clearTimeout(this.heartbeatTimer);
|
||||
throw this.leaseError;
|
||||
}
|
||||
return operation();
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
private scheduleHeartbeat(): void {
|
||||
if (this.closing || this.leaseError) return;
|
||||
this.heartbeatTimer = setTimeout(async () => {
|
||||
this.heartbeatTimer = undefined;
|
||||
try {
|
||||
await this.operations.enqueue(() => {
|
||||
if (this.closing || this.leaseError) return;
|
||||
this.db.transaction(() => {
|
||||
const now = Date.now();
|
||||
if (!renewWriterLease(this.db, this.metadata.id, this.lease, now, now + this.leaseOptions.ttlMs)) {
|
||||
this.leaseError = lostWriterError(this.metadata.id);
|
||||
}
|
||||
});
|
||||
});
|
||||
} catch {
|
||||
// A transient heartbeat failure is retried. Every write still verifies ownership transactionally.
|
||||
} finally {
|
||||
this.scheduleHeartbeat();
|
||||
}
|
||||
}, this.leaseOptions.heartbeatIntervalMs);
|
||||
this.heartbeatTimer.unref();
|
||||
}
|
||||
|
||||
async getMetadata(): Promise<SqliteSessionMetadata> {
|
||||
return decodeSessionMetadata(requireSessionRow(this.db, this.metadata.id), this.metadata.path);
|
||||
}
|
||||
|
||||
isForSession(sessionId: string): boolean {
|
||||
return this.metadata.id === sessionId;
|
||||
}
|
||||
|
||||
async getLanes(): Promise<{ lane: string; leafId: string | null }[]> {
|
||||
return readLanes(this.db, this.metadata.id).map((row) => ({ lane: row.lane, leafId: row.leaf_id }));
|
||||
}
|
||||
|
||||
async createLane(lane: string, at: string | null): Promise<void> {
|
||||
return this.enqueueWrite(() => {
|
||||
if (readLane(this.db, this.metadata.id, lane)) {
|
||||
throw new SessionError("already_exists", `Lane already exists: ${lane}`);
|
||||
}
|
||||
if (at !== null && !readEntryRow(this.db, this.metadata.id, at)) {
|
||||
throw new SessionError("not_found", `Entry not found: ${at}`);
|
||||
}
|
||||
const seq = getNextSequence(this.db, this.metadata.id);
|
||||
insertLane(this.db, this.metadata.id, seq, lane, at);
|
||||
advanceSequence(this.db, this.metadata.id, seq);
|
||||
});
|
||||
}
|
||||
|
||||
async moveLane(lane: string, to: string | null): Promise<void> {
|
||||
return this.enqueueWrite(() => {
|
||||
if (!readLane(this.db, this.metadata.id, lane))
|
||||
throw new SessionError("invalid_lane", `Lane not found: ${lane}`);
|
||||
if (to !== null && !readEntryRow(this.db, this.metadata.id, to)) {
|
||||
throw new SessionError("not_found", `Entry not found: ${to}`);
|
||||
}
|
||||
const seq = getNextSequence(this.db, this.metadata.id);
|
||||
updateLane(this.db, this.metadata.id, seq, lane, to);
|
||||
advanceSequence(this.db, this.metadata.id, seq);
|
||||
});
|
||||
}
|
||||
|
||||
async appendEntry<TEntry extends Entry>(entry: ProvisionedEntry<TEntry>, lane: string): Promise<TEntry> {
|
||||
return this.enqueueWrite(() => {
|
||||
const parentId = readLaneHead(this.db, this.metadata.id, lane).leafId;
|
||||
assertUnusedId(this.db, this.metadata.id, entry.id);
|
||||
const seq = getNextSequence(this.db, this.metadata.id);
|
||||
const committed = { ...entry, parentId, seq, timestamp: Date.now() } as Entry;
|
||||
insertEntryRow(this.db, this.metadata.id, {
|
||||
seq,
|
||||
id: committed.id,
|
||||
parentId: committed.parentId,
|
||||
type: committed.type,
|
||||
timestamp: committed.timestamp,
|
||||
payload: JSON.stringify(entryPayload(committed)),
|
||||
});
|
||||
setLaneLeaf(this.db, this.metadata.id, lane, committed.id);
|
||||
appendEntryToBranchCache(
|
||||
this.db,
|
||||
this.metadata.id,
|
||||
committed.id,
|
||||
seq,
|
||||
committed.type,
|
||||
committed.type === "custom" ? committed.customType : null,
|
||||
committed.parentId,
|
||||
);
|
||||
if (committed.type === "message") incrementMessageCount(this.db, this.metadata.id);
|
||||
advanceSequence(this.db, this.metadata.id, seq);
|
||||
return structuredClone(committed as TEntry);
|
||||
});
|
||||
}
|
||||
|
||||
async appendRecord<TRecord extends LaneRecord>(record: NewRecord<TRecord>): Promise<TRecord>;
|
||||
async appendRecord(record: NewRecord): Promise<LaneRecord> {
|
||||
return this.enqueueWrite(() => {
|
||||
if (!readLane(this.db, this.metadata.id, record.lane)) {
|
||||
throw new SessionError("invalid_lane", `Lane not found: ${record.lane}`);
|
||||
}
|
||||
assertUnusedId(this.db, this.metadata.id, record.id);
|
||||
const seq = getNextSequence(this.db, this.metadata.id);
|
||||
const committed: LaneRecord = { ...record, seq, timestamp: Date.now() };
|
||||
if (record.type === "operation_started") {
|
||||
startLaneOperation(this.db, this.metadata.id, record.lane, record.id);
|
||||
}
|
||||
appendRecordRow(this.db, this.metadata.id, {
|
||||
seq,
|
||||
id: record.id,
|
||||
lane: record.lane,
|
||||
runId: recordRunId(record),
|
||||
type: record.type,
|
||||
opKind: recordOpKind(record),
|
||||
timestamp: committed.timestamp,
|
||||
payload: JSON.stringify(record),
|
||||
});
|
||||
if (record.type === "operation_finished") {
|
||||
finishLaneOperation(this.db, this.metadata.id, record.lane, record.runId);
|
||||
}
|
||||
if (record.type === "usage") addUsageToStats(this.db, this.metadata.id, record.usage);
|
||||
advanceSequence(this.db, this.metadata.id, seq);
|
||||
return structuredClone(committed);
|
||||
});
|
||||
}
|
||||
|
||||
async getEntry(id: string): Promise<Entry | undefined> {
|
||||
const row = readEntryRow(this.db, this.metadata.id, id);
|
||||
return row ? decodeEntry(row) : undefined;
|
||||
}
|
||||
|
||||
async findEntries(query: EntryQuery = {}): Promise<Entry[]> {
|
||||
const sqlType = query.type ?? (query.customType === undefined ? undefined : "custom");
|
||||
const sqlLimit = query.customType === undefined ? query.limit : undefined;
|
||||
const rows = readEntryRows(this.db, this.metadata.id, {
|
||||
cursor: query.cursor,
|
||||
limit: sqlLimit,
|
||||
order: query.order,
|
||||
type: sqlType,
|
||||
});
|
||||
const entries = rows.map(decodeEntry).filter((entry) => matchesEntryQuery(entry, query));
|
||||
return query.limit === undefined ? entries : entries.slice(0, query.limit);
|
||||
}
|
||||
|
||||
async findEntriesOnBranch(query: EntryQuery & BranchBounds & { start: string }): Promise<Entry[]> {
|
||||
const cached = readCachedBranch(this.db, this.metadata.id, query.start);
|
||||
if (!cached) {
|
||||
if (!readEntryRow(this.db, this.metadata.id, query.start))
|
||||
throw new SessionError("not_found", `Entry not found: ${query.start}`);
|
||||
throw new SessionError("invalid_entry", `Branch cache missing entry ${query.start}`);
|
||||
}
|
||||
const rows = queryCachedBranchRows(this.db, this.metadata.id, cached, query);
|
||||
validateCachedBranchRows(rows, query);
|
||||
const entries = rows
|
||||
.map(entryRowFromCached)
|
||||
.map(decodeEntry)
|
||||
.filter((entry) => matchesEntryQuery(entry, query));
|
||||
return query.limit === undefined ? entries : entries.slice(0, query.limit);
|
||||
}
|
||||
|
||||
async findRecords(query: RecordQuery = {}): Promise<LaneRecord[]> {
|
||||
const rows = readRecordRows(this.db, this.metadata.id, query);
|
||||
return rows.map(decodeRecord);
|
||||
}
|
||||
|
||||
async findOpenOperations(lane: string, options?: { limit?: number }): Promise<OperationStartedRecord[]> {
|
||||
const rows = readOpenOperationRows(this.db, this.metadata.id, lane, options);
|
||||
|
||||
return rows.map((row) => {
|
||||
const record = decodeRecord(row);
|
||||
if (record.type !== "operation_started") {
|
||||
throw new SessionError("storage", "Expected operation_started record");
|
||||
}
|
||||
return record;
|
||||
});
|
||||
}
|
||||
|
||||
async getLog(options: LogOptions = {}): Promise<LogItem[]> {
|
||||
const afterSeq = options.afterSeq ?? 0;
|
||||
const limit = options.limit;
|
||||
const entryRows = readEntryRows(this.db, this.metadata.id, { afterSeq, order: "oldestFirst", limit });
|
||||
const recordRows = readRecordRows(this.db, this.metadata.id, { afterSeq, order: "oldestFirst", limit });
|
||||
const laneRows = readLaneMoveRows(this.db, this.metadata.id, { afterSeq, limit });
|
||||
const factRows = readFactRows(this.db, this.metadata.id, { afterSeq, limit });
|
||||
|
||||
const logRows: { seq: number; decode: () => LogItem }[] = [
|
||||
...entryRows.map((row) => ({
|
||||
seq: row.seq,
|
||||
decode: () => ({ kind: "entry" as const, seq: row.seq, entry: decodeEntry(row) }),
|
||||
})),
|
||||
...recordRows.map((row) => ({
|
||||
seq: row.seq,
|
||||
decode: () => ({ kind: "record" as const, seq: row.seq, record: decodeRecord(row) }),
|
||||
})),
|
||||
...laneRows.map((row) => ({
|
||||
seq: row.seq,
|
||||
decode: () => ({ kind: "lane" as const, seq: row.seq, lane: row.lane, leafId: row.leaf_id }),
|
||||
})),
|
||||
...factRows.map((row) => ({
|
||||
seq: row.seq,
|
||||
decode: () => {
|
||||
if (row.kind === "name")
|
||||
return {
|
||||
kind: "fact" as const,
|
||||
seq: row.seq,
|
||||
fact: "name" as const,
|
||||
name: row.value === null ? undefined : (JSON.parse(row.value) as string),
|
||||
};
|
||||
return {
|
||||
kind: "fact" as const,
|
||||
seq: row.seq,
|
||||
fact: "label" as const,
|
||||
targetId: row.key ?? "",
|
||||
label: row.value === null ? undefined : (JSON.parse(row.value) as string),
|
||||
};
|
||||
},
|
||||
})),
|
||||
].sort((left, right) => left.seq - right.seq);
|
||||
const selectedRows = options.limit === undefined ? logRows : logRows.slice(0, options.limit);
|
||||
return selectedRows.map((row) => row.decode());
|
||||
}
|
||||
|
||||
async getName(): Promise<string | undefined> {
|
||||
const row = readLatestFact(this.db, this.metadata.id, "name", null);
|
||||
return row?.value === undefined || row.value === null ? undefined : (JSON.parse(row.value) as string);
|
||||
}
|
||||
|
||||
async setName(name: string | undefined): Promise<void> {
|
||||
return this.enqueueWrite(() => {
|
||||
const seq = getNextSequence(this.db, this.metadata.id);
|
||||
appendFact(this.db, this.metadata.id, seq, "name", null, name === undefined ? null : JSON.stringify(name));
|
||||
advanceSequence(this.db, this.metadata.id, seq);
|
||||
});
|
||||
}
|
||||
|
||||
async getLabel(id: string): Promise<string | undefined> {
|
||||
const row = readLatestFact(this.db, this.metadata.id, "label", id);
|
||||
return row?.value === undefined || row.value === null ? undefined : (JSON.parse(row.value) as string);
|
||||
}
|
||||
|
||||
async setLabel(id: string, label: string | undefined): Promise<void> {
|
||||
return this.enqueueWrite(() => {
|
||||
if (!readEntryRow(this.db, this.metadata.id, id)) {
|
||||
throw new SessionError("not_found", `Entry not found: ${id}`);
|
||||
}
|
||||
const seq = getNextSequence(this.db, this.metadata.id);
|
||||
appendFact(this.db, this.metadata.id, seq, "label", id, label === undefined ? null : JSON.stringify(label));
|
||||
advanceSequence(this.db, this.metadata.id, seq);
|
||||
});
|
||||
}
|
||||
|
||||
async getStats(): Promise<SessionStats> {
|
||||
return readStats(this.db, this.metadata.id);
|
||||
}
|
||||
}
|
||||
|
||||
function claimStorage(
|
||||
db: SqliteDatabase,
|
||||
metadata: SqliteSessionMetadata,
|
||||
leaseOptions: ResolvedWriterLeaseOptions,
|
||||
onRelease: () => void,
|
||||
): SqliteSessionStorage {
|
||||
requireSessionRow(db, metadata.id);
|
||||
const claimed = db.transaction(() => {
|
||||
const lease = claimWriterLease(db, metadata.id, leaseOptions);
|
||||
const row = requireSessionRow(db, metadata.id);
|
||||
readLanes(db, metadata.id);
|
||||
return { lease, row };
|
||||
});
|
||||
return new SqliteSessionStorage(
|
||||
db,
|
||||
decodeSessionMetadata(claimed.row, metadata.path),
|
||||
claimed.lease,
|
||||
leaseOptions,
|
||||
onRelease,
|
||||
);
|
||||
}
|
||||
|
||||
export class SqliteSessionRepository
|
||||
implements
|
||||
SessionRepository<SqliteSessionMetadata, SqliteSessionCreateOptions, SqliteSessionListOptions>,
|
||||
AsyncDisposable
|
||||
{
|
||||
private databasePath: string | undefined;
|
||||
private database: SqliteDatabase | undefined;
|
||||
private databasePromise: Promise<SqliteDatabase> | undefined;
|
||||
private readonly operations = new SerialOperationQueue();
|
||||
private readonly activeStorages = new Set<SqliteSessionStorage>();
|
||||
private readonly options: SqliteSessionRepositoryOptions;
|
||||
private readonly leaseOptions: ResolvedWriterLeaseOptions;
|
||||
|
||||
constructor(options: SqliteSessionRepositoryOptions) {
|
||||
this.options = options;
|
||||
this.leaseOptions = resolveWriterLeaseOptions(options.writerLease);
|
||||
}
|
||||
|
||||
private async releaseStoragesForSession(sessionId: string): Promise<void> {
|
||||
for (const storage of [...this.activeStorages]) {
|
||||
if (storage.isForSession(sessionId)) await storage.release();
|
||||
}
|
||||
}
|
||||
|
||||
private sessionFromLease(
|
||||
db: SqliteDatabase,
|
||||
metadata: SqliteSessionMetadata,
|
||||
lease: WriterLease,
|
||||
): Session<SqliteSessionMetadata> {
|
||||
let storage: SqliteSessionStorage;
|
||||
storage = new SqliteSessionStorage(db, metadata, lease, this.leaseOptions, () => {
|
||||
this.activeStorages.delete(storage);
|
||||
});
|
||||
this.activeStorages.add(storage);
|
||||
return new Session(storage);
|
||||
}
|
||||
|
||||
private claimSession(db: SqliteDatabase, metadata: SqliteSessionMetadata): Session<SqliteSessionMetadata> {
|
||||
const active = [...this.activeStorages].find((storage) => storage.isForSession(metadata.id));
|
||||
if (active) {
|
||||
readLanes(db, metadata.id);
|
||||
return new Session(active);
|
||||
}
|
||||
let storage: SqliteSessionStorage;
|
||||
storage = claimStorage(db, metadata, this.leaseOptions, () => {
|
||||
this.activeStorages.delete(storage);
|
||||
});
|
||||
this.activeStorages.add(storage);
|
||||
return new Session(storage);
|
||||
}
|
||||
|
||||
async create(options: SqliteSessionCreateOptions): Promise<Session<SqliteSessionMetadata>> {
|
||||
return this.operations.enqueue(async () => {
|
||||
const db = await this.getDatabase();
|
||||
const path = await this.getDatabasePath();
|
||||
const id = options.id ?? uuidv7();
|
||||
if (sessionExists(db, id)) throw new SessionError("already_exists", `Session already exists: ${id}`);
|
||||
const createdAt = Date.now();
|
||||
const lease = db.transaction(() => {
|
||||
insertSessionRow(db, {
|
||||
id,
|
||||
createdAt,
|
||||
cwd: options.cwd,
|
||||
parentSessionId: options.parentSessionId,
|
||||
metadata: options.metadata,
|
||||
});
|
||||
createSequence(db, id);
|
||||
createStats(db, id);
|
||||
createInitialLane(db, id);
|
||||
return claimWriterLease(db, id, this.leaseOptions);
|
||||
});
|
||||
const row = requireSessionRow(db, id);
|
||||
return this.sessionFromLease(db, decodeSessionMetadata(row, path), lease);
|
||||
});
|
||||
}
|
||||
|
||||
async open(metadata: SqliteSessionMetadata): Promise<Session<SqliteSessionMetadata>> {
|
||||
return this.operations.enqueue(async () => this.claimSession(await this.getDatabase(), metadata));
|
||||
}
|
||||
|
||||
/** Rebuilds this session's private branch-read cache from canonical entry parent links. */
|
||||
async repairBranchCache(metadata: SqliteSessionMetadata): Promise<void> {
|
||||
return this.operations.enqueue(async () => {
|
||||
await this.releaseStoragesForSession(metadata.id);
|
||||
const db = await this.getDatabase();
|
||||
db.transaction(() => {
|
||||
const lease = claimWriterLease(db, metadata.id, this.leaseOptions);
|
||||
requireSessionRow(db, metadata.id);
|
||||
rebuildBranchCache(db, metadata.id);
|
||||
releaseWriterLease(db, metadata.id, lease);
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
/** Reads the session catalog without acquiring or renewing per-session writer leases. */
|
||||
async list(options: SqliteSessionListOptions = {}): Promise<SqliteSessionMetadata[]> {
|
||||
return this.operations.enqueue(async () => {
|
||||
const path = await this.getDatabasePath();
|
||||
if (!resultOrThrow(await this.options.env.exists(path), `Failed to check database ${path}`)) return [];
|
||||
const db = await this.getDatabase();
|
||||
const rows = readSessionRows(db, options);
|
||||
return rows.map((row) => decodeSessionMetadata(row, path));
|
||||
});
|
||||
}
|
||||
|
||||
async delete(metadata: SqliteSessionMetadata): Promise<void> {
|
||||
return this.operations.enqueue(async () => {
|
||||
await this.releaseStoragesForSession(metadata.id);
|
||||
const db = await this.getDatabase();
|
||||
db.transaction(() => {
|
||||
if (!sessionExists(db, metadata.id)) {
|
||||
deleteWriterLease(db, metadata.id);
|
||||
return;
|
||||
}
|
||||
claimWriterLease(db, metadata.id, this.leaseOptions);
|
||||
deleteBranchCache(db, metadata.id);
|
||||
deleteFactRows(db, metadata.id);
|
||||
deleteLaneRows(db, metadata.id);
|
||||
deleteRecordRows(db, metadata.id);
|
||||
deleteEntryRows(db, metadata.id);
|
||||
deleteWriterLease(db, metadata.id);
|
||||
deleteStats(db, metadata.id);
|
||||
deleteSequence(db, metadata.id);
|
||||
deleteSessionRow(db, metadata.id);
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
async fork(
|
||||
source: SqliteSessionMetadata,
|
||||
options: ForkOptions & SqliteSessionCreateOptions,
|
||||
): Promise<Session<SqliteSessionMetadata>> {
|
||||
return this.operations.enqueue(async () => {
|
||||
const db = await this.getDatabase();
|
||||
const path = await this.getDatabasePath();
|
||||
const sourceMetadata = decodeSessionMetadata(requireSessionRow(db, source.id), path);
|
||||
const id = options.id ?? uuidv7();
|
||||
if (sessionExists(db, id)) throw new SessionError("already_exists", `Session already exists: ${id}`);
|
||||
|
||||
const entries: EntryRow[] = [];
|
||||
const lanes: { lane: string; leafId: string | null }[] = [];
|
||||
const branchTips: string[] = [];
|
||||
let branchForkTargetId: string | null = null;
|
||||
|
||||
if (options.scope === "tree") {
|
||||
entries.push(...readEntryRows(db, source.id, { order: "oldestFirst" }));
|
||||
lanes.push(...readLanes(db, source.id).map((row) => ({ lane: row.lane, leafId: row.leaf_id })));
|
||||
branchTips.push(...readBranchTipIds(db, source.id));
|
||||
} else {
|
||||
const main = readLane(db, source.id, "main");
|
||||
if (!main) throw new SessionError("invalid_lane", "Lane not found: main");
|
||||
const selectedEntryId = options.entryId ?? main.leaf_id;
|
||||
if (selectedEntryId !== null) {
|
||||
const target = readEntryRow(db, source.id, selectedEntryId);
|
||||
if (!target || target.type !== "message") {
|
||||
throw new SessionError(
|
||||
"invalid_fork_target",
|
||||
`Fork target is not a message entry: ${selectedEntryId}`,
|
||||
);
|
||||
}
|
||||
const position = options.position ?? (options.entryId === undefined ? "at" : "before");
|
||||
branchForkTargetId = position === "at" ? target.id : target.parent_id;
|
||||
}
|
||||
lanes.push({ lane: "main", leafId: branchForkTargetId });
|
||||
if (branchForkTargetId !== null) {
|
||||
const cached = readCachedBranch(db, source.id, branchForkTargetId);
|
||||
if (!cached) {
|
||||
throw new SessionError(
|
||||
"invalid_fork_target",
|
||||
`Fork target is not on a cached branch: ${branchForkTargetId}`,
|
||||
);
|
||||
}
|
||||
const rows = queryCachedBranchRows(db, source.id, cached, { order: "oldestFirst" });
|
||||
entries.push(...rows.map(entryRowFromCached));
|
||||
branchTips.push(branchForkTargetId);
|
||||
}
|
||||
}
|
||||
|
||||
const copiedIds = new Set(entries.map((entry) => entry.id));
|
||||
const latestName = readLatestFact(db, source.id, "name", null);
|
||||
const latestLabels = readLatestLabelFacts(db, source.id);
|
||||
const labelsToCopy = latestLabels.filter(
|
||||
(row) => options.scope === "tree" || (row.key !== null && copiedIds.has(row.key)),
|
||||
);
|
||||
const createdAt = Date.now();
|
||||
const metadata = options.metadata ?? sourceMetadata.metadata;
|
||||
let lease: WriterLease;
|
||||
|
||||
try {
|
||||
lease = db.transaction(() => {
|
||||
insertSessionRow(db, {
|
||||
id,
|
||||
createdAt,
|
||||
cwd: options.cwd,
|
||||
parentSessionId: options.parentSessionId ?? source.id,
|
||||
metadata,
|
||||
});
|
||||
createSequence(db, id);
|
||||
createStats(db, id, entries.filter((entry) => entry.type === "message").length);
|
||||
|
||||
let nextSeq = 1;
|
||||
const allocateSeq = () => nextSeq++;
|
||||
for (const entry of entries) {
|
||||
insertEntryRow(db, id, {
|
||||
seq: allocateSeq(),
|
||||
id: entry.id,
|
||||
parentId: entry.parent_id,
|
||||
type: entry.type,
|
||||
timestamp: entry.timestamp,
|
||||
payload: entry.payload,
|
||||
});
|
||||
}
|
||||
|
||||
if (options.scope === "tree") {
|
||||
for (const lane of lanes) insertLane(db, id, allocateSeq(), lane.lane, lane.leafId);
|
||||
} else {
|
||||
createInitialLane(db, id, "main", branchForkTargetId);
|
||||
}
|
||||
|
||||
if (latestName?.value !== undefined && latestName.value !== null) {
|
||||
appendFact(db, id, allocateSeq(), "name", null, latestName.value);
|
||||
}
|
||||
for (const label of labelsToCopy) appendFact(db, id, allocateSeq(), "label", label.key, label.value);
|
||||
|
||||
setNextSequence(db, id, nextSeq);
|
||||
for (const tip of branchTips) buildCachedBranch(db, id, tip);
|
||||
return claimWriterLease(db, id, this.leaseOptions);
|
||||
});
|
||||
} catch (error) {
|
||||
if (error instanceof SessionError) throw error;
|
||||
throw new SessionError(
|
||||
"storage",
|
||||
`Failed to fork SQLite session ${id}`,
|
||||
error instanceof Error ? error : undefined,
|
||||
);
|
||||
}
|
||||
|
||||
const row = requireSessionRow(db, id);
|
||||
return this.sessionFromLease(db, decodeSessionMetadata(row, path), lease);
|
||||
});
|
||||
}
|
||||
|
||||
async close(): Promise<void> {
|
||||
await this.operations.drain();
|
||||
for (const storage of [...this.activeStorages]) await storage.release();
|
||||
if (this.database) this.database.close();
|
||||
this.database = undefined;
|
||||
this.databasePromise = undefined;
|
||||
}
|
||||
|
||||
async [Symbol.asyncDispose](): Promise<void> {
|
||||
await this.close();
|
||||
}
|
||||
|
||||
private async getDatabasePath(): Promise<string> {
|
||||
this.databasePath ??= resultOrThrow(
|
||||
await this.options.env.absolutePath(this.options.databasePath),
|
||||
`Failed to resolve SQLite sessions database ${this.options.databasePath}`,
|
||||
);
|
||||
return this.databasePath;
|
||||
}
|
||||
|
||||
private async getDatabase(): Promise<SqliteDatabase> {
|
||||
if (!this.databasePromise) this.databasePromise = this.openDatabase();
|
||||
this.database = await this.databasePromise;
|
||||
return this.database;
|
||||
}
|
||||
|
||||
private async openDatabase(): Promise<SqliteDatabase> {
|
||||
const path = await this.getDatabasePath();
|
||||
resultOrThrow(
|
||||
await this.options.env.createDir(getParentPath(path), { recursive: true }),
|
||||
`Failed to create SQLite sessions directory ${path}`,
|
||||
);
|
||||
const db = await this.options.sqlite.open(path);
|
||||
try {
|
||||
configureSqliteDatabase(db);
|
||||
await applyMigrations(db);
|
||||
return db;
|
||||
} catch (error) {
|
||||
db.close();
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -1,194 +0,0 @@
|
|||
import type {
|
||||
FileError,
|
||||
Result,
|
||||
SessionSearch,
|
||||
SessionSearchHit,
|
||||
SessionSearchOptions,
|
||||
} from "@earendil-works/pi-agent-core";
|
||||
import { SessionError } from "@earendil-works/pi-agent-core";
|
||||
import { applyMigrations } from "./migrations.ts";
|
||||
import { sql } from "./sql.ts";
|
||||
import { decodeSessionMetadata, type SessionRow } from "./storage/sessions.ts";
|
||||
import type {
|
||||
SqliteDatabase,
|
||||
SqliteDatabaseFactory,
|
||||
SqliteSessionMetadata,
|
||||
SqliteSessionRepositoryEnv,
|
||||
} from "./types.ts";
|
||||
|
||||
function getFileSystemResultOrThrow<TValue>(result: Result<TValue, FileError>, message: string): TValue {
|
||||
if (!result.ok) {
|
||||
const code = result.error.code === "not_found" ? "not_found" : "storage";
|
||||
throw new SessionError(code, `${message}: ${result.error.message}`, result.error);
|
||||
}
|
||||
return result.value;
|
||||
}
|
||||
|
||||
function getParentPath(path: string): string {
|
||||
const normalized = path.replace(/[\\/]+$/, "");
|
||||
const lastSlash = Math.max(normalized.lastIndexOf("/"), normalized.lastIndexOf("\\"));
|
||||
if (lastSlash < 0) return ".";
|
||||
if (lastSlash === 0) return normalized.slice(0, 1);
|
||||
return normalized.slice(0, lastSlash);
|
||||
}
|
||||
|
||||
function throwIfAborted(signal: AbortSignal | undefined): void {
|
||||
if (!signal?.aborted) return;
|
||||
if (signal.reason instanceof Error) throw signal.reason;
|
||||
const error = new Error("The operation was aborted");
|
||||
error.name = "AbortError";
|
||||
throw error;
|
||||
}
|
||||
|
||||
function configureSqliteDatabase(db: SqliteDatabase): void {
|
||||
sql`PRAGMA journal_mode=WAL`.exec(db);
|
||||
sql`PRAGMA synchronous=FULL`.exec(db);
|
||||
sql`PRAGMA busy_timeout=5000`.exec(db);
|
||||
}
|
||||
|
||||
export interface SqliteSessionSearchOptions {
|
||||
env: Pick<SqliteSessionRepositoryEnv, "absolutePath" | "createDir">;
|
||||
sqlite: SqliteDatabaseFactory;
|
||||
databasePath: string;
|
||||
}
|
||||
|
||||
function tableExists(db: SqliteDatabase, name: string): boolean {
|
||||
return !!sql`SELECT 1 AS found FROM sqlite_master WHERE type = 'table' AND name = ${name} LIMIT 1`.get<{
|
||||
found: number;
|
||||
}>(db);
|
||||
}
|
||||
|
||||
function rebuildSearchIndex(db: SqliteDatabase): void {
|
||||
sql`INSERT INTO session_search_fts(session_search_fts) VALUES('rebuild')`.run(db);
|
||||
}
|
||||
|
||||
function ensureSearchSchema(db: SqliteDatabase): void {
|
||||
const ftsExists = tableExists(db, "session_search_fts");
|
||||
const entriesExist = tableExists(db, "entries");
|
||||
db.transaction(() => {
|
||||
sql`
|
||||
CREATE VIRTUAL TABLE IF NOT EXISTS session_search_fts USING fts5(
|
||||
payload,
|
||||
content = 'entries',
|
||||
content_rowid = 'rowid',
|
||||
tokenize = 'trigram remove_diacritics 1'
|
||||
);
|
||||
CREATE TRIGGER IF NOT EXISTS session_search_fts_ai AFTER INSERT ON entries BEGIN
|
||||
INSERT INTO session_search_fts(rowid, payload) VALUES (new.rowid, new.payload);
|
||||
END;
|
||||
CREATE TRIGGER IF NOT EXISTS session_search_fts_ad AFTER DELETE ON entries BEGIN
|
||||
INSERT INTO session_search_fts(session_search_fts, rowid, payload) VALUES('delete', old.rowid, old.payload);
|
||||
END;
|
||||
CREATE TRIGGER IF NOT EXISTS session_search_fts_au AFTER UPDATE OF payload ON entries BEGIN
|
||||
INSERT INTO session_search_fts(session_search_fts, rowid, payload) VALUES('delete', old.rowid, old.payload);
|
||||
INSERT INTO session_search_fts(rowid, payload) VALUES (new.rowid, new.payload);
|
||||
END;
|
||||
`.exec(db);
|
||||
if (!ftsExists && entriesExist) rebuildSearchIndex(db);
|
||||
});
|
||||
}
|
||||
|
||||
export interface SqliteSessionSearchHit extends SessionSearchHit {
|
||||
readonly metadata: SqliteSessionMetadata;
|
||||
readonly timestamp: number;
|
||||
readonly score: number;
|
||||
}
|
||||
|
||||
/** SQLite FTS search over a co-located canonical session database. */
|
||||
class SqliteSessionSearch implements SessionSearch<SqliteSessionSearchHit> {
|
||||
private readonly options: SqliteSessionSearchOptions;
|
||||
private databasePath: string | undefined;
|
||||
|
||||
constructor(options: SqliteSessionSearchOptions) {
|
||||
this.options = options;
|
||||
}
|
||||
|
||||
private async getDatabasePath(): Promise<string> {
|
||||
if (!this.databasePath) {
|
||||
this.databasePath = getFileSystemResultOrThrow(
|
||||
await this.options.env.absolutePath(this.options.databasePath),
|
||||
`Failed to resolve SQLite search database ${this.options.databasePath}`,
|
||||
);
|
||||
}
|
||||
return this.databasePath;
|
||||
}
|
||||
|
||||
private async openDatabase(): Promise<SqliteDatabase> {
|
||||
const path = await this.getDatabasePath();
|
||||
const directory = getParentPath(path);
|
||||
getFileSystemResultOrThrow(
|
||||
await this.options.env.createDir(directory, { recursive: true }),
|
||||
`Failed to create SQLite search directory ${directory}`,
|
||||
);
|
||||
const db = await this.options.sqlite.open(path);
|
||||
try {
|
||||
configureSqliteDatabase(db);
|
||||
await applyMigrations(db);
|
||||
ensureSearchSchema(db);
|
||||
return db;
|
||||
} catch (error) {
|
||||
db.close();
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
async *search(text: string, options: SessionSearchOptions = {}): AsyncIterable<SqliteSessionSearchHit> {
|
||||
const queryText = text.trim();
|
||||
if (!queryText || (options.limit !== undefined && options.limit <= 0)) return;
|
||||
if (options.entryTypes?.length === 0) return;
|
||||
throwIfAborted(options.signal);
|
||||
const db = await this.openDatabase();
|
||||
try {
|
||||
const query = `"${queryText.replaceAll('"', '""')}"`;
|
||||
const predicates = ["session_search_fts MATCH ?"];
|
||||
const params: unknown[] = [query];
|
||||
if (options.entryTypes !== undefined) {
|
||||
predicates.push(`se.type IN (${options.entryTypes.map(() => "?").join(", ")})`);
|
||||
params.push(...options.entryTypes);
|
||||
}
|
||||
const rows = db
|
||||
.prepare(
|
||||
`SELECT s.id, s.created_at, s.metadata, s.cwd, s.parent_session_id,
|
||||
name_fact.seq IS NOT NULL AS has_session_name,
|
||||
name_fact.value AS session_name,
|
||||
se.id AS entry_id, se.timestamp, bm25(session_search_fts) AS score
|
||||
FROM session_search_fts
|
||||
JOIN entries AS se ON se.rowid = session_search_fts.rowid
|
||||
JOIN sessions AS s ON s.id = se.session_id
|
||||
LEFT JOIN facts AS name_fact
|
||||
ON name_fact.session_id = s.id
|
||||
AND name_fact.kind = 'name'
|
||||
AND name_fact.key IS NULL
|
||||
AND name_fact.seq = (
|
||||
SELECT MAX(f.seq)
|
||||
FROM facts AS f
|
||||
WHERE f.session_id = s.id AND f.kind = 'name' AND f.key IS NULL
|
||||
)
|
||||
WHERE ${predicates.join(" AND ")}
|
||||
ORDER BY score
|
||||
LIMIT ?`,
|
||||
)
|
||||
.iterate<SessionRow & { entry_id: string; timestamp: number; score: number }>(
|
||||
...params,
|
||||
options.limit ?? -1,
|
||||
);
|
||||
const path = await this.getDatabasePath();
|
||||
for (const row of rows) {
|
||||
throwIfAborted(options.signal);
|
||||
yield {
|
||||
sessionId: row.id,
|
||||
metadata: decodeSessionMetadata(row, path),
|
||||
entryId: row.entry_id,
|
||||
timestamp: row.timestamp,
|
||||
score: row.score,
|
||||
};
|
||||
}
|
||||
} finally {
|
||||
db.close();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export function createSqliteSessionSearch(options: SqliteSessionSearchOptions): SessionSearch<SqliteSessionSearchHit> {
|
||||
return new SqliteSessionSearch(options);
|
||||
}
|
||||
|
|
@ -1,174 +0,0 @@
|
|||
import { type Entry, SessionError } from "@earendil-works/pi-agent-core";
|
||||
import { joinSqlFragments, sql } from "../sql.ts";
|
||||
import type { SqliteDatabase } from "../types.ts";
|
||||
|
||||
/** Derived root-to-tip branch cache membership. Canonical parent links remain in entries. */
|
||||
export interface CachedBranch {
|
||||
branchId: string;
|
||||
leafSeq: number;
|
||||
}
|
||||
|
||||
export interface CachedBranchEntryRow {
|
||||
session_id: string;
|
||||
id: string;
|
||||
entry_seq: number;
|
||||
parent_id: string | null;
|
||||
type: Entry["type"];
|
||||
timestamp: number;
|
||||
payload: string;
|
||||
}
|
||||
|
||||
export interface CachedBranchQuery {
|
||||
type?: Entry["type"];
|
||||
customType?: string;
|
||||
stopAtType?: Entry["type"];
|
||||
stopAtId?: string;
|
||||
cursor?: { afterSeq: number };
|
||||
order?: "newestFirst" | "oldestFirst";
|
||||
limit?: number;
|
||||
}
|
||||
|
||||
interface BranchPathEntryRow {
|
||||
id: string;
|
||||
seq: number;
|
||||
parent_id: string | null;
|
||||
type: Entry["type"];
|
||||
payload: string;
|
||||
}
|
||||
|
||||
export function readCachedBranch(db: SqliteDatabase, sessionId: string, leafId: string) {
|
||||
const membership = sql`SELECT branch_id, entry_seq
|
||||
FROM branch_entries
|
||||
WHERE session_id = ${sessionId} AND entry_id = ${leafId}
|
||||
ORDER BY branch_id
|
||||
LIMIT 1`.get<{ branch_id: string; entry_seq: number }>(db);
|
||||
if (!membership) return undefined;
|
||||
return { branchId: membership.branch_id, leafSeq: membership.entry_seq };
|
||||
}
|
||||
|
||||
export function queryCachedBranchRows(
|
||||
db: SqliteDatabase,
|
||||
sessionId: string,
|
||||
branch: CachedBranch,
|
||||
query: CachedBranchQuery,
|
||||
) {
|
||||
const oldestFirst = query.order === "oldestFirst";
|
||||
const stopPredicates: ReturnType<typeof sql>[] = [];
|
||||
if (query.stopAtType !== undefined) stopPredicates.push(sql`stop.entry_type = ${query.stopAtType}`);
|
||||
if (query.stopAtId !== undefined) stopPredicates.push(sql`stop.entry_id = ${query.stopAtId}`);
|
||||
|
||||
const aggregate = oldestFirst ? sql`MIN` : sql`MAX`;
|
||||
const boundaryComparison = oldestFirst ? sql`<=` : sql`>=`;
|
||||
const cursorComparison = oldestFirst ? sql`>` : sql`<`;
|
||||
const direction = oldestFirst ? sql`ASC` : sql`DESC`;
|
||||
const boundary =
|
||||
stopPredicates.length === 0
|
||||
? sql``
|
||||
: sql`SELECT ${aggregate}(stop.entry_seq)
|
||||
FROM branch_entries AS stop
|
||||
WHERE stop.session_id = ${sessionId}
|
||||
AND stop.branch_id = ${branch.branchId}
|
||||
AND stop.entry_seq <= ${branch.leafSeq}
|
||||
AND (${joinSqlFragments(stopPredicates, " OR ")})`;
|
||||
|
||||
const predicates = [
|
||||
sql`b.session_id = ${sessionId}`,
|
||||
sql`b.branch_id = ${branch.branchId}`,
|
||||
sql`b.entry_seq <= ${branch.leafSeq}`,
|
||||
];
|
||||
if (stopPredicates.length > 0) {
|
||||
predicates.push(
|
||||
sql`b.entry_seq ${boundaryComparison} COALESCE((${boundary}), ${oldestFirst ? branch.leafSeq : 0})`,
|
||||
);
|
||||
}
|
||||
if (query.cursor !== undefined) predicates.push(sql`b.entry_seq ${cursorComparison} ${query.cursor.afterSeq}`);
|
||||
if (query.type !== undefined) predicates.push(sql`b.entry_type = ${query.type}`);
|
||||
if (query.customType !== undefined) predicates.push(sql`b.custom_type = ${query.customType}`);
|
||||
const limit = query.limit === undefined ? sql`` : sql` LIMIT ${query.limit}`;
|
||||
|
||||
return sql`SELECT e.session_id, e.id, e.seq AS entry_seq, e.parent_id, e.type, e.timestamp, e.payload
|
||||
FROM branch_entries AS b
|
||||
JOIN entries AS e ON e.session_id = b.session_id AND e.id = b.entry_id
|
||||
WHERE ${joinSqlFragments(predicates, " AND ")}
|
||||
ORDER BY b.entry_seq ${direction}${limit}`.all<CachedBranchEntryRow>(db);
|
||||
}
|
||||
|
||||
export function deleteBranchEntries(db: SqliteDatabase, sessionId: string) {
|
||||
sql`DELETE FROM branch_entries WHERE session_id = ${sessionId}`.run(db);
|
||||
}
|
||||
|
||||
export function insertBranchEntry(
|
||||
db: SqliteDatabase,
|
||||
sessionId: string,
|
||||
branchId: string,
|
||||
entryId: string,
|
||||
entrySeq: number,
|
||||
entryType: string,
|
||||
customType: string | null,
|
||||
) {
|
||||
sql`INSERT INTO branch_entries
|
||||
(session_id, branch_id, entry_id, entry_seq, entry_type, custom_type)
|
||||
VALUES (${sessionId}, ${branchId}, ${entryId}, ${entrySeq}, ${entryType}, ${customType})`.run(db);
|
||||
}
|
||||
|
||||
function customTypeFromPayload(row: BranchPathEntryRow): string | null {
|
||||
if (row.type !== "custom") return null;
|
||||
try {
|
||||
const payload = JSON.parse(row.payload) as unknown;
|
||||
if (typeof payload !== "object" || payload === null || Array.isArray(payload)) {
|
||||
throw new Error("Payload is not an object");
|
||||
}
|
||||
const customType = (payload as { customType?: unknown }).customType;
|
||||
if (typeof customType !== "string") throw new Error("Invalid custom payload");
|
||||
return customType;
|
||||
} catch (error) {
|
||||
throw new SessionError(
|
||||
"invalid_entry",
|
||||
`Invalid SQLite session entry ${row.id}: failed to decode entry ${row.id}`,
|
||||
error instanceof Error ? error : undefined,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
export function insertBranchEntriesForPath(db: SqliteDatabase, sessionId: string, branchId: string, leafId: string) {
|
||||
const path: BranchPathEntryRow[] = [];
|
||||
const seen = new Set<string>();
|
||||
let entryId: string | null = leafId;
|
||||
|
||||
while (entryId !== null) {
|
||||
if (seen.has(entryId)) throw new SessionError("invalid_entry", `Entry parent cycle at ${entryId}`);
|
||||
seen.add(entryId);
|
||||
const row: BranchPathEntryRow | undefined = sql`SELECT id, seq, parent_id, type, payload
|
||||
FROM entries
|
||||
WHERE session_id = ${sessionId} AND id = ${entryId}`.get<BranchPathEntryRow>(db);
|
||||
if (!row) throw new SessionError("invalid_entry", `Entry ${entryId} not found`);
|
||||
path.push(row);
|
||||
entryId = row.parent_id;
|
||||
}
|
||||
|
||||
for (const row of path.reverse()) {
|
||||
insertBranchEntry(db, sessionId, branchId, row.id, row.seq, row.type, customTypeFromPayload(row));
|
||||
}
|
||||
}
|
||||
|
||||
export function readBranchContainingEntry(db: SqliteDatabase, sessionId: string, entryId: string) {
|
||||
const row = sql`SELECT b.branch_id, b.entry_seq
|
||||
FROM branch_entries AS b
|
||||
WHERE b.session_id = ${sessionId} AND b.entry_id = ${entryId}
|
||||
ORDER BY b.branch_id
|
||||
LIMIT 1`.get<{ branch_id: string; entry_seq: number }>(db);
|
||||
return row === undefined ? undefined : { branchId: row.branch_id, entrySeq: row.entry_seq };
|
||||
}
|
||||
|
||||
export function copyBranchEntriesThroughSeq(
|
||||
db: SqliteDatabase,
|
||||
sessionId: string,
|
||||
targetBranchId: string,
|
||||
sourceBranchId: string,
|
||||
throughSeq: number,
|
||||
) {
|
||||
sql`INSERT INTO branch_entries (session_id, branch_id, entry_id, entry_seq, entry_type, custom_type)
|
||||
SELECT session_id, ${targetBranchId}, entry_id, entry_seq, entry_type, custom_type
|
||||
FROM branch_entries
|
||||
WHERE session_id = ${sessionId} AND branch_id = ${sourceBranchId} AND entry_seq <= ${throughSeq}`.run(db);
|
||||
}
|
||||
|
|
@ -1,35 +0,0 @@
|
|||
import { sql } from "../sql.ts";
|
||||
import type { SqliteDatabase } from "../types.ts";
|
||||
|
||||
export function readBranchTipIds(db: SqliteDatabase, sessionId: string) {
|
||||
return sql`SELECT tip_id FROM branch_tips WHERE session_id = ${sessionId} ORDER BY tip_id`
|
||||
.all<{ tip_id: string }>(db)
|
||||
.map((row) => row.tip_id);
|
||||
}
|
||||
|
||||
export function readBranchTipBranchId(db: SqliteDatabase, sessionId: string, tipId: string) {
|
||||
const tip = sql`SELECT branch_id FROM branch_tips WHERE session_id = ${sessionId} AND tip_id = ${tipId}`.get<{
|
||||
branch_id: string;
|
||||
}>(db);
|
||||
return tip?.branch_id;
|
||||
}
|
||||
|
||||
export function insertBranchTip(db: SqliteDatabase, sessionId: string, tipId: string, branchId: string) {
|
||||
sql`INSERT INTO branch_tips (session_id, tip_id, branch_id) VALUES (${sessionId}, ${tipId}, ${branchId})`.run(db);
|
||||
}
|
||||
|
||||
export function updateBranchTip(
|
||||
db: SqliteDatabase,
|
||||
sessionId: string,
|
||||
branchId: string,
|
||||
oldTipId: string,
|
||||
newTipId: string,
|
||||
) {
|
||||
const result = sql`UPDATE branch_tips SET tip_id = ${newTipId}
|
||||
WHERE session_id = ${sessionId} AND branch_id = ${branchId} AND tip_id = ${oldTipId}`.run(db);
|
||||
return result.changes === 1;
|
||||
}
|
||||
|
||||
export function deleteBranchTips(db: SqliteDatabase, sessionId: string) {
|
||||
sql`DELETE FROM branch_tips WHERE session_id = ${sessionId}`.run(db);
|
||||
}
|
||||
|
|
@ -1,78 +0,0 @@
|
|||
import type { Entry, EntryOrder } from "@earendil-works/pi-agent-core";
|
||||
import { sql } from "../sql.ts";
|
||||
import type { SqliteDatabase } from "../types.ts";
|
||||
|
||||
export interface EntryRow {
|
||||
session_id: string;
|
||||
seq: number;
|
||||
id: string;
|
||||
parent_id: string | null;
|
||||
type: Entry["type"];
|
||||
timestamp: number;
|
||||
payload: string;
|
||||
}
|
||||
|
||||
export interface NewEntryRow {
|
||||
seq: number;
|
||||
id: string;
|
||||
parentId: string | null;
|
||||
type: Entry["type"];
|
||||
timestamp: number;
|
||||
payload: string;
|
||||
}
|
||||
|
||||
export function entryPayload(entry: Entry): Record<string, unknown> {
|
||||
const { type: _type, id: _id, seq: _seq, parentId: _parentId, timestamp: _timestamp, ...payload } = entry;
|
||||
return payload;
|
||||
}
|
||||
|
||||
export function insertEntryRow(db: SqliteDatabase, sessionId: string, entry: NewEntryRow) {
|
||||
sql`INSERT INTO entries (session_id, id, seq, parent_id, type, timestamp, payload)
|
||||
VALUES (${sessionId}, ${entry.id}, ${entry.seq}, ${entry.parentId}, ${entry.type}, ${entry.timestamp}, ${entry.payload})`.run(
|
||||
db,
|
||||
);
|
||||
}
|
||||
|
||||
export function readEntryRow(db: SqliteDatabase, sessionId: string, entryId: string) {
|
||||
return sql`SELECT session_id, seq, id, parent_id, type, timestamp, payload
|
||||
FROM entries
|
||||
WHERE session_id = ${sessionId} AND id = ${entryId}`.get<EntryRow>(db);
|
||||
}
|
||||
|
||||
export function readEntryRows(
|
||||
db: SqliteDatabase,
|
||||
sessionId: string,
|
||||
options: {
|
||||
afterSeq?: number;
|
||||
cursor?: { afterSeq: number };
|
||||
type?: Entry["type"];
|
||||
order?: EntryOrder;
|
||||
limit?: number;
|
||||
} = {},
|
||||
) {
|
||||
const oldestFirst = options.order === "oldestFirst";
|
||||
const after = options.afterSeq === undefined ? sql`` : sql` AND seq > ${options.afterSeq}`;
|
||||
const cursor =
|
||||
options.cursor === undefined
|
||||
? sql``
|
||||
: oldestFirst
|
||||
? sql` AND seq > ${options.cursor.afterSeq}`
|
||||
: sql` AND seq < ${options.cursor.afterSeq}`;
|
||||
const type = options.type === undefined ? sql`` : sql` AND type = ${options.type}`;
|
||||
const direction = oldestFirst ? sql`ASC` : sql`DESC`;
|
||||
const limit = options.limit === undefined ? sql`` : sql` LIMIT ${options.limit}`;
|
||||
return sql`SELECT session_id, seq, id, parent_id, type, timestamp, payload
|
||||
FROM entries
|
||||
WHERE session_id = ${sessionId}${after}${cursor}${type}
|
||||
ORDER BY seq ${direction}${limit}`.all<EntryRow>(db);
|
||||
}
|
||||
|
||||
export function idExistsInEntries(db: SqliteDatabase, sessionId: string, id: string) {
|
||||
return !!sql`SELECT 1 AS found FROM entries WHERE session_id = ${sessionId} AND id = ${id} LIMIT 1`.get<{
|
||||
found: number;
|
||||
}>(db);
|
||||
}
|
||||
|
||||
export function deleteEntryRows(db: SqliteDatabase, sessionId: string) {
|
||||
sql`DELETE FROM entries WHERE session_id = ${sessionId}`.run(db);
|
||||
}
|
||||
|
|
@ -1,64 +0,0 @@
|
|||
import { sql } from "../sql.ts";
|
||||
import type { SqliteDatabase } from "../types.ts";
|
||||
|
||||
export interface FactRow {
|
||||
session_id: string;
|
||||
seq: number;
|
||||
kind: string;
|
||||
key: string | null;
|
||||
value: string | null;
|
||||
}
|
||||
|
||||
export function appendFact(
|
||||
db: SqliteDatabase,
|
||||
sessionId: string,
|
||||
seq: number,
|
||||
kind: string,
|
||||
key: string | null,
|
||||
value: string | null,
|
||||
) {
|
||||
sql`INSERT INTO facts (session_id, seq, kind, key, value) VALUES (${sessionId}, ${seq}, ${kind}, ${key}, ${value})`.run(
|
||||
db,
|
||||
);
|
||||
}
|
||||
|
||||
export function readLatestFact(db: SqliteDatabase, sessionId: string, kind: string, key: string | null) {
|
||||
return sql`SELECT session_id, seq, kind, key, value
|
||||
FROM facts INDEXED BY idx_facts_session_kind_key_seq
|
||||
WHERE session_id = ${sessionId} AND kind = ${kind} AND key IS ${key}
|
||||
ORDER BY seq DESC
|
||||
LIMIT 1`.get<FactRow>(db);
|
||||
}
|
||||
|
||||
export function readLatestLabelFacts(db: SqliteDatabase, sessionId: string) {
|
||||
return sql`SELECT f.key, f.value
|
||||
FROM facts AS f INDEXED BY idx_facts_session_kind_key_seq
|
||||
WHERE f.session_id = ${sessionId}
|
||||
AND f.kind = 'label'
|
||||
AND f.value IS NOT NULL
|
||||
AND f.seq = (
|
||||
SELECT MAX(candidate.seq)
|
||||
FROM facts AS candidate INDEXED BY idx_facts_session_kind_key_seq
|
||||
WHERE candidate.session_id = f.session_id
|
||||
AND candidate.kind = f.kind
|
||||
AND candidate.key IS f.key
|
||||
)
|
||||
ORDER BY f.key`.all<{ key: string; value: string }>(db);
|
||||
}
|
||||
|
||||
export function readFactRows(
|
||||
db: SqliteDatabase,
|
||||
sessionId: string,
|
||||
options: { afterSeq?: number; limit?: number } = {},
|
||||
) {
|
||||
const after = options.afterSeq === undefined ? sql`` : sql` AND seq > ${options.afterSeq}`;
|
||||
const limit = options.limit === undefined ? sql`` : sql` LIMIT ${options.limit}`;
|
||||
return sql`SELECT session_id, seq, kind, key, value
|
||||
FROM facts
|
||||
WHERE session_id = ${sessionId}${after}
|
||||
ORDER BY seq${limit}`.all<FactRow>(db);
|
||||
}
|
||||
|
||||
export function deleteFactRows(db: SqliteDatabase, sessionId: string) {
|
||||
sql`DELETE FROM facts WHERE session_id = ${sessionId}`.run(db);
|
||||
}
|
||||
|
|
@ -1,124 +0,0 @@
|
|||
import { SessionError } from "@earendil-works/pi-agent-core";
|
||||
import { sql } from "../sql.ts";
|
||||
import type { SqliteDatabase } from "../types.ts";
|
||||
|
||||
export interface LaneRow {
|
||||
session_id: string;
|
||||
lane: string;
|
||||
leaf_id: string | null;
|
||||
open_operation_id: string | null;
|
||||
}
|
||||
|
||||
export interface LaneMoveRow {
|
||||
session_id: string;
|
||||
seq: number;
|
||||
lane: string;
|
||||
leaf_id: string | null;
|
||||
}
|
||||
|
||||
export function createInitialLane(db: SqliteDatabase, sessionId: string, lane = "main", leafId: string | null = null) {
|
||||
sql`INSERT INTO lanes (session_id, lane, leaf_id, open_operation_id)
|
||||
VALUES (${sessionId}, ${lane}, ${leafId}, NULL)`.run(db);
|
||||
}
|
||||
|
||||
export function readLanes(db: SqliteDatabase, sessionId: string) {
|
||||
const rows = sql`SELECT
|
||||
l.session_id,
|
||||
l.lane,
|
||||
l.leaf_id,
|
||||
l.open_operation_id,
|
||||
(l.leaf_id IS NULL OR EXISTS (
|
||||
SELECT 1 FROM entries AS e WHERE e.session_id = l.session_id AND e.id = l.leaf_id
|
||||
)) AS leaf_exists
|
||||
FROM lanes AS l
|
||||
WHERE l.session_id = ${sessionId}
|
||||
ORDER BY l.lane`.all<LaneRow & { leaf_exists: number }>(db);
|
||||
for (const row of rows) {
|
||||
if (row.leaf_exists === 0) {
|
||||
throw new SessionError("storage", `Lane ${row.lane} points at missing entry ${row.leaf_id}`);
|
||||
}
|
||||
}
|
||||
return rows.map(({ session_id, lane, leaf_id, open_operation_id }) => ({
|
||||
session_id,
|
||||
lane,
|
||||
leaf_id,
|
||||
open_operation_id,
|
||||
}));
|
||||
}
|
||||
|
||||
export function readLane(db: SqliteDatabase, sessionId: string, lane: string) {
|
||||
return sql`SELECT session_id, lane, leaf_id, open_operation_id
|
||||
FROM lanes
|
||||
WHERE session_id = ${sessionId} AND lane = ${lane}`.get<LaneRow>(db);
|
||||
}
|
||||
|
||||
export function readLaneHead(db: SqliteDatabase, sessionId: string, lane: string) {
|
||||
const row = sql`SELECT
|
||||
l.leaf_id,
|
||||
(l.leaf_id IS NULL OR EXISTS (
|
||||
SELECT 1 FROM entries AS e WHERE e.session_id = l.session_id AND e.id = l.leaf_id
|
||||
)) AS leaf_exists
|
||||
FROM lanes AS l
|
||||
WHERE l.session_id = ${sessionId} AND l.lane = ${lane}`.get<{
|
||||
leaf_id: string | null;
|
||||
leaf_exists: number;
|
||||
}>(db);
|
||||
if (!row) throw new SessionError("invalid_lane", `Lane not found: ${lane}`);
|
||||
if (row.leaf_exists === 0) throw new SessionError("storage", `Entry ${row.leaf_id} not found`);
|
||||
return { leafId: row.leaf_id };
|
||||
}
|
||||
|
||||
export function createLane(db: SqliteDatabase, sessionId: string, seq: number, lane: string, leafId: string | null) {
|
||||
sql`INSERT INTO lanes (session_id, lane, leaf_id, open_operation_id)
|
||||
VALUES (${sessionId}, ${lane}, ${leafId}, NULL)`.run(db);
|
||||
appendLaneMove(db, sessionId, seq, lane, leafId);
|
||||
}
|
||||
|
||||
export function moveLane(db: SqliteDatabase, sessionId: string, seq: number, lane: string, leafId: string | null) {
|
||||
const result = sql`UPDATE lanes SET leaf_id = ${leafId} WHERE session_id = ${sessionId} AND lane = ${lane}`.run(db);
|
||||
if (result.changes !== 1) throw new SessionError("invalid_lane", `Lane not found: ${lane}`);
|
||||
appendLaneMove(db, sessionId, seq, lane, leafId);
|
||||
}
|
||||
|
||||
export function setLaneLeaf(db: SqliteDatabase, sessionId: string, lane: string, leafId: string | null) {
|
||||
const result = sql`UPDATE lanes SET leaf_id = ${leafId} WHERE session_id = ${sessionId} AND lane = ${lane}`.run(db);
|
||||
if (result.changes !== 1) throw new SessionError("invalid_lane", `Lane not found: ${lane}`);
|
||||
}
|
||||
|
||||
export function startLaneOperation(db: SqliteDatabase, sessionId: string, lane: string, runId: string) {
|
||||
const result = sql`UPDATE lanes SET open_operation_id = ${runId}
|
||||
WHERE session_id = ${sessionId} AND lane = ${lane} AND open_operation_id IS NULL`.run(db);
|
||||
if (result.changes === 1) return;
|
||||
const current = readLane(db, sessionId, lane);
|
||||
if (!current) throw new SessionError("invalid_lane", `Lane not found: ${lane}`);
|
||||
throw new SessionError("storage", `Lane ${lane} already has an open operation ${current.open_operation_id}`);
|
||||
}
|
||||
|
||||
export function finishLaneOperation(db: SqliteDatabase, sessionId: string, lane: string, runId: string) {
|
||||
sql`UPDATE lanes SET open_operation_id = NULL
|
||||
WHERE session_id = ${sessionId} AND lane = ${lane} AND open_operation_id = ${runId}`.run(db);
|
||||
}
|
||||
|
||||
export function readLaneMoveRows(
|
||||
db: SqliteDatabase,
|
||||
sessionId: string,
|
||||
options: { afterSeq?: number; limit?: number } = {},
|
||||
) {
|
||||
const after = options.afterSeq === undefined ? sql`` : sql` AND seq > ${options.afterSeq}`;
|
||||
const limit = options.limit === undefined ? sql`` : sql` LIMIT ${options.limit}`;
|
||||
return sql`SELECT session_id, seq, lane, leaf_id
|
||||
FROM lane_moves
|
||||
WHERE session_id = ${sessionId}${after}
|
||||
ORDER BY seq${limit}`.all<LaneMoveRow>(db);
|
||||
}
|
||||
|
||||
export function deleteLaneRows(db: SqliteDatabase, sessionId: string) {
|
||||
sql`DELETE FROM lane_moves WHERE session_id = ${sessionId}`.run(db);
|
||||
sql`DELETE FROM lanes WHERE session_id = ${sessionId}`.run(db);
|
||||
}
|
||||
|
||||
function appendLaneMove(db: SqliteDatabase, sessionId: string, seq: number, lane: string, leafId: string | null) {
|
||||
sql`INSERT INTO lane_moves (session_id, seq, lane, leaf_id) VALUES (${sessionId}, ${seq}, ${lane}, ${leafId})`.run(
|
||||
db,
|
||||
);
|
||||
}
|
||||
|
|
@ -1,95 +0,0 @@
|
|||
import { SessionError } from "@earendil-works/pi-agent-core";
|
||||
import { joinSqlFragments, sql } from "../sql.ts";
|
||||
import type { SqliteDatabase } from "../types.ts";
|
||||
|
||||
export interface RecordRow {
|
||||
session_id: string;
|
||||
seq: number;
|
||||
id: string;
|
||||
lane: string;
|
||||
run_id: string | null;
|
||||
type: string;
|
||||
op_kind: string | null;
|
||||
timestamp: number;
|
||||
payload: string;
|
||||
}
|
||||
|
||||
export interface NewRecordRow {
|
||||
seq: number;
|
||||
id: string;
|
||||
lane: string;
|
||||
runId?: string;
|
||||
type: string;
|
||||
opKind?: string;
|
||||
timestamp: number;
|
||||
payload: string;
|
||||
}
|
||||
|
||||
export function appendRecordRow(db: SqliteDatabase, sessionId: string, record: NewRecordRow) {
|
||||
sql`INSERT INTO records
|
||||
(session_id, seq, id, lane, run_id, type, op_kind, timestamp, payload)
|
||||
VALUES (${sessionId}, ${record.seq}, ${record.id}, ${record.lane}, ${record.runId ?? null}, ${record.type}, ${record.opKind ?? null}, ${record.timestamp}, ${record.payload})`.run(
|
||||
db,
|
||||
);
|
||||
}
|
||||
|
||||
export function idExistsInRecords(db: SqliteDatabase, sessionId: string, id: string) {
|
||||
return !!sql`SELECT 1 AS found FROM records WHERE session_id = ${sessionId} AND id = ${id} LIMIT 1`.get<{
|
||||
found: number;
|
||||
}>(db);
|
||||
}
|
||||
|
||||
export function deleteRecordRows(db: SqliteDatabase, sessionId: string) {
|
||||
sql`DELETE FROM records WHERE session_id = ${sessionId}`.run(db);
|
||||
}
|
||||
|
||||
export function readRecordRows(
|
||||
db: SqliteDatabase,
|
||||
sessionId: string,
|
||||
query: {
|
||||
lane?: string;
|
||||
type?: string;
|
||||
runId?: string;
|
||||
operationKind?: string;
|
||||
afterSeq?: number;
|
||||
order?: "newestFirst" | "oldestFirst";
|
||||
limit?: number;
|
||||
} = {},
|
||||
) {
|
||||
const predicates = [sql`session_id = ${sessionId}`];
|
||||
if (query.lane !== undefined) predicates.push(sql`lane = ${query.lane}`);
|
||||
if (query.type !== undefined) predicates.push(sql`type = ${query.type}`);
|
||||
if (query.runId !== undefined) predicates.push(sql`run_id = ${query.runId}`);
|
||||
if (query.operationKind !== undefined) predicates.push(sql`op_kind = ${query.operationKind}`);
|
||||
if (query.afterSeq !== undefined) predicates.push(sql`seq > ${query.afterSeq}`);
|
||||
const direction = query.order === "oldestFirst" ? sql`ASC` : sql`DESC`;
|
||||
const limit = query.limit === undefined ? sql`` : sql` LIMIT ${query.limit}`;
|
||||
return sql`SELECT session_id, seq, id, lane, run_id, type, op_kind, timestamp, payload
|
||||
FROM records
|
||||
WHERE ${joinSqlFragments(predicates, " AND ")}
|
||||
ORDER BY seq ${direction}${limit}`.all<RecordRow>(db);
|
||||
}
|
||||
|
||||
export function readOpenOperationRows(
|
||||
db: SqliteDatabase,
|
||||
sessionId: string,
|
||||
lane: string,
|
||||
_options: { limit?: number } = {},
|
||||
): RecordRow[] {
|
||||
const laneRow = sql`SELECT open_operation_id FROM lanes WHERE session_id = ${sessionId} AND lane = ${lane}`.get<{
|
||||
open_operation_id: string | null;
|
||||
}>(db);
|
||||
if (!laneRow?.open_operation_id) return [];
|
||||
|
||||
const record = sql`SELECT session_id, seq, id, lane, run_id, type, op_kind, timestamp, payload
|
||||
FROM records
|
||||
WHERE session_id = ${sessionId}
|
||||
AND id = ${laneRow.open_operation_id}`.get<RecordRow>(db);
|
||||
if (!record) {
|
||||
throw new SessionError("storage", `Lane ${lane} points at missing open operation ${laneRow.open_operation_id}`);
|
||||
}
|
||||
if (record.lane !== lane || record.type !== "operation_started") {
|
||||
throw new SessionError("storage", `Lane ${lane} points at invalid open operation ${laneRow.open_operation_id}`);
|
||||
}
|
||||
return [record];
|
||||
}
|
||||
|
|
@ -1,29 +0,0 @@
|
|||
import { SessionError } from "@earendil-works/pi-agent-core";
|
||||
import { sql } from "../sql.ts";
|
||||
import type { SqliteDatabase } from "../types.ts";
|
||||
|
||||
export function createSequence(db: SqliteDatabase, sessionId: string, nextSeq = 1) {
|
||||
sql`INSERT INTO session_sequences (session_id, next_seq) VALUES (${sessionId}, ${nextSeq})`.run(db);
|
||||
}
|
||||
|
||||
export function getNextSequence(db: SqliteDatabase, sessionId: string) {
|
||||
const sequenceRow = sql`SELECT next_seq FROM session_sequences WHERE session_id = ${sessionId}`.get<{
|
||||
next_seq: number;
|
||||
}>(db);
|
||||
if (!sequenceRow) {
|
||||
throw new SessionError("storage", `Missing sequence row for session ${sessionId}`);
|
||||
}
|
||||
return sequenceRow.next_seq;
|
||||
}
|
||||
|
||||
export function setNextSequence(db: SqliteDatabase, sessionId: string, nextSeq: number) {
|
||||
sql`UPDATE session_sequences SET next_seq = ${nextSeq} WHERE session_id = ${sessionId}`.run(db);
|
||||
}
|
||||
|
||||
export function advanceSequence(db: SqliteDatabase, sessionId: string, seq: number) {
|
||||
setNextSequence(db, sessionId, seq + 1);
|
||||
}
|
||||
|
||||
export function deleteSequence(db: SqliteDatabase, sessionId: string) {
|
||||
sql`DELETE FROM session_sequences WHERE session_id = ${sessionId}`.run(db);
|
||||
}
|
||||
|
|
@ -1,54 +0,0 @@
|
|||
import { SessionError, type SessionStats } from "@earendil-works/pi-agent-core";
|
||||
import type { Usage } from "@earendil-works/pi-ai";
|
||||
import { sql } from "../sql.ts";
|
||||
import type { SqliteDatabase } from "../types.ts";
|
||||
|
||||
export interface SessionStatsRow {
|
||||
session_id: string;
|
||||
message_count: number;
|
||||
cached_tokens: number;
|
||||
uncached_tokens: number;
|
||||
total_tokens: number;
|
||||
cost_total: number;
|
||||
}
|
||||
|
||||
export function createStats(db: SqliteDatabase, sessionId: string, messageCount = 0): void {
|
||||
sql`INSERT INTO session_stats
|
||||
(session_id, message_count, cached_tokens, uncached_tokens, total_tokens, cost_total)
|
||||
VALUES (${sessionId}, ${messageCount}, 0, 0, 0, 0)`.run(db);
|
||||
}
|
||||
|
||||
export function readStats(db: SqliteDatabase, sessionId: string): SessionStats {
|
||||
const row = sql`SELECT session_id, message_count, cached_tokens, uncached_tokens, total_tokens, cost_total
|
||||
FROM session_stats
|
||||
WHERE session_id = ${sessionId}`.get<SessionStatsRow>(db);
|
||||
if (!row) throw new SessionError("storage", `Missing stats row for session ${sessionId}`);
|
||||
return {
|
||||
messageCount: row.message_count,
|
||||
cachedTokens: row.cached_tokens,
|
||||
uncachedTokens: row.uncached_tokens,
|
||||
totalTokens: row.total_tokens,
|
||||
costTotal: row.cost_total,
|
||||
};
|
||||
}
|
||||
|
||||
export function incrementMessageCount(db: SqliteDatabase, sessionId: string): void {
|
||||
const result = sql`UPDATE session_stats SET message_count = message_count + 1 WHERE session_id = ${sessionId}`.run(
|
||||
db,
|
||||
);
|
||||
if (result.changes !== 1) throw new SessionError("storage", `Missing stats row for session ${sessionId}`);
|
||||
}
|
||||
|
||||
export function addUsageToStats(db: SqliteDatabase, sessionId: string, usage: Usage): void {
|
||||
const result = sql`UPDATE session_stats
|
||||
SET cached_tokens = cached_tokens + ${usage.cacheRead},
|
||||
uncached_tokens = uncached_tokens + ${usage.input + usage.cacheWrite},
|
||||
total_tokens = total_tokens + ${usage.totalTokens},
|
||||
cost_total = cost_total + ${usage.cost.total}
|
||||
WHERE session_id = ${sessionId}`.run(db);
|
||||
if (result.changes !== 1) throw new SessionError("storage", `Missing stats row for session ${sessionId}`);
|
||||
}
|
||||
|
||||
export function deleteStats(db: SqliteDatabase, sessionId: string): void {
|
||||
sql`DELETE FROM session_stats WHERE session_id = ${sessionId}`.run(db);
|
||||
}
|
||||
|
|
@ -1,131 +0,0 @@
|
|||
import { assertJsonSerializable, SessionError } from "@earendil-works/pi-agent-core";
|
||||
import { sql } from "../sql.ts";
|
||||
import type { SqliteDatabase, SqliteSessionMetadata } from "../types.ts";
|
||||
|
||||
export interface SessionRow {
|
||||
id: string;
|
||||
created_at: number;
|
||||
metadata: string | null;
|
||||
cwd: string;
|
||||
parent_session_id: string | null;
|
||||
has_session_name: number;
|
||||
session_name: string | null;
|
||||
}
|
||||
|
||||
export interface NewSessionRow {
|
||||
id: string;
|
||||
createdAt: number;
|
||||
cwd: string;
|
||||
parentSessionId?: string;
|
||||
metadata?: Record<string, unknown>;
|
||||
}
|
||||
|
||||
function parseMetadata(metadata: string | null, sessionId: string): Record<string, unknown> | undefined {
|
||||
if (metadata === null) return undefined;
|
||||
let parsed: unknown;
|
||||
try {
|
||||
parsed = JSON.parse(metadata);
|
||||
} catch (error) {
|
||||
throw new SessionError(
|
||||
"storage",
|
||||
`Invalid SQLite session ${sessionId}: metadata is not valid JSON`,
|
||||
error instanceof Error ? error : undefined,
|
||||
);
|
||||
}
|
||||
if (typeof parsed !== "object" || parsed === null || Array.isArray(parsed)) {
|
||||
throw new SessionError("storage", `Invalid SQLite session ${sessionId}: metadata must be an object`);
|
||||
}
|
||||
return parsed as Record<string, unknown>;
|
||||
}
|
||||
|
||||
export function sessionExists(db: SqliteDatabase, sessionId: string) {
|
||||
return !!sql`SELECT 1 AS found FROM sessions WHERE id = ${sessionId}`.get<{ found: number }>(db);
|
||||
}
|
||||
|
||||
function serializeMetadata(metadata: Record<string, unknown> | undefined): string | null {
|
||||
if (metadata === undefined) return null;
|
||||
if (typeof metadata !== "object" || metadata === null || Array.isArray(metadata)) {
|
||||
throw new SessionError("invalid_payload", "SQLite session metadata must be an object");
|
||||
}
|
||||
assertJsonSerializable(metadata);
|
||||
return JSON.stringify(metadata);
|
||||
}
|
||||
|
||||
export function insertSessionRow(db: SqliteDatabase, session: NewSessionRow) {
|
||||
sql`INSERT INTO sessions (id, created_at, metadata, cwd, parent_session_id)
|
||||
VALUES (${session.id}, ${session.createdAt}, ${serializeMetadata(session.metadata)}, ${session.cwd}, ${session.parentSessionId ?? null})`.run(
|
||||
db,
|
||||
);
|
||||
}
|
||||
|
||||
export function readSessionRow(db: SqliteDatabase, sessionId: string) {
|
||||
return sql`SELECT s.id, s.created_at, s.metadata, s.cwd, s.parent_session_id,
|
||||
name_fact.seq IS NOT NULL AS has_session_name,
|
||||
name_fact.value AS session_name
|
||||
FROM sessions AS s
|
||||
LEFT JOIN facts AS name_fact
|
||||
ON name_fact.session_id = s.id
|
||||
AND name_fact.kind = 'name'
|
||||
AND name_fact.key IS NULL
|
||||
AND name_fact.seq = (
|
||||
SELECT MAX(f.seq)
|
||||
FROM facts AS f
|
||||
WHERE f.session_id = s.id AND f.kind = 'name' AND f.key IS NULL
|
||||
)
|
||||
WHERE s.id = ${sessionId}`.get<SessionRow>(db);
|
||||
}
|
||||
|
||||
export function readSessionRows(db: SqliteDatabase, options: { cwd?: string } = {}) {
|
||||
const where = options.cwd === undefined ? sql`` : sql`WHERE s.cwd = ${options.cwd}`;
|
||||
return sql`SELECT s.id, s.created_at, s.metadata, s.cwd, s.parent_session_id,
|
||||
name_fact.seq IS NOT NULL AS has_session_name,
|
||||
name_fact.value AS session_name
|
||||
FROM sessions AS s
|
||||
LEFT JOIN facts AS name_fact
|
||||
ON name_fact.session_id = s.id
|
||||
AND name_fact.kind = 'name'
|
||||
AND name_fact.key IS NULL
|
||||
AND name_fact.seq = (
|
||||
SELECT MAX(f.seq)
|
||||
FROM facts AS f
|
||||
WHERE f.session_id = s.id AND f.kind = 'name' AND f.key IS NULL
|
||||
)
|
||||
${where}
|
||||
ORDER BY s.created_at DESC`.all<SessionRow>(db);
|
||||
}
|
||||
|
||||
export function deleteSessionRow(db: SqliteDatabase, sessionId: string) {
|
||||
sql`DELETE FROM sessions WHERE id = ${sessionId}`.run(db);
|
||||
}
|
||||
|
||||
function parseSessionName(value: string | null, sessionId: string): string | undefined {
|
||||
if (value === null) return undefined;
|
||||
let parsed: unknown;
|
||||
try {
|
||||
parsed = JSON.parse(value);
|
||||
} catch (error) {
|
||||
throw new SessionError(
|
||||
"storage",
|
||||
`Invalid SQLite session ${sessionId}: name is not valid JSON`,
|
||||
error instanceof Error ? error : undefined,
|
||||
);
|
||||
}
|
||||
if (typeof parsed !== "string") {
|
||||
throw new SessionError("storage", `Invalid SQLite session ${sessionId}: name must be a string`);
|
||||
}
|
||||
return parsed;
|
||||
}
|
||||
|
||||
export function decodeSessionMetadata(row: SessionRow, path: string): SqliteSessionMetadata {
|
||||
const metadata = parseMetadata(row.metadata, row.id);
|
||||
const name = row.has_session_name === 0 ? undefined : parseSessionName(row.session_name, row.id);
|
||||
return {
|
||||
id: row.id,
|
||||
createdAt: row.created_at,
|
||||
...(name === undefined ? {} : { name }),
|
||||
cwd: row.cwd,
|
||||
path,
|
||||
parentSessionId: row.parent_session_id ?? undefined,
|
||||
metadata,
|
||||
};
|
||||
}
|
||||
|
|
@ -1,58 +0,0 @@
|
|||
import { sql } from "../sql.ts";
|
||||
import type { SqliteDatabase } from "../types.ts";
|
||||
|
||||
export interface WriterLease {
|
||||
ownerId: string;
|
||||
fence: number;
|
||||
expiresAtMs: number;
|
||||
}
|
||||
|
||||
interface WriterLeaseRow {
|
||||
owner_id: string;
|
||||
fence: number;
|
||||
expires_at_ms: number;
|
||||
}
|
||||
|
||||
export function acquireWriterLease(
|
||||
db: SqliteDatabase,
|
||||
sessionId: string,
|
||||
ownerId: string,
|
||||
now: number,
|
||||
expiresAtMs: number,
|
||||
) {
|
||||
const row = sql`INSERT INTO writer_leases (session_id, owner_id, fence, expires_at_ms)
|
||||
VALUES (${sessionId}, ${ownerId}, 1, ${expiresAtMs})
|
||||
ON CONFLICT(session_id) DO UPDATE SET
|
||||
owner_id = excluded.owner_id,
|
||||
fence = writer_leases.fence + 1,
|
||||
expires_at_ms = excluded.expires_at_ms
|
||||
WHERE writer_leases.expires_at_ms <= ${now}
|
||||
RETURNING owner_id, fence, expires_at_ms`.get<WriterLeaseRow>(db);
|
||||
return row === undefined ? undefined : { ownerId: row.owner_id, fence: row.fence, expiresAtMs: row.expires_at_ms };
|
||||
}
|
||||
|
||||
export function renewWriterLease(
|
||||
db: SqliteDatabase,
|
||||
sessionId: string,
|
||||
lease: WriterLease,
|
||||
now: number,
|
||||
expiresAtMs: number,
|
||||
) {
|
||||
const result = sql`UPDATE writer_leases
|
||||
SET expires_at_ms = ${expiresAtMs}
|
||||
WHERE session_id = ${sessionId}
|
||||
AND owner_id = ${lease.ownerId}
|
||||
AND fence = ${lease.fence}
|
||||
AND expires_at_ms > ${now}`.run(db);
|
||||
if (result.changes === 1) lease.expiresAtMs = expiresAtMs;
|
||||
return result.changes === 1;
|
||||
}
|
||||
|
||||
export function releaseWriterLease(db: SqliteDatabase, sessionId: string, lease: WriterLease) {
|
||||
sql`DELETE FROM writer_leases
|
||||
WHERE session_id = ${sessionId} AND owner_id = ${lease.ownerId} AND fence = ${lease.fence}`.run(db);
|
||||
}
|
||||
|
||||
export function deleteWriterLease(db: SqliteDatabase, sessionId: string) {
|
||||
sql`DELETE FROM writer_leases WHERE session_id = ${sessionId}`.run(db);
|
||||
}
|
||||
|
|
@ -1,10 +1,6 @@
|
|||
import type { FileSystem, SessionCreateOptions, SessionMetadata } from "@earendil-works/pi-agent-core";
|
||||
|
||||
/** Result of a prepared SQLite statement execution. */
|
||||
export interface SqliteRunResult {
|
||||
/** Number of rows changed by the statement. */
|
||||
changes: number;
|
||||
/** Inserted row id when the backend exposes one. */
|
||||
lastInsertRowid?: number;
|
||||
}
|
||||
|
||||
|
|
@ -21,32 +17,10 @@ export interface SqliteDatabase {
|
|||
exec(sql: string): void;
|
||||
prepare(sql: string): SqliteStatement;
|
||||
/** Runs a synchronous write transaction. The callback must not return a promise. */
|
||||
transaction<T>(fn: () => T): T;
|
||||
transaction<T>(callback: () => T): T;
|
||||
close(): void;
|
||||
}
|
||||
|
||||
export interface SqliteDatabaseFactory {
|
||||
open(path: string): Promise<SqliteDatabase>;
|
||||
}
|
||||
|
||||
export interface SqliteSessionMetadata extends SessionMetadata {
|
||||
cwd: string;
|
||||
path: string;
|
||||
parentSessionId?: string;
|
||||
/** Current session name projected from SQLite global facts. */
|
||||
name?: string;
|
||||
/** Opaque application-owned metadata. */
|
||||
metadata?: Record<string, unknown>;
|
||||
}
|
||||
|
||||
export interface SqliteSessionCreateOptions extends SessionCreateOptions {
|
||||
cwd: string;
|
||||
parentSessionId?: string;
|
||||
metadata?: Record<string, unknown>;
|
||||
}
|
||||
|
||||
export interface SqliteSessionListOptions {
|
||||
cwd?: string;
|
||||
}
|
||||
|
||||
export type SqliteSessionRepositoryEnv = Pick<FileSystem, "absolutePath" | "createDir" | "exists">;
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue