Merge remote-tracking branch 'origin/master' into session-query-search

# Conflicts:
#	docs/config-catalog.md
#	docs/cordis-catalog/services.md
#	docs/module-graph.md
#	packages/session-persistence/session-persistence-jsonl/README.md
#	packages/session-persistence/session-persistence-jsonl/src/index.ts
#	packages/session-persistence/session-persistence-sqlite/README.md
#	packages/session-persistence/session-persistence-sqlite/src/index.ts
#	packages/session-persistence/session-persistence-sqlite/src/schema.ts
#	packages/session-persistence/session-persistence-sqlite/tests/sqlite.spec.ts
#	packages/session-persistence/session-persistence/README.md
#	packages/session-persistence/session-persistence/src/coordinator.ts
#	packages/session-persistence/session-persistence/tests/persistence.spec.ts
#	packages/session-query/session-query/README.md
This commit is contained in:
Hypatia May
2026-07-15 16:17:59 +08:00
841 changed files with 18816 additions and 16117 deletions

View File

@@ -8,14 +8,14 @@ A SQLite durable session-persistence backend — a second `SessionPersistence` i
Each `SessionEvent` maps 1:1 onto a row in an `events` table `(session_id, seq, type, time, data, source_event_seqs, surface_op)``data` is the event payload as JSON text, so the row shape is the event verbatim (including `assistant/chunk`, keeping `seq` contiguous). The two `TEXT` columns `source_event_seqs` and `surface_op` are nullable; they store the event's optional surface-metadata fields (see [session surface](../../../docs/rfc/implemented/architecture/2026-06-18-session-surface.md)). Out-of-log metadata (`SessionHeader`) and a monotonic per-log revision live in a `sessions` row; a singleton state row carries the immutable store id. A `sessions` row is written only by the first `append` — its existence is the lazy-materialization signal (`list` reports exactly the sessions that have a row).
The repo's `engines.node` is `^22.19.0 || >=24.0.0` (Node 22.19+ or 24+), matching the LTS floor required by the installed Pi adapter dependency; `node:sqlite` itself ships without the `--experimental-sqlite` flag from Node 22.13 (LTS) and 23.4 / 24 (Current) on. The range deliberately excludes Node 23 because that line is non-LTS/EOL and still has flagged runtime features before 23.6. The database opens with `foreign_keys = ON` (so `ON DELETE CASCADE` drops a session's events with its row) and the configured `journal_mode` (default `wal`; pick a rollback-journal mode like `delete` on filesystems where WAL's shared-memory files do not work, e.g. network mounts). 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 any other, incompatible build (a non-current `user_version`, older or newer) is rejected rather than opened against an unknown layout — there is no migration (unreleased software).
The repository's Node range supports unflagged `node:sqlite`. The database enables foreign keys and uses the configured journal mode (`wal` by default; use a rollback mode where WAL shared-memory files are unsuitable). `PRAGMA user_version` stores the table-layout version; databases with any other version are rejected because this unreleased format has no migrations.
## 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 `list()` (which reports exactly the sessions that have a row).
- **Lightweight revisions.** `listSnapshots()` combines the database's immutable random store id and physical file identity with the monotonic revision stored beside each session header; an in-memory database uses the store id alone. Append and mutating load repair increment the local counter in the same transaction as their event changes, so unchanged same-file observations are stable, independent stores and file replacements cannot collide on a local counter, and no full-log count or parse is required.
- **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 (an error `tool/result` for every assistant tool call left unanswered, 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 `list()` — the same as the JSONL backend, whose file likewise survives a first append that never reached `turn/end`.
- **Interrupted-turn close on load.** `load()` implements the shared [crash-recovery contract](../../../docs/rfc/implemented/architecture/2026-06-14-session-persistence.md): preserve the valid interrupted turn, append its synthetic closing events in one transaction, and remove only a torn tail row. Committed parse errors or sequence gaps make the session unloadable. Because recovery mutates stored rows, the next append starts from a balanced log and accurate cursor.
- **Lightweight revisions.** `listSnapshots()` combines an immutable store identity, the database file identity, and a per-session counter incremented in each mutating transaction. This keeps unchanged observations stable without parsing event rows and prevents independent stores from sharing a revision accidentally.
## Configuration (schemastery)
@@ -29,3 +29,18 @@ interface Config {
## Write path
Like the JSONL backend, the plugin also installs the `session/event` → buffer → `session/flush` drain: it copies each already-frozen event into a persistence-owned buffer, 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.
## Model Experience
### Resumed conversation history
**What the model sees**: SQLite storage contributes no live prompt or schema. Loading restores the same surface history as JSONL and preserves prior headers for reconstruction; the new loop composes its current envelope. Each unanswered call in interrupted rows is balanced with the exact error text `Tool call interrupted by a crash; no result was recorded.` Row metadata and raw chunks are not messages.
**Token effect**: Zero live-request tokens. Resume restores retained history and pays the current envelope, plus the quoted repair result for each interrupted call.
## Known Limitations and Deferred Work
- **`DatabaseSync` is synchronous** — every append transaction blocks the event loop for its duration; acceptable for local stores, a throughput ceiling for busy multi-session servers.
- **Write contention has no wait or retry policy** — the backend sets no busy timeout and retries no locked-database error, so another connection holding a write transaction makes the operation reject immediately.
- **Only the current `SCHEMA_VERSION` opens** — a database with any other schema version is rejected rather than migrated (unreleased software; no persisted user data to preserve).
- **Nothing deletes stored sessions** — rows accumulate until removed externally (the seam has no deletion surface; `ON DELETE CASCADE` is wired for such out-of-band cleanup).

View File

@@ -24,7 +24,7 @@
"peerDependencies": {
"@deepseek-ai/dsh-session": "^0.0.1",
"@deepseek-ai/dsh-session-persistence": "^0.0.1",
"cordis": "^4.0.0-rc.6"
"cordis": "^4.0.0-rc.7"
},
"dependencies": {
"schemastery": "^3.18.0"
@@ -32,6 +32,6 @@
"devDependencies": {
"@deepseek-ai/dsh-session": "workspace:^",
"@deepseek-ai/dsh-session-persistence": "workspace:^",
"cordis": "^4.0.0-rc.6"
"cordis": "^4.0.0-rc.7"
}
}

View File

@@ -1,19 +1,7 @@
/**
* 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, source_event_seqs, surface_op)`.
*
* Like the JSONL backend it supplies ONLY the storage primitives (the
* {@link PersistenceBackend} hooks below — INSERT/DELETE/SELECT inside
* transactions); all the write-path orchestration lives in the backend-agnostic
* {@link PersistenceCoordinator} this class composes. The stateful public
* {@link SessionPersistence} methods delegate to the coordinator.
*
* SQLite durable session-persistence backend. It maps each session header and
* event to rows, and delegates write-path orchestration to
* {@link PersistenceCoordinator}.
* @module @deepseek-ai/dsh-session-persistence-sqlite
*/
@@ -27,7 +15,7 @@ import {
SessionPersistence, SessionPersistenceRevision, PersistenceCoordinator,
type PersistenceBackend, type SessionPersistenceSnapshot, type StoredPrefix,
} from '@deepseek-ai/dsh-session-persistence'
import type { Session, SessionEvent, SurfaceEventType, SessionId, SessionHeader } from '@deepseek-ai/dsh-session'
import type { SessionEvent, SurfaceEventType, SessionId, SessionHeader } from '@deepseek-ai/dsh-session'
import {
type JournalMode, openDatabase, rowToMeta, scanRows, type EventRow, type SessionRow,
} from './schema.ts'
@@ -91,10 +79,8 @@ export class SessionPersistenceSqlite extends SessionPersistence implements Pers
constructor(ctx: Context, public config: Config) {
super(ctx)
// Open the database asynchronously (the parent directory may need creating);
// every hook awaits `ready` first. Opening synchronously would force a sync
// mkdir and block plugin apply. schemastery (static Config) has already
// filled `journalMode`; the cast records that runtime fact.
// Open asynchronously so directory creation does not block plugin apply;
// every storage hook awaits the same readiness promise.
this.ready = this.openDb(config.path, (config as Required<Config>).journalMode)
this.coordinator = new PersistenceCoordinator<number>(this.ctx, this)
}
@@ -140,18 +126,8 @@ export class SessionPersistenceSqlite extends SessionPersistence implements Pers
return this.coordinator.load(id)
}
// `list` is BOTH the public service method and the PersistenceBackend hook —
// one method (the SELECT below). The coordinator adds no orchestration for
// listing, so routing it through the coordinator would just recurse. Defined
// once, in the "PersistenceBackend hooks" section.
/**
* The per-session init promises, exposed for white-box tests that await a
* specific session's onCreated (there is no public API to await one init).
*/
get inits(): Map<Session, Promise<void>> {
return this.coordinator.inits
}
// One method serves both public `list` and the backend hook; delegating it to
// the coordinator would call this hook recursively.
// --- PersistenceBackend hooks (the SQLite storage primitives) ---

View File

@@ -60,23 +60,11 @@ export interface EventRow {
export type JournalMode = 'wal' | 'delete' | 'truncate' | 'persist'
/**
* Open the database at `path` and apply the schema + pragmas. `foreign_keys`
* makes `ON DELETE CASCADE` drop a session's events with its row; the
* `journal_mode` pragma is set from the plugin's `journalMode` config (`wal`
* default — 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 whose version is NOT the
* current one (written by a different, incompatible build — older or newer) is
* REJECTED rather than opened against a layout this build does not understand.
* There are no migrations: an incompatible layout is rejected. The current
* persistence-state row carries an immutable random store id, the sessions row
* carries every header field plus its monotonic snapshot revision, and the
* events row carries the complete surface metadata.
* Open the database and apply its schema and pragmas. A zero `user_version` is
* stamped with {@link SCHEMA_VERSION}; every other non-current version rejects
* rather than being migrated in place.
* @param path - the SQLite database file to open (created when absent).
* @param journalMode - the journal pragma to apply — a closed in-code union, validated by the plugin Config.
* @param journalMode - validated journal pragma.
* @returns the open handle with pragmas applied and all three tables ensured.
*/
export function openDatabase(path: string, journalMode: JournalMode): DatabaseSync {
@@ -92,8 +80,7 @@ export function openDatabase(path: string, journalMode: JournalMode): DatabaseSy
function configureDatabase(db: DatabaseSync, path: string, journalMode: JournalMode): void {
db.exec('PRAGMA foreign_keys = ON')
// journalMode is a closed in-code union (validated by the plugin Config), not
// user-controlled SQL — safe to interpolate (PRAGMA takes no bound params).
// The validated union is safe to interpolate into a non-bindable PRAGMA.
db.exec(`PRAGMA journal_mode = ${journalMode.toUpperCase()}`)
// `PRAGMA user_version` always returns exactly one row { user_version }.
const { user_version: onDisk } = db.prepare('PRAGMA user_version').get() as { user_version: number }
@@ -101,9 +88,7 @@ function configureDatabase(db: DatabaseSync, path: string, journalMode: JournalM
throw new Error(`session database at "${path}" has schema version ${onDisk}, incompatible with this build (${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).
// Stamp fresh or pre-versioning databases.
db.exec(`PRAGMA user_version = ${SCHEMA_VERSION}`)
}
db.exec(`
@@ -179,28 +164,11 @@ export function rowToEvent(row: EventRow): 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).
* Find the preserved prefix of ordered event rows. Fully written rows in an
* interrupted final turn remain in the prefix. The first unparsable row or seq
* gap after the last `turn/end` marks a tolerated torn tail; the same hole in
* the committed region rejects.
*
* 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 (the session-persistence RFC). 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.
* @param rows - one session's event rows, ordered by seq ascending.
* @returns the preserved event prefix, plus `tornFrom` — the seq the physical
* delete starts at — when a torn tail exists.
@@ -224,12 +192,8 @@ export function scanRows(rows: readonly EventRow[]): { preserved: SessionEvent[]
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).
// Preserve the contiguous prefix, including a complete interrupted turn;
// holes through the last committed boundary throw, while later holes stop.
const preserved: SessionEvent[] = []
for (let i = 0; i < rows.length; i++) {
const p = parsed[i]

View File

@@ -14,6 +14,19 @@ import { runCoordinatorContract, type CoordinatorFixture } from '../../session-p
const dirs: string[] = []
afterEach(async () => { for (const d of dirs.splice(0)) await rm(d, { recursive: true, force: true }) })
async function expectParallelFlushError(promise: Promise<unknown>, message: RegExp): Promise<void> {
try {
await promise
} catch (error) {
expect(error).toBeInstanceOf(AggregateError)
const [cause] = (error as AggregateError).errors as unknown[]
expect(cause).toBeInstanceOf(Error)
expect((cause as Error).message).toMatch(message)
return
}
throw new Error('expected parallel flush to reject')
}
async function freshDbPath(): Promise<string> {
const dir = await mkdtemp(join(tmpdir(), 'dsh-sqlite-'))
dirs.push(dir)
@@ -28,8 +41,7 @@ async function backend(path = ':memory:'): Promise<{ ctx: Context; dispose: () =
return { ctx, dispose: () => fiber.dispose() }
}
// The payoff: the SAME backend-agnostic contract the JSONL backend runs, now
// proving the SQLite backend satisfies identical semantics.
// Run the same backend-agnostic contract as JSONL to pin identical semantics.
runPersistenceContract('sqlite', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
@@ -40,11 +52,8 @@ runPersistenceContract('sqlite', async () => {
}
})
// Run the shared coordinator orchestration suite against the real SQLite backend.
// A FILE-backed db (not :memory:) is the shared storage scope so two mounted
// instances see the same rows (HMR/reload). `corruptTail` INSERTs a row past the
// committed seq whose `data` is invalid JSON — a never-committed torn tail that
// drives the coordinator's commitRepair-with-tornMarker branch over real db rows.
// A file-backed database lets two mounts share rows across reload. `corruptTail` inserts invalid
// JSON past the committed seq, exercising coordinator repair against real database rows.
runCoordinatorContract('sqlite', async (): Promise<CoordinatorFixture> => {
const dir = await mkdtemp(join(tmpdir(), 'dsh-sqlite-coord-'))
const path = join(dir, 'sessions.db')
@@ -66,9 +75,9 @@ runCoordinatorContract('sqlite', async (): Promise<CoordinatorFixture> => {
})
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. Surface
// fields are serialized to their nullable columns so a round trip is faithful.
// 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. Surface metadata is serialized to
// its nullable columns so the conversion remains faithful.
const rows = (events: SessionEvent[]): EventRow[] =>
events.map((e) => {
const se = e as SessionEvent<SurfaceEventType>
@@ -256,11 +265,7 @@ describe('SessionPersistenceSqlite: durability and crash semantics', () => {
})
it('rejects a sibling v3 database (the merge-collided version) rather than opening it against missing columns', async () => {
// Two unmerged branches each shipped a DISTINCT layout under user_version 3
// (one added only `seed_length`, the other only the surface columns). The
// the current build rejects every older layout; an on-disk v3 is ambiguous and is missing at least one
// of this build's columns, so it MUST be rejected, not opened. Stamp a v3
// database and confirm the version check refuses it.
// Version 3 identified two incompatible sibling layouts, so it is always rejected.
const path = await freshDbPath()
openDatabase(path, 'wal').close() // creates + stamps user_version = SCHEMA_VERSION
const db = openDatabase(path, 'wal')
@@ -277,11 +282,9 @@ describe('SessionPersistenceSqlite: durability and crash semantics', () => {
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.
// A torn row after the last committed turn has invalid JSON. `scanRows` locates the boundary
// from seq/type columns without parsing the tail, preserves the committed prefix, and load
// deletes the row; invalid JSON inside the committed region would remain fatal.
const db = openDatabase(path, 'wal')
db.prepare('INSERT INTO events (session_id, seq, type, time, data) VALUES (?, 6, ?, 7, ?)')
.run(m.id, 'turn/start', '{not valid json')
@@ -473,7 +476,7 @@ describe('SessionPersistenceSqlite: edge cases', () => {
}, { 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 expectParallelFlushError(ctx.parallel('session/flush', session), /id collision/)
await ctx.fiber.dispose()
})
})