feat(session-persistence-sqlite): preserve interrupted turns on load, don't truncate

Mirror the JSONL backend's crash-recovery contract change: load() now PRESERVES
the real events of an interrupted final turn (a turn can be huge — truncating it
would destroy work) and durably CLOSES the orphaned turn with synthetic boundary
events (step/end if open, then turn/end {interrupted}) inside one transaction
that also deletes any torn tail row. load() is therefore mutating; the deferred
truncation-repair on the next append is gone. Replaces cutAtLastTurnEnd with
scanRows (longest preserved prefix + torn-tail offset). Updates the sqlite tests
and README to the preserve-and-close semantics; the shared runPersistenceContract
suite now holds both backends to identical interrupted-turn behavior.

packages/session-persistence-sqlite/README.md

@@ -1,6 +1,6 @@
 # @deepseek-ai/dsh-session-persistence-sqlite
 
-A SQLite durable session-persistence backend — a second `SessionPersistence` implementation ([ADR 0018](../../docs/adr/0018-session-persistence.md)), built to validate that the abstract seam and the shared `runPersistenceContract` suite are genuinely backend-agnostic. It satisfies the SAME contract as `dsh-session-persistence-jsonl` (append-only, contiguous-seq, lazy materialization, crash-tail-on-load), expressed over `node:sqlite` rows instead of file bytes.
+A SQLite durable session-persistence backend — a second `SessionPersistence` implementation ([ADR 0018](../../docs/adr/0018-session-persistence.md)), built to validate that the abstract seam and the shared `runPersistenceContract` suite are genuinely backend-agnostic. It satisfies the SAME contract as `dsh-session-persistence-jsonl` (append-only, contiguous-seq, lazy materialization, interrupted-turn close on load), expressed over `node:sqlite` rows instead of file bytes.
 
 > **TODO:** this backend talks to `node:sqlite` directly. If a cordis database service (`cordis/db` / a `@cordisjs` SQL driver plugin) is adopted, route through that instead of holding a raw `DatabaseSync` here — the contract surface (`SessionPersistence`) would not change, only the storage driver.
 
@@ -12,9 +12,9 @@ The repo targets Node ≥ 24 (the root `engines` field), which includes the stab
 
 ## Contract semantics over rows
 
-- **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.
+- **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. (`load()` already balanced the stored log, so `append` never has to repair a crash tail.)
 - **Lazy materialization.** `create()` records intent in memory only — no row is written until the first `append`. A created-but-never-appended session has no `sessions` row, so it is absent from `has()`/`list()` (which report exactly the sessions that have a row).
-- **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. A session materialized by a partial first turn (all-tail, zero committed events) keeps its metadata row and stays present in `has()`/`list()` — the same as the JSONL backend, whose file likewise survives a first append that never reached `turn/end`.
+- **Interrupted-turn close on load.** `load()` reads every stored event ordered by `seq` and finds the longest seq-contiguous, parseable prefix — INCLUDING the real events of an interrupted final turn after the last `turn/end` (the loop only flushes at `turn/end`, so a process killed mid-turn leaves real, fully-written rows past it). A single turn can be huge in a long-horizon task, so those events are **preserved, never truncated**: `load()` CLOSES the orphaned turn by durably appending the minimal synthetic boundary events (a `step/end` if a step was open, then a `turn/end` carrying `{ kind: 'interrupted' }`), inside one transaction that also DELETEs any never-fully-written torn tail row. `load()` is therefore mutating — after it the stored rows are balanced and the cursor is truthful, so the next `append` continues cleanly. The boundary (last `turn/end`, torn-tail detection) is computed from the `seq`/`type` columns so a malformed `data` in a torn tail row is never parsed (discarded, not unloadable). A parse error or `seq` gap inside the committed region (at or before the last real `turn/end`) makes the session unloadable. A session whose only turn never closed keeps its metadata row and stays present in `has()`/`list()` — the same as the JSONL backend, whose file likewise survives a first append that never reached `turn/end`.
 
 ## Configuration (schemastery)
 

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

@@ -4,10 +4,10 @@
  * A SECOND {@link SessionPersistence} implementation, built to validate that
  * the abstract seam + the shared `runPersistenceContract` suite are genuinely
  * backend-agnostic: the same append-only / contiguous-seq / lazy-materialization
- * / crash-tail-on-load semantics the JSONL backend expresses over file bytes,
- * expressed here over `node:sqlite` rows. Each `SessionEvent` maps 1:1 onto a
- * row `(session_id, seq, type, time, data)`; `append` is an INSERT inside a
- * transaction that asserts the contiguous-seq contract; the mutable
+ * / interrupted-turn-close-on-load semantics the JSONL backend expresses over
+ * file bytes, expressed here over `node:sqlite` rows. Each `SessionEvent` maps
+ * 1:1 onto a row `(session_id, seq, type, time, data)`; `append` is an INSERT
+ * inside a transaction that asserts the contiguous-seq contract; the mutable
  * `SessionSummary` lives in the `sessions` metadata row.
  *
  * Like the JSONL backend it is also the write-path plugin: it installs the
@@ -25,10 +25,10 @@ import { DatabaseSync } from 'node:sqlite'
 import { mkdir } from 'node:fs/promises'
 import { dirname, resolve } from 'node:path'
 import { SessionPersistence } from '@deepseek-ai/dsh-session-persistence'
-import { isJsonValue } from '@deepseek-ai/dsh-session'
+import { isJsonValue, interruptedTurnClosers } from '@deepseek-ai/dsh-session'
 import type { Session, SessionEvent, SessionId, SessionMeta, SessionSummary } from '@deepseek-ai/dsh-session'
 import {
-  cutAtLastTurnEnd, openDatabase, rowToEvent, rowToMeta, type EventRow, type SessionRow,
+  openDatabase, rowToMeta, scanRows, type EventRow, type SessionRow,
 } from './schema.ts'
 
 export { SCHEMA_VERSION } from './schema.ts'
@@ -50,14 +50,6 @@ 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
 }
@@ -179,24 +171,19 @@ export class SessionPersistenceSqlite extends SessionPersistence {
       }
     }
 
-    // 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.
+    // 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. (A crash
+    // tail is already gone: load() physically deletes the torn fragment and
+    // durably closes the interrupted turn before returning, so by the time any
+    // append runs the stored log is balanced and contiguous.)
     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))
@@ -210,7 +197,6 @@ export class SessionPersistenceSqlite extends SessionPersistence {
       this.db.exec('ROLLBACK')
       throw error
     }
-    delete state.repairFrom
     state.materialized = true
     state.cursor += events.length
   }
@@ -226,42 +212,68 @@ export class SessionPersistenceSqlite extends SessionPersistence {
     const meta = rowToMeta(row)
     this.assertVersion(meta)
 
-    // 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).
+    // Read every stored row ordered by seq, then scan for the preserved prefix:
+    // the longest seq-contiguous, parseable run, INCLUDING the real events of an
+    // interrupted final turn after the last turn/end (a turn can be huge — they
+    // are never truncated). scanRows works off the seq+type COLUMNS for the
+    // last-turn/end boundary, so a malformed `data` in a torn tail row is
+    // discarded (not unloadable); only a parse error / seq gap in the COMMITTED
+    // region (at or before the last turn/end) throws (genuine corruption).
     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 { committed, cutTail } = cutAtLastTurnEnd(eventRows)
-    const events = committed.map(rowToEvent)
+    const { preserved, tornFrom } = scanRows(eventRows)
 
-    // 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.
-    //
-    // The metadata row stays as-is even when committed.length === 0 (an all-tail
-    // crash): the session WAS materialized by the partial append, so its row
-    // exists and has()/list() report it present — the same as the JSONL backend,
-    // whose file likewise survives a first append that never reached turn/end.
-    //
-    // 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.
+    // Crash-recovery (mutating load, same as the JSONL backend): if the log ended
+    // mid-turn, close it DURING load so disk, the returned log, and the cursor all
+    // agree — both append routes then continue with no special-casing. Synthesize
+    // the boundary events (a step/end if a step was open, then a
+    // turn/end {kind:'interrupted'}); the interrupted turn's real events are
+    // preserved, never truncated (ADR 0018).
+    const closers = interruptedTurnClosers(preserved)
+    const balanced = [...preserved, ...closers]
+
+    // Physically repair the stored log inside one transaction: DELETE the torn
+    // tail fragment (if any), then INSERT the synthetic closers. After COMMIT the
+    // stored rows == balanced, so the cursor is truthful and the next append
+    // continues cleanly with no deferred repair. The metadata row stays as-is
+    // even when preserved.length === 0 (an all-tail crash): the session WAS
+    // materialized by the partial append, so has()/list() still report it — the
+    // same as the JSONL backend, whose file likewise survives a first append that
+    // never reached turn/end.
+    if (tornFrom !== undefined || closers.length > 0) {
+      this.db.exec('BEGIN')
+      try {
+        if (tornFrom !== undefined) {
+          this.db.prepare('DELETE FROM events WHERE session_id = ? AND seq >= ?').run(id, tornFrom)
+        }
+        if (closers.length > 0) {
+          const insertEvent = this.db.prepare('INSERT INTO events (session_id, seq, type, time, data) VALUES (?, ?, ?, ?, ?)')
+          for (const event of closers) {
+            insertEvent.run(id, event.seq, event.type, event.time, JSON.stringify(event.data))
+          }
+        }
+        this.db.exec('COMMIT')
+      } catch (error) {
+        // The DELETE+INSERT cannot collide (a row at a closer's seq is preserved
+        // or deleted as torn first); this rolls back a DB-level failure (disk
+        // full, etc.), unreachable in test.
+        /* v8 ignore start */
+        this.db.exec('ROLLBACK')
+        throw error
+        /* v8 ignore stop */
+      }
+    }
+
+    // Record state at the balanced 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,
+      cursor: balanced.length,
       materialized: true,
-      ...cutTail ? { repairFrom: committed.length } : {},
     })
-    return { meta, events }
+    return { meta, events: balanced }
   }
 
   async list(): Promise<SessionMeta[]> {
@@ -482,14 +494,17 @@ export class SessionPersistenceSqlite extends SessionPersistence {
     if (seed.length > 0) await this.append(id, seed)
   }
 
-  /** The committed events for a session id (last-turn/end cut applied). */
+  /** The preserved events for a session id (torn tail excluded, turn NOT yet closed). */
   private eventsFor(id: SessionId): SessionEvent[] {
     const rows = this.db
       .prepare('SELECT seq, type, time, data FROM events WHERE session_id = ? ORDER BY seq')
       .all(id) as unknown as EventRow[]
-    // 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)
+    // Scan on seq+type columns, parsing `data` only for the preserved prefix (a
+    // malformed torn tail must not throw here — same as loadCore). Returns the
+    // preserved events WITHOUT the synthetic closers, so a collision check
+    // compares a live seed against the real on-disk events, mirroring the JSONL
+    // backend's scanLog use in onCreated.
+    return scanRows(rows).preserved
   }
 
   /** Whether a live session's seed reproduces the first `cursor` stored events. */

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

@@ -122,41 +122,69 @@ 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 (items after it) was cut.
+ * The preserved prefix of an ordered event-row list (mirrors the JSONL
+ * backend's `scanLog`): the longest prefix of complete, seq-contiguous,
+ * parseable rows, PLUS the seq from which a never-committed torn tail must be
+ * deleted (or `undefined` if the whole list is intact).
  *
- * 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`.
+ * A crash can leave a durable log whose final turn never closed: real,
+ * fully-written rows sit after the last `turn/end`. Those are PRESERVED — a
+ * single turn can be huge in a long-horizon task, so truncating it would
+ * destroy real work; the backend closes the orphaned open turn with a synthetic
+ * `turn/end {kind:'interrupted'}` on load (ADR 0018). The ONLY thing excluded is
+ * a torn trailing fragment — a row whose `data` never parses, or a seq gap —
+ * AFTER the last committed `turn/end`; that bounds the preserved region and its
+ * seq is returned as `tornFrom` so `load` can physically delete it.
  *
- * 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).
- * The committed region MUST be contiguous (`item.seq === i`); a gap there means
- * committed data was lost and the session is unloadable.
+ * The last `turn/end` is computed from the `type` COLUMN (never parsing tail
+ * `data`), so a malformed `data` in an uncommitted tail row is discarded rather
+ * than making the session unloadable. A parse error or seq gap AT OR BEFORE the
+ * last committed `turn/end` is committed-data corruption and throws.
+ *
+ * This relies on the session-log invariant that every event lives inside a turn
+ * (`Session.append` enforces it): only the final turn can be open, so the
+ * preserved tail is at most one unclosed turn.
  */
-export function cutAtLastTurnEnd<T extends { seq: number; type: string }>(
-  items: readonly T[],
-): { committed: T[]; cutTail: boolean } {
-  let lastTurnEnd = -1
-  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: items.length > 0 }
-  }
-  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})`)
+export function scanRows(rows: readonly EventRow[]): { preserved: SessionEvent[]; tornFrom?: number } {
+  // Pass 1: parse each row's data; a row whose data is not valid JSON is a hole.
+  // (The seq/type COLUMNS are always present even when `data` is corrupt.)
+  interface Parsed { ok: boolean; event?: SessionEvent }
+  const parsed: Parsed[] = rows.map((row) => {
+    try {
+      return { ok: true, event: rowToEvent(row) }
+    } catch {
+      return { ok: false }
     }
   })
-  return { committed, cutTail: lastTurnEnd < items.length - 1 }
+
+  // The last index that is a valid `turn/end` — the last fully-committed
+  // boundary (the loop flushes only at turn/end).
+  let lastTurnEnd = -1
+  for (let i = parsed.length - 1; i >= 0; i--) {
+    if (parsed[i]?.ok && rows[i]?.type === 'turn/end') { lastTurnEnd = i; break }
+  }
+
+  // Walk the longest PREFIX of complete, seq-contiguous, parseable rows
+  // (row i has seq === i). This includes the fully-written rows of an
+  // interrupted final turn AFTER the last turn/end — real work, never
+  // truncated. The walk stops at the first hole:
+  //   - at or before the last committed turn/end → committed corruption (throw);
+  //   - after it (or no committed turn/end) → tolerated torn tail (stop).
+  const preserved: SessionEvent[] = []
+  for (let i = 0; i < rows.length; i++) {
+    const p = parsed[i]
+    if (!p?.ok || p.event === undefined) {
+      if (i <= lastTurnEnd) throw new Error(`corrupt session log: unparsable committed event at seq ${rows[i]?.seq}`)
+      break // torn tail fragment after the last turn/end — stop, tolerate
+    }
+    if (p.event.seq !== i) {
+      if (i <= lastTurnEnd) throw new Error(`corrupt session log: seq gap in committed region (expected ${i}, got ${p.event.seq})`)
+      break // gap after the last turn/end — torn tail, stop
+    }
+    preserved.push(p.event)
+  }
+
+  // Any rows past the preserved prefix are a never-committed torn tail; their
+  // first seq is the deletion point for load's physical repair.
+  return preserved.length < rows.length ? { preserved, tornFrom: preserved.length } : { preserved }
 }

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

@@ -6,7 +6,7 @@ import { join } from 'node:path'
 import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
 import type { Session, SessionEvent, SessionMeta } from '@deepseek-ai/dsh-session'
 import SessionPersistenceSqlite, { SCHEMA_VERSION } from '@deepseek-ai/dsh-session-persistence-sqlite'
-import { cutAtLastTurnEnd, openDatabase } from '../src/schema.ts'
+import { openDatabase, scanRows, type EventRow } from '../src/schema.ts'
 import { runPersistenceContract, meta, oneTurnLog } from '../../session-persistence/tests/contract.ts'
 
 const dirs: string[] = []
@@ -38,49 +38,78 @@ runPersistenceContract('sqlite', async () => {
   }
 })
 
-describe('cutAtLastTurnEnd', () => {
-  it('returns the prefix through the last complete turn/end and flags a cut tail', () => {
-    const log = oneTurnLog()
-    const withTail: SessionEvent[] = [
-      ...log,
+describe('scanRows', () => {
+  // scanRows works off EventRows (data is a JSON string column); build them from
+  // SessionEvents so the unit tests read in terms of the event vocabulary.
+  const rows = (events: SessionEvent[]): EventRow[] =>
+    events.map(e => ({ seq: e.seq, type: e.type, time: e.time, data: JSON.stringify(e.data) }))
+
+  it('preserves the full log when it ends exactly on a turn/end (no torn tail)', () => {
+    const { preserved, tornFrom } = scanRows(rows(oneTurnLog()))
+    expect(preserved).toEqual(oneTurnLog())
+    expect(tornFrom).toBeUndefined()
+  })
+
+  it('PRESERVES the real events of an interrupted turn after the last turn/end', () => {
+    // turn 1 committed (0..5) + a crashed turn 2 (turn/start 6, step/start 7, no
+    // close): all 8 rows are intact, so the whole prefix is preserved and there
+    // is no torn fragment to delete. (load() then synthesizes the closers.)
+    const withOpenTurn: SessionEvent[] = [
+      ...oneTurnLog(),
       { type: 'turn/start', seq: 6, time: 7, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } },
-      { type: 'user/message', seq: 7, time: 8, data: { content: [{ type: 'text', text: 'q2' }], source: { kind: 'user' } } },
+      { type: 'step/start', seq: 7, time: 8, data: { turn: 2, step: 1 } },
     ]
-    const { committed, cutTail } = cutAtLastTurnEnd(withTail)
-    expect(committed).toEqual(log)
-    expect(cutTail).toBe(true)
+    const { preserved, tornFrom } = scanRows(rows(withOpenTurn))
+    expect(preserved.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7])
+    expect(tornFrom).toBeUndefined()
   })
 
-  it('treats a log with no turn/end as fully uncommitted', () => {
-    const partial: SessionEvent[] = [
+  it('preserves the contiguous prefix and flags a torn tail at a seq gap', () => {
+    // A gap after seq 0 (no committed turn/end): seq 0 is the preserved
+    // interrupted-turn event; the gap bounds it and marks the torn fragment.
+    const gapped: 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: 'hi' }], source: { kind: 'user' } } },
+      { type: 'step/start', seq: 2, time: 2, data: { turn: 1, step: 1 } }, // seq 1 missing
     ]
-    expect(cutAtLastTurnEnd(partial)).toEqual({ committed: [], cutTail: true })
+    const { preserved, tornFrom } = scanRows(rows(gapped))
+    expect(preserved.map(e => e.seq)).toEqual([0])
+    expect(tornFrom).toBe(1)
   })
 
-  it('reports no cut when the log ends exactly on a turn/end', () => {
-    const { committed, cutTail } = cutAtLastTurnEnd(oneTurnLog())
-    expect(committed).toEqual(oneTurnLog())
-    expect(cutTail).toBe(false)
+  it('an empty log preserves nothing and has no torn tail', () => {
+    expect(scanRows([])).toEqual({ preserved: [] })
   })
 
-  it('an empty log is committed-empty with no tail', () => {
-    expect(cutAtLastTurnEnd([])).toEqual({ committed: [], cutTail: false })
-  })
-
-  it('throws on a seq gap inside the committed region', () => {
+  it('throws on a seq gap inside the committed region (before the last turn/end)', () => {
     const gapped: SessionEvent[] = [
       { type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } },
       { type: 'step/start', seq: 2, time: 2, data: { turn: 1, step: 1 } }, // seq 1 missing
       { type: 'turn/end', seq: 3, time: 3, data: { turn: 1, reason: { kind: 'completed' } } },
     ]
-    expect(() => cutAtLastTurnEnd(gapped)).toThrow(/seq gap in committed region/)
+    expect(() => scanRows(rows(gapped))).toThrow(/seq gap in committed region/)
+  })
+
+  it('throws on an unparsable row inside the committed region', () => {
+    const withCorruptCommitted: EventRow[] = [
+      { seq: 0, type: 'turn/start', time: 1, data: '{not json' }, // corrupt, sits before a turn/end
+      { seq: 1, type: 'turn/end', time: 2, data: JSON.stringify({ turn: 1, reason: { kind: 'completed' } }) },
+    ]
+    expect(() => scanRows(withCorruptCommitted)).toThrow(/unparsable committed event/)
+  })
+
+  it('tolerates an unparsable torn-tail row after the last turn/end', () => {
+    const withCorruptTail: EventRow[] = [
+      ...rows(oneTurnLog()),
+      { seq: 6, type: 'turn/start', time: 7, data: '{not json' }, // torn fragment, no committed turn/end after
+    ]
+    const { preserved, tornFrom } = scanRows(withCorruptTail)
+    expect(preserved).toEqual(oneTurnLog())
+    expect(tornFrom).toBe(6)
   })
 })
 
 describe('SessionPersistenceSqlite: durability and crash semantics', () => {
-  it('a crash tail (rows after the last turn/end) is excluded on load and repaired on the next append', async () => {
+  it('an interrupted turn (rows after the last turn/end) is PRESERVED and closed during load', async () => {
     const path = await freshDbPath()
     const m = meta('crash')
     // Run 1: persist a complete turn, then a half-written second turn (no turn/end).
@@ -91,37 +120,44 @@ describe('SessionPersistenceSqlite: durability and crash semantics', () => {
     await ctx1.sessionPersistence.append(m.id, oneTurnLog())
     await ctx1.sessionPersistence.append(m.id, [
       { type: 'turn/start', seq: 6, time: 7, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } },
-      { type: 'user/message', seq: 7, time: 8, data: { content: [{ type: 'text', text: 'q2' }], source: { kind: 'user' } } },
+      { type: 'step/start', seq: 7, time: 8, data: { turn: 2, step: 1 } },
     ])
     await fiber1.dispose()
 
-    // Run 2: load returns only the committed first turn (tail excluded).
+    // Run 2: load PRESERVES the interrupted turn's real events (a turn can be huge
+    // — never truncated) and closes the orphaned turn with synthetic boundary
+    // events: step/end (the step was open) then turn/end {interrupted}.
     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())
+    expect(loaded.events.map(e => e.type)).toEqual([
+      'turn/start', 'user/message', 'step/start', 'assistant/message', 'step/end', 'turn/end', // turn 1
+      'turn/start', 'step/start', 'step/end', 'turn/end', // turn 2: real events + synthetic closers
+    ])
+    expect(loaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7, 8, 9])
+    const last = loaded.events.at(-1)!
+    expect(last.type === 'turn/end' && last.data.reason).toEqual({ kind: 'interrupted' })
 
-    // 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.
+    // load durably closed the turn, so the next append continues at the balanced
+    // length (seq 10) and a reload round-trips identically.
     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' } } },
+      { type: 'turn/start', seq: 10, time: 9, data: { turn: 3, trigger: { kind: 'message', source: { kind: 'user' } } } },
+      { type: 'turn/end', seq: 11, time: 10, data: { turn: 3, reason: { kind: 'completed' } } },
     ])
     const reloaded = await ctx2.sessionPersistence.load(m.id)
-    expect(reloaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7])
+    expect(reloaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11])
     await fiber2.dispose()
   })
 
-  it('load() is non-mutating: the crash tail rows survive until the next append repairs them', async () => {
+  it('load() durably closes the interrupted turn: the synthetic closers are on disk after load', async () => {
     const path = await freshDbPath()
-    const m = meta('load-nonmutating')
+    const m = meta('load-closes')
     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).
+    // Hand-write an interrupted turn (turn/start 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' } } }))
@@ -129,17 +165,20 @@ describe('SessionPersistenceSqlite: durability and crash semantics', () => {
 
     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.
+    // turn 2's real turn/start (seq 6) is preserved + a synthetic turn/end (seq 7).
+    expect(loaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7])
+    expect(loaded.events.at(-1)!.type).toBe('turn/end')
+    // load() is mutating: the synthetic turn/end MUST be on disk so the stored log
+    // is balanced and the cursor is truthful (contract: load closes, not defers).
     const probe = openDatabase(path)
-    const tailRows = probe.prepare('SELECT seq FROM events WHERE session_id = ? AND seq >= 6').all(m.id)
+    const stored = probe.prepare('SELECT seq, type FROM events WHERE session_id = ? ORDER BY seq').all(m.id) as { seq: number; type: string }[]
     probe.close()
-    expect(tailRows).toHaveLength(1)
+    expect(stored.map(r => r.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7])
+    expect(stored.at(-1)!.type).toBe('turn/end')
     await b2.dispose()
   })
 
-  it('all-tail load: a session materialized by a partial first turn stays present (JSONL parity), load returns zero committed events', async () => {
+  it('all-tail load: a session whose only turn never closed is preserved and closed on load', async () => {
     const path = await freshDbPath()
     const m = meta('all-tail')
     const b1 = await backend(path)
@@ -152,14 +191,13 @@ describe('SessionPersistenceSqlite: durability and crash semantics', () => {
     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
-    // load returns zero events — but the session WAS materialized (its metadata
-    // row exists), so has()/list() still report it present, matching the JSONL
-    // backend whose file likewise survives a first append that never reached
-    // turn/end. The orphaned tail rows are removed by the next append's repair.
+    // A fresh backend loads it: the interrupted (only) turn's real events are
+    // preserved and closed with a synthetic turn/end {interrupted} — NOT
+    // truncated. The session was materialized, so has()/list() report it present.
     const b2 = await backend(path)
     const loaded = await b2.ctx.sessionPersistence.load(m.id)
-    expect(loaded.events).toEqual([])
+    expect(loaded.events.map(e => e.type)).toEqual(['turn/start', 'user/message', 'turn/end'])
+    expect(loaded.events.at(-1)!.type === 'turn/end' && loaded.events.at(-1)!.data).toMatchObject({ reason: { kind: 'interrupted' } })
     expect(await b2.ctx.sessionPersistence.has(m.id)).toBe(true)
     expect((await b2.ctx.sessionPersistence.list()).map(x => x.id)).toContain(m.id)
     await b2.dispose()
@@ -208,10 +246,11 @@ describe('SessionPersistenceSqlite: durability and crash semantics', () => {
     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`).
+    // Hand-insert a torn 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 torn tail must be discarded. scanRows finds the last
+    // turn/end on the seq+type columns (never parsing tail `data`), so the
+    // unparsable row after it bounds the preserved prefix and is deleted by load.
     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')
@@ -219,8 +258,8 @@ describe('SessionPersistenceSqlite: durability and crash semantics', () => {
 
     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.
+    expect(loaded.events).toEqual(oneTurnLog()) // torn tail discarded, committed intact (turn 1 already balanced → no closers)
+    // load physically deleted the corrupt tail row, 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' } } },