From ba6190f82009d3603246f8ba72aa724100b2c934 Mon Sep 17 00:00:00 2001 From: Cristina Poncela Cubeiro <140309543+cristinaponcela@users.noreply.github.com> Date: Wed, 12 Aug 2026 17:06:11 +0200 Subject: [PATCH] feat: session helpers for repo --- .../src/sqlite/session/registers.ts | 11 +++ .../src/sqlite/session/session-row.ts | 82 +++++++++++++++++++ .../src/sqlite/session/writer-lease.ts | 31 +++++++ 3 files changed, 124 insertions(+) create mode 100644 packages/session-backends/sqlite-node/src/sqlite/session/registers.ts create mode 100644 packages/session-backends/sqlite-node/src/sqlite/session/session-row.ts create mode 100644 packages/session-backends/sqlite-node/src/sqlite/session/writer-lease.ts diff --git a/packages/session-backends/sqlite-node/src/sqlite/session/registers.ts b/packages/session-backends/sqlite-node/src/sqlite/session/registers.ts new file mode 100644 index 000000000..b352b960f --- /dev/null +++ b/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); +} diff --git a/packages/session-backends/sqlite-node/src/sqlite/session/session-row.ts b/packages/session-backends/sqlite-node/src/sqlite/session/session-row.ts new file mode 100644 index 000000000..cad6a32f3 --- /dev/null +++ b/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(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); +} diff --git a/packages/session-backends/sqlite-node/src/sqlite/session/writer-lease.ts b/packages/session-backends/sqlite-node/src/sqlite/session/writer-lease.ts new file mode 100644 index 000000000..b25134d9a --- /dev/null +++ b/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(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); +}