sscoble/pi-mono
1
0
Fork 0
mirror of https://github.com/badlogic/pi-mono.git synced 2026-11-05 23:37:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

feat(agent): add SQLite SQL template queries

Browse source
This commit is contained in:
Armin Ronacher 2026-08-06 18:10:42 +02:00
parent c03d78bdce
commit 055ecce080
19 changed files with 417 additions and 455 deletions
Download patch file Download diff file Expand all files Collapse all files

4
packages/session-backends/sqlite-node/CHANGELOG.md
View file

@ -2,6 +2,10 @@
## [Unreleased]
### Added
- Added the composable, parameterized `sql` template tag for SQLite queries.
## [0.84.0] - 2026-08-06
### Breaking Changes

7
packages/session-backends/sqlite-node/src/index.ts
View file

@ -1,5 +1,6 @@
import type { SQLInputValue } from "node:sqlite";
import { DatabaseSync } from "node:sqlite";
import { sql } from "./sqlite/sql.ts";
import type { SqliteDatabase, SqliteDatabaseFactory, SqliteRunResult, SqliteStatement } from "./sqlite/types.ts";
function isNamedParameters(value: unknown): value is Record<string, SQLInputValue> {
@ -65,17 +66,17 @@ class NodeSqliteDatabase implements SqliteDatabase {
}
transaction<T>(fn: () => T): T {
this.db.exec("BEGIN IMMEDIATE");
sql`BEGIN IMMEDIATE`.exec(this);
try {
const result = fn();
if (isAsyncResult(result)) {
throw new TypeError("SQLite transaction callbacks must be synchronous");
}
this.db.exec("COMMIT");
sql`COMMIT`.exec(this);
return result;
} catch (error) {
try {
this.db.exec("ROLLBACK");
sql`ROLLBACK`.exec(this);
} catch {
// Ignore rollback errors to rethrow original error.
}

28
packages/session-backends/sqlite-node/src/sqlite/branch-cache.ts
View file

@ -1,5 +1,6 @@
import { SessionError } from "@earendil-works/pi-agent-core";
import { uuidv7 } from "@earendil-works/pi-ai";
import { sql } from "./sql.ts";
import {
copyBranchEntriesThroughSeq,
deleteBranchEntries,
@ -7,7 +8,6 @@ import {
insertBranchEntry,
readBranchContainingEntry,
} from "./storage/branch-entries.ts";
import { deleteBranchTips, insertBranchTip, readBranchTipBranchId, updateBranchTip } from "./storage/branch-tips.ts";
import type { SqliteDatabase } from "./types.ts";
@ -17,32 +17,28 @@ export function deleteBranchCache(db: SqliteDatabase, sessionId: string) {
}
export function rebuildBranchCache(db: SqliteDatabase, sessionId: string) {
const tips = db
.prepare(
`SELECT leaf.id
FROM entries AS leaf
WHERE leaf.session_id = ?
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 }>(sessionId);
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) {
db.exec("SAVEPOINT build_branch_cache");
sql`SAVEPOINT build_branch_cache`.exec(db);
try {
const branchId = uuidv7();
insertBranchEntriesForPath(db, sessionId, branchId, leafId);
insertBranchTip(db, sessionId, leafId, branchId);
db.exec("RELEASE SAVEPOINT build_branch_cache");
sql`RELEASE SAVEPOINT build_branch_cache`.exec(db);
} catch (error) {
try {
db.exec("ROLLBACK TO SAVEPOINT build_branch_cache");
db.exec("RELEASE SAVEPOINT build_branch_cache");
sql`ROLLBACK TO SAVEPOINT build_branch_cache`.exec(db);
sql`RELEASE SAVEPOINT build_branch_cache`.exec(db);
} catch {
// Preserve the original build failure.
}

1
packages/session-backends/sqlite-node/src/sqlite/index.ts
View file

@ -5,6 +5,7 @@ export {
type SqliteWriterLeaseOptions,
} from "./repo.ts";
export * from "./search-backend.ts";
export * from "./sql.ts";
export type {
SqliteDatabase,
SqliteDatabaseFactory,

12
packages/session-backends/sqlite-node/src/sqlite/migrations.ts
View file

@ -1,5 +1,6 @@
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 {
@ -23,28 +24,25 @@ export async function loadMigrations(): Promise<SqliteMigration[]> {
}
function ensureMigrationsTable(db: SqliteDatabase): void {
db.exec(`
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 = db.prepare("SELECT id FROM migrations ORDER BY applied_at, id").all<{ id: string }>();
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);
db.prepare("INSERT INTO migrations (id, applied_at) VALUES (?, ?)").run(
migration.id,
new Date().toISOString(),
);
sql`INSERT INTO migrations (id, applied_at) VALUES (${migration.id}, ${new Date().toISOString()})`.run(db);
});
applied.add(migration.id);
}

7
packages/session-backends/sqlite-node/src/sqlite/repo.ts
View file

@ -20,6 +20,7 @@ import {
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 {
@ -169,9 +170,9 @@ function getParentPath(path: string): string {
}
function configureSqliteDatabase(db: SqliteDatabase): void {
db.exec("PRAGMA journal_mode=WAL");
db.exec("PRAGMA synchronous=FULL");
db.exec("PRAGMA busy_timeout=5000");
sql`PRAGMA journal_mode=WAL`.exec(db);
sql`PRAGMA synchronous=FULL`.exec(db);
sql`PRAGMA busy_timeout=5000`.exec(db);
}
function timestampToText(timestamp: number): string {

62
packages/session-backends/sqlite-node/src/sqlite/search-backend.ts
View file

@ -1,6 +1,7 @@
import type { SessionSearch, SessionSearchHit, SessionSearchOptions } from "@earendil-works/pi-agent-core";
import { getFileSystemResultOrThrow } 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,
@ -18,9 +19,9 @@ function getParentPath(path: string): string {
}
function configureSqliteDatabase(db: SqliteDatabase): void {
db.exec("PRAGMA journal_mode=WAL");
db.exec("PRAGMA synchronous=FULL");
db.exec("PRAGMA busy_timeout=5000");
sql`PRAGMA journal_mode=WAL`.exec(db);
sql`PRAGMA synchronous=FULL`.exec(db);
sql`PRAGMA busy_timeout=5000`.exec(db);
}
export interface SqliteSessionSearchOptions {
@ -30,14 +31,14 @@ export interface SqliteSessionSearchOptions {
}
function tableExists(db: SqliteDatabase, name: string): boolean {
return !!db
.prepare("SELECT 1 AS found FROM sqlite_master WHERE type = 'table' AND name = ? LIMIT 1")
.get<{ found: number }>(name);
return !!sql`SELECT 1 AS found FROM sqlite_master WHERE type = 'table' AND name = ${name} LIMIT 1`.get<{
found: number;
}>(db);
}
function ensureSearchSchema(db: SqliteDatabase): void {
const ftsExists = tableExists(db, "session_search_fts");
db.exec(`
sql`
CREATE VIRTUAL TABLE IF NOT EXISTS session_search_fts USING fts5(
payload,
content = 'entries',
@ -54,8 +55,8 @@ CREATE TRIGGER IF NOT EXISTS session_search_fts_au AFTER UPDATE OF payload ON en
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;
`);
if (!ftsExists) db.exec("INSERT INTO session_search_fts(session_search_fts) VALUES('rebuild')");
`.exec(db);
if (!ftsExists) sql`INSERT INTO session_search_fts(session_search_fts) VALUES('rebuild')`.exec(db);
}
/** SQLite FTS search over a co-located canonical session database. */
@ -102,32 +103,25 @@ class SqliteSessionSearch implements SessionSearch<SqliteSessionMetadata> {
const db = await this.openDatabase();
try {
const query = `"${text.replaceAll('"', '""')}"`;
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 session_search_fts MATCH ? AND (? IS NULL OR s.cwd = ?)
ORDER BY score`,
const cwd = options.cwd ?? null;
const rows = 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,
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
)
.all<SessionRow & { entry_id: string; timestamp: string; score: number }>(
query,
options.cwd ?? null,
options.cwd ?? null,
);
WHERE session_search_fts MATCH ${query} AND (${cwd} IS NULL OR s.cwd = ${cwd})
ORDER BY score`.all<SessionRow & { entry_id: string; timestamp: string; score: number }>(db);
const path = await this.getDatabasePath();
return rows.map((row) => ({
metadata: decodeSessionMetadata(row, path),

62
packages/session-backends/sqlite-node/src/sqlite/sql.ts Normal file
View file

@ -0,0 +1,62 @@
import type { SqliteDatabase, SqliteRunResult } from "./types.ts";
type SqlTemplateValue = unknown | SqlQuery;
/** A parameterized SQLite query produced by {@link sql}. */
export class SqlQuery {
readonly queryText: string;
readonly params: readonly unknown[];
constructor(queryText: string, params: readonly unknown[] = []) {
this.queryText = queryText;
this.params = params;
}
exec(db: SqliteDatabase): void {
if (this.params.length > 0) throw new TypeError("SQLite exec queries cannot have parameters");
db.exec(this.queryText);
}
run(db: SqliteDatabase): SqliteRunResult {
return db.prepare(this.queryText).run(...this.params);
}
get<TRow extends object>(db: SqliteDatabase): TRow | undefined {
return db.prepare(this.queryText).get<TRow>(...this.params);
}
all<TRow extends object>(db: SqliteDatabase): TRow[] {
return db.prepare(this.queryText).all<TRow>(...this.params);
}
}
/** Builds a parameterized query. Nested queries are inlined; other interpolations become `?` parameters. */
export function sql(strings: TemplateStringsArray, ...values: SqlTemplateValue[]): SqlQuery {
let queryText = strings[0] ?? "";
const params: unknown[] = [];
for (let index = 0; index < values.length; index++) {
const value = values[index];
if (value instanceof SqlQuery) {
queryText += value.queryText;
params.push(...value.params);
} else {
queryText += "?";
params.push(value);
}
queryText += strings[index + 1] ?? "";
}
return new SqlQuery(queryText, params);
}
/** Joins trusted query fragments while preserving their parameter order. */
export function joinSqlFragments(fragments: readonly SqlQuery[], separator: string): SqlQuery {
let queryText = "";
const params: unknown[] = [];
for (let index = 0; index < fragments.length; index++) {
if (index > 0) queryText += separator;
const fragment = fragments[index]!;
queryText += fragment.queryText;
params.push(...fragment.params);
}
return new SqlQuery(queryText, params);
}

134
packages/session-backends/sqlite-node/src/sqlite/storage/branch-entries.ts
View file

@ -1,4 +1,5 @@
import type { Entry } 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. */
@ -24,11 +25,11 @@ export interface CachedBranchQuery {
}
export function readCachedBranch(db: SqliteDatabase, sessionId: string, leafId: string) {
const membership = db
.prepare(
"SELECT branch_id, entry_seq FROM branch_entries WHERE session_id = ? AND entry_id = ? ORDER BY branch_id LIMIT 1",
)
.get<{ branch_id: string; entry_seq: number }>(sessionId, leafId);
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 };
}
@ -40,46 +41,45 @@ export function queryCachedBranchRows(
query: CachedBranchQuery,
) {
const oldestFirst = query.order === "oldestFirst";
const boundaryParams: unknown[] = [sessionId, branch.branchId, branch.leafSeq];
const stopPredicates: string[] = [];
if (query.stopAtType !== undefined) {
stopPredicates.push("stop_entry.type = ?");
boundaryParams.push(query.stopAtType);
}
if (query.stopAtId !== undefined) {
stopPredicates.push("stop.entry_id = ?");
boundaryParams.push(query.stopAtId);
}
const stopPredicates = [];
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 boundary = stopPredicates.length
? `WITH boundary AS (
SELECT ${oldestFirst ? "MIN" : "MAX"}(stop.entry_seq) AS entry_seq
FROM branch_entries AS stop
JOIN entries AS stop_entry
ON stop_entry.session_id = stop.session_id AND stop_entry.id = stop.entry_id
WHERE stop.session_id = ? AND stop.branch_id = ? AND stop.entry_seq <= ?
AND (${stopPredicates.join(" OR ")})
)`
: "";
const range = stopPredicates.length
? `AND b.entry_seq ${oldestFirst ? "<=" : ">="} COALESCE(
(SELECT entry_seq FROM boundary), ${oldestFirst ? branch.leafSeq : 0}
)`
: "";
const sql = `${boundary}
const aggregate = oldestFirst ? sql`MIN` : sql`MAX`;
const comparison = oldestFirst ? sql`<=` : sql`>=`;
const direction = oldestFirst ? sql`ASC` : sql`DESC`;
const boundary =
stopPredicates.length === 0
? sql``
: sql`WITH boundary AS (
SELECT ${aggregate}(stop.entry_seq) AS entry_seq
FROM branch_entries AS stop
JOIN entries AS stop_entry
ON stop_entry.session_id = stop.session_id AND stop_entry.id = stop.entry_id
WHERE stop.session_id = ${sessionId}
AND stop.branch_id = ${branch.branchId}
AND stop.entry_seq <= ${branch.leafSeq}
AND (${joinSqlFragments(stopPredicates, " OR ")})
)`;
const range =
stopPredicates.length === 0
? sql``
: sql`AND b.entry_seq ${comparison} COALESCE(
(SELECT entry_seq FROM boundary), ${oldestFirst ? branch.leafSeq : 0}
)`;
return sql`${boundary}
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 b.session_id = ? AND b.branch_id = ? AND b.entry_seq <= ?
WHERE b.session_id = ${sessionId}
AND b.branch_id = ${branch.branchId}
AND b.entry_seq <= ${branch.leafSeq}
${range}
ORDER BY b.entry_seq ${oldestFirst ? "ASC" : "DESC"}`;
const params = [...(stopPredicates.length === 0 ? [] : boundaryParams), sessionId, branch.branchId, branch.leafSeq];
return db.prepare(sql).all<CachedBranchEntryRow>(...params);
ORDER BY b.entry_seq ${direction}`.all<CachedBranchEntryRow>(db);
}
export function deleteBranchEntries(db: SqliteDatabase, sessionId: string) {
db.prepare("DELETE FROM branch_entries WHERE session_id = ?").run(sessionId);
sql`DELETE FROM branch_entries WHERE session_id = ${sessionId}`.run(db);
}
export function insertBranchEntry(
@ -91,42 +91,34 @@ export function insertBranchEntry(
entryType: string,
customType: string | null,
) {
db.prepare(
`INSERT INTO branch_entries
sql`INSERT INTO branch_entries
(session_id, branch_id, entry_id, entry_seq, entry_type, custom_type)
VALUES (?, ?, ?, ?, ?, ?)`,
).run(sessionId, branchId, entryId, entrySeq, entryType, customType);
VALUES (${sessionId}, ${branchId}, ${entryId}, ${entrySeq}, ${entryType}, ${customType})`.run(db);
}
export function insertBranchEntriesForPath(db: SqliteDatabase, sessionId: string, branchId: string, leafId: string) {
db.prepare(
`WITH RECURSIVE path(id, entry_seq, parent_id, type, custom_type) AS (
SELECT id, seq, parent_id, type,
CASE WHEN type = 'custom' THEN json_extract(payload, '$.customType') ELSE NULL END
FROM entries
WHERE session_id = ? AND id = ?
UNION ALL
SELECT parent.id, parent.seq, parent.parent_id, parent.type,
CASE WHEN parent.type = 'custom' THEN json_extract(parent.payload, '$.customType') ELSE NULL END
FROM entries AS parent
JOIN path AS child ON child.parent_id = parent.id
WHERE parent.session_id = ?
)
INSERT INTO branch_entries (session_id, branch_id, entry_id, entry_seq, entry_type, custom_type)
SELECT ?, ?, id, entry_seq, type, custom_type FROM path`,
).run(sessionId, leafId, sessionId, sessionId, branchId);
sql`WITH RECURSIVE path(id, entry_seq, parent_id, type, custom_type) AS (
SELECT id, seq, parent_id, type,
CASE WHEN type = 'custom' THEN json_extract(payload, '$.customType') ELSE NULL END
FROM entries
WHERE session_id = ${sessionId} AND id = ${leafId}
UNION ALL
SELECT parent.id, parent.seq, parent.parent_id, parent.type,
CASE WHEN parent.type = 'custom' THEN json_extract(parent.payload, '$.customType') ELSE NULL END
FROM entries AS parent
JOIN path AS child ON child.parent_id = parent.id
WHERE parent.session_id = ${sessionId}
)
INSERT INTO branch_entries (session_id, branch_id, entry_id, entry_seq, entry_type, custom_type)
SELECT ${sessionId}, ${branchId}, id, entry_seq, type, custom_type FROM path`.run(db);
}
export function readBranchContainingEntry(db: SqliteDatabase, sessionId: string, entryId: string) {
const row = db
.prepare(
`SELECT b.branch_id, b.entry_seq
FROM branch_entries AS b
WHERE b.session_id = ? AND b.entry_id = ?
ORDER BY b.branch_id
LIMIT 1`,
)
.get<{ branch_id: string; entry_seq: number }>(sessionId, entryId);
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 };
}
@ -137,10 +129,8 @@ export function copyBranchEntriesThroughSeq(
sourceBranchId: string,
throughSeq: number,
) {
db.prepare(
`INSERT INTO branch_entries (session_id, branch_id, entry_id, entry_seq, entry_type, custom_type)
SELECT session_id, ?, entry_id, entry_seq, entry_type, custom_type
FROM branch_entries
WHERE session_id = ? AND branch_id = ? AND entry_seq <= ?`,
).run(targetBranchId, sessionId, sourceBranchId, throughSeq);
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);
}

25
packages/session-backends/sqlite-node/src/sqlite/storage/branch-tips.ts
View file

@ -1,25 +1,21 @@
import { sql } from "../sql.ts";
import type { SqliteDatabase } from "../types.ts";
export function readBranchTipIds(db: SqliteDatabase, sessionId: string) {
return db
.prepare("SELECT tip_id FROM branch_tips WHERE session_id = ? ORDER BY tip_id")
.all<{ tip_id: string }>(sessionId)
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 = db
.prepare("SELECT branch_id FROM branch_tips WHERE session_id = ? AND tip_id = ?")
.get<{ branch_id: string }>(sessionId, tipId);
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) {
db.prepare("INSERT INTO branch_tips (session_id, tip_id, branch_id) VALUES (?, ?, ?)").run(
sessionId,
tipId,
branchId,
);
sql`INSERT INTO branch_tips (session_id, tip_id, branch_id) VALUES (${sessionId}, ${tipId}, ${branchId})`.run(db);
}
export function updateBranchTip(
@ -29,12 +25,11 @@ export function updateBranchTip(
oldTipId: string,
newTipId: string,
) {
const result = db
.prepare("UPDATE branch_tips SET tip_id = ? WHERE session_id = ? AND branch_id = ? AND tip_id = ?")
.run(newTipId, sessionId, branchId, oldTipId);
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) {
db.prepare("DELETE FROM branch_tips WHERE session_id = ?").run(sessionId);
sql`DELETE FROM branch_tips WHERE session_id = ${sessionId}`.run(db);
}

48
packages/session-backends/sqlite-node/src/sqlite/storage/entries.ts
View file

@ -1,4 +1,5 @@
import type { Entry, EntryOrder } from "@earendil-works/pi-agent-core";
import { sql } from "../sql.ts";
import type { SqliteDatabase } from "../types.ts";
export interface EntryRow {
@ -25,22 +26,17 @@ export function entryPayload(entry: Entry): Record<string, unknown> {
return payload;
}
function orderedSql(order: EntryOrder | undefined): string {
return order === "oldestFirst" ? "ASC" : "DESC";
}
export function insertEntryRow(db: SqliteDatabase, sessionId: string, entry: NewEntryRow) {
db.prepare(
"INSERT INTO entries (session_id, id, seq, parent_id, type, timestamp, payload) VALUES (?, ?, ?, ?, ?, ?, ?)",
).run(sessionId, entry.id, entry.seq, entry.parentId, entry.type, entry.timestamp, entry.payload);
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 db
.prepare(
"SELECT session_id, seq, id, parent_id, type, timestamp, payload FROM entries WHERE session_id = ? AND id = ?",
)
.get<EntryRow>(sessionId, entryId);
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(
@ -48,28 +44,20 @@ export function readEntryRows(
sessionId: string,
options: { afterSeq?: number; order?: EntryOrder } = {},
) {
const predicates = ["session_id = ?"];
const params: unknown[] = [sessionId];
if (options.afterSeq !== undefined) {
predicates.push("seq > ?");
params.push(options.afterSeq);
}
return db
.prepare(
`SELECT session_id, seq, id, parent_id, type, timestamp, payload
FROM entries
WHERE ${predicates.join(" AND ")}
ORDER BY seq ${orderedSql(options.order)}`,
)
.all<EntryRow>(...params);
const after = options.afterSeq === undefined ? sql`` : sql` AND seq > ${options.afterSeq}`;
const direction = options.order === "oldestFirst" ? sql`ASC` : sql`DESC`;
return sql`SELECT session_id, seq, id, parent_id, type, timestamp, payload
FROM entries
WHERE session_id = ${sessionId}${after}
ORDER BY seq ${direction}`.all<EntryRow>(db);
}
export function idExistsInEntries(db: SqliteDatabase, sessionId: string, id: string) {
return !!db
.prepare("SELECT 1 AS found FROM entries WHERE session_id = ? AND id = ? LIMIT 1")
.get<{ found: number }>(sessionId, id);
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) {
db.prepare("DELETE FROM entries WHERE session_id = ?").run(sessionId);
sql`DELETE FROM entries WHERE session_id = ${sessionId}`.run(db);
}

60
packages/session-backends/sqlite-node/src/sqlite/storage/facts.ts
View file

@ -1,3 +1,4 @@
import { sql } from "../sql.ts";
import type { SqliteDatabase } from "../types.ts";
export interface FactRow {
@ -16,58 +17,37 @@ export function appendFact(
key: string | null,
value: string | null,
) {
db.prepare("INSERT INTO facts (session_id, seq, kind, key, value) VALUES (?, ?, ?, ?, ?)").run(
sessionId,
seq,
kind,
key,
value,
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 db
.prepare(
`SELECT session_id, seq, kind, key, value
FROM facts
WHERE session_id = ? AND kind = ? AND key IS ?
ORDER BY seq DESC
LIMIT 1`,
)
.get<FactRow>(sessionId, kind, key);
return sql`SELECT session_id, seq, kind, key, value
FROM facts
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 db
.prepare(
`SELECT key, value FROM (
SELECT key, value, ROW_NUMBER() OVER (PARTITION BY key ORDER BY seq DESC) AS rank
FROM facts
WHERE session_id = ? AND kind = 'label'
)
WHERE rank = 1 AND value IS NOT NULL
ORDER BY key`,
return sql`SELECT key, value FROM (
SELECT key, value, ROW_NUMBER() OVER (PARTITION BY key ORDER BY seq DESC) AS rank
FROM facts
WHERE session_id = ${sessionId} AND kind = 'label'
)
.all<{ key: string; value: string }>(sessionId);
WHERE rank = 1 AND value IS NOT NULL
ORDER BY key`.all<{ key: string; value: string }>(db);
}
export function readFactRows(db: SqliteDatabase, sessionId: string, options: { afterSeq?: number } = {}) {
const predicates = ["session_id = ?"];
const params: unknown[] = [sessionId];
if (options.afterSeq !== undefined) {
predicates.push("seq > ?");
params.push(options.afterSeq);
}
return db
.prepare(
`SELECT session_id, seq, kind, key, value
FROM facts
WHERE ${predicates.join(" AND ")}
ORDER BY seq`,
)
.all<FactRow>(...params);
const after = options.afterSeq === undefined ? sql`` : sql` AND seq > ${options.afterSeq}`;
return sql`SELECT session_id, seq, kind, key, value
FROM facts
WHERE session_id = ${sessionId}${after}
ORDER BY seq`.all<FactRow>(db);
}
export function deleteFactRows(db: SqliteDatabase, sessionId: string) {
db.prepare("DELETE FROM facts WHERE session_id = ?").run(sessionId);
sql`DELETE FROM facts WHERE session_id = ${sessionId}`.run(db);
}

116
packages/session-backends/sqlite-node/src/sqlite/storage/lanes.ts
View file

@ -1,4 +1,5 @@
import { SessionError } from "@earendil-works/pi-agent-core";
import { sql } from "../sql.ts";
import type { SqliteDatabase } from "../types.ts";
export interface LaneRow {
@ -16,29 +17,22 @@ export interface LaneMoveRow {
}
export function createInitialLane(db: SqliteDatabase, sessionId: string, lane = "main", leafId: string | null = null) {
db.prepare("INSERT INTO lanes (session_id, lane, leaf_id, open_operation_id) VALUES (?, ?, ?, NULL)").run(
sessionId,
lane,
leafId,
);
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 = db
.prepare(
`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 = ?
ORDER BY l.lane`,
)
.all<LaneRow & { leaf_exists: number }>(sessionId);
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}`);
@ -53,56 +47,47 @@ export function readLanes(db: SqliteDatabase, sessionId: string) {
}
export function readLane(db: SqliteDatabase, sessionId: string, lane: string) {
return db
.prepare("SELECT session_id, lane, leaf_id, open_operation_id FROM lanes WHERE session_id = ? AND lane = ?")
.get<LaneRow>(sessionId, lane);
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 = db
.prepare(
`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 = ? AND l.lane = ?`,
)
.get<{ leaf_id: string | null; leaf_exists: number }>(sessionId, lane);
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) {
db.prepare("INSERT INTO lanes (session_id, lane, leaf_id, open_operation_id) VALUES (?, ?, ?, NULL)").run(
sessionId,
lane,
leafId,
);
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 = db
.prepare("UPDATE lanes SET leaf_id = ? WHERE session_id = ? AND lane = ?")
.run(leafId, sessionId, lane);
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 = db
.prepare("UPDATE lanes SET leaf_id = ? WHERE session_id = ? AND lane = ?")
.run(leafId, sessionId, lane);
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 = db
.prepare("UPDATE lanes SET open_operation_id = ? WHERE session_id = ? AND lane = ? AND open_operation_id IS NULL")
.run(runId, sessionId, lane);
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}`);
@ -110,38 +95,25 @@ export function startLaneOperation(db: SqliteDatabase, sessionId: string, lane:
}
export function finishLaneOperation(db: SqliteDatabase, sessionId: string, lane: string, runId: string) {
db.prepare(
"UPDATE lanes SET open_operation_id = NULL WHERE session_id = ? AND lane = ? AND open_operation_id = ?",
).run(sessionId, lane, runId);
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 } = {}) {
const predicates = ["session_id = ?"];
const params: unknown[] = [sessionId];
if (options.afterSeq !== undefined) {
predicates.push("seq > ?");
params.push(options.afterSeq);
}
return db
.prepare(
`SELECT session_id, seq, lane, leaf_id
FROM lane_moves
WHERE ${predicates.join(" AND ")}
ORDER BY seq`,
)
.all<LaneMoveRow>(...params);
const after = options.afterSeq === undefined ? sql`` : sql` AND seq > ${options.afterSeq}`;
return sql`SELECT session_id, seq, lane, leaf_id
FROM lane_moves
WHERE session_id = ${sessionId}${after}
ORDER BY seq`.all<LaneMoveRow>(db);
}
export function deleteLaneRows(db: SqliteDatabase, sessionId: string) {
db.prepare("DELETE FROM lane_moves WHERE session_id = ?").run(sessionId);
db.prepare("DELETE FROM lanes WHERE session_id = ?").run(sessionId);
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) {
db.prepare("INSERT INTO lane_moves (session_id, seq, lane, leaf_id) VALUES (?, ?, ?, ?)").run(
sessionId,
seq,
lane,
leafId,
sql`INSERT INTO lane_moves (session_id, seq, lane, leaf_id) VALUES (${sessionId}, ${seq}, ${lane}, ${leafId})`.run(
db,
);
}

88
packages/session-backends/sqlite-node/src/sqlite/storage/records.ts
View file

@ -1,4 +1,5 @@
import { SessionError } from "@earendil-works/pi-agent-core";
import { joinSqlFragments, sql } from "../sql.ts";
import type { SqliteDatabase } from "../types.ts";
export interface RecordRow {
@ -25,31 +26,21 @@ export interface NewRecordRow {
}
export function appendRecordRow(db: SqliteDatabase, sessionId: string, record: NewRecordRow) {
db.prepare(
`INSERT INTO records
sql`INSERT INTO records
(session_id, seq, id, lane, run_id, type, op_kind, timestamp, payload)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`,
).run(
sessionId,
record.seq,
record.id,
record.lane,
record.runId ?? null,
record.type,
record.opKind ?? null,
record.timestamp,
record.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 !!db
.prepare("SELECT 1 AS found FROM records WHERE session_id = ? AND id = ? LIMIT 1")
.get<{ found: number }>(sessionId, id);
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) {
db.prepare("DELETE FROM records WHERE session_id = ?").run(sessionId);
sql`DELETE FROM records WHERE session_id = ${sessionId}`.run(db);
}
export function readRecordRows(
@ -65,39 +56,18 @@ export function readRecordRows(
limit?: number;
} = {},
) {
const predicates = ["session_id = ?"];
const params: unknown[] = [sessionId];
if (query.lane !== undefined) {
predicates.push("lane = ?");
params.push(query.lane);
}
if (query.type !== undefined) {
predicates.push("type = ?");
params.push(query.type);
}
if (query.runId !== undefined) {
predicates.push("run_id = ?");
params.push(query.runId);
}
if (query.operationKind !== undefined) {
predicates.push("op_kind = ?");
params.push(query.operationKind);
}
if (query.afterSeq !== undefined) {
predicates.push("seq > ?");
params.push(query.afterSeq);
}
const limit = query.limit === undefined ? "" : " LIMIT ?";
if (query.limit !== undefined) params.push(query.limit);
const direction = query.order === "oldestFirst" ? "ASC" : "DESC";
return db
.prepare(
`SELECT session_id, seq, id, lane, run_id, type, op_kind, timestamp, payload
FROM records
WHERE ${predicates.join(" AND ")}
ORDER BY seq ${direction}${limit}`,
)
.all<RecordRow>(...params);
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(
@ -106,19 +76,15 @@ export function readOpenOperationRows(
lane: string,
_options: { limit?: number } = {},
): RecordRow[] {
const laneRow = db
.prepare("SELECT open_operation_id FROM lanes WHERE session_id = ? AND lane = ?")
.get<{ open_operation_id: string | null }>(sessionId, lane);
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 = db
.prepare(
`SELECT session_id, seq, id, lane, run_id, type, op_kind, timestamp, payload
FROM records
WHERE session_id = ?
AND id = ?`,
)
.get<RecordRow>(sessionId, laneRow.open_operation_id);
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}`);
}

13
packages/session-backends/sqlite-node/src/sqlite/storage/session-sequences.ts
View file

@ -1,14 +1,15 @@
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) {
db.prepare("INSERT INTO session_sequences (session_id, next_seq) VALUES (?, ?)").run(sessionId, nextSeq);
sql`INSERT INTO session_sequences (session_id, next_seq) VALUES (${sessionId}, ${nextSeq})`.run(db);
}
export function getNextSequence(db: SqliteDatabase, sessionId: string) {
const sequenceRow = db
.prepare("SELECT next_seq FROM session_sequences WHERE session_id = ?")
.get<{ next_seq: number }>(sessionId);
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}`);
}
@ -16,7 +17,7 @@ export function getNextSequence(db: SqliteDatabase, sessionId: string) {
}
export function setNextSequence(db: SqliteDatabase, sessionId: string, nextSeq: number) {
db.prepare("UPDATE session_sequences SET next_seq = ? WHERE session_id = ?").run(nextSeq, sessionId);
sql`UPDATE session_sequences SET next_seq = ${nextSeq} WHERE session_id = ${sessionId}`.run(db);
}
export function advanceSequence(db: SqliteDatabase, sessionId: string, seq: number) {
@ -24,5 +25,5 @@ export function advanceSequence(db: SqliteDatabase, sessionId: string, seq: numb
}
export function deleteSequence(db: SqliteDatabase, sessionId: string) {
db.prepare("DELETE FROM session_sequences WHERE session_id = ?").run(sessionId);
sql`DELETE FROM session_sequences WHERE session_id = ${sessionId}`.run(db);
}

41
packages/session-backends/sqlite-node/src/sqlite/storage/session-stats.ts
View file

@ -1,5 +1,6 @@
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 {
@ -12,21 +13,15 @@ export interface SessionStatsRow {
}
export function createStats(db: SqliteDatabase, sessionId: string, messageCount = 0): void {
db.prepare(
`INSERT INTO session_stats
sql`INSERT INTO session_stats
(session_id, message_count, cached_tokens, uncached_tokens, total_tokens, cost_total)
VALUES (?, ?, 0, 0, 0, 0)`,
).run(sessionId, messageCount);
VALUES (${sessionId}, ${messageCount}, 0, 0, 0, 0)`.run(db);
}
export function readStats(db: SqliteDatabase, sessionId: string): SessionStats {
const row = db
.prepare(
`SELECT session_id, message_count, cached_tokens, uncached_tokens, total_tokens, cost_total
FROM session_stats
WHERE session_id = ?`,
)
.get<SessionStatsRow>(sessionId);
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,
@ -38,26 +33,22 @@ export function readStats(db: SqliteDatabase, sessionId: string): SessionStats {
}
export function incrementMessageCount(db: SqliteDatabase, sessionId: string): void {
const result = db
.prepare("UPDATE session_stats SET message_count = message_count + 1 WHERE session_id = ?")
.run(sessionId);
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 = db
.prepare(
`UPDATE session_stats
SET cached_tokens = cached_tokens + ?,
uncached_tokens = uncached_tokens + ?,
total_tokens = total_tokens + ?,
cost_total = cost_total + ?
WHERE session_id = ?`,
)
.run(usage.cacheRead, usage.input + usage.cacheWrite, usage.totalTokens, usage.cost.total, sessionId);
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 {
db.prepare("DELETE FROM session_stats WHERE session_id = ?").run(sessionId);
sql`DELETE FROM session_stats WHERE session_id = ${sessionId}`.run(db);
}

82
packages/session-backends/sqlite-node/src/sqlite/storage/sessions.ts
View file

@ -1,4 +1,5 @@
import { assertJsonSerializable, SessionError } from "@earendil-works/pi-agent-core";
import { sql } from "../sql.ts";
import type { SqliteDatabase, SqliteSessionMetadata } from "../types.ts";
export interface SessionRow {
@ -38,7 +39,7 @@ function parseMetadata(metadata: string | null, sessionId: string): Record<strin
}
export function sessionExists(db: SqliteDatabase, sessionId: string) {
return !!db.prepare("SELECT 1 AS found FROM sessions WHERE id = ?").get<{ found: number }>(sessionId);
return !!sql`SELECT 1 AS found FROM sessions WHERE id = ${sessionId}`.get<{ found: number }>(db);
}
function serializeMetadata(metadata: Record<string, unknown> | undefined): string | null {
@ -51,61 +52,50 @@ function serializeMetadata(metadata: Record<string, unknown> | undefined): strin
}
export function insertSessionRow(db: SqliteDatabase, session: NewSessionRow) {
db.prepare("INSERT INTO sessions (id, created_at, metadata, cwd, parent_session_id) VALUES (?, ?, ?, ?, ?)").run(
session.id,
session.createdAt,
serializeMetadata(session.metadata),
session.cwd,
session.parentSessionId ?? null,
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 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
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 = ?`,
)
.get<SessionRow>(sessionId);
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 ? "" : "WHERE s.cwd = ?";
return 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
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>(...(options.cwd === undefined ? [] : [options.cwd]));
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) {
db.prepare("DELETE FROM sessions WHERE id = ?").run(sessionId);
sql`DELETE FROM sessions WHERE id = ${sessionId}`.run(db);
}
function parseSessionName(value: string | null, sessionId: string): string {

43
packages/session-backends/sqlite-node/src/sqlite/storage/writer-leases.ts
View file

@ -1,3 +1,4 @@
import { sql } from "../sql.ts";
import type { SqliteDatabase } from "../types.ts";
export interface WriterLease {
@ -19,18 +20,14 @@ export function acquireWriterLease(
now: number,
expiresAtMs: number,
) {
const row = db
.prepare(
`INSERT INTO writer_leases (session_id, owner_id, fence, expires_at_ms)
VALUES (?, ?, 1, ?)
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 <= ?
RETURNING owner_id, fence, expires_at_ms`,
)
.get<WriterLeaseRow>(sessionId, ownerId, expiresAtMs, now);
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 };
}
@ -41,25 +38,21 @@ export function renewWriterLease(
now: number,
expiresAtMs: number,
) {
const result = db
.prepare(
`UPDATE writer_leases
SET expires_at_ms = ?
WHERE session_id = ? AND owner_id = ? AND fence = ? AND expires_at_ms > ?`,
)
.run(expiresAtMs, sessionId, lease.ownerId, lease.fence, now);
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) {
db.prepare("DELETE FROM writer_leases WHERE session_id = ? AND owner_id = ? AND fence = ?").run(
sessionId,
lease.ownerId,
lease.fence,
);
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) {
db.prepare("DELETE FROM writer_leases WHERE session_id = ?").run(sessionId);
sql`DELETE FROM writer_leases WHERE session_id = ${sessionId}`.run(db);
}

39
packages/session-backends/sqlite-node/test/sql.test.ts Normal file
View file

@ -0,0 +1,39 @@
import { describe, expect, it } from "vitest";
import { createNodeSqliteFactory, joinSqlFragments, sql } from "../src/index.ts";
describe("sql", () => {
it("composes SQLite queries without renumbering parameters", async () => {
const db = await createNodeSqliteFactory().open(":memory:");
try {
sql`CREATE TABLE entries (id TEXT PRIMARY KEY, kind TEXT NOT NULL, active INTEGER NOT NULL)`.exec(db);
sql`INSERT INTO entries (id, kind, active) VALUES (${"one"}, ${"message"}, ${1})`.run(db);
sql`INSERT INTO entries (id, kind, active) VALUES (${"two"}, ${"message"}, ${0})`.run(db);
const filters = joinSqlFragments([sql`kind = ${"message"}`, sql`active = ${1}`], " AND ");
expect(sql`SELECT id FROM entries WHERE ${filters} LIMIT ${10}`.all<{ id: string }>(db)).toEqual([
{ id: "one" },
]);
} finally {
db.close();
}
});
it("executes parameterized queries", async () => {
const db = await createNodeSqliteFactory().open(":memory:");
try {
sql`CREATE TABLE values_table (id INTEGER PRIMARY KEY, value TEXT NOT NULL)`.exec(db);
sql`INSERT INTO values_table (id, value) VALUES (${1}, ${"one"})`.run(db);
sql`INSERT INTO values_table (id, value) VALUES (${2}, ${"two"})`.run(db);
expect(sql`SELECT value FROM values_table WHERE id = ${1}`.get<{ value: string }>(db)).toEqual({
value: "one",
});
expect(sql`SELECT value FROM values_table ORDER BY id`.all<{ value: string }>(db)).toEqual([
{ value: "one" },
{ value: "two" },
]);
} finally {
db.close();
}
});
});