fix(sqlite): enforce integer session metadata
This commit is contained in:
Hypatia May
@@ -19,16 +19,16 @@ Key choices recorded here because they are durable, contested, and surprising:
|
||||
|
||||
- **The canonical durable log persists every `SessionEvent` verbatim, including `assistant/chunk`.** `deriveMessages()` skips chunks, and a chunk-filtered rollout (Codex's `policy.rs`) is tempting — but `seq = log.length` and the load-validation `events[i].seq === i` require a *contiguous* log; filtering chunks out would leave holes and break both the contract and resume. A chunk-filtered projection is possible later as a derived view with its own renumbering, but it is NOT the canonical log.
|
||||
- **Append-only; a crashed turn is closed, never truncated.** Flushed events are never rewritten. The [semantic checkpoint policy](../bug-fix/2026-07-21-semantic-session-checkpoints.md) drains the request before model dispatch, a recorded top-level call before tool dispatch, and the complete response/result batch after a step; the loop drains the final turn boundary. Because one interrupted turn may contain substantial valid work, `load` preserves its contiguous, parseable events and appends risk-classified error results for unanswered assistant calls, a missing `step/end`, and `turn/end` with `{ kind: 'interrupted' }`. The synthetic results keep resumed provider transcripts valid. Only an incomplete final record is discarded; a parse error or sequence gap at or before the last real `turn/end` is corruption and makes the session unloadable.
|
||||
- **File backend canonical, DB backend a proven drop-in.** `SessionEvent` maps 1:1 onto a row `(session_id, seq, type, time, data)` — `append` is INSERT (in a transaction asserting the contiguous-seq contract), `load` is SELECT … ORDER BY seq. `dsh-session-persistence-sqlite` is exactly this: a `SessionPersistence` subclass with no interface change (opencode runs this exact shape on SQLite/WAL), and it passes the same `runPersistenceContract` suite as the JSONL backend — so the contract holds both backends to identical semantics (lazy materialization, interrupted-turn close on load, contiguous-seq), expressed once over file bytes and once over rows.
|
||||
- **Metadata is out-of-log.** Format version, cwd, and lineage are storage concerns, not replayable conversation state, so they live in a `SessionHeader` owned by `dsh-session` and attached to a `Session` via a new readonly `session.header` — never in `SessionEventMap`, never reaching `deriveMessages()`. The alternative (a merge-extensible `session/meta` event as log line 0) was rejected: an in-log event would ride along with a seeded/forked session for free, but metadata is not replayable state, so the explicit out-of-log header seam is the cleaner cost. (The header was originally split into an immutable `SessionHeader` plus a mutable `SessionSummary` whose union was `SessionMeta`; the mutable summary was later removed as dead state — see [Drop the mutable session summary](../simplification/2026-06-19-drop-mutable-session-summary.md).)
|
||||
- **File backend canonical, DB backend a proven drop-in.** `SessionEvent` maps 1:1 onto a row `(session_id, seq, type, time, data)` — `append` is INSERT (in a transaction asserting the contiguous-seq contract), `load` is SELECT … ORDER BY seq. `dsh-session-persistence-sqlite` is exactly this: a `SessionPersistence` subclass with no interface change (opencode runs this exact shape on SQLite/WAL), and it passes the same `runPersistenceContract` suite as the JSONL backend — so the contract holds both backends to identical semantics (lazy materialization, interrupted-turn close on load, contiguous-seq), expressed once over file bytes and once over rows. Its database carries a dedicated application id and monotonic schema version. A pristine file creates all tables and stamps both header values in one transaction; an unversioned file with any user-defined schema object or application identity, a foreign current-version identity, and every non-current version reject before journal-mode mutation.
|
||||
- **Metadata is out-of-log.** Format version, cwd, and lineage are storage concerns, not replayable conversation state, so they live in a `SessionHeader` owned by `dsh-session` and attached to a `Session` via a new readonly `session.header` — never in `SessionEventMap`, never reaching `deriveMessages()`. `createdAt` is non-negative safe-integer Unix epoch milliseconds: live creation and persistence registration reject fractional values, JSONL validates the decoded header, and SQLite stores it in a strict `INTEGER` column. The alternative (a merge-extensible `session/meta` event as log line 0) was rejected: an in-log event would ride along with a seeded/forked session for free, but metadata is not replayable state, so the explicit out-of-log header seam is the cleaner cost. (The header was originally split into an immutable `SessionHeader` plus a mutable `SessionSummary` whose union was `SessionMeta`; the mutable summary was later removed as dead state — see [Drop the mutable session summary](../simplification/2026-06-19-drop-mutable-session-summary.md).)
|
||||
- **`ctx.agents.create()` and `ctx.agents.resume()` are async factories; resume additionally crosses the persistence boundary.** `ctx.agents.resume({ resumeSessionId })` awaits `ctx.sessionPersistence.load`, recreates the live session with the loaded events (so `lastTurnNumber`/`deriveMessages` continue), and registers the fresh agent under the exact resumed id. The agent-loop does NOT hard-inject `sessionPersistence` (that would pend non-persistent demos forever); `resume` rejects with a clear error when it is absent.
|
||||
|
||||
## Alternatives considered
|
||||
|
||||
Each key choice above records its rejected alternative where the choice is stated: a **chunk-filtered canonical log** (Codex's `policy.rs` shape) — breaks the contiguous-seq contract; **truncating a crashed turn** — silently destroys a long autonomous run's real work; an **in-log `session/meta` event as line 0** — metadata is not replayable state; **hard-injecting `sessionPersistence` into the loop** — would pend non-persistent demos forever.
|
||||
Each key choice above records its rejected alternative where the choice is stated: a **chunk-filtered canonical log** (Codex's `policy.rs` shape) — breaks the contiguous-seq contract; **truncating a crashed turn** — silently destroys a long autonomous run's real work; an **in-log `session/meta` event as line 0** — metadata is not replayable state; **finite fractional `createdAt` values** — have no producer and diverge from integer Unix-millisecond storage and query columns; **adopting a non-pristine unversioned SQLite file** — can overwrite unrelated objects or identity; **hard-injecting `sessionPersistence` into the loop** — would pend non-persistent demos forever.
|
||||
|
||||
Format versioning: the header carries a `version`; `load` rejects any non-current version (no migration — the pre-release session format is pinned at `SESSION_FORMAT_VERSION = 0` and absorbs shape churn, per the AGENTS.md pre-release stance). Stated honestly: append-only + flush is robust to partial trailing writes (tolerated on load) but not to fsync-less power loss mid-line; a DB/WAL backend is the stronger option later.
|
||||
|
||||
## Consequences
|
||||
|
||||
Two new packages and the metadata seam in `dsh-session` (`session.header`, the `create(id?, options?)` signature). Bought: durable resume/fork, a read/replay path, crash tolerance, and the foundation the ACP `session/load` ([ACP support](../feature/2026-06-14-acp-agent-client-protocol.md)) needs — all over the existing event-sourced log, with the backend swappable behind one interface. The reusable `runPersistenceContract` suite holds every backend to the same append-only, contiguous-seq, lazy-materialization, and serializability semantics. Persisting the full log also settles event fidelity: `assistant/chunk` remains verbatim.
|
||||
Two new packages and the metadata seam in `dsh-session` (`session.header`, the `create(id?, options?)` signature). Bought: durable resume/fork, a read/replay path, crash tolerance, and the foundation the ACP `session/load` ([ACP support](../feature/2026-06-14-acp-agent-client-protocol.md)) needs — all over the existing event-sourced log, with the backend swappable behind one interface. The reusable `runPersistenceContract` suite holds every backend to the same append-only, contiguous-seq, lazy-materialization, integer-metadata, and serializability semantics. Persisting the full log also settles event fidelity: `assistant/chunk` remains verbatim. SQLite initialization either commits its complete owned schema and header identity or leaves no partial schema to strand on the next open.
|
||||
@@ -36,7 +36,7 @@ One serialized operation reads the provider-neutral `SessionPersistence` snapsho
|
||||
|
||||
Persisted documents survive restarts. Live sessions use connection-local TEMP tables, shadow the persisted base for the same id, and reveal that base on detach. Closing the database drops live rows. Unmounting persistence hides durable rows without treating absence as authoritative deletion; remounting observes and reconciles the backend again. Conflicting immutable live and durable headers fail rather than combining sources.
|
||||
|
||||
The derived schema has its own application id and monotonic schema version. A recognized incompatible version resets only this derived database. A database with a foreign application id or unrecognized user tables is refused before journal-mode mutation, which prevents an accidentally configured canonical session database from being changed. On POSIX filesystems, missing directories and database files are created owner-only so new SQLite sidecars inherit that mode; existing modes are preserved. One service in one process exclusively owns a derived-index path; cross-process writers are unsupported because generations and live TEMP shadow state are connection-owned.
|
||||
The derived schema has its own application id and monotonic schema version. Persistent and TEMP session metadata store the integer `SessionHeader.createdAt` contract in strict `INTEGER` columns. A recognized incompatible version resets only this derived database. A database with a foreign application id or unrecognized user tables is refused before journal-mode mutation, which prevents an accidentally configured canonical session database from being changed. On POSIX filesystems, missing directories and database files are created owner-only so new SQLite sidecars inherit that mode; existing modes are preserved. One service in one process exclusively owns a derived-index path; cross-process writers are unsupported because generations and live TEMP shadow state are connection-owned.
|
||||
|
||||
Cancellation rejects queued operations and caller waits around asynchronous source observation without committing an aborted observation. Node's synchronous `DatabaseSync` MATCH call cannot be interrupted once it is executing on the JavaScript thread, so the service checks the signal at serialized boundaries but does not promise mid-statement preemption.
|
||||
|
||||
|
||||
@@ -1238,7 +1238,7 @@ fork(source: SessionForkSource, boundary?: number, childSessionId?: SessionId):
|
||||
|
||||
Types: [CreateSessionOptions](../core-data-structures/persistence.md) · [OutOfBandSessionEventType](../core-data-structures/session.md) · [Session](../core-data-structures/session.md) · [SessionEvent](../core-data-structures/core.md) · [SessionEventMap](../core-data-structures/session.md) · [SessionId](../core-data-structures/core.md) · [TurnTrigger](../core-data-structures/session.md)
|
||||
|
||||
Source: [`packages/core/session/src/index.ts:605`](../../packages/core/session/src/index.ts)
|
||||
Source: [`packages/core/session/src/index.ts:607`](../../packages/core/session/src/index.ts)
|
||||
|
||||
## `ctx.sessionTitle` — `SessionTitleService`
|
||||
|
||||
|
||||
@@ -53,7 +53,7 @@ interface SessionHeader {
|
||||
readonly version: number
|
||||
/** The session's id (mirrors the {@link Session}'s id). */
|
||||
readonly id: SessionId
|
||||
/** Unix epoch milliseconds when the session was created. */
|
||||
/** Non-negative safe-integer Unix epoch milliseconds when the session was created. */
|
||||
readonly createdAt: number
|
||||
/** Absolute working directory the session was created in (if any). */
|
||||
readonly cwd?: string
|
||||
|
||||
@@ -129,8 +129,10 @@ function snapshotSessionHeader(id: SessionId, source?: SessionHeader): SessionHe
|
||||
if (record.id !== id) {
|
||||
throw new Error(`session header id "${String(record.id)}" does not match session id "${id}"`)
|
||||
}
|
||||
if (typeof record.createdAt !== 'number' || !Number.isFinite(record.createdAt)) {
|
||||
throw new Error('session header createdAt must be a finite number')
|
||||
if (typeof record.createdAt !== 'number'
|
||||
|| !Number.isSafeInteger(record.createdAt)
|
||||
|| record.createdAt < 0) {
|
||||
throw new Error('session header createdAt must be a non-negative safe integer')
|
||||
}
|
||||
if (record.cwd !== undefined) {
|
||||
if (typeof record.cwd !== 'string') throw new Error('session header cwd must be a string')
|
||||
|
||||
@@ -36,7 +36,7 @@ export interface SessionHeader {
|
||||
readonly version: number
|
||||
/** The session's id (mirrors the {@link Session}'s id). */
|
||||
readonly id: SessionId
|
||||
/** Unix epoch milliseconds when the session was created. */
|
||||
/** Non-negative safe-integer Unix epoch milliseconds when the session was created. */
|
||||
readonly createdAt: number
|
||||
/** Absolute working directory the session was created in (if any). */
|
||||
readonly cwd?: string
|
||||
|
||||
@@ -757,7 +757,7 @@ describe('Session', () => {
|
||||
{ header: 1, error: /not a plain JSON record/ },
|
||||
{ header: null, error: /not a plain JSON record/ },
|
||||
{ header: { ...base, version: 1 }, error: /header version/ },
|
||||
{ header: { ...base, createdAt: '123' }, error: /createdAt must be a finite number/ },
|
||||
{ header: { ...base, createdAt: '123' }, error: /createdAt must be a non-negative safe integer/ },
|
||||
{ header: { ...base, cwd: 1 }, error: /header cwd must be a string/ },
|
||||
{ header: { ...base, cwd: 'relative' }, error: /header cwd must be an absolute path/ },
|
||||
{ header: { ...base, parentSession: 1 }, error: /header parentSession must be a string/ },
|
||||
@@ -962,7 +962,7 @@ describe('SessionStore', () => {
|
||||
await ctx.plugin(SessionStore)
|
||||
const session = ctx.sessions.create(SessionId('plain'))
|
||||
expect(session.header).toMatchObject({ version: SESSION_FORMAT_VERSION, id: 'plain' })
|
||||
expect(typeof session.header.createdAt).toBe('number')
|
||||
expect(Number.isSafeInteger(session.header.createdAt)).toBe(true)
|
||||
expect(session.header.cwd).toBeUndefined()
|
||||
expect(session.header.parentSession).toBeUndefined()
|
||||
})
|
||||
@@ -1001,7 +1001,10 @@ describe('SessionStore', () => {
|
||||
{ meta: { parentSession: 1n }, error: /header is not losslessly JSON-serializable/ },
|
||||
{ meta: { cwd: 1 }, error: /header cwd must be a string/ },
|
||||
{ meta: { parentSession: 1 }, error: /header parentSession must be a string/ },
|
||||
{ meta: { createdAt: '123' }, error: /header createdAt must be a finite number/ },
|
||||
{ meta: { createdAt: '123' }, error: /header createdAt must be a non-negative safe integer/ },
|
||||
{ meta: { createdAt: 1.5 }, error: /header createdAt must be a non-negative safe integer/ },
|
||||
{ meta: { createdAt: -1 }, error: /header createdAt must be a non-negative safe integer/ },
|
||||
{ meta: { createdAt: Number.MAX_SAFE_INTEGER + 1 }, error: /header createdAt must be a non-negative safe integer/ },
|
||||
{ meta: { seedLength: '1' }, error: /seedLength must be a non-negative safe integer/ },
|
||||
{ meta: { seedLength: 0.5 }, error: /seedLength must be a non-negative safe integer/ },
|
||||
{ meta: { seedLength: -1 }, error: /seedLength must be a non-negative safe integer/ },
|
||||
|
||||
@@ -84,6 +84,8 @@ function isHeaderLine(value: unknown): value is HeaderLine {
|
||||
&& typeof (value as { version?: unknown }).version === 'number'
|
||||
&& typeof (value as { id?: unknown }).id === 'string'
|
||||
&& typeof (value as { createdAt?: unknown }).createdAt === 'number'
|
||||
&& Number.isSafeInteger((value as { createdAt: number }).createdAt)
|
||||
&& (value as { createdAt: number }).createdAt >= 0
|
||||
&& typeof (value as { delegationDepth?: unknown }).delegationDepth === 'number'
|
||||
&& Number.isSafeInteger((value as { delegationDepth: number }).delegationDepth)
|
||||
&& (value as { delegationDepth: number }).delegationDepth >= 0
|
||||
|
||||
@@ -559,6 +559,21 @@ describe('SessionPersistenceJsonl: scanLog unit', () => {
|
||||
expect(() => scanLog(Buffer.from('{"type":"event"}\n'))).toThrow(/session header/)
|
||||
})
|
||||
|
||||
it.each([
|
||||
['fractional', 1.5],
|
||||
['negative', -1],
|
||||
['unsafe', Number.MAX_SAFE_INTEGER + 1],
|
||||
])('rejects a session header with a %s createdAt', (_label, createdAt) => {
|
||||
const log = JSON.stringify({
|
||||
type: 'session',
|
||||
version: 0,
|
||||
id: 'invalid-created-at',
|
||||
createdAt,
|
||||
delegationDepth: 0,
|
||||
}) + '\n'
|
||||
expect(() => scanLog(Buffer.from(log))).toThrow(/session header/)
|
||||
})
|
||||
|
||||
it.each([
|
||||
['missing', undefined],
|
||||
['a string', '1'],
|
||||
|
||||
@@ -8,9 +8,9 @@ A SQLite durable session-persistence backend — a second `SessionPersistence` i
|
||||
|
||||
## Storage model
|
||||
|
||||
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).
|
||||
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; `createdAt` is a non-negative safe integer stored in a strict `INTEGER` column. 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. 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.
|
||||
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 application_id` identifies the canonical persistence database, and `PRAGMA user_version` stores its layout version. A fresh database must have no application identity or user-defined schema objects; initialization creates every table and stamps both pragmas in one transaction. Non-pristine unversioned databases, foreign application identities, and every non-current version reject before journal-mode mutation because this unreleased format has no migrations.
|
||||
|
||||
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 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).
|
||||
- **Only a pristine new database or the current owned `SCHEMA_VERSION` opens** — unversioned schema objects, foreign application identities, and every other schema version are 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,10 @@ 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 = 9
|
||||
export const SCHEMA_VERSION = 10
|
||||
|
||||
/** SQLite application id protecting unrelated databases from persistence writes. */
|
||||
export const SESSION_PERSISTENCE_SQLITE_APPLICATION_ID = 0x44534850
|
||||
|
||||
/**
|
||||
* A row of the `sessions` table — the out-of-log metadata ({@link SessionHeader}).
|
||||
@@ -86,53 +89,77 @@ function configureDatabase(db: DatabaseSync, path: string, journalMode: JournalM
|
||||
db.exec('PRAGMA foreign_keys = ON')
|
||||
// `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_%'",
|
||||
const { application_id: applicationId } = db.prepare('PRAGMA application_id').get() as { application_id: number }
|
||||
const { count: userObjectCount } = db.prepare(
|
||||
"SELECT COUNT(*) AS count FROM sqlite_schema WHERE 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 && (applicationId !== 0 || userObjectCount > 0)) {
|
||||
throw new Error(`session database at "${path}" has an unversioned schema or application identity`)
|
||||
}
|
||||
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 === SCHEMA_VERSION && applicationId !== SESSION_PERSISTENCE_SQLITE_APPLICATION_ID) {
|
||||
throw new Error(
|
||||
`session database at "${path}" has application id ${applicationId}, expected ${SESSION_PERSISTENCE_SQLITE_APPLICATION_ID}`,
|
||||
)
|
||||
}
|
||||
// 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),
|
||||
store_id TEXT NOT NULL
|
||||
) STRICT
|
||||
`)
|
||||
db.prepare(
|
||||
'INSERT OR IGNORE INTO persistence_state (singleton, store_id) VALUES (1, ?)',
|
||||
).run(randomUUID())
|
||||
db.exec(`
|
||||
CREATE TABLE IF NOT EXISTS sessions (
|
||||
id TEXT PRIMARY KEY,
|
||||
version INTEGER NOT NULL,
|
||||
created_at REAL NOT NULL,
|
||||
cwd TEXT,
|
||||
parent_session TEXT,
|
||||
seed_length INTEGER,
|
||||
delegation_depth INTEGER,
|
||||
incarnation TEXT NOT NULL,
|
||||
revision INTEGER NOT NULL
|
||||
) STRICT
|
||||
`)
|
||||
db.exec(`
|
||||
CREATE TABLE IF NOT EXISTS events (
|
||||
session_id TEXT NOT NULL REFERENCES sessions(id) ON DELETE CASCADE,
|
||||
seq INTEGER NOT NULL,
|
||||
type TEXT NOT NULL,
|
||||
time INTEGER NOT NULL,
|
||||
data TEXT NOT NULL,
|
||||
source_event_seqs TEXT,
|
||||
surface_op TEXT,
|
||||
PRIMARY KEY (session_id, seq)
|
||||
) STRICT
|
||||
`)
|
||||
if (onDisk === 0) db.exec(`PRAGMA user_version = ${SCHEMA_VERSION}`)
|
||||
let began = false
|
||||
try {
|
||||
db.exec('BEGIN IMMEDIATE')
|
||||
began = true
|
||||
db.exec(`
|
||||
CREATE TABLE IF NOT EXISTS persistence_state (
|
||||
singleton INTEGER PRIMARY KEY CHECK (singleton = 1),
|
||||
store_id TEXT NOT NULL
|
||||
) STRICT;
|
||||
|
||||
CREATE TABLE IF NOT EXISTS sessions (
|
||||
id TEXT PRIMARY KEY,
|
||||
version INTEGER NOT NULL,
|
||||
created_at INTEGER NOT NULL,
|
||||
cwd TEXT,
|
||||
parent_session TEXT,
|
||||
seed_length INTEGER,
|
||||
delegation_depth INTEGER,
|
||||
incarnation TEXT NOT NULL,
|
||||
revision INTEGER NOT NULL
|
||||
) STRICT;
|
||||
|
||||
CREATE TABLE IF NOT EXISTS events (
|
||||
session_id TEXT NOT NULL REFERENCES sessions(id) ON DELETE CASCADE,
|
||||
seq INTEGER NOT NULL,
|
||||
type TEXT NOT NULL,
|
||||
time INTEGER NOT NULL,
|
||||
data TEXT NOT NULL,
|
||||
source_event_seqs TEXT,
|
||||
surface_op TEXT,
|
||||
PRIMARY KEY (session_id, seq)
|
||||
) STRICT
|
||||
`)
|
||||
db.prepare(
|
||||
'INSERT OR IGNORE INTO persistence_state (singleton, store_id) VALUES (1, ?)',
|
||||
).run(randomUUID())
|
||||
if (onDisk === 0) {
|
||||
db.exec(`PRAGMA application_id = ${SESSION_PERSISTENCE_SQLITE_APPLICATION_ID}`)
|
||||
db.exec(`PRAGMA user_version = ${SCHEMA_VERSION}`)
|
||||
}
|
||||
db.exec('COMMIT')
|
||||
} catch (error: unknown) {
|
||||
/* v8 ignore next -- a BEGIN failure leaves no transaction to roll back. */
|
||||
if (began) {
|
||||
/* v8 ignore next 5 -- preserve the original schema failure if SQLite also refuses rollback. */
|
||||
try {
|
||||
db.exec('ROLLBACK')
|
||||
} catch {
|
||||
// The original SQLite failure remains the actionable cause.
|
||||
}
|
||||
}
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -141,6 +168,9 @@ function configureDatabase(db: DatabaseSync, path: string, journalMode: JournalM
|
||||
* @returns the header, `NULL` columns mapped to omitted optional fields.
|
||||
*/
|
||||
export function rowToMeta(row: SessionRow): SessionHeader {
|
||||
if (!Number.isSafeInteger(row.created_at) || row.created_at < 0) {
|
||||
throw new Error('stored session createdAt must be a non-negative safe integer')
|
||||
}
|
||||
return {
|
||||
version: row.version,
|
||||
id: row.id as SessionId,
|
||||
|
||||
@@ -8,7 +8,14 @@ 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'
|
||||
import { openDatabase, rowToEvent, scanRows, type EventRow } from '../src/schema.ts'
|
||||
import {
|
||||
openDatabase,
|
||||
rowToEvent,
|
||||
rowToMeta,
|
||||
scanRows,
|
||||
SESSION_PERSISTENCE_SQLITE_APPLICATION_ID,
|
||||
type EventRow,
|
||||
} from '../src/schema.ts'
|
||||
import { runPersistenceContract, meta, oneTurnLog, appendLog } from '../../session-persistence/tests/contract.ts'
|
||||
import { runCoordinatorContract, type CoordinatorFixture } from '../../session-persistence/tests/coordinator-contract.ts'
|
||||
|
||||
@@ -151,6 +158,22 @@ describe('scanRows', () => {
|
||||
})
|
||||
})
|
||||
|
||||
describe('rowToMeta', () => {
|
||||
it('rejects fractional stored creation metadata', () => {
|
||||
expect(() => rowToMeta({
|
||||
id: 'fractional',
|
||||
version: 0,
|
||||
created_at: 1.5,
|
||||
cwd: null,
|
||||
parent_session: null,
|
||||
seed_length: null,
|
||||
incarnation: 'fractional',
|
||||
revision: 1,
|
||||
delegation_depth: null,
|
||||
})).toThrow('stored session createdAt must be a non-negative safe integer')
|
||||
})
|
||||
})
|
||||
|
||||
describe('SessionPersistenceSqlite: durability and crash semantics', () => {
|
||||
it('rejects a stored v0 log containing a legacy request/header-delta event', async () => {
|
||||
const path = await freshDbPath()
|
||||
@@ -305,13 +328,13 @@ 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 () => {
|
||||
it('rejects a table-backed 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/)
|
||||
expect(() => openDatabase(path, 'wal')).toThrow(/unversioned schema or application identity/)
|
||||
|
||||
const unchanged = new DatabaseSync(path)
|
||||
expect(unchanged.prepare('PRAGMA user_version').get()).toEqual({ user_version: 0 })
|
||||
@@ -322,6 +345,86 @@ describe('SessionPersistenceSqlite: durability and crash semantics', () => {
|
||||
unchanged.close()
|
||||
})
|
||||
|
||||
it('rejects view-only and foreign-application unversioned databases without mutation', async () => {
|
||||
const viewPath = await freshDbPath()
|
||||
const viewOnly = new DatabaseSync(viewPath)
|
||||
viewOnly.exec('CREATE VIEW foreign_view AS SELECT 1 AS value')
|
||||
viewOnly.close()
|
||||
|
||||
expect(() => openDatabase(viewPath, 'wal')).toThrow(/unversioned schema or application identity/)
|
||||
const unchangedView = new DatabaseSync(viewPath)
|
||||
expect(unchangedView.prepare('PRAGMA journal_mode').get()).toEqual({ journal_mode: 'delete' })
|
||||
expect(unchangedView.prepare(
|
||||
"SELECT type FROM sqlite_schema WHERE name = 'foreign_view'",
|
||||
).get()).toEqual({ type: 'view' })
|
||||
unchangedView.close()
|
||||
|
||||
const applicationPath = await freshDbPath()
|
||||
const foreignApplication = new DatabaseSync(applicationPath)
|
||||
foreignApplication.exec('PRAGMA application_id = 12345')
|
||||
foreignApplication.close()
|
||||
|
||||
expect(() => openDatabase(applicationPath, 'wal')).toThrow(/unversioned schema or application identity/)
|
||||
const unchangedApplication = new DatabaseSync(applicationPath)
|
||||
expect(unchangedApplication.prepare('PRAGMA application_id').get()).toEqual({ application_id: 12345 })
|
||||
expect(unchangedApplication.prepare('PRAGMA user_version').get()).toEqual({ user_version: 0 })
|
||||
expect(unchangedApplication.prepare('PRAGMA journal_mode').get()).toEqual({ journal_mode: 'delete' })
|
||||
unchangedApplication.close()
|
||||
})
|
||||
|
||||
it('rejects a current-version database with a foreign application identity', async () => {
|
||||
const path = await freshDbPath()
|
||||
const foreign = new DatabaseSync(path)
|
||||
foreign.exec('PRAGMA application_id = 12345')
|
||||
foreign.exec(`PRAGMA user_version = ${SCHEMA_VERSION}`)
|
||||
foreign.close()
|
||||
|
||||
expect(() => openDatabase(path, 'wal')).toThrow(/has application id 12345/)
|
||||
|
||||
const unchanged = new DatabaseSync(path)
|
||||
expect(unchanged.prepare('PRAGMA application_id').get()).toEqual({ application_id: 12345 })
|
||||
expect(unchanged.prepare('PRAGMA user_version').get()).toEqual({ user_version: SCHEMA_VERSION })
|
||||
expect(unchanged.prepare('PRAGMA journal_mode').get()).toEqual({ journal_mode: 'delete' })
|
||||
unchanged.close()
|
||||
})
|
||||
|
||||
it('rolls back schema objects and identity stamps when initialization fails', async () => {
|
||||
const path = await freshDbPath()
|
||||
const conflicting = new DatabaseSync(path)
|
||||
conflicting.exec(`PRAGMA application_id = ${SESSION_PERSISTENCE_SQLITE_APPLICATION_ID}`)
|
||||
conflicting.exec(`PRAGMA user_version = ${SCHEMA_VERSION}`)
|
||||
conflicting.exec("CREATE VIEW persistence_state AS SELECT 1 AS singleton, 'foreign' AS store_id")
|
||||
conflicting.close()
|
||||
|
||||
expect(() => openDatabase(path, 'wal')).toThrow()
|
||||
|
||||
const unchanged = new DatabaseSync(path)
|
||||
expect(unchanged.prepare(
|
||||
"SELECT type FROM sqlite_schema WHERE name = 'persistence_state'",
|
||||
).get()).toEqual({ type: 'view' })
|
||||
expect(unchanged.prepare(
|
||||
"SELECT type FROM sqlite_schema WHERE name = 'sessions'",
|
||||
).get()).toBeUndefined()
|
||||
expect(unchanged.prepare(
|
||||
"SELECT type FROM sqlite_schema WHERE name = 'events'",
|
||||
).get()).toBeUndefined()
|
||||
expect(unchanged.prepare('PRAGMA application_id').get())
|
||||
.toEqual({ application_id: SESSION_PERSISTENCE_SQLITE_APPLICATION_ID })
|
||||
expect(unchanged.prepare('PRAGMA user_version').get()).toEqual({ user_version: SCHEMA_VERSION })
|
||||
unchanged.close()
|
||||
})
|
||||
|
||||
it('stamps the persistence application identity with the schema version', async () => {
|
||||
const path = await freshDbPath()
|
||||
openDatabase(path, 'wal').close()
|
||||
|
||||
const db = new DatabaseSync(path)
|
||||
expect(db.prepare('PRAGMA application_id').get())
|
||||
.toEqual({ application_id: SESSION_PERSISTENCE_SQLITE_APPLICATION_ID })
|
||||
expect(db.prepare('PRAGMA user_version').get()).toEqual({ user_version: SCHEMA_VERSION })
|
||||
db.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()
|
||||
@@ -460,7 +563,7 @@ describe('SessionPersistenceSqlite: durability and crash semantics', () => {
|
||||
})
|
||||
|
||||
it('exposes the schema version constant', () => {
|
||||
expect(SCHEMA_VERSION).toBe(9)
|
||||
expect(SCHEMA_VERSION).toBe(10)
|
||||
})
|
||||
|
||||
it('keeps the revision stable for an empty repair hook', async () => {
|
||||
|
||||
@@ -178,6 +178,9 @@ export class PersistenceCoordinator<TornMarker = unknown> {
|
||||
if (snapshot === undefined) {
|
||||
return Promise.reject(new TypeError('session metadata must be losslessly JSON-serializable'))
|
||||
}
|
||||
if (!Number.isSafeInteger(snapshot.createdAt) || snapshot.createdAt < 0) {
|
||||
return Promise.reject(new TypeError('session metadata createdAt must be a non-negative safe integer'))
|
||||
}
|
||||
return this.serialize(snapshot.id, () => this.createCore(snapshot))
|
||||
}
|
||||
|
||||
|
||||
@@ -84,15 +84,17 @@ export function runPersistenceContract(name: string, make: () => Promise<Contrac
|
||||
}
|
||||
})
|
||||
|
||||
it('round-trips a finite fractional creation timestamp', async () => {
|
||||
it('rejects a fractional creation timestamp without reserving its session id', 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())
|
||||
await expect(persistence.create(m))
|
||||
.rejects.toThrow('session metadata createdAt must be a non-negative safe integer')
|
||||
|
||||
const loaded = await persistence.load(m.id)
|
||||
expect(loaded.meta.createdAt).toBe(1.5)
|
||||
const valid = meta('fractional-created-at')
|
||||
await persistence.create(valid)
|
||||
await persistence.append(valid.id, oneTurnLog())
|
||||
expect((await persistence.load(valid.id)).meta.createdAt).toBe(valid.createdAt)
|
||||
} finally {
|
||||
await dispose()
|
||||
}
|
||||
|
||||
@@ -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 = 4
|
||||
export const SESSION_QUERY_SQLITE_SCHEMA_VERSION = 5
|
||||
|
||||
/** 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 REAL NOT NULL,
|
||||
created_at INTEGER 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 REAL NOT NULL,
|
||||
created_at INTEGER NOT NULL,
|
||||
cwd TEXT,
|
||||
parent_session TEXT,
|
||||
seed_length INTEGER,
|
||||
|
||||
@@ -167,22 +167,6 @@ 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