mirror of
https://github.com/deepseek-ai/deepseek-harness
synced 2026-08-15 21:04:50 +00:00
Merge pull request #35 from deepseek-ai/split/session-persistence-sqlite
feat(session-persistence-sqlite): second backend validating the abstraction (split 5/5)
This commit is contained in:
Tianyi Cui
GitHub
@@ -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; a crashed turn is closed, never truncated.** 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 durable log whose final turn never closed: real, fully-written events sit after the last `turn/end`. **A single turn can be huge in a long-horizon task** (many steps, large tool output spanning a long autonomous run), so discarding the interrupted turn would silently destroy a large amount of real work — truncating a turn is wrong. Instead, on reload `load` PRESERVES those events and CLOSES the orphaned turn by durably appending the minimal synthetic boundary events: an error `tool/result` for every `tool-call` the crash left unanswered, then a `step/end` if a step was still open, then a `turn/end` carrying the merge-extensible `{ kind: 'interrupted' }` reason (a marker that records the turn was cut short by a crash, not completed by the model — no loop ever emits it). The synthetic tool results matter for resume correctness: the loop logs the `assistant/message` (carrying the `tool-call` blocks) BEFORE running the tools, so a crash mid-tool leaves calls without results; `deriveMessages()` would then replay a dangling assistant tool-call, which every provider rejects as an invalid transcript on the next request. Answering each orphaned call with an error result keeps the rehydrated history valid. `load` returns the balanced log, so a resumed session is immediately usable. The ONLY thing discarded is a never-fully-written **torn tail fragment** — a final record whose bytes (JSONL) or row were never completely flushed; that fragment is not a valid event and is dropped before the synthetic closers are written. A parse error or `seq` gap in the COMMITTED region (at or before the last real `turn/end`) is genuine corruption and makes the session unloadable.
|
||||
- **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, interrupted-turn close 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 clear error when it is absent.
|
||||
|
||||
|
||||
@@ -85,7 +85,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, crash recovery that PRESERVES an interrupted turn (closing it with a synthetic `turn/end {interrupted}` rather than truncating — a turn can be huge), 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, crash recovery that PRESERVES an interrupted turn (closing it with a synthetic `turn/end {interrupted}` rather than truncating — a turn can be huge), 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)
|
||||
|
||||
@@ -231,7 +231,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.
|
||||
|
||||
@@ -20,6 +20,8 @@ graph TD
|
||||
invariants --> session
|
||||
session-persistence-jsonl --> session
|
||||
session-persistence-jsonl --> session-persistence
|
||||
session-persistence-sqlite --> session
|
||||
session-persistence-sqlite --> session-persistence
|
||||
tools --> agent
|
||||
tools --> llm
|
||||
tools --> system-prompt
|
||||
@@ -48,6 +50,7 @@ graph TD
|
||||
| `session-persistence` | `session` |
|
||||
| `invariants` | `agent`, `llm`, `session` |
|
||||
| `session-persistence-jsonl` | `session`, `session-persistence` |
|
||||
| `session-persistence-sqlite` | `session`, `session-persistence` |
|
||||
| `tools` | `agent`, `llm`, `system-prompt` |
|
||||
| `agent-loop` | `agent`, `llm`, `session`, `session-persistence`, `system-prompt`, `tools` |
|
||||
| `tool-bash` | `agent`, `bash`, `llm`, `tools` |
|
||||
|
||||
29
packages/session-persistence-sqlite/README.md
Normal file
29
packages/session-persistence-sqlite/README.md
Normal file
@@ -0,0 +1,29 @@
|
||||
# @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, 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.
|
||||
|
||||
## 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. 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.
|
||||
|
||||
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 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).
|
||||
- **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)
|
||||
|
||||
```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.
|
||||
35
packages/session-persistence-sqlite/package.json
Normal file
35
packages/session-persistence-sqlite/package.json
Normal file
@@ -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": "workspace:^",
|
||||
"@deepseek-ai/dsh-session-persistence": "workspace:^",
|
||||
"cordis": "^4.0.0-rc.6"
|
||||
}
|
||||
}
|
||||
536
packages/session-persistence-sqlite/src/index.ts
Normal file
536
packages/session-persistence-sqlite/src/index.ts
Normal file
@@ -0,0 +1,536 @@
|
||||
/**
|
||||
* 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
|
||||
* / 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
|
||||
* `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, interruptedTurnClosers } from '@deepseek-ai/dsh-session'
|
||||
import type { Session, SessionEvent, SessionId, SessionMeta, SessionSummary } from '@deepseek-ai/dsh-session'
|
||||
import {
|
||||
openDatabase, rowToMeta, scanRows, 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<Config> = z.object({
|
||||
path: z.string().required(),
|
||||
})
|
||||
|
||||
private db!: DatabaseSync
|
||||
private ready: Promise<void>
|
||||
/** Backend bookkeeping keyed by session id (NOT the live Session object). */
|
||||
private states = new Map<string, SessionState>()
|
||||
/** Write-behind buffers keyed by the live Session (write path). */
|
||||
private buffers = new Map<Session, SessionEvent[]>()
|
||||
/** Per-session serialization chain (keyed by session id). */
|
||||
private chains = new Map<string, Promise<unknown>>()
|
||||
/** Per-session init promise (onCreated), keyed by the LIVE Session object. */
|
||||
private inits = new Map<Session, Promise<void>>()
|
||||
|
||||
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<void> {
|
||||
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<void> {
|
||||
const snapshot: SessionMeta = { ...meta }
|
||||
return this.serialize(snapshot.id, () => this.createCore(snapshot))
|
||||
}
|
||||
|
||||
private async createCore(meta: SessionMeta): Promise<void> {
|
||||
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 })
|
||||
}
|
||||
|
||||
// `async` so the synchronous validate/clone below reject (not throw) per the
|
||||
// Promise<void> contract — callers use `await expect(...).rejects`.
|
||||
async append(id: SessionId, events: readonly SessionEvent[]): Promise<void> {
|
||||
// Validate serializability BEFORE cloning so a bad event surfaces the typed
|
||||
// "non-JSON-serializable" error rather than an opaque DataCloneError from
|
||||
// structuredClone. Then deep-snapshot the batch HERE, before the op waits
|
||||
// behind the per-session chain: a caller that passes a live array (e.g.
|
||||
// session.events) and mutates it — OR mutates an event inside it — before
|
||||
// the op runs would otherwise have those changes persisted, or advance the
|
||||
// cursor past what was written. The clone is taken at call time (before the
|
||||
// first await), matching the JSONL backend.
|
||||
assertSerializable(events)
|
||||
const batch = events.map(e => structuredClone(e))
|
||||
return this.serialize(id, () => this.appendCore(id, batch))
|
||||
}
|
||||
|
||||
private async appendCore(id: SessionId, events: readonly SessionEvent[]): Promise<void> {
|
||||
await this.ready
|
||||
if (events.length === 0) return
|
||||
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. (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 {
|
||||
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 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 { preserved, tornFrom } = scanRows(eventRows)
|
||||
|
||||
// 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: balanced.length,
|
||||
materialized: true,
|
||||
})
|
||||
return { meta, events: balanced }
|
||||
}
|
||||
|
||||
async list(): Promise<SessionMeta[]> {
|
||||
await this.ready
|
||||
// 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')
|
||||
.all() as unknown as SessionRow[]
|
||||
return rows.map(rowToMeta)
|
||||
}
|
||||
|
||||
async has(id: SessionId): Promise<boolean> {
|
||||
await this.ready
|
||||
const state = this.states.get(id)
|
||||
if (state?.materialized) return true
|
||||
// A metadata row exists iff the session was materialized by a first append.
|
||||
return this.rowFor(id) !== undefined
|
||||
}
|
||||
|
||||
delete(id: SessionId): Promise<void> {
|
||||
return this.serialize(id, () => this.deleteCore(id))
|
||||
}
|
||||
|
||||
private async deleteCore(id: SessionId): Promise<void> {
|
||||
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<SessionSummary>): Promise<void> {
|
||||
return this.serialize(id, () => this.updateCore(id, summary))
|
||||
}
|
||||
|
||||
private async updateCore(id: SessionId, summary: Partial<SessionSummary>): Promise<void> {
|
||||
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. 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)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
|
||||
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
|
||||
`).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<SessionState> {
|
||||
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<T>(id: SessionId, op: () => Promise<T>): Promise<T> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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) {
|
||||
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 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[]
|
||||
// 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. */
|
||||
private async seedMatchesPersisted(id: SessionId, seed: readonly SessionEvent[], cursor: number): Promise<boolean> {
|
||||
await this.ready
|
||||
if (cursor === 0) return true
|
||||
return seedCoversPrefix(seed, this.eventsFor(id).slice(0, cursor))
|
||||
}
|
||||
|
||||
private async flush(session: Session): Promise<void> {
|
||||
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<void> {
|
||||
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
|
||||
190
packages/session-persistence-sqlite/src/schema.ts
Normal file
190
packages/session-persistence-sqlite/src/schema.ts
Normal file
@@ -0,0 +1,190 @@
|
||||
/**
|
||||
* 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`). 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
|
||||
version: number
|
||||
created_at: number
|
||||
cwd: string | null
|
||||
parent_session: string | null
|
||||
updated_at: number
|
||||
title: string | null
|
||||
first_prompt: string | null
|
||||
}
|
||||
|
||||
/** 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).
|
||||
*
|
||||
* 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,
|
||||
version INTEGER NOT NULL,
|
||||
created_at INTEGER NOT NULL,
|
||||
cwd TEXT,
|
||||
parent_session TEXT,
|
||||
updated_at INTEGER NOT NULL,
|
||||
title TEXT,
|
||||
first_prompt TEXT
|
||||
) 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 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).
|
||||
*
|
||||
* 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 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 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 }
|
||||
}
|
||||
})
|
||||
|
||||
// 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 }
|
||||
}
|
||||
752
packages/session-persistence-sqlite/tests/sqlite.spec.ts
Normal file
752
packages/session-persistence-sqlite/tests/sqlite.spec.ts
Normal file
@@ -0,0 +1,752 @@
|
||||
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 { openDatabase, scanRows, type EventRow } 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<string> {
|
||||
const dir = await mkdtemp(join(tmpdir(), 'dsh-sqlite-'))
|
||||
dirs.push(dir)
|
||||
return join(dir, 'sessions.db')
|
||||
}
|
||||
|
||||
/** A context with the session store + SQLite backend, plus a teardown. */
|
||||
async function backend(path = ':memory:'): Promise<{ ctx: Context; dispose: () => Promise<void> }> {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
const fiber = await ctx.plugin(SessionPersistenceSqlite, { path })
|
||||
return { ctx, dispose: () => fiber.dispose() }
|
||||
}
|
||||
|
||||
// The payoff: the SAME backend-agnostic contract the JSONL backend runs, now
|
||||
// proving the SQLite backend satisfies identical semantics.
|
||||
runPersistenceContract('sqlite', async () => {
|
||||
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('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: 'step/start', seq: 7, time: 8, data: { turn: 2, step: 1 } },
|
||||
]
|
||||
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('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: 'step/start', seq: 2, time: 2, data: { turn: 1, step: 1 } }, // seq 1 missing
|
||||
]
|
||||
const { preserved, tornFrom } = scanRows(rows(gapped))
|
||||
expect(preserved.map(e => e.seq)).toEqual([0])
|
||||
expect(tornFrom).toBe(1)
|
||||
})
|
||||
|
||||
it('an empty log preserves nothing and has no torn tail', () => {
|
||||
expect(scanRows([])).toEqual({ preserved: [] })
|
||||
})
|
||||
|
||||
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(() => 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('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).
|
||||
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: 'step/start', seq: 7, time: 8, data: { turn: 2, step: 1 } },
|
||||
])
|
||||
await fiber1.dispose()
|
||||
|
||||
// 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.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' })
|
||||
|
||||
// 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: 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, 8, 9, 10, 11])
|
||||
await fiber2.dispose()
|
||||
})
|
||||
|
||||
it('load() durably closes the interrupted turn: the synthetic closers are on disk after load', async () => {
|
||||
const path = await freshDbPath()
|
||||
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 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' } } }))
|
||||
db.close()
|
||||
|
||||
const b2 = await backend(path)
|
||||
const loaded = await b2.ctx.sessionPersistence.load(m.id)
|
||||
// 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 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(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 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)
|
||||
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 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.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()
|
||||
})
|
||||
|
||||
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)
|
||||
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 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')
|
||||
db.close()
|
||||
|
||||
const b2 = await backend(path)
|
||||
const loaded = await b2.ctx.sessionPersistence.load(m.id)
|
||||
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' } } },
|
||||
])
|
||||
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)
|
||||
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) VALUES (?, ?, ?, ?)')
|
||||
.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', () => {
|
||||
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<Session, Promise<void>> }).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<Session, Promise<void>> }).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()
|
||||
})
|
||||
})
|
||||
15
packages/session-persistence-sqlite/tsconfig.json
Normal file
15
packages/session-persistence-sqlite/tsconfig.json
Normal file
@@ -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" }
|
||||
]
|
||||
}
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
16
pnpm-lock.yaml
generated
16
pnpm-lock.yaml
generated
@@ -215,6 +215,22 @@ importers:
|
||||
specifier: ^4.0.0-rc.6
|
||||
version: 4.0.0-rc.6(@cordisjs/plugin-include@1.0.4)(@cordisjs/plugin-loader@1.0.0-rc.4)
|
||||
|
||||
packages/session-persistence-sqlite:
|
||||
dependencies:
|
||||
schemastery:
|
||||
specifier: ^3.18.0
|
||||
version: 3.18.0
|
||||
devDependencies:
|
||||
'@deepseek-ai/dsh-session':
|
||||
specifier: workspace:^
|
||||
version: link:../session
|
||||
'@deepseek-ai/dsh-session-persistence':
|
||||
specifier: workspace:^
|
||||
version: link:../session-persistence
|
||||
cordis:
|
||||
specifier: ^4.0.0-rc.6
|
||||
version: 4.0.0-rc.6(@cordisjs/plugin-include@1.0.4)(@cordisjs/plugin-loader@1.0.0-rc.4)
|
||||
|
||||
packages/system-prompt:
|
||||
devDependencies:
|
||||
'@deepseek-ai/dsh-llm':
|
||||
|
||||
@@ -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',
|
||||
|
||||
@@ -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"],
|
||||
|
||||
@@ -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" },
|
||||
|
||||
@@ -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"],
|
||||
|
||||
Reference in New Issue
Block a user