From 9126697d87624c9db711073253bfe3e982e0e18f Mon Sep 17 00:00:00 2001 From: Tianyi Cui <53024+tianyicui@users.noreply.github.com> Date: Mon, 15 Jun 2026 21:45:21 +0800 Subject: [PATCH 1/5] feat(session-persistence-sqlite): second backend validating the abstraction MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add a SQLite SessionPersistence backend (node:sqlite), a SECOND implementation built to prove the abstract seam + the shared runPersistenceContract suite are genuinely backend-agnostic. Each SessionEvent maps 1:1 onto an events row (session_id, seq, type, time, data); append is an INSERT inside a transaction asserting the contiguous-seq contract; the mutable SessionSummary lives in the sessions metadata row. It satisfies the SAME contract semantics as the JSONL backend, expressed over rows instead of file bytes: - Lazy materialization: create() records intent in memory; no row until the first append (a never-appended session is absent from has()/list() via a materialized flag set inside the first append transaction). - Crash-tail-on-load: load() returns events only through the last complete turn/end and deletes the uncommitted tail; a seq gap in the committed region makes the session unloadable. - Transactional append: a mid-batch failure (a UNIQUE seq collision from a concurrent writer) rolls back entirely, keeping the cursor truthful. Like the JSONL backend it is also the write-path plugin (session/event → buffer → session/flush drain, onCreated seed/adopt/collision handling, HMR seeding, dispose-to-quiescence). The package runs the shared runPersistenceContract suite plus SQLite-specific tests (transaction rollback, crash-tail cut, schema version, HMR adoption). Docs flip every "SQLite is future/deferred" reference (ADR 0016, architecture.md, the persistence module doc + README) to "implemented; the contract holds both backends to identical semantics". --- docs/adr/0016-session-persistence.md | 2 +- docs/architecture.md | 3 +- packages/session-persistence-sqlite/README.md | 27 + .../session-persistence-sqlite/package.json | 35 ++ .../session-persistence-sqlite/src/index.ts | 482 ++++++++++++++ .../session-persistence-sqlite/src/schema.ts | 135 ++++ .../tests/sqlite.spec.ts | 595 ++++++++++++++++++ .../session-persistence-sqlite/tsconfig.json | 15 + packages/session-persistence/README.md | 2 +- packages/session-persistence/src/index.ts | 9 +- scripts/publint-all.ts | 1 + tsconfig.base.json | 1 + tsconfig.build.json | 1 + tsconfig.typecheck.json | 1 + yarn.lock | 15 + 15 files changed, 1317 insertions(+), 7 deletions(-) create mode 100644 packages/session-persistence-sqlite/README.md create mode 100644 packages/session-persistence-sqlite/package.json create mode 100644 packages/session-persistence-sqlite/src/index.ts create mode 100644 packages/session-persistence-sqlite/src/schema.ts create mode 100644 packages/session-persistence-sqlite/tests/sqlite.spec.ts create mode 100644 packages/session-persistence-sqlite/tsconfig.json diff --git a/docs/adr/0016-session-persistence.md b/docs/adr/0016-session-persistence.md index 4ab7ad6e37..d4ac30df40 100644 --- a/docs/adr/0016-session-persistence.md +++ b/docs/adr/0016-session-persistence.md @@ -19,7 +19,7 @@ Key choices recorded here because they are durable, contested, and surprising: - **The canonical durable log persists every `SessionEvent` verbatim, including `assistant/chunk`.** `deriveMessages()` skips chunks, and a chunk-filtered rollout (Codex's `policy.rs`) is tempting — but `seq = log.length` and the load-validation `events[i].seq === i` require a *contiguous* log; filtering chunks out would leave holes and break both the contract and resume. A chunk-filtered projection is possible later as a derived view with its own renumbering, but it is NOT the canonical log. - **Append-only with a single exception.** Committed events — those at or below a flushed `turn/end` — are never rewritten. The loop only flushes at `turn/end`, so a crash can leave a half-written final turn below the last checkpoint; `load` returns events only up to the **last complete `turn/end`**, and the first post-load `append` runs a one-time **truncation-repair** (`ftruncate` + `fsync`) that physically discards only that never-committed crash tail before writing. -- **File backend canonical, DB backend a drop-in.** `SessionEvent` maps 1:1 onto a row `(session_id, seq, type, time, data)` — `append` is INSERT (in a transaction asserting the contiguous-seq contract), `load` is SELECT … ORDER BY seq. A future `dsh-session-persistence-sqlite` is a `SessionPersistence` subclass with no interface change (opencode runs this exact shape on SQLite/WAL). +- **File backend canonical, DB backend a proven drop-in.** `SessionEvent` maps 1:1 onto a row `(session_id, seq, type, time, data)` — `append` is INSERT (in a transaction asserting the contiguous-seq contract), `load` is SELECT … ORDER BY seq. `dsh-session-persistence-sqlite` is exactly this: a `SessionPersistence` subclass with no interface change (opencode runs this exact shape on SQLite/WAL), and it passes the same `runPersistenceContract` suite as the JSONL backend — so the contract holds both backends to identical semantics (lazy materialization, crash-tail-on-load, contiguous-seq), expressed once over file bytes and once over rows. - **Metadata is out-of-log.** Format version, cwd, and lineage are storage concerns, not replayable conversation state, so they live in a `SessionMeta` (`SessionHeader & SessionSummary`) owned by `dsh-session` and attached to a `Session` via a new readonly `session.header` — never in `SessionEventMap`, never reaching `deriveMessages()`. The alternative (a merge-extensible `session/meta` event as log line 0) was rejected: an in-log event would ride along with a seeded/forked session for free, but metadata is not replayable state, so the explicit out-of-log header seam is the cleaner cost. - **Resume is an async factory, not a change to synchronous create.** `ctx.agents.resume({ resumeSessionId })` awaits `ctx.sessionPersistence.load`, recreates the live session with the loaded events (so `lastTurnNumber`/`deriveMessages` continue), and starts a fresh agent on the resumed id (NOT `${agentId}-session`). The agent-loop does NOT hard-inject `sessionPersistence` (that would pend non-persistent demos forever); `resume` rejects with a typed error when it is absent. diff --git a/docs/architecture.md b/docs/architecture.md index 85398a9f7c..829283b729 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -95,7 +95,7 @@ A `Session` is an append-only log of typed `SessionEvent`s — the single source Replay/fork = `ctx.sessions.create(id, { seed: seedEvents })`. Trace/telemetry = listen to `session/event`. -**Durability seam**: `session/event` is a synchronous notification; persistence plugins buffer (write-behind) and drain at the awaited `session/flush` checkpoint the loop fires at every turn end. The durable backend is a real **capability seam**: the abstract `SessionPersistence` service (`dsh-session-persistence`, `ctx.sessionPersistence`) defines create/append/load/list/update over the existing `SessionEvent` (no parallel persisted type), and `dsh-session-persistence-jsonl` is the first implementation — an append-only JSONL log per session with crash-safe atomic writes, truncation-repair of a never-committed crash tail, and a read/replay path. Session metadata (format version, cwd, lineage) travels separately as `SessionMeta`, attached to a `Session` via `session.header`. Resuming a persisted session into a live agent is `ctx.agents.resume({ resumeSessionId })`. A SQLite/WAL backend is a future drop-in `SessionPersistence` subclass (the row shape `(session_id, seq, type, time, data)` maps 1:1 onto `SessionEvent`). +**Durability seam**: `session/event` is a synchronous notification; persistence plugins buffer (write-behind) and drain at the awaited `session/flush` checkpoint the loop fires at every turn end. The durable backend is a real **capability seam**: the abstract `SessionPersistence` service (`dsh-session-persistence`, `ctx.sessionPersistence`) defines create/append/load/list/update over the existing `SessionEvent` (no parallel persisted type), and `dsh-session-persistence-jsonl` is the first implementation — an append-only JSONL log per session with crash-safe atomic writes, truncation-repair of a never-committed crash tail, and a read/replay path. Session metadata (format version, cwd, lineage) travels separately as `SessionMeta`, attached to a `Session` via `session.header`. Resuming a persisted session into a live agent is `ctx.agents.resume({ resumeSessionId })`. A second backend, `dsh-session-persistence-sqlite` (`node:sqlite`, one row per `SessionEvent` — the row shape `(session_id, seq, type, time, data)` maps 1:1 onto it), passes the same `runPersistenceContract` suite, proving the seam is genuinely backend-agnostic. ## Prompt assembly (dsh-system-prompt) @@ -241,7 +241,6 @@ Code skeletons for the three plugin shapes (tool, hook/permission-gate, UI) and Tracked here deliberately — each is designed-for but not implemented: - **Sub-agent spawn/fork semantics** (seam: `AgentLoop.create()`); inter-agent channels beyond `send`/`steer`/events. -- **SQLite/WAL persistence backend** — a drop-in `SessionPersistence` subclass (the abstract seam + the JSONL backend landed; see the durability-seam paragraph). - **Compaction implementation** (auto thresholds, summarization prompts) on the `agent/request` seam, with its session-event types added by declaration merging. - **Parallel tool execution** (concurrency-safety hints on ToolDefinition). - **Session branching/tree** (pi-style entry tree) if needed beyond seed-based forking. diff --git a/packages/session-persistence-sqlite/README.md b/packages/session-persistence-sqlite/README.md new file mode 100644 index 0000000000..5af385e63e --- /dev/null +++ b/packages/session-persistence-sqlite/README.md @@ -0,0 +1,27 @@ +# @deepseek-ai/dsh-session-persistence-sqlite + +A SQLite durable session-persistence backend — a second `SessionPersistence` implementation ([ADR 0016](../../docs/adr/0016-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. + +## Storage model + +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`. + +## 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. +- **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. + +## Configuration (schemastery) + +```ts +interface Config { + path: string // SQLite database file path, or ':memory:' for an in-process DB +} +``` + +## Write path + +Like the JSONL backend, the plugin also installs the `session/event` → buffer → `session/flush` drain: it snapshots each event when buffered (the live `session.events` object is mutable), persists a fork's seed once on `session/created`, keeps a per-session write cursor so a resumed session never re-appends stored events, and seeds existing live sessions on apply (HMR does not replay `session/created`). Dispose awaits every in-flight init + final drain and then closes the database, so no write lands after teardown. diff --git a/packages/session-persistence-sqlite/package.json b/packages/session-persistence-sqlite/package.json new file mode 100644 index 0000000000..0cbefc4f00 --- /dev/null +++ b/packages/session-persistence-sqlite/package.json @@ -0,0 +1,35 @@ +{ + "name": "@deepseek-ai/dsh-session-persistence-sqlite", + "description": "SQLite durable session persistence backend for the DeepSeek Harness", + "version": "0.0.1", + "private": true, + "type": "module", + "main": "lib/index.js", + "types": "lib/index.d.ts", + "exports": { + ".": { + "types": "./lib/index.d.ts", + "default": "./lib/index.js" + }, + "./src/*": "./src/*", + "./package.json": "./package.json" + }, + "files": [ + "lib", + "src" + ], + "license": "BSD-3-Clause", + "peerDependencies": { + "@deepseek-ai/dsh-session": "^0.0.1", + "@deepseek-ai/dsh-session-persistence": "^0.0.1", + "cordis": "^4.0.0-rc.6" + }, + "dependencies": { + "schemastery": "^3.18.0" + }, + "devDependencies": { + "@deepseek-ai/dsh-session": "^0.0.1", + "@deepseek-ai/dsh-session-persistence": "^0.0.1", + "cordis": "^4.0.0-rc.6" + } +} diff --git a/packages/session-persistence-sqlite/src/index.ts b/packages/session-persistence-sqlite/src/index.ts new file mode 100644 index 0000000000..f279e4684e --- /dev/null +++ b/packages/session-persistence-sqlite/src/index.ts @@ -0,0 +1,482 @@ +/** + * SQLite durable session-persistence backend (`@deepseek-ai/dsh-session-persistence-sqlite`). + * + * 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 + * `SessionSummary` lives in the `sessions` metadata row. + * + * Like the JSONL backend it is also the write-path plugin: it installs the + * `session/event` → buffer → `session/flush` drain, persists a fork's seed once + * on `session/created`, keeps a per-session write cursor so a resumed session + * never re-appends stored events, and seeds existing live sessions on apply + * (HMR does not replay `session/created`). + * + * @module @deepseek-ai/dsh-session-persistence-sqlite + */ + +import { Context } from 'cordis' +import z from 'schemastery' +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 type { Session, SessionEvent, SessionId, SessionMeta, SessionSummary } from '@deepseek-ai/dsh-session' +import { + cutAtLastTurnEnd, openDatabase, rowToEvent, rowToMeta, type EventRow, type SessionRow, +} from './schema.ts' + +export { SCHEMA_VERSION } from './schema.ts' + +/** Plugin configuration. */ +export interface Config { + /** + * Filesystem path to the SQLite database file. The special value `:memory:` + * opens an in-process database (tests); a file path is created (with parent + * dirs) on construction. + */ + path: string +} + +/** Backend bookkeeping for a session id (NOT the live Session object). */ +interface SessionState { + meta: SessionMeta + /** Next seq to write — equals the number of committed events. */ + cursor: number + /** Whether the session has at least one persisted event (materialized). */ + materialized: boolean + /** The live Session that owns this state (collision detection); see onCreated. */ + owner?: Session +} + +/** + * Whether a live session's `seed` reproduces a persisted `prefix` exactly (the + * prefix is no longer than the seed and each event DEEP-equals the seed event + * at the same index). Distinguishes a session legitimately continuing a + * persisted log (HMR re-seeing its own session, or a resume) from a different + * session that merely reuses the id. Mirrors the JSONL backend's check; both + * sides are JSON-serializable by contract, so `JSON.stringify` is a sound + * canonical form. + */ +function seedCoversPrefix(seed: readonly SessionEvent[], prefix: readonly SessionEvent[]): boolean { + return prefix.length <= seed.length + && prefix.every((e, i) => { + const s = seed[i] + return s !== undefined && JSON.stringify(s) === JSON.stringify(e) + }) +} + +/** Reject non-JSON-serializable `event.data`, naming the offending type. */ +function assertSerializable(events: readonly SessionEvent[]): void { + for (const event of events) { + if (!isJsonValue(event.data)) { + throw new Error(`event "${event.type}" carries non-JSON-serializable data (seq ${event.seq})`) + } + } +} + +/** + * The SQLite persistence backend. Load as a plugin; it registers as + * `ctx.sessionPersistence` and installs the write-path listeners. + */ +export class SessionPersistenceSqlite extends SessionPersistence { + static inject = ['sessions'] + + static Config: z = z.object({ + path: z.string().required(), + }) + + private db!: DatabaseSync + private ready: Promise + /** Backend bookkeeping keyed by session id (NOT the live Session object). */ + private states = new Map() + /** Write-behind buffers keyed by the live Session (write path). */ + private buffers = new Map() + /** Per-session serialization chain (keyed by session id). */ + private chains = new Map>() + /** Per-session init promise (onCreated), keyed by the LIVE Session object. */ + private inits = new Map>() + + constructor(ctx: Context, public config: Config) { + super(ctx) + // Open the database asynchronously (the parent directory may need creating); + // every backend op awaits `ready` first. Opening synchronously in the ctor + // would force a sync mkdir and block plugin apply. + this.ready = this.openDb(config.path) + this.installWritePath() + } + + private async openDb(path: string): Promise { + if (path !== ':memory:') { + const abs = resolve(path) + await mkdir(dirname(abs), { recursive: true, mode: 0o700 }) + this.db = openDatabase(abs) + } else { + this.db = openDatabase(path) + } + } + + // --- SessionPersistence backend surface (all serialized per session id) --- + + create(meta: SessionMeta): Promise { + const snapshot: SessionMeta = { ...meta } + return this.serialize(snapshot.id, () => this.createCore(snapshot)) + } + + private async createCore(meta: SessionMeta): Promise { + await this.ready + if (this.states.has(meta.id)) { + throw new Error(`session "${meta.id}" already exists in this backend`) + } + if (this.rowFor(meta.id) !== undefined) { + throw new Error(`session "${meta.id}" already has a persisted row; load/resume it instead of creating`) + } + // Lazy: record intent in memory only. No row until the first append, so an + // abandoned (never-appended) session leaves nothing behind and stays absent + // from has()/list(). + this.states.set(meta.id, { meta, cursor: 0, materialized: false }) + } + + append(id: SessionId, events: readonly SessionEvent[]): Promise { + return this.serialize(id, () => this.appendCore(id, events)) + } + + private async appendCore(id: SessionId, events: readonly SessionEvent[]): Promise { + 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) + + // Contiguity contract: each event's seq must continue the stored log. + for (const [i, event] of events.entries()) { + if (event.seq !== state.cursor + i) { + throw new Error(`append seq mismatch for "${id}": expected ${state.cursor + i} at index ${i}, got ${event.seq}`) + } + } + + // 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. + const insertEvent = this.db.prepare( + 'INSERT INTO events (session_id, seq, type, time, data) VALUES (?, ?, ?, ?, ?)', + ) + this.db.exec('BEGIN') + try { + 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)) + } + // Bump updatedAt on every append (the mutable summary lives in the row). + const updatedAt = Date.now() + this.db.prepare('UPDATE sessions SET updated_at = ? WHERE id = ?').run(updatedAt, id) + this.db.exec('COMMIT') + state.meta = { ...state.meta, updatedAt } + } catch (error) { + this.db.exec('ROLLBACK') + throw error + } + state.materialized = true + state.cursor += events.length + } + + load(id: SessionId): Promise<{ meta: SessionMeta; events: SessionEvent[] }> { + return this.serialize(id, () => this.loadCore(id)) + } + + private async loadCore(id: SessionId): Promise<{ meta: SessionMeta; events: SessionEvent[] }> { + await this.ready + const row = this.rowFor(id) + if (row === undefined) throw new Error(`session "${id}" not found`) + 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). + 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) + + // 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) + } + + // 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 }) + return { meta, events: committed } + } + + async list(): Promise { + await this.ready + // Materialized rows only: a created-but-never-appended (lazy) session has no + // row at all, and a load that cut every event back to zero leaves + // materialized = 0. Both are excluded, matching has(). + const rows = this.db + .prepare('SELECT * FROM sessions WHERE materialized = 1') + .all() as unknown as SessionRow[] + return rows.map(rowToMeta) + } + + async has(id: SessionId): Promise { + await this.ready + const state = this.states.get(id) + if (state?.materialized) return true + const row = this.rowFor(id) + return row !== undefined && row.materialized === 1 + } + + delete(id: SessionId): Promise { + return this.serialize(id, () => this.deleteCore(id)) + } + + private async deleteCore(id: SessionId): Promise { + await this.ready + // ON DELETE CASCADE drops the session's events with its row. + this.db.prepare('DELETE FROM sessions WHERE id = ?').run(id) + this.states.delete(id) + } + + update(id: SessionId, summary: Partial): Promise { + return this.serialize(id, () => this.updateCore(id, summary)) + } + + private async updateCore(id: SessionId, summary: Partial): Promise { + await this.ready + let state = this.states.get(id) + if (state === undefined) state = await this.adopt(id) + const nextMeta: SessionMeta = { ...state.meta, ...summary } + // update's only durable effect is the summary fields; the event log is + // untouched. If the row is not materialized yet (a lazy session updated + // before its first append) there is nothing to write — keep the pending + // summary in memory so the materializing append carries it. + if (state.materialized) this.writeRow(nextMeta) + state.meta = nextMeta + } + + // --- row helpers --- + + /** Fetch a session's row, or undefined if absent. */ + private rowFor(id: SessionId): SessionRow | undefined { + const row = this.db.prepare('SELECT * FROM sessions WHERE id = ?').get(id) as unknown as SessionRow | undefined + return row + } + + /** + * Insert-or-replace a session's metadata row, marked materialized. The only + * callers are the first materializing `append` and a post-materialization + * `update` — a row is written only once a session has durable events, so + * `materialized` is always 1 (a never-appended session has no row at all). + */ + private writeRow(meta: SessionMeta): void { + this.db.prepare(` + INSERT INTO sessions (id, version, created_at, cwd, parent_session, updated_at, title, first_prompt, materialized) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, 1) + ON CONFLICT(id) DO UPDATE SET + version = excluded.version, + created_at = excluded.created_at, + cwd = excluded.cwd, + parent_session = excluded.parent_session, + updated_at = excluded.updated_at, + title = excluded.title, + first_prompt = excluded.first_prompt, + materialized = excluded.materialized + `).run( + meta.id, + meta.version, + meta.createdAt, + meta.cwd ?? null, + meta.parentSession ?? null, + meta.updatedAt, + meta.title ?? null, + meta.firstPrompt ?? null, + ) + } + + /** Build a state for a session present in the DB but not yet in memory. */ + private async adopt(id: SessionId): Promise { + await this.loadCore(id) // sets the state; load (serialized) would deadlock + const state = this.states.get(id) + /* v8 ignore next -- loadCore always sets the state for the id */ + if (!state) throw new Error(`failed to adopt session "${id}"`) + return state + } + + private assertVersion(meta: SessionMeta): void { + if (meta.version !== 1) { + throw new Error(`unsupported session format version ${meta.version} for "${meta.id}" (only v1 is supported)`) + } + } + + /** + * Run `op` after any in-flight operation for the same session id, so writes + * for one session never interleave. Errors do not poison the chain. NOTE: + * serialized public methods must NOT call each other (deadlock); they call + * the unserialized `*Core` helpers instead. + */ + private serialize(id: SessionId, op: () => Promise): Promise { + const prior = this.chains.get(id) ?? Promise.resolve() + const next = prior.then(op, op) + this.chains.set(id, next.then(() => undefined, () => undefined)) + return next + } + + // --- write path (session/event → flush drain) --- + + private installWritePath(): void { + const ctx = this.ctx + + ctx.on('session/created', (session) => { void this.initFor(session) }) + + // Snapshot + buffer every event (the live object is mutable; clone so a + // later in-place mutation cannot rewrite a buffered event). Serializability + // is guaranteed at the source (Session.append), so structuredClone is safe. + ctx.on('session/event', (session, event) => { + let buffer = this.buffers.get(session) + if (!buffer) this.buffers.set(session, buffer = []) + buffer.push(structuredClone(event)) + }) + + ctx.on('session/flush', session => this.flush(session)) + + // Dispose must reach quiescence: await every init + final drain, then close + // the database, BEFORE returning, so no write lands after teardown. + ctx.effect(() => async () => { + await Promise.allSettled([...this.inits.values()]) + await Promise.allSettled([...this.buffers.keys()].map(s => this.flush(s))) + await Promise.allSettled([...this.chains.values()]) + await this.ready + this.db.close() + }, 'session-persistence-sqlite write path') + + // HMR: a hot reload does not replay session/created, so seed existing live + // sessions (mirrors dsh-invariants and the JSONL backend). + for (const session of ctx.sessions.list()) void this.initFor(session) + } + + /** Start (once) the async init for a session and remember its promise. */ + private initFor(session: Session): Promise { + const existing = this.inits.get(session) + if (existing) return existing + const seed = session.events.map(e => structuredClone(e)) + const p = this.onCreated(session, seed) + p.catch(() => { /* observed by flush/dispose via the stored promise */ }) + this.inits.set(session, p) + return p + } + + /** + * On session/created: sync the backend's state to a live Session. Cases + * mirror the JSONL backend: + * 1. Already tracked → no-op (or claim ownerless state if the seed matches). + * 2. A row EXISTS and is a seq-aligned PREFIX of the live events → adopt + * (HMR/resume), persisting any live suffix beyond the stored prefix. + * 3. A row EXISTS but is NOT a prefix → reject (id collision). + * 4. No row → a genuinely new session: register meta (lazy) + persist seed. + */ + private async onCreated(session: Session, seed: readonly SessionEvent[]): Promise { + await this.ready + const id = session.header.id + const tracked = this.states.get(id) + if (tracked !== undefined) { + /* v8 ignore next -- initFor dedupes per session object; same-object re-entry can't occur */ + if (tracked.owner === session) return + if (tracked.owner === undefined) { + // Ownerless state from a public create()/load(). The first live session + // claims it ONLY if its seed reproduces the persisted prefix. + if (!await this.seedMatchesPersisted(id, seed, tracked.cursor)) { + throw new Error(`session "${id}" is already persisted with ${tracked.cursor} event(s) that do not match this live session (id collision)`) + } + tracked.owner = session + const suffix = seed.slice(tracked.cursor) + if (suffix.length > 0) await this.append(id, suffix) + return + } + // Owned by a DIFFERENT live session. Reclaim ONLY a truly-abandoned id + // (never materialized, no pending buffer); else it is a real collision. + const ownerBuffer = this.buffers.get(tracked.owner) + if (!tracked.materialized && !ownerBuffer?.length) { + this.states.delete(id) + } else { + throw new Error(`session "${id}" is already bound to a different live session in this backend (id collision)`) + } + } + + const row = this.rowFor(id) + if (row !== undefined && row.materialized === 1) { + const stored = this.eventsFor(id) + if (!seedCoversPrefix(seed, stored)) { + throw new Error(`session "${id}" already has a persisted log that does not match this live session (id collision)`) + } + await this.serialize(id, () => this.loadCore(id)) + const adopted = this.states.get(id) + /* v8 ignore next -- loadCore always sets the state for the id */ + if (adopted !== undefined) adopted.owner = session + const suffix = seed.slice(stored.length) + if (suffix.length > 0) await this.append(id, suffix) + return + } + + // case 4: a genuinely new session. + const meta: SessionMeta = { ...session.header, updatedAt: Date.now() } + await this.create(meta) + const created = this.states.get(id) + /* v8 ignore next -- create() always sets the state for the id */ + if (created !== undefined) created.owner = session + if (seed.length > 0) await this.append(id, seed) + } + + /** The committed events for a session id (last-turn/end cut applied). */ + 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[] + return cutAtLastTurnEnd(rows.map(rowToEvent)).committed + } + + /** Whether a live session's seed reproduces the first `cursor` stored events. */ + private async seedMatchesPersisted(id: SessionId, seed: readonly SessionEvent[], cursor: number): Promise { + await this.ready + if (cursor === 0) return true + return seedCoversPrefix(seed, this.eventsFor(id).slice(0, cursor)) + } + + private async flush(session: Session): Promise { + await this.inits.get(session) + await this.serialize(session.header.id, () => this.drain(session)) + } + + /** Drain a session's write buffer to the database. Caller serializes per id. */ + private async drain(session: Session): Promise { + const buffer = this.buffers.get(session) + if (!buffer?.length) return + const batch = buffer.slice() + const state = this.states.get(session.header.id) + /* v8 ignore next -- state is always set by the awaited init before flush */ + const cursor = state?.cursor ?? 0 + const fresh = batch.filter(e => e.seq >= cursor) + if (fresh.length > 0) await this.appendCore(session.header.id, fresh) + buffer.splice(0, batch.length) + } +} + +export default SessionPersistenceSqlite diff --git a/packages/session-persistence-sqlite/src/schema.ts b/packages/session-persistence-sqlite/src/schema.ts new file mode 100644 index 0000000000..5f20b5c08b --- /dev/null +++ b/packages/session-persistence-sqlite/src/schema.ts @@ -0,0 +1,135 @@ +/** + * Schema + load-time helpers for the SQLite session-persistence backend: the + * DDL (a `sessions` metadata table and a 1:1 `events` row per `SessionEvent`), + * the database open/configure step, and the last-`turn/end` cut that gives the + * SQLite backend the SAME crash-tail-on-load semantics as the JSONL backend. + * + * @module dsh-session-persistence-sqlite/schema + */ + +import { DatabaseSync } from 'node:sqlite' +import type { SessionEvent, SessionId, SessionMeta } from '@deepseek-ai/dsh-session' + +/** + * The on-disk schema version. Bumped only on a breaking change to the table + * layout; orthogonal to a session's own `version` (which versions the EVENT + * vocabulary, stored per session in the `sessions` row). + */ +export const SCHEMA_VERSION = 1 + +/** + * A row of the `sessions` table — the out-of-log metadata (`SessionMeta`) plus + * the `materialized` flag that implements lazy materialization (a created-but- + * never-appended session has `materialized = 0` and is excluded from + * `has`/`list`, mirroring the JSONL backend's "no file until first append"). + */ +export interface SessionRow { + id: string + version: number + created_at: number + cwd: string | null + parent_session: string | null + updated_at: number + title: string | null + first_prompt: string | null + materialized: number +} + +/** An `events` table row: one `SessionEvent` mapped 1:1 (`data` is JSON text). */ +export interface EventRow { + seq: number + type: string + time: number + data: string +} + +/** + * Open the database at `path` and apply the schema + pragmas. `foreign_keys` + * 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). + */ +export function openDatabase(path: string): DatabaseSync { + const db = new DatabaseSync(path) + db.exec('PRAGMA foreign_keys = ON') + db.exec('PRAGMA journal_mode = WAL') + db.exec(` + CREATE TABLE IF NOT EXISTS sessions ( + id TEXT PRIMARY KEY, + version INTEGER NOT NULL, + created_at INTEGER NOT NULL, + cwd TEXT, + parent_session TEXT, + updated_at INTEGER NOT NULL, + title TEXT, + first_prompt TEXT, + materialized INTEGER NOT NULL DEFAULT 0 + ) STRICT + `) + db.exec(` + CREATE TABLE IF NOT EXISTS events ( + session_id TEXT NOT NULL REFERENCES sessions(id) ON DELETE CASCADE, + seq INTEGER NOT NULL, + type TEXT NOT NULL, + time INTEGER NOT NULL, + data TEXT NOT NULL, + PRIMARY KEY (session_id, seq) + ) STRICT + `) + return db +} + +/** Reconstruct the full {@link SessionMeta} from a `sessions` row. */ +export function rowToMeta(row: SessionRow): SessionMeta { + return { + version: row.version, + id: row.id as SessionId, + createdAt: row.created_at, + updatedAt: row.updated_at, + ...row.cwd !== null ? { cwd: row.cwd } : {}, + ...row.parent_session !== null ? { parentSession: row.parent_session as SessionId } : {}, + ...row.title !== null ? { title: row.title } : {}, + ...row.first_prompt !== null ? { firstPrompt: row.first_prompt } : {}, + } +} + +/** Reconstruct a {@link SessionEvent} from an `events` row (parses `data`). */ +export function rowToEvent(row: EventRow): SessionEvent { + return { + type: row.type, + seq: row.seq, + time: row.time, + data: JSON.parse(row.data) as SessionEvent['data'], + } as 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 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. + */ +export function cutAtLastTurnEnd(events: readonly SessionEvent[]): { committed: SessionEvent[]; cutTail: boolean } { + let lastTurnEnd = -1 + events.forEach((event, i) => { + if (event.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 } + } + 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})`) + } + }) + return { committed, cutTail: lastTurnEnd < events.length - 1 } +} diff --git a/packages/session-persistence-sqlite/tests/sqlite.spec.ts b/packages/session-persistence-sqlite/tests/sqlite.spec.ts new file mode 100644 index 0000000000..854013e03a --- /dev/null +++ b/packages/session-persistence-sqlite/tests/sqlite.spec.ts @@ -0,0 +1,595 @@ +import { afterEach, describe, expect, it } from 'vitest' +import { Context } from 'cordis' +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +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 { runPersistenceContract, meta, oneTurnLog } from '../../session-persistence/tests/contract.ts' + +const dirs: string[] = [] +afterEach(async () => { for (const d of dirs.splice(0)) await rm(d, { recursive: true, force: true }) }) + +async function freshDbPath(): Promise { + const dir = await mkdtemp(join(tmpdir(), 'dsh-sqlite-')) + dirs.push(dir) + return join(dir, 'sessions.db') +} + +// The payoff: the SAME backend-agnostic contract the JSONL backend runs, now +// proving the SQLite backend satisfies identical semantics. +runPersistenceContract('sqlite', async () => { + const ctx = new Context() + await ctx.plugin(SessionStore) + const fiber = await ctx.plugin(SessionPersistenceSqlite, { path: ':memory:' }) + return { + persistence: ctx.sessionPersistence, + dispose: async () => { await fiber.dispose() }, + } +}) + +describe('cutAtLastTurnEnd', () => { + it('returns the prefix through the last complete turn/end and flags a cut tail', () => { + const log = oneTurnLog() + const withTail: SessionEvent[] = [ + ...log, + { 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' } } }, + ] + const { committed, cutTail } = cutAtLastTurnEnd(withTail) + expect(committed).toEqual(log) + expect(cutTail).toBe(true) + }) + + it('treats a log with no turn/end as fully uncommitted', () => { + const partial: 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' } } }, + ] + expect(cutAtLastTurnEnd(partial)).toEqual({ committed: [], cutTail: true }) + }) + + 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 is committed-empty with no tail', () => { + expect(cutAtLastTurnEnd([])).toEqual({ committed: [], cutTail: false }) + }) + + it('throws on a seq gap inside the committed region', () => { + 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/) + }) +}) + +describe('SessionPersistenceSqlite: durability and crash semantics', () => { + it('a crash tail (rows after the last turn/end) is excluded and deleted on 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). + const ctx1 = new Context() + await ctx1.plugin(SessionStore) + const fiber1 = await ctx1.plugin(SessionPersistenceSqlite, { path }) + await ctx1.sessionPersistence.create(m) + 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' } } }, + ]) + await fiber1.dispose() + + // Run 2: load returns only the committed first turn; the tail is gone. + 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. + 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' } } }, + ]) + const reloaded = await ctx2.sessionPersistence.load(m.id) + expect(reloaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7]) + await fiber2.dispose() + }) + + it('append rolls back the whole batch on a mid-batch seq collision (transaction)', async () => { + const ctx = new Context() + await ctx.plugin(SessionStore) + const fiber = await ctx.plugin(SessionPersistenceSqlite, { path: ':memory:' }) + const m = meta('rollback') + await ctx.sessionPersistence.create(m) + await ctx.sessionPersistence.append(m.id, oneTurnLog()) // seqs 0..5 + + // A batch that re-states an already-stored seq must be rejected and leave + // the stored log unchanged (the UNIQUE (session_id, seq) constraint fires + // inside the transaction → ROLLBACK). + await expect(ctx.sessionPersistence.append(m.id, oneTurnLog())).rejects.toThrow() + const loaded = await ctx.sessionPersistence.load(m.id) + expect(loaded.events).toEqual(oneTurnLog()) // unchanged + await fiber.dispose() + }) + + it('persists across separate backend instances over the same file', async () => { + const path = await freshDbPath() + const m = meta('persist', '/proj') + const ctx1 = new Context() + await ctx1.plugin(SessionStore) + const fiber1 = await ctx1.plugin(SessionPersistenceSqlite, { path }) + await ctx1.sessionPersistence.create(m) + await ctx1.sessionPersistence.append(m.id, oneTurnLog()) + await ctx1.sessionPersistence.update(m.id, { title: 'T', firstPrompt: 'hi' }) + await fiber1.dispose() + + const ctx2 = new Context() + await ctx2.plugin(SessionStore) + const fiber2 = await ctx2.plugin(SessionPersistenceSqlite, { path }) + expect((await ctx2.sessionPersistence.list()).map(x => x.id)).toContain(m.id) + const loaded = await ctx2.sessionPersistence.load(m.id) + expect(loaded.meta).toMatchObject({ id: m.id, cwd: '/proj', title: 'T', firstPrompt: 'hi' }) + expect(loaded.events).toEqual(oneTurnLog()) + await fiber2.dispose() + }) + + it('rejects an unknown format version on load', async () => { + const path = await freshDbPath() + // Materialize a row with version 2 directly via the real schema. + const db = openDatabase(path) + db.prepare('INSERT INTO sessions (id, version, created_at, updated_at, materialized) VALUES (?, ?, ?, ?, 1)') + .run('v2', 2, 1, 1) + db.close() + + const ctx = new Context() + await ctx.plugin(SessionStore) + const fiber = await ctx.plugin(SessionPersistenceSqlite, { path }) + await expect(ctx.sessionPersistence.load(SessionId('v2'))).rejects.toThrow(/version 2/) + await fiber.dispose() + }) + + it('create rejects a duplicate id (in memory and on a persisted row)', async () => { + const path = await freshDbPath() + const m = meta('dup') + const ctx = new Context() + await ctx.plugin(SessionStore) + const fiber = await ctx.plugin(SessionPersistenceSqlite, { path }) + await ctx.sessionPersistence.create(m) + // Same in-memory state. + await expect(ctx.sessionPersistence.create(m)).rejects.toThrow(/already exists/) + await ctx.sessionPersistence.append(m.id, oneTurnLog()) + await fiber.dispose() + + // A fresh instance over the same file sees the persisted row. + const ctx2 = new Context() + await ctx2.plugin(SessionStore) + const fiber2 = await ctx2.plugin(SessionPersistenceSqlite, { path }) + await expect(ctx2.sessionPersistence.create(m)).rejects.toThrow(/already has a persisted row/) + await fiber2.dispose() + }) + + it('exposes the schema version constant', () => { + expect(SCHEMA_VERSION).toBe(1) + }) +}) + +describe('SessionPersistenceSqlite: write path (session/event → flush)', () => { + function send(session: Session, events: SessionEvent[]): void { + for (const e of events) session.append(e.type, e.data) + } + + it('persists a turn appended through the live session on flush', async () => { + const ctx = new Context() + await ctx.plugin(SessionStore) + const fiber = await ctx.plugin(SessionPersistenceSqlite, { path: ':memory:' }) + const session = ctx.sessions.create('w1') + send(session, oneTurnLog()) + await ctx.parallel('session/flush', session) + const loaded = await ctx.sessionPersistence.load(SessionId('w1')) + expect(loaded.events.map(e => e.type)).toEqual(oneTurnLog().map(e => e.type)) + await fiber.dispose() + }) + + it('a resumed session does not re-append its seed', async () => { + const path = await freshDbPath() + // Run 1: persist a full turn through the live session. + const ctx1 = new Context() + await ctx1.plugin(SessionStore) + const fiber1 = await ctx1.plugin(SessionPersistenceSqlite, { path }) + const s1 = ctx1.sessions.create('resume') + for (const e of oneTurnLog()) s1.append(e.type, e.data) + await ctx1.parallel('session/flush', s1) + await fiber1.dispose() + + // Run 2: reconstruct the live session from the loaded log (seed), then add a + // second turn. The seed must NOT be re-appended (no UNIQUE collision), and + // the second turn continues the seq. + const ctx2 = new Context() + await ctx2.plugin(SessionStore) + const fiber2 = await ctx2.plugin(SessionPersistenceSqlite, { path }) + const { events } = await ctx2.sessionPersistence.load(SessionId('resume')) + const s2 = ctx2.sessions.create('resume', { seed: events }) + s2.append('turn/start', { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } }) + s2.append('turn/end', { turn: 2, reason: { kind: 'completed' } }) + await ctx2.parallel('session/flush', s2) + const reloaded = await ctx2.sessionPersistence.load(SessionId('resume')) + expect(reloaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7]) + await fiber2.dispose() + }) + + it('HMR: applying the plugin seeds existing live sessions', async () => { + const ctx = new Context() + await ctx.plugin(SessionStore) + const session = ctx.sessions.create('hmr') + for (const e of oneTurnLog()) session.append(e.type, e.data) + // Plugin applied AFTER the session already has events. + const fiber = await ctx.plugin(SessionPersistenceSqlite, { path: ':memory:' }) + await ctx.parallel('session/flush', session) + expect(await ctx.sessionPersistence.has(SessionId('hmr'))).toBe(true) + await fiber.dispose() + }) + + it('dispose drains a pending buffer before closing the database', async () => { + const path = await freshDbPath() + const ctx = new Context() + await ctx.plugin(SessionStore) + const fiber = await ctx.plugin(SessionPersistenceSqlite, { path }) + const session = ctx.sessions.create('drain') + for (const e of oneTurnLog()) session.append(e.type, e.data) + // No explicit flush — dispose must drain the buffer. + await fiber.dispose() + + const ctx2 = new Context() + await ctx2.plugin(SessionStore) + const fiber2 = await ctx2.plugin(SessionPersistenceSqlite, { path }) + expect(await ctx2.sessionPersistence.has(SessionId('drain'))).toBe(true) + await fiber2.dispose() + }) + + it('rejects a different live session colliding on a persisted id', async () => { + const path = await freshDbPath() + const ctx1 = new Context() + await ctx1.plugin(SessionStore) + const fiber1 = await ctx1.plugin(SessionPersistenceSqlite, { path }) + const s1 = ctx1.sessions.create('collide') + for (const e of oneTurnLog()) s1.append(e.type, e.data) + await ctx1.parallel('session/flush', s1) + await fiber1.dispose() + + // A fresh, unrelated session reusing the id (no seed) must be rejected. + const ctx2 = new Context() + await ctx2.plugin(SessionStore) + const fiber2 = await ctx2.plugin(SessionPersistenceSqlite, { path }) + const s2 = ctx2.sessions.create('collide') + s2.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) + await expect(ctx2.parallel('session/flush', s2)).rejects.toThrow(/id collision/) + await fiber2.dispose() + }) + + it('update before the first append keeps the summary in memory and the session lazy', async () => { + const ctx = new Context() + await ctx.plugin(SessionStore) + const fiber = await ctx.plugin(SessionPersistenceSqlite, { path: ':memory:' }) + const m = meta('lazy-update') + await ctx.sessionPersistence.create(m) + await ctx.sessionPersistence.update(m.id, { title: 'pending' }) + // Still lazy: no materialized row yet. + expect(await ctx.sessionPersistence.has(m.id)).toBe(false) + // The first append materializes and carries the pending title. + await ctx.sessionPersistence.append(m.id, oneTurnLog()) + const loaded = await ctx.sessionPersistence.load(m.id) + expect(loaded.meta.title).toBe('pending') + await fiber.dispose() + }) +}) + +describe('SessionPersistenceSqlite: edge cases', () => { + async function backend(path = ':memory:'): Promise<{ ctx: Context; dispose: () => Promise }> { + 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') + await ctx.sessionPersistence.create(m) + await ctx.sessionPersistence.append(m.id, []) + expect(await ctx.sessionPersistence.has(m.id)).toBe(false) // still lazy + await dispose() + }) + + it('load rejects a missing session', async () => { + const { ctx, dispose } = await backend() + await expect(ctx.sessionPersistence.load(SessionId('nope'))).rejects.toThrow(/not found/) + await dispose() + }) + + it('delete of a non-existent session is a no-op', async () => { + const { ctx, dispose } = await backend() + await ctx.sessionPersistence.delete(SessionId('ghost')) + expect(await ctx.sessionPersistence.has(SessionId('ghost'))).toBe(false) + await dispose() + }) + + it('append adopts a session that exists only in the DB (fresh instance)', async () => { + const path = await freshDbPath() + const m = meta('adopt-append') + const b1 = await backend(path) + await b1.ctx.sessionPersistence.create(m) + await b1.ctx.sessionPersistence.append(m.id, oneTurnLog()) + await b1.dispose() + + // A fresh instance appends a second turn WITHOUT a prior create/load: append + // must adopt the on-disk row (cursor = stored length) and continue the seq. + const b2 = await backend(path) + await b2.ctx.sessionPersistence.append(m.id, [ + { type: 'turn/start', seq: 6, time: 7, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } }, + { type: 'turn/end', seq: 7, time: 8, data: { turn: 2, reason: { kind: 'completed' } } }, + ]) + const loaded = await b2.ctx.sessionPersistence.load(m.id) + expect(loaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7]) + await b2.dispose() + }) + + it('update adopts a session that exists only in the DB (fresh instance)', async () => { + const path = await freshDbPath() + const m = meta('adopt-update') + const b1 = await backend(path) + await b1.ctx.sessionPersistence.create(m) + await b1.ctx.sessionPersistence.append(m.id, oneTurnLog()) + await b1.dispose() + + const b2 = await backend(path) + await b2.ctx.sessionPersistence.update(m.id, { title: 'after restart' }) + const loaded = await b2.ctx.sessionPersistence.load(m.id) + expect(loaded.meta.title).toBe('after restart') + await b2.dispose() + }) + + it('append rolls back and rethrows when an event INSERT fails inside the transaction', async () => { + const path = await freshDbPath() + const m = meta('rollback-insert') + const b1 = await backend(path) + await b1.ctx.sessionPersistence.create(m) + await b1.ctx.sessionPersistence.append(m.id, oneTurnLog()) + + // A SECOND backend over the same file loads the session first, so it adopts + // cursor 6 (the committed length) into its OWN in-memory state. + const b2 = await backend(path) + await b2.ctx.sessionPersistence.load(m.id) // cursor 6 in b2 + const turn2: SessionEvent[] = [ + { type: 'turn/start', seq: 6, time: 7, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } }, + { type: 'turn/end', seq: 7, time: 8, data: { turn: 2, reason: { kind: 'completed' } } }, + ] + // b1 commits seq 6..7 first. + await b1.ctx.sessionPersistence.append(m.id, turn2) + // b2 still thinks its cursor is 6, so this batch passes the contiguity check + // but its INSERT of seq 6 hits the UNIQUE (session_id, seq) constraint + // mid-transaction → ROLLBACK + rethrow. + await expect(b2.ctx.sessionPersistence.append(m.id, turn2)).rejects.toThrow(/UNIQUE/) + // b1's turn is intact; b2's rolled-back attempt left nothing extra. + const loaded = await b1.ctx.sessionPersistence.load(m.id) + expect(loaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7]) + await b1.dispose() + await b2.dispose() + }) + + it('round-trips a header with parentSession (fork lineage)', async () => { + const { ctx, dispose } = await backend() + const m: SessionMeta = { ...meta('child'), parentSession: SessionId('parent') } + await ctx.sessionPersistence.create(m) + await ctx.sessionPersistence.append(m.id, oneTurnLog()) + const loaded = await ctx.sessionPersistence.load(m.id) + expect(loaded.meta.parentSession).toBe(SessionId('parent')) + await dispose() + }) + + it('a fresh live session reusing a previously-loaded id is rejected (ownerless guard)', async () => { + const path = await freshDbPath() + const m = meta('ownerless') + const b1 = await backend(path) + await b1.ctx.sessionPersistence.create(m) + await b1.ctx.sessionPersistence.append(m.id, oneTurnLog()) + await b1.dispose() + + const b2 = await backend(path) + // load() leaves ownerless state with cursor 6. + await b2.ctx.sessionPersistence.load(m.id) + // A fresh, unrelated live session reusing the id has a shorter/non-matching + // seed → its onCreated must reject rather than graft onto the loaded prefix. + const s = b2.ctx.sessions.create('ownerless') + s.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) + await expect(b2.ctx.parallel('session/flush', s)).rejects.toThrow(/id collision/) + await b2.dispose() + }) + + it('a live session whose seed matches the loaded prefix claims ownerless state and persists the suffix', async () => { + const path = await freshDbPath() + const m = meta('claim') + const b1 = await backend(path) + await b1.ctx.sessionPersistence.create(m) + await b1.ctx.sessionPersistence.append(m.id, oneTurnLog()) + await b1.dispose() + + const b2 = await backend(path) + const { events } = await b2.ctx.sessionPersistence.load(m.id) // ownerless, cursor 6 + // A live session seeded with the loaded log PLUS a new turn claims the state + // and persists only the suffix. + const s = b2.ctx.sessions.create('claim', { seed: [ + ...events, + { type: 'turn/start', seq: 6, time: 7, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } }, + { type: 'turn/end', seq: 7, time: 8, data: { turn: 2, reason: { kind: 'completed' } } }, + ] }) + await b2.ctx.parallel('session/flush', s) + const loaded = await b2.ctx.sessionPersistence.load(m.id) + expect(loaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7]) + await b2.dispose() + }) + + it('an abandoned lazy session (never materialized) releases its id for reuse', async () => { + const { ctx, dispose } = await backend() + const inits = (ctx.sessionPersistence as unknown as { inits: Map> }).inits + let first!: Session + const firstFiber = await ctx.plugin(Object.assign((inner: Context) => { + first = inner.sessions.create('reuse') + }, { inject: ['sessions'] })) + await inits.get(first) // let the lazy create register the state + await firstFiber.dispose() // disposed before any append → never materialized + + let reuse!: Session + await ctx.plugin(Object.assign((inner: Context) => { + reuse = inner.sessions.create('reuse') + }, { inject: ['sessions'] })) + await expect(inits.get(reuse)).resolves.toBeUndefined() + reuse.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) + reuse.append('turn/end', { turn: 1, reason: { kind: 'completed' } }) + await ctx.parallel('session/flush', reuse) + expect(await ctx.sessionPersistence.has(SessionId('reuse'))).toBe(true) + await dispose() + }) + + it('does NOT reclaim an id whose abandoned owner still has buffered (unflushed) events', async () => { + const { ctx, dispose } = await backend() + const inits = (ctx.sessionPersistence as unknown as { inits: Map> }).inits + let first!: Session + const firstFiber = await ctx.plugin(Object.assign((inner: Context) => { + first = inner.sessions.create('buffered') + }, { inject: ['sessions'] })) + await inits.get(first) + // Append a turn but do NOT flush — events sit in the write-behind buffer. + first.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) + first.append('turn/end', { turn: 1, reason: { kind: 'completed' } }) + await firstFiber.dispose() // disposed before flush; not materialized, buffer pending + + let reuse!: Session + await ctx.plugin(Object.assign((inner: Context) => { + reuse = inner.sessions.create('buffered') + }, { inject: ['sessions'] })) + await expect(inits.get(reuse)).rejects.toThrow(/already bound to a different live session/) + await dispose() + }) + + it('initFor is idempotent: re-emitting session/created does not re-initialize', async () => { + const { ctx, dispose } = await backend() + const session = ctx.sessions.create('idem') + ctx.emit('session/created', session) // second create event for the same object + session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) + session.append('turn/end', { turn: 1, reason: { kind: 'completed' } }) + await ctx.parallel('session/flush', session) + expect(await ctx.sessionPersistence.has(SessionId('idem'))).toBe(true) + await dispose() + }) + + it('a live session claims cursor-0 ownerless state created via the public API and persists its seed', async () => { + const { ctx, dispose } = await backend() + // create() registers ownerless state with cursor 0 (no events yet). + await ctx.sessionPersistence.create(meta('cursor0')) + // A live session reusing that id, seeded with a turn, claims the ownerless + // state (cursor 0 trivially matches any seed) and persists the whole seed. + const s = ctx.sessions.create('cursor0', { seed: [ + { type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } }, + { type: 'turn/end', seq: 1, time: 2, data: { turn: 1, reason: { kind: 'completed' } } }, + ] }) + await ctx.parallel('session/flush', s) + const loaded = await ctx.sessionPersistence.load(SessionId('cursor0')) + expect(loaded.events.map(e => e.seq)).toEqual([0, 1]) + await dispose() + }) + + it('HMR: reloading the backend adopts a still-live, already-materialized session', async () => { + const path = await freshDbPath() + const ctx = new Context() + await ctx.plugin(SessionStore) + // The session lives in its OWN fiber so it survives the backend reload. + let session!: Session + await ctx.plugin(Object.assign((inner: Context) => { + session = inner.sessions.create('hmr-adopt') + }, { inject: ['sessions'] })) + + // Backend instance 1 materializes the session on disk. + const backend1 = await ctx.plugin(SessionPersistenceSqlite, { path }) + session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) + session.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }) + session.append('turn/end', { turn: 1, reason: { kind: 'completed' } }) + await ctx.parallel('session/flush', session) + + // Hot-reload: dispose instance 1, plug in instance 2 over the SAME file + // while the session stays live. Instance 2 has an empty states map but the + // row is materialized on disk and is a prefix of the live events — it must + // ADOPT (not reject), and a second turn then persists. + await backend1.dispose() + await ctx.plugin(SessionPersistenceSqlite, { path }) + session.append('turn/start', { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } }) + session.append('turn/end', { turn: 2, reason: { kind: 'completed' } }) + await expect(ctx.parallel('session/flush', session)).resolves.not.toThrow() + + const loaded = await ctx.sessionPersistence.load(SessionId('hmr-adopt')) + expect(loaded.events.filter(e => e.type === 'turn/start')).toHaveLength(2) + await ctx.fiber.dispose() + }) + + it('HMR: adoption persists the live SUFFIX that was ahead of the on-disk prefix', async () => { + const path = await freshDbPath() + const ctx = new Context() + await ctx.plugin(SessionStore) + let session!: Session + await ctx.plugin(Object.assign((inner: Context) => { + session = inner.sessions.create('hmr-suffix') + }, { inject: ['sessions'] })) + + const backend1 = await ctx.plugin(SessionPersistenceSqlite, { path }) + session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) + session.append('turn/end', { turn: 1, reason: { kind: 'completed' } }) + await ctx.parallel('session/flush', session) + + // Append turn 2 to the LIVE session, then dispose instance 1 WITHOUT + // flushing turn 2: it is now ONLY in the live session's events. + await backend1.dispose() + session.append('turn/start', { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } }) + session.append('turn/end', { turn: 2, reason: { kind: 'completed' } }) + + // Instance 2 adopts the on-disk prefix (turn 1) and MUST persist the live + // suffix (turn 2) carried in the session's events. + await ctx.plugin(SessionPersistenceSqlite, { path }) + await ctx.parallel('session/flush', session) + const loaded = await ctx.sessionPersistence.load(SessionId('hmr-suffix')) + expect(loaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3]) + expect(loaded.events.filter(e => e.type === 'turn/start')).toHaveLength(2) + await ctx.fiber.dispose() + }) + + it('HMR: a DIFFERENT session colliding with a materialized on-disk id is rejected', async () => { + const path = await freshDbPath() + // Instance 1 materializes a session and disposes. + const b1 = await backend(path) + const s1 = b1.ctx.sessions.create('hmr-collide') + for (const e of oneTurnLog()) s1.append(e.type, e.data) + await b1.ctx.parallel('session/flush', s1) + await b1.dispose() + + // A fresh context with an UNRELATED live session reusing the id meets a + // materialized row that is NOT a prefix of its events → reject. + const ctx = new Context() + await ctx.plugin(SessionStore) + let session!: Session + await ctx.plugin(Object.assign((inner: Context) => { + session = inner.sessions.create('hmr-collide') + }, { inject: ['sessions'] })) + session.append('turn/start', { turn: 9, trigger: { kind: 'message', source: { kind: 'user' } } }) + await ctx.plugin(SessionPersistenceSqlite, { path }) + await expect(ctx.parallel('session/flush', session)).rejects.toThrow(/id collision/) + await ctx.fiber.dispose() + }) +}) diff --git a/packages/session-persistence-sqlite/tsconfig.json b/packages/session-persistence-sqlite/tsconfig.json new file mode 100644 index 0000000000..3595f989bd --- /dev/null +++ b/packages/session-persistence-sqlite/tsconfig.json @@ -0,0 +1,15 @@ +{ + "extends": "../../tsconfig.base.json", + "compilerOptions": { + "rootDir": "src", + "outDir": "lib" + }, + "include": ["src"], + "references": [ + { "path": "../../vendor/cosmokit" }, + { "path": "../../vendor/cordis" }, + { "path": "../../vendor/schemastery" }, + { "path": "../session" }, + { "path": "../session-persistence" } + ] +} diff --git a/packages/session-persistence/README.md b/packages/session-persistence/README.md index aa30d7c028..77ee9a4e2d 100644 --- a/packages/session-persistence/README.md +++ b/packages/session-persistence/README.md @@ -26,7 +26,7 @@ The persisted unit IS the existing `SessionEvent` (event-sourced model — the l Import `runPersistenceContract` from `tests/contract.ts` and call it with a factory that yields a fresh, empty backend plus a teardown. Every backend is held to the same append-only / contiguous-seq / lazy-materialization / serializability semantics; a backend's own spec adds implementation-specific tests (crash repair, path sanitization) on top. -> **TODO (validate the abstraction with a second backend):** `dsh-session-persistence-jsonl` is currently the only implementation, so the interface and `runPersistenceContract` are only proven against one storage model. A second backend — a SQLite implementation (`dsh-session-persistence-sqlite`), where each `SessionEvent` maps 1:1 onto a row `(session_id, seq, type, time, data)` — would run the SAME `runPersistenceContract` suite and so prove the seam is genuinely backend-agnostic (lazy materialization, crash-tail-on-load, contiguous-seq all expressed against a transactional store rather than an append-only file). +Two backends run this suite: `dsh-session-persistence-jsonl` (append-only file log) and `dsh-session-persistence-sqlite` (`node:sqlite`, each `SessionEvent` one row `(session_id, seq, type, time, data)`). Both passing the same contract is the proof that the seam is genuinely backend-agnostic — lazy materialization, crash-tail-on-load, and contiguous-seq hold identically over file bytes and over a transactional store. ## Metadata types diff --git a/packages/session-persistence/src/index.ts b/packages/session-persistence/src/index.ts index 8a7d1da992..f08ac2e1f0 100644 --- a/packages/session-persistence/src/index.ts +++ b/packages/session-persistence/src/index.ts @@ -4,9 +4,12 @@ * list, and update sessions — without saying HOW. Implementations subclass * {@link SessionPersistence} and register themselves as the * `sessionPersistence` service; `@deepseek-ai/dsh-session-persistence-jsonl` - * (an append-only JSONL log per session) is the first. Future backends swap in - * SQLite/WAL, an object store, or a remote service without touching the - * consumers (the write-path plugin, the agent-loop resume seam). + * (an append-only JSONL log per session) is the first and + * `@deepseek-ai/dsh-session-persistence-sqlite` (`node:sqlite`, one row per + * event) is a second that validates the seam is backend-agnostic by passing + * the same `runPersistenceContract` suite. Further backends swap in an object + * store or a remote service without touching the consumers (the write-path + * plugin, the agent-loop resume seam). * * The persisted unit IS the existing {@link SessionEvent} — there is no * parallel "persisted message" type the log must be converted to and from diff --git a/scripts/publint-all.ts b/scripts/publint-all.ts index 8cd274647a..5b84790263 100644 --- a/scripts/publint-all.ts +++ b/scripts/publint-all.ts @@ -8,6 +8,7 @@ const packages = [ 'packages/session', 'packages/session-persistence', 'packages/session-persistence-jsonl', + 'packages/session-persistence-sqlite', 'packages/system-prompt', 'packages/tools', 'packages/agent', diff --git a/tsconfig.base.json b/tsconfig.base.json index d9010a3c6b..9d2e8bd18d 100644 --- a/tsconfig.base.json +++ b/tsconfig.base.json @@ -38,6 +38,7 @@ "@deepseek-ai/dsh-session": ["./packages/session/src"], "@deepseek-ai/dsh-session-persistence": ["./packages/session-persistence/src"], "@deepseek-ai/dsh-session-persistence-jsonl": ["./packages/session-persistence-jsonl/src"], + "@deepseek-ai/dsh-session-persistence-sqlite": ["./packages/session-persistence-sqlite/src"], "@deepseek-ai/dsh-system-prompt": ["./packages/system-prompt/src"], "@deepseek-ai/dsh-tools": ["./packages/tools/src"], "@deepseek-ai/dsh-agent": ["./packages/agent/src"], diff --git a/tsconfig.build.json b/tsconfig.build.json index 92e10bca73..874351ad84 100644 --- a/tsconfig.build.json +++ b/tsconfig.build.json @@ -14,6 +14,7 @@ { "path": "./packages/session" }, { "path": "./packages/session-persistence" }, { "path": "./packages/session-persistence-jsonl" }, + { "path": "./packages/session-persistence-sqlite" }, { "path": "./packages/system-prompt" }, { "path": "./packages/agent" }, { "path": "./packages/tools" }, diff --git a/tsconfig.typecheck.json b/tsconfig.typecheck.json index da1d1ba7b6..6051a68410 100644 --- a/tsconfig.typecheck.json +++ b/tsconfig.typecheck.json @@ -20,6 +20,7 @@ "@deepseek-ai/dsh-session": ["./packages/session/src"], "@deepseek-ai/dsh-session-persistence": ["./packages/session-persistence/src"], "@deepseek-ai/dsh-session-persistence-jsonl": ["./packages/session-persistence-jsonl/src"], + "@deepseek-ai/dsh-session-persistence-sqlite": ["./packages/session-persistence-sqlite/src"], "@deepseek-ai/dsh-system-prompt": ["./packages/system-prompt/src"], "@deepseek-ai/dsh-tools": ["./packages/tools/src"], "@deepseek-ai/dsh-agent": ["./packages/agent/src"], diff --git a/yarn.lock b/yarn.lock index 8b2339874e..328a61174c 100644 --- a/yarn.lock +++ b/yarn.lock @@ -706,6 +706,21 @@ __metadata: languageName: unknown linkType: soft +"@deepseek-ai/dsh-session-persistence-sqlite@workspace:packages/session-persistence-sqlite": + version: 0.0.0-use.local + resolution: "@deepseek-ai/dsh-session-persistence-sqlite@workspace:packages/session-persistence-sqlite" + dependencies: + "@deepseek-ai/dsh-session": "npm:^0.0.1" + "@deepseek-ai/dsh-session-persistence": "npm:^0.0.1" + cordis: "npm:^4.0.0-rc.6" + schemastery: "npm:^3.18.0" + peerDependencies: + "@deepseek-ai/dsh-session": ^0.0.1 + "@deepseek-ai/dsh-session-persistence": ^0.0.1 + cordis: ^4.0.0-rc.6 + languageName: unknown + linkType: soft + "@deepseek-ai/dsh-session-persistence@npm:^0.0.1, @deepseek-ai/dsh-session-persistence@workspace:packages/session-persistence": version: 0.0.0-use.local resolution: "@deepseek-ai/dsh-session-persistence@workspace:packages/session-persistence" From ccbc4f533fe06a03c54485342be93940ff40064a Mon Sep 17 00:00:00 2001 From: Tianyi Cui <53024+tianyicui@users.noreply.github.com> Date: Mon, 15 Jun 2026 22:23:36 +0800 Subject: [PATCH 2/5] fix(session-persistence-sqlite): JSONL parity on append snapshot + corrupt-tail load MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- .../session-persistence-sqlite/src/index.ts | 42 +++++++---- .../session-persistence-sqlite/src/schema.ts | 37 ++++++---- .../tests/sqlite.spec.ts | 70 +++++++++++++++++-- 3 files changed, 114 insertions(+), 35 deletions(-) diff --git a/packages/session-persistence-sqlite/src/index.ts b/packages/session-persistence-sqlite/src/index.ts index f279e4684e..8a2a326f02 100644 --- a/packages/session-persistence-sqlite/src/index.ts +++ b/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 { - return this.serialize(id, () => this.appendCore(id, events)) + // `async` so the synchronous validate/clone below reject (not throw) per the + // Promise contract — callers use `await expect(...).rejects`. + async append(id: SessionId, events: readonly SessionEvent[]): Promise { + // 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 { 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 { @@ -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. */ diff --git a/packages/session-persistence-sqlite/src/schema.ts b/packages/session-persistence-sqlite/src/schema.ts index 5f20b5c08b..5e3029936c 100644 --- a/packages/session-persistence-sqlite/src/schema.ts +++ b/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( + 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 } } diff --git a/packages/session-persistence-sqlite/tests/sqlite.spec.ts b/packages/session-persistence-sqlite/tests/sqlite.spec.ts index 854013e03a..f392c4edcc 100644 --- a/packages/session-persistence-sqlite/tests/sqlite.spec.ts +++ b/packages/session-persistence-sqlite/tests/sqlite.spec.ts @@ -18,6 +18,14 @@ async function freshDbPath(): Promise { 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 }> { + 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 }> { - 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') From f1dac1b1ed36331d15e132410ca5213f8c6c5229 Mon Sep 17 00:00:00 2001 From: Tianyi Cui <53024+tianyicui@users.noreply.github.com> Date: Tue, 16 Jun 2026 00:05:25 +0800 Subject: [PATCH 3/5] fix(session-persistence-sqlite): defer crash-tail repair to append; fix all-tail materialized flag; persist schema version (review #35) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 +- .../session-persistence-sqlite/src/index.ts | 62 ++++++++++++----- .../session-persistence-sqlite/src/schema.ts | 19 ++++++ .../tests/sqlite.spec.ts | 67 +++++++++++++++++-- 4 files changed, 132 insertions(+), 22 deletions(-) diff --git a/packages/session-persistence-sqlite/README.md b/packages/session-persistence-sqlite/README.md index 5af385e63e..1cd3de593e 100644 --- a/packages/session-persistence-sqlite/README.md +++ b/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) diff --git a/packages/session-persistence-sqlite/src/index.ts b/packages/session-persistence-sqlite/src/index.ts index 8a2a326f02..51a03475dd 100644 --- a/packages/session-persistence-sqlite/src/index.ts +++ b/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 } } diff --git a/packages/session-persistence-sqlite/src/schema.ts b/packages/session-persistence-sqlite/src/schema.ts index 5e3029936c..ff01f65fe2 100644 --- a/packages/session-persistence-sqlite/src/schema.ts +++ b/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, diff --git a/packages/session-persistence-sqlite/tests/sqlite.spec.ts b/packages/session-persistence-sqlite/tests/sqlite.spec.ts index f392c4edcc..92a389687f 100644 --- a/packages/session-persistence-sqlite/tests/sqlite.spec.ts +++ b/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) From 3bef6b38c7dab583acc2b87c290a6ff8ce4fb702 Mon Sep 17 00:00:00 2001 From: Tianyi Cui <53024+tianyicui@users.noreply.github.com> Date: Tue, 16 Jun 2026 20:35:01 +0800 Subject: [PATCH 4/5] refactor(session-persistence-sqlite): drop the materialized column; use row existence as the signal (review #35) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The `materialized` INTEGER column was redundant: create()/update() already keep a lazy session in memory and write no row, so a `sessions` row is written only by the first append. Its EXISTENCE is the materialization signal — has()/list() now report exactly the sessions that have a row, matching the JSONL backend's "file exists ⇔ materialized". The column only existed to force has()/list() to FALSE for an all-tail crash (a partial first turn, zero committed events). That actually DIVERGED from the JSONL backend, whose file (and thus has()=true) survives a first append that never reached turn/end. Removing the column drops that special case: an all-tail session keeps its row and stays present, the same as JSONL. The orphaned tail rows are still removed by the deferred truncation-repair on the next append, and load() stays non-mutating. Also: add a TODO to route through a cordis db service if one is adopted, and correct the README's Node-version framing to the repo's engines (>=24). --- packages/session-persistence-sqlite/README.md | 10 ++-- .../session-persistence-sqlite/src/index.ts | 49 ++++++++----------- .../session-persistence-sqlite/src/schema.ts | 13 +++-- .../tests/sqlite.spec.ts | 13 +++-- 4 files changed, 41 insertions(+), 44 deletions(-) diff --git a/packages/session-persistence-sqlite/README.md b/packages/session-persistence-sqlite/README.md index 9c0ac07b39..dd72ddfde4 100644 --- a/packages/session-persistence-sqlite/README.md +++ b/packages/session-persistence-sqlite/README.md @@ -2,17 +2,19 @@ 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. +> **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. + ## Storage model -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. +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. A `sessions` row is written only by the first `append` — its existence is the lazy-materialization signal (`has`/`list` report exactly the sessions that have a row), so no separate column is needed. -`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. +The repo targets Node ≥ 24 (the root `engines` field), which includes the stable `node:sqlite` module. 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 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), 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. +- **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`. ## Configuration (schemastery) diff --git a/packages/session-persistence-sqlite/src/index.ts b/packages/session-persistence-sqlite/src/index.ts index 51a03475dd..45d0215384 100644 --- a/packages/session-persistence-sqlite/src/index.ts +++ b/packages/session-persistence-sqlite/src/index.ts @@ -245,18 +245,12 @@ export class SessionPersistenceSqlite extends SessionPersistence { // 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) - } - + // + // 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 @@ -264,7 +258,7 @@ export class SessionPersistenceSqlite extends SessionPersistence { this.states.set(id, { meta: { ...meta }, cursor: committed.length, - materialized, + materialized: true, ...cutTail ? { repairFrom: committed.length } : {}, }) return { meta, events } @@ -272,11 +266,11 @@ export class SessionPersistenceSqlite extends SessionPersistence { async list(): Promise { await this.ready - // Materialized rows only: a created-but-never-appended (lazy) session has no - // row at all, and a load that cut every event back to zero leaves - // materialized = 0. Both are excluded, matching has(). + // Every metadata row is a materialized session: the row is written only by + // the first append (a created-but-never-appended session has no row), so + // listing all rows is exactly the materialized set. const rows = this.db - .prepare('SELECT * FROM sessions WHERE materialized = 1') + .prepare('SELECT * FROM sessions') .all() as unknown as SessionRow[] return rows.map(rowToMeta) } @@ -285,8 +279,8 @@ export class SessionPersistenceSqlite extends SessionPersistence { await this.ready const state = this.states.get(id) if (state?.materialized) return true - const row = this.rowFor(id) - return row !== undefined && row.materialized === 1 + // A metadata row exists iff the session was materialized by a first append. + return this.rowFor(id) !== undefined } delete(id: SessionId): Promise { @@ -326,15 +320,15 @@ export class SessionPersistenceSqlite extends SessionPersistence { } /** - * Insert-or-replace a session's metadata row, marked materialized. The only - * callers are the first materializing `append` and a post-materialization - * `update` — a row is written only once a session has durable events, so - * `materialized` is always 1 (a never-appended session has no row at all). + * Insert-or-replace a session's metadata row. The only callers are the first + * materializing `append` and a post-materialization `update`, so writing the + * row IS the materialization (its existence is the signal `has`/`list` read); + * a never-appended session has no row at all. */ private writeRow(meta: SessionMeta): void { this.db.prepare(` - INSERT INTO sessions (id, version, created_at, cwd, parent_session, updated_at, title, first_prompt, materialized) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, 1) + INSERT INTO sessions (id, version, created_at, cwd, parent_session, updated_at, title, first_prompt) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(id) DO UPDATE SET version = excluded.version, created_at = excluded.created_at, @@ -342,8 +336,7 @@ export class SessionPersistenceSqlite extends SessionPersistence { parent_session = excluded.parent_session, updated_at = excluded.updated_at, title = excluded.title, - first_prompt = excluded.first_prompt, - materialized = excluded.materialized + first_prompt = excluded.first_prompt `).run( meta.id, meta.version, @@ -466,7 +459,7 @@ export class SessionPersistenceSqlite extends SessionPersistence { } const row = this.rowFor(id) - if (row !== undefined && row.materialized === 1) { + if (row !== undefined) { const stored = this.eventsFor(id) if (!seedCoversPrefix(seed, stored)) { throw new Error(`session "${id}" already has a persisted log that does not match this live session (id collision)`) diff --git a/packages/session-persistence-sqlite/src/schema.ts b/packages/session-persistence-sqlite/src/schema.ts index ff01f65fe2..c3ef9ca051 100644 --- a/packages/session-persistence-sqlite/src/schema.ts +++ b/packages/session-persistence-sqlite/src/schema.ts @@ -18,10 +18,11 @@ import type { SessionEvent, SessionId, SessionMeta } from '@deepseek-ai/dsh-sess export const SCHEMA_VERSION = 1 /** - * A row of the `sessions` table — the out-of-log metadata (`SessionMeta`) plus - * the `materialized` flag that implements lazy materialization (a created-but- - * never-appended session has `materialized = 0` and is excluded from - * `has`/`list`, mirroring the JSONL backend's "no file until first append"). + * A row of the `sessions` table — the out-of-log metadata (`SessionMeta`). The + * row's EXISTENCE is the materialization signal: it is written only by the + * first `append` (lazy materialization), so a created-but-never-appended + * session has no row and is absent from `has`/`list`, mirroring the JSONL + * backend's "no file until first append". */ export interface SessionRow { id: string @@ -32,7 +33,6 @@ export interface SessionRow { updated_at: number title: string | null first_prompt: string | null - materialized: number } /** An `events` table row: one `SessionEvent` mapped 1:1 (`data` is JSON text). */ @@ -81,8 +81,7 @@ export function openDatabase(path: string): DatabaseSync { parent_session TEXT, updated_at INTEGER NOT NULL, title TEXT, - first_prompt TEXT, - materialized INTEGER NOT NULL DEFAULT 0 + first_prompt TEXT ) STRICT `) db.exec(` diff --git a/packages/session-persistence-sqlite/tests/sqlite.spec.ts b/packages/session-persistence-sqlite/tests/sqlite.spec.ts index 92a389687f..6ab4dae30b 100644 --- a/packages/session-persistence-sqlite/tests/sqlite.spec.ts +++ b/packages/session-persistence-sqlite/tests/sqlite.spec.ts @@ -139,7 +139,7 @@ describe('SessionPersistenceSqlite: durability and crash semantics', () => { await b2.dispose() }) - it('all-tail load: a session whose only content is a crash tail is absent from has()/list()', async () => { + it('all-tail load: a session materialized by a partial first turn stays present (JSONL parity), load returns zero committed events', async () => { const path = await freshDbPath() const m = meta('all-tail') const b1 = await backend(path) @@ -153,12 +153,15 @@ describe('SessionPersistenceSqlite: durability and crash semantics', () => { 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. + // 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. 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) + 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() }) @@ -269,7 +272,7 @@ describe('SessionPersistenceSqlite: durability and crash semantics', () => { const path = await freshDbPath() // Materialize a row with version 2 directly via the real schema. const db = openDatabase(path) - db.prepare('INSERT INTO sessions (id, version, created_at, updated_at, materialized) VALUES (?, ?, ?, ?, 1)') + db.prepare('INSERT INTO sessions (id, version, created_at, updated_at) VALUES (?, ?, ?, ?)') .run('v2', 2, 1, 1) db.close() From ce4b32ace9f46d8967d7bffd788e7aa14b4a2cf8 Mon Sep 17 00:00:00 2001 From: Tianyi Cui <53024+tianyicui@users.noreply.github.com> Date: Tue, 16 Jun 2026 22:56:17 +0800 Subject: [PATCH 5/5] feat(session-persistence-sqlite): preserve interrupted turns on load, don't truncate MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 6 +- .../session-persistence-sqlite/src/index.ts | 135 +++++++++------- .../session-persistence-sqlite/src/schema.ts | 92 +++++++---- .../tests/sqlite.spec.ts | 149 +++++++++++------- 4 files changed, 232 insertions(+), 150 deletions(-) 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' } } },