From 9970e045554ba88bc466139cfc92eb65717db251 Mon Sep 17 00:00:00 2001 From: Ali Mohammad Date: Thu, 1 Oct 2026 10:32:57 +0300 Subject: [PATCH] LOU-W5: SQLite store for sessions, checkpoints and approvals Adds SqliteStore (new subpath @loushy/build-ai-agent/sqlite, not in the root barrel): one node:sqlite file implementing SessionStore, CheckpointStore and ApprovalStore. Versioned schema via PRAGMA user_version with forward migrations in a transaction, WAL + busy timeout for multi-process sharing, prepared parameterized statements, opaque JSON payloads so snapshot fields added by other tickets round-trip, prune({ olderThanMs }), clear errors for corrupt files / use after close / missing node:sqlite. Shared contract suites run against the in-memory, file and SQLite implementations. Docs: Stores section in docs/sessions.md and a README bullet. Co-Authored-By: Claude Sonnet 5.5 --- README.md | 1 + docs/sessions.md | 44 +++ llms-full.txt | 45 +++ package.json | 5 + src/providers/importGraph.test.ts | 3 +- src/storage/sqlite/SqliteStore.test.ts | 305 ++++++++++++++++++ src/storage/sqlite/SqliteStore.ts | 97 ++++++ .../sqlite/__fixtures__/storeContracts.ts | 161 +++++++++ src/storage/sqlite/connection.ts | 65 ++++ src/storage/sqlite/driver.ts | 53 +++ src/storage/sqlite/index.ts | 6 + src/storage/sqlite/migrations.ts | 72 +++++ src/storage/sqlite/stores.ts | 113 +++++++ tsup.config.ts | 1 + 14 files changed, 970 insertions(+), 1 deletion(-) create mode 100644 src/storage/sqlite/SqliteStore.test.ts create mode 100644 src/storage/sqlite/SqliteStore.ts create mode 100644 src/storage/sqlite/__fixtures__/storeContracts.ts create mode 100644 src/storage/sqlite/connection.ts create mode 100644 src/storage/sqlite/driver.ts create mode 100644 src/storage/sqlite/index.ts create mode 100644 src/storage/sqlite/migrations.ts create mode 100644 src/storage/sqlite/stores.ts diff --git a/README.md b/README.md index d79abfcd8..eb01a8ac1 100644 --- a/README.md +++ b/README.md @@ -23,6 +23,7 @@ dashboard — any provider, any deploy target, no lock-in. - **Durable execution** — pass a `sessionId` + `checkpointStore` and a crash mid-conversation resumes instead of restarting - **Cancellation** — pass an `AbortSignal` (`agent.send(input, { signal })`) to stop a run; it resolves with `finishReason: 'aborted'` and the transcript so far, and the signal reaches the provider, tools and delegated agents - **Sessions** — `agent.session()` keeps a multi-turn conversation (in memory, or persisted with `FileSessionStore`) +- **SQLite store** — `new SqliteStore('./.loushy/agent.db')` from `@loushy/build-ai-agent/sqlite` keeps sessions, checkpoints and approvals in one durable, transactional file (built-in `node:sqlite`, no native dependency) — see [Sessions](docs/sessions.md#stores) - **Project instructions** — `createAgent({ projectInstructions: true })` appends the nearest `AGENTS.md` / `CLAUDE.md` to the instructions (opt-in) - **Parallel tool calls** — when the model asks for several tools in one turn they run concurrently (cap it with `toolConcurrency`, or `1` for sequential), and results still reach the transcript in the model's call order - **Skills** — `defineSkill()` / `loadSkills('./skills')`: only each skill's name and description sit in the system prompt; the model loads the full markdown on demand through an auto-registered `load_skill` tool diff --git a/docs/sessions.md b/docs/sessions.md index ca1454fa9..e61f07800 100644 --- a/docs/sessions.md +++ b/docs/sessions.md @@ -77,6 +77,50 @@ Session ids must match `^[A-Za-z0-9_-]{1,128}$` (they become file names, so `../x` and `a/b` are refused with an error that says so). Implement `SessionStore` yourself to keep transcripts in a database or Redis. +## Stores + +Sessions, durable-execution checkpoints and approvals each have a store +interface (`SessionStore`, `CheckpointStore`, `ApprovalStore`). Pick the +implementation by where the process runs: + +| Store | Sessions | Checkpoints | Approvals | Use it when | +| --- | --- | --- | --- | --- | +| In memory | `MemorySessionStore` | (supply your own) | (supply your own) | Tests, scripts, one process that never restarts | +| Files | `FileSessionStore(dir)` | `LocalStorageCheckpointStore` | `StorageServiceApprovalStore` | One machine, you want plain inspectable files | +| SQLite | `store.sessions` | `store.checkpoints` | `store.approvals` | A Node server: one durable, transactional file, shared safely by several processes | +| Cloudflare KV | - | `KVCheckpointStore` | - | Workers deployments (see [Deployment](deployment.md)) | + +`SqliteStore` keeps all three in one database file, using Node's built-in +`node:sqlite` (no native dependency; needs Node 22.13 or newer, and it is not +re-exported from the root entry, so importing the SDK never loads it): + +```ts +import { createAgent, AgentExecutor } from '@loushy/build-ai-agent'; +import { SqliteStore } from '@loushy/build-ai-agent/sqlite'; + +const store = new SqliteStore('./.loushy/agent.db'); // or ':memory:' +const session = createAgent({ provider }).session({ id: 'user-42', store: store.sessions }); +await session.send('Hello'); + +// Same file, same store, for durable runs with approvals: +// AgentExecutor.execute({ ..., sessionId, checkpointStore: store.checkpoints, approvalStore: store.approvals }) + +store.prune({ olderThanMs: 7 * 24 * 60 * 60 * 1000 }); // { sessions, checkpoints, approvals } deleted +store.close(); +``` + +- The directory is created if missing. The schema is versioned with + `PRAGMA user_version` and migrated on open; a database written by a newer + release is refused. +- WAL mode and a 5 s busy timeout let two processes share the file. An approval + can be resolved by only one of them. +- Checkpoints and approval snapshots are stored as opaque JSON, so new fields + round-trip unchanged. +- `prune()` removes sessions and checkpoints not updated within `olderThanMs`, + and approvals resolved that long ago; unresolved approvals are kept. +- A file that is not a SQLite database fails with an error naming the path; + using the store after `close()` throws a clear error. + ## Project instructions Not part of sessions, but often wanted together: see "Project instructions" in diff --git a/llms-full.txt b/llms-full.txt index 44a60ab1e..b2c07765d 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -21,6 +21,7 @@ dashboard — any provider, any deploy target, no lock-in. - **Durable execution** — pass a `sessionId` + `checkpointStore` and a crash mid-conversation resumes instead of restarting - **Cancellation** — pass an `AbortSignal` (`agent.send(input, { signal })`) to stop a run; it resolves with `finishReason: 'aborted'` and the transcript so far, and the signal reaches the provider, tools and delegated agents - **Sessions** — `agent.session()` keeps a multi-turn conversation (in memory, or persisted with `FileSessionStore`) +- **SQLite store** — `new SqliteStore('./.loushy/agent.db')` from `@loushy/build-ai-agent/sqlite` keeps sessions, checkpoints and approvals in one durable, transactional file (built-in `node:sqlite`, no native dependency) — see [Sessions](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/sessions.md#stores) - **Project instructions** — `createAgent({ projectInstructions: true })` appends the nearest `AGENTS.md` / `CLAUDE.md` to the instructions (opt-in) - **Parallel tool calls** — when the model asks for several tools in one turn they run concurrently (cap it with `toolConcurrency`, or `1` for sequential), and results still reach the transcript in the model's call order - **Skills** — `defineSkill()` / `loadSkills('./skills')`: only each skill's name and description sit in the system prompt; the model loads the full markdown on demand through an auto-registered `load_skill` tool @@ -2504,6 +2505,50 @@ Session ids must match `^[A-Za-z0-9_-]{1,128}$` (they become file names, so `../x` and `a/b` are refused with an error that says so). Implement `SessionStore` yourself to keep transcripts in a database or Redis. +## Stores + +Sessions, durable-execution checkpoints and approvals each have a store +interface (`SessionStore`, `CheckpointStore`, `ApprovalStore`). Pick the +implementation by where the process runs: + +| Store | Sessions | Checkpoints | Approvals | Use it when | +| --- | --- | --- | --- | --- | +| In memory | `MemorySessionStore` | (supply your own) | (supply your own) | Tests, scripts, one process that never restarts | +| Files | `FileSessionStore(dir)` | `LocalStorageCheckpointStore` | `StorageServiceApprovalStore` | One machine, you want plain inspectable files | +| SQLite | `store.sessions` | `store.checkpoints` | `store.approvals` | A Node server: one durable, transactional file, shared safely by several processes | +| Cloudflare KV | - | `KVCheckpointStore` | - | Workers deployments (see [Deployment](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/deployment.md)) | + +`SqliteStore` keeps all three in one database file, using Node's built-in +`node:sqlite` (no native dependency; needs Node 22.13 or newer, and it is not +re-exported from the root entry, so importing the SDK never loads it): + +```ts +import { createAgent, AgentExecutor } from '@loushy/build-ai-agent'; +import { SqliteStore } from '@loushy/build-ai-agent/sqlite'; + +const store = new SqliteStore('./.loushy/agent.db'); // or ':memory:' +const session = createAgent({ provider }).session({ id: 'user-42', store: store.sessions }); +await session.send('Hello'); + +// Same file, same store, for durable runs with approvals: +// AgentExecutor.execute({ ..., sessionId, checkpointStore: store.checkpoints, approvalStore: store.approvals }) + +store.prune({ olderThanMs: 7 * 24 * 60 * 60 * 1000 }); // { sessions, checkpoints, approvals } deleted +store.close(); +``` + +- The directory is created if missing. The schema is versioned with + `PRAGMA user_version` and migrated on open; a database written by a newer + release is refused. +- WAL mode and a 5 s busy timeout let two processes share the file. An approval + can be resolved by only one of them. +- Checkpoints and approval snapshots are stored as opaque JSON, so new fields + round-trip unchanged. +- `prune()` removes sessions and checkpoints not updated within `olderThanMs`, + and approvals resolved that long ago; unresolved approvals are kept. +- A file that is not a SQLite database fails with an error naming the path; + using the store after `close()` throws a clear error. + ## Project instructions Not part of sessions, but often wanted together: see "Project instructions" in diff --git a/package.json b/package.json index fb05b7b9a..4b6c8be55 100644 --- a/package.json +++ b/package.json @@ -58,6 +58,11 @@ "import": "./dist/execution/hooks.mjs", "require": "./dist/execution/hooks.js" }, + "./sqlite": { + "types": "./dist/storage/sqlite/index.d.ts", + "import": "./dist/storage/sqlite/index.mjs", + "require": "./dist/storage/sqlite/index.js" + }, "./triggers": { "types": "./dist/triggers/index.d.ts", "import": "./dist/triggers/index.mjs", diff --git a/src/providers/importGraph.test.ts b/src/providers/importGraph.test.ts index 26393464b..a4bf59ba2 100644 --- a/src/providers/importGraph.test.ts +++ b/src/providers/importGraph.test.ts @@ -19,9 +19,10 @@ const NEVER_LOADED_AT_IMPORT = [ 'undici', 'dockerode', '@modelcontextprotocol/sdk', + 'node:sqlite', // LOU-W5: only the lazily-loaded /sqlite subpath may use it, and only when a store is constructed ]; -const ENTRY_POINTS = ['index', 'core/index', 'tools/index', 'tools/mcp/index', 'flows/index', 'testing/index']; +const ENTRY_POINTS = ['index', 'core/index', 'tools/index', 'tools/mcp/index', 'flows/index', 'testing/index', 'storage/sqlite/index']; /** Records every module specifier resolved while `body` runs, via synchronous module hooks. */ const RECORDER = ` diff --git a/src/storage/sqlite/SqliteStore.test.ts b/src/storage/sqlite/SqliteStore.test.ts new file mode 100644 index 000000000..08d6a16b3 --- /dev/null +++ b/src/storage/sqlite/SqliteStore.test.ts @@ -0,0 +1,305 @@ +/** + * LOU-W5: SqliteStore - contract suites, persistence, migration, prune, + * multi-instance access, error handling and an approval pause/resume that + * crosses a close/reopen. + */ +import { describe, it, expect, afterEach, vi } from 'vitest'; +import { mkdtempSync, rmSync, writeFileSync, existsSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +// Vite 5 cannot resolve the bare 'node:sqlite' builtin; load it the way the library does. +const DatabaseSync = loadDatabaseSync(); +import { SqliteStore } from './index'; +import { migrate, MIGRATIONS } from './migrations'; +import { loadDatabaseSync } from './driver'; +import { + describeApprovalStoreContract, + describeCheckpointStoreContract, + describeSessionStoreContract, + makeCheckpoint, + makePending, + makeSnapshot, +} from './__fixtures__/storeContracts'; +import { MemorySessionStore, FileSessionStore } from '../../session/sessionStore'; +import type { Checkpoint, CheckpointStore } from '../../execution/checkpoint'; +import type { ApprovalStore, ResolvedApproval } from '../../execution/ApprovalGate'; +import { AgentExecutor } from '../../execution/AgentExecutor'; +import { resumeAfterApproval } from '../../execution/resume'; +import { ToolRegistry } from '../../tools'; +import { AgentBuilder } from '../../core'; +import { AgentType } from '../../types'; +import { mockModel } from '../../testing'; + +const tempDirs: string[] = []; +const stores: SqliteStore[] = []; + +function tempDir(): string { + const dir = mkdtempSync(join(tmpdir(), 'loushy-sqlite-')); + tempDirs.push(dir); + return dir; +} + +function open(path: string): SqliteStore { + const store = new SqliteStore(path); + stores.push(store); + return store; +} + +afterEach(() => { + while (stores.length) stores.pop()?.close(); + while (tempDirs.length) rmSync(tempDirs.pop() as string, { recursive: true, force: true }); +}); + +function inMemoryCheckpointStore(): CheckpointStore { + const map = new Map(); + return { + async save(id, checkpoint) { + map.set(id, JSON.stringify(checkpoint)); + }, + async load(id) { + const raw = map.get(id); + return raw === undefined ? null : (JSON.parse(raw) as Checkpoint); + }, + async delete(id) { + map.delete(id); + }, + }; +} + +function inMemoryApprovalStore(): ApprovalStore { + const map = new Map(); + return { + async save(pending, snapshot) { + map.set(pending.id, JSON.stringify({ pending, snapshot })); + }, + async resolve(id) { + const raw = map.get(id); + if (raw === undefined) return null; + map.delete(id); + return JSON.parse(raw) as ResolvedApproval; + }, + }; +} + +describeSessionStoreContract('MemorySessionStore', () => new MemorySessionStore()); +describeSessionStoreContract('FileSessionStore', () => new FileSessionStore(tempDir())); +describeSessionStoreContract('SqliteStore.sessions', () => open(':memory:').sessions); +describeCheckpointStoreContract('in-memory', inMemoryCheckpointStore); +describeCheckpointStoreContract('SqliteStore.checkpoints', () => open(':memory:').checkpoints); +describeApprovalStoreContract('in-memory', inMemoryApprovalStore); +describeApprovalStoreContract('SqliteStore.approvals', () => open(':memory:').approvals); + +describe('SqliteStore persistence', () => { + it('keeps everything across close and reopen of a file', async () => { + const file = join(tempDir(), 'agent.db'); + const first = new SqliteStore(file); + await first.sessions.save('s1', [{ role: 'user', content: 'remember me' }]); + await first.checkpoints.save('run-1', makeCheckpoint()); + const pending = makePending('ap-1'); + await first.approvals.save(pending, makeSnapshot(pending)); + first.close(); + + const second = open(file); + expect(await second.sessions.load('s1')).toEqual([{ role: 'user', content: 'remember me' }]); + expect(await second.checkpoints.load('run-1')).toEqual(makeCheckpoint()); + expect((await second.approvals.resolve('ap-1'))?.pending).toEqual(pending); + }); + + it('creates a missing directory and uses WAL mode', () => { + const file = join(tempDir(), 'nested', 'deeper', 'agent.db'); + const store = open(file); + expect(existsSync(file)).toBe(true); + expect(store.path).toBe(file); + const raw = new DatabaseSync(file); + expect(raw.prepare('PRAGMA journal_mode').get()).toEqual({ journal_mode: 'wal' }); + raw.close(); + }); +}); + +describe('SqliteStore migrations', () => { + it('upgrades a v0 (empty) database to the current version', async () => { + const file = join(tempDir(), 'v0.db'); + const raw = new DatabaseSync(file); + expect(raw.prepare('PRAGMA user_version').get()).toEqual({ user_version: 0 }); + raw.close(); + + const store = open(file); + await store.sessions.save('s', []); + const check = new DatabaseSync(file); + expect(check.prepare('PRAGMA user_version').get()).toEqual({ user_version: MIGRATIONS.length }); + check.close(); + }); + + it('applies only the missing steps, and rolls back a failing one', () => { + const db = new DatabaseSync(':memory:'); + const steps = ['CREATE TABLE a (x INTEGER)', 'CREATE TABLE b (y INTEGER)']; + expect(migrate(db, steps.slice(0, 1))).toBe(1); + expect(migrate(db, steps)).toBe(2); + const broken = [...steps, 'CREATE TABLE c (z INTEGER); THIS IS NOT SQL']; + expect(() => migrate(db, broken)).toThrow(); + expect(db.prepare('PRAGMA user_version').get()).toEqual({ user_version: 2 }); + expect(db.prepare("SELECT name FROM sqlite_master WHERE name = 'c'").get()).toBeUndefined(); + db.close(); + }); + + it('refuses a database written by a newer version', () => { + const file = join(tempDir(), 'future.db'); + const raw = new DatabaseSync(file); + raw.exec('PRAGMA user_version = 999'); + raw.close(); + expect(() => new SqliteStore(file)).toThrow(/newer than this library supports/); + }); +}); + +describe('SqliteStore.prune', () => { + it('deletes stale sessions, checkpoints and resolved approvals only', async () => { + const file = join(tempDir(), 'prune.db'); + const store = open(file); + await store.sessions.save('old', []); + await store.sessions.save('fresh', []); + await store.checkpoints.save('old', makeCheckpoint()); + await store.checkpoints.save('fresh', makeCheckpoint()); + for (const id of ['resolved-old', 'resolved-fresh', 'pending-old']) { + const pending = makePending(id); + await store.approvals.save(pending, makeSnapshot(pending)); + } + await store.approvals.resolve('resolved-old'); + await store.approvals.resolve('resolved-fresh'); + + const raw = new DatabaseSync(file); + const age = 10 * 60 * 1000; + raw.prepare('UPDATE sessions SET updated_at = updated_at - ? WHERE id = ?').run(age, 'old'); + raw.prepare('UPDATE checkpoints SET updated_at = updated_at - ? WHERE session_id = ?').run(age, 'old'); + raw.prepare('UPDATE approvals SET resolved_at = resolved_at - ? WHERE id = ?').run(age, 'resolved-old'); + raw.prepare('UPDATE approvals SET updated_at = updated_at - ? WHERE id = ?').run(age, 'pending-old'); + raw.close(); + + expect(store.prune({ olderThanMs: 60 * 1000 })).toEqual({ sessions: 1, checkpoints: 1, approvals: 1 }); + expect(await store.sessions.load('old')).toBeUndefined(); + expect(await store.sessions.load('fresh')).toEqual([]); + expect(await store.checkpoints.load('old')).toBeNull(); + expect(await store.checkpoints.load('fresh')).not.toBeNull(); + expect(await store.approvals.resolve('resolved-old')).toBeNull(); + expect(await store.approvals.resolve('pending-old')).not.toBeNull(); + expect(store.prune({ olderThanMs: 60 * 1000 })).toEqual({ sessions: 0, checkpoints: 0, approvals: 0 }); + }); + + it('rejects a negative or non-finite age', () => { + const store = open(':memory:'); + expect(() => store.prune({ olderThanMs: -1 })).toThrow(/olderThanMs/); + expect(() => store.prune({ olderThanMs: Number.NaN })).toThrow(/olderThanMs/); + }); +}); + +describe('SqliteStore sharing a file', () => { + it('lets two instances see each other and resolve an approval only once', async () => { + const file = join(tempDir(), 'shared.db'); + const a = open(file); + const b = open(file); + await a.sessions.save('s', [{ role: 'user', content: 'from a' }]); + expect(await b.sessions.load('s')).toEqual([{ role: 'user', content: 'from a' }]); + + const pending = makePending('ap'); + await a.approvals.save(pending, makeSnapshot(pending)); + const results = await Promise.all([a.approvals.resolve('ap'), b.approvals.resolve('ap')]); + expect(results.filter((r) => r !== null)).toHaveLength(1); + }); + + it('survives interleaved writes from two instances', async () => { + const file = join(tempDir(), 'busy.db'); + const a = open(file); + const b = open(file); + const writes: Promise[] = []; + for (let i = 0; i < 50; i++) { + writes.push((i % 2 ? a : b).checkpoints.save(`run-${i}`, makeCheckpoint({ stepIndex: i }))); + } + await Promise.all(writes); + expect((await a.checkpoints.load('run-49'))?.stepIndex).toBe(49); + expect((await b.checkpoints.load('run-0'))?.stepIndex).toBe(0); + }); + + it('migrates a fresh file once when two instances open it', () => { + const file = join(tempDir(), 'race.db'); + const a = open(file); + const b = open(file); + expect(a.path).toBe(b.path); + }); +}); + +describe('SqliteStore errors', () => { + it('names the path of a file that is not a SQLite database', () => { + const file = join(tempDir(), 'garbage.db'); + writeFileSync(file, 'this is definitely not a sqlite database, just some text '.repeat(40)); + expect(() => new SqliteStore(file)).toThrow(new RegExp(`Could not open SQLite database at .*garbage\\.db`)); + }); + + it('gives a clear error when used after close(), and close() is idempotent', async () => { + const store = new SqliteStore(':memory:'); + const { sessions, checkpoints, approvals } = store; + store.close(); + store.close(); + await expect(sessions.load('a')).rejects.toThrow(/is closed/); + await expect(checkpoints.save('a', makeCheckpoint())).rejects.toThrow(/is closed/); + await expect(approvals.resolve('a')).rejects.toThrow(/is closed/); + expect(() => store.prune({ olderThanMs: 0 })).toThrow(/is closed/); + }); + + it('throws a clear Node-version error when node:sqlite is unavailable', () => { + const original = process.getBuiltinModule; + vi.spyOn(process, 'getBuiltinModule').mockImplementation(((id: string) => + id === 'node:sqlite' ? undefined : original(id)) as typeof process.getBuiltinModule); + try { + expect(() => new SqliteStore(':memory:')).toThrow(/needs the built-in 'node:sqlite' module \(Node >= 22/); + expect(() => loadDatabaseSync()).toThrow(/Upgrade Node/); + } finally { + vi.restoreAllMocks(); + } + }); +}); + +describe('SqliteStore end to end', () => { + it('resumes an approval after the store is closed and reopened', async () => { + const file = join(tempDir(), 'e2e.db'); + const execute = vi.fn().mockResolvedValue({ charged: true }); + const registry = new ToolRegistry(); + registry.register('chargeCard', { + displayName: 'Charge Card', + tool: { description: 'Charge a card', parameters: {}, execute } as never, + needsApproval: true, + }); + const agent = AgentBuilder.create() + .setType(AgentType.SmartAssistant) + .setName('Test Agent') + .addTool('chargeCard', { tool: 'chargeCard', options: {} }) + .build(); + const model = () => mockModel([{ toolCalls: [{ name: 'chargeCard', args: {} }] }, 'All done']); + + const first = new SqliteStore(file); + const paused = await AgentExecutor.execute({ + agent, + input: 'charge it', + provider: model(), + toolRegistry: registry, + sessionId: 'run-1', + checkpointStore: first.checkpoints, + approvalStore: first.approvals, + }); + expect(paused.finishReason).toBe('awaiting-approval'); + expect(execute).not.toHaveBeenCalled(); + first.close(); + + // "new process": fresh store instances, and a model that only has the post-approval turn. + const second = open(file); + const resumed = await resumeAfterApproval( + { id: paused.approvalId as string, approved: true }, + second.approvals, + registry, + mockModel(['All done']), + {}, + second.checkpoints + ); + expect(execute).toHaveBeenCalledTimes(1); + expect(resumed.finishReason).toBe('stop'); + expect(await second.approvals.resolve(paused.approvalId as string)).toBeNull(); + }); +}); diff --git a/src/storage/sqlite/SqliteStore.ts b/src/storage/sqlite/SqliteStore.ts new file mode 100644 index 000000000..2a721e50d --- /dev/null +++ b/src/storage/sqlite/SqliteStore.ts @@ -0,0 +1,97 @@ +import type { CheckpointStore } from '../../execution/checkpoint'; +import type { ApprovalStore } from '../../execution/ApprovalGate'; +import type { SessionStore } from '../../session/sessionStore'; +import { Connection } from './connection'; +import { SqliteApprovalStore, SqliteCheckpointStore, SqliteSessionStore, Statements } from './stores'; + +/** Options for {@link SqliteStore.prune}. */ +export interface PruneOptions { + /** Delete rows last updated more than this many milliseconds ago. */ + olderThanMs: number; +} + +/** How many rows {@link SqliteStore.prune} deleted, per kind. */ +export interface PruneResult { + sessions: number; + checkpoints: number; + approvals: number; +} + +/** + * One SQLite file holding sessions, checkpoints and approvals: durable, + * transactional and zero-ops. Uses Node's built-in `node:sqlite` (no native + * dependency; Node >= 22.13). WAL mode plus a busy timeout let several + * processes share one file. + * + * @example + * ```ts + * import { SqliteStore } from '@loushy/build-ai-agent/sqlite'; + * + * const store = new SqliteStore('./.loushy/agent.db'); // or ':memory:' + * const session = agent.session({ id: 'user-42', store: store.sessions }); + * await AgentExecutor.execute({ + * agent, input, provider, toolRegistry, + * sessionId: 'run-1', + * checkpointStore: store.checkpoints, + * approvalStore: store.approvals, + * }); + * store.close(); + * ``` + */ +export class SqliteStore { + /** Session transcripts, for `agent.session({ store })`. */ + readonly sessions: SessionStore; + /** Durable-execution checkpoints, for `checkpointStore`. */ + readonly checkpoints: CheckpointStore; + /** Pending tool approvals, for `approvalStore`. */ + readonly approvals: ApprovalStore; + private readonly connection: Connection; + private readonly sql: Statements; + + /** + * @param path database file (its directory is created if missing) or `':memory:'` + * @throws if `node:sqlite` is unavailable, or `path` is not a SQLite database + */ + constructor(path: string) { + this.connection = Connection.open(path); + this.sql = new Statements(this.connection); + this.sessions = new SqliteSessionStore(this.connection); + this.checkpoints = new SqliteCheckpointStore(this.connection); + this.approvals = new SqliteApprovalStore(this.connection); + } + + /** Absolute path of the database file (or `':memory:'`). */ + get path(): string { + return this.connection.path; + } + + /** + * Delete stale sessions and checkpoints (by last update) and approvals that + * were resolved more than `olderThanMs` ago. Unresolved approvals are kept. + * + * @example + * ```ts + * const removed = store.prune({ olderThanMs: 7 * 24 * 60 * 60 * 1000 }); + * console.log(removed); // { sessions: 3, checkpoints: 1, approvals: 0 } + * ``` + */ + prune({ olderThanMs }: PruneOptions): PruneResult { + if (!Number.isFinite(olderThanMs) || olderThanMs < 0) { + throw new Error(`prune: olderThanMs must be a non-negative number of milliseconds, got ${olderThanMs}.`); + } + const cutoff = Date.now() - olderThanMs; + return this.connection.transaction(() => { + const run = (sql: string): number => Number(this.sql.get(sql).run(cutoff).changes); + return { + sessions: run('DELETE FROM sessions WHERE updated_at < ?'), + checkpoints: run('DELETE FROM checkpoints WHERE updated_at < ?'), + approvals: run('DELETE FROM approvals WHERE resolved_at IS NOT NULL AND resolved_at < ?'), + }; + }); + } + + /** Close the database. Any later use of the store or its parts throws. Safe to call twice. */ + close(): void { + this.connection.close(); + } +} diff --git a/src/storage/sqlite/__fixtures__/storeContracts.ts b/src/storage/sqlite/__fixtures__/storeContracts.ts new file mode 100644 index 000000000..4f8c07a85 --- /dev/null +++ b/src/storage/sqlite/__fixtures__/storeContracts.ts @@ -0,0 +1,161 @@ +/** + * Shared contract suites (LOU-W5): the same assertions run against every + * implementation of SessionStore, CheckpointStore and ApprovalStore. + * `factory` returns a fresh, empty store (and may be async). + */ +import { describe, it, expect } from 'vitest'; +import type { SessionStore } from '../../../session/sessionStore'; +import type { Checkpoint, CheckpointStore } from '../../../execution/checkpoint'; +import type { ApprovalStore, ExecutionSnapshot, PendingApproval } from '../../../execution/ApprovalGate'; +import { AgentType } from '../../../types'; +import type { Message } from '../../../providers/llm'; + +type Factory = () => T | Promise; + +const convo: Message[] = [ + { role: 'user', content: 'hi' }, + { role: 'assistant', content: 'hello' }, +]; + +export function describeSessionStoreContract(name: string, factory: Factory): void { + describe(`SessionStore contract: ${name}`, () => { + it('returns undefined for an id that was never saved', async () => { + expect(await (await factory()).load('nope')).toBeUndefined(); + }); + + it('round-trips a transcript and replaces it on save', async () => { + const store = await factory(); + await store.save('a', convo); + expect(await store.load('a')).toEqual(convo); + await store.save('a', [convo[0]]); + expect(await store.load('a')).toEqual([convo[0]]); + }); + + it('keeps sessions isolated and tolerates delete of a missing id', async () => { + const store = await factory(); + await store.save('a', convo); + await store.save('b', [convo[0]]); + await store.delete('a'); + await store.delete('missing'); + expect(await store.load('a')).toBeUndefined(); + expect(await store.load('b')).toEqual([convo[0]]); + }); + + it('does not alias the saved array', async () => { + const store = await factory(); + const messages = structuredClone(convo); + await store.save('a', messages); + messages[0].content = 'tampered'; + expect((await store.load('a'))?.[0].content).toBe('hi'); + }); + + it('refuses invalid session ids', async () => { + const store = await factory(); + await expect(store.save('../x', convo)).rejects.toThrow(/Invalid session id/); + await expect(store.load('a/b')).rejects.toThrow(/Invalid session id/); + }); + }); +} + +export function makeCheckpoint(overrides: Partial = {}): Checkpoint { + return { + agentId: 'agent-1', + sessionId: 'run-1', + stepIndex: 2, + messages: convo, + toolCalls: [], + usage: { promptTokens: 1, completionTokens: 2, totalTokens: 3 }, + ...overrides, + }; +} + +export function describeCheckpointStoreContract(name: string, factory: Factory): void { + describe(`CheckpointStore contract: ${name}`, () => { + it('returns null when nothing was saved', async () => { + expect(await (await factory()).load('run-1')).toBeNull(); + }); + + it('round-trips and overwrites a checkpoint, including businessState', async () => { + const store = await factory(); + const first = makeCheckpoint({ businessState: { orderId: 7 } }); + await store.save('run-1', first); + expect(await store.load('run-1')).toEqual(first); + const second = makeCheckpoint({ stepIndex: 3 }); + await store.save('run-1', second); + expect(await store.load('run-1')).toEqual(second); + }); + + it('keeps sessions isolated and tolerates delete of a missing id', async () => { + const store = await factory(); + await store.save('a', makeCheckpoint({ sessionId: 'a' })); + await store.save('b', makeCheckpoint({ sessionId: 'b' })); + await store.delete('a'); + await store.delete('missing'); + expect(await store.load('a')).toBeNull(); + expect((await store.load('b'))?.sessionId).toBe('b'); + }); + + it('round-trips fields it does not know about (opaque JSON)', async () => { + const store = await factory(); + const extended = { ...makeCheckpoint(), futureField: { nested: [1, 'two'] } } as Checkpoint; + await store.save('run-1', extended); + expect(await store.load('run-1')).toEqual(extended); + }); + }); +} + +export function makePending(id: string, overrides: Partial = {}): PendingApproval { + return { + id, + toolCallId: `call-${id}`, + toolName: 'chargeCard', + args: { amount: 100 }, + agentId: 'agent-1', + createdAt: '2026-01-01T00:00:00.000Z', + ...overrides, + }; +} + +export function makeSnapshot(pending: PendingApproval): ExecutionSnapshot { + return { + agent: { id: 'agent-1', name: 'Test Agent', agentType: AgentType.SmartAssistant }, + currentMessages: convo, + pendingToolCall: pending, + steps: 1, + sessionId: 'run-1', + }; +} + +export function describeApprovalStoreContract(name: string, factory: Factory): void { + describe(`ApprovalStore contract: ${name}`, () => { + it('returns null for an id that was never saved', async () => { + expect(await (await factory()).resolve('nope')).toBeNull(); + }); + + it('resolves a saved approval exactly once', async () => { + const store = await factory(); + const pending = makePending('a'); + await store.save(pending, makeSnapshot(pending)); + expect(await store.resolve('a')).toEqual({ pending, snapshot: makeSnapshot(pending) }); + expect(await store.resolve('a')).toBeNull(); + }); + + it('keeps concurrent saves isolated', async () => { + const store = await factory(); + const a = makePending('a'); + const b = makePending('b', { toolName: 'sendEmail' }); + await Promise.all([store.save(a, makeSnapshot(a)), store.save(b, makeSnapshot(b))]); + const [ra, rb] = await Promise.all([store.resolve('a'), store.resolve('b')]); + expect(ra?.pending).toEqual(a); + expect(rb?.pending).toEqual(b); + }); + + it('round-trips snapshot fields it does not know about (opaque JSON)', async () => { + const store = await factory(); + const pending = makePending('a'); + const snapshot = { ...makeSnapshot(pending), futureField: { x: [true] } } as ExecutionSnapshot; + await store.save(pending, snapshot); + expect((await store.resolve('a'))?.snapshot).toEqual(snapshot); + }); + }); +} diff --git a/src/storage/sqlite/connection.ts b/src/storage/sqlite/connection.ts new file mode 100644 index 000000000..e9c92e624 --- /dev/null +++ b/src/storage/sqlite/connection.ts @@ -0,0 +1,65 @@ +import { mkdirSync } from 'node:fs'; +import { dirname, resolve } from 'node:path'; +import { loadDatabaseSync, type SqlDatabase } from './driver'; +import { migrate } from './migrations'; + +/** Milliseconds a writer waits for another process's lock before failing. */ +const BUSY_TIMEOUT_MS = 5000; + +/** One open database shared by the three stores, with open-state and transaction helpers. */ +export class Connection { + private database: SqlDatabase | undefined; + + private constructor( + readonly path: string, + database: SqlDatabase + ) { + this.database = database; + } + + /** Open (creating the directory and file as needed), configure and migrate. */ + static open(path: string): Connection { + const DatabaseSync = loadDatabaseSync(); + const location = path === ':memory:' ? path : resolve(path); + if (location !== ':memory:') mkdirSync(dirname(location), { recursive: true }); + const database = new DatabaseSync(location, { timeout: BUSY_TIMEOUT_MS }); + try { + if (location !== ':memory:') database.exec('PRAGMA journal_mode = WAL'); + database.exec('PRAGMA synchronous = NORMAL'); + migrate(database); + } catch (error) { + database.close(); + throw new Error(`Could not open SQLite database at ${location}: ${(error as Error).message}`, { + cause: error, + }); + } + return new Connection(location, database); + } + + /** The open database; throws once {@link close} has been called. */ + get db(): SqlDatabase { + if (!this.database) { + throw new Error(`SqliteStore for ${this.path} is closed. Create a new SqliteStore to keep using it.`); + } + return this.database; + } + + /** Run `work` in one write transaction (`BEGIN IMMEDIATE`), rolling back if it throws. */ + transaction(work: () => T): T { + const db = this.db; + db.exec('BEGIN IMMEDIATE'); + try { + const result = work(); + db.exec('COMMIT'); + return result; + } catch (error) { + db.exec('ROLLBACK'); + throw error; + } + } + + close(): void { + this.database?.close(); + this.database = undefined; + } +} diff --git a/src/storage/sqlite/driver.ts b/src/storage/sqlite/driver.ts new file mode 100644 index 000000000..9ec4feccd --- /dev/null +++ b/src/storage/sqlite/driver.ts @@ -0,0 +1,53 @@ +/** + * Minimal structural types for the parts of `node:sqlite` this package uses, + * plus the lazy loader. `@types/node` older than 22.5 has no `node:sqlite` + * typings, and loading it lazily keeps `import '@loushy/build-ai-agent/sqlite'` + * safe on Node versions that lack it. + */ + +/** A row returned by a query. */ +export type SqlRow = Record; + +/** A bind parameter. */ +export type SqlValue = string | number | null; + +/** The subset of `StatementSync` used here. */ +export interface SqlStatement { + run(...params: SqlValue[]): { changes: number | bigint }; + get(...params: SqlValue[]): SqlRow | undefined; +} + +/** The subset of `DatabaseSync` used here. */ +export interface SqlDatabase { + exec(sql: string): void; + prepare(sql: string): SqlStatement; + close(): void; +} + +/** The constructor shape of `DatabaseSync`. */ +export type SqlDatabaseConstructor = new (path: string, options?: { timeout?: number }) => SqlDatabase; + +/** Oldest Node release in the 22 line that ships `node:sqlite` without a flag. */ +const REQUIRED_NODE = '22.13.0'; + +/** + * Load `DatabaseSync` from `node:sqlite`. Uses `process.getBuiltinModule` so it + * works from both the ESM and the CJS build, and throws a clear error on a + * Node that has no `node:sqlite`. + */ +export function loadDatabaseSync(): SqlDatabaseConstructor { + let loaded: { DatabaseSync?: SqlDatabaseConstructor } | undefined; + try { + loaded = process.getBuiltinModule?.('node:sqlite') as typeof loaded; + } catch { + loaded = undefined; + } + if (!loaded?.DatabaseSync) { + throw new Error( + `SqliteStore needs the built-in 'node:sqlite' module (Node >= ${REQUIRED_NODE}), ` + + `but this runtime (Node ${process.versions.node}) does not provide it. ` + + 'Upgrade Node, or use FileSessionStore / LocalStorageCheckpointStore instead.' + ); + } + return loaded.DatabaseSync; +} diff --git a/src/storage/sqlite/index.ts b/src/storage/sqlite/index.ts new file mode 100644 index 000000000..601e5a622 --- /dev/null +++ b/src/storage/sqlite/index.ts @@ -0,0 +1,6 @@ +/** + * `@loushy/build-ai-agent/sqlite`: one SQLite file for sessions, checkpoints + * and approvals. Deliberately NOT re-exported from the root entry. + */ +export { SqliteStore } from './SqliteStore'; +export type { PruneOptions, PruneResult } from './SqliteStore'; diff --git a/src/storage/sqlite/migrations.ts b/src/storage/sqlite/migrations.ts new file mode 100644 index 000000000..dbc8538b5 --- /dev/null +++ b/src/storage/sqlite/migrations.ts @@ -0,0 +1,72 @@ +import type { SqlDatabase } from './driver'; + +/** + * Forward-only schema migrations. `MIGRATIONS[n]` upgrades a database from + * `user_version` n to n + 1. Append new entries; never edit shipped ones. + * + * Payloads are opaque JSON text so new fields on sessions, checkpoints and + * approval snapshots round-trip without schema changes. + */ +export const MIGRATIONS: readonly string[] = [ + ` + CREATE TABLE sessions ( + id TEXT PRIMARY KEY, + payload TEXT NOT NULL, + created_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL + ); + CREATE INDEX sessions_updated_at ON sessions (updated_at); + CREATE TABLE checkpoints ( + session_id TEXT PRIMARY KEY, + payload TEXT NOT NULL, + created_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL + ); + CREATE INDEX checkpoints_updated_at ON checkpoints (updated_at); + CREATE TABLE approvals ( + id TEXT PRIMARY KEY, + payload TEXT NOT NULL, + created_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL, + resolved_at INTEGER + ); + CREATE INDEX approvals_resolved_at ON approvals (resolved_at); + `, +]; + +function readVersion(db: SqlDatabase): number { + const row = db.prepare('PRAGMA user_version').get(); + return Number(row?.user_version ?? 0); +} + +/** + * Bring `db` up to `migrations.length`, in one transaction. The version is + * re-read after the write lock is taken, so two processes opening a fresh + * file at once migrate it exactly once. A database from a newer release is + * refused rather than guessed at. + * + * @returns the schema version after migrating + */ +export function migrate(db: SqlDatabase, migrations: readonly string[] = MIGRATIONS): number { + const target = migrations.length; + if (readVersion(db) === target) return target; + db.exec('BEGIN IMMEDIATE'); + try { + const current = readVersion(db); + if (current > target) { + throw new Error( + `Database schema version ${current} is newer than this library supports (${target}). ` + + 'Upgrade @loushy/build-ai-agent.' + ); + } + for (let version = current; version < target; version++) { + db.exec(migrations[version]); + } + db.exec(`PRAGMA user_version = ${target}`); + db.exec('COMMIT'); + } catch (error) { + db.exec('ROLLBACK'); + throw error; + } + return target; +} diff --git a/src/storage/sqlite/stores.ts b/src/storage/sqlite/stores.ts new file mode 100644 index 000000000..8e8d78f8b --- /dev/null +++ b/src/storage/sqlite/stores.ts @@ -0,0 +1,113 @@ +import type { Message } from '../../providers/llm'; +import type { Checkpoint, CheckpointStore } from '../../execution/checkpoint'; +import type { + ApprovalStore, + ExecutionSnapshot, + PendingApproval, + ResolvedApproval, +} from '../../execution/ApprovalGate'; +import { assertSessionId, type SessionStore } from '../../session/sessionStore'; +import type { Connection } from './connection'; +import type { SqlStatement } from './driver'; + +/** Prepared statements, prepared once each; `connection.db` throws a clear error after close. */ +export class Statements { + private readonly cache = new Map(); + constructor(private readonly connection: Connection) {} + + get(sql: string): SqlStatement { + const db = this.connection.db; + let statement = this.cache.get(sql); + if (!statement) { + statement = db.prepare(sql); + this.cache.set(sql, statement); + } + return statement; + } +} + +const parse = (row: { payload?: unknown } | undefined): T | undefined => + row === undefined ? undefined : (JSON.parse(String(row.payload)) as T); + +// `table` and `key` are fixed identifiers from this file, never user input. +const upsert = (table: string, key: string): string => + `INSERT INTO ${table} (${key}, payload, created_at, updated_at) VALUES (?, ?, ?, ?) + ON CONFLICT(${key}) DO UPDATE SET payload = excluded.payload, updated_at = excluded.updated_at`; + +/** `SessionStore` over the `sessions` table. */ +export class SqliteSessionStore implements SessionStore { + private readonly sql: Statements; + constructor(connection: Connection) { + this.sql = new Statements(connection); + } + + async load(id: string): Promise { + assertSessionId(id); + return parse(this.sql.get('SELECT payload FROM sessions WHERE id = ?').get(id)); + } + + async save(id: string, messages: readonly Message[]): Promise { + assertSessionId(id); + const now = Date.now(); + this.sql.get(upsert('sessions', 'id')).run(id, JSON.stringify(messages), now, now); + } + + async delete(id: string): Promise { + assertSessionId(id); + this.sql.get('DELETE FROM sessions WHERE id = ?').run(id); + } +} + +/** `CheckpointStore` over the `checkpoints` table. The checkpoint is stored as opaque JSON. */ +export class SqliteCheckpointStore implements CheckpointStore { + private readonly sql: Statements; + constructor(connection: Connection) { + this.sql = new Statements(connection); + } + + async save(sessionId: string, checkpoint: Checkpoint): Promise { + const now = Date.now(); + this.sql.get(upsert('checkpoints', 'session_id')).run(sessionId, JSON.stringify(checkpoint), now, now); + } + + async load(sessionId: string): Promise { + const row = this.sql.get('SELECT payload FROM checkpoints WHERE session_id = ?').get(sessionId); + return parse(row) ?? null; + } + + async delete(sessionId: string): Promise { + this.sql.get('DELETE FROM checkpoints WHERE session_id = ?').run(sessionId); + } +} + +/** + * `ApprovalStore` over the `approvals` table. `resolve` claims the row in one + * transaction (so two processes cannot both resume it) and stamps + * `resolved_at`; the row stays until `prune()` removes it. + */ +export class SqliteApprovalStore implements ApprovalStore { + private readonly sql: Statements; + constructor(private readonly connection: Connection) { + this.sql = new Statements(connection); + } + + async save(pending: PendingApproval, snapshot: ExecutionSnapshot): Promise { + const record: ResolvedApproval = { pending, snapshot }; + const now = Date.now(); + this.connection.transaction(() => { + this.sql.get(upsert('approvals', 'id')).run(pending.id, JSON.stringify(record), now, now); + // Saving again re-opens an approval that was already resolved. + this.sql.get('UPDATE approvals SET resolved_at = NULL WHERE id = ?').run(pending.id); + }); + } + + async resolve(id: string): Promise { + return this.connection.transaction(() => { + const row = this.sql.get('SELECT payload FROM approvals WHERE id = ? AND resolved_at IS NULL').get(id); + if (row === undefined) return null; + const now = Date.now(); + this.sql.get('UPDATE approvals SET resolved_at = ?, updated_at = ? WHERE id = ?').run(now, now, id); + return parse(row) ?? null; + }); + } +} diff --git a/tsup.config.ts b/tsup.config.ts index 06b88c252..431baba1d 100644 --- a/tsup.config.ts +++ b/tsup.config.ts @@ -34,6 +34,7 @@ export default defineConfig({ 'cli/doctor': 'src/cli/doctor.ts', 'execution/otel': 'src/execution/otel.ts', 'execution/hooks': 'src/execution/hooks.ts', + 'storage/sqlite/index': 'src/storage/sqlite/index.ts', 'triggers/index': 'src/triggers/index.ts', }, format: ['cjs', 'esm'],