Coder/deepseek-harness
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/master' into codex/simp-session-dead-surface

Browse Source 
# Conflicts:
#	docs/cordis-catalog/services.md
#	packages/core/session/README.md
#	packages/core/session/tests/derived-cache.spec.ts
#	packages/session-persistence/session-persistence-jsonl/tests/jsonl.spec.ts
#	packages/session-persistence/session-persistence/src/coordinator.ts
#	packages/session-persistence/session-persistence/src/index.ts
This commit is contained in:
Tianyi Cui
2026-07-14 17:24:02 +08:00
parent 60260b5d5b fb126e164a
commit 34a059c9bc
562 changed files with 4437 additions and 12564 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
packages/session-persistence/session-persistence-jsonl/README.md
View File

@@ -23,12 +23,12 @@ The JSONL durable session-persistence backend — a concrete `SessionPersistence
- **Lazy materialization.** `create(meta)` writes nothing; on the first `append`, the backend writes and `fsync`s a temporary file, publishes it without overwrite via a hard link, then `fsync`s the directory. A created-but-never-appended session leaves nothing on disk and is absent from `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/architecture/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.
- **Crash recovery — preserve valid tail work.** `load` keeps the contiguous valid prefix of an interrupted final turn. It truncates from the first unparsable or sequence-gapped uncommitted record, then appends the synthetic tool, step, and turn closers required by the shared [persistence contract](../../../docs/rfc/implemented/architecture/2026-06-14-session-persistence.md); the same defect at or before the last committed `turn/end` rejects.
- **Contiguous-seq.** `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.
## 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` (copy each already-frozen event into the persistence-owned write-behind buffer), and `session/flush`/dispose (drain that 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.
The plugin buffers frozen session events and drains them on flush or disposal. A per-session cursor prevents resumed sessions from re-appending stored events, and live sessions are seeded when the plugin loads. Operations for one session are serialized; disposal waits for initialization and the final drain so no write lands after teardown.
## Model Experience

72
packages/session-persistence/session-persistence-jsonl/src/format.ts
View File

@@ -72,19 +72,13 @@ function isHeaderLine(value: unknown): value is HeaderLine {
}
/**
* 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.
* 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.
* Safe code units remain literal; every other unit, including `~`, becomes
* `~XXXX`. Operating on code units preserves lone surrogates, while special-
* casing `.` and `..` prevents traversal by an otherwise safe whole segment.
*
* 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.
* @param raw - the string to encode; must be non-empty (throws on `''`).
* @returns the escaped single path segment, decodable back to `raw`.
*/
@@ -142,38 +136,18 @@ export function eventLine(event: SessionEvent): string {
/**
* 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`).
* 0). Fully written events in an interrupted final turn remain part of the
* prefix. The first unparsable record or seq gap after the last `turn/end`
* marks a tolerated torn tail; the same hole in the committed region rejects.
*
* A crash can leave a durable log whose final turn never closed: real,
* fully-written 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.
* @param buffer - the raw bytes of the log file (header line first).
* @returns the header, the preserved event prefix, and `committedBytes` — the
* byte offset the next append truncates any torn tail to.
*/
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).
// Track complete lines by byte offset: a non-newline tail is torn and ignored,
// and a running counter avoids rescanning a long multi-byte log.
const lines: { text: string; endByte: number }[] = []
let start = 0
let byteOffset = 0
@@ -201,14 +175,8 @@ export function scanLog(buffer: Buffer): { meta: SessionHeader; events: SessionE
}
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.
// Parse every complete record first so the last valid `turn/end` determines
// whether an earlier hole is committed corruption or an uncommitted tail.
interface Parsed { ok: boolean; event?: SessionEvent; endByte: number }
const parsed: Parsed[] = eventEntries.map((entry) => {
try {
@@ -226,18 +194,8 @@ export function scanLog(buffer: Buffer): { meta: SessionHeader; events: SessionE
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.
// Preserve the contiguous prefix, including a complete interrupted turn;
// holes through the last committed boundary throw, while later holes stop.
const preserved: SessionEvent[] = []
for (let i = 0; i < parsed.length; i++) {
const p = parsed[i]

110
packages/session-persistence/session-persistence-jsonl/src/index.ts
View File

@@ -1,19 +1,7 @@
/**
* 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 four public
* {@link SessionPersistence} methods delegate to the coordinator.
*
* JSONL durable session-persistence backend. It stores a header and contiguous
* events in one append-only file per session, and delegates orchestration to
* {@link PersistenceCoordinator}.
* @module @deepseek-ai/dsh-session-persistence-jsonl
*/
@@ -41,13 +29,7 @@ export interface Config {
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`.)
*/
/** Whether a filesystem error means absence; every non-ENOENT failure must surface. */
function isENOENT(error: unknown): boolean {
return (error as NodeJS.ErrnoException | null)?.code === 'ENOENT'
}
@@ -65,13 +47,9 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
})
/**
* 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.
* Backend label for coordinator diagnostics and effects. It shadows
* `Service.name` without changing the service key captured by the base
* constructor.
*/
override readonly name = 'session-persistence-jsonl'
@@ -80,10 +58,7 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
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.
// Resolve once so later process.cwd() changes cannot split one backend across roots.
this.root = resolve(config.root)
this.coordinator = new PersistenceCoordinator<number>(this.ctx, this)
}
@@ -105,16 +80,13 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
return this.coordinator.load(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.
// One method serves both public `list` and the backend hook; delegating it to
// the coordinator would call this hook recursively.
/* jscpd:ignore-end */
// --- PersistenceBackend hooks (the file-bytes storage primitives) ---
/** Read a stored prefix by id across ALL cwd buckets (cwd unknown). */
/** Read a stored prefix by id across all cwd buckets when cwd is unknown. */
async loadStored(id: SessionId): Promise<StoredPrefix<number> | undefined> {
const file = await this.findLog(id)
if (file === undefined) return undefined
@@ -122,11 +94,8 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
}
/**
* 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.
* Read a stored prefix within one cwd for HMR adoption. `undefined` names the
* no-cwd bucket rather than an unknown cwd, so this never scans other buckets.
*/
async loadLive(id: SessionId, cwd: string | undefined): Promise<StoredPrefix<number> | undefined> {
const path = logPath(this.root, cwd, id)
@@ -135,10 +104,8 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
}
/**
* 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.
* Read a stored prefix and convert torn-tail state to the byte offset the
* coordinator can round-trip without knowing the file format.
*/
private async readPrefix(path: string): Promise<StoredPrefix<number>> {
const buffer = await readFile(path)
@@ -174,9 +141,7 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
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).
// Read only headers so listing scales with session count, not log size.
const first = await this.readFirstLine(`${dir}/${name}`)
if (first === undefined) continue // empty/half-written file
const meta = parseHeaderMeta(first)
@@ -197,10 +162,7 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
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.)
// Materialization is the first write; an existing log is an id collision.
/* 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)`)
@@ -217,28 +179,22 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
} 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.
// Publish with link()+unlink(): unlike rename(), link fails if another
// process materialized the same id first.
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.
// Remove an unpublished temp on failure. After publication, defer cleanup
// until the directory entry is durable so cleanup cannot reject a live log.
/* 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.
// The published link becomes crash-durable only after its directory fsync.
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.
// 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.
try {
await rm(tmp, { force: true })
} catch {
@@ -257,11 +213,9 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
}
/**
* 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.
* Append and fsync event lines. On a partial write or sync failure, restore the
* previous size before rethrowing because the unchanged cursor will retry the
* batch; leaving partial bytes would create duplicate sequence numbers.
*/
private async appendLines(meta: SessionHeader, events: readonly SessionEvent[]): Promise<void> {
const path = logPath(this.root, meta.cwd, meta.id)
@@ -323,10 +277,8 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
}
/**
* Find a session's log file by id across ALL cwd buckets — the any-cwd scan
* for `loadStored` (resume identifies 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.
* Find a session by id across cwd buckets for resume. Cwd-scoped HMR adoption
* bypasses this scan so a no-cwd session cannot claim another bucket.
*/
private async findLog(id: SessionId): Promise<{ path: string; cwd: string | undefined } | undefined> {
const target = encodeSegment(id) + '.jsonl'
@@ -347,9 +299,7 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
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.
// Only an absent root means no sessions; rethrow every other I/O failure.
if (isENOENT(error)) return []
throw error
}

55
packages/session-persistence/session-persistence-jsonl/tests/jsonl.spec.ts
View File

@@ -54,12 +54,8 @@ runPersistenceContract('jsonl', async () => {
}
})
// 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.
// Two mounts share this temp root to exercise reload. `corruptTail` appends a partial,
// newline-less fragment past the committed region so coordinator repair runs on real file bytes.
runCoordinatorContract('jsonl', async (): Promise<CoordinatorFixture> => {
const dir = await mkdtemp(join(tmpdir(), 'dsh-jsonl-coord-'))
return {
@@ -347,10 +343,9 @@ describe('SessionPersistenceJsonl: scanLog unit', () => {
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.
// 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. `loadCore`, not this scanner, later closes the orphaned turn.
expect(scanLog(Buffer.from(log)).events.map(e => e.seq)).toEqual([0])
})
@@ -470,9 +465,8 @@ describe('SessionPersistenceJsonl: edge cases', () => {
})
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.
// A tolerated extra field makes this valid header exceed the 8192-byte read buffer, proving
// `readFirstLine` accumulates chunks before `list()` parses it.
const bucket = join(root, '_no-cwd')
await mkdir(bucket, { recursive: true })
const bigHeader = JSON.stringify({ type: 'session', version: 0, id: 'big', createdAt: 1, pad: 'x'.repeat(9000) })
@@ -492,10 +486,8 @@ describe('SessionPersistenceJsonl: edge cases', () => {
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.
// A new Session object reuses the id. Object-keyed initialization must run independently,
// detect the disk collision, and reject instead of appending through session A's stale cursor.
let b!: Session
await ctx.plugin(Object.assign((inner: Context) => {
b = inner.sessions.create(SessionId('reuse'), { meta: { cwd: '/a' } })
@@ -512,13 +504,9 @@ describe('SessionPersistenceJsonl: edge cases', () => {
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.
// Backend 2 creates a no-cwd session whose id exists only in `/w`. Exact `loadLive(id,
// undefined)` must not adopt across buckets; the any-cwd collision check then rejects instead
// of grafting no-cwd events onto a log with mismatched cwd.
const ctx2 = new Context()
await ctx2.plugin(SessionStore)
await ctx2.plugin(SessionPersistenceJsonl, { root })
@@ -583,9 +571,8 @@ describe('SessionPersistenceJsonl: edge cases', () => {
})
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.
// A durable backend must not collapse a storage fault to "no sessions". Making the root a
// regular file forces ENOTDIR from `readdir`, which must propagate.
const filePath = join(root, 'not-a-dir')
await writeFile(filePath, 'x')
const ctx2 = new Context()
@@ -596,11 +583,8 @@ describe('SessionPersistenceJsonl: edge cases', () => {
})
it('loadLive surfaces a non-ENOENT lookup error (ENOTDIR) instead of reporting absent', async () => {
// A non-ENOENT error from the per-id open() must surface, not be collapsed to
// "not found" (which would let live-adoption proceed under a false absence
// assumption). A live session's onCreated 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.
// A non-ENOENT per-id open error must surface rather than become "not found" and permit false
// live adoption. Making the cwd bucket a regular file forces ENOTDIR for its child log path.
const cwd = '/x'
const ctx2 = new Context()
await ctx2.plugin(SessionStore)
@@ -718,10 +702,9 @@ describe('SessionPersistenceJsonl: edge cases', () => {
it('Session.append rejects a non-serializable event at the source (never enters the log)', () => {
const session = ctx.sessions.create(SessionId('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.
// 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 error therefore surfaces synchronously at append, not later during backend flush.
expect(() => {
session.append('user/message', { content: [{ type: 'text', text: 'bad' }], source: { kind: 'user' }, bad: 1n } as never, { surfaceOp: 'append' })
}).toThrow(/non-JSON-serializable/)

5
packages/session-persistence/session-persistence-sqlite/README.md
View File

@@ -8,13 +8,13 @@ A SQLite durable session-persistence backend — a second `SessionPersistence` i
Each `SessionEvent` maps 1:1 onto a row in an `events` table `(session_id, seq, type, time, data, source_event_seqs, surface_op)` — `data` is the event payload as JSON text, so the row shape is the event verbatim (including `assistant/chunk`, keeping `seq` contiguous). The two `TEXT` columns `source_event_seqs` and `surface_op` are nullable; they store the event's optional surface-metadata fields (see [session surface](../../../docs/rfc/implemented/architecture/2026-06-18-session-surface.md)). Out-of-log metadata (`SessionHeader`) lives in a `sessions` row. A `sessions` row is written only by the first `append` — its existence is the lazy-materialization signal (`list` reports exactly the sessions that have a row), so no separate column is needed.
The repo's `engines.node` is `^22.19.0 || >=24.0.0` (Node 22.19+ or 24+), matching the LTS floor required by the installed Pi adapter dependency; `node:sqlite` itself ships without the `--experimental-sqlite` flag from Node 22.13 (LTS) and 23.4 / 24 (Current) on. The range deliberately excludes Node 23 because that line is non-LTS/EOL and still has flagged runtime features before 23.6. The database opens with `foreign_keys = ON` (so `ON DELETE CASCADE` drops a session's events with its row) and the configured `journal_mode` (default `wal`; pick a rollback-journal mode like `delete` on filesystems where WAL's shared-memory files do not work, e.g. network mounts). The table-layout version is stored in `PRAGMA user_version` and checked on open: a fresh database is stamped with the current `SCHEMA_VERSION`; a database written by any other, incompatible build (a non-current `user_version`, older or newer) is rejected rather than opened against an unknown layout — there is no migration (unreleased software).
The repository's Node range supports unflagged `node:sqlite`. The database enables foreign keys and uses the configured journal mode (`wal` by default; use a rollback mode where WAL shared-memory files are unsuitable). `PRAGMA user_version` stores the table-layout version; databases with any other version are rejected because this unreleased format has no migrations.
## Contract semantics over rows
- **Append = a transaction.** `append` runs `BEGIN`/`COMMIT` around the batch: it materializes the `sessions` row (if still lazy) and INSERTs every event, asserting the contiguous-seq contract first (the first event's `seq` must equal the stored next-seq). A mid-batch failure (a UNIQUE violation on a duplicated seq) rolls back entirely, so the stored log and the in-memory cursor stay consistent. (`load()` already balanced the stored log, so `append` never has to repair a crash tail.)
- **Lazy materialization.** `create()` records intent in memory only — no row is written until the first `append`. A created-but-never-appended session has no `sessions` row, so it is absent from `list()` (which reports exactly the sessions that have a row).
- **Interrupted-turn close on load.** `load()` reads every stored event ordered by `seq` and finds the longest seq-contiguous, parseable prefix — INCLUDING the real events of an interrupted final turn after the last `turn/end` (the loop only flushes at `turn/end`, so a process killed mid-turn leaves real, fully-written rows past it). A single turn can be huge in a long-horizon task, so those events are **preserved, never truncated**: `load()` CLOSES the orphaned turn by durably appending the minimal synthetic boundary events (an error `tool/result` for every assistant tool call left unanswered, a `step/end` if a step was open, then a `turn/end` carrying `{ kind: 'interrupted' }`), inside one transaction that also DELETEs any never-fully-written torn tail row. `load()` is therefore mutating — after it the stored rows are balanced and the cursor is truthful, so the next `append` continues cleanly. The boundary (last `turn/end`, torn-tail detection) is computed from the `seq`/`type` columns so a malformed `data` in a torn tail row is never parsed (discarded, not unloadable). A parse error or `seq` gap inside the committed region (at or before the last real `turn/end`) makes the session unloadable. A session whose only turn never closed keeps its metadata row and stays present in `list()` — the same as the JSONL backend, whose file likewise survives a first append that never reached `turn/end`.
- **Interrupted-turn close on load.** `load()` implements the shared [crash-recovery contract](../../../docs/rfc/implemented/architecture/2026-06-14-session-persistence.md): preserve the valid interrupted turn, append its synthetic closing events in one transaction, and remove only a torn tail row. Committed parse errors or sequence gaps make the session unloadable. Because recovery mutates stored rows, the next append starts from a balanced log and accurate cursor.
## Configuration (schemastery)
@@ -39,7 +39,6 @@ Like the JSONL backend, the plugin also installs the `session/event` → buffer
## Known Limitations and Deferred Work
- **Raw `node:sqlite`, pending a cordis database service** — the backend holds a `DatabaseSync` directly; if a `cordis/db` / `@cordisjs` SQL driver is adopted, the storage driver routes through it (the `SessionPersistence` contract would not change) — a marked TODO.
- **`DatabaseSync` is synchronous** — every append transaction blocks the event loop for its duration; acceptable for local stores, a throughput ceiling for busy multi-session servers.
- **Write contention has no wait or retry policy** — the backend sets no busy timeout and retries no locked-database error, so another connection holding a write transaction makes the operation reject immediately.
- **Only the current `SCHEMA_VERSION` opens** — a database with any other schema version is rejected rather than migrated (unreleased software; no persisted user data to preserve).

30
packages/session-persistence/session-persistence-sqlite/src/index.ts
View File

@@ -1,19 +1,7 @@
/**
* SQLite durable session-persistence backend (`@deepseek-ai/dsh-session-persistence-sqlite`).
*
* A SECOND {@link SessionPersistence} implementation, built to validate that the
* abstract seam + the shared `runPersistenceContract` suite are genuinely
* backend-agnostic: the same append-only / contiguous-seq / lazy-materialization
* / interrupted-turn-close-on-load semantics the JSONL backend expresses over
* file bytes, expressed here over `node:sqlite` rows. Each `SessionEvent` maps
* 1:1 onto a row `(session_id, seq, type, time, data, source_event_seqs, surface_op)`.
*
* Like the JSONL backend it supplies ONLY the storage primitives (the
* {@link PersistenceBackend} hooks below — INSERT/DELETE/SELECT inside
* transactions); all the write-path orchestration lives in the backend-agnostic
* {@link PersistenceCoordinator} this class composes. The four public
* {@link SessionPersistence} methods delegate to the coordinator.
*
* SQLite durable session-persistence backend. It maps each session header and
* event to rows, and delegates write-path orchestration to
* {@link PersistenceCoordinator}.
* @module @deepseek-ai/dsh-session-persistence-sqlite
*/
@@ -89,10 +77,8 @@ export class SessionPersistenceSqlite extends SessionPersistence implements Pers
constructor(ctx: Context, public config: Config) {
super(ctx)
// Open the database asynchronously (the parent directory may need creating);
// every hook awaits `ready` first. Opening synchronously would force a sync
// mkdir and block plugin apply. schemastery (static Config) has already
// filled `journalMode`; the cast records that runtime fact.
// Open asynchronously so directory creation does not block plugin apply;
// every storage hook awaits the same readiness promise.
this.ready = this.openDb(config.path, (config as Required<Config>).journalMode)
this.coordinator = new PersistenceCoordinator<number>(this.ctx, this)
}
@@ -121,10 +107,8 @@ export class SessionPersistenceSqlite extends SessionPersistence implements Pers
return this.coordinator.load(id)
}
// `list` is BOTH the public service method and the PersistenceBackend hook —
// one method (the SELECT below). The coordinator adds no orchestration for
// listing, so routing it through the coordinator would just recurse. Defined
// once, in the "PersistenceBackend hooks" section.
// One method serves both public `list` and the backend hook; delegating it to
// the coordinator would call this hook recursively.
// --- PersistenceBackend hooks (the SQLite storage primitives) ---

65
packages/session-persistence/session-persistence-sqlite/src/schema.ts
View File

@@ -56,35 +56,17 @@ export interface EventRow {
export type JournalMode = 'wal' | 'delete' | 'truncate' | 'persist'
/**
* Open the database at `path` and apply the schema + pragmas. `foreign_keys`
* makes `ON DELETE CASCADE` drop a session's events with its row; the
* `journal_mode` pragma is set from the plugin's `journalMode` config (`wal`
* default — the durability model the ADR records; the row shape maps 1:1
* onto `SessionEvent`; opencode runs this exact shape on SQLite/WAL).
*
* The table-layout version is persisted in SQLite's `PRAGMA user_version` and
* checked on open: a fresh database (user_version 0) is stamped with the
* current {@link SCHEMA_VERSION}; an existing database whose version is NOT the
* current one (written by a different, incompatible build — older or newer) is
* REJECTED rather than opened against a layout this build does not understand.
* There are no migrations: an earlier layout is not upgraded in place — it is
* rejected. v1 had a different `sessions` shape; v2 lacked all of
* `seed_length`/`source_event_seqs`/`surface_op`. v3 is SKIPPED: two unmerged
* branches each shipped a DISTINCT v3 (one adding only `seed_length`, the other
* adding only the surface columns), so an on-disk v3 is ambiguous — it could be
* either sibling layout, neither of which has all of this build's columns. v4
* is the merged layout carrying every column; bumping past the collided v3
* makes the version check reject both sibling v3 databases instead of opening
* one against columns it does not have.
* Open the database and apply its schema and pragmas. A zero `user_version` is
* stamped with {@link SCHEMA_VERSION}; every other non-current version rejects
* rather than being migrated in place.
* @param path - the SQLite database file to open (created when absent).
* @param journalMode - the journal pragma to apply — a closed in-code union, validated by the plugin Config.
* @param journalMode - validated journal pragma.
* @returns the open handle with pragmas applied and both tables ensured.
*/
export function openDatabase(path: string, journalMode: JournalMode): DatabaseSync {
const db = new DatabaseSync(path)
db.exec('PRAGMA foreign_keys = ON')
// journalMode is a closed in-code union (validated by the plugin Config), not
// user-controlled SQL — safe to interpolate (PRAGMA takes no bound params).
// The validated union is safe to interpolate into a non-bindable PRAGMA.
db.exec(`PRAGMA journal_mode = ${journalMode.toUpperCase()}`)
// `PRAGMA user_version` always returns exactly one row { user_version }.
const { user_version: onDisk } = db.prepare('PRAGMA user_version').get() as { user_version: number }
@@ -93,9 +75,7 @@ export function openDatabase(path: string, journalMode: JournalMode): DatabaseSy
throw new Error(`session database at "${path}" has schema version ${onDisk}, incompatible with this build (${SCHEMA_VERSION})`)
}
if (onDisk === 0) {
// Fresh (or pre-versioning) database: stamp the current layout version.
// PRAGMA does not accept bound parameters, so interpolate the integer
// constant (SCHEMA_VERSION is a trusted in-code number, not user input).
// Stamp fresh or pre-versioning databases.
db.exec(`PRAGMA user_version = ${SCHEMA_VERSION}`)
}
db.exec(`
@@ -162,28 +142,11 @@ export function rowToEvent(row: EventRow): SessionEvent {
}
/**
* The preserved prefix of an ordered event-row list (mirrors the JSONL
* backend's `scanLog`): the longest prefix of complete, seq-contiguous,
* parseable rows, PLUS the seq from which a never-committed torn tail must be
* deleted (or `undefined` if the whole list is intact).
* Find the preserved prefix of ordered event rows. Fully written rows in an
* interrupted final turn remain in the prefix. The first unparsable row or seq
* gap after the last `turn/end` marks a tolerated torn tail; the same hole in
* the committed region rejects.
*
* A crash can leave a durable log whose final turn never closed: real,
* fully-written rows sit after the last `turn/end`. Those are PRESERVED — a
* single turn can be huge in a long-horizon task, so truncating it would
* destroy real work; the backend closes the orphaned open turn with a synthetic
* `turn/end {kind:'interrupted'}` on load (the session-persistence RFC). The ONLY thing excluded is
* a torn trailing fragment — a row whose `data` never parses, or a seq gap —
* AFTER the last committed `turn/end`; that bounds the preserved region and its
* seq is returned as `tornFrom` so `load` can physically delete it.
*
* The last `turn/end` is computed from the `type` COLUMN (never parsing tail
* `data`), so a malformed `data` in an uncommitted tail row is discarded rather
* than making the session unloadable. A parse error or seq gap AT OR BEFORE the
* last committed `turn/end` is committed-data corruption and throws.
*
* This relies on the session-log invariant that every event lives inside a turn
* (`Session.append` enforces it): only the final turn can be open, so the
* preserved tail is at most one unclosed turn.
* @param rows - one session's event rows, ordered by seq ascending.
* @returns the preserved event prefix, plus `tornFrom` — the seq the physical
* delete starts at — when a torn tail exists.
@@ -207,12 +170,8 @@ export function scanRows(rows: readonly EventRow[]): { preserved: SessionEvent[]
if (parsed[i]?.ok && rows[i]?.type === 'turn/end') { lastTurnEnd = i; break }
}
// Walk the longest PREFIX of complete, seq-contiguous, parseable rows
// (row i has seq === i). This includes the fully-written rows of an
// interrupted final turn AFTER the last turn/end — real work, never
// truncated. The walk stops at the first hole:
// - at or before the last committed turn/end → committed corruption (throw);
// - after it (or no committed turn/end) → tolerated torn tail (stop).
// Preserve the contiguous prefix, including a complete interrupted turn;
// holes through the last committed boundary throw, while later holes stop.
const preserved: SessionEvent[] = []
for (let i = 0; i < rows.length; i++) {
const p = parsed[i]

32
packages/session-persistence/session-persistence-sqlite/tests/sqlite.spec.ts
View File

@@ -28,8 +28,7 @@ async function backend(path = ':memory:'): Promise<{ ctx: Context; dispose: () =
return { ctx, dispose: () => fiber.dispose() }
}
// The payoff: the SAME backend-agnostic contract the JSONL backend runs, now
// proving the SQLite backend satisfies identical semantics.
// Run the same backend-agnostic contract as JSONL to pin identical semantics.
runPersistenceContract('sqlite', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
@@ -40,11 +39,8 @@ runPersistenceContract('sqlite', async () => {
}
})
// Run the shared coordinator orchestration suite against the real SQLite backend.
// A FILE-backed db (not :memory:) is the shared storage scope so two mounted
// instances see the same rows (HMR/reload). `corruptTail` INSERTs a row past the
// committed seq whose `data` is invalid JSON — a never-committed torn tail that
// drives the coordinator's commitRepair-with-tornMarker branch over real db rows.
// A file-backed database lets two mounts share rows across reload. `corruptTail` inserts invalid
// JSON past the committed seq, exercising coordinator repair against real database rows.
runCoordinatorContract('sqlite', async (): Promise<CoordinatorFixture> => {
const dir = await mkdtemp(join(tmpdir(), 'dsh-sqlite-coord-'))
const path = join(dir, 'sessions.db')
@@ -66,9 +62,9 @@ runCoordinatorContract('sqlite', async (): Promise<CoordinatorFixture> => {
})
describe('scanRows', () => {
// scanRows works off EventRows (data is a JSON string column); build them from
// SessionEvents so the unit tests read in terms of the event vocabulary. Surface
// fields are serialized to their nullable columns so a round trip is faithful.
// scanRows works off EventRows (data is a JSON string column); build them from SessionEvents
// so the unit tests read in terms of the event vocabulary. Surface metadata is serialized to
// its nullable columns so the conversion remains faithful.
const rows = (events: SessionEvent[]): EventRow[] =>
events.map((e) => {
const se = e as SessionEvent<SurfaceEventType>
@@ -256,11 +252,9 @@ describe('SessionPersistenceSqlite: durability and crash semantics', () => {
})
it('rejects a sibling v3 database (the merge-collided version) rather than opening it against missing columns', async () => {
// Two unmerged branches each shipped a DISTINCT layout under user_version 3
// (one added only `seed_length`, the other only the surface columns). The
// merged build is v4; an on-disk v3 is ambiguous and is missing at least one
// of this build's columns, so it MUST be rejected, not opened. Stamp a v3
// database and confirm the version check refuses it.
// Two unmerged branches each shipped a DISTINCT layout under user_version 3 (one added only
// `seed_length`, the other only the surface columns). The merged v4 cannot interpret that
// ambiguous, incomplete layout and must reject it.
const path = await freshDbPath()
openDatabase(path, 'wal').close() // creates + stamps user_version = SCHEMA_VERSION (4)
const db = openDatabase(path, 'wal')
@@ -277,11 +271,9 @@ describe('SessionPersistenceSqlite: durability and crash semantics', () => {
await b1.ctx.sessionPersistence.append(m.id, oneTurnLog()) // committed: seqs 0..5
await b1.dispose()
// Hand-insert a torn tail row (seq 6, no closing turn/end) whose `data` is
// invalid JSON. The contract: only a parse error in the COMMITTED region is
// unloadable; a torn tail must be discarded. scanRows finds the last
// turn/end on the seq+type columns (never parsing tail `data`), so the
// unparsable row after it bounds the preserved prefix and is deleted by load.
// A torn row after the last committed turn has invalid JSON. `scanRows` locates the boundary
// from seq/type columns without parsing the tail, preserves the committed prefix, and load
// deletes the row; invalid JSON inside the committed region would remain fatal.
const db = openDatabase(path, 'wal')
db.prepare('INSERT INTO events (session_id, seq, type, time, data) VALUES (?, 6, ?, 7, ?)')
.run(m.id, 'turn/start', '{not valid json')

4
packages/session-persistence/session-persistence/README.md
View File

@@ -22,9 +22,7 @@ The persisted unit IS the existing `SessionEvent` (event-sourced model — the l
## The write coordinator
The two first-party backends were byte-identical (or same-algorithm) for ALL of their write-path orchestration — the in-memory bookkeeping (per-id state, write-behind buffers, per-id serialization chains, per-session init promises), the `session/event` → buffer → `session/flush` drain, lazy materialization, crash-tail repair on load, the four `session/created` adoption cases (new / HMR-adopt / collision / ownerless-claim), and dispose-time quiescence. Only the STORAGE primitives differed (write bytes vs. INSERT rows).
`PersistenceCoordinator` owns that orchestration once. A first-party backend composes one (`new PersistenceCoordinator(ctx, this)`), implements the small `PersistenceBackend` hook interface, and delegates its four public service methods to the coordinator. The correctness-heavy orchestration therefore has one implementation and one place for fixes.
`PersistenceCoordinator` owns per-id state, write-behind buffers and serialization, the `session/event` → `session/flush` drain, lazy materialization, crash-tail repair, session adoption, and quiescent disposal. A first-party backend composes one, implements the small `PersistenceBackend` storage hook interface, and delegates its four public service methods. JSONL and SQLite therefore share lifecycle correctness while retaining different storage primitives; see the [coordinator RFC](../../../docs/rfc/implemented/architecture/2026-06-18-shared-persistence-write-coordinator.md).
The `PersistenceBackend<TornMarker>` hooks (the only seam between the coordinator and storage):

92
packages/session-persistence/session-persistence/src/coordinator.ts
View File

@@ -1,26 +1,7 @@
/**
* The backend-agnostic write-path orchestration shared by every first-party
* {@link SessionPersistence} backend.
*
* Every durable backend needs the same orchestration: the in-memory bookkeeping
* (the per-id state, the write-behind buffers, the per-id serialization chains,
* the per-session init promises), the `session/event` → buffer → `session/flush`
* drain, lazy materialization, crash-tail repair on load, the four
* `session/created` adoption cases (new / HMR-adopt / collision /
* ownerless-claim), and dispose-time quiescence. Only the STORAGE primitives are
* backend-specific (file bytes for `dsh-session-persistence-jsonl`, `node:sqlite`
* rows for `dsh-session-persistence-sqlite`). {@link PersistenceCoordinator} owns
* the orchestration; a backend supplies the storage primitives as a small
* {@link PersistenceBackend} hook object.
*
* The abstract {@link SessionPersistence} service's public API is independent of
* this: a backend IS a `SessionPersistence` (its four public methods delegate to
* a coordinator it composes), so a third-party backend MAY implement the service
* directly without using the coordinator at all.
*
* See the write-coordinator RFC (docs/rfc/implemented/architecture/2026-06-18-shared-persistence-write-coordinator.md)
* for the design rationale (composition over inheritance, the opaque torn marker).
*
* Shared buffering, serialization, adoption, repair, and disposal orchestration
* for first-party backends. Third-party backends may implement the public
* persistence seam directly.
* @module @deepseek-ai/dsh-session-persistence/coordinator
*/
@@ -29,16 +10,9 @@ import { interruptedTurnClosers, SESSION_FORMAT_VERSION, snapshotJsonValue } fro
import type { Session, SessionEvent, SessionId, SessionHeader } from '@deepseek-ai/dsh-session'
/**
* A stored session's durable prefix as read back from a backend: its
* {@link SessionHeader}, the preserved (seq-contiguous, parseable) event prefix,
* and an OPAQUE `tornMarker` that is present iff a never-committed torn tail must
* be truncated before further writes.
*
* The coordinator NEVER inspects `tornMarker`'s value — it only tests
* `!== undefined` (is there a tail to repair?) and passes the value back to
* {@link PersistenceBackend.commitRepair}. Each backend chooses its own marker
* type: the JSONL backend uses the byte offset to truncate to, the SQLite
* backend uses the seq to delete from (both happen to be `number`).
* A stored session's header, valid contiguous event prefix, and optional opaque
* torn-tail marker. The coordinator only checks marker presence and returns its
* value to {@link PersistenceBackend.commitRepair}; each backend owns the type.
*/
export interface StoredPrefix<TornMarker = unknown> {
meta: SessionHeader
@@ -111,16 +85,9 @@ interface SessionState {
/** The next seq the backend expects to append (the stored log length). */
cursor: number
/**
* Whether the backend has physically written this session (a JSONL file /
* SQLite row exists). `create()` registers state LAZILY — cursor 0,
* materialized false, nothing on disk — so an empty session leaves no
* artifact and the FIRST `appendBatch` writes the header + its events in ONE
* transaction (the "a row exists ⇔ it has events" invariant `list`
* relies on; a separate up-front materialize could crash leaving a row with
* zero events). The flag is the only signal that distinguishes a session
* registered-but-never-written from one durably present, which the reclaim
* path needs (an abandoned id with no artifact AND no buffered events is free
* to reuse; a materialized one is a real collision).
* Whether lazy creation has produced a durable artifact. The first append
* atomically materializes the header with events; reclaim logic uses this to
* distinguish an unused id from a persisted collision.
*/
materialized: boolean
/**
@@ -174,13 +141,9 @@ export class PersistenceCoordinator<TornMarker = unknown> {
*/
private chains = new Map<SessionId, Promise<unknown>>()
/**
* Per-session init promise (onCreated). Keyed by the LIVE Session OBJECT, not
* its id: a disposed fiber's session can be replaced by a different live
* Session reusing the same id (HMR, an ACP reconnect), and an id-keyed cache
* would hand the new object the old object's init promise.
*
* Flush is the public observation boundary for initialization; callers do
* not inspect this bookkeeping directly.
* Init promises keyed by live session object, preventing an id-reusing
* replacement from inheriting stale initialization. Flush is the public
* observation boundary; callers do not inspect this bookkeeping directly.
*/
private inits = new Map<Session, Promise<void>>()
@@ -191,16 +154,11 @@ export class PersistenceCoordinator<TornMarker = unknown> {
// --- public surface (the backend's service methods delegate here) ---
/**
* Register a new session's metadata (lazy: no physical write until the first
* {@link append}). Rejects if the id is already tracked or already persisted.
* @param meta - the header (id, version, cwd, lineage) to record; materialized
* as a detached lossless-JSON snapshot at call time.
* Register detached session metadata for lazy creation on the first append.
* @param meta - header to snapshot; duplicate tracked or persisted ids reject.
*/
create(meta: SessionHeader): Promise<void> {
// Snapshot the metadata at call time: the op runs later (behind the
// per-session chain) and the snapshot is stored as the lazy state, so keeping
// the caller's object by reference would let a later mutation of `id`/`cwd`
// register under one key but materialize under a different path/header.
// Snapshot before queueing so caller mutation cannot diverge the key and header.
const snapshot = snapshotJsonValue(meta)
if (snapshot === undefined) {
return Promise.reject(new TypeError('session metadata must be losslessly JSON-serializable'))
@@ -281,32 +239,20 @@ export class PersistenceCoordinator<TornMarker = unknown> {
const { meta, events, tornMarker } = stored
this.assertVersion(meta)
// Crash-recovery: if the log ended mid-turn (real, preserved events but no
// closing turn/end), close it durably DURING load so disk, the returned log,
// and the cursor all agree. The interrupted turn's real events are preserved,
// never truncated (a turn can be huge — the session-persistence RFC); only a
// never-fully-written torn tail fragment is discarded.
// Preserve complete interrupted events and synthesize only missing closers.
const closers = interruptedTurnClosers(events)
const balanced = [...events, ...closers]
// Make the repair durable (truncate the torn tail + append the synthetic
// closers) BEFORE recording state — commitRepair takes `meta` directly, so
// there is no state-path ordering dependency (uniform across backends).
// Repair storage before publishing coordinator state.
if (tornMarker !== undefined || closers.length > 0) {
await this.backend.commitRepair(meta, tornMarker, closers)
}
// The state keeps its OWN copy of the meta; the returned value is separate so
// a consumer mutating loaded.meta cannot corrupt the backend's metadata.
// Keep coordinator metadata detached from the returned record.
this.states.set(id, { meta: { ...meta }, cursor: balanced.length, materialized: true })
return { meta, events: balanced }
}
// NOTE: there is deliberately no coordinator `list()`. Listing needs none of
// the coordinator's orchestration (no per-id serialization, no cursor, no
// in-memory state) — it is a pure read of stored metadata. A backend's public
// `list()` IS the {@link PersistenceBackend.list} hook (one method); routing it
// through the coordinator would only forward to that same hook, so the
// coordinator stays out of the listing path entirely.
// Listing is a direct backend read and needs no coordinator state.
// --- per-id serialization + adoption helpers ---

79
packages/session-persistence/session-persistence/src/index.ts
View File

@@ -1,23 +1,7 @@
/**
* The durable session-persistence seam (`ctx.sessionPersistence`): an abstract
* service defining WHAT a persistence backend does — durably store, reload,
* and list sessions — without saying HOW. Implementations subclass
* {@link SessionPersistence} and register themselves as the
* `sessionPersistence` service; `@deepseek-ai/dsh-session-persistence-jsonl`
* (an append-only JSONL log per session) is the first and
* `@deepseek-ai/dsh-session-persistence-sqlite` (`node:sqlite`, one row per
* event) is a second that validates the seam is backend-agnostic by passing
* the same `runPersistenceContract` suite. Further backends swap in an object
* store or a remote service without touching the consumers (the write-path
* plugin, the agent-loop resume seam).
*
* The persisted unit IS the existing {@link SessionEvent} — there is no
* parallel "persisted message" type the log must be converted to and from
* (faithful to the event-sourced model: the log is the single source of
* truth). Metadata that is NOT replayable conversation state (format version,
* cwd, lineage, seed boundary) travels separately as {@link SessionHeader},
* which is owned by `dsh-session` and re-exported here.
*
* Durable session-persistence seam (`ctx.sessionPersistence`). Backends store
* {@link SessionEvent}s as the event-sourced log and carry non-replayable
* {@link SessionHeader} metadata separately.
* @module @deepseek-ai/dsh-session-persistence
*/
@@ -38,32 +22,10 @@ declare module 'cordis' {
}
/**
* Abstract durable session-persistence service. Subclass, implement the
* abstract methods, and load the subclass as a plugin — it registers as
* `ctx.sessionPersistence` (one implementation per context; loading a second
* throws, cordis' standard duplicate-service behavior).
*
* Contracts every implementation MUST honor (a DB backend asserts them inside
* a transaction; a file backend appends at EOF):
*
* - **Append-only; a crashed turn is closed, not truncated.** Committed events
* — those at or below a flushed `turn/end` — are never rewritten. A crash can
* leave an unclosed final turn whose events are real (and possibly large);
* {@link load} preserves them and closes the orphaned turn with synthetic
* boundary events (see {@link load}). Only a never-fully-written torn tail
* fragment is discarded.
* - **Contiguous seq.** A persisted log is contiguous: `events[i].seq === i`.
* {@link load} rejects a parse error or a `seq` gap in the COMMITTED region
* (unloadable); {@link append}'s first event `seq` MUST equal the backend's
* stored next-seq (after `load` has balanced any interrupted turn).
* - **JSON-serializable events.** `SessionEventMap` is merge-extensible, so
* {@link append} materializes each complete batch through the shared
* lossless-JSON boundary before buffering it. The public `session.events`
* view is immutable, but persistence still snapshots direct/replay callers at
* this independent trust boundary.
* - **Durability.** {@link append} returns only once the batch is durable
* (the file backend fsyncs; a DB commits). {@link create} MAY defer the
* physical write until the first {@link append} (lazy materialization).
* Durable append-only session storage. Implementations preserve contiguous,
* losslessly JSON-serializable events; {@link append} resolves only after
* durability, and {@link load} balances a complete interrupted tail without
* rewriting committed events.
*/
export abstract class SessionPersistence extends Service {
constructor(ctx: Context) {
@@ -91,29 +53,12 @@ export abstract class SessionPersistence extends Service {
abstract append(id: SessionId, events: readonly SessionEvent[]): Promise<void>
/**
* Reload a session: its {@link SessionHeader} plus the event log up to the last
* durable checkpoint. Returns `meta` AND `events` so the live session is
* reconstructed with its `cwd`/lineage, not just its log.
*
* The loop only flushes at `turn/end`, so a crash can leave a durable log
* whose final turn never closed: real, fully-written events sit after the last
* `turn/end`. Those events are PRESERVED — a single turn can be huge in a
* long-horizon task, so truncating it would destroy real work — and `load`
* CLOSES the orphaned turn by durably appending the minimal synthetic boundary
* events: an error `tool/result` for every `tool-call` the crash left
* unanswered (so the rehydrated history is a valid provider transcript — a
* dangling assistant tool-call is otherwise rejected), then a `step/end` if a
* step was open, then a `turn/end` carrying the `{ kind: 'interrupted' }`
* reason. The returned `events` therefore end on a balanced `turn/end` and are
* immediately usable as a session seed. Only a never-fully-written TORN tail
* fragment (a half-written final record) is discarded. Returned events are
* contiguous (`events[i].seq === i`); a parse error or a `seq` gap in the
* COMMITTED region (at or before the last real `turn/end`) makes the session
* unloadable (reject). Rejects an unknown format `version`. See the session-persistence RFC for
* the crash-recovery contract.
* Load a header and balanced contiguous log. A complete interrupted final
* turn is preserved and durably closed with missing tool errors plus any open
* step and turn boundaries; only a torn final record is discarded. Unknown
* versions and corruption in the committed prefix reject.
* @param id - the persisted session to reload.
* @returns the header plus the event log, ending on a balanced `turn/end` —
* immediately usable as a session seed.
* @returns the header and a log ending on a balanced `turn/end`.
*/
abstract load(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }>

20
packages/session-persistence/session-persistence/tests/contract.ts
View File

@@ -43,15 +43,9 @@ export function oneTurnLog(): SessionEvent[] {
}
/**
* Append a whole event log to a LIVE session, event by event, forwarding the
* surface metadata each event already carries. A bare `append(e.type, e.data)`
* over a `SessionEvent[]` widens the type argument to the union, where the
* typed overload's mandatory-marker rule collapses to optional — and `append`'s
* runtime guard then rejects a surface-eligible event with no marker. This
* helper forwards the `surfaceOp`/`sourceEventSeqs` VERBATIM from the source
* event (it does not synthesize a default), so a well-formed recorded log
* round-trips through a live session intact and a fixture that forgot a marker
* still trips the guard.
* Append recorded events to a live session while forwarding surface metadata verbatim. The broad
* `SessionEvent` union makes the typed marker optional, but the runtime guard must still reject a
* surface event whose fixture omitted it; this helper never synthesizes a default.
*/
export function appendLog(session: Session, events: readonly SessionEvent[]): void {
for (const e of events) {
@@ -222,10 +216,10 @@ export function runPersistenceContract(name: string, make: () => Promise<Contrac
it('append rejects non-JSON-serializable event data, naming the event type', async () => {
const { persistence, dispose } = await make()
try {
// Every value `isJsonValue` rejects must be rejected by the backend, not
// just BigInt — otherwise a backend could pass this contract while still
// accepting values that corrupt the durable round-trip. Each is a
// plugin-added `extra` field on a single user/message (seq 0).
// Every value `isJsonValue` rejects must be rejected by the backend, not just BigInt —
// otherwise a backend could pass this contract while still accepting values that
// corrupt the durable round-trip. Each value is carried in a plugin-added field on one
// user message so the contract covers the complete JSON-value boundary.
const cyclic: Record<string, unknown> = { type: 'text', text: 'x' }
cyclic['self'] = cyclic
const badValues: unknown[] = [

74
packages/session-persistence/session-persistence/tests/coordinator-contract.ts
View File

@@ -1,28 +1,11 @@
/**
* Reusable ORCHESTRATION suite for any backend that composes a
* {@link PersistenceCoordinator}. Where {@link runPersistenceContract} (in
* contract.ts) pins the public read/write SEMANTICS, this suite pins the
* coordinator's WRITE-PATH ORCHESTRATION — the behavior that is identical across
* every first-party backend because it lives in the shared coordinator, not in
* the storage primitives: the `session/created` → `session/event` →
* `session/flush` → dispose drain, lazy materialization, fork-seed persistence,
* the four `onCreated` adoption cases (new / HMR-adopt / collision /
* ownerless-claim), crash-tail repair on load, and dispose-time quiescence.
*
* A backend imports {@link runCoordinatorContract} and calls it with a
* {@link CoordinatorFixture} factory that knows how to (a) mount the REAL
* backend plugin on a {@link Context} over a SHARED storage scope (so HMR/reload
* tests can dispose one instance and mount another over the same bytes/rows),
* and (b) inject a never-committed torn tail for one session
* ({@link CoordinatorFixture.corruptTail}) so the through-coordinator torn-tail
* repair branch is exercised against real storage. The suite drives everything
* through the PUBLIC {@link SessionPersistence} API + the cordis SessionStore
* write path — never the storage primitives directly — so it runs unchanged for
* every backend (memory / jsonl / sqlite).
*
* Each scenario lives here once and runs once per backend through the fixture;
* the per-backend specs keep ONLY their storage-mechanics tests.
* Shared write-path orchestration contract for backends using {@link PersistenceCoordinator}.
* Unlike the public storage-semantics suite in `contract.ts`, it covers SessionStore event wiring,
* lazy creation, fork seed persistence, four adoption/collision cases, crash-tail repair, reload,
* flush, and disposal quiescence through public APIs rather than storage primitives.
*
* Each real backend supplies a shared storage scope and optional torn-tail injector; backend specs
* retain only storage-mechanics tests, while these scenarios run once per backend.
* @module @deepseek-ai/dsh-session-persistence/tests/coordinator-contract
*/
@@ -38,25 +21,12 @@ import { meta, oneTurnLog, appendLog } from './contract.ts'
* the suite mounts/disposes backend instances on it and cleans it up at the end.
*/
export interface CoordinatorFixture {
/**
* Mount the REAL backend plugin (via `ctx.plugin`, the Loader path) on `ctx`,
* over THIS fixture's shared storage scope. Returns the plugin fiber so the
* suite can dispose a single instance (HMR/reload) while the storage — and any
* still-live session in another fiber — survives. The caller has already
* mounted `SessionStore` on `ctx`.
*/
/** Mount the real backend through `ctx.plugin` over shared storage and return only that fiber. */
mount: (ctx: Context) => Promise<Fiber>
/**
* Inject a NEVER-COMMITTED torn tail into the backend's storage for `id` at
* the given `cwd` (the cwd the session was created with): a half-written
* record past the committed region (JSONL: a partial line with no newline;
* SQLite: a row with invalid `data` JSON past the committed seq). This drives
* the coordinator's `loadCore` `tornMarker !== undefined` → `commitRepair`
* branch against real storage.
*
* OMITTED by a backend that structurally has no torn tails (memory): the
* torn-tail scenario then self-skips (asserted explicitly in the suite).
* Inject a never-committed partial record after the durable region so `loadCore` reaches
* `commitRepair`. Omit only when the backend structurally cannot produce torn tails.
*/
corruptTail?: (id: SessionId, cwd: string | undefined) => Promise<void>
@@ -118,10 +88,9 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
})
it('round-trips the seed boundary (seedLength) through persistence', async () => {
// A forked child records how many leading events were inherited via the
// seed; the boundary must survive a reload (so a resume/replay can tell the
// inherited prefix from the child's own events). Both backends carry it on
// the header — JSONL on the header line, SQLite in the seed_length column.
// A forked child records how many leading events were inherited via the seed; the
// boundary must survive a reload (so a resume/replay can tell the inherited prefix from
// the child's own events). JSONL stores it in the header; SQLite uses `seed_length`.
const fix = await makeFixture()
const { ctx, fiber } = await freshCtx(fix)
try {
@@ -207,10 +176,10 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
})
it('resume: a re-created session seeded with the loaded log does not re-append its seed and continues the seq', async () => {
// Separate backend lifecycles distinguish persisted-seed adoption from an in-memory continuation.
const fix = await makeFixture()
const first = await freshCtx(fix)
try {
// First lifecycle: persist a session through the store.
const s1 = first.ctx.sessions.create(SessionId('resumed'), { meta: { cwd: WORK } })
send(s1, oneTurnLog())
await first.ctx.parallel('session/flush', s1)
@@ -218,9 +187,6 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
await first.fiber.dispose()
}
// Second lifecycle: a NEW backend instance + a session re-created with the
// same id SEEDED with the loaded events. onCreated adopts the stored log
// (does not re-persist the seed); a new turn appends at seq 6.
const second = await freshCtx(fix)
try {
const loaded = await second.ctx.sessionPersistence.load(SessionId('resumed'))
@@ -231,7 +197,6 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
await second.ctx.parallel('session/flush', s2)
const reloaded = await second.ctx.sessionPersistence.load(SessionId('resumed'))
// 6 original + 2 new, contiguous, no duplicated seed.
expect(reloaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7])
} finally {
await second.fiber.dispose()
@@ -298,10 +263,9 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
await ctx.parallel('session/flush', session)
// Hot-reload: dispose instance 1, mount instance 2 over the SAME storage
// while the session stays live. Instance 2 has an empty states map but the
// log is materialized and is a prefix of the live events — it must ADOPT
// (not reject). A second turn appended after reload then persists.
// Hot-reload: dispose instance 1, mount instance 2 over the same storage while the
// session stays live. The new instance has no coordinator state but must adopt the
// materialized prefix, then persist another turn rather than rejecting it as a collision.
await backend1.dispose()
await fix.mount(ctx)
session.append('turn/start', { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } })
@@ -393,9 +357,9 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
await first.fiber.dispose()
}
// A FRESH backend + a NEW live session with the same id but NO explicit
// resume. onCreated treats it as new; create() rejects because a log already
// exists. The rejection surfaces via the init promise (flush awaits it).
// A fresh backend + a NEW live session with the same id but NO explicit resume. onCreated
// treats it as new; create() rejects because a log already exists, and `flush()` surfaces
// that initialization rejection.
const second = await freshCtx(fix)
try {
const s2 = second.ctx.sessions.create(SessionId('collide'), { meta: { cwd: WORK } })

27
packages/session-persistence/session-persistence/tests/persistence.spec.ts
View File

@@ -16,18 +16,10 @@ type MemoryStore = Map<string, { meta: SessionHeader; events: SessionEvent[] }>
interface MemoryConfig { store?: MemoryStore }
/**
* A trivial in-memory {@link SessionPersistence} that composes a
* {@link PersistenceCoordinator} over a dependency-free `Map`-backed
* {@link PersistenceBackend}. It is BOTH the coordinator's reference vehicle
* (the simplest possible storage — a `Map<id, {meta, events}>` with no torn
* tails, so `tornMarker` is always undefined) and the cover for the abstract
* base's constructor + service registration. The real durable backends are
* `@deepseek-ai/dsh-session-persistence-jsonl` / `-sqlite`.
*
* The store can be supplied via config so two backend instances share one Map —
* the in-RAM analogue of two backends over the same file/db, which the
* coordinator orchestration suite's HMR/reload tests need (a fresh instance with
* an empty in-memory states map adopting an already-materialized session).
* Reference {@link PersistenceCoordinator} vehicle and abstract-service coverage, backed by a
* dependency-free map with atomic writes and no torn-tail marker. Supplying the map lets multiple
* instances share materialized sessions, the in-memory analogue of reload over one file/database;
* durable behavior is covered by the JSONL and SQLite backends.
*/
class MemoryPersistence extends SessionPersistence implements PersistenceBackend<never> {
static inject = ['sessions']
@@ -83,8 +75,7 @@ class MemoryPersistence extends SessionPersistence implements PersistenceBackend
}
const existing = this.store.get(m.id)
if (!existing) {
// First batch: `_isMaterialized` is false (the coordinator only omits
// materialization on the first batch); writing the entry IS the materialization.
// The coordinator sends the first batch for materialization; later batches append.
this.store.set(m.id, { meta: structuredClone(m), events: structuredClone(events) as SessionEvent[] })
} else {
existing.events.push(...structuredClone(events) as SessionEvent[])
@@ -117,12 +108,8 @@ runPersistenceContract('memory', async () => {
}
})
// Run the shared coordinator orchestration suite against the in-memory backend.
// A per-fixture Map is the shared "storage", so two mounted instances see the
// same materialized sessions (HMR/reload). `corruptTail` is OMITTED: a Map store
// writes atomically in RAM and has no torn tails, so the suite's torn-tail test
// self-skips (and asserts the omission). The real torn-tail repair branch is
// covered by the jsonl/sqlite fixtures, which CAN inject one.
// Each fixture shares one map across mounts. No `corruptTail` is supplied because map writes are
// atomic; the suite asserts that skip while JSONL and SQLite cover the repair branch.
runCoordinatorContract('memory', async (): Promise<CoordinatorFixture> => {
const store: MemoryStore = new Map()
return {