fix(session-persistence-sqlite): defer crash-tail repair to append; fix all-tail materialized flag; persist schema version (review #35)

- load() no longer DELETEs the crash tail — it stays non-mutating w.r.t.
  the event log and records a repair point (repairFrom). The next append
  runs the DELETE inside its own transaction before inserting. This makes
  the SQLite backend honor the SAME public contract as JSONL (load returns
  the committed prefix; the subsequent append performs the one-time
  physical truncation-repair), instead of mutating during load.

- All-tail load: when the discarded crash tail was the session's only
  committed content (committed.length === 0), the metadata row still read
  materialized = 1 from the prior append, so has()/list() reported a
  session load() had just emptied. load() now flips the row's materialized
  flag to 0 (metadata only — the orphaned event rows are still removed by
  the deferred repair), so has()/list() are immediately consistent.

- Schema version: openDatabase now stores SCHEMA_VERSION in PRAGMA
  user_version on a fresh database and rejects opening one whose
  user_version is newer than this build supports, protecting against a
  future incompatible layout.

Regression tests: load is non-mutating (tail rows survive until the next
append), all-tail load makes has()/list() false, and a newer-schema
database is rejected on open.

packages/session-persistence-sqlite/README.md

@@ -6,13 +6,13 @@ 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)` — `data` is the event payload as JSON text, so the row shape is the event verbatim (including `assistant/chunk`, keeping `seq` contiguous). Out-of-log metadata (`SessionMeta`) lives in a `sessions` row, including the mutable `SessionSummary` fields (`updatedAt`, `title`, `firstPrompt`) that `update()` rewrites without touching the event log.
 
-`node:sqlite` requires Node ≥ 22.5 (this repo runs Node ≥ 24); the database opens with `foreign_keys = ON` (so `ON DELETE CASCADE` drops a session's events with its row) and `journal_mode = WAL`.
+`node:sqlite` requires Node ≥ 22.5 (this repo runs Node ≥ 24); the database opens with `foreign_keys = ON` (so `ON DELETE CASCADE` drops a session's events with its row) and `journal_mode = WAL`. The table-layout version is stored in `PRAGMA user_version` and checked on open: a fresh database is stamped with the current `SCHEMA_VERSION`; a database written by a newer, incompatible build (higher `user_version`) is rejected rather than opened against an unknown layout.
 
 ## Contract semantics over rows
 
-- **Append = a transaction.** `append` runs `BEGIN`/`COMMIT` around the batch: it materializes the `sessions` row (if still lazy) and INSERTs every event, asserting the contiguous-seq contract first (the first event's `seq` must equal the stored next-seq). A mid-batch failure (a UNIQUE violation on a duplicated seq) rolls back entirely, so the stored log and the in-memory cursor stay consistent.
+- **Append = a transaction.** `append` runs `BEGIN`/`COMMIT` around the batch: it runs any deferred crash-tail repair, materializes the `sessions` row (if still lazy), and INSERTs every event, asserting the contiguous-seq contract first (the first event's `seq` must equal the stored next-seq). A mid-batch failure (a UNIQUE violation on a duplicated seq) rolls back entirely, so the stored log and the in-memory cursor stay consistent.
 - **Lazy materialization.** `create()` records intent in memory only — no row is written until the first `append`. A created-but-never-appended session is absent from `has()`/`list()` (a `materialized` flag on the row, set inside the first append transaction; `has`/`list` filter to materialized rows).
-- **Crash-tail-on-load.** `load()` reads every stored event ordered by `seq` and returns only the prefix through the **last complete `turn/end`** (the `SessionPersistence.load` contract). A batch that landed without its closing `turn/end` (a process killed mid-turn) is an uncommitted tail and is deleted on load; a `seq` gap inside the committed region makes the session unloadable.
+- **Crash-tail-on-load.** `load()` reads every stored event ordered by `seq` and returns only the prefix through the **last complete `turn/end`** (the `SessionPersistence.load` contract), computed from the `seq`/`type` columns so a malformed `data` in the uncommitted tail is never parsed. A batch that landed without its closing `turn/end` (a process killed mid-turn) is an uncommitted tail: `load()` stays non-mutating w.r.t. the event log and records a repair point; the **next `append`** physically DELETEs the orphaned rows inside its transaction (the one-time truncation-repair, matching the JSONL backend and the abstract contract). A `seq` gap inside the committed region makes the session unloadable. If the discarded tail was the session's only committed content, `load()` also flips the metadata row's `materialized` flag to 0 so `has()`/`list()` immediately stop reporting the now-empty session.
 
 ## Configuration (schemastery)
 

packages/session-persistence-sqlite/src/index.ts

@@ -50,6 +50,14 @@ interface SessionState {
   cursor: number
   /** Whether the session has at least one persisted event (materialized). */
   materialized: boolean
+  /**
+   * If a load found a crash tail, the seq from which the next {@link append}
+   * must DELETE before inserting (the one-time truncation-repair). load() stays
+   * non-mutating w.r.t. the event log — it only records this marker — so the
+   * public contract matches the JSONL backend: load returns the committed
+   * prefix; the subsequent append performs the physical repair.
+   */
+  repairFrom?: number
   /** The live Session that owns this state (collision detection); see onCreated. */
   owner?: Session
 }
@@ -171,16 +179,24 @@ export class SessionPersistenceSqlite extends SessionPersistence {
       }
     }
 
-    // The transaction is the durability + atomicity boundary: materialize the
-    // sessions row (if lazy) and INSERT every event, or roll back entirely. A
-    // BEGIN/COMMIT around the batch means a mid-batch failure (a UNIQUE
-    // violation on a duplicated seq from a concurrent writer) leaves the stored
-    // log untouched, so the cursor stays truthful and a retry is clean.
+    // The transaction is the durability + atomicity boundary: run any deferred
+    // crash-tail repair, materialize the sessions row (if lazy), and INSERT
+    // every event, or roll back entirely. A BEGIN/COMMIT around the batch means
+    // a mid-batch failure (a UNIQUE violation on a duplicated seq from a
+    // concurrent writer) leaves the stored log untouched, so the cursor stays
+    // truthful and a retry is clean.
     const insertEvent = this.db.prepare(
       'INSERT INTO events (session_id, seq, type, time, data) VALUES (?, ?, ?, ?, ?)',
     )
     this.db.exec('BEGIN')
     try {
+      // One-time truncation-repair: a prior load() found a crash tail and
+      // deferred its physical removal to here (load stays non-mutating). DELETE
+      // the orphaned rows (seq >= repairFrom) before inserting, inside the same
+      // transaction, so the repair + first new append commit atomically.
+      if (state.repairFrom !== undefined) {
+        this.db.prepare('DELETE FROM events WHERE session_id = ? AND seq >= ?').run(id, state.repairFrom)
+      }
       if (!state.materialized) this.writeRow(state.meta)
       for (const event of events) {
         insertEvent.run(id, event.seq, event.type, event.time, JSON.stringify(event.data))
@@ -194,6 +210,7 @@ export class SessionPersistenceSqlite extends SessionPersistence {
       this.db.exec('ROLLBACK')
       throw error
     }
+    delete state.repairFrom
     state.materialized = true
     state.cursor += events.length
   }
@@ -223,18 +240,33 @@ export class SessionPersistenceSqlite extends SessionPersistence {
     const { committed, cutTail } = cutAtLastTurnEnd(eventRows)
     const events = committed.map(rowToEvent)
 
-    // Physically discard the crash tail so the stored log matches what load
-    // returned (the next append continues at the committed length). Mirrors the
-    // JSONL truncation-repair, but done eagerly here (a DELETE is transactional;
-    // there is no half-written-line hazard to defer past).
-    if (cutTail) {
-      this.db.prepare('DELETE FROM events WHERE session_id = ? AND seq >= ?').run(id, committed.length)
+    // Do NOT delete the crash tail here: load() stays non-mutating w.r.t. the
+    // event log, matching the abstract contract and the JSONL backend (load
+    // returns the committed prefix; the next append performs the one-time
+    // physical repair). Record the repair point so the next appendCore DELETEs
+    // the orphaned tail inside its own transaction before inserting.
+    const materialized = committed.length > 0
+    if (committed.length === 0 && row.materialized === 1) {
+      // All-tail discard: the only committed events were a crash tail, so the
+      // session now has NO committed events. The metadata row, however, still
+      // reads materialized = 1 from the prior append — which would make has()
+      // and list() report a session that load() just emptied. Correct the
+      // materialized FLAG (metadata, not the event log) so has()/list() are
+      // immediately consistent. The orphaned tail rows are still removed by the
+      // deferred repair on the next append.
+      this.db.prepare('UPDATE sessions SET materialized = 0 WHERE id = ?').run(id)
     }
 
-    // Record state so a later append continues at the committed length. The
-    // state keeps its OWN copy of the meta; the returned value is separate so a
-    // consumer mutating loaded.meta cannot corrupt the backend's row metadata.
-    this.states.set(id, { meta: { ...meta }, cursor: committed.length, materialized: committed.length > 0 })
+    // Record state so a later append continues at the committed length and runs
+    // the deferred tail repair. The state keeps its OWN copy of the meta; the
+    // returned value is separate so a consumer mutating loaded.meta cannot
+    // corrupt the backend's row metadata.
+    this.states.set(id, {
+      meta: { ...meta },
+      cursor: committed.length,
+      materialized,
+      ...cutTail ? { repairFrom: committed.length } : {},
+    })
     return { meta, events }
   }
 

packages/session-persistence-sqlite/src/schema.ts

@@ -48,11 +48,30 @@ export interface EventRow {
  * makes `ON DELETE CASCADE` drop a session's events with its row; `journal_mode
  * = WAL` matches the durability model the ADR records (the row shape maps 1:1
  * onto `SessionEvent`; opencode runs this exact shape on SQLite/WAL).
+ *
+ * The table-layout version is persisted in SQLite's `PRAGMA user_version` and
+ * checked on open: a fresh database (user_version 0) is stamped with the
+ * current {@link SCHEMA_VERSION}; an existing database with a NEWER version
+ * (written by a future, incompatible build) is rejected rather than opened
+ * against a layout this build does not understand. (An older-but-compatible
+ * version would be migrated here when migrations exist; v1 has none.)
  */
 export function openDatabase(path: string): DatabaseSync {
   const db = new DatabaseSync(path)
   db.exec('PRAGMA foreign_keys = ON')
   db.exec('PRAGMA journal_mode = WAL')
+  // `PRAGMA user_version` always returns exactly one row { user_version }.
+  const { user_version: onDisk } = db.prepare('PRAGMA user_version').get() as { user_version: number }
+  if (onDisk > SCHEMA_VERSION) {
+    db.close()
+    throw new Error(`session database at "${path}" has schema version ${onDisk}, newer than this build supports (${SCHEMA_VERSION})`)
+  }
+  if (onDisk === 0) {
+    // Fresh (or pre-versioning) database: stamp the current layout version.
+    // PRAGMA does not accept bound parameters, so interpolate the integer
+    // constant (SCHEMA_VERSION is a trusted in-code number, not user input).
+    db.exec(`PRAGMA user_version = ${SCHEMA_VERSION}`)
+  }
   db.exec(`
     CREATE TABLE IF NOT EXISTS sessions (
       id             TEXT PRIMARY KEY,

packages/session-persistence-sqlite/tests/sqlite.spec.ts

@@ -80,7 +80,7 @@ describe('cutAtLastTurnEnd', () => {
 })
 
 describe('SessionPersistenceSqlite: durability and crash semantics', () => {
-  it('a crash tail (rows after the last turn/end) is excluded and deleted on load', async () => {
+  it('a crash tail (rows after the last turn/end) is excluded on load and repaired on the next append', async () => {
     const path = await freshDbPath()
     const m = meta('crash')
     // Run 1: persist a complete turn, then a half-written second turn (no turn/end).
@@ -95,15 +95,16 @@ describe('SessionPersistenceSqlite: durability and crash semantics', () => {
     ])
     await fiber1.dispose()
 
-    // Run 2: load returns only the committed first turn; the tail is gone.
+    // Run 2: load returns only the committed first turn (tail excluded).
     const ctx2 = new Context()
     await ctx2.plugin(SessionStore)
     const fiber2 = await ctx2.plugin(SessionPersistenceSqlite, { path })
     const loaded = await ctx2.sessionPersistence.load(m.id)
     expect(loaded.events).toEqual(oneTurnLog())
 
-    // The next append continues at seq 6 (the committed length) and the cut
-    // tail was physically deleted, so there is no UNIQUE collision.
+    // The next append continues at seq 6 and performs the deferred truncation-
+    // repair inside its transaction (DELETE seq >= 6 before inserting), so the
+    // orphaned tail rows are gone and there is no UNIQUE collision.
     await ctx2.sessionPersistence.append(m.id, [
       { type: 'turn/start', seq: 6, time: 9, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } },
       { type: 'turn/end', seq: 7, time: 10, data: { turn: 2, reason: { kind: 'completed' } } },
@@ -113,6 +114,64 @@ describe('SessionPersistenceSqlite: durability and crash semantics', () => {
     await fiber2.dispose()
   })
 
+  it('load() is non-mutating: the crash tail rows survive until the next append repairs them', async () => {
+    const path = await freshDbPath()
+    const m = meta('load-nonmutating')
+    const b1 = await backend(path)
+    await b1.ctx.sessionPersistence.create(m)
+    await b1.ctx.sessionPersistence.append(m.id, oneTurnLog()) // seqs 0..5
+    await b1.dispose()
+    // Hand-write an uncommitted tail (seq 6, no turn/end).
+    const db = openDatabase(path)
+    db.prepare('INSERT INTO events (session_id, seq, type, time, data) VALUES (?, 6, ?, 7, ?)')
+      .run(m.id, 'turn/start', JSON.stringify({ turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } }))
+    db.close()
+
+    const b2 = await backend(path)
+    const loaded = await b2.ctx.sessionPersistence.load(m.id)
+    expect(loaded.events).toEqual(oneTurnLog())
+    // load() must NOT have deleted the tail row (contract: load returns the
+    // prefix; the next append repairs). Verify the row is still on disk.
+    const probe = openDatabase(path)
+    const tailRows = probe.prepare('SELECT seq FROM events WHERE session_id = ? AND seq >= 6').all(m.id)
+    probe.close()
+    expect(tailRows).toHaveLength(1)
+    await b2.dispose()
+  })
+
+  it('all-tail load: a session whose only content is a crash tail is absent from has()/list()', async () => {
+    const path = await freshDbPath()
+    const m = meta('all-tail')
+    const b1 = await backend(path)
+    await b1.ctx.sessionPersistence.create(m)
+    // A first turn that NEVER completed: turn/start + user/message, no turn/end.
+    await b1.ctx.sessionPersistence.append(m.id, [
+      { type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } },
+      { type: 'user/message', seq: 1, time: 2, data: { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } } },
+    ])
+    expect(await b1.ctx.sessionPersistence.has(m.id)).toBe(true) // materialized
+    await b1.dispose()
+
+    // A fresh backend loads it: the committed prefix is empty (no turn/end), so
+    // the session has no committed content. has()/list() must NOT report it.
+    const b2 = await backend(path)
+    const loaded = await b2.ctx.sessionPersistence.load(m.id)
+    expect(loaded.events).toEqual([])
+    expect(await b2.ctx.sessionPersistence.has(m.id)).toBe(false)
+    expect((await b2.ctx.sessionPersistence.list()).map(x => x.id)).not.toContain(m.id)
+    await b2.dispose()
+  })
+
+  it('rejects opening a database whose schema version is newer than this build', async () => {
+    const path = await freshDbPath()
+    openDatabase(path).close() // stamp user_version = SCHEMA_VERSION
+    // Bump user_version past what this build supports.
+    const db = openDatabase(path)
+    db.exec(`PRAGMA user_version = ${SCHEMA_VERSION + 1}`)
+    db.close()
+    expect(() => openDatabase(path)).toThrow(/newer than this build/)
+  })
+
   it('append snapshots the batch: mutating an event after the call does not corrupt the persisted copy', async () => {
     const ctx = new Context()
     await ctx.plugin(SessionStore)