Reorganize packages into a modular hierarchy
Move the 18 flat packages/<name> packages into role-grouped dirs: core/, llm/, bash/, session-persistence/, ui/, support/. Group dirs are pure containers; each package keeps its @deepseek-ai/dsh-* name. Collapse the per-package tsconfig paths maps (base + typecheck) into one @deepseek-ai/dsh-* wildcard with a candidate per group, and derive the publint list from the hierarchy. Update all depth-coupled globs/configs (workspace, tsdown, vitest, eslint, knip, tsconfig includes/refs, per-package tsconfigs, generators, doc-script scopes, type-equiv manifest) and the cross-package/script relative imports in tests. Fix doc-typecheck's workspacePaths() to parse tsconfig JSONC via the TypeScript API instead of a regex comment-strip, which corrupted the new wildcard `/*/` path candidates. WIP: doc cross-links and package/RFC docs still to update.
This commit is contained in:
Tianyi Cui
@@ -0,0 +1,32 @@
|
||||
# @deepseek-ai/dsh-session-persistence-jsonl
|
||||
|
||||
The JSONL durable session-persistence backend — a concrete `SessionPersistence` (the `dsh-session-persistence` seam). One append-only `.jsonl` event log per session.
|
||||
|
||||
## On-disk layout
|
||||
|
||||
```
|
||||
<root>/
|
||||
cwd-<sha256(cwd)[:12]>/ # per-project bucket (or _no-cwd/ when no cwd)
|
||||
<encoded-id>.jsonl # header line + one SessionEvent per line (verbatim)
|
||||
```
|
||||
|
||||
- The first `.jsonl` line is the immutable `SessionHeader` tagged `{ type: 'session', version, id, cwd?, createdAt, parentSession? }`; every subsequent line is one `SessionEvent` JSON, **verbatim including `assistant/chunk`** so `seq` stays contiguous (`events[i].seq === i`).
|
||||
- Session ids are unvalidated branded strings, so they are percent-encoded to a single safe path segment before use (no traversal, no collision).
|
||||
|
||||
## Config
|
||||
|
||||
| Key | Type | Notes |
|
||||
|---|---|---|
|
||||
| `root` | `string` (required) | Root directory for all session files. **No default** — a `process.cwd()` default would scatter files as the process's cwd changes (bash calls, subprocesses). |
|
||||
|
||||
## Durability and crash semantics
|
||||
|
||||
- **Lazy materialization.** `create(meta)` writes nothing; the `.jsonl` (header + first batch) is written atomically (temp-write + `fsync` + rename) on the first `append`. A created-but-never-appended session leaves nothing on disk and is absent from `has`/`list`.
|
||||
- **Append-only.** Committed events (at or below a flushed `turn/end`) are never rewritten. Subsequent appends are line appends at EOF + `fsync`.
|
||||
- **Crash recovery — close, don't truncate.** A crash can leave a log whose final turn never closed (real events after the last `turn/end`). `load` PRESERVES those events (a turn can be huge — they are real work) and closes the orphaned turn by durably appending synthetic boundary events: an error `tool/result` for every `tool-call` the crash left unanswered (the loop logs the assistant message before running the tools, so a mid-tool crash leaves dangling calls — and `deriveMessages()` would replay an assistant tool-call with no result, which providers reject), then a `step/end` if a step was open, then `turn/end {kind:'interrupted'}`, returning a balanced log. Only a never-fully-written **torn tail fragment** (a final line with no newline / unparseable) is `ftruncate`d away before the closers are written. See [session persistence](../../docs/rfc/implemented/2026-06-14-session-persistence.md).
|
||||
- **Contiguous-seq.** `load` rejects a mid-log parse error or `seq` gap (unloadable); `append` rejects a batch whose first `seq` does not continue the stored log, and rejects non-JSON-serializable `event.data` naming the offending event type.
|
||||
- **Format version.** Only v1 is supported; `load` rejects an unknown version. While the harness is unreleased a format change bumps the version and rejects non-current logs — there is no migration (no persisted user data to preserve).
|
||||
|
||||
## Write path
|
||||
|
||||
The plugin generalizes the example `session-jsonl.ts`: it subscribes to `session/created` (capture the header; persist a fork's seed once), `session/event` (snapshot each event when buffering — the live `session.events` object is mutable), and `session/flush`/dispose (drain the write-behind buffer through `append`). A per-session write cursor means a resumed session never re-appends already-stored events. Existing live sessions are seeded on plugin apply (HMR does not replay `session/created`). All backend operations for one session are serialized, and disposal awaits quiescence (every init + final drain) before returning, so no write lands after teardown.
|
||||
@@ -0,0 +1,35 @@
|
||||
{
|
||||
"name": "@deepseek-ai/dsh-session-persistence-jsonl",
|
||||
"description": "JSONL 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"
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,249 @@
|
||||
/**
|
||||
* On-disk format helpers for the JSONL session-persistence backend: path
|
||||
* sanitization (a {@link SessionId} is an unvalidated branded string, so it
|
||||
* MUST be encoded before use in a path — no traversal, no collision), the
|
||||
* per-cwd directory layout, header-line (de)serialization, and the
|
||||
* truncation-repair offset computation.
|
||||
*
|
||||
* @module dsh-session-persistence-jsonl/format
|
||||
*/
|
||||
|
||||
import { createHash } from 'node:crypto'
|
||||
import { join } from 'node:path'
|
||||
import type { SessionEvent, SessionHeader, SessionId } from '@deepseek-ai/dsh-session'
|
||||
|
||||
/**
|
||||
* The first line of a session's `.jsonl` file: the immutable
|
||||
* {@link SessionHeader} tagged as a `session` record so a reader can tell it
|
||||
* apart from an event line.
|
||||
*/
|
||||
export interface HeaderLine {
|
||||
type: 'session'
|
||||
version: number
|
||||
id: SessionId
|
||||
createdAt: number
|
||||
cwd?: string
|
||||
parentSession?: SessionId
|
||||
}
|
||||
|
||||
/** Build the header line object from a {@link SessionHeader}. */
|
||||
export function toHeaderLine(header: SessionHeader): HeaderLine {
|
||||
return {
|
||||
type: 'session',
|
||||
version: header.version,
|
||||
id: header.id,
|
||||
createdAt: header.createdAt,
|
||||
...header.cwd !== undefined ? { cwd: header.cwd } : {},
|
||||
...header.parentSession !== undefined ? { parentSession: header.parentSession } : {},
|
||||
}
|
||||
}
|
||||
|
||||
/** Parse a header line back into a {@link SessionHeader}. */
|
||||
export function fromHeaderLine(line: HeaderLine): SessionHeader {
|
||||
return {
|
||||
version: line.version,
|
||||
id: line.id,
|
||||
createdAt: line.createdAt,
|
||||
...line.cwd !== undefined ? { cwd: line.cwd } : {},
|
||||
...line.parentSession !== undefined ? { parentSession: line.parentSession } : {},
|
||||
}
|
||||
}
|
||||
|
||||
/** Type guard: a parsed first line is a well-formed session header. */
|
||||
function isHeaderLine(value: unknown): value is HeaderLine {
|
||||
return (
|
||||
typeof value === 'object' && value !== null
|
||||
&& (value as { type?: unknown }).type === 'session'
|
||||
&& typeof (value as { version?: unknown }).version === 'number'
|
||||
&& typeof (value as { id?: unknown }).id === 'string'
|
||||
&& typeof (value as { createdAt?: unknown }).createdAt === 'number'
|
||||
)
|
||||
}
|
||||
|
||||
/**
|
||||
* Encode an arbitrary string as a single safe path segment, injectively over
|
||||
* ALL JS (UTF-16) strings — including lone surrogates. A {@link SessionId} is
|
||||
* an unvalidated branded string, so this neutralizes `../`, absolute paths,
|
||||
* NUL, and separators before any filesystem use.
|
||||
*
|
||||
* Each UTF-16 code unit is either kept literal (the safe set `[A-Za-z0-9_-]`)
|
||||
* or escaped as `~XXXX` (its 4-hex-digit code unit). `~` is itself escaped, so
|
||||
* the mapping is injective and reversible: distinct inputs never collide. We
|
||||
* iterate code UNITS (`charCodeAt`), not code points, so a lone surrogate
|
||||
* escapes to a distinct `~XXXX` instead of being normalized to U+FFFD (which
|
||||
* `Buffer.from(…, 'utf8')` would do, breaking injectivity). `.` is in the safe
|
||||
* set for readability but the whole-segment tokens `.`/`..` are escaped so they
|
||||
* can never traverse.
|
||||
*/
|
||||
export function encodeSegment(raw: string): string {
|
||||
if (raw.length === 0) throw new Error('cannot encode an empty path segment')
|
||||
if (raw === '.') return '~002E'
|
||||
if (raw === '..') return '~002E~002E'
|
||||
let out = ''
|
||||
for (let i = 0; i < raw.length; i++) {
|
||||
const code = raw.charCodeAt(i)
|
||||
const ch = String.fromCharCode(code)
|
||||
if (ch !== '~' && /^[A-Za-z0-9._-]$/.test(ch)) {
|
||||
out += ch
|
||||
} else {
|
||||
out += '~' + code.toString(16).toUpperCase().padStart(4, '0')
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
/**
|
||||
* The directory a session's files live in: the configured root, then a per-cwd
|
||||
* subdirectory so sessions group by project. The cwd subdir is a stable hash
|
||||
* (short, collision-resistant, filesystem-safe) plus an encoded suffix for
|
||||
* readability; sessions without a cwd go in a shared `_no-cwd` bucket.
|
||||
*/
|
||||
export function sessionDir(root: string, cwd: string | undefined): string {
|
||||
if (cwd === undefined) return join(root, '_no-cwd')
|
||||
const hash = createHash('sha256').update(cwd).digest('hex').slice(0, 12)
|
||||
return join(root, `cwd-${hash}`)
|
||||
}
|
||||
|
||||
/** The append-only event-log file path for a session. */
|
||||
export function logPath(root: string, cwd: string | undefined, id: SessionId): string {
|
||||
return join(sessionDir(root, cwd), `${encodeSegment(id)}.jsonl`)
|
||||
}
|
||||
|
||||
/** Serialize one event as a JSONL line (no trailing newline). */
|
||||
export function eventLine(event: SessionEvent): string {
|
||||
return JSON.stringify(event)
|
||||
}
|
||||
|
||||
/**
|
||||
* Parse a JSONL log buffer into its preserved event prefix (the header is line
|
||||
* 0). Returns the longest prefix of complete, seq-contiguous events plus the
|
||||
* byte offset of the end of the last preserved line (`committedBytes`).
|
||||
*
|
||||
* A crash can leave a durable log whose final turn never closed: real,
|
||||
* fully-written events sit after the last `turn/end`. Those are PRESERVED (a
|
||||
* single turn can be huge in a long-horizon task — truncating it would destroy
|
||||
* real work); the backend closes the orphaned open turn with a synthetic
|
||||
* `turn/end {kind:'interrupted'}` on reload (the session-persistence RFC). Only a TORN trailing
|
||||
* fragment — a final line never fully flushed (no newline, unparseable, or a
|
||||
* seq gap) — is excluded; it bounds the preserved region. A parse error or seq
|
||||
* gap AT OR BEFORE the last committed `turn/end` is committed-data corruption
|
||||
* and makes the session unloadable (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 scanLog(buffer: Buffer): { meta: SessionHeader; events: SessionEvent[]; committedBytes: number } {
|
||||
const text = buffer.toString('utf8')
|
||||
// Split into complete (newline-terminated) lines, tracking the byte offset of
|
||||
// each line's end so the truncation point is exact (multi-byte chars make the
|
||||
// char offset differ from the byte offset). A trailing line with no newline is
|
||||
// an uncommitted crash fragment and is ignored — it is below the last
|
||||
// turn/end by construction (the loop only flushes whole lines).
|
||||
//
|
||||
// Track the byte offset with a RUNNING accumulator (`endByte`), adding each
|
||||
// line's byte length as we go. Recomputing `Buffer.byteLength(text.slice(0, i))`
|
||||
// per newline would rescan the whole prefix every time — O(n²) over a long
|
||||
// log (one assistant/chunk line per token makes that pathological).
|
||||
const lines: { text: string; endByte: number }[] = []
|
||||
let start = 0
|
||||
let byteOffset = 0
|
||||
for (let i = 0; i < text.length; i++) {
|
||||
if (text[i] === '\n') {
|
||||
const lineText = text.slice(start, i)
|
||||
byteOffset += Buffer.byteLength(lineText, 'utf8') + 1 // +1 for the '\n' (a 1-byte char)
|
||||
lines.push({ text: lineText, endByte: byteOffset })
|
||||
start = i + 1
|
||||
}
|
||||
}
|
||||
|
||||
const [headerEntry, ...eventEntries] = lines
|
||||
if (headerEntry === undefined) throw new Error('empty or header-less session log')
|
||||
|
||||
// Line 0 is the header.
|
||||
let parsedHeader: unknown
|
||||
try {
|
||||
parsedHeader = JSON.parse(headerEntry.text)
|
||||
} catch {
|
||||
throw new Error('corrupt session log: header line is not valid JSON')
|
||||
}
|
||||
if (!isHeaderLine(parsedHeader)) {
|
||||
throw new Error('corrupt session log: first line is not a session header')
|
||||
}
|
||||
const headerLine = parsedHeader
|
||||
|
||||
// Find the committed region: the prefix up to and including the LAST complete
|
||||
// `turn/end` in the WHOLE log. Two passes so a crash tail after the last
|
||||
// turn/end is tolerated, but corruption/gaps AT OR BEFORE the last committed
|
||||
// turn/end make the log unloadable (committed data must never be silently
|
||||
// dropped).
|
||||
//
|
||||
// Pass 1: parse every line that parses, recording (parsedOk, seq, isTurnEnd,
|
||||
// endByte) per line index. Lines that fail to parse are holes.
|
||||
interface Parsed { ok: boolean; event?: SessionEvent; endByte: number }
|
||||
const parsed: Parsed[] = eventEntries.map((entry) => {
|
||||
try {
|
||||
return { ok: true, event: JSON.parse(entry.text) as SessionEvent, endByte: entry.endByte }
|
||||
} catch {
|
||||
return { ok: false, endByte: entry.endByte }
|
||||
}
|
||||
})
|
||||
|
||||
// The last index (into eventEntries) 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--) {
|
||||
const p = parsed[i]
|
||||
if (p?.ok && p.event?.type === 'turn/end') { lastTurnEnd = i; break }
|
||||
}
|
||||
|
||||
// Walk the longest PREFIX of complete, seq-contiguous, parseable event lines
|
||||
// (line i is a parsed event with seq === i). This is the preservable region:
|
||||
// it includes any fully-written events of an interrupted final turn AFTER the
|
||||
// last turn/end — those are real, durably-written work and must NOT be
|
||||
// truncated (a single turn can be huge in a long-horizon task; the orphaned
|
||||
// open turn is closed with a synthetic turn/end on reload, not discarded —
|
||||
// the session-persistence RFC). The walk stops at the first hole (unparseable line or seq gap):
|
||||
// - if that hole is AT OR BEFORE the last committed turn/end, committed data
|
||||
// was damaged → the session is unloadable (throw);
|
||||
// - if it is AFTER (or there is no committed turn/end yet), it is the
|
||||
// tolerated crash boundary — a torn final line never fully flushed — and
|
||||
// it simply bounds the preserved tail.
|
||||
const preserved: SessionEvent[] = []
|
||||
for (let i = 0; i < parsed.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 line ${i + 1}`)
|
||||
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 at line ${i + 1} (expected ${i}, got ${p.event.seq})`)
|
||||
break // gap after the last turn/end — torn tail, stop
|
||||
}
|
||||
preserved.push(p.event)
|
||||
}
|
||||
|
||||
// committedBytes = end of the last PRESERVED line (header if none): the next
|
||||
// append truncates any torn bytes past this point before writing the
|
||||
// synthetic closers + new events.
|
||||
const lastPreserved = parsed[preserved.length - 1]
|
||||
const committedBytes = preserved.length > 0 && lastPreserved ? lastPreserved.endByte : headerEntry.endByte
|
||||
return { meta: fromHeaderLine(headerLine), events: preserved, committedBytes }
|
||||
}
|
||||
|
||||
/**
|
||||
* Parse just the header line of a log into a {@link SessionHeader}, or
|
||||
* `undefined` if it is missing/not a header. Used by `list()` to read session
|
||||
* metadata WITHOUT parsing the whole log: a session picker scales with the
|
||||
* number of sessions, not the total size of every conversation.
|
||||
*/
|
||||
export function parseHeaderMeta(firstLine: string): SessionHeader | undefined {
|
||||
let parsed: unknown
|
||||
try {
|
||||
parsed = JSON.parse(firstLine)
|
||||
} catch {
|
||||
return undefined
|
||||
}
|
||||
if (!isHeaderLine(parsed)) return undefined
|
||||
return fromHeaderLine(parsed)
|
||||
}
|
||||
@@ -0,0 +1,395 @@
|
||||
/**
|
||||
* JSONL durable session-persistence backend (`@deepseek-ai/dsh-session-persistence-jsonl`).
|
||||
*
|
||||
* One append-only `.jsonl` event log per session (a header line then one
|
||||
* `SessionEvent` per line, verbatim including `assistant/chunk` so `seq` stays
|
||||
* contiguous), with lazy materialization (no file until the first `append`),
|
||||
* atomic first write, and load-time repair of a never-committed crash tail.
|
||||
*
|
||||
* The backend supplies ONLY the file-bytes storage primitives (the
|
||||
* {@link PersistenceBackend} hooks below); all the write-path orchestration
|
||||
* (the `session/event` → buffer → `session/flush` drain, per-session
|
||||
* serialization, write cursors, fork-seed persistence, HMR live-adoption,
|
||||
* crash-repair sequencing, dispose quiescence) lives in the backend-agnostic
|
||||
* {@link PersistenceCoordinator} this class composes. The six public
|
||||
* {@link SessionPersistence} methods delegate to the coordinator.
|
||||
*
|
||||
* @module @deepseek-ai/dsh-session-persistence-jsonl
|
||||
*/
|
||||
|
||||
import { Context } from 'cordis'
|
||||
import z from 'schemastery'
|
||||
import { open, mkdir, readFile, readdir, link, rm, truncate } from 'node:fs/promises'
|
||||
import { dirname, resolve } from 'node:path'
|
||||
import { randomBytes } from 'node:crypto'
|
||||
import {
|
||||
SessionPersistence, PersistenceCoordinator,
|
||||
type PersistenceBackend, type StoredPrefix,
|
||||
} from '@deepseek-ai/dsh-session-persistence'
|
||||
import type { Session, SessionEvent, SessionId, SessionHeader } from '@deepseek-ai/dsh-session'
|
||||
import {
|
||||
encodeSegment, eventLine, logPath, parseHeaderMeta, scanLog, sessionDir, toHeaderLine,
|
||||
} from './format.ts'
|
||||
|
||||
export interface Config {
|
||||
/**
|
||||
* Root directory for all session files. Required (no default): a default of
|
||||
* `process.cwd()` would scatter session files as the process's cwd changes
|
||||
* (bash calls, subprocesses). Sessions group under per-cwd subdirectories.
|
||||
*/
|
||||
root: string
|
||||
}
|
||||
|
||||
/**
|
||||
* Whether `error` is a "no such file/directory" (`ENOENT`) failure — the ONLY
|
||||
* filesystem error that legitimately means "this session/root is absent" for a
|
||||
* durable backend. Any OTHER error (`EACCES`, `ENOTDIR`, transient I/O) must
|
||||
* surface rather than be silently reported as absence. (A NodeJS filesystem
|
||||
* rejection carries a string `code`.)
|
||||
*/
|
||||
function isENOENT(error: unknown): boolean {
|
||||
return (error as NodeJS.ErrnoException | null)?.code === 'ENOENT'
|
||||
}
|
||||
|
||||
/**
|
||||
* The JSONL persistence backend. Load as a plugin; it registers as
|
||||
* `ctx.sessionPersistence` and (via the coordinator) installs the write-path
|
||||
* listeners. Its torn-tail marker is the byte offset to truncate the log to.
|
||||
*/
|
||||
export class SessionPersistenceJsonl extends SessionPersistence implements PersistenceBackend<number> {
|
||||
static inject = ['sessions']
|
||||
|
||||
static Config: z<Config> = z.object({
|
||||
root: z.string().required(),
|
||||
})
|
||||
|
||||
/**
|
||||
* Backend label for the coordinator's dispose-failure AggregateError and
|
||||
* effect name. NOTE: this intentionally shadows cordis `Service.name` (which
|
||||
* the base sets to `'sessionPersistence'`). The service is registered under the
|
||||
* fixed key the Service constructor captured (`reflect.provide('sessionPersistence', …)`),
|
||||
* not via `this.name`, so overwriting the instance field with the backend label
|
||||
* does not affect `ctx.sessionPersistence` resolution — it only relabels the
|
||||
* dispose diagnostics, which is exactly what {@link PersistenceBackend.name} is for.
|
||||
*/
|
||||
override readonly name = 'session-persistence-jsonl'
|
||||
|
||||
private root: string
|
||||
private coordinator: PersistenceCoordinator<number>
|
||||
|
||||
constructor(ctx: Context, public config: Config) {
|
||||
super(ctx)
|
||||
// Resolve the configured root to an ABSOLUTE path ONCE, here. A relative root
|
||||
// would otherwise re-resolve against `process.cwd()` at every later
|
||||
// readdir/open — so if any plugin or test changed cwd between create, append,
|
||||
// and load, one session's files could split across directories.
|
||||
this.root = resolve(config.root)
|
||||
this.coordinator = new PersistenceCoordinator<number>(this.ctx, this)
|
||||
}
|
||||
|
||||
// --- SessionPersistence service surface (delegated to the coordinator) ---
|
||||
|
||||
create(meta: SessionHeader): Promise<void> {
|
||||
return this.coordinator.create(meta)
|
||||
}
|
||||
|
||||
append(id: SessionId, events: readonly SessionEvent[]): Promise<void> {
|
||||
return this.coordinator.append(id, events)
|
||||
}
|
||||
|
||||
load(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }> {
|
||||
return this.coordinator.load(id)
|
||||
}
|
||||
|
||||
has(id: SessionId): Promise<boolean> {
|
||||
return this.coordinator.has(id)
|
||||
}
|
||||
|
||||
delete(id: SessionId): Promise<void> {
|
||||
return this.coordinator.delete(id)
|
||||
}
|
||||
|
||||
// `list` is BOTH the public service method and the PersistenceBackend hook —
|
||||
// one method, the bucket walk below. The coordinator adds no orchestration for
|
||||
// listing (no per-id serialization, no cursor), so it would just call back into
|
||||
// this same method; routing it through the coordinator would 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
|
||||
}
|
||||
|
||||
// --- PersistenceBackend hooks (the file-bytes storage primitives) ---
|
||||
|
||||
/** Read a stored prefix by id across ALL cwd buckets (cwd unknown). */
|
||||
async loadStored(id: SessionId): Promise<StoredPrefix<number> | undefined> {
|
||||
const file = await this.findLog(id)
|
||||
if (file === undefined) return undefined
|
||||
return this.readPrefix(file.path)
|
||||
}
|
||||
|
||||
/**
|
||||
* Read a stored prefix SCOPED to `cwd` (HMR live-adoption must not cross cwd).
|
||||
* `undefined` is the DEFINITE "no-cwd" bucket, NOT "unknown" — a live session
|
||||
* with no cwd may only adopt a persisted no-cwd log, never a same-id log that
|
||||
* lives in some other cwd bucket. So this looks at exactly `logPath(cwd)`
|
||||
* (which maps `undefined` → the `_no-cwd` bucket), never the all-buckets scan.
|
||||
*/
|
||||
async loadLive(id: SessionId, cwd: string | undefined): Promise<StoredPrefix<number> | undefined> {
|
||||
const path = logPath(this.root, cwd, id)
|
||||
if (!await this.exists(path)) return undefined
|
||||
return this.readPrefix(path)
|
||||
}
|
||||
|
||||
/**
|
||||
* Read and scan a session's log file into a {@link StoredPrefix}. Folds the
|
||||
* torn-tail comparison HERE so the `tornMarker` is the byte offset to truncate
|
||||
* to (or `undefined` when nothing is torn) — the coordinator never sees the
|
||||
* raw byteLength.
|
||||
*/
|
||||
private async readPrefix(path: string): Promise<StoredPrefix<number>> {
|
||||
const buffer = await readFile(path)
|
||||
const { meta, events, committedBytes } = scanLog(buffer)
|
||||
return {
|
||||
meta,
|
||||
events,
|
||||
...committedBytes < buffer.byteLength ? { tornMarker: committedBytes } : {},
|
||||
}
|
||||
}
|
||||
|
||||
/** Durably append a batch, lazily materializing the file when not yet present. */
|
||||
async appendBatch(meta: SessionHeader, events: readonly SessionEvent[], isMaterialized: boolean): Promise<void> {
|
||||
if (isMaterialized) {
|
||||
await this.appendLines(meta, events)
|
||||
} else {
|
||||
await this.materialize(meta, events)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Make a crash repair durable: truncate the torn tail to `tornMarker` bytes (if
|
||||
* any), then append the synthetic `closers` (if any). Two fsync'd steps — the
|
||||
* seam does not require this to be atomic.
|
||||
*/
|
||||
async commitRepair(meta: SessionHeader, tornMarker: number | undefined, closers: readonly SessionEvent[]): Promise<void> {
|
||||
if (tornMarker !== undefined) await this.repair(meta, tornMarker)
|
||||
if (closers.length > 0) await this.appendLines(meta, closers)
|
||||
}
|
||||
|
||||
/** Remove a session's log file (the coordinator clears its in-memory state). */
|
||||
async deleteStored(id: SessionId): Promise<void> {
|
||||
const file = await this.findLog(id)
|
||||
if (file) await rm(file.path, { force: true })
|
||||
}
|
||||
|
||||
/** List all stored sessions' metadata (header line only — no full-log parse). */
|
||||
async list(): Promise<SessionHeader[]> {
|
||||
const metas: SessionHeader[] = []
|
||||
for (const dir of await this.listCwdDirs()) {
|
||||
for (const name of await this.listJsonl(dir)) {
|
||||
// Read ONLY the header line, not the whole log: a session picker must
|
||||
// scale with the number of sessions, not the total size of every
|
||||
// conversation (the log persists every assistant/chunk verbatim).
|
||||
const first = await this.readFirstLine(`${dir}/${name}`)
|
||||
if (first === undefined) continue // empty/half-written file
|
||||
const meta = parseHeaderMeta(first)
|
||||
if (meta === undefined) continue // not a session header
|
||||
metas.push(meta)
|
||||
}
|
||||
}
|
||||
return metas
|
||||
}
|
||||
|
||||
// --- materialization / append / repair (file mechanics) ---
|
||||
|
||||
/** Atomically write the header line + first batch (temp-write, fsync, rename). */
|
||||
private async materialize(meta: SessionHeader, events: readonly SessionEvent[]): Promise<void> {
|
||||
const dir = sessionDir(this.root, meta.cwd)
|
||||
await mkdir(this.root, { recursive: true, mode: 0o700 })
|
||||
await this.syncDir(dirname(this.root))
|
||||
await mkdir(dir, { recursive: true, mode: 0o700 })
|
||||
await this.syncDir(this.root)
|
||||
const finalPath = logPath(this.root, meta.cwd, meta.id)
|
||||
// Never rename over an existing committed log: materialize is the FIRST write
|
||||
// of a session the backend believes is new. A file here means a different
|
||||
// session shares this id on disk — reject loudly. (createCore already guards
|
||||
// the create path, so this is unreachable-in-practice TOCTOU defense.)
|
||||
/* v8 ignore next 3 -- createCore guards collisions before materialize; this is a TOCTOU backstop */
|
||||
if (await this.exists(finalPath)) {
|
||||
throw new Error(`refusing to materialize "${meta.id}": a log already exists on disk (load/resume it instead)`)
|
||||
}
|
||||
const header = JSON.stringify(toHeaderLine(meta))
|
||||
const body = events.map(eventLine).join('\n')
|
||||
const content = header + '\n' + body + '\n'
|
||||
|
||||
const tmp = `${finalPath}.${randomBytes(6).toString('hex')}.tmp`
|
||||
const handle = await open(tmp, 'wx', 0o600)
|
||||
try {
|
||||
await handle.writeFile(content)
|
||||
await handle.sync()
|
||||
} finally {
|
||||
await handle.close()
|
||||
}
|
||||
// Publish via link()+unlink(), NOT rename(): link fails with EEXIST if the
|
||||
// final path already exists, so two processes materializing the same id
|
||||
// concurrently cannot clobber each other. rename() would silently overwrite.
|
||||
let linked = false
|
||||
try {
|
||||
await link(tmp, finalPath)
|
||||
linked = true
|
||||
} finally {
|
||||
// If link FAILED, the temp is the only reference and must be removed before
|
||||
// the original error propagates. If it SUCCEEDED, defer temp cleanup to
|
||||
// AFTER the publish is durable (below) so a temp-rm failure can never reject
|
||||
// a session whose log already published.
|
||||
/* v8 ignore next -- link failure is the TOCTOU/IO race guarded above; not reachable in test */
|
||||
if (!linked) await rm(tmp, { force: true })
|
||||
}
|
||||
// link() succeeded — the log is published. fsync the directory so the new
|
||||
// entry survives a power loss: the new link is not crash-durable until the
|
||||
// parent directory's metadata is synced.
|
||||
await this.syncDir(dir)
|
||||
// Best-effort temp cleanup: the log is already published and durable, so a
|
||||
// failure to remove the (now-redundant) temp hard link must NOT reject the
|
||||
// append. Swallow only the rm failure; nothing else of consequence runs here.
|
||||
try {
|
||||
await rm(tmp, { force: true })
|
||||
} catch {
|
||||
/* v8 ignore next -- redundant temp link; publish already durable, rm failure is an unreachable IO edge */
|
||||
}
|
||||
}
|
||||
|
||||
/** fsync a directory so a just-created/renamed entry inside it is crash-durable. */
|
||||
private async syncDir(dir: string): Promise<void> {
|
||||
const handle = await open(dir, 'r')
|
||||
try {
|
||||
await handle.sync()
|
||||
} finally {
|
||||
await handle.close()
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Append event lines at EOF and fsync. On a write/sync failure AFTER the kernel
|
||||
* accepted some bytes (ENOSPC, an fsync error), truncate the file back to its
|
||||
* pre-append size before rethrowing: the cursor is unchanged, so the batch will
|
||||
* be retried, and without this rollback the retry would append AFTER the partial
|
||||
* bytes — producing duplicate seqs that make `scanLog` see a gap.
|
||||
*/
|
||||
private async appendLines(meta: SessionHeader, events: readonly SessionEvent[]): Promise<void> {
|
||||
const path = logPath(this.root, meta.cwd, meta.id)
|
||||
const handle = await open(path, 'a')
|
||||
try {
|
||||
const { size: before } = await handle.stat()
|
||||
try {
|
||||
await handle.writeFile(events.map(eventLine).join('\n') + '\n')
|
||||
await handle.sync()
|
||||
} catch (error) {
|
||||
// Roll back whatever bytes landed so a retry starts from a clean EOF.
|
||||
await handle.truncate(before)
|
||||
await handle.sync()
|
||||
throw error
|
||||
}
|
||||
} finally {
|
||||
await handle.close()
|
||||
}
|
||||
}
|
||||
|
||||
/** Truncate the log file to `offset` bytes and fsync (discard the crash tail). */
|
||||
private async repair(meta: SessionHeader, offset: number): Promise<void> {
|
||||
const path = logPath(this.root, meta.cwd, meta.id)
|
||||
await truncate(path, offset)
|
||||
const handle = await open(path, 'r+')
|
||||
try {
|
||||
await handle.sync()
|
||||
} finally {
|
||||
await handle.close()
|
||||
}
|
||||
}
|
||||
|
||||
// --- discovery helpers ---
|
||||
|
||||
/**
|
||||
* Read the first newline-terminated line of a file without loading the whole
|
||||
* file. Returns undefined if the file is empty or has no complete first line.
|
||||
* Reads in bounded chunks so a huge log costs only the header read.
|
||||
*/
|
||||
private async readFirstLine(path: string): Promise<string | undefined> {
|
||||
const handle = await open(path, 'r')
|
||||
try {
|
||||
const chunks: Buffer[] = []
|
||||
const buf = Buffer.alloc(8192)
|
||||
for (;;) {
|
||||
const { bytesRead } = await handle.read(buf, 0, buf.length, null)
|
||||
if (bytesRead === 0) return undefined // EOF with no newline → no complete line
|
||||
const slice = buf.subarray(0, bytesRead)
|
||||
const nl = slice.indexOf(0x0a)
|
||||
if (nl !== -1) {
|
||||
chunks.push(slice.subarray(0, nl))
|
||||
return Buffer.concat(chunks).toString('utf8')
|
||||
}
|
||||
chunks.push(Buffer.from(slice))
|
||||
}
|
||||
} finally {
|
||||
await handle.close()
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Find a session's log file by id across ALL cwd buckets — the any-cwd scan
|
||||
* for `loadStored`/`deleteStored` (resume and removal identify a session by id
|
||||
* alone). The cwd-scoped lookup (`loadLive`) does NOT use this; it goes
|
||||
* straight to `logPath(cwd)` so a no-cwd session can't match a real-cwd bucket.
|
||||
*/
|
||||
private async findLog(id: SessionId): Promise<{ path: string; cwd: string | undefined } | undefined> {
|
||||
const target = encodeSegment(id) + '.jsonl'
|
||||
for (const dir of await this.listCwdDirs()) {
|
||||
const path = `${dir}/${target}`
|
||||
if (await this.exists(path)) {
|
||||
// Recover the cwd from the header so the caller has the session's bucket.
|
||||
const { meta } = scanLog(await readFile(path))
|
||||
return { path, cwd: meta.cwd }
|
||||
}
|
||||
}
|
||||
return undefined
|
||||
}
|
||||
|
||||
/** The cwd-bucket directories under the root (absolute paths). */
|
||||
private async listCwdDirs(): Promise<string[]> {
|
||||
try {
|
||||
const entries = await readdir(this.root, { withFileTypes: true })
|
||||
return entries.filter(e => e.isDirectory()).map(e => `${this.root}/${e.name}`)
|
||||
} catch (error) {
|
||||
// ENOENT = the root has not been created yet → genuinely no sessions. Any
|
||||
// other error (EACCES, ENOTDIR, transient I/O) must NOT be reported as "no
|
||||
// sessions" — a durable backend cannot silently pretend state is absent.
|
||||
if (isENOENT(error)) return []
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
private async listJsonl(dir: string): Promise<string[]> {
|
||||
const entries = await readdir(dir)
|
||||
return entries.filter(n => n.endsWith('.jsonl'))
|
||||
}
|
||||
|
||||
private async exists(path: string): Promise<boolean> {
|
||||
try {
|
||||
const handle = await open(path, 'r')
|
||||
await handle.close()
|
||||
return true
|
||||
} catch (error) {
|
||||
// Only ENOENT means absent. A permission/I/O error must surface, not be
|
||||
// collapsed to `false` — otherwise load() reports "not found" and collision
|
||||
// checks proceed under a false absence assumption.
|
||||
if (isENOENT(error)) return false
|
||||
throw error
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export default SessionPersistenceJsonl
|
||||
@@ -0,0 +1,714 @@
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
import { Context } from 'cordis'
|
||||
import { appendFile, mkdtemp, mkdir, rm, readFile, writeFile, readdir, stat } 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 } from '@deepseek-ai/dsh-session'
|
||||
import SessionPersistenceJsonl from '@deepseek-ai/dsh-session-persistence-jsonl'
|
||||
import { encodeSegment, logPath, scanLog, sessionDir } from '../src/format.ts'
|
||||
import { runPersistenceContract, meta, oneTurnLog } from '../../session-persistence/tests/contract.ts'
|
||||
import { runCoordinatorContract, type CoordinatorFixture } from '../../session-persistence/tests/coordinator-contract.ts'
|
||||
|
||||
let root: string
|
||||
const dirs: string[] = []
|
||||
|
||||
async function freshRoot(): Promise<string> {
|
||||
const dir = await mkdtemp(join(tmpdir(), 'dsh-jsonl-'))
|
||||
dirs.push(dir)
|
||||
return dir
|
||||
}
|
||||
|
||||
afterEach(async () => {
|
||||
for (const d of dirs.splice(0)) await rm(d, { recursive: true, force: true })
|
||||
})
|
||||
|
||||
// Run the shared backend contract against the real JSONL backend.
|
||||
runPersistenceContract('jsonl', async () => {
|
||||
const dir = await mkdtemp(join(tmpdir(), 'dsh-jsonl-'))
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
const fiber = await ctx.plugin(SessionPersistenceJsonl, { root: dir })
|
||||
return {
|
||||
persistence: ctx.sessionPersistence,
|
||||
dispose: async () => {
|
||||
await fiber.dispose()
|
||||
await rm(dir, { recursive: true, force: true })
|
||||
},
|
||||
}
|
||||
})
|
||||
|
||||
// Run the shared coordinator orchestration suite against the real JSONL backend.
|
||||
// One temp root is the shared storage scope (two mounted instances over the same
|
||||
// root = HMR/reload). `corruptTail` appends a partial, newline-less fragment to
|
||||
// the session's .jsonl past the committed region — a never-committed torn tail
|
||||
// that drives the coordinator's commitRepair-with-tornMarker branch over real
|
||||
// file bytes.
|
||||
runCoordinatorContract('jsonl', async (): Promise<CoordinatorFixture> => {
|
||||
const dir = await mkdtemp(join(tmpdir(), 'dsh-jsonl-coord-'))
|
||||
return {
|
||||
mount: async (ctx) => {
|
||||
const fiber = await ctx.plugin(SessionPersistenceJsonl, { root: dir })
|
||||
return fiber
|
||||
},
|
||||
corruptTail: async (id, cwd) => {
|
||||
// A half-written record with no trailing newline: scanLog treats it as an
|
||||
// uncommitted crash fragment and reports committedBytes < byteLength, so
|
||||
// the coordinator sees a tornMarker to truncate.
|
||||
await appendFile(logPath(dir, cwd, id), '{"type":"assistant/chunk","seq":8,"ti')
|
||||
},
|
||||
cleanup: async () => { await rm(dir, { recursive: true, force: true }) },
|
||||
}
|
||||
})
|
||||
|
||||
describe('SessionPersistenceJsonl: format helpers', () => {
|
||||
it('encodeSegment neutralizes traversal, separators, and absolute paths', () => {
|
||||
expect(encodeSegment('..')).toBe('~002E~002E')
|
||||
expect(encodeSegment('.')).toBe('~002E')
|
||||
expect(encodeSegment('a/b')).toBe('a~002Fb')
|
||||
expect(encodeSegment('/etc/passwd')).toBe('~002Fetc~002Fpasswd')
|
||||
expect(encodeSegment('a\u0000b')).toBe('a~0000b')
|
||||
expect(encodeSegment('plain-ID_1.2')).toBe('plain-ID_1.2') // safe chars pass through
|
||||
expect(encodeSegment('a~b')).toBe('a~007Eb') // ~ itself is escaped
|
||||
})
|
||||
|
||||
it('encodeSegment is injective over UTF-16, incl. lone surrogates', () => {
|
||||
// Distinct lone surrogates must NOT collide (Buffer.from would normalize
|
||||
// both to U+FFFD; code-unit escaping keeps them distinct).
|
||||
const hi = encodeSegment(String.fromCharCode(0xD800))
|
||||
const lo = encodeSegment(String.fromCharCode(0xDC00))
|
||||
expect(hi).toBe('~D800')
|
||||
expect(lo).toBe('~DC00')
|
||||
expect(hi).not.toBe(lo)
|
||||
// A literal "~002F" input cannot collide with the encoding of "/".
|
||||
expect(encodeSegment('~002F')).not.toBe(encodeSegment('/'))
|
||||
})
|
||||
|
||||
it('encodeSegment rejects an empty id', () => {
|
||||
expect(() => encodeSegment('')).toThrow(/empty/)
|
||||
})
|
||||
})
|
||||
|
||||
describe('SessionPersistenceJsonl: durability and crash semantics', () => {
|
||||
let ctx: Context
|
||||
beforeEach(async () => {
|
||||
root = await freshRoot()
|
||||
ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
await ctx.plugin(SessionPersistenceJsonl, { root })
|
||||
})
|
||||
afterEach(async () => { await ctx.fiber.dispose() })
|
||||
|
||||
it('lazy materialization: create() writes no file until the first append', async () => {
|
||||
const m = meta('lazy', '/work')
|
||||
await ctx.sessionPersistence.create(m)
|
||||
// nothing on disk yet
|
||||
const dir = sessionDir(root, '/work')
|
||||
await expect(stat(logPath(root, '/work', m.id))).rejects.toThrow()
|
||||
expect(await ctx.sessionPersistence.has(m.id)).toBe(false)
|
||||
|
||||
await ctx.sessionPersistence.append(m.id, oneTurnLog())
|
||||
// now materialized
|
||||
expect((await stat(logPath(root, '/work', m.id))).isFile()).toBe(true)
|
||||
expect(await ctx.sessionPersistence.has(m.id)).toBe(true)
|
||||
void dir
|
||||
})
|
||||
|
||||
it('round-trip is byte-identical (incl. assistant/chunk verbatim)', async () => {
|
||||
const m = meta('chunks')
|
||||
const log: SessionEvent[] = [
|
||||
{ type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } },
|
||||
{ type: 'step/start', seq: 1, time: 2, data: { turn: 1, step: 1 } },
|
||||
{ type: 'assistant/chunk', seq: 2, time: 3, data: { turn: 1, step: 1, chunk: { type: 'text-delta', index: 0, text: 'he' } } },
|
||||
{ type: 'assistant/chunk', seq: 3, time: 4, data: { turn: 1, step: 1, chunk: { type: 'text-delta', index: 0, text: 'llo' } } },
|
||||
{ type: 'assistant/message', seq: 4, time: 5, data: { turn: 1, step: 1, content: [{ type: 'text', text: 'hello' }] } },
|
||||
{ type: 'step/end', seq: 5, time: 6, data: { turn: 1, step: 1 } },
|
||||
{ type: 'turn/end', seq: 6, time: 7, data: { turn: 1, reason: { kind: 'completed' } } },
|
||||
]
|
||||
await ctx.sessionPersistence.create(m)
|
||||
await ctx.sessionPersistence.append(m.id, log)
|
||||
const loaded = await ctx.sessionPersistence.load(m.id)
|
||||
expect(loaded.events).toEqual(log) // chunks preserved, contiguous seqs
|
||||
})
|
||||
|
||||
it('crash recovery: load preserves the interrupted turn and closes it with a synthetic turn/end {interrupted}', async () => {
|
||||
const m = meta('crash', '/proj')
|
||||
await ctx.sessionPersistence.create(m)
|
||||
await ctx.sessionPersistence.append(m.id, oneTurnLog()) // seqs 0..5, turn/end at 5
|
||||
|
||||
// Simulate a crash mid-second-turn: append raw lines that are NOT closed by
|
||||
// a turn/end (turn/start + step/start are fully written), plus a final
|
||||
// partial line with no newline (a torn fragment never fully flushed).
|
||||
const path = logPath(root, '/proj', m.id)
|
||||
await writeFile(path, [
|
||||
JSON.stringify({ type: 'turn/start', seq: 6, time: 8, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } }),
|
||||
JSON.stringify({ type: 'step/start', seq: 7, time: 9, data: { turn: 2, step: 1 } }),
|
||||
'{"type":"assistant/chunk","seq":8,"ti', // truncated partial line (no newline)
|
||||
].join('\n'), { flag: 'a' })
|
||||
|
||||
// load PRESERVES the interrupted turn's real events (turn/start 6, step/start
|
||||
// 7) — a turn can be huge, so they must not be truncated — and durably closes
|
||||
// the orphaned turn with synthetic step/end (8) + turn/end {interrupted} (9).
|
||||
const loaded = await ctx.sessionPersistence.load(m.id)
|
||||
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' })
|
||||
const stepEnd = loaded.events[8]!
|
||||
expect(stepEnd.type).toBe('step/end')
|
||||
// the torn seq-8 chunk fragment did not survive
|
||||
expect(loaded.events.some(e => e.type === 'assistant/chunk' && e.seq === 8)).toBe(false)
|
||||
|
||||
// The next append continues at seq 10 (the balanced length).
|
||||
const turn3 = [
|
||||
{ type: 'turn/start', seq: 10, time: 11, data: { turn: 3, trigger: { kind: 'message', source: { kind: 'user' } } } },
|
||||
{ type: 'turn/end', seq: 11, time: 12, data: { turn: 3, reason: { kind: 'completed' } } },
|
||||
] as SessionEvent[]
|
||||
await ctx.sessionPersistence.append(m.id, turn3)
|
||||
const reloaded = await ctx.sessionPersistence.load(m.id)
|
||||
expect(reloaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11])
|
||||
})
|
||||
|
||||
it('committed events are never rewritten: only the crash tail is repaired', async () => {
|
||||
const m = meta('append-only')
|
||||
await ctx.sessionPersistence.create(m)
|
||||
await ctx.sessionPersistence.append(m.id, oneTurnLog())
|
||||
const before = await readFile(logPath(root, undefined, m.id), 'utf8')
|
||||
const committedPrefix = before // the whole committed log
|
||||
|
||||
// A crash tail then a repair-append.
|
||||
await writeFile(logPath(root, undefined, m.id), '\n{"partial', { flag: 'a' })
|
||||
await ctx.sessionPersistence.load(m.id)
|
||||
await ctx.sessionPersistence.append(m.id, [
|
||||
{ type: 'turn/start', seq: 6, time: 9, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } },
|
||||
{ type: 'turn/end', seq: 7, time: 10, data: { turn: 2, reason: { kind: 'completed' } } },
|
||||
] as SessionEvent[])
|
||||
const after = await readFile(logPath(root, undefined, m.id), 'utf8')
|
||||
// the committed prefix is byte-for-byte intact at the head of the file
|
||||
expect(after.startsWith(committedPrefix)).toBe(true)
|
||||
})
|
||||
|
||||
it('a failed appendLines truncates partial bytes so a retry has no seq gap', async () => {
|
||||
const m = meta('truncate-retry')
|
||||
await ctx.sessionPersistence.create(m)
|
||||
await ctx.sessionPersistence.append(m.id, oneTurnLog()) // materialized, seqs 0..5
|
||||
const sizeBefore = (await stat(logPath(root, undefined, m.id))).size
|
||||
|
||||
// Force the NEXT fsync (inside appendLines) to fail once, AFTER writeFile
|
||||
// has already put bytes on disk — simulating an ENOSPC/fsync error
|
||||
// mid-append. The recovery truncate() also fsyncs, so allow that one.
|
||||
const handle = await (await import('node:fs/promises')).open(logPath(root, undefined, m.id), 'r')
|
||||
const proto = Object.getPrototypeOf(handle) as { sync: () => Promise<void> }
|
||||
await handle.close()
|
||||
const realSync = proto.sync
|
||||
let failed = false
|
||||
const spy = vi.spyOn(proto, 'sync').mockImplementation(async function (this: unknown) {
|
||||
if (!failed) { failed = true; throw new Error('simulated fsync ENOSPC') }
|
||||
return realSync.call(this)
|
||||
})
|
||||
|
||||
const turn2 = [
|
||||
{ type: 'turn/start', seq: 6, time: 9, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } },
|
||||
{ type: 'turn/end', seq: 7, time: 10, data: { turn: 2, reason: { kind: 'completed' } } },
|
||||
] as SessionEvent[]
|
||||
// The append rejects, but the partial bytes are truncated back: the file is
|
||||
// its pre-append size and the cursor is unchanged.
|
||||
await expect(ctx.sessionPersistence.append(m.id, turn2)).rejects.toThrow(/ENOSPC/)
|
||||
expect((await stat(logPath(root, undefined, m.id))).size).toBe(sizeBefore)
|
||||
spy.mockRestore()
|
||||
|
||||
// The retry now succeeds with NO seq gap — the log is contiguous 0..7.
|
||||
await ctx.sessionPersistence.append(m.id, turn2)
|
||||
const loaded = await ctx.sessionPersistence.load(m.id)
|
||||
expect(loaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7])
|
||||
})
|
||||
|
||||
it('load returns a meta copy: mutating it does not corrupt backend pathing', async () => {
|
||||
const m = meta('meta-copy', '/proj')
|
||||
await ctx.sessionPersistence.create(m)
|
||||
await ctx.sessionPersistence.append(m.id, oneTurnLog())
|
||||
const loaded = await ctx.sessionPersistence.load(m.id)
|
||||
// A consumer mutates the returned meta's cwd. The backend's stored pathing
|
||||
// metadata must be unaffected, so a later append still finds the right log.
|
||||
loaded.meta.cwd = '/evil'
|
||||
await ctx.sessionPersistence.append(m.id, [
|
||||
{ type: 'turn/start', seq: 6, time: 9, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } },
|
||||
{ type: 'turn/end', seq: 7, time: 10, data: { turn: 2, reason: { kind: 'completed' } } },
|
||||
] as SessionEvent[])
|
||||
// The append landed in the ORIGINAL /proj log, not beside an /evil path.
|
||||
const reloaded = await ctx.sessionPersistence.load(m.id)
|
||||
expect(reloaded.meta.cwd).toBe('/proj')
|
||||
expect(reloaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7])
|
||||
})
|
||||
|
||||
it('rejects a re-append of an already-stored seq', async () => {
|
||||
const m = meta('reappend')
|
||||
await ctx.sessionPersistence.create(m)
|
||||
await ctx.sessionPersistence.append(m.id, oneTurnLog())
|
||||
await expect(ctx.sessionPersistence.append(m.id, oneTurnLog())).rejects.toThrow(/seq mismatch/)
|
||||
})
|
||||
|
||||
it('path-traversal session ids are neutralized (no escape from root)', async () => {
|
||||
const evil = SessionId('../../etc/pwn')
|
||||
const m = { version: 1, id: evil, createdAt: 1 }
|
||||
await ctx.sessionPersistence.create(m)
|
||||
await ctx.sessionPersistence.append(evil, oneTurnLog())
|
||||
// The file lives UNDER root, not at ../../etc.
|
||||
const all: string[] = []
|
||||
async function walk(dir: string): Promise<void> {
|
||||
for (const e of await readdir(dir, { withFileTypes: true })) {
|
||||
const p = join(dir, e.name)
|
||||
if (e.isDirectory()) await walk(p)
|
||||
else all.push(p)
|
||||
}
|
||||
}
|
||||
await walk(root)
|
||||
expect(all.length).toBeGreaterThan(0)
|
||||
expect(all.every(p => p.startsWith(root))).toBe(true)
|
||||
})
|
||||
})
|
||||
|
||||
describe('SessionPersistenceJsonl: write path (session/event → flush)', () => {
|
||||
it('concurrent sessions do not cross buffers', async () => {
|
||||
root = await freshRoot()
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
await ctx.plugin(SessionPersistenceJsonl, { root })
|
||||
|
||||
const a = ctx.sessions.create('sa')
|
||||
const b = ctx.sessions.create('sb')
|
||||
a.append('user/message', { content: [{ type: 'text', text: 'A' }], source: { kind: 'user' } })
|
||||
b.append('user/message', { content: [{ type: 'text', text: 'B' }], source: { kind: 'user' } })
|
||||
a.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
|
||||
b.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
|
||||
await ctx.parallel('session/flush', a)
|
||||
await ctx.parallel('session/flush', b)
|
||||
|
||||
const la = await ctx.sessionPersistence.load(SessionId('sa'))
|
||||
const lb = await ctx.sessionPersistence.load(SessionId('sb'))
|
||||
expect(JSON.stringify(la.events)).toContain('"A"')
|
||||
expect(JSON.stringify(la.events)).not.toContain('"B"')
|
||||
expect(JSON.stringify(lb.events)).toContain('"B"')
|
||||
expect(JSON.stringify(lb.events)).not.toContain('"A"')
|
||||
await ctx.fiber.dispose()
|
||||
})
|
||||
|
||||
})
|
||||
|
||||
|
||||
describe('SessionPersistenceJsonl: scanLog unit', () => {
|
||||
it('rejects a header-less / empty log', () => {
|
||||
expect(() => scanLog(Buffer.from(''))).toThrow()
|
||||
})
|
||||
|
||||
it('rejects a corrupt header line', () => {
|
||||
expect(() => scanLog(Buffer.from('not json\n'))).toThrow(/header/)
|
||||
})
|
||||
|
||||
it('rejects a non-session first line', () => {
|
||||
expect(() => scanLog(Buffer.from('{"type":"event"}\n'))).toThrow(/session header/)
|
||||
})
|
||||
|
||||
it('a seq gap after the last turn/end bounds the preserved tail (torn fragment tolerated)', () => {
|
||||
const log = [
|
||||
JSON.stringify({ type: 'session', version: 1, id: 'g', createdAt: 1 }),
|
||||
JSON.stringify({ type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } }),
|
||||
JSON.stringify({ type: 'step/start', seq: 2, time: 2, data: { turn: 1, step: 1 } }), // gap: missing seq 1
|
||||
].join('\n') + '\n'
|
||||
// No committed turn/end, so the gap is a tolerated crash boundary: scanLog
|
||||
// PRESERVES the contiguous prefix (turn/start seq 0) — real interrupted-turn
|
||||
// work, not discarded — and stops at the gap. The orphaned open turn is
|
||||
// closed by loadCore's synthetic turn/end, not here.
|
||||
expect(scanLog(Buffer.from(log)).events.map(e => e.seq)).toEqual([0])
|
||||
})
|
||||
|
||||
it('rejects a seq gap BEFORE a later committed turn/end (committed data damaged)', () => {
|
||||
const log = [
|
||||
JSON.stringify({ type: 'session', version: 1, id: 'g2', createdAt: 1 }),
|
||||
JSON.stringify({ type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } }),
|
||||
JSON.stringify({ type: 'step/start', seq: 2, time: 2, data: { turn: 1, step: 1 } }), // gap: missing seq 1
|
||||
JSON.stringify({ type: 'turn/end', seq: 3, time: 3, data: { turn: 1, reason: { kind: 'completed' } } }),
|
||||
].join('\n') + '\n'
|
||||
// A turn/end exists, so the prefix up to it is committed — but it has a hole.
|
||||
// Truncating it would silently drop committed data → unloadable.
|
||||
expect(() => scanLog(Buffer.from(log))).toThrow(/seq gap in committed region/)
|
||||
})
|
||||
|
||||
it('rejects a corrupt line BEFORE a later committed turn/end (committed data damaged)', () => {
|
||||
const log = [
|
||||
JSON.stringify({ type: 'session', version: 1, id: 'c', createdAt: 1 }),
|
||||
'{not json', // corrupt, sits in the committed region (a turn/end follows)
|
||||
JSON.stringify({ type: 'turn/end', seq: 1, time: 2, data: { turn: 1, reason: { kind: 'completed' } } }),
|
||||
].join('\n') + '\n'
|
||||
expect(() => scanLog(Buffer.from(log))).toThrow(/unparsable committed event/)
|
||||
})
|
||||
|
||||
it('a header-only log (no event lines at all) preserves nothing — committedBytes is the header', () => {
|
||||
const log = JSON.stringify({ type: 'session', version: 1, id: 'h0', createdAt: 1 }) + '\n'
|
||||
const scanned = scanLog(Buffer.from(log))
|
||||
expect(scanned.events).toEqual([])
|
||||
// committedBytes falls back to the header line's end (no preserved events).
|
||||
expect(scanned.committedBytes).toBe(Buffer.byteLength(log, 'utf8'))
|
||||
})
|
||||
|
||||
it('a corrupt line after the last turn/end bounds the preserved tail', () => {
|
||||
const log = [
|
||||
JSON.stringify({ type: 'session', version: 1, id: 'c2', createdAt: 1 }),
|
||||
JSON.stringify({ type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } }),
|
||||
'{not json', // corrupt crash fragment, no turn/end committed
|
||||
].join('\n') + '\n'
|
||||
// The contiguous prefix (turn/start seq 0) is preserved; the corrupt
|
||||
// fragment after it is the tolerated crash boundary.
|
||||
expect(scanLog(Buffer.from(log)).events.map(e => e.seq)).toEqual([0])
|
||||
})
|
||||
|
||||
it('tolerates a seq gap AFTER a turn/end (uncommitted tail)', () => {
|
||||
const log = [
|
||||
JSON.stringify({ type: 'session', version: 1, id: 't', createdAt: 1 }),
|
||||
JSON.stringify({ type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } }),
|
||||
JSON.stringify({ type: 'turn/end', seq: 1, time: 2, data: { turn: 1, reason: { kind: 'completed' } } }),
|
||||
JSON.stringify({ type: 'step/start', seq: 9, time: 3, data: { turn: 2, step: 1 } }), // gap in uncommitted tail
|
||||
].join('\n') + '\n'
|
||||
const { events } = scanLog(Buffer.from(log))
|
||||
expect(events.map(e => e.seq)).toEqual([0, 1]) // tail dropped
|
||||
})
|
||||
})
|
||||
|
||||
describe('SessionPersistenceJsonl: edge cases', () => {
|
||||
let ctx: Context
|
||||
beforeEach(async () => {
|
||||
root = await freshRoot()
|
||||
ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
await ctx.plugin(SessionPersistenceJsonl, { root })
|
||||
})
|
||||
afterEach(async () => { await ctx.fiber.dispose() })
|
||||
|
||||
it('append rejects non-JSON-serializable undefined-producing data', async () => {
|
||||
const m = meta('undef')
|
||||
await ctx.sessionPersistence.create(m)
|
||||
// A value whose JSON.stringify yields undefined (a bare function as data).
|
||||
const bad = [{ type: 'user/message', seq: 0, time: 1, data: (() => 0) as unknown }] as unknown as SessionEvent[]
|
||||
await expect(ctx.sessionPersistence.append(m.id, bad)).rejects.toThrow(/non-JSON-serializable/)
|
||||
})
|
||||
|
||||
it('create snapshots its meta: mutating the caller object after the call is ignored', async () => {
|
||||
const m = meta('create-snap', '/orig')
|
||||
const p = ctx.sessionPersistence.create(m)
|
||||
// Mutate the caller's meta object immediately after calling create.
|
||||
m.cwd = '/mutated'
|
||||
await p
|
||||
await ctx.sessionPersistence.append(SessionId('create-snap'), oneTurnLog())
|
||||
// The log materialized under the ORIGINAL cwd, not the mutated one.
|
||||
expect((await stat(logPath(root, '/orig', SessionId('create-snap')))).isFile()).toBe(true)
|
||||
await expect(stat(logPath(root, '/mutated', SessionId('create-snap')))).rejects.toThrow()
|
||||
})
|
||||
|
||||
it('list discovers sessions across multiple cwd buckets', async () => {
|
||||
await ctx.sessionPersistence.create(meta('p1', '/projA'))
|
||||
await ctx.sessionPersistence.append(SessionId('p1'), oneTurnLog())
|
||||
await ctx.sessionPersistence.create(meta('p2', '/projB'))
|
||||
await ctx.sessionPersistence.append(SessionId('p2'), oneTurnLog())
|
||||
await ctx.sessionPersistence.create(meta('p3')) // no cwd → _no-cwd bucket
|
||||
await ctx.sessionPersistence.append(SessionId('p3'), oneTurnLog())
|
||||
|
||||
const ids = (await ctx.sessionPersistence.list()).map(x => x.id).sort()
|
||||
expect(ids).toEqual(['p1', 'p2', 'p3'])
|
||||
})
|
||||
|
||||
it('list on an empty root returns nothing', async () => {
|
||||
expect(await ctx.sessionPersistence.list()).toEqual([])
|
||||
})
|
||||
|
||||
it('list skips empty and non-header .jsonl files (metadata-only read)', async () => {
|
||||
// A real session…
|
||||
await ctx.sessionPersistence.create(meta('real', '/p'))
|
||||
await ctx.sessionPersistence.append(SessionId('real'), oneTurnLog())
|
||||
// …alongside two junk files in the _no-cwd bucket: an EMPTY file (readFirstLine
|
||||
// returns undefined) and a file whose first line is not a session header
|
||||
// (parseHeaderMeta returns undefined). Both are skipped, not listed.
|
||||
const bucket = join(root, '_no-cwd')
|
||||
await mkdir(bucket, { recursive: true })
|
||||
await writeFile(join(bucket, 'empty.jsonl'), '')
|
||||
await writeFile(join(bucket, 'notheader.jsonl'), '{"type":"turn/start"}\n')
|
||||
await writeFile(join(bucket, 'badjson.jsonl'), 'not json at all\n')
|
||||
|
||||
const ids = (await ctx.sessionPersistence.list()).map(x => x.id).sort()
|
||||
expect(ids).toEqual(['real'])
|
||||
})
|
||||
|
||||
it('list reads a header line longer than the 8KB read chunk', async () => {
|
||||
// readFirstLine accumulates across reads when the first line exceeds its
|
||||
// buffer. Plant a valid header whose line is > 8192 bytes (a long extra
|
||||
// field is tolerated by the header type guard) and confirm list() reads it.
|
||||
const bucket = join(root, '_no-cwd')
|
||||
await mkdir(bucket, { recursive: true })
|
||||
const bigHeader = JSON.stringify({ type: 'session', version: 1, id: 'big', createdAt: 1, pad: 'x'.repeat(9000) })
|
||||
await writeFile(join(bucket, 'big.jsonl'), bigHeader + '\n')
|
||||
const ids = (await ctx.sessionPersistence.list()).map(x => x.id)
|
||||
expect(ids).toContain('big')
|
||||
})
|
||||
|
||||
it('has() finds a session on disk under an unknown cwd (cross-bucket scan)', async () => {
|
||||
const m = meta('scan-me', '/somewhere')
|
||||
await ctx.sessionPersistence.create(m)
|
||||
await ctx.sessionPersistence.append(m.id, oneTurnLog())
|
||||
// A fresh backend with no in-memory state → has() must scan disk buckets.
|
||||
const ctx2 = new Context()
|
||||
await ctx2.plugin(SessionStore)
|
||||
await ctx2.plugin(SessionPersistenceJsonl, { root })
|
||||
expect(await ctx2.sessionPersistence.has(m.id)).toBe(true)
|
||||
expect(await ctx2.sessionPersistence.has(SessionId('absent'))).toBe(false)
|
||||
await ctx2.fiber.dispose()
|
||||
})
|
||||
|
||||
it('a DIFFERENT live session object reusing a disposed id gets its own init (no stale cache)', async () => {
|
||||
// Session A materializes a log under id "reuse".
|
||||
const sessFiberA = await ctx.plugin(Object.assign((inner: Context) => {
|
||||
const a = inner.sessions.create('reuse', { meta: { cwd: '/a' } })
|
||||
for (const e of oneTurnLog()) a.append(e.type, e.data)
|
||||
}, { inject: ['sessions'] }))
|
||||
// Drain A, then dispose ITS fiber (the live session A is gone) while the
|
||||
// backend stays loaded.
|
||||
for (const s of ctx.sessions.list()) await ctx.parallel('session/flush', s)
|
||||
await sessFiberA.dispose()
|
||||
|
||||
// A NEW live Session object reuses id "reuse". The init cache is keyed by
|
||||
// the Session OBJECT, so this gets its OWN onCreated (not A's stale promise)
|
||||
// — which detects the on-disk collision and rejects, rather than silently
|
||||
// appending the new session's events onto A's log under a stale cursor.
|
||||
const backend = ctx.sessionPersistence as unknown as { inits: Map<Session, Promise<void>> }
|
||||
let b!: Session
|
||||
await ctx.plugin(Object.assign((inner: Context) => {
|
||||
b = inner.sessions.create('reuse', { meta: { cwd: '/a' } })
|
||||
}, { inject: ['sessions'] }))
|
||||
await expect(backend.inits.get(b)).rejects.toThrow(/already bound to a different live session|already has a persisted log on disk/)
|
||||
})
|
||||
|
||||
it('a NO-CWD live session does NOT cross-cwd-adopt a same-id log from a real cwd bucket (loadLive is scope-exact)', async () => {
|
||||
// Backend 1: materialize a log under id "x" in the cwd "/w" bucket, then
|
||||
// dispose the WHOLE backend (so backend 2 mounts with an EMPTY states map —
|
||||
// the HMR/reload path where onCreated goes through loadLive, not a tracked
|
||||
// collision).
|
||||
await ctx.sessionPersistence.create(meta('x', '/w'))
|
||||
await ctx.sessionPersistence.append(SessionId('x'), oneTurnLog())
|
||||
await ctx.fiber.dispose()
|
||||
|
||||
// Backend 2 over the SAME root. A live no-cwd session reuses id "x". Because
|
||||
// loadLive(id, undefined) is the DEFINITE no-cwd bucket (NOT an all-buckets
|
||||
// scan), case-2 adoption does NOT match the "/w" log — so it would NOT
|
||||
// silently graft the no-cwd events onto the "/w" log with a mismatched cwd
|
||||
// (the bug a non-scope-exact loadLive caused). It falls through to the
|
||||
// new-session path, where createCore's any-cwd collision probe (loadStored)
|
||||
// catches the duplicate id and REJECTS — the id is taken in another bucket.
|
||||
const ctx2 = new Context()
|
||||
await ctx2.plugin(SessionStore)
|
||||
await ctx2.plugin(SessionPersistenceJsonl, { root })
|
||||
const backend = ctx2.sessionPersistence as unknown as { inits: Map<Session, Promise<void>> }
|
||||
let b!: Session
|
||||
await ctx2.plugin(Object.assign((inner: Context) => {
|
||||
b = inner.sessions.create('x') // no cwd
|
||||
}, { inject: ['sessions'] }))
|
||||
await expect(backend.inits.get(b)).rejects.toThrow(/already has a persisted log on disk/)
|
||||
|
||||
// The "/w" log is untouched — no no-cwd events were grafted onto it, and no
|
||||
// `_no-cwd` log for "x" was created.
|
||||
const inW = scanLog(await readFile(logPath(root, '/w', SessionId('x'))))
|
||||
expect(inW.meta.cwd).toBe('/w')
|
||||
expect(inW.events).toHaveLength(6)
|
||||
await expect(stat(logPath(root, undefined, SessionId('x')))).rejects.toThrow()
|
||||
await ctx2.fiber.dispose()
|
||||
})
|
||||
|
||||
it('a seed with matching seq/type/time but DIFFERENT data is rejected (deep prefix compare)', async () => {
|
||||
// Materialize and load (ownerless, cursor = 6).
|
||||
await ctx.sessionPersistence.create(meta('divergent', '/a'))
|
||||
await ctx.sessionPersistence.append(SessionId('divergent'), oneTurnLog())
|
||||
await ctx.sessionPersistence.load(SessionId('divergent'))
|
||||
|
||||
const backend = ctx.sessionPersistence as unknown as { inits: Map<Session, Promise<void>> }
|
||||
// A seed that keeps every seq/type/time but mutates a payload must NOT be
|
||||
// accepted as "the same session" — otherwise drain filters those seqs as
|
||||
// already persisted and the divergent payload is silently lost.
|
||||
const tampered = oneTurnLog()
|
||||
const userMsg = tampered[1]
|
||||
if (userMsg?.type === 'user/message') userMsg.data.content = [{ type: 'text', text: 'DIFFERENT' }]
|
||||
let bad!: Session
|
||||
await ctx.plugin(Object.assign((inner: Context) => {
|
||||
bad = inner.sessions.create('divergent', { seed: tampered, meta: { cwd: '/a' } })
|
||||
}, { inject: ['sessions'] }))
|
||||
await expect(backend.inits.get(bad)).rejects.toThrow(/do not match this live session|already has a persisted log/)
|
||||
})
|
||||
|
||||
it('a second live session reusing a bound id is rejected', async () => {
|
||||
// A live session materializes and owns the id.
|
||||
const firstFiber = await ctx.plugin(Object.assign((inner: Context) => {
|
||||
const a = inner.sessions.create('bound', { meta: { cwd: '/a' } })
|
||||
a.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
a.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
|
||||
}, { inject: ['sessions'] }))
|
||||
for (const s of ctx.sessions.list()) await ctx.parallel('session/flush', s)
|
||||
await firstFiber.dispose()
|
||||
|
||||
const backend = ctx.sessionPersistence as unknown as { inits: Map<Session, Promise<void>> }
|
||||
let second!: Session
|
||||
await ctx.plugin(Object.assign((inner: Context) => {
|
||||
second = inner.sessions.create('bound', { meta: { cwd: '/a' } })
|
||||
}, { inject: ['sessions'] }))
|
||||
await expect(backend.inits.get(second))
|
||||
.rejects.toThrow(/already bound to a different live session|already has a persisted log|do not match/)
|
||||
})
|
||||
|
||||
it('list returns nothing when the root directory does not exist', async () => {
|
||||
const ctx2 = new Context()
|
||||
await ctx2.plugin(SessionStore)
|
||||
await ctx2.plugin(SessionPersistenceJsonl, { root: join(root, 'does-not-exist-yet') })
|
||||
expect(await ctx2.sessionPersistence.list()).toEqual([])
|
||||
await ctx2.fiber.dispose()
|
||||
})
|
||||
|
||||
it('list surfaces a non-ENOENT root error (ENOTDIR) instead of reporting no sessions', async () => {
|
||||
// A durable backend must NOT collapse a storage fault to "no sessions". Point
|
||||
// the root at a regular FILE: readdir then fails with ENOTDIR, which must
|
||||
// propagate rather than be swallowed as an empty listing.
|
||||
const filePath = join(root, 'not-a-dir')
|
||||
await writeFile(filePath, 'x')
|
||||
const ctx2 = new Context()
|
||||
await ctx2.plugin(SessionStore)
|
||||
await ctx2.plugin(SessionPersistenceJsonl, { root: filePath })
|
||||
await expect(ctx2.sessionPersistence.list()).rejects.toThrow(/ENOTDIR/)
|
||||
await ctx2.fiber.dispose()
|
||||
})
|
||||
|
||||
it('exists() surfaces a non-ENOENT lookup error (ENOTDIR) instead of reporting absent', async () => {
|
||||
// Same contract on the existence path: a non-ENOENT error from the per-id
|
||||
// open() must surface, not be collapsed to "not found" (which would let a
|
||||
// collision check proceed under a false absence assumption). A LAZY session
|
||||
// (created, never appended) keeps its cwd in state, so has() reaches
|
||||
// loadLive(id, cwd) → exists(logPath). Make that cwd's bucket DIRECTORY a
|
||||
// regular file: open()ing `bucket/<id>.jsonl` under it then fails ENOTDIR.
|
||||
const cwd = '/x'
|
||||
const ctx2 = new Context()
|
||||
await ctx2.plugin(SessionStore)
|
||||
await ctx2.plugin(SessionPersistenceJsonl, { root })
|
||||
await ctx2.sessionPersistence.create(meta('exists-fault', cwd)) // lazy: no bucket yet
|
||||
await writeFile(sessionDir(root, cwd), 'x') // bucket path is now a FILE
|
||||
await expect(ctx2.sessionPersistence.has(SessionId('exists-fault'))).rejects.toThrow(/ENOTDIR/)
|
||||
await ctx2.fiber.dispose()
|
||||
})
|
||||
|
||||
it('append() to a disk-only session adopts it and repairs a crash tail', async () => {
|
||||
// Persist a session, then corrupt its tail, all through ONE backend.
|
||||
const m = meta('disk-append', '/d')
|
||||
await ctx.sessionPersistence.create(m)
|
||||
await ctx.sessionPersistence.append(m.id, oneTurnLog())
|
||||
await writeFile(logPath(root, '/d', m.id), '\n{"partial crash', { flag: 'a' })
|
||||
|
||||
// A FRESH backend with no in-memory state: append directly (no prior load)
|
||||
// → append must adopt from disk, and the adopt's load schedules a repair
|
||||
// that the same append then performs before writing.
|
||||
const ctx2 = new Context()
|
||||
await ctx2.plugin(SessionStore)
|
||||
await ctx2.plugin(SessionPersistenceJsonl, { root })
|
||||
await ctx2.sessionPersistence.append(m.id, [
|
||||
{ type: 'turn/start', seq: 6, time: 9, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } },
|
||||
{ type: 'turn/end', seq: 7, time: 10, data: { turn: 2, reason: { kind: 'completed' } } },
|
||||
] as SessionEvent[])
|
||||
const loaded = await ctx2.sessionPersistence.load(m.id)
|
||||
expect(loaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7])
|
||||
await ctx2.fiber.dispose()
|
||||
})
|
||||
|
||||
it('a header-only log (open turn, no turn/end) preserves the open turn on load and closes it', async () => {
|
||||
// A session whose only durable content is an unclosed first turn. scanLog
|
||||
// preserves the turn/start; loadCore closes it with a synthetic
|
||||
// turn/end {interrupted} so the returned log is balanced.
|
||||
const m = meta('open-turn', '/h')
|
||||
await ctx.sessionPersistence.create(m)
|
||||
await ctx.sessionPersistence.append(m.id, [
|
||||
{ type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } },
|
||||
] as SessionEvent[])
|
||||
const { events } = await ctx.sessionPersistence.load(m.id)
|
||||
expect(events.map(e => e.type)).toEqual(['turn/start', 'turn/end'])
|
||||
const end = events[1]!
|
||||
expect(end.type === 'turn/end' && end.data.reason).toEqual({ kind: 'interrupted' })
|
||||
})
|
||||
|
||||
|
||||
it('createCore rejects an id already on disk under a DIFFERENT cwd bucket', async () => {
|
||||
// Persist the id under cwd A.
|
||||
const a = meta('dup-id', '/projA')
|
||||
await ctx.sessionPersistence.create(a)
|
||||
await ctx.sessionPersistence.append(a.id, oneTurnLog())
|
||||
// A fresh backend creating the SAME id under cwd B must still refuse: load/
|
||||
// has identify by id across all buckets, so a second log would make resume
|
||||
// nondeterministic. create scans every bucket, not just meta.cwd's.
|
||||
const ctx2 = new Context()
|
||||
await ctx2.plugin(SessionStore)
|
||||
await ctx2.plugin(SessionPersistenceJsonl, { root })
|
||||
await expect(ctx2.sessionPersistence.create(meta('dup-id', '/projB')))
|
||||
.rejects.toThrow(/already has a persisted log on disk/)
|
||||
await ctx2.fiber.dispose()
|
||||
})
|
||||
|
||||
it('flush keeps buffered events when the append fails (no silent loss)', async () => {
|
||||
root = await freshRoot()
|
||||
const ctx2 = new Context()
|
||||
await ctx2.plugin(SessionStore)
|
||||
await ctx2.plugin(SessionPersistenceJsonl, { root })
|
||||
const session = ctx2.sessions.create('flush-fail')
|
||||
// A full turn lands in the write-behind buffer.
|
||||
session.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } })
|
||||
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
|
||||
// Make the durable materialize fail on the next flush.
|
||||
const backend = ctx2.sessionPersistence as unknown as { materialize: (...args: unknown[]) => Promise<void> }
|
||||
const origMat = backend.materialize.bind(backend)
|
||||
backend.materialize = () => Promise.reject(new Error('disk full'))
|
||||
await expect(ctx2.parallel('session/flush', session)).rejects.toThrow(/disk full/)
|
||||
// The events are STILL buffered (not silently dropped): a retry persists them.
|
||||
backend.materialize = origMat
|
||||
await ctx2.parallel('session/flush', session)
|
||||
const loaded = await ctx2.sessionPersistence.load(SessionId('flush-fail'))
|
||||
expect(loaded.events.map(e => e.seq)).toEqual([0, 1])
|
||||
await ctx2.fiber.dispose()
|
||||
})
|
||||
|
||||
it('rejects non-JSON event data: BigInt, function, circular, Map, undefined property', async () => {
|
||||
const m = meta('serial')
|
||||
await ctx.sessionPersistence.create(m)
|
||||
const bad = (extra: unknown) => [{ type: 'user/message', seq: 0, time: 1, data: { content: [{ type: 'text', text: 'x' }], source: { kind: 'user' }, extra } }] as unknown as SessionEvent[]
|
||||
await expect(ctx.sessionPersistence.append(m.id, bad(1n))).rejects.toThrow(/non-JSON-serializable/)
|
||||
await expect(ctx.sessionPersistence.append(m.id, bad(() => 0))).rejects.toThrow(/non-JSON-serializable/)
|
||||
await expect(ctx.sessionPersistence.append(m.id, bad(Symbol('s')))).rejects.toThrow(/non-JSON-serializable/)
|
||||
await expect(ctx.sessionPersistence.append(m.id, bad(new Map()))).rejects.toThrow(/non-JSON-serializable/)
|
||||
await expect(ctx.sessionPersistence.append(m.id, bad(undefined))).rejects.toThrow(/non-JSON-serializable/)
|
||||
await expect(ctx.sessionPersistence.append(m.id, bad(Infinity))).rejects.toThrow(/non-JSON-serializable/)
|
||||
// a circular structure
|
||||
const circ: Record<string, unknown> = {}
|
||||
circ.self = circ
|
||||
await expect(ctx.sessionPersistence.append(m.id, bad(circ))).rejects.toThrow(/non-JSON-serializable/)
|
||||
// The session was never materialized by any of the rejected appends.
|
||||
expect(await ctx.sessionPersistence.has(m.id)).toBe(false)
|
||||
})
|
||||
|
||||
it('accepts well-formed JSON values (null, booleans, nested arrays/objects)', async () => {
|
||||
const m = meta('json-ok')
|
||||
await ctx.sessionPersistence.create(m)
|
||||
const ev = [{ type: 'user/message', seq: 0, time: 1, data: { content: [{ type: 'text', text: 'x' }], source: { kind: 'user' }, extra: { a: null, b: true, c: [1, 2, { d: 'nested' }] } } }] as unknown as SessionEvent[]
|
||||
await ctx.sessionPersistence.append(m.id, ev)
|
||||
expect(await ctx.sessionPersistence.has(m.id)).toBe(true)
|
||||
})
|
||||
|
||||
it('Session.append rejects a non-serializable event at the source (never enters the log)', () => {
|
||||
const session = ctx.sessions.create('reject-bad')
|
||||
// Serializability is enforced at the source: Session.append throws on a
|
||||
// BigInt-bearing event BEFORE it enters session.events, so the durable log
|
||||
// can never diverge from the live log. The throw surfaces at the caller's
|
||||
// append site, not asynchronously in a backend flush.
|
||||
expect(() => {
|
||||
session.append('user/message', { content: [{ type: 'text', text: 'bad' }], source: { kind: 'user' }, bad: 1n } as never)
|
||||
}).toThrow(/non-JSON-serializable/)
|
||||
// The bad event was rejected, so the log stayed empty.
|
||||
expect(session.events.length).toBe(0)
|
||||
})
|
||||
|
||||
})
|
||||
@@ -0,0 +1,27 @@
|
||||
{
|
||||
"extends": "../../../tsconfig.base.json",
|
||||
"compilerOptions": {
|
||||
"rootDir": "src",
|
||||
"outDir": "lib"
|
||||
},
|
||||
"include": [
|
||||
"src"
|
||||
],
|
||||
"references": [
|
||||
{
|
||||
"path": "../../../vendor/cosmokit"
|
||||
},
|
||||
{
|
||||
"path": "../../../vendor/cordis"
|
||||
},
|
||||
{
|
||||
"path": "../../../vendor/schemastery"
|
||||
},
|
||||
{
|
||||
"path": "../../core/session"
|
||||
},
|
||||
{
|
||||
"path": "../../session-persistence/session-persistence"
|
||||
}
|
||||
]
|
||||
}
|
||||
@@ -0,0 +1,29 @@
|
||||
# @deepseek-ai/dsh-session-persistence-sqlite
|
||||
|
||||
A SQLite durable session-persistence backend — a second `SessionPersistence` implementation ([session persistence](../../docs/rfc/implemented/2026-06-14-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 (`SessionHeader`) lives in a `sessions` row. 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 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).
|
||||
|
||||
## 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 (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 `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.
|
||||
@@ -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"
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,258 @@
|
||||
/**
|
||||
* 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)`.
|
||||
*
|
||||
* 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 six public
|
||||
* {@link SessionPersistence} methods delegate to the coordinator.
|
||||
*
|
||||
* @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, PersistenceCoordinator,
|
||||
type PersistenceBackend, type StoredPrefix,
|
||||
} from '@deepseek-ai/dsh-session-persistence'
|
||||
import type { Session, SessionEvent, SessionId, SessionHeader } 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
|
||||
}
|
||||
|
||||
/**
|
||||
* The SQLite persistence backend. Load as a plugin; it registers as
|
||||
* `ctx.sessionPersistence` and (via the coordinator) installs the write-path
|
||||
* listeners. Its torn-tail marker is the seq to delete from.
|
||||
*/
|
||||
export class SessionPersistenceSqlite extends SessionPersistence implements PersistenceBackend<number> {
|
||||
static inject = ['sessions']
|
||||
|
||||
static Config: z<Config> = z.object({
|
||||
path: z.string().required(),
|
||||
})
|
||||
|
||||
/**
|
||||
* Backend label for the coordinator's dispose diagnostics. Intentionally
|
||||
* shadows cordis `Service.name` (set to `'sessionPersistence'` by the base);
|
||||
* see the JSONL backend for why this does not affect service resolution.
|
||||
*/
|
||||
override readonly name = 'session-persistence-sqlite'
|
||||
|
||||
private db!: DatabaseSync
|
||||
private ready: Promise<void>
|
||||
private coordinator: PersistenceCoordinator<number>
|
||||
|
||||
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.
|
||||
this.ready = this.openDb(config.path)
|
||||
this.coordinator = new PersistenceCoordinator<number>(this.ctx, this)
|
||||
}
|
||||
|
||||
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 service surface (delegated to the coordinator) ---
|
||||
|
||||
create(meta: SessionHeader): Promise<void> {
|
||||
return this.coordinator.create(meta)
|
||||
}
|
||||
|
||||
append(id: SessionId, events: readonly SessionEvent[]): Promise<void> {
|
||||
return this.coordinator.append(id, events)
|
||||
}
|
||||
|
||||
load(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }> {
|
||||
return this.coordinator.load(id)
|
||||
}
|
||||
|
||||
has(id: SessionId): Promise<boolean> {
|
||||
return this.coordinator.has(id)
|
||||
}
|
||||
|
||||
delete(id: SessionId): Promise<void> {
|
||||
return this.coordinator.delete(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
|
||||
}
|
||||
|
||||
// --- PersistenceBackend hooks (the SQLite storage primitives) ---
|
||||
|
||||
/** Read a stored prefix by id (ids are globally unique — no scope to scan). */
|
||||
loadStored(id: SessionId): Promise<StoredPrefix<number> | undefined> {
|
||||
return this.readPrefix(id)
|
||||
}
|
||||
|
||||
/** Read a stored prefix; `cwd` is ignored (the id is globally unique in SQLite). */
|
||||
loadLive(id: SessionId, _cwd: string | undefined): Promise<StoredPrefix<number> | undefined> {
|
||||
return this.readPrefix(id)
|
||||
}
|
||||
|
||||
/**
|
||||
* Read a session's row + ordered events into a {@link StoredPrefix}. The
|
||||
* torn-tail marker is the seq from which a never-committed tail must be deleted
|
||||
* (`scanRows` already returns it as `number | undefined`).
|
||||
*/
|
||||
private async readPrefix(id: SessionId): Promise<StoredPrefix<number> | undefined> {
|
||||
await this.ready
|
||||
const row = this.rowFor(id)
|
||||
if (row === undefined) return undefined
|
||||
const meta = rowToMeta(row)
|
||||
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)
|
||||
return { meta, events: preserved, ...tornFrom !== undefined ? { tornMarker: tornFrom } : {} }
|
||||
}
|
||||
|
||||
/**
|
||||
* Durably append a batch in ONE transaction: materialize the sessions row (if
|
||||
* lazy) and INSERT every event, or roll back entirely. The transaction is the
|
||||
* atomicity + durability boundary, so a mid-batch failure (a UNIQUE violation
|
||||
* on a duplicated seq) leaves the stored log untouched.
|
||||
*/
|
||||
async appendBatch(meta: SessionHeader, events: readonly SessionEvent[], isMaterialized: boolean): Promise<void> {
|
||||
await this.ready
|
||||
const insertEvent = this.db.prepare(
|
||||
'INSERT INTO events (session_id, seq, type, time, data) VALUES (?, ?, ?, ?, ?)',
|
||||
)
|
||||
this.db.exec('BEGIN')
|
||||
try {
|
||||
if (!isMaterialized) this.writeRow(meta)
|
||||
for (const event of events) {
|
||||
insertEvent.run(meta.id, event.seq, event.type, event.time, JSON.stringify(event.data))
|
||||
}
|
||||
this.db.exec('COMMIT')
|
||||
} catch (error) {
|
||||
this.db.exec('ROLLBACK')
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Make a crash repair durable in ONE transaction: DELETE the torn tail (from
|
||||
* `tornMarker`) and INSERT the synthetic `closers`. After COMMIT the stored rows
|
||||
* == the balanced log.
|
||||
*/
|
||||
async commitRepair(meta: SessionHeader, tornMarker: number | undefined, closers: readonly SessionEvent[]): Promise<void> {
|
||||
await this.ready
|
||||
this.db.exec('BEGIN')
|
||||
try {
|
||||
if (tornMarker !== undefined) {
|
||||
this.db.prepare('DELETE FROM events WHERE session_id = ? AND seq >= ?').run(meta.id, tornMarker)
|
||||
}
|
||||
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(meta.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 */
|
||||
}
|
||||
}
|
||||
|
||||
/** Remove a session's row (ON DELETE CASCADE drops its events). */
|
||||
async deleteStored(id: SessionId): Promise<void> {
|
||||
await this.ready
|
||||
this.db.prepare('DELETE FROM sessions WHERE id = ?').run(id)
|
||||
}
|
||||
|
||||
/** List all materialized sessions' metadata (every row is a materialized session). */
|
||||
async list(): Promise<SessionHeader[]> {
|
||||
await this.ready
|
||||
const rows = this.db
|
||||
.prepare('SELECT * FROM sessions')
|
||||
.all() as unknown as SessionRow[]
|
||||
return rows.map(rowToMeta)
|
||||
}
|
||||
|
||||
/** Close the database handle (awaited by the coordinator's dispose, post-drain). */
|
||||
async close(): Promise<void> {
|
||||
await this.ready
|
||||
this.db.close()
|
||||
}
|
||||
|
||||
// --- row helpers ---
|
||||
|
||||
/** Fetch a session's row, or undefined if absent. */
|
||||
private rowFor(id: SessionId): SessionRow | undefined {
|
||||
return this.db.prepare('SELECT * FROM sessions WHERE id = ?').get(id) as unknown as SessionRow | undefined
|
||||
}
|
||||
|
||||
/**
|
||||
* Insert-or-replace a session's metadata row. The only caller is the first
|
||||
* materializing `appendBatch`, so writing the row IS the materialization (its
|
||||
* existence is the signal `has`/`list` read).
|
||||
*/
|
||||
private writeRow(meta: SessionHeader): void {
|
||||
this.db.prepare(`
|
||||
INSERT INTO sessions (id, version, created_at, cwd, parent_session)
|
||||
VALUES (?, ?, ?, ?, ?)
|
||||
ON CONFLICT(id) DO UPDATE SET
|
||||
version = excluded.version,
|
||||
created_at = excluded.created_at,
|
||||
cwd = excluded.cwd,
|
||||
parent_session = excluded.parent_session
|
||||
`).run(
|
||||
meta.id,
|
||||
meta.version,
|
||||
meta.createdAt,
|
||||
meta.cwd ?? null,
|
||||
meta.parentSession ?? null,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
export default SessionPersistenceSqlite
|
||||
@@ -0,0 +1,182 @@
|
||||
/**
|
||||
* 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, SessionHeader } 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 = 2
|
||||
|
||||
/**
|
||||
* A row of the `sessions` table — the out-of-log metadata ({@link SessionHeader}).
|
||||
* 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
|
||||
}
|
||||
|
||||
/** 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 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: v1 had a different `sessions` layout and is not
|
||||
* upgraded in place.
|
||||
*/
|
||||
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 !== 0 && onDisk !== SCHEMA_VERSION) {
|
||||
db.close()
|
||||
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).
|
||||
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
|
||||
) 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 {@link SessionHeader} from a `sessions` row. */
|
||||
export function rowToMeta(row: SessionRow): SessionHeader {
|
||||
return {
|
||||
version: row.version,
|
||||
id: row.id as SessionId,
|
||||
createdAt: row.created_at,
|
||||
...row.cwd !== null ? { cwd: row.cwd } : {},
|
||||
...row.parent_session !== null ? { parentSession: row.parent_session as SessionId } : {},
|
||||
}
|
||||
}
|
||||
|
||||
/** 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 (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.
|
||||
*/
|
||||
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 }
|
||||
}
|
||||
@@ -0,0 +1,375 @@
|
||||
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 from '@deepseek-ai/dsh-session'
|
||||
import type { Session, SessionEvent } 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'
|
||||
import { runCoordinatorContract, type CoordinatorFixture } from '../../session-persistence/tests/coordinator-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() },
|
||||
}
|
||||
})
|
||||
|
||||
// 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.
|
||||
runCoordinatorContract('sqlite', async (): Promise<CoordinatorFixture> => {
|
||||
const dir = await mkdtemp(join(tmpdir(), 'dsh-sqlite-coord-'))
|
||||
const path = join(dir, 'sessions.db')
|
||||
return {
|
||||
mount: async ctx => ctx.plugin(SessionPersistenceSqlite, { path }),
|
||||
corruptTail: async (id) => {
|
||||
// A row past the committed region whose `data` does not parse: scanRows
|
||||
// bounds the preserved prefix at it and returns its seq as tornFrom, which
|
||||
// the backend surfaces to the coordinator as the tornMarker to delete from.
|
||||
const db = openDatabase(path)
|
||||
const next = (db.prepare('SELECT COALESCE(MAX(seq), -1) + 1 AS n FROM events WHERE session_id = ?')
|
||||
.get(id) as { n: number }).n
|
||||
db.prepare('INSERT INTO events (session_id, seq, type, time, data) VALUES (?, ?, ?, ?, ?)')
|
||||
.run(id, next, 'assistant/chunk', 99, '{not valid json')
|
||||
db.close()
|
||||
},
|
||||
cleanup: async () => { await rm(dir, { recursive: true, force: true }) },
|
||||
}
|
||||
})
|
||||
|
||||
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 not the current build (newer OR older)', async () => {
|
||||
const path = await freshDbPath()
|
||||
openDatabase(path).close() // stamp user_version = SCHEMA_VERSION
|
||||
// Bump user_version past what this build supports.
|
||||
const dbNewer = openDatabase(path)
|
||||
dbNewer.exec(`PRAGMA user_version = ${SCHEMA_VERSION + 1}`)
|
||||
dbNewer.close()
|
||||
expect(() => openDatabase(path)).toThrow(/incompatible with this build/)
|
||||
|
||||
// A stale OLDER version (e.g. a pre-summary-drop v1 DB) is also rejected —
|
||||
// we do not migrate (unreleased software, no backward-compat).
|
||||
const olderPath = await freshDbPath()
|
||||
openDatabase(olderPath).close()
|
||||
const dbOlder = openDatabase(olderPath)
|
||||
dbOlder.exec('PRAGMA user_version = 1')
|
||||
dbOlder.close()
|
||||
expect(() => openDatabase(olderPath)).toThrow(/incompatible with this build/)
|
||||
})
|
||||
|
||||
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 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' })
|
||||
expect(loaded.events).toEqual(oneTurnLog())
|
||||
await fiber2.dispose()
|
||||
})
|
||||
|
||||
it('exposes the schema version constant', () => {
|
||||
expect(SCHEMA_VERSION).toBe(2)
|
||||
})
|
||||
})
|
||||
|
||||
describe('SessionPersistenceSqlite: edge cases', () => {
|
||||
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('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()
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,27 @@
|
||||
{
|
||||
"extends": "../../../tsconfig.base.json",
|
||||
"compilerOptions": {
|
||||
"rootDir": "src",
|
||||
"outDir": "lib"
|
||||
},
|
||||
"include": [
|
||||
"src"
|
||||
],
|
||||
"references": [
|
||||
{
|
||||
"path": "../../../vendor/cosmokit"
|
||||
},
|
||||
{
|
||||
"path": "../../../vendor/cordis"
|
||||
},
|
||||
{
|
||||
"path": "../../../vendor/schemastery"
|
||||
},
|
||||
{
|
||||
"path": "../../core/session"
|
||||
},
|
||||
{
|
||||
"path": "../../session-persistence/session-persistence"
|
||||
}
|
||||
]
|
||||
}
|
||||
@@ -0,0 +1,21 @@
|
||||
{
|
||||
"extends": "../../../tsconfig.base.json",
|
||||
"compilerOptions": {
|
||||
"rootDir": "src",
|
||||
"outDir": "lib"
|
||||
},
|
||||
"include": [
|
||||
"src"
|
||||
],
|
||||
"references": [
|
||||
{
|
||||
"path": "../../../vendor/cosmokit"
|
||||
},
|
||||
{
|
||||
"path": "../../../vendor/cordis"
|
||||
},
|
||||
{
|
||||
"path": "../../core/session"
|
||||
}
|
||||
]
|
||||
}
|
||||
@@ -1,13 +0,0 @@
|
||||
{
|
||||
"extends": "../../tsconfig.base.json",
|
||||
"compilerOptions": {
|
||||
"rootDir": "src",
|
||||
"outDir": "lib"
|
||||
},
|
||||
"include": ["src"],
|
||||
"references": [
|
||||
{ "path": "../../vendor/cosmokit" },
|
||||
{ "path": "../../vendor/cordis" },
|
||||
{ "path": "../session" }
|
||||
]
|
||||
}
|
||||
Reference in New Issue
Block a user