fix(sqlite): preserve session metadata contracts
This commit is contained in:
Hypatia May
@@ -10,7 +10,7 @@ A SQLite durable session-persistence backend — a second `SessionPersistence` i
|
||||
|
||||
Each `SessionEvent` maps 1:1 onto a row in an `events` table `(session_id, seq, type, time, data, source_event_seqs, surface_op)` — `data` is the event payload as JSON text, so the row shape is the event verbatim (including `assistant/chunk`, keeping `seq` contiguous). The two `TEXT` columns `source_event_seqs` and `surface_op` are nullable; they store the event's optional surface-metadata fields (see [session surface](../../../.agents/notes/implemented/architecture/2026-06-18-session-surface.md)). Out-of-log metadata (`SessionHeader`), a per-materialization incarnation id, and a monotonic per-log revision live in a `sessions` row; a singleton state row carries the immutable store id. A `sessions` row is written only by the first `append` — its existence is the lazy-materialization signal (`list` reports exactly the sessions that have a row).
|
||||
|
||||
The repository's Node range supports unflagged `node:sqlite`. The database enables foreign keys and uses the configured journal mode (`wal` by default; use a rollback mode where WAL shared-memory files are unsuitable). `PRAGMA user_version` stores the table-layout version; databases with any other version are rejected because this unreleased format has no migrations.
|
||||
The repository's Node range supports unflagged `node:sqlite`. The database enables foreign keys and uses the configured journal mode (`wal` by default; use a rollback mode where WAL shared-memory files are unsuitable). `PRAGMA user_version` stores the table-layout version. A fresh empty database is initialized at the current version; nonempty unversioned databases and every other version are rejected because this unreleased format has no migrations. Rejection occurs before changing journal mode or stamping the file.
|
||||
|
||||
On filesystems with POSIX modes, the backend requests mode `0700` for missing directories and exclusively creates a missing database with mode `0600` before SQLite opens it; the process umask may further restrict both. New WAL, shared-memory, and persistent rollback-journal sidecars receive the database's resulting owner-only mode. Existing directories, database files, and sidecars keep their modes; filesystem setup errors other than an existing database fail initialization. These defaults prevent incidental exposure through a permissive process umask, but do not protect database confidentiality or integrity when another principal can replace the database entry in its parent directory.
|
||||
|
||||
@@ -55,5 +55,5 @@ SQLite storage does not mutate live request prefixes. A resumed loop can reuse p
|
||||
|
||||
- **`DatabaseSync` is synchronous** — every append transaction blocks the event loop for its duration; acceptable for local stores, a throughput ceiling for busy multi-session servers.
|
||||
- **Write contention has no wait or retry policy** — the backend sets no busy timeout and retries no locked-database error, so another connection holding a write transaction makes the operation reject immediately.
|
||||
- **Only the current `SCHEMA_VERSION` opens** — a database with any other schema version is rejected rather than migrated (unreleased software; no persisted user data to preserve).
|
||||
- **Only an empty new database or the current `SCHEMA_VERSION` opens** — a nonempty unversioned database or any other schema version is rejected rather than migrated (unreleased software; no persisted user data to preserve).
|
||||
- **Nothing deletes stored sessions** — rows accumulate until removed externally (the seam has no deletion surface; `ON DELETE CASCADE` is wired for such out-of-band cleanup).
|
||||
@@ -17,7 +17,7 @@ import type { SessionEvent, SessionId, SessionHeader, SurfaceOp } from '@deepsee
|
||||
* layout; orthogonal to a session's own `version` (which versions the EVENT
|
||||
* vocabulary, stored per session in the `sessions` row).
|
||||
*/
|
||||
export const SCHEMA_VERSION = 8
|
||||
export const SCHEMA_VERSION = 9
|
||||
|
||||
/**
|
||||
* A row of the `sessions` table — the out-of-log metadata ({@link SessionHeader}).
|
||||
@@ -63,9 +63,10 @@ export interface EventRow {
|
||||
export type JournalMode = 'wal' | 'delete' | 'truncate' | 'persist'
|
||||
|
||||
/**
|
||||
* Open the database and apply its schema and pragmas. A zero `user_version` is
|
||||
* stamped with {@link SCHEMA_VERSION}; every other non-current version rejects
|
||||
* rather than being migrated in place.
|
||||
* Open the database and apply its schema and pragmas. An empty database with a
|
||||
* zero `user_version` is initialized at {@link SCHEMA_VERSION}; a nonempty
|
||||
* unversioned database and every other non-current version reject rather than
|
||||
* being migrated in place.
|
||||
* @param path - the SQLite database file to open (created when absent).
|
||||
* @param journalMode - validated journal pragma.
|
||||
* @returns the open handle with pragmas applied and all three tables ensured.
|
||||
@@ -83,17 +84,20 @@ export function openDatabase(path: string, journalMode: JournalMode): DatabaseSy
|
||||
|
||||
function configureDatabase(db: DatabaseSync, path: string, journalMode: JournalMode): void {
|
||||
db.exec('PRAGMA foreign_keys = ON')
|
||||
// The validated union is safe to interpolate into a non-bindable PRAGMA.
|
||||
db.exec(`PRAGMA journal_mode = ${journalMode.toUpperCase()}`)
|
||||
// `PRAGMA user_version` always returns exactly one row { user_version }.
|
||||
const { user_version: onDisk } = db.prepare('PRAGMA user_version').get() as { user_version: number }
|
||||
const { count: userTableCount } = db.prepare(
|
||||
"SELECT COUNT(*) AS count FROM sqlite_schema WHERE type = 'table' AND name NOT LIKE 'sqlite_%'",
|
||||
).get() as { count: number }
|
||||
if (onDisk === 0 && userTableCount > 0) {
|
||||
throw new Error(`session database at "${path}" has a nonempty unversioned schema`)
|
||||
}
|
||||
if (onDisk !== 0 && onDisk !== SCHEMA_VERSION) {
|
||||
throw new Error(`session database at "${path}" has schema version ${onDisk}, incompatible with this build (${SCHEMA_VERSION})`)
|
||||
}
|
||||
if (onDisk === 0) {
|
||||
// Stamp fresh or pre-versioning databases.
|
||||
db.exec(`PRAGMA user_version = ${SCHEMA_VERSION}`)
|
||||
}
|
||||
// The validated union is safe to interpolate into a non-bindable PRAGMA.
|
||||
// Apply it only after rejecting incompatible existing databases.
|
||||
db.exec(`PRAGMA journal_mode = ${journalMode.toUpperCase()}`)
|
||||
db.exec(`
|
||||
CREATE TABLE IF NOT EXISTS persistence_state (
|
||||
singleton INTEGER PRIMARY KEY CHECK (singleton = 1),
|
||||
@@ -107,7 +111,7 @@ function configureDatabase(db: DatabaseSync, path: string, journalMode: JournalM
|
||||
CREATE TABLE IF NOT EXISTS sessions (
|
||||
id TEXT PRIMARY KEY,
|
||||
version INTEGER NOT NULL,
|
||||
created_at INTEGER NOT NULL,
|
||||
created_at REAL NOT NULL,
|
||||
cwd TEXT,
|
||||
parent_session TEXT,
|
||||
seed_length INTEGER,
|
||||
@@ -128,6 +132,7 @@ function configureDatabase(db: DatabaseSync, path: string, journalMode: JournalM
|
||||
PRIMARY KEY (session_id, seq)
|
||||
) STRICT
|
||||
`)
|
||||
if (onDisk === 0) db.exec(`PRAGMA user_version = ${SCHEMA_VERSION}`)
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -4,6 +4,7 @@ import { existsSync } from 'node:fs'
|
||||
import { chmod, mkdtemp, rm, stat, symlink, writeFile } from 'node:fs/promises'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { dirname, join } from 'node:path'
|
||||
import { DatabaseSync } from 'node:sqlite'
|
||||
import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
|
||||
import type { Session, SessionEvent, SurfaceEvent, SurfaceEventType } from '@deepseek-ai/dsh-session'
|
||||
import SessionPersistenceSqlite, { SCHEMA_VERSION } from '@deepseek-ai/dsh-session-persistence-sqlite'
|
||||
@@ -304,6 +305,23 @@ describe('SessionPersistenceSqlite: durability and crash semantics', () => {
|
||||
expect(() => openDatabase(olderPath, 'wal')).toThrow(/incompatible with this build/)
|
||||
})
|
||||
|
||||
it('rejects a nonempty unversioned database before stamping or changing journal mode', async () => {
|
||||
const path = await freshDbPath()
|
||||
const legacy = new DatabaseSync(path)
|
||||
legacy.exec('CREATE TABLE sessions (id TEXT PRIMARY KEY)')
|
||||
legacy.close()
|
||||
|
||||
expect(() => openDatabase(path, 'wal')).toThrow(/nonempty unversioned schema/)
|
||||
|
||||
const unchanged = new DatabaseSync(path)
|
||||
expect(unchanged.prepare('PRAGMA user_version').get()).toEqual({ user_version: 0 })
|
||||
expect(unchanged.prepare('PRAGMA journal_mode').get()).toEqual({ journal_mode: 'delete' })
|
||||
expect(unchanged.prepare(
|
||||
"SELECT name FROM sqlite_schema WHERE type = 'table' AND name = 'sessions'",
|
||||
).get()).toEqual({ name: 'sessions' })
|
||||
unchanged.close()
|
||||
})
|
||||
|
||||
it('rejects a sibling v3 database (the merge-collided version) rather than opening it against missing columns', async () => {
|
||||
// Version 3 identified two incompatible sibling layouts, so it is always rejected.
|
||||
const path = await freshDbPath()
|
||||
@@ -442,7 +460,7 @@ describe('SessionPersistenceSqlite: durability and crash semantics', () => {
|
||||
})
|
||||
|
||||
it('exposes the schema version constant', () => {
|
||||
expect(SCHEMA_VERSION).toBe(8)
|
||||
expect(SCHEMA_VERSION).toBe(9)
|
||||
})
|
||||
|
||||
it('keeps the revision stable for an empty repair hook', async () => {
|
||||
|
||||
@@ -84,6 +84,20 @@ export function runPersistenceContract(name: string, make: () => Promise<Contrac
|
||||
}
|
||||
})
|
||||
|
||||
it('round-trips a finite fractional creation timestamp', async () => {
|
||||
const { persistence, dispose } = await make()
|
||||
try {
|
||||
const m = { ...meta('fractional-created-at'), createdAt: 1.5 }
|
||||
await persistence.create(m)
|
||||
await persistence.append(m.id, oneTurnLog())
|
||||
|
||||
const loaded = await persistence.load(m.id)
|
||||
expect(loaded.meta.createdAt).toBe(1.5)
|
||||
} finally {
|
||||
await dispose()
|
||||
}
|
||||
})
|
||||
|
||||
it('crash recovery: load preserves an interrupted (unclosed) turn and closes it with turn/end {interrupted}', async () => {
|
||||
const { persistence, dispose } = await make()
|
||||
try {
|
||||
|
||||
@@ -5,7 +5,7 @@ import { mkdir, open } from 'node:fs/promises'
|
||||
import { dirname, resolve } from 'node:path'
|
||||
|
||||
/** Current derived-index schema version. Incompatible versions reset in place. */
|
||||
export const SESSION_QUERY_SQLITE_SCHEMA_VERSION = 3
|
||||
export const SESSION_QUERY_SQLITE_SCHEMA_VERSION = 4
|
||||
|
||||
/** SQLite application id protecting unrelated databases from derived resets. */
|
||||
export const SESSION_QUERY_SQLITE_APPLICATION_ID = 0x44534851
|
||||
@@ -112,7 +112,7 @@ function ensurePersistentSchema(db: DatabaseSync): void {
|
||||
CREATE TABLE IF NOT EXISTS persisted_sessions (
|
||||
id TEXT PRIMARY KEY,
|
||||
version INTEGER NOT NULL,
|
||||
created_at INTEGER NOT NULL,
|
||||
created_at REAL NOT NULL,
|
||||
cwd TEXT,
|
||||
parent_session TEXT,
|
||||
seed_length INTEGER,
|
||||
@@ -141,7 +141,7 @@ function ensureTemporarySchema(db: DatabaseSync): void {
|
||||
CREATE TEMP TABLE IF NOT EXISTS live_sessions (
|
||||
id TEXT PRIMARY KEY,
|
||||
version INTEGER NOT NULL,
|
||||
created_at INTEGER NOT NULL,
|
||||
created_at REAL NOT NULL,
|
||||
cwd TEXT,
|
||||
parent_session TEXT,
|
||||
seed_length INTEGER,
|
||||
|
||||
@@ -167,6 +167,22 @@ async function liveContext(config: ConstructorParameters<typeof SessionQuerySqli
|
||||
}
|
||||
|
||||
describe('SQLite session search', () => {
|
||||
it('indexes finite fractional creation timestamps from live and persisted sources', async () => {
|
||||
const persisted = header('fractional-persisted', 1.5)
|
||||
TestPersistence.reset([{ meta: persisted, events: messageEvents('persisted fractional') }])
|
||||
const ctx = await liveContext()
|
||||
await ctx.plugin(TestPersistence)
|
||||
const live = ctx.sessions.create(SessionId('fractional-live'), {
|
||||
seed: messageEvents('live fractional'),
|
||||
meta: { createdAt: 2.5 },
|
||||
})
|
||||
|
||||
await expect(ctx.sessionQuery.searchSessions({ query: 'persisted' }))
|
||||
.resolves.toMatchObject({ items: [{ header: { id: persisted.id, createdAt: 1.5 } }] })
|
||||
await expect(ctx.sessionQuery.searchSessions({ query: 'live' }))
|
||||
.resolves.toMatchObject({ items: [{ header: { id: live.id, createdAt: 2.5 } }] })
|
||||
})
|
||||
|
||||
it('searches two-character Unicode61 tokens in live-only sessions', async () => {
|
||||
const ctx = await liveContext({ path: ':memory:', snippetChars: 20 })
|
||||
const session = ctx.sessions.create(SessionId('live'), {
|
||||
|
||||
Reference in New Issue
Block a user