feat: session helpers for repo

packages/session-backends/sqlite-node/src/sqlite/session/registers.ts

@@ -0,0 +1,11 @@
+import type { LaneState } from "@earendil-works/pi-agent-core";
+import { sql } from "../sql.ts";
+import type { SqliteDatabase } from "../types.ts";
+
+export function insertInitialMainLaneRegisters(db: SqliteDatabase): void {
+	const laneState = { currentOperationId: null, pendingNextRun: [] } satisfies LaneState;
+	sql`INSERT INTO registers (namespace, key, seq, value)
+		VALUES (${"lane.leaf"}, ${"main"}, ${1}, ${JSON.stringify(null)})`.run(db);
+	sql`INSERT INTO registers (namespace, key, seq, value)
+		VALUES (${"lane.state"}, ${"main"}, ${2}, ${JSON.stringify(laneState)})`.run(db);
+}

packages/session-backends/sqlite-node/src/sqlite/session/session-row.ts

@@ -0,0 +1,82 @@
+import type { SessionMetadata } 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 SessionRow {
+	rowid: number;
+	created_at: number;
+	parent_session_id: string | null;
+	storage_version: number;
+	metadata: string | null;
+	message_count: number;
+	usage_payload: string;
+	next_seq: number;
+}
+
+export interface SqliteSessionMetadata extends SessionMetadata {
+	path: string;
+}
+
+function zeroUsage(): Usage {
+	return {
+		input: 0,
+		output: 0,
+		cacheRead: 0,
+		cacheWrite: 0,
+		totalTokens: 0,
+		cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
+	};
+}
+
+export function readSingleSessionRow(db: SqliteDatabase): SessionRow {
+	const rows = sql`SELECT rowid, created_at, parent_session_id, storage_version, metadata,
+			message_count, usage_payload, next_seq
+		FROM session`.all<SessionRow>(db);
+	if (rows.length !== 1) throw new Error(`Expected exactly one session row, found ${rows.length}`);
+	return rows[0]!;
+}
+
+export function readSessionRowCount(db: SqliteDatabase): number {
+	return sql`SELECT rowid FROM session`.all<{ rowid: number }>(db).length;
+}
+
+export function metadataFromSessionRow(
+	path: string,
+	id: string,
+	row: SessionRow,
+	currentStorageVersion: number,
+): SqliteSessionMetadata {
+	if (row.storage_version > currentStorageVersion) {
+		throw new Error(`SQLite session storage version ${row.storage_version} is newer than ${currentStorageVersion}`);
+	}
+	if (row.storage_version < currentStorageVersion) {
+		throw new Error(`SQLite session storage version ${row.storage_version} requires migrations`);
+	}
+	return {
+		id,
+		createdAt: row.created_at,
+		storageVersion: row.storage_version,
+		...(row.parent_session_id === null ? {} : { parentSessionId: row.parent_session_id }),
+		path,
+	};
+}
+
+export function insertSessionRow(
+	db: SqliteDatabase,
+	metadata: SqliteSessionMetadata,
+	storageVersion: number,
+	nextSeq: number,
+): void {
+	sql`INSERT INTO session
+			(created_at, parent_session_id, storage_version, metadata, message_count, usage_payload, next_seq)
+		VALUES (
+			${metadata.createdAt},
+			${metadata.parentSessionId ?? null},
+			${storageVersion},
+			${null},
+			${0},
+			${JSON.stringify(zeroUsage())},
+			${nextSeq}
+		)`.run(db);
+}

packages/session-backends/sqlite-node/src/sqlite/session/writer-lease.ts

@@ -0,0 +1,31 @@
+import { sql } from "../sql.ts";
+import type { SqliteDatabase } from "../types.ts";
+
+export interface WriterLeaseRow {
+	rowid: number;
+	owner_id: string;
+	fence: number;
+	expires_at_ms: number;
+}
+
+export function claimWriterLease(db: SqliteDatabase, ownerId: string, now: number, leaseMs: number): WriterLeaseRow {
+	const rows = sql`SELECT rowid, owner_id, fence, expires_at_ms FROM writer_lease`.all<WriterLeaseRow>(db);
+	if (rows.length > 1) throw new Error(`Expected at most one writer lease row, found ${rows.length}`);
+	const expiresAt = now + leaseMs;
+	const current = rows[0];
+	if (current === undefined) {
+		sql`INSERT INTO writer_lease (owner_id, fence, expires_at_ms) VALUES (${ownerId}, ${1}, ${expiresAt})`.run(db);
+		return { rowid: 1, owner_id: ownerId, fence: 1, expires_at_ms: expiresAt };
+	}
+	if (current.owner_id !== ownerId && current.expires_at_ms > now) {
+		throw new Error(`SQLite session is already claimed by writer ${current.owner_id}`);
+	}
+	const fence = current.owner_id === ownerId ? current.fence : current.fence + 1;
+	sql`UPDATE writer_lease SET owner_id = ${ownerId}, fence = ${fence}, expires_at_ms = ${expiresAt}
+		WHERE rowid = ${current.rowid}`.run(db);
+	return { rowid: current.rowid, owner_id: ownerId, fence, expires_at_ms: expiresAt };
+}
+
+export function releaseWriterLease(db: SqliteDatabase, ownerId: string, fence: number): void {
+	sql`DELETE FROM writer_lease WHERE owner_id = ${ownerId} AND fence = ${fence}`.run(db);
+}