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

refactor(session-persistence): extract a shared write coordinator

Browse Source 
The JSONL and SQLite backends were byte-identical (or same-algorithm) for ALL
of their write-path orchestration — the four maps (states/buffers/chains/inits),
installWritePath, initFor, onCreated's four adoption cases, flush, drain,
serialize, adopt/adoptLivePrefix, assertVersion, and the create/append/load/
has/delete skeletons. Only the storage primitives (write bytes vs INSERT rows)
differed, so every fix landed twice.

Extract that orchestration into a PersistenceCoordinator in the seam package.
Each backend composes one (new PersistenceCoordinator(ctx, this)), implements a
small PersistenceBackend hook interface (loadStored, loadLive, appendBatch,
commitRepair, deleteStored, list, optional close), and delegates its six public
service methods to it. Composition, not inheritance — a backend exposes only the
hooks, can't reach the coordinator's private state, and the public
SessionPersistence API is unchanged so a third-party backend may still implement
it directly.

The crash-repair torn-tail token is OPAQUE: the coordinator computes the
synthetic closers (it owns interruptedTurnClosers) but only tests
`tornMarker !== undefined` and round-trips it to commitRepair, never inspecting
it (JSONL = byte offset, SQLite = seq). loadStored vs loadLive stay distinct so
HMR adoption is cwd-scoped (a same-id log at a different cwd is a collision, not
a resume). appendBatch carries meta so lazy-materialize + first-batch commit
atomically (no separate materialize hook).

Tests: the duplicated orchestration tests (adoption, HMR, collision,
dispose-drain, crash-tail) move into one runCoordinatorContract suite run once
per backend (memory + jsonl + sqlite) via hook fixtures; per-backend specs keep
only storage mechanics. A through-coordinator torn-tail test per real backend
keeps the commitRepair-with-marker branch covered under the 100% gate.

Net -112 lines (the dedup outweighs the new coordinator + shared suite); 100%
coverage; backends shrank ~1200 lines of duplicated churn. Migrates the
write-coordinator RFC proposed -> implemented.
This commit is contained in:
Tianyi Cui
2026-06-20 03:47:28 +08:00
parent 31af23b4fe
commit ab02e9acec
13 changed files with 1879 additions and 1991 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,20 +1,18 @@
/**
* JSONL durable session-persistence backend (`@deepseek-ai/dsh-session-persistence-jsonl`).
*
* Two concerns in one plugin:
* 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.
*
* 1. **The backend** — a concrete {@link SessionPersistence}: one append-only
* `.jsonl` event log per session (a header line then one `SessionEvent` per
* line, verbatim including `assistant/chunk` so `seq` stays contiguous).
* Lazy materialization (no file until the first `append`), atomic first
* write, and load-time repair of a never-committed crash tail.
*
* 2. **The write path** — the `session/event` → buffer → `session/flush` drain
* that generalizes the example `session-jsonl.ts`: snapshot each event when
* it is buffered (the live `session.events` object is mutable), persist
* forks once on `session/created`, maintain a per-session write cursor so a
* resumed session never re-appends already-stored events, and seed existing
* live sessions on plugin apply (HMR does not replay `session/created`).
* 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
*/
@@ -25,9 +23,9 @@ import { open, mkdir, readFile, readdir, link, rm, truncate } from 'node:fs/prom
import { dirname, resolve } from 'node:path'
import { randomBytes } from 'node:crypto'
import {
SessionPersistence, assertSerializable, seedCoversPrefix,
SessionPersistence, PersistenceCoordinator,
type PersistenceBackend, type StoredPrefix,
} from '@deepseek-ai/dsh-session-persistence'
import { interruptedTurnClosers } from '@deepseek-ai/dsh-session'
import type { Session, SessionEvent, SessionId, SessionHeader } from '@deepseek-ai/dsh-session'
import {
encodeSegment, eventLine, logPath, parseHeaderMeta, scanLog, sessionDir, toHeaderLine,
@@ -42,267 +40,152 @@ export interface Config {
root: string
}
/** Per-session write state held by the backend's in-memory bookkeeping. */
interface SessionState {
meta: SessionHeader
/** The next seq the backend expects to append (the stored log length). */
cursor: number
/** Whether the `.jsonl` file has been physically materialized. */
materialized: boolean
/**
* The live Session this state was bound to via `onCreated`, if any. Used to
* detect a DIFFERENT live session reusing a tracked id (a collision): state
* created through the public `create()`/`load()` API has no owner, but state
* bound to a live session lets `onCreated` reject a second, unrelated session
* object on the same id instead of silently no-opping (which would leave the
* new session's events to be dropped against the old cursor).
*/
owner?: Session
}
/**
* 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: masking it would let
* `list()` report no sessions, `load()` report "not found", and collision
* checks proceed under a false absence assumption — all unsafe for durable
* persistence. (A NodeJS filesystem rejection carries a string `code`.)
* 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'
}
async function settledErrors(promises: Iterable<Promise<unknown>>): Promise<unknown[]> {
const settled = await Promise.allSettled([...promises])
const errors: unknown[] = []
for (const result of settled) {
if (result.status === 'rejected') errors.push(result.reason)
}
return errors
}
/**
* The JSONL persistence backend. Load as a plugin; it registers as
* `ctx.sessionPersistence` and installs the write-path listeners.
* `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 {
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
/** Backend bookkeeping keyed by session id (NOT the live Session object). */
private states = new Map<string, SessionState>()
/** Write-behind buffers keyed by the live Session (write path). */
private buffers = new Map<Session, SessionEvent[]>()
/**
* Per-session serialization: every backend operation chains onto the prior
* one for the same id, so concurrent flushes / a flush racing onCreated never
* interleave file writes or read a half-built state. Keyed by session id.
*/
private chains = new Map<string, 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 — skipping
* onCreated for the new session, so its events start at seq 0 while flush
* filters against the stale cursor and silently drops them. Keying by object
* gives each live Session its own init. flush awaits it before appending.
*/
private inits = new Map<Session, Promise<void>>()
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 (the examples use `./.sessions`) 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. Pinning it at construction makes all paths
// stable regardless of later cwd changes.
// 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.installWritePath()
this.coordinator = new PersistenceCoordinator<number>(this.ctx, this)
}
// --- SessionPersistence backend surface (all serialized per session id) ---
// --- SessionPersistence service surface (delegated to the coordinator) ---
create(meta: SessionHeader): Promise<void> {
// Snapshot the metadata at call time: the op runs later (behind the
// per-session chain) and the snapshot is also 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. A shallow copy is enough — SessionHeader is a flat record.
const snapshot: SessionHeader = { ...meta }
return this.serialize(snapshot.id, () => this.createCore(snapshot))
return this.coordinator.create(meta)
}
private async createCore(meta: SessionHeader): Promise<void> {
// Do NOT clobber an existing session. If we already track it, or a log
// exists on disk under this id, refuse — the SessionId IS the identity, and
// silently resetting state (cursor 0, materialized false) over committed
// data would let the next append rename over the existing log.
if (this.states.has(meta.id)) {
throw new Error(`session "${meta.id}" already exists in this backend`)
}
// Scan ALL cwd buckets (pass undefined), not just meta.cwd's: load/has/adopt
// identify a session by id alone and search every bucket, so an id already
// persisted under a DIFFERENT cwd must still block creation here. Probing
// only meta.cwd's bucket would let two logs share one id and make resume
// (which picks the first matching bucket) nondeterministic.
if (await this.findLog(meta.id, undefined) !== undefined) {
throw new Error(`session "${meta.id}" already has a persisted log on disk; load/resume it instead of creating`)
}
// Pure lazy: record intent only. No file until the first append, so an
// abandoned (never-appended) session leaves nothing on disk and stays
// absent from has()/list().
this.states.set(meta.id, { meta, cursor: 0, materialized: false })
}
/**
* Run `op` after any in-flight operation for the same session id, so writes
* for one session never interleave (two flushes, a flush racing a load, an
* update racing an append). Errors do not poison the chain — the next op
* still runs. NOTE: serialized public methods must NOT call each other (that
* would deadlock on the same chain); they call the unserialized `*Core`
* helpers instead.
*/
private serialize<T>(id: SessionId, op: () => Promise<T>): Promise<T> {
const prior = this.chains.get(id) ?? Promise.resolve()
const next = prior.then(op, op)
// Keep the chain alive but swallow this op's rejection for the NEXT waiter
// (the caller still sees the real rejection via `next`).
this.chains.set(id, next.then(() => undefined, () => undefined))
return next
}
// `async` so the synchronous validate/clone below reject (not throw) per the
// Promise<void> contract — callers use `await expect(...).rejects`.
async append(id: SessionId, events: readonly SessionEvent[]): Promise<void> {
// Validate serializability BEFORE cloning, so a bad event surfaces the typed
// "non-JSON-serializable" error rather than an opaque DataCloneError from
// structuredClone below. (In an async method this throw becomes a rejection,
// honoring the Promise<void> contract rather than throwing synchronously.)
assertSerializable(events)
// Deep-snapshot the batch here, BEFORE the op waits behind the per-session
// chain: the op may await before serializing, so a caller that passes a live
// array (e.g. session.events) and mutates it — OR mutates an event object
// inside it — before the op runs would otherwise have those changes
// persisted, or advance the cursor past what was actually written.
// structuredClone covers both the array and the event objects (safe now that
// serializability is checked above). The clone happens synchronously (before
// the first await), so it is taken at call time.
const batch = events.map(e => structuredClone(e))
return this.serialize(id, () => this.appendCore(id, batch))
}
private async appendCore(id: SessionId, events: readonly SessionEvent[]): Promise<void> {
if (events.length === 0) return
assertSerializable(events)
let state = this.states.get(id)
if (state === undefined) state = await this.adopt(id) // calls loadCore, not load
// Contiguity contract: each event's seq must continue the stored log.
for (const [i, event] of events.entries()) {
if (event.seq !== state.cursor + i) {
throw new Error(`append seq mismatch for "${id}": expected ${state.cursor + i} at index ${i}, got ${event.seq}`)
}
}
if (!state.materialized) {
await this.materialize(state, events)
} else {
await this.appendLines(state, events)
}
// The durable event log is the transaction: advance the cursor as soon as
// the log write commits.
state.cursor += events.length
append(id: SessionId, events: readonly SessionEvent[]): Promise<void> {
return this.coordinator.append(id, events)
}
load(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }> {
return this.serialize(id, () => this.loadCore(id))
return this.coordinator.load(id)
}
private async loadCore(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }> {
const cwd = this.states.get(id)?.meta.cwd
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> {
return this.readPrefix(id, undefined)
}
/** Read a stored prefix SCOPED to `cwd` (HMR live-adoption must not cross cwd). */
async loadLive(id: SessionId, cwd: string | undefined): Promise<StoredPrefix<number> | undefined> {
return this.readPrefix(id, cwd)
}
/**
* Read and scan a session's log 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(id: SessionId, cwd: string | undefined): Promise<StoredPrefix<number> | undefined> {
const file = await this.findLog(id, cwd)
if (file === undefined) throw new Error(`session "${id}" not found`)
if (file === undefined) return undefined
const buffer = await readFile(file.path)
const { meta, events, committedBytes } = scanLog(buffer)
this.assertVersion(meta)
// Crash-recovery: if the log ended mid-turn (an open turn with real,
// preserved events but no closing turn/end), close it durably DURING load so
// disk, the returned log, and the cursor all agree — both append routes then
// continue with no special-casing. Synthesize the boundary events (a
// step/end if a step was open, then a turn/end {kind:'interrupted'}); the
// interrupted turn's real events are preserved, never truncated (a turn can
// be huge — the session-persistence RFC).
const closers = interruptedTurnClosers(events)
const balanced = [...events, ...closers]
// Set state BEFORE the repair writes so they can resolve the log path.
const needsTorn = committedBytes < buffer.byteLength
const state: SessionState = {
meta: { ...meta },
cursor: events.length,
materialized: true,
return {
meta,
events,
...committedBytes < buffer.byteLength ? { tornMarker: committedBytes } : {},
}
this.states.set(id, state)
if (needsTorn) {
// Discard the torn trailing fragment (a final line never fully flushed)
// before writing the closers, so the closers land at a clean EOF.
await this.repair(state, committedBytes)
}
if (closers.length > 0) {
// Durably append the synthetic closers, then advance the cursor to the
// balanced length. After this, disk == balanced and the next append (live
// or direct) continues cleanly.
await this.appendLines(state, closers)
state.cursor = balanced.length
}
return { meta, events: balanced }
}
private async adoptLiveDiskPrefix(
session: Session,
seed: readonly SessionEvent[],
file: { path: string; cwd: string | undefined },
): Promise<void> {
const buffer = await readFile(file.path)
const { meta, events, committedBytes } = scanLog(buffer)
this.assertVersion(meta)
if (!seedCoversPrefix(seed, events)) {
throw new Error(`session "${session.header.id}" already has a persisted log on disk that does not match this live session (id collision)`)
/** 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)
}
const state: SessionState = {
meta: { ...meta },
cursor: events.length,
materialized: true,
owner: session,
}
this.states.set(session.header.id, state)
if (committedBytes < buffer.byteLength) {
await this.repair(state, committedBytes)
}
const suffix = seed.slice(events.length)
if (suffix.length > 0) await this.appendCore(session.header.id, suffix)
}
/**
* 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, undefined)
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, so a
// full scanLog here would be O(total history)).
// 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)
@@ -313,11 +196,119 @@ export class SessionPersistenceJsonl extends SessionPersistence {
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
* (a half-written log). Reads in bounded chunks so a huge log costs only the
* header read.
* 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')
@@ -340,145 +331,6 @@ export class SessionPersistenceJsonl extends SessionPersistence {
}
}
async has(id: SessionId): Promise<boolean> {
const state = this.states.get(id)
if (state?.materialized) return true
const cwd = state?.meta.cwd
return (await this.findLog(id, cwd)) !== undefined
}
delete(id: SessionId): Promise<void> {
return this.serialize(id, () => this.deleteCore(id))
}
private async deleteCore(id: SessionId): Promise<void> {
const cwd = this.states.get(id)?.meta.cwd
const file = await this.findLog(id, cwd)
if (file) await rm(file.path, { force: true })
this.states.delete(id)
}
// --- materialization / append / repair ---
/** Atomically write the header line + first batch (temp-write, fsync, rename). */
private async materialize(state: SessionState, events: readonly SessionEvent[]): Promise<void> {
const dir = sessionDir(this.root, state.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, state.meta.cwd, state.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 rather than
// clobber committed data. (createCore already guards the create path before
// this point, so this is unreachable-in-practice defense-in-depth against a
// TOCTOU/fork race; ignored for coverage.)
/* 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 "${state.meta.id}": a log already exists on disk (load/resume it instead)`)
}
const header = JSON.stringify(toHeaderLine(state.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 (both could pass the exists() check
// above, but only one link() wins). rename() would silently overwrite the
// log the other process just committed.
let linked = false
try {
await link(tmp, finalPath)
linked = true
} finally {
// If link FAILED (EEXIST on a race, or any I/O error), the temp is the
// only reference and must be removed before the original error propagates.
// If link SUCCEEDED, the temp cleanup is deferred to AFTER the publish is
// durable (below) so a temp-rm failure can never reject a session whose
// log already published — that would leave state.materialized false and
// wedge every retry on the exists() backstop above.
/* 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: on POSIX filesystems the new link is not
// crash-durable until the parent directory's metadata is synced. The seam
// contract is "append returns once durable", and materialize is the first
// append's write — so the directory entry must be durable before we return.
await this.syncDir(dir)
state.materialized = true
// 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. A leftover `*.tmp` is harmless — it is never read, and the next
// materialize of this id is guarded by exists()/link(). Swallow only the
// rm failure; nothing else of consequence runs in the try.
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: `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 and render the session unloadable.
*/
private async appendLines(state: SessionState, events: readonly SessionEvent[]): Promise<void> {
const path = logPath(this.root, state.meta.cwd, state.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(state: SessionState, offset: number): Promise<void> {
const path = logPath(this.root, state.meta.cwd, state.meta.id)
await truncate(path, offset)
const handle = await open(path, 'r+')
try {
await handle.sync()
} finally {
await handle.close()
}
}
// --- discovery helpers ---
/** Find a session's log file across cwd buckets (when cwd is unknown). */
private async findLog(id: SessionId, cwd: string | undefined): Promise<{ path: string; cwd: string | undefined } | undefined> {
if (cwd !== undefined) {
@@ -490,8 +342,7 @@ export class SessionPersistenceJsonl extends SessionPersistence {
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 location (which `findLog` was given an unknown cwd for).
// 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 }
}
@@ -505,10 +356,9 @@ export class SessionPersistenceJsonl extends SessionPersistence {
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 persisted
// state is absent on a storage fault.
// 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
}
@@ -526,244 +376,12 @@ export class SessionPersistenceJsonl extends SessionPersistence {
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.
// collapsed to `false` — otherwise load() reports "not found" and collision
// checks proceed under a false absence assumption.
if (isENOENT(error)) return false
throw error
}
}
/** Build a state for a session discovered on disk but not yet in memory. */
private async adopt(id: SessionId): Promise<SessionState> {
// loadCore (NOT load) — adopt runs inside an already-serialized op, so
// re-entering the chain via the public load() would deadlock.
await this.loadCore(id)
const state = this.states.get(id)
/* v8 ignore next -- loadCore always sets the state for the id */
if (!state) throw new Error(`failed to adopt session "${id}"`)
return state
}
private assertVersion(meta: SessionHeader): void {
if (meta.version !== 1) {
throw new Error(`unsupported session format version ${meta.version} for "${meta.id}" (only v1 is supported)`)
}
}
// --- write path (session/event → flush drain) ---
private installWritePath(): void {
const ctx = this.ctx
// Capture the header on creation; persist a fork's seed once. Record the
// init promise so flush/dispose can await it (onCreated is async).
ctx.on('session/created', (session) => { void this.initFor(session) })
// Snapshot + buffer every event (the live object is mutable; clone so a
// later in-place mutation of session.events cannot rewrite a buffered
// event). Serializability is guaranteed at the source — `Session.append`
// rejects non-JSON-serializable data before the event ever enters the log
// or this emit — so structuredClone here can never hit a non-cloneable
// value, and the durable log can never diverge from session.events.
ctx.on('session/event', (session, event) => {
let buffer = this.buffers.get(session)
if (!buffer) this.buffers.set(session, buffer = [])
buffer.push(structuredClone(event))
})
// Drain to the backend at the durability checkpoint.
ctx.on('session/flush', session => this.flush(session))
// Dispose must reach quiescence: await every session's init + final drain
// BEFORE returning, so no write lands after teardown (orphan rename/ENOENT).
ctx.effect(() => async () => {
const errors = [
...await settledErrors(this.inits.values()),
...await settledErrors([...this.buffers.keys()].map(s => this.flush(s))),
...await settledErrors(this.chains.values()),
]
if (errors.length > 0) {
throw new AggregateError(errors, 'session-persistence-jsonl dispose failed')
}
}, 'session-persistence-jsonl write path')
// HMR: a hot reload does not replay session/created, so seed existing live
// sessions (mirrors dsh-invariants).
for (const session of ctx.sessions.list()) void this.initFor(session)
}
/** Start (once) the async init for a session and remember its promise. */
private initFor(session: Session): Promise<void> {
const existing = this.inits.get(session)
if (existing) return existing
// Snapshot the seed SYNCHRONOUSLY here — initFor runs inside the
// `session/created` emit, before any later `append` adds non-seed events.
// A clone freezes it against later mutation of the live event objects.
const seed = session.events.map(e => structuredClone(e))
const p = this.onCreated(session, seed)
// Attach a no-op rejection handler so a failing init (e.g. an id collision)
// does not surface as an unhandled rejection if no flush observes `p` before
// it rejects. The REAL error is still delivered: flush/dispose await the
// same `p` from the map and see the rejection there.
p.catch(() => { /* observed by flush/dispose via the stored promise */ })
this.inits.set(session, p)
return p
}
/**
* Whether a live `session`'s `seed` reproduces the first `cursor` persisted
* events. Reads the on-disk committed prefix and compares. A `cursor` of 0
* (nothing persisted yet) trivially matches. Used when a live session claims
* ownerless state left by a prior `load()`/`create()` — to reject a fresh,
* unrelated session that reuses the id and would otherwise have its seq
* 0..cursor-1 events filtered as already-written.
*/
private async seedMatchesPersisted(session: Session, seed: readonly SessionEvent[], cursor: number): Promise<boolean> {
if (cursor === 0) return true
const onDisk = await this.findLog(session.header.id, session.header.cwd)
/* v8 ignore next -- a cursor > 0 means the log was materialized, so it exists */
if (onDisk === undefined) return false
const { events: diskEvents } = scanLog(await readFile(onDisk.path))
return seedCoversPrefix(seed, diskEvents.slice(0, cursor))
}
/**
* On session/created: sync the backend's in-memory state to a live Session.
*
* Cases, by whether this backend tracks the id and whether a log is on disk:
* 1. Already in `states` (created here, or a prior load/resume) → no-op.
* 2. Not tracked, a log EXISTS on disk, and it is a seq-aligned PREFIX of the
* live session's current events → ADOPT it (HMR/reload): a fresh backend
* instance (empty `states`) meets a live session whose log a previous
* instance materialized; the live object already carries that history (it
* is the source of truth this run), so we continue from the stored length
* instead of re-creating. This keeps persistence alive across hot reload.
* 3. Not tracked, a log EXISTS on disk, but it is NOT a prefix of the live
* session's events → REJECT: a different session collides on the id. The
* SessionId is the identity, so two unrelated sessions sharing one is a
* bug, not a resume — fail loudly rather than clobber committed data.
* 4. Not tracked and NO log on disk → a genuinely new session: register its
* meta (lazy) and persist its `seed` once.
*
* The public `create(meta)` API is stricter still (rejects ANY on-disk id):
* there the caller asserts "brand new", so even a prefix match is a bug.
*
* The seed events were copied into the Session by its constructor WITHOUT
* emitting session/event, so the write-behind buffer never sees them — the
* one explicit `append(seed)` below is the only persistence of the seed.
* Events appended AFTER creation flow through the session/event buffer and
* are persisted by flush (filtered by the write cursor), never here.
*/
private async onCreated(session: Session, seed: readonly SessionEvent[]): Promise<void> {
const id = session.header.id
const tracked = this.states.get(id)
if (tracked !== undefined) {
// case 1: already tracked.
// (owner === session is a defensive same-object guard: initFor dedupes by
// session object, so onCreated never actually runs twice for one session.)
/* v8 ignore next -- initFor dedupes per session object; same-object re-entry can't occur */
if (tracked.owner === session) return
if (tracked.owner === undefined) {
// Ownerless state was created via the public create()/load() API. The
// FIRST live session to arrive claims it — but ONLY if its seed is the
// already-persisted prefix. A load() for preview leaves cursor at the
// persisted length; a fresh, unrelated session reusing that id has a
// seed shorter than (or not matching) that prefix, so flush would filter
// its seq 0..cursor-1 events as already-written and silently graft the
// new conversation onto the old log. Verify the seed covers the cursor.
if (!await this.seedMatchesPersisted(session, seed, tracked.cursor)) {
throw new Error(`session "${id}" is already persisted with ${tracked.cursor} event(s) that do not match this live session (id collision)`)
}
tracked.owner = session
// Persist the live seed SUFFIX beyond the persisted prefix. Constructor
// seed events (from sessions.create(id, { seed })) never emit
// session/event, so the write-behind buffer never sees them — without
// this they would be lost and a later flush would seq-mismatch. (cursor
// is 0 for a public create(), so this covers the whole seed there.)
const suffix = seed.slice(tracked.cursor)
if (suffix.length > 0) await this.append(id, suffix)
return
}
// The state is owned by a DIFFERENT live session. We may reclaim the id
// ONLY if that owner left nothing behind: never materialized a log (cursor
// 0, not materialized) AND has no write-behind buffer still pending. A
// session that appended events but was disposed before its first flush is
// NOT materialized yet but DOES have buffered events — reclaiming then
// would let that stale buffer drain against the new session's state
// (persisting old events under the new id, or dropping the new session's
// seq-0 events). Such an owner, and any materialized owner, is a real
// collision and rejects; only a truly-abandoned (artifact-free) id is
// freed, honoring lazy materialization's "leaves nothing behind" promise.
const ownerBuffer = this.buffers.get(tracked.owner)
if (!tracked.materialized && !ownerBuffer?.length) {
this.states.delete(id)
} else {
throw new Error(`session "${id}" is already bound to a different live session in this backend (id collision)`)
}
}
const onDisk = await this.findLog(id, session.header.cwd)
if (onDisk !== undefined) {
// case 2: adopt a LIVE prefix. Do NOT route through loadCore(): loadCore
// crash-repairs open turns as interrupted, which is right for a true load
// after a crash but wrong for HMR while the live Session is still the
// authority and may append the real step/turn end later.
await this.serialize(id, () => this.adoptLiveDiskPrefix(session, seed, onDisk))
return
}
// case 4: a genuinely new session. Register its meta (lazy), then persist
// its seed (events present at creation time) once.
const meta: SessionHeader = { ...session.header }
await this.create(meta)
// Bind this state to the live session so a later DIFFERENT session reusing
// the id is detected as a collision (case 1) rather than silently no-opped.
const created = this.states.get(id)
/* v8 ignore next -- create() always sets the state for the id */
if (created !== undefined) created.owner = session
if (seed.length > 0) {
await this.append(id, seed)
}
}
private async flush(session: Session): Promise<void> {
// Wait for the session's init (onCreated) to finish so the state/cursor and
// any fork-seed persistence are in place before we drain. Awaiting the same
// promise initFor stored also surfaces an init failure (e.g. an id
// collision) here, where the caller of session/flush observes it.
await this.inits.get(session)
// Serialize the WHOLE drain (read cursor → append → splice) on the
// per-session chain. Two concurrent flushes (e.g. an idle inject()'s
// fire-and-forget flush racing an explicit checkpoint) would otherwise both
// read the same cursor, both compute the same `fresh` slice, and the second
// append would seq-mismatch against the cursor the first already advanced.
await this.serialize(session.header.id, () => this.drain(session))
}
/** Drain a session's write buffer to disk. Caller serializes this per id. */
private async drain(session: Session): Promise<void> {
const buffer = this.buffers.get(session)
if (!buffer?.length) return
// Copy WITHOUT removing: the buffer is the only durable-pending copy of
// these events (session/event does not re-emit). Splicing before the append
// means a failed append (disk error, or a seq mismatch after a dropped bad
// event) permanently loses a completed turn. Drain the buffer only AFTER
// the append commits; events pushed during the await sit past batch.length
// and survive the prefix splice, so a retry/dispose re-drains the rest.
const batch = buffer.slice()
const state = this.states.get(session.header.id)
// Only append events at or beyond the write cursor (a resumed session's
// seed is already on disk; the cursor was set to the loaded length). flush
// awaits the init above, which always sets state, so the `?? 0` fallback is
// a defensive guard that never fires in practice.
/* v8 ignore next -- state is always set by the awaited init before flush */
const cursor = state?.cursor ?? 0
const fresh = batch.filter(e => e.seq >= cursor)
// appendCore (NOT the serialized append) — drain already runs inside the
// per-session chain, so re-entering it via append() would deadlock.
if (fresh.length > 0) await this.appendCore(session.header.id, fresh)
buffer.splice(0, batch.length)
}
}
export default SessionPersistenceJsonl

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

@@ -4,10 +4,11 @@ import { appendFile, mkdtemp, mkdir, rm, readFile, writeFile, readdir, stat } fr
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
import type { Session, SessionEvent, SessionHeader } 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[] = []
@@ -37,6 +38,29 @@ 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.
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')
@@ -198,36 +222,6 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => {
expect(loaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7])
})
it('append snapshots its batch: mutating the caller array after the call is ignored', async () => {
const m = meta('snapshot')
await ctx.sessionPersistence.create(m)
const events = oneTurnLog() // seqs 0..5
const p = ctx.sessionPersistence.append(m.id, events)
// Mutate the caller's array immediately after calling append (before the
// queued op runs). The backend must persist the snapshot taken at call time,
// not the mutated array.
events.push({ type: 'turn/start', seq: 6, time: 99, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } })
await p
const loaded = await ctx.sessionPersistence.load(m.id)
expect(loaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5]) // not 0..6
})
it('append deep-snapshots event objects: mutating an event after the call is ignored', async () => {
const m = meta('deep-snapshot')
await ctx.sessionPersistence.create(m)
const events = oneTurnLog()
const userMsg = events[1] // the user/message event
const p = ctx.sessionPersistence.append(m.id, events)
// Mutate an event OBJECT (not just the array) after calling append. The deep
// snapshot taken at call time must shield the persisted data.
if (userMsg?.type === 'user/message') userMsg.data.content = [{ type: 'text', text: 'MUTATED' }]
await p
const loaded = await ctx.sessionPersistence.load(m.id)
const persisted = JSON.stringify(loaded.events)
expect(persisted).toContain('hi') // original content
expect(persisted).not.toContain('MUTATED')
})
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)
@@ -246,25 +240,6 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => {
expect(reloaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7])
})
it('rejects an unknown format version on load', async () => {
const m = meta('v2')
await ctx.sessionPersistence.create(m)
await ctx.sessionPersistence.append(m.id, oneTurnLog())
// Corrupt the header version on disk.
const path = logPath(root, undefined, m.id)
const lines = (await readFile(path, 'utf8')).split('\n')
const header = JSON.parse(lines[0]!) as { version: number }
header.version = 2
lines[0] = JSON.stringify(header)
await writeFile(path, lines.join('\n'))
// Fresh backend (no in-memory state) → must reject on load.
const ctx2 = new Context()
await ctx2.plugin(SessionStore)
await ctx2.plugin(SessionPersistenceJsonl, { root })
await expect(ctx2.sessionPersistence.load(m.id)).rejects.toThrow(/version/)
await ctx2.fiber.dispose()
})
it('rejects a re-append of an already-stored seq', async () => {
const m = meta('reappend')
await ctx.sessionPersistence.create(m)
@@ -293,62 +268,6 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => {
})
describe('SessionPersistenceJsonl: write path (session/event → flush)', () => {
it('persists a live session driven through the store, surviving reload', async () => {
root = await freshRoot()
const ctx = new Context()
await ctx.plugin(SessionStore)
await ctx.plugin(SessionPersistenceJsonl, { root })
const session = ctx.sessions.create('live', { meta: { cwd: '/w' } })
for (const e of oneTurnLog()) session.append(e.type, e.data)
await ctx.parallel('session/flush', session)
const loaded = await ctx.sessionPersistence.load(SessionId('live'))
expect(loaded.events).toHaveLength(6)
expect(loaded.meta.cwd).toBe('/w')
await ctx.fiber.dispose()
})
it('snapshot-on-buffer: mutating an event after session/event does not corrupt the persisted copy', async () => {
root = await freshRoot()
const ctx = new Context()
await ctx.plugin(SessionStore)
await ctx.plugin(SessionPersistenceJsonl, { root })
const session = ctx.sessions.create('mutate')
const ev = session.append('user/message', { content: [{ type: 'text', text: 'original' }], source: { kind: 'user' } })
// Mutate the live event object AFTER it was buffered.
;(ev.data as { content: { type: 'text'; text: string }[] }).content[0]!.text = 'HACKED'
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
await ctx.parallel('session/flush', session)
const loaded = await ctx.sessionPersistence.load(SessionId('mutate'))
const first = loaded.events[0]
expect(first?.type === 'user/message' && (first.data.content[0] as { text: string }).text).toBe('original')
await ctx.fiber.dispose()
})
it('fork: a seeded new session persists its seed once', async () => {
root = await freshRoot()
const ctx = new Context()
await ctx.plugin(SessionStore)
await ctx.plugin(SessionPersistenceJsonl, { root })
const seed = oneTurnLog()
// A fork: a brand-new id whose seed came from elsewhere.
const forked = ctx.sessions.create('forked', { seed })
// onCreated persisted the seed asynchronously; wait a tick.
await new Promise(r => setTimeout(r, 10))
const loaded = await ctx.sessionPersistence.load(SessionId('forked'))
expect(loaded.events).toEqual(seed)
// A flush with no NEW events must not double-write.
await ctx.parallel('session/flush', forked)
const reloaded = await ctx.sessionPersistence.load(SessionId('forked'))
expect(reloaded.events).toEqual(seed)
await ctx.fiber.dispose()
})
it('concurrent sessions do not cross buffers', async () => {
root = await freshRoot()
const ctx = new Context()
@@ -373,112 +292,6 @@ describe('SessionPersistenceJsonl: write path (session/event → flush)', () =>
await ctx.fiber.dispose()
})
it('HMR: applying the plugin seeds existing live sessions', async () => {
root = await freshRoot()
const ctx = new Context()
await ctx.plugin(SessionStore)
// A session exists BEFORE the persistence plugin is applied.
const session = ctx.sessions.create('pre-existing')
session.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } })
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
await ctx.plugin(SessionPersistenceJsonl, { root })
// The plugin seeded it on apply; a subsequent flush persists its events.
await ctx.parallel('session/flush', session)
const loaded = await ctx.sessionPersistence.load(SessionId('pre-existing'))
expect(loaded.events.length).toBeGreaterThanOrEqual(2)
await ctx.fiber.dispose()
})
it('HMR: dispose drains remaining buffers', async () => {
root = await freshRoot()
const ctx = new Context()
await ctx.plugin(SessionStore)
let session!: Session
const fiber = await ctx.plugin(SessionPersistenceJsonl, { root })
const sessFiber = await ctx.plugin(Object.assign((inner: Context) => {
session = inner.sessions.create('drain')
}, { inject: ['sessions'] }))
session.append('user/message', { content: [{ type: 'text', text: 'buffered' }], source: { kind: 'user' } })
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
// No explicit flush — dispose must drain.
await fiber.dispose()
await sessFiber.dispose()
// A fresh backend reads what the disposed one drained.
const ctx2 = new Context()
await ctx2.plugin(SessionStore)
await ctx2.plugin(SessionPersistenceJsonl, { root })
const loaded = await ctx2.sessionPersistence.load(SessionId('drain'))
expect(loaded.events.length).toBeGreaterThanOrEqual(2)
await ctx2.fiber.dispose()
})
it('HMR: reloading the backend adopts a still-live, already-materialized session', async () => {
root = await freshRoot()
const ctx = new Context()
await ctx.plugin(SessionStore)
// The session lives in its OWN fiber so it survives the backend reload.
let session!: Session
await ctx.plugin(Object.assign((inner: Context) => {
session = inner.sessions.create('hmr-adopt')
}, { inject: ['sessions'] }))
// Backend instance 1 materializes the session on disk.
const backend1 = await ctx.plugin(SessionPersistenceJsonl, { root })
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
session.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } })
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
await ctx.parallel('session/flush', session)
// Hot-reload the backend: dispose instance 1, plug in instance 2 over the
// SAME root while the session stays live. Instance 2 has an empty states
// map but the log is on disk — it must ADOPT (not reject) so flush keeps
// working. A second turn appended after reload then persists.
await backend1.dispose()
await ctx.plugin(SessionPersistenceJsonl, { root })
session.append('turn/start', { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } })
session.append('user/message', { content: [{ type: 'text', text: 'again' }], source: { kind: 'user' } })
session.append('turn/end', { turn: 2, reason: { kind: 'completed' } })
await expect(ctx.parallel('session/flush', session)).resolves.not.toThrow()
const loaded = await ctx.sessionPersistence.load(SessionId('hmr-adopt'))
expect(loaded.events.filter(e => e.type === 'turn/start')).toHaveLength(2)
await ctx.fiber.dispose()
})
it('HMR: adoption persists the live SUFFIX that was ahead of the on-disk prefix', async () => {
root = await freshRoot()
const ctx = new Context()
await ctx.plugin(SessionStore)
let session!: Session
await ctx.plugin(Object.assign((inner: Context) => {
session = inner.sessions.create('hmr-suffix')
}, { inject: ['sessions'] }))
// Instance 1 flushes turn 1 to disk.
const backend1 = await ctx.plugin(SessionPersistenceJsonl, { root })
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
await ctx.parallel('session/flush', session)
// Append turn 2 to the LIVE session, then dispose instance 1 WITHOUT
// flushing turn 2. Turn 2 is now ONLY in the live session's events; the new
// backend never buffered it via session/event.
await backend1.dispose()
session.append('turn/start', { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } })
session.append('turn/end', { turn: 2, reason: { kind: 'completed' } })
// Instance 2 adopts the on-disk prefix (turn 1) and MUST also persist the
// live suffix (turn 2) carried in the session's events — otherwise turn 2 is
// lost and a later flush would mismatch.
await ctx.plugin(SessionPersistenceJsonl, { root })
await ctx.parallel('session/flush', session)
const loaded = await ctx.sessionPersistence.load(SessionId('hmr-suffix'))
expect(loaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3])
expect(loaded.events.filter(e => e.type === 'turn/start')).toHaveLength(2)
await ctx.fiber.dispose()
})
})
@@ -570,17 +383,6 @@ describe('SessionPersistenceJsonl: edge cases', () => {
})
afterEach(async () => { await ctx.fiber.dispose() })
it('load rejects a missing session', async () => {
await expect(ctx.sessionPersistence.load(SessionId('nope'))).rejects.toThrow(/not found/)
})
it('append of an empty batch is a no-op', async () => {
const m = meta('empty-batch')
await ctx.sessionPersistence.create(m)
await ctx.sessionPersistence.append(m.id, [])
expect(await ctx.sessionPersistence.has(m.id)).toBe(false)
})
it('append rejects non-JSON-serializable undefined-producing data', async () => {
const m = meta('undef')
await ctx.sessionPersistence.create(m)
@@ -589,60 +391,6 @@ describe('SessionPersistenceJsonl: edge cases', () => {
await expect(ctx.sessionPersistence.append(m.id, bad)).rejects.toThrow(/non-JSON-serializable/)
})
it('delete of a non-existent session is a no-op', async () => {
await expect(ctx.sessionPersistence.delete(SessionId('ghost'))).resolves.toBeUndefined()
})
it('an abandoned lazy session (never materialized) releases its id for reuse', async () => {
// A live session is created then disposed BEFORE its first append: cursor 0,
// never materialized, nothing on disk. A new live session reusing the id
// must reclaim it (lazy materialization promises no lingering artifact),
// not wedge on an "already bound" collision until restart.
const backend = ctx.sessionPersistence as unknown as { inits: Map<Session, Promise<void>> }
let firstSession!: Session
const firstFiber = await ctx.plugin(Object.assign((inner: Context) => {
firstSession = inner.sessions.create('abandoned', { meta: { cwd: '/a' } })
}, { inject: ['sessions'] }))
await backend.inits.get(firstSession) // let the lazy create register the state
await firstFiber.dispose() // disposed before any append → never materialized
let reuse!: Session
await ctx.plugin(Object.assign((inner: Context) => {
reuse = inner.sessions.create('abandoned', { meta: { cwd: '/a' } })
}, { inject: ['sessions'] }))
// The new session claims the id without error and can persist a turn.
await expect(backend.inits.get(reuse)).resolves.toBeUndefined()
reuse.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
reuse.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
await ctx.parallel('session/flush', reuse)
const loaded = await ctx.sessionPersistence.load(SessionId('abandoned'))
expect(loaded.events.map(e => e.seq)).toEqual([0, 1])
})
it('does NOT reclaim an id whose abandoned owner still has buffered (unflushed) events', async () => {
// A session that appended events but was disposed BEFORE its first flush is
// not materialized yet but still holds a write-behind buffer. Reusing the id
// must be rejected (not reclaimed), or the stale buffer would drain against
// the new session — persisting old events under the new id or dropping the
// new session's seq-0 events.
const backend = ctx.sessionPersistence as unknown as { inits: Map<Session, Promise<void>> }
let first!: Session
const firstFiber = await ctx.plugin(Object.assign((inner: Context) => {
first = inner.sessions.create('buffered', { meta: { cwd: '/a' } })
}, { inject: ['sessions'] }))
await backend.inits.get(first)
// Append a turn but do NOT flush — events sit in the write-behind buffer.
first.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
first.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
await firstFiber.dispose() // disposed before flush; not materialized, buffer pending
let reuse!: Session
await ctx.plugin(Object.assign((inner: Context) => {
reuse = inner.sessions.create('buffered', { meta: { cwd: '/a' } })
}, { inject: ['sessions'] }))
await expect(backend.inits.get(reuse)).rejects.toThrow(/already bound to a different live session/)
})
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)
@@ -713,81 +461,6 @@ describe('SessionPersistenceJsonl: edge cases', () => {
await ctx2.fiber.dispose()
})
it('resume/adopt: a live session whose id is already on disk continues from the stored length', async () => {
// First lifecycle: persist a session through the store.
const s1 = ctx.sessions.create('resumed', { meta: { cwd: '/r' } })
for (const e of oneTurnLog()) s1.append(e.type, e.data)
await ctx.parallel('session/flush', s1)
// Second lifecycle: a NEW backend + a session re-created with the same id
// and SEEDED with the loaded events (the resume path). onCreated must adopt
// the on-disk log (not re-persist the seed), and a new turn appends at seq 6.
const ctx2 = new Context()
await ctx2.plugin(SessionStore)
await ctx2.plugin(SessionPersistenceJsonl, { root })
const loaded = await ctx2.sessionPersistence.load(SessionId('resumed'))
const s2 = ctx2.sessions.create('resumed', { seed: loaded.events, meta: { cwd: '/r' } })
await new Promise(r => setTimeout(r, 10)) // let onCreated adopt
// Append a fresh turn through the live session.
s2.append('turn/start', { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } })
s2.append('turn/end', { turn: 2, reason: { kind: 'completed' } })
await ctx2.parallel('session/flush', s2)
const reloaded = await ctx2.sessionPersistence.load(SessionId('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])
await ctx2.fiber.dispose()
})
it('HMR adoption does not crash-repair an active open turn as interrupted', async () => {
const dir = await freshRoot()
const hmr = new Context()
await hmr.plugin(SessionStore)
const first = await hmr.plugin(SessionPersistenceJsonl, { root: dir })
const session = hmr.sessions.create('hmr-open', { meta: { cwd: '/hmr' } })
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
session.append('step/start', { turn: 1, step: 1 })
await hmr.parallel('session/flush', session)
await first.dispose()
await appendFile(logPath(dir, '/hmr', SessionId('hmr-open')), '{"torn":')
const second = await hmr.plugin(SessionPersistenceJsonl, { root: dir })
session.append('step/end', { turn: 1, step: 1 })
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
await hmr.parallel('session/flush', session)
const loaded = await hmr.sessionPersistence.load(SessionId('hmr-open'))
expect(loaded.events.map(e => e.type)).toEqual(['turn/start', 'step/start', 'step/end', 'turn/end'])
expect(loaded.events.at(-1)).toMatchObject({ type: 'turn/end', data: { reason: { kind: 'completed' } } })
await second.dispose()
await hmr.fiber.dispose()
})
it('a NEW live session whose id collides with an on-disk log is rejected, not silently adopted', async () => {
// Persist a session on disk.
const s1 = ctx.sessions.create('collide', { meta: { cwd: '/a' } })
for (const e of oneTurnLog()) s1.append(e.type, e.data)
await ctx.parallel('session/flush', s1)
const before = await readFile(logPath(root, '/a', SessionId('collide')), 'utf8')
// A FRESH backend + a NEW live session with the same id but NO explicit
// load/resume. onCreated must NOT adopt-from-disk (resume is explicit); it
// treats this as a new session and create() rejects because a log already
// exists on disk. The rejection surfaces via the init promise (flush awaits
// it); the on-disk committed log is left byte-for-byte intact.
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>> }
const s2 = ctx2.sessions.create('collide', { meta: { cwd: '/a' } })
// The init for the new live session rejects (observed via the per-session
// init map and, in production, via flush which awaits the same promise).
await expect(backend.inits.get(s2)).rejects.toThrow(/already has a persisted log on disk/)
// The committed log is untouched (no clobber).
expect(await readFile(logPath(root, '/a', SessionId('collide')), 'utf8')).toBe(before)
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) => {
@@ -811,98 +484,6 @@ describe('SessionPersistenceJsonl: edge cases', () => {
await expect(backend.inits.get(b)).rejects.toThrow(/already bound to a different live session|already has a persisted log on disk/)
})
it('a live session claims cursor-0 ownerless state created via the public API', async () => {
// create() registers ownerless state with cursor 0 (lazy, nothing persisted
// yet). A live session with that id then arrives and claims it without a
// prefix check (cursor 0 matches trivially), persisting its seed.
await ctx.sessionPersistence.create(meta('lazy-claim', '/a'))
const backend = ctx.sessionPersistence as unknown as { inits: Map<Session, Promise<void>> }
let live!: Session
await ctx.plugin(Object.assign((inner: Context) => {
live = inner.sessions.create('lazy-claim', { meta: { cwd: '/a' } })
}, { inject: ['sessions'] }))
await expect(backend.inits.get(live)).resolves.toBeUndefined()
live.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
live.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
await ctx.parallel('session/flush', live)
const loaded = await ctx.sessionPersistence.load(SessionId('lazy-claim'))
expect(loaded.events.map(e => e.seq)).toEqual([0, 1])
})
it('a fresh session reusing a previously-loaded id is rejected (ownerless guard)', async () => {
// Materialize a log, then load() it into the backend's state WITHOUT a live
// session — leaving state.owner undefined and cursor at the persisted length
// (the public preview path).
await ctx.sessionPersistence.create(meta('preview', '/a'))
await ctx.sessionPersistence.append(SessionId('preview'), oneTurnLog())
await ctx.sessionPersistence.load(SessionId('preview'))
const backend = ctx.sessionPersistence as unknown as { inits: Map<Session, Promise<void>> }
// A FRESH (empty-seed) live session reusing that id must be rejected: its
// seq 0..cursor-1 events would otherwise be filtered as already-persisted
// and its conversation grafted onto the old log.
let fresh!: Session
await ctx.plugin(Object.assign((inner: Context) => {
fresh = inner.sessions.create('preview', { meta: { cwd: '/a' } })
}, { inject: ['sessions'] }))
await expect(backend.inits.get(fresh)).rejects.toThrow(/do not match this live session|already has a persisted log/)
})
it('a session whose seed matches the loaded prefix claims ownerless state', async () => {
// Materialize a log and load it (ownerless state, cursor = 6).
await ctx.sessionPersistence.create(meta('match', '/a'))
await ctx.sessionPersistence.append(SessionId('match'), oneTurnLog())
await ctx.sessionPersistence.load(SessionId('match'))
const backend = ctx.sessionPersistence as unknown as { inits: Map<Session, Promise<void>> }
// A live session SEEDED with the persisted log legitimately continues it —
// its seed reproduces the loaded prefix, so it claims the ownerless state.
let cont!: Session
await ctx.plugin(Object.assign((inner: Context) => {
cont = inner.sessions.create('match', { seed: oneTurnLog(), meta: { cwd: '/a' } })
}, { inject: ['sessions'] }))
await expect(backend.inits.get(cont)).resolves.toBeUndefined()
})
it('claiming ownerless state persists the seed suffix beyond the prefix', async () => {
// Materialize a one-turn log and load it (ownerless state, cursor = 6).
await ctx.sessionPersistence.create(meta('suffix-claim', '/a'))
await ctx.sessionPersistence.append(SessionId('suffix-claim'), oneTurnLog())
await ctx.sessionPersistence.load(SessionId('suffix-claim'))
const backend = ctx.sessionPersistence as unknown as { inits: Map<Session, Promise<void>> }
// A live session seeded with the prefix PLUS a second turn (seqs 6,7). The
// suffix (constructor seed, never emits session/event) must be persisted on
// claim, not lost.
const seed = [
...oneTurnLog(),
{ 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' } } },
] as SessionEvent[]
let cont!: Session
await ctx.plugin(Object.assign((inner: Context) => {
cont = inner.sessions.create('suffix-claim', { seed, meta: { cwd: '/a' } })
}, { inject: ['sessions'] }))
await backend.inits.get(cont)
const loaded = await ctx.sessionPersistence.load(SessionId('suffix-claim'))
expect(loaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7])
})
it('claiming cursor-0 ownerless state persists the whole constructor seed', async () => {
// create() registers ownerless state with cursor 0 (lazy, nothing on disk).
await ctx.sessionPersistence.create(meta('lazy-seed', '/a'))
const backend = ctx.sessionPersistence as unknown as { inits: Map<Session, Promise<void>> }
// A live session seeded with a full turn claims it; the whole seed (cursor
// is 0) must be persisted.
let cont!: Session
await ctx.plugin(Object.assign((inner: Context) => {
cont = inner.sessions.create('lazy-seed', { seed: oneTurnLog(), meta: { cwd: '/a' } })
}, { inject: ['sessions'] }))
await backend.inits.get(cont)
const loaded = await ctx.sessionPersistence.load(SessionId('lazy-seed'))
expect(loaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5])
})
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'))
@@ -942,14 +523,6 @@ describe('SessionPersistenceJsonl: edge cases', () => {
.rejects.toThrow(/already bound to a different live session|already has a persisted log|do not match/)
})
it('round-trips a header with parentSession (fork lineage)', async () => {
const m: SessionHeader = { version: 1, id: SessionId('forked-child'), createdAt: 1, parentSession: SessionId('the-parent') }
await ctx.sessionPersistence.create(m)
await ctx.sessionPersistence.append(m.id, oneTurnLog())
const loaded = await ctx.sessionPersistence.load(m.id)
expect(loaded.meta.parentSession).toBe('the-parent')
})
it('list returns nothing when the root directory does not exist', async () => {
const ctx2 = new Context()
await ctx2.plugin(SessionStore)
@@ -1025,48 +598,6 @@ describe('SessionPersistenceJsonl: edge cases', () => {
expect(end.type === 'turn/end' && end.data.reason).toEqual({ kind: 'interrupted' })
})
it('initFor is idempotent: a re-seeded existing session is not re-initialized', async () => {
const session = ctx.sessions.create('idem', { meta: { cwd: '/i' } })
session.append('user/message', { content: [{ type: 'text', text: 'x' }], source: { kind: 'user' } })
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
await ctx.parallel('session/flush', session)
// Re-emit session/created for the SAME live session (idempotent initFor).
ctx.emit('session/created', session)
await ctx.parallel('session/flush', session)
const loaded = await ctx.sessionPersistence.load(SessionId('idem'))
expect(loaded.events).toHaveLength(2) // not doubled
})
it('flush before init resolves with no state uses cursor 0', async () => {
// Drive a fork (seed) flush where the buffer holds the seed; the fresh
// events filter against cursor. Exercises the state-undefined cursor path.
const session = ctx.sessions.create('flush-nostate')
// Append directly to the live session and flush IMMEDIATELY, before the
// async onCreated init has necessarily set state.
session.append('user/message', { content: [{ type: 'text', text: 'q' }], source: { kind: 'user' } })
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
await ctx.parallel('session/flush', session)
const loaded = await ctx.sessionPersistence.load(SessionId('flush-nostate'))
expect(loaded.events).toHaveLength(2)
})
it('createCore rejects creating an id this backend already tracks', async () => {
await ctx.sessionPersistence.create(meta('dup'))
await expect(ctx.sessionPersistence.create(meta('dup'))).rejects.toThrow(/already exists in this backend/)
})
it('createCore rejects creating an id whose log already exists on disk', async () => {
const m = meta('on-disk', '/od')
await ctx.sessionPersistence.create(m)
await ctx.sessionPersistence.append(m.id, oneTurnLog())
// A fresh backend (no in-memory state) must refuse to create over the log.
const ctx2 = new Context()
await ctx2.plugin(SessionStore)
await ctx2.plugin(SessionPersistenceJsonl, { root })
await expect(ctx2.sessionPersistence.create(meta('on-disk', '/od'))).rejects.toThrow(/already has a persisted log on disk/)
await ctx2.fiber.dispose()
})
it('createCore rejects an id already on disk under a DIFFERENT cwd bucket', async () => {
// Persist the id under cwd A.