fix(session-persistence-sqlite): JSONL parity on append snapshot + corrupt-tail load

Two parity gaps with the JSONL backend found in review:

- append() now validates serializability and structuredClones the batch
  synchronously at call time, BEFORE waiting behind the per-session
  chain. A caller that mutates the passed array (or an event inside it)
  after the call can no longer corrupt the persisted copy or advance the
  cursor past what was written. Matches the JSONL backend.

- load() now computes the last-turn/end cut from the seq+type COLUMNS
  only (cutAtLastTurnEnd is generic over {seq,type}); event `data` is
  JSON-parsed only for the committed prefix, never for the uncommitted
  tail. A malformed `data` in a crash tail is discarded, not treated as
  unloadable — only a parse error/gap in the COMMITTED region is
  unloadable (the SessionPersistence.load contract). Matches scanLog.

Regression tests for both.

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

@@ -142,16 +142,25 @@ export class SessionPersistenceSqlite extends SessionPersistence {
     this.states.set(meta.id, { meta, cursor: 0, materialized: false })
   }
 
-  append(id: SessionId, events: readonly SessionEvent[]): Promise<void> {
-    return this.serialize(id, () => this.appendCore(id, events))
+  // `async` so the synchronous validate/clone below reject (not throw) per the
+  // Promise<void> contract — callers use `await expect(...).rejects`.
+  async append(id: SessionId, events: readonly SessionEvent[]): Promise<void> {
+    // Validate serializability BEFORE cloning so a bad event surfaces the typed
+    // "non-JSON-serializable" error rather than an opaque DataCloneError from
+    // structuredClone. Then deep-snapshot the batch HERE, before the op waits
+    // behind the per-session chain: a caller that passes a live array (e.g.
+    // session.events) and mutates it — OR mutates an event inside it — before
+    // the op runs would otherwise have those changes persisted, or advance the
+    // cursor past what was written. The clone is taken at call time (before the
+    // first await), matching the JSONL backend.
+    assertSerializable(events)
+    const batch = events.map(e => structuredClone(e))
+    return this.serialize(id, () => this.appendCore(id, batch))
   }
 
   private async appendCore(id: SessionId, events: readonly SessionEvent[]): Promise<void> {
     await this.ready
     if (events.length === 0) return
-    // Validate serializability up front so a bad event surfaces the typed error
-    // (rather than failing later inside the INSERT loop, mid-transaction).
-    assertSerializable(events)
     let state = this.states.get(id)
     if (state === undefined) state = await this.adopt(id)
 
@@ -200,16 +209,19 @@ export class SessionPersistenceSqlite extends SessionPersistence {
     const meta = rowToMeta(row)
     this.assertVersion(meta)
 
-    // Read every stored event ordered by seq, then cut at the last complete
-    // turn/end — the same crash-tail semantics as the JSONL backend. A row that
-    // landed without its closing turn/end (process killed mid-turn) is an
-    // uncommitted tail and is excluded; a seq gap inside the committed region
-    // makes the session unloadable (cutAtLastTurnEnd throws).
+    // Read every stored row ordered by seq, then cut at the last complete
+    // turn/end — the same crash-tail semantics as the JSONL backend. The cut is
+    // computed from seq+type COLUMNS only, so a malformed `data` in the
+    // uncommitted tail is discarded (not unloadable); only `data` in the
+    // COMMITTED prefix is parsed (rowToEvent), where a parse error correctly
+    // surfaces. A row that landed without its closing turn/end is an
+    // uncommitted tail and is excluded; a seq gap in the committed region makes
+    // the session unloadable (cutAtLastTurnEnd throws).
     const eventRows = this.db
       .prepare('SELECT seq, type, time, data FROM events WHERE session_id = ? ORDER BY seq')
       .all(id) as unknown as EventRow[]
-    const all = eventRows.map(rowToEvent)
-    const { committed, cutTail } = cutAtLastTurnEnd(all)
+    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
@@ -223,7 +235,7 @@ export class SessionPersistenceSqlite extends SessionPersistence {
     // 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 })
-    return { meta, events: committed }
+    return { meta, events }
   }
 
   async list(): Promise<SessionMeta[]> {
@@ -450,7 +462,9 @@ export class SessionPersistenceSqlite extends SessionPersistence {
     const rows = this.db
       .prepare('SELECT seq, type, time, data FROM events WHERE session_id = ? ORDER BY seq')
       .all(id) as unknown as EventRow[]
-    return cutAtLastTurnEnd(rows.map(rowToEvent)).committed
+    // Cut on seq+type columns, then parse `data` only for the committed prefix
+    // (a malformed tail must not throw here — same as loadCore).
+    return cutAtLastTurnEnd(rows).committed.map(rowToEvent)
   }
 
   /** Whether a live session's seed reproduces the first `cursor` stored events. */

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

@@ -105,31 +105,40 @@ export function rowToEvent(row: EventRow): SessionEvent {
 
 /**
  * The committed prefix of an ordered event list: everything up to and including
- * the LAST `turn/end`, plus whether a crash tail (events after it) was cut.
+ * the LAST `turn/end`, plus whether a crash tail (items after it) was cut.
+ *
+ * Generic over anything carrying `seq` + `type` (an {@link EventRow} or a
+ * {@link SessionEvent}) so the cut is computed from those COLUMNS alone — the
+ * caller parses each row's `data` only for the committed items it returns,
+ * never for the tail. This matters for the contract: a malformed `data` in an
+ * uncommitted crash tail must be discarded, not make the session unloadable —
+ * only a parse error / gap in the COMMITTED region is unloadable (see
+ * `SessionPersistence.load`). Mirrors the JSONL backend's `scanLog`, which
+ * likewise tolerates a corrupt tail after the last committed `turn/end`.
  *
  * The loop only flushes at `turn/end`, so the last `turn/end` is the last
  * durable boundary; anything after it is a never-committed crash tail (a batch
  * that landed without its closing `turn/end`, e.g. a process killed mid-turn).
- * This is the SQLite analogue of the JSONL backend's `scanLog` truncation point
- * — the SAME contract (`SessionPersistence.load`), expressed over rows rather
- * than file bytes. The committed region MUST be contiguous (`events[i].seq ===
- * i`); a gap there means committed data was lost and the session is unloadable.
+ * The committed region MUST be contiguous (`item.seq === i`); a gap there means
+ * committed data was lost and the session is unloadable.
  */
-export function cutAtLastTurnEnd(events: readonly SessionEvent[]): { committed: SessionEvent[]; cutTail: boolean } {
+export function cutAtLastTurnEnd<T extends { seq: number; type: string }>(
+  items: readonly T[],
+): { committed: T[]; cutTail: boolean } {
   let lastTurnEnd = -1
-  events.forEach((event, i) => {
-    if (event.type === 'turn/end') lastTurnEnd = i
+  items.forEach((item, i) => {
+    if (item.type === 'turn/end') lastTurnEnd = i
   })
   // No committed turn/end anywhere: the whole list is an uncommitted first-turn
   // tail. Nothing is committed (mirrors scanLog returning zero events).
   if (lastTurnEnd < 0) {
-    return { committed: [], cutTail: events.length > 0 }
+    return { committed: [], cutTail: items.length > 0 }
   }
-  const committed = events.slice(0, lastTurnEnd + 1)
-  committed.forEach((event, i) => {
-    if (event.seq !== i) {
-      throw new Error(`corrupt session log: seq gap in committed region at index ${i} (expected ${i}, got ${event.seq})`)
+  const committed = items.slice(0, lastTurnEnd + 1)
+  committed.forEach((item, i) => {
+    if (item.seq !== i) {
+      throw new Error(`corrupt session log: seq gap in committed region at index ${i} (expected ${i}, got ${item.seq})`)
     }
   })
-  return { committed, cutTail: lastTurnEnd < events.length - 1 }
+  return { committed, cutTail: lastTurnEnd < items.length - 1 }
 }

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

@@ -18,6 +18,14 @@ async function freshDbPath(): Promise<string> {
   return join(dir, 'sessions.db')
 }
 
+/** A context with the session store + SQLite backend, plus a teardown. */
+async function backend(path = ':memory:'): Promise<{ ctx: Context; dispose: () => Promise<void> }> {
+  const ctx = new Context()
+  await ctx.plugin(SessionStore)
+  const fiber = await ctx.plugin(SessionPersistenceSqlite, { path })
+  return { ctx, dispose: () => fiber.dispose() }
+}
+
 // The payoff: the SAME backend-agnostic contract the JSONL backend runs, now
 // proving the SQLite backend satisfies identical semantics.
 runPersistenceContract('sqlite', async () => {
@@ -105,6 +113,61 @@ describe('SessionPersistenceSqlite: durability and crash semantics', () => {
     await fiber2.dispose()
   })
 
+  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)
+    const fiber = await ctx.plugin(SessionPersistenceSqlite, { path: ':memory:' })
+    const m = meta('snapshot')
+    await ctx.sessionPersistence.create(m)
+    const batch: SessionEvent[] = [
+      { 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: 'original' }], source: { kind: 'user' } } },
+      { type: 'turn/end', seq: 2, time: 3, data: { turn: 1, reason: { kind: 'completed' } } },
+    ]
+    const p = ctx.sessionPersistence.append(m.id, batch)
+    // Mutate the live array AND an event's data AFTER the call but before it
+    // drains behind the per-session chain. The snapshot taken at call time must
+    // shield the persisted copy.
+    ;(batch[1]!.data as { content: { type: 'text'; text: string }[] }).content[0]!.text = 'HACKED'
+    batch.push({ type: 'user/message', seq: 3, time: 4, data: { content: [{ type: 'text', text: 'injected' }], source: { kind: 'user' } } })
+    await p
+    const loaded = await ctx.sessionPersistence.load(m.id)
+    expect(loaded.events).toHaveLength(3) // the pushed event was not persisted
+    const um = loaded.events[1]
+    expect(um?.type === 'user/message' && (um.data.content[0] as { text: string }).text).toBe('original')
+    await fiber.dispose()
+  })
+
+  it('a corrupt-JSON row in the uncommitted tail is discarded on load, not unloadable', async () => {
+    const path = await freshDbPath()
+    const m = meta('corrupt-tail')
+    const b1 = await backend(path)
+    await b1.ctx.sessionPersistence.create(m)
+    await b1.ctx.sessionPersistence.append(m.id, oneTurnLog()) // committed: seqs 0..5
+    await b1.dispose()
+
+    // Hand-insert an uncommitted tail row (seq 6, no closing turn/end) whose
+    // `data` is invalid JSON. The contract: only a parse error in the COMMITTED
+    // region is unloadable; a corrupt tail must be discarded (load cuts at the
+    // last turn/end using seq+type columns, never parsing tail `data`).
+    const db = openDatabase(path)
+    db.prepare('INSERT INTO events (session_id, seq, type, time, data) VALUES (?, 6, ?, 7, ?)')
+      .run(m.id, 'turn/start', '{not valid json')
+    db.close()
+
+    const b2 = await backend(path)
+    const loaded = await b2.ctx.sessionPersistence.load(m.id)
+    expect(loaded.events).toEqual(oneTurnLog()) // tail discarded, committed intact
+    // The corrupt tail row was physically deleted, so a fresh append continues.
+    await b2.ctx.sessionPersistence.append(m.id, [
+      { type: 'turn/start', seq: 6, time: 8, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } },
+      { type: 'turn/end', seq: 7, time: 9, data: { turn: 2, reason: { kind: 'completed' } } },
+    ])
+    const reloaded = await b2.ctx.sessionPersistence.load(m.id)
+    expect(reloaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7])
+    await b2.dispose()
+  })
+
   it('append rolls back the whole batch on a mid-batch seq collision (transaction)', async () => {
     const ctx = new Context()
     await ctx.plugin(SessionStore)
@@ -294,13 +357,6 @@ describe('SessionPersistenceSqlite: write path (session/event → flush)', () =>
 })
 
 describe('SessionPersistenceSqlite: edge cases', () => {
-  async function backend(path = ':memory:'): Promise<{ ctx: Context; dispose: () => Promise<void> }> {
-    const ctx = new Context()
-    await ctx.plugin(SessionStore)
-    const fiber = await ctx.plugin(SessionPersistenceSqlite, { path })
-    return { ctx, dispose: () => fiber.dispose() }
-  }
-
   it('append of an empty batch is a no-op', async () => {
     const { ctx, dispose } = await backend()
     const m = meta('empty-batch')