diff --git a/packages/session-persistence-sqlite/README.md b/packages/session-persistence-sqlite/README.md index dd72ddfde4..3a7a3c8163 100644 --- a/packages/session-persistence-sqlite/README.md +++ b/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) diff --git a/packages/session-persistence-sqlite/src/index.ts b/packages/session-persistence-sqlite/src/index.ts index 45d0215384..bd4b4d9435 100644 --- a/packages/session-persistence-sqlite/src/index.ts +++ b/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 { @@ -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. */ diff --git a/packages/session-persistence-sqlite/src/schema.ts b/packages/session-persistence-sqlite/src/schema.ts index c3ef9ca051..1dad51698b 100644 --- a/packages/session-persistence-sqlite/src/schema.ts +++ b/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( - 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 } } diff --git a/packages/session-persistence-sqlite/tests/sqlite.spec.ts b/packages/session-persistence-sqlite/tests/sqlite.spec.ts index 6ab4dae30b..44d788ab72 100644 --- a/packages/session-persistence-sqlite/tests/sqlite.spec.ts +++ b/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' } } },