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

Merge latest invariant registration gate

Browse Source 
# Conflicts:
#	docs/event-producer-consumer.md
This commit is contained in:
Tianyi Cui
2026-07-20 20:20:10 +08:00
parent 9cbe705b15 5f806b96f0
commit 8ee54bcb22
184 changed files with 1882 additions and 364 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,31 +1,39 @@
# @deepseek-ai/dsh-session-persistence-jsonl
The JSONL durable session-persistence backend — a concrete `SessionPersistence` (the `dsh-session-persistence` seam). One append-only `.jsonl` event log per session.
The JSONL durable session-persistence backend — a concrete `SessionPersistence` (the `dsh-session-persistence` seam). Each session has one append-only logical JSONL log, stored as `.jsonl.zstd` by default or raw `.jsonl` when compression is disabled.
## On-disk layout
```
<root>/
cwd-<sha256(cwd)[:12]>/ # per-project bucket (or _no-cwd/ when no cwd)
<encoded-id>.jsonl # header line + one SessionEvent per line (verbatim)
<encoded-id>.jsonl.zstd # default: checksummed header frame + append frames
<encoded-id>.jsonl # only with compression: 'none'
```
- The first `.jsonl` line is the immutable `SessionHeader` tagged `{ type: 'session', version, id, cwd?, createdAt, parentSession?, seedLength? }`; every subsequent line is one `SessionEvent` JSON, **verbatim including `assistant/chunk`** so `seq` stays contiguous (`events[i].seq === i`).
- Session ids are unvalidated branded strings, so they are percent-encoded to a single safe path segment before use (no traversal, no collision).
- The first logical line is the immutable `SessionHeader` tagged `{ type: 'session', version, id, cwd?, createdAt, parentSession?, seedLength?, delegationDepth }`. `delegationDepth` is required on disk and is `0` for a top-level session; a missing or invalid value rejects the log. Every subsequent line is one `SessionEvent` JSON, **verbatim including `assistant/chunk`** so `seq` stays contiguous (`events[i].seq === i`).
- Session ids are unvalidated branded strings, so they are injectively escaped to a single safe path segment before use (no traversal, no collision).
## Config
| Key | Type | Notes |
|---|---|---|
| `root` | `string` (required) | Root directory for all session files. **No default** — a `process.cwd()` default would scatter files as the process's cwd changes (bash calls, subprocesses). |
| `compression` | `'zstd' \| 'none'` | Defaults to `'zstd'`; `'none'` retains newline-delimited UTF-8 text. |
`locate(meta)` returns `{ kind: 'jsonl', path }` using the resolved absolute root and the same cwd-bucket/id encoding as materialization. It performs no filesystem I/O: the target can be returned before the file exists, and an existing file contains only the last flushed prefix.
## Physical encoding
The default artifact is a standard concatenation of independent [Zstandard frames](../../../.agents/notes/implemented/architecture/2026-07-19-zstandard-jsonl-session-logs.md): one checksummed frame containing only the header line, followed by one checksummed frame per durable append batch. The backend uses Node's built-in Zstandard API with its default compression level and exposes no level knob. Listing reads and validates only the header frame. `compression: 'none'` keeps the same logical lines in the original raw representation.
A root belongs to one encoding. Startup discovery and targeted lookup reject the opposite suffix with an error naming the incompatible artifact and instructing the caller to select the matching mode or a separate root. There is no migration, mixed-root fallback, or dual write.
## Durability and crash semantics
- **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 when the host supports it. 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 — 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](../../../.agents/notes/implemented/architecture/2026-06-14-session-persistence.md); the same defect at or before the last committed `turn/end` rejects.
- **Append-only.** Committed events (at or below a flushed `turn/end`) are never rewritten. Subsequent raw batches append lines; compressed batches append one frame. Both paths `fsync`, and a caught write or sync failure rolls the file back to its prior byte length.
- **Crash recovery — preserve valid tail work.** `load` validates every complete compressed frame and scans their decompressed JSONL. If the last frame is structurally incomplete, the reader keeps its complete decoded records, truncates from that frame's start, and re-encodes those records with the synthetic tool, step, and turn closers required by the shared [persistence contract](../../../.agents/notes/implemented/architecture/2026-06-14-session-persistence.md). Raw mode truncates from its first incomplete line. A checksum/decompression failure in a complete frame, or a defect at or before the last committed `turn/end`, is corruption and 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
@@ -50,7 +58,8 @@ JSONL storage does not mutate live request prefixes. A resumed loop can reuse pr
## Known Limitations and Deferred Work
- **Only the current `SESSION_FORMAT_VERSION` (v0) loads** — the on-disk format is pre-release/unstable: a breaking format change is absorbed at v0 and non-current logs are rejected; there is no migration.
- **Only the configured encoding and current `SESSION_FORMAT_VERSION` (v0) load** — changing compression requires a separate/fresh root or selecting the legacy raw mode; the pre-release format has no migration.
- **Compressed files are not directly line-readable** — use the backend to load them, or select `compression: 'none'` before writing a fresh root when text fixtures or external line readers are required.
- **Nothing deletes session files** — logs accumulate under `root` until removed externally (the seam has no deletion surface).
- **Single-process assumption** — per-session serialization and the write cursor live in this process; two processes appending to the same `root` are not coordinated.
- **Initial materialization requires hard-link support** — first append uses `link()` so same-id races fail instead of overwriting a committed log; a filesystem that cannot create hard links cannot host this backend.

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

@@ -12,8 +12,20 @@ import { createHash } from 'node:crypto'
import { join } from 'node:path'
import type { SessionEvent, SessionHeader, SessionId } from '@deepseek-ai/dsh-session'
/** Physical encoding selected for JSONL session artifacts. */
export type JsonlCompression = 'zstd' | 'none'
/**
* The first line of a session's `.jsonl` file: the immutable
* Return the artifact suffix for one physical encoding.
* @param compression - configured JSONL artifact encoding.
* @returns `.jsonl.zstd` for Zstandard or `.jsonl` for plaintext.
*/
export function logSuffix(compression: JsonlCompression): '.jsonl.zstd' | '.jsonl' {
return compression === 'zstd' ? '.jsonl.zstd' : '.jsonl'
}
/**
* The first JSONL record of a session artifact: the immutable
* {@link SessionHeader} tagged as a `session` record so a reader can tell it
* apart from an event line.
*/
@@ -25,6 +37,7 @@ export interface HeaderLine {
cwd?: string
parentSession?: SessionId
seedLength?: number
delegationDepth: number
}
/**
@@ -41,6 +54,7 @@ export function toHeaderLine(header: SessionHeader): HeaderLine {
...header.cwd !== undefined ? { cwd: header.cwd } : {},
...header.parentSession !== undefined ? { parentSession: header.parentSession } : {},
...header.seedLength !== undefined ? { seedLength: header.seedLength } : {},
delegationDepth: header.delegationDepth ?? 0,
}
}
@@ -57,6 +71,7 @@ export function fromHeaderLine(line: HeaderLine): SessionHeader {
...line.cwd !== undefined ? { cwd: line.cwd } : {},
...line.parentSession !== undefined ? { parentSession: line.parentSession } : {},
...line.seedLength !== undefined ? { seedLength: line.seedLength } : {},
delegationDepth: line.delegationDepth,
}
}
@@ -68,6 +83,10 @@ function isHeaderLine(value: unknown): value is HeaderLine {
&& typeof (value as { version?: unknown }).version === 'number'
&& typeof (value as { id?: unknown }).id === 'string'
&& typeof (value as { createdAt?: unknown }).createdAt === 'number'
&& typeof (value as { delegationDepth?: unknown }).delegationDepth === 'number'
&& Number.isSafeInteger((value as { delegationDepth: number }).delegationDepth)
&& (value as { delegationDepth: number }).delegationDepth >= 0
&& !Object.is((value as { delegationDepth: number }).delegationDepth, -0)
)
}
@@ -119,10 +138,16 @@ export function sessionDir(root: string, cwd: string | undefined): string {
* @param root - the backend's session root directory.
* @param cwd - the session's project directory (picks the per-cwd bucket; `undefined` → `_no-cwd`).
* @param id - the session id, path-encoded via {@link encodeSegment} before filesystem use.
* @returns the session's `.jsonl` log file path.
* @param compression - physical artifact encoding and filename suffix.
* @returns the session's configured JSONL artifact path.
*/
export function logPath(root: string, cwd: string | undefined, id: SessionId): string {
return join(sessionDir(root, cwd), `${encodeSegment(id)}.jsonl`)
export function logPath(
root: string,
cwd: string | undefined,
id: SessionId,
compression: JsonlCompression,
): string {
return join(sessionDir(root, cwd), `${encodeSegment(id)}${logSuffix(compression)}`)
}
/**

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

@@ -17,8 +17,20 @@ import {
} from '@deepseek-ai/dsh-session-persistence'
import type { SessionEvent, SessionId, SessionHeader } from '@deepseek-ai/dsh-session'
import {
encodeSegment, eventLine, logPath, parseHeaderMeta, scanLog, sessionDir, toHeaderLine,
encodeSegment, eventLine, logPath, logSuffix, parseHeaderMeta, scanLog, sessionDir, toHeaderLine,
type JsonlCompression,
} from './format.ts'
import { compressZstdFrame, decompressZstdFrame, scanZstdFrames } from './zstd.ts'
export type { JsonlCompression } from './format.ts'
const DEFAULT_COMPRESSION: JsonlCompression = 'zstd'
/** Loader schema for the JSONL artifact's physical encoding. */
export const JsonlCompressionSchema: z<JsonlCompression> = z.union([
z.const('zstd'),
z.const('none'),
]).default(DEFAULT_COMPRESSION)
/** Plugin config: where the JSONL backend keeps its session logs (`root` is required — no default). */
export interface Config {
@@ -28,6 +40,14 @@ export interface Config {
* (bash calls, subprocesses). Sessions group under per-cwd subdirectories.
*/
root: string
/** Physical encoding; defaults to checksummed Zstandard frames. */
compression?: JsonlCompression
}
/** Opaque coordinator token for replacing bytes recovered from a torn frame. */
interface JsonlTornMarker {
truncateTo: number
recoveredEvents: SessionEvent[]
}
/** Whether a filesystem error means absence; every non-ENOENT failure must surface. */
@@ -38,13 +58,15 @@ function isENOENT(error: unknown): boolean {
/**
* The JSONL persistence backend. Load as a plugin; it registers as
* `ctx.sessionPersistence` and (via the coordinator) installs the write-path
* listeners. Its torn-tail marker is the byte offset to truncate the log to.
* listeners. Its torn-tail marker carries the byte offset and any events
* recovered from an incomplete final Zstandard frame.
*/
export class SessionPersistenceJsonl extends SessionPersistence implements PersistenceBackend<number> {
export class SessionPersistenceJsonl extends SessionPersistence implements PersistenceBackend<JsonlTornMarker> {
static inject = ['sessions']
static Config: z<Config> = z.object({
root: z.string().required(),
compression: JsonlCompressionSchema,
})
/**
@@ -55,7 +77,9 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
override readonly name = 'session-persistence-jsonl'
private root: string
private coordinator: PersistenceCoordinator<number>
private compression: JsonlCompression
private coordinator: PersistenceCoordinator<JsonlTornMarker>
private rootEncodingCheck: Promise<void> | undefined
/** Runtime host platform used to decide whether directory sync is supported. */
readonly internals: { platform: NodeJS.Platform } = { platform: process.platform }
@@ -64,7 +88,8 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
super(ctx)
// 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)
this.compression = config.compression ?? DEFAULT_COMPRESSION
this.coordinator = new PersistenceCoordinator<JsonlTornMarker>(this.ctx, this)
}
// Each backend keeps the typed service surface beside its storage hooks;
@@ -74,7 +99,7 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
/** Resolve the absolute target path without touching the filesystem. */
locate(meta: SessionHeader): SessionLocation {
return { kind: 'jsonl', path: logPath(this.root, meta.cwd, meta.id) }
return { kind: 'jsonl', path: logPath(this.root, meta.cwd, meta.id, this.compression) }
}
create(meta: SessionHeader): Promise<void> {
@@ -96,7 +121,8 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
// --- PersistenceBackend hooks (the file-bytes storage primitives) ---
/** Read a stored prefix by id across all cwd buckets when cwd is unknown. */
async loadStored(id: SessionId): Promise<StoredPrefix<number> | undefined> {
async loadStored(id: SessionId): Promise<StoredPrefix<JsonlTornMarker> | undefined> {
await this.ensureRootEncoding()
const file = await this.findLog(id)
if (file === undefined) return undefined
return this.readPrefix(file.path)
@@ -106,28 +132,85 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
* 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)
if (!await this.exists(path)) return undefined
async loadLive(id: SessionId, cwd: string | undefined): Promise<StoredPrefix<JsonlTornMarker> | undefined> {
await this.ensureRootEncoding()
const path = logPath(this.root, cwd, id, this.compression)
if (!await this.exists(path)) {
await this.rejectOppositeArtifact(cwd, id)
return undefined
}
return this.readPrefix(path)
}
/**
* Read a stored prefix and convert torn-tail state to the byte offset the
* coordinator can round-trip without knowing the file format.
* Read a stored prefix and convert torn-tail state to the opaque marker the
* coordinator can round-trip without knowing the physical encoding.
*/
private async readPrefix(path: string): Promise<StoredPrefix<number>> {
private async readPrefix(path: string): Promise<StoredPrefix<JsonlTornMarker>> {
const buffer = await readFile(path)
if (this.compression === 'zstd') return this.readZstdPrefix(buffer)
const { meta, events, committedBytes } = scanLog(buffer)
return {
meta,
events,
...committedBytes < buffer.byteLength ? { tornMarker: committedBytes } : {},
...committedBytes < buffer.byteLength
? { tornMarker: { truncateTo: committedBytes, recoveredEvents: [] } }
: {},
}
}
/** Decode complete frames and retain complete JSONL records from a torn final frame. */
private async readZstdPrefix(buffer: Buffer): Promise<StoredPrefix<JsonlTornMarker>> {
const { frames, tornStart } = scanZstdFrames(buffer)
if (frames.length === 0) throw new Error('empty or header-less Zstandard session log')
const plaintextFrames: Buffer[] = []
for (const frame of frames) {
try {
plaintextFrames.push(await decompressZstdFrame(buffer.subarray(frame.start, frame.end)))
} catch (error) {
throw new Error(`corrupt Zstandard session log: frame at byte ${frame.start} failed validation`, { cause: error })
}
}
const headerFrame = plaintextFrames[0]
if (headerFrame === undefined || headerFrame.length === 0 || headerFrame.indexOf(0x0A) !== headerFrame.length - 1) {
throw new Error('corrupt Zstandard session log: first frame is not exactly one header line')
}
const completePlaintext = Buffer.concat(plaintextFrames)
const completePrefix = scanLog(completePlaintext)
if (completePrefix.committedBytes !== completePlaintext.length) {
throw new Error('corrupt Zstandard session log: complete frame contains a torn JSONL record')
}
if (tornStart === undefined) {
return { meta: completePrefix.meta, events: completePrefix.events }
}
let recoveredPlaintext: Buffer = Buffer.alloc(0)
try {
recoveredPlaintext = await decompressZstdFrame(buffer.subarray(tornStart))
} catch {
// A structurally incomplete final frame may end before Node's decoder can
// emit any plaintext; the complete prior frames remain recoverable.
}
const recoveredPrefix = scanLog(Buffer.concat([completePlaintext, recoveredPlaintext]))
/* v8 ignore next 3 -- appending plaintext cannot shorten the already-scanned complete prefix */
if (recoveredPrefix.events.length < completePrefix.events.length) {
throw new Error('corrupt Zstandard session log: recovered prefix does not extend complete frames')
}
return {
meta: recoveredPrefix.meta,
events: recoveredPrefix.events,
tornMarker: {
truncateTo: tornStart,
recoveredEvents: recoveredPrefix.events.slice(completePrefix.events.length),
},
}
}
/** Durably append a batch, lazily materializing the file when not yet present. */
async appendBatch(meta: SessionHeader, events: readonly SessionEvent[], isMaterialized: boolean): Promise<void> {
await this.ensureRootEncoding()
if (isMaterialized) {
await this.appendLines(meta, events)
} else {
@@ -136,22 +219,30 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
}
/**
* 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.
* Make a crash repair durable: truncate a torn tail, restore complete events
* decoded from it, then append synthetic closers. 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)
async commitRepair(
meta: SessionHeader,
tornMarker: JsonlTornMarker | undefined,
closers: readonly SessionEvent[],
): Promise<void> {
if (tornMarker !== undefined) await this.repair(meta, tornMarker.truncateTo)
const repairedEvents = [...(tornMarker?.recoveredEvents ?? []), ...closers]
if (repairedEvents.length > 0) await this.appendLines(meta, repairedEvents)
}
/** List all stored sessions' metadata (header line only — no full-log parse). */
async list(): Promise<SessionHeader[]> {
await this.ensureRootEncoding()
const metas: SessionHeader[] = []
for (const dir of await this.listCwdDirs()) {
for (const name of await this.listJsonl(dir)) {
for (const name of await this.listArtifacts(dir)) {
// Read only headers so listing scales with session count, not log size.
const first = await this.readFirstLine(`${dir}/${name}`)
const first = this.compression === 'zstd'
? await this.readFirstZstdLine(`${dir}/${name}`)
: await this.readFirstLine(`${dir}/${name}`)
if (first === undefined) continue // empty/half-written file
const meta = parseHeaderMeta(first)
if (meta === undefined) continue // not a session header
@@ -170,15 +261,14 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
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)
const finalPath = logPath(this.root, meta.cwd, meta.id, this.compression)
// 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)`)
}
const header = JSON.stringify(toHeaderLine(meta))
const body = events.map(eventLine).join('\n')
const content = header + '\n' + body + '\n'
await this.rejectOppositeArtifact(meta.cwd, meta.id)
const content = await this.encodeMaterialization(meta, events)
const tmp = `${finalPath}.${randomBytes(6).toString('hex')}.tmp`
const handle = await open(tmp, 'wx', 0o600)
@@ -211,6 +301,22 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
}
}
/** Encode the header and first batch without combining their frame boundaries. */
private async encodeMaterialization(meta: SessionHeader, events: readonly SessionEvent[]): Promise<Buffer | string> {
const header = JSON.stringify(toHeaderLine(meta)) + '\n'
const body = events.map(eventLine).join('\n') + '\n'
if (this.compression === 'none') return header + body
const headerFrame = await compressZstdFrame(header)
const eventFrame = await compressZstdFrame(body)
return Buffer.concat([headerFrame, eventFrame])
}
/** Encode one durable append batch in the configured physical representation. */
private async encodeEventBatch(events: readonly SessionEvent[]): Promise<Buffer | string> {
const body = events.map(eventLine).join('\n') + '\n'
return this.compression === 'zstd' ? compressZstdFrame(body) : body
}
/** fsync a directory when the host exposes that durability primitive. */
private async syncDir(dir: string): Promise<void> {
const handle = await open(dir, 'r')
@@ -234,12 +340,13 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
* 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)
const content = await this.encodeEventBatch(events)
const path = logPath(this.root, meta.cwd, meta.id, this.compression)
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.writeFile(content)
await handle.sync()
} catch (error) {
// Roll back whatever bytes landed so a retry starts from a clean EOF.
@@ -254,7 +361,7 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
/** 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)
const path = logPath(this.root, meta.cwd, meta.id, this.compression)
await truncate(path, offset)
const handle = await open(path, 'r+')
try {
@@ -292,17 +399,47 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
}
}
/** Read and validate only the independently compressed header frame. */
private async readFirstZstdLine(path: string): Promise<string | undefined> {
const handle = await open(path, 'r')
try {
let content = Buffer.alloc(0)
const chunk = Buffer.alloc(8192)
for (;;) {
const { bytesRead } = await handle.read(chunk, 0, chunk.length, null)
if (bytesRead === 0) return undefined
content = Buffer.concat([content, chunk.subarray(0, bytesRead)])
const first = scanZstdFrames(content, 1).frames[0]
if (first === undefined) continue
let plaintext: Buffer
try {
plaintext = await decompressZstdFrame(content.subarray(first.start, first.end))
} catch (error) {
throw new Error('corrupt Zstandard session log: header frame failed validation', { cause: error })
}
if (plaintext.length === 0 || plaintext.indexOf(0x0A) !== plaintext.length - 1) {
throw new Error('corrupt Zstandard session log: first frame is not exactly one header line')
}
return plaintext.subarray(0, -1).toString('utf8')
}
} finally {
await handle.close()
}
}
/**
* 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'
const target = encodeSegment(id) + logSuffix(this.compression)
for (const dir of await this.listCwdDirs()) {
const path = `${dir}/${target}`
const opposite = `${dir}/${encodeSegment(id)}${logSuffix(this.oppositeCompression())}`
if (await this.exists(opposite)) throw this.encodingMismatch(opposite)
if (await this.exists(path)) {
// Recover the cwd from the header so the caller has the session's bucket.
const { meta } = scanLog(await readFile(path))
const { meta } = await this.readPrefix(path)
return { path, cwd: meta.cwd }
}
}
@@ -321,9 +458,45 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
}
}
private async listJsonl(dir: string): Promise<string[]> {
private async listArtifacts(dir: string): Promise<string[]> {
const entries = await readdir(dir)
return entries.filter(n => n.endsWith('.jsonl'))
const oppositeSuffix = logSuffix(this.oppositeCompression())
const incompatible = entries.find(name => name.endsWith(oppositeSuffix))
if (incompatible !== undefined) throw this.encodingMismatch(`${dir}/${incompatible}`)
const suffix = logSuffix(this.compression)
return entries.filter(name => name.endsWith(suffix))
}
/** Reject a root that already belongs to the other physical encoding. */
private ensureRootEncoding(): Promise<void> {
this.rootEncodingCheck ??= this.checkRootEncoding()
return this.rootEncodingCheck
}
private async checkRootEncoding(): Promise<void> {
const oppositeSuffix = logSuffix(this.oppositeCompression())
for (const dir of await this.listCwdDirs()) {
const entries = await readdir(dir)
const incompatible = entries.find(name => name.endsWith(oppositeSuffix))
if (incompatible !== undefined) throw this.encodingMismatch(`${dir}/${incompatible}`)
}
}
private async rejectOppositeArtifact(cwd: string | undefined, id: SessionId): Promise<void> {
const path = logPath(this.root, cwd, id, this.oppositeCompression())
if (await this.exists(path)) throw this.encodingMismatch(path)
}
private oppositeCompression(): JsonlCompression {
return this.compression === 'zstd' ? 'none' : 'zstd'
}
private encodingMismatch(path: string): Error {
return new Error(
`session artifact ${JSON.stringify(path)} uses ${logSuffix(this.oppositeCompression())}, `
+ `but this backend is configured for compression ${JSON.stringify(this.compression)}; `
+ 'use a separate root or select the matching compression mode',
)
}
private async exists(path: string): Promise<boolean> {

116
packages/session-persistence/session-persistence-jsonl/src/zstd.ts Normal file
View File

@@ -0,0 +1,116 @@
/**
* Zstandard frame primitives for the JSONL persistence backend. The backend
* owns a concatenated-frame container so it can append and recover batches
* without exposing compression mechanics through the persistence seam.
* @module dsh-session-persistence-jsonl/zstd
*/
import { constants, zstdCompress, zstdDecompress, type ZstdOptions } from 'node:zlib'
import { promisify } from 'node:util'
const ZSTD_MAGIC = 0xFD2FB528
const zstdCompressAsync = promisify(zstdCompress)
const zstdDecompressAsync = promisify(zstdDecompress)
const CHECKSUM_OPTIONS: ZstdOptions = {
params: { [constants.ZSTD_c_checksumFlag]: 1 },
}
/** Byte range occupied by one structurally complete Zstandard frame. */
export interface ZstdFrameRange {
/** Inclusive frame start. */
start: number
/** Exclusive frame end. */
end: number
}
/** Structural scan result for a concatenated Zstandard stream. */
export interface ZstdFrameScan {
/** Complete frames in file order. */
frames: ZstdFrameRange[]
/** Start of an incomplete final frame, when EOF interrupts one. */
tornStart?: number
}
/**
* Locate complete frames without decompressing their blocks. Invalid complete
* structure rejects; EOF inside the final frame returns its start for repair.
* @param buffer - complete bytes currently present in the session artifact.
* @param maxFrames - optional complete-frame limit for metadata-only readers.
* @returns complete frame ranges and an optional incomplete-final-frame start.
*/
export function scanZstdFrames(buffer: Buffer, maxFrames = Number.POSITIVE_INFINITY): ZstdFrameScan {
const frames: ZstdFrameRange[] = []
let offset = 0
while (offset < buffer.length) {
const start = offset
if (buffer.length - offset < 4) return { frames, tornStart: start }
if (buffer.readUInt32LE(offset) !== ZSTD_MAGIC) {
throw new Error(`corrupt Zstandard session log: invalid frame magic at byte ${offset}`)
}
offset += 4
if (offset === buffer.length) return { frames, tornStart: start }
const descriptor = buffer.readUInt8(offset)
offset += 1
if ((descriptor & 0x18) !== 0) {
throw new Error(`corrupt Zstandard session log: reserved frame-header bit at byte ${offset - 1}`)
}
const contentSizeFlag = descriptor >>> 6
const singleSegment = (descriptor & 0x20) !== 0
const checksum = (descriptor & 0x04) !== 0
const dictionaryFlag = descriptor & 0x03
const dictionaryBytes = dictionaryFlag === 3 ? 4 : dictionaryFlag
const contentSizeBytes = contentSizeFlag === 0
? (singleSegment ? 1 : 0)
: 1 << contentSizeFlag
const remainingHeaderBytes = (singleSegment ? 0 : 1) + dictionaryBytes + contentSizeBytes
if (buffer.length - offset < remainingHeaderBytes) return { frames, tornStart: start }
offset += remainingHeaderBytes
for (;;) {
if (buffer.length - offset < 3) return { frames, tornStart: start }
const blockHeader = buffer.readUIntLE(offset, 3)
offset += 3
const lastBlock = (blockHeader & 1) !== 0
const blockType = (blockHeader >>> 1) & 0x03
const blockSize = blockHeader >>> 3
if (blockType === 0x03) {
throw new Error(`corrupt Zstandard session log: reserved block type at byte ${offset - 3}`)
}
const payloadBytes = blockType === 0x01 ? 1 : blockSize
if (buffer.length - offset < payloadBytes) return { frames, tornStart: start }
offset += payloadBytes
if (lastBlock) break
}
if (checksum) {
if (buffer.length - offset < 4) return { frames, tornStart: start }
offset += 4
}
frames.push({ start, end: offset })
if (frames.length === maxFrames) return { frames }
}
return { frames }
}
/**
* Compress one independently decodable, checksummed Zstandard frame.
* @param input - JSONL bytes for a header or durable event batch.
* @returns the complete encoded frame.
*/
export async function compressZstdFrame(input: Buffer | string): Promise<Buffer> {
return zstdCompressAsync(input, CHECKSUM_OPTIONS)
}
/**
* Decompress one complete frame or the available prefix of a torn final frame.
* Complete-frame checksums are validated by Node's decoder.
* @param input - bytes beginning at a Zstandard frame boundary.
* @returns plaintext produced from the available input.
*/
export async function decompressZstdFrame(input: Buffer): Promise<Buffer> {
return zstdDecompressAsync(input)
}

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

@@ -38,6 +38,10 @@ async function freshRoot(): Promise<string> {
return dir
}
function rawLogPath(root: string, cwd: string | undefined, id: SessionId): string {
return logPath(root, cwd, id, 'none')
}
afterEach(async () => {
vi.restoreAllMocks()
for (const d of dirs.splice(0)) await rm(d, { recursive: true, force: true })
@@ -68,11 +72,11 @@ function appendClosedTurn(session: Session): void {
}
// Run the shared backend contract against the real JSONL backend.
runPersistenceContract('jsonl', async () => {
runPersistenceContract('jsonl-none', async () => {
const dir = await mkdtemp(join(tmpdir(), 'dsh-jsonl-'))
const ctx = new Context()
await ctx.plugin(SessionStore)
const fiber = await ctx.plugin(SessionPersistenceJsonl, { root: dir })
const fiber = await ctx.plugin(SessionPersistenceJsonl, { root: dir, compression: 'none' })
return {
persistence: ctx.sessionPersistence,
dispose: async () => {
@@ -84,18 +88,18 @@ runPersistenceContract('jsonl', async () => {
// 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> => {
runCoordinatorContract('jsonl-none', async (): Promise<CoordinatorFixture> => {
const dir = await mkdtemp(join(tmpdir(), 'dsh-jsonl-coord-'))
return {
mount: async (ctx) => {
const fiber = await ctx.plugin(SessionPersistenceJsonl, { root: dir })
const fiber = await ctx.plugin(SessionPersistenceJsonl, { root: dir, compression: 'none' })
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')
await appendFile(rawLogPath(dir, cwd, id), '{"type":"assistant/chunk","seq":8,"ti')
},
cleanup: async () => { await rm(dir, { recursive: true, force: true }) },
}
@@ -132,11 +136,14 @@ describe('SessionPersistenceJsonl: format helpers', () => {
const absoluteRoot = await freshRoot()
const ctx = new Context()
await ctx.plugin(SessionStore)
const fiber = await ctx.plugin(SessionPersistenceJsonl, { root: relative(process.cwd(), absoluteRoot) })
const fiber = await ctx.plugin(SessionPersistenceJsonl, {
root: relative(process.cwd(), absoluteRoot),
compression: 'none',
})
const m = meta('relative-location', '/work')
expect(ctx.sessionPersistence.locate(m)).toEqual({
kind: 'jsonl',
path: logPath(resolve(absoluteRoot), '/work', m.id),
path: rawLogPath(resolve(absoluteRoot), '/work', m.id),
})
await fiber.dispose()
})
@@ -148,26 +155,26 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => {
root = await freshRoot()
ctx = new Context()
await ctx.plugin(SessionStore)
await ctx.plugin(SessionPersistenceJsonl, { root })
await ctx.plugin(SessionPersistenceJsonl, { root, compression: 'none' })
})
afterEach(async () => { await ctx.fiber.dispose() })
it('lazy materialization: create() writes no file until the first append', async () => {
const m = meta('lazy', '/work')
const location = ctx.sessionPersistence.locate(m)
expect(location).toEqual({ kind: 'jsonl', path: logPath(root, '/work', m.id) })
expect(location).toEqual({ kind: 'jsonl', path: rawLogPath(root, '/work', m.id) })
expect(isAbsolute(location!.path)).toBe(true)
await ctx.sessionPersistence.create(m)
// locate() is a pure target-path calculation: neither it nor create()
// materializes a file before the first append.
const dir = sessionDir(root, '/work')
await expect(stat(logPath(root, '/work', m.id))).rejects.toThrow()
await expect(stat(rawLogPath(root, '/work', m.id))).rejects.toThrow()
expect((await ctx.sessionPersistence.list()).map(h => h.id)).not.toContain(m.id)
await ctx.sessionPersistence.append(m.id, oneTurnLog())
// now materialized
expect((await stat(logPath(root, '/work', m.id))).isFile()).toBe(true)
expect((await stat(rawLogPath(root, '/work', m.id))).isFile()).toBe(true)
expect((await ctx.sessionPersistence.list()).map(h => h.id)).toContain(m.id)
void dir
})
@@ -189,7 +196,7 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => {
}
const childLocation = ctx.sessionPersistence.locate(child)
expect(childLocation?.path).not.toBe(parentLocation?.path)
expect(childLocation).toEqual({ kind: 'jsonl', path: logPath(root, '/work', child.id) })
expect(childLocation).toEqual({ kind: 'jsonl', path: rawLogPath(root, '/work', child.id) })
})
it('round-trip is byte-identical (incl. assistant/chunk verbatim)', async () => {
@@ -211,7 +218,7 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => {
it('rejects a stored v0 log containing a legacy request/header-delta event', async () => {
const m = meta('legacy-header-delta', '/legacy')
const path = logPath(root, m.cwd, m.id)
const path = rawLogPath(root, m.cwd, m.id)
await mkdir(sessionDir(root, m.cwd), { recursive: true })
await writeFile(path, [
JSON.stringify(toHeaderLine(m)),
@@ -226,7 +233,7 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => {
it('rejects a stored v0 full header carrying the legacy fallback reason', async () => {
const m = meta('legacy-header-fallback', '/legacy')
const path = logPath(root, m.cwd, m.id)
const path = rawLogPath(root, m.cwd, m.id)
await mkdir(sessionDir(root, m.cwd), { recursive: true })
await writeFile(path, [
JSON.stringify(toHeaderLine(m)),
@@ -268,7 +275,7 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => {
// Simulate a crash mid-second-turn: append raw lines that are NOT closed by
// a turn/end (turn/start + step/start are fully written), plus a final
// partial line with no newline (a torn fragment never fully flushed).
const path = logPath(root, '/proj', m.id)
const path = rawLogPath(root, '/proj', m.id)
await writeFile(path, [
JSON.stringify({ type: 'turn/start', seq: 6, time: 8, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } }),
JSON.stringify({ type: 'step/start', seq: 7, time: 9, data: { turn: 2, step: 1 } }),
@@ -301,17 +308,17 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => {
const m = meta('append-only')
await ctx.sessionPersistence.create(m)
await ctx.sessionPersistence.append(m.id, oneTurnLog())
const before = await readFile(logPath(root, undefined, m.id), 'utf8')
const before = await readFile(rawLogPath(root, undefined, m.id), 'utf8')
const committedPrefix = before // the whole committed log
// A crash tail then a repair-append.
await writeFile(logPath(root, undefined, m.id), '\n{"partial', { flag: 'a' })
await writeFile(rawLogPath(root, undefined, m.id), '\n{"partial', { flag: 'a' })
await ctx.sessionPersistence.load(m.id)
await ctx.sessionPersistence.append(m.id, [
{ type: 'turn/start', seq: 6, time: 9, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } },
{ type: 'turn/end', seq: 7, time: 10, data: { turn: 2, reason: { kind: 'completed' } } },
] as SessionEvent[])
const after = await readFile(logPath(root, undefined, m.id), 'utf8')
const after = await readFile(rawLogPath(root, undefined, m.id), 'utf8')
// the committed prefix is byte-for-byte intact at the head of the file
expect(after.startsWith(committedPrefix)).toBe(true)
})
@@ -320,12 +327,12 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => {
const m = meta('truncate-retry')
await ctx.sessionPersistence.create(m)
await ctx.sessionPersistence.append(m.id, oneTurnLog()) // materialized, seqs 0..5
const sizeBefore = (await stat(logPath(root, undefined, m.id))).size
const sizeBefore = (await stat(rawLogPath(root, undefined, m.id))).size
// Force the NEXT fsync (inside appendLines) to fail once, AFTER writeFile
// has already put bytes on disk — simulating an ENOSPC/fsync error
// mid-append. The recovery truncate() also fsyncs, so allow that one.
const handle = await (await import('node:fs/promises')).open(logPath(root, undefined, m.id), 'r')
const handle = await (await import('node:fs/promises')).open(rawLogPath(root, undefined, m.id), 'r')
const proto = Object.getPrototypeOf(handle) as { sync: () => Promise<void> }
await handle.close()
const realSync = proto.sync
@@ -342,7 +349,7 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => {
// The append rejects, but the partial bytes are truncated back: the file is
// its pre-append size and the cursor is unchanged.
await expect(ctx.sessionPersistence.append(m.id, turn2)).rejects.toThrow(/ENOSPC/)
expect((await stat(logPath(root, undefined, m.id))).size).toBe(sizeBefore)
expect((await stat(rawLogPath(root, undefined, m.id))).size).toBe(sizeBefore)
spy.mockRestore()
// The retry now succeeds with NO seq gap — the log is contiguous 0..7.
@@ -423,7 +430,7 @@ describe('SessionPersistenceJsonl: write path (session/event → flush)', () =>
root = await freshRoot()
const ctx = new Context()
await ctx.plugin(SessionStore)
await ctx.plugin(SessionPersistenceJsonl, { root })
await ctx.plugin(SessionPersistenceJsonl, { root, compression: 'none' })
const a = ctx.sessions.create(SessionId('sa'))
const b = ctx.sessions.create(SessionId('sb'))
@@ -461,9 +468,30 @@ describe('SessionPersistenceJsonl: scanLog unit', () => {
expect(() => scanLog(Buffer.from('{"type":"event"}\n'))).toThrow(/session header/)
})
it.each([
['missing', undefined],
['a string', '1'],
['fractional', 1.5],
['negative', -1],
])('rejects a session header with %s delegationDepth', (_label, delegationDepth) => {
const log = JSON.stringify({
type: 'session',
version: 0,
id: 'invalid-depth',
createdAt: 1,
...delegationDepth === undefined ? {} : { delegationDepth },
}) + '\n'
expect(() => scanLog(Buffer.from(log))).toThrow(/session header/)
})
it('rejects a session header with negative-zero delegationDepth', () => {
const log = '{"type":"session","version":0,"id":"invalid-depth","createdAt":1,"delegationDepth":-0}\n'
expect(() => scanLog(Buffer.from(log))).toThrow(/session header/)
})
it('a seq gap after the last turn/end bounds the preserved tail (torn fragment tolerated)', () => {
const log = [
JSON.stringify({ type: 'session', version: 0, id: 'g', createdAt: 1 }),
JSON.stringify({ type: 'session', version: 0, id: 'g', createdAt: 1, delegationDepth: 0 }),
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'
@@ -475,7 +503,7 @@ describe('SessionPersistenceJsonl: scanLog unit', () => {
it('rejects a seq gap BEFORE a later committed turn/end (committed data damaged)', () => {
const log = [
JSON.stringify({ type: 'session', version: 0, id: 'g2', createdAt: 1 }),
JSON.stringify({ type: 'session', version: 0, id: 'g2', createdAt: 1, delegationDepth: 0 }),
JSON.stringify({ type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } }),
JSON.stringify({ type: 'step/start', seq: 2, time: 2, data: { turn: 1, step: 1 } }), // gap: missing seq 1
JSON.stringify({ type: 'turn/end', seq: 3, time: 3, data: { turn: 1, reason: { kind: 'completed' } } }),
@@ -487,7 +515,7 @@ describe('SessionPersistenceJsonl: scanLog unit', () => {
it('rejects a corrupt line BEFORE a later committed turn/end (committed data damaged)', () => {
const log = [
JSON.stringify({ type: 'session', version: 0, id: 'c', createdAt: 1 }),
JSON.stringify({ type: 'session', version: 0, id: 'c', createdAt: 1, delegationDepth: 0 }),
'{not json', // corrupt, sits in the committed region (a turn/end follows)
JSON.stringify({ type: 'turn/end', seq: 1, time: 2, data: { turn: 1, reason: { kind: 'completed' } } }),
].join('\n') + '\n'
@@ -495,7 +523,7 @@ describe('SessionPersistenceJsonl: scanLog unit', () => {
})
it('a header-only log (no event lines at all) preserves nothing — committedBytes is the header', () => {
const log = JSON.stringify({ type: 'session', version: 0, id: 'h0', createdAt: 1 }) + '\n'
const log = JSON.stringify({ type: 'session', version: 0, id: 'h0', createdAt: 1, delegationDepth: 0 }) + '\n'
const scanned = scanLog(Buffer.from(log))
expect(scanned.events).toEqual([])
// committedBytes falls back to the header line's end (no preserved events).
@@ -504,7 +532,7 @@ describe('SessionPersistenceJsonl: scanLog unit', () => {
it('a corrupt line after the last turn/end bounds the preserved tail', () => {
const log = [
JSON.stringify({ type: 'session', version: 0, id: 'c2', createdAt: 1 }),
JSON.stringify({ type: 'session', version: 0, id: 'c2', createdAt: 1, delegationDepth: 0 }),
JSON.stringify({ type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } }),
'{not json', // corrupt crash fragment, no turn/end committed
].join('\n') + '\n'
@@ -515,7 +543,7 @@ describe('SessionPersistenceJsonl: scanLog unit', () => {
it('tolerates a seq gap AFTER a turn/end (uncommitted tail)', () => {
const log = [
JSON.stringify({ type: 'session', version: 0, id: 't', createdAt: 1 }),
JSON.stringify({ type: 'session', version: 0, id: 't', createdAt: 1, delegationDepth: 0 }),
JSON.stringify({ type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } }),
JSON.stringify({ type: 'turn/end', seq: 1, time: 2, data: { turn: 1, reason: { kind: 'completed' } } }),
JSON.stringify({ type: 'step/start', seq: 9, time: 3, data: { turn: 2, step: 1 } }), // gap in uncommitted tail
@@ -531,7 +559,7 @@ describe('SessionPersistenceJsonl: edge cases', () => {
root = await freshRoot()
ctx = new Context()
await ctx.plugin(SessionStore)
await ctx.plugin(SessionPersistenceJsonl, { root })
await ctx.plugin(SessionPersistenceJsonl, { root, compression: 'none' })
})
afterEach(async () => { await ctx.fiber.dispose() })
@@ -551,8 +579,8 @@ describe('SessionPersistenceJsonl: edge cases', () => {
await p
await ctx.sessionPersistence.append(SessionId('create-snap'), oneTurnLog())
// The log materialized under the ORIGINAL cwd, not the mutated one.
expect((await stat(logPath(root, '/orig', SessionId('create-snap')))).isFile()).toBe(true)
await expect(stat(logPath(root, '/mutated', SessionId('create-snap')))).rejects.toThrow()
expect((await stat(rawLogPath(root, '/orig', SessionId('create-snap')))).isFile()).toBe(true)
await expect(stat(rawLogPath(root, '/mutated', SessionId('create-snap')))).rejects.toThrow()
})
it('list discovers sessions across multiple cwd buckets', async () => {
@@ -593,7 +621,7 @@ describe('SessionPersistenceJsonl: edge cases', () => {
// `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) })
const bigHeader = JSON.stringify({ type: 'session', version: 0, id: 'big', createdAt: 1, delegationDepth: 0, pad: 'x'.repeat(9000) })
await writeFile(join(bucket, 'big.jsonl'), bigHeader + '\n')
const ids = (await ctx.sessionPersistence.list()).map(x => x.id)
expect(ids).toContain('big')
@@ -633,7 +661,7 @@ describe('SessionPersistenceJsonl: edge cases', () => {
// of grafting no-cwd events onto a log with mismatched cwd.
const ctx2 = new Context()
await ctx2.plugin(SessionStore)
await ctx2.plugin(SessionPersistenceJsonl, { root })
await ctx2.plugin(SessionPersistenceJsonl, { root, compression: 'none' })
let b!: Session
await ctx2.plugin(Object.assign((inner: Context) => {
b = inner.sessions.create(SessionId('x')) // no cwd
@@ -642,10 +670,10 @@ describe('SessionPersistenceJsonl: edge cases', () => {
// The "/w" log is untouched — no no-cwd events were grafted onto it, and no
// `_no-cwd` log for "x" was created.
const inW = scanLog(await readFile(logPath(root, '/w', SessionId('x'))))
const inW = scanLog(await readFile(rawLogPath(root, '/w', SessionId('x'))))
expect(inW.meta.cwd).toBe('/w')
expect(inW.events).toHaveLength(6)
await expect(stat(logPath(root, undefined, SessionId('x')))).rejects.toThrow()
await expect(stat(rawLogPath(root, undefined, SessionId('x')))).rejects.toThrow()
await ctx2.fiber.dispose()
})
@@ -689,7 +717,10 @@ describe('SessionPersistenceJsonl: edge cases', () => {
it('list returns nothing when the root directory does not exist', async () => {
const ctx2 = new Context()
await ctx2.plugin(SessionStore)
await ctx2.plugin(SessionPersistenceJsonl, { root: join(root, 'does-not-exist-yet') })
await ctx2.plugin(SessionPersistenceJsonl, {
root: join(root, 'does-not-exist-yet'),
compression: 'none',
})
expect(await ctx2.sessionPersistence.list()).toEqual([])
await ctx2.fiber.dispose()
})
@@ -701,7 +732,7 @@ describe('SessionPersistenceJsonl: edge cases', () => {
await writeFile(filePath, 'x')
const ctx2 = new Context()
await ctx2.plugin(SessionStore)
await ctx2.plugin(SessionPersistenceJsonl, { root: filePath })
await ctx2.plugin(SessionPersistenceJsonl, { root: filePath, compression: 'none' })
await expect(ctx2.sessionPersistence.list()).rejects.toThrow(/ENOTDIR/)
await ctx2.fiber.dispose()
})
@@ -712,7 +743,7 @@ describe('SessionPersistenceJsonl: edge cases', () => {
const cwd = '/x'
const ctx2 = new Context()
await ctx2.plugin(SessionStore)
await ctx2.plugin(SessionPersistenceJsonl, { root })
await ctx2.plugin(SessionPersistenceJsonl, { root, compression: 'none' })
await writeFile(sessionDir(root, cwd), 'x') // bucket path is now a FILE
let s!: Session
await ctx2.plugin(Object.assign((inner: Context) => {
@@ -727,14 +758,14 @@ describe('SessionPersistenceJsonl: edge cases', () => {
const m = meta('disk-append', '/d')
await ctx.sessionPersistence.create(m)
await ctx.sessionPersistence.append(m.id, oneTurnLog())
await writeFile(logPath(root, '/d', m.id), '\n{"partial crash', { flag: 'a' })
await writeFile(rawLogPath(root, '/d', m.id), '\n{"partial crash', { flag: 'a' })
// A FRESH backend with no in-memory state: append directly (no prior load)
// → append must adopt from disk, and the adopt's load schedules a repair
// that the same append then performs before writing.
const ctx2 = new Context()
await ctx2.plugin(SessionStore)
await ctx2.plugin(SessionPersistenceJsonl, { root })
await ctx2.plugin(SessionPersistenceJsonl, { root, compression: 'none' })
await ctx2.sessionPersistence.append(m.id, [
{ type: 'turn/start', seq: 6, time: 9, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } },
{ type: 'turn/end', seq: 7, time: 10, data: { turn: 2, reason: { kind: 'completed' } } },
@@ -770,7 +801,7 @@ describe('SessionPersistenceJsonl: edge cases', () => {
// nondeterministic. create scans every bucket, not just meta.cwd's.
const ctx2 = new Context()
await ctx2.plugin(SessionStore)
await ctx2.plugin(SessionPersistenceJsonl, { root })
await ctx2.plugin(SessionPersistenceJsonl, { root, compression: 'none' })
await expect(ctx2.sessionPersistence.create(meta('dup-id', '/projB')))
.rejects.toThrow(/already has a persisted log on disk/)
await ctx2.fiber.dispose()
@@ -780,7 +811,7 @@ describe('SessionPersistenceJsonl: edge cases', () => {
root = await freshRoot()
const ctx2 = new Context()
await ctx2.plugin(SessionStore)
await ctx2.plugin(SessionPersistenceJsonl, { root })
await ctx2.plugin(SessionPersistenceJsonl, { root, compression: 'none' })
const session = ctx2.sessions.create(SessionId('flush-fail'))
// A full turn lands in the write-behind buffer.
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })

24
packages/session-persistence/session-persistence-jsonl/tests/zstd.compat.spec.ts Normal file
View File

@@ -0,0 +1,24 @@
import { describe, expect, it } from 'vitest'
import { compressZstdFrame, decompressZstdFrame, scanZstdFrames } from '../src/zstd.ts'
describe('JSONL Zstandard compatibility', () => {
it('round-trips concatenated checksummed frames through the built-in Node API', async () => {
const encoded = Buffer.concat([
await compressZstdFrame('{"type":"session","version":0,"id":"compat","createdAt":1}\n'),
await compressZstdFrame('{"type":"turn/start","seq":0,"turn":1}\n'),
])
const { frames, tornStart } = scanZstdFrames(encoded)
expect(tornStart).toBeUndefined()
expect(frames).toHaveLength(2)
expect(frames.map(frame => encoded.subarray(frame.start, frame.start + 4).toString('hex')))
.toEqual(['28b52ffd', '28b52ffd'])
const decoded = await Promise.all(frames.map(frame => decompressZstdFrame(encoded.subarray(frame.start, frame.end))))
expect(Buffer.concat(decoded).toString()).toContain('"type":"turn/start"')
const eventFrame = encoded.subarray(frames[1]!.start, frames[1]!.end)
const missingChecksumByte = eventFrame.subarray(0, -1)
expect(scanZstdFrames(missingChecksumByte)).toEqual({ frames: [], tornStart: 0 })
expect((await decompressZstdFrame(missingChecksumByte)).toString()).toContain('"type":"turn/start"')
})
})

483
packages/session-persistence/session-persistence-jsonl/tests/zstd.spec.ts Normal file
View File

@@ -0,0 +1,483 @@
import { afterEach, describe, expect, it, vi } from 'vitest'
import { Context } from 'cordis'
import { appendFile, mkdir, mkdtemp, open, readFile, readdir, rm, stat, writeFile } from 'node:fs/promises'
import type { FileHandle } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
import type { SessionEvent } from '@deepseek-ai/dsh-session'
import SessionPersistenceJsonl from '@deepseek-ai/dsh-session-persistence-jsonl'
import { eventLine, logPath, scanLog, sessionDir, toHeaderLine, type JsonlCompression } from '../src/format.ts'
import { compressZstdFrame, decompressZstdFrame, scanZstdFrames } from '../src/zstd.ts'
import { runPersistenceContract, meta, oneTurnLog } from '../../session-persistence/tests/contract.ts'
import { runCoordinatorContract, type CoordinatorFixture } from '../../session-persistence/tests/coordinator-contract.ts'
const MAGIC = Buffer.from([0x28, 0xB5, 0x2F, 0xFD])
const roots: string[] = []
const contexts: Context[] = []
async function freshRoot(prefix = 'dsh-jsonl-zstd-'): Promise<string> {
const root = await mkdtemp(join(tmpdir(), prefix))
roots.push(root)
return root
}
async function mount(root: string, compression?: JsonlCompression): Promise<Context> {
const ctx = new Context()
contexts.push(ctx)
await ctx.plugin(SessionStore)
await ctx.plugin(SessionPersistenceJsonl, {
root,
...(compression === undefined ? {} : { compression }),
})
return ctx
}
async function decodeCompleteFrames(buffer: Buffer): Promise<Buffer> {
const { frames, tornStart } = scanZstdFrames(buffer)
expect(tornStart).toBeUndefined()
const plaintext: Buffer[] = []
for (const frame of frames) {
plaintext.push(await decompressZstdFrame(buffer.subarray(frame.start, frame.end)))
}
return Buffer.concat(plaintext)
}
async function tornFrame(
plaintext: string,
accepts: (decoded: string) => boolean,
): Promise<Buffer> {
const frame = await compressZstdFrame(plaintext)
const candidateEnds = [
frame.length - 1,
frame.length - 4,
...[0.9, 0.75, 0.6, 0.5, 0.4, 0.25].map(ratio => Math.floor(frame.length * ratio)),
]
for (const end of candidateEnds) {
const candidate = frame.subarray(0, end)
if (scanZstdFrames(candidate).tornStart !== 0) continue
try {
const decoded = (await decompressZstdFrame(candidate)).toString('utf8')
if (accepts(decoded)) return candidate
} catch {
// Some early cuts precede the first decodable block; keep searching for
// a cut that exercises partial-plaintext recovery.
}
}
throw new Error('test fixture could not produce the requested torn Zstandard frame')
}
function deterministicNoise(length: number): string {
let state = 0x12345678
let output = ''
for (let index = 0; index < length; index++) {
state = (Math.imul(state, 1_664_525) + 1_013_904_223) >>> 0
output += String.fromCharCode(33 + (state % 90))
}
return output
}
function emptyStructuralFrame(descriptor: number): Buffer {
const contentSizeFlag = descriptor >>> 6
const singleSegment = (descriptor & 0x20) !== 0
const dictionaryBytes = [0, 1, 2, 4][descriptor & 0x03]!
const contentSizeBytes = contentSizeFlag === 0 ? (singleSegment ? 1 : 0) : 1 << contentSizeFlag
const variableHeader = Buffer.alloc((singleSegment ? 0 : 1) + dictionaryBytes + contentSizeBytes)
const lastEmptyRawBlock = Buffer.from([1, 0, 0])
const checksum = (descriptor & 0x04) === 0 ? Buffer.alloc(0) : Buffer.alloc(4)
return Buffer.concat([MAGIC, Buffer.from([descriptor]), variableHeader, lastEmptyRawBlock, checksum])
}
afterEach(async () => {
vi.restoreAllMocks()
for (const ctx of contexts.splice(0).reverse()) await ctx.fiber.dispose()
for (const root of roots.splice(0)) await rm(root, { recursive: true, force: true })
})
runPersistenceContract('jsonl-zstd', async () => {
const root = await mkdtemp(join(tmpdir(), 'dsh-jsonl-zstd-contract-'))
const ctx = new Context()
await ctx.plugin(SessionStore)
const fiber = await ctx.plugin(SessionPersistenceJsonl, { root })
return {
persistence: ctx.sessionPersistence,
dispose: async () => {
await fiber.dispose()
await rm(root, { recursive: true, force: true })
},
}
})
runCoordinatorContract('jsonl-zstd', async (): Promise<CoordinatorFixture> => {
const root = await mkdtemp(join(tmpdir(), 'dsh-jsonl-zstd-coordinator-'))
return {
mount: async ctx => ctx.plugin(SessionPersistenceJsonl, { root }),
corruptTail: async (id, cwd) => {
const line = JSON.stringify({
type: 'assistant/chunk',
seq: 8,
time: 9,
data: { turn: 2, step: 1, chunk: { type: 'text-delta', index: 0, text: deterministicNoise(300_000) } },
}) + '\n'
const partial = await tornFrame(line, decoded => decoded.length > 0 && !decoded.endsWith('\n'))
await appendFile(logPath(root, cwd, id, 'zstd'), partial)
},
cleanup: async () => { await rm(root, { recursive: true, force: true }) },
}
})
describe('Zstandard frame structure', () => {
it('scans concatenated checksummed frames and honors a frame limit', async () => {
const first = await compressZstdFrame('header\n')
const second = await compressZstdFrame('event\n')
const stream = Buffer.concat([first, second])
expect(scanZstdFrames(Buffer.alloc(0))).toEqual({ frames: [] })
expect(scanZstdFrames(stream)).toEqual({
frames: [{ start: 0, end: first.length }, { start: first.length, end: stream.length }],
})
expect(scanZstdFrames(stream, 1)).toEqual({ frames: [{ start: 0, end: first.length }] })
expect(first[4]! & 0x04).toBe(0x04)
expect(second[4]! & 0x04).toBe(0x04)
expect((await decompressZstdFrame(first)).toString()).toBe('header\n')
})
it('distinguishes incomplete frame regions from invalid complete structure', () => {
expect(scanZstdFrames(MAGIC.subarray(0, 2))).toEqual({ frames: [], tornStart: 0 })
expect(scanZstdFrames(MAGIC)).toEqual({ frames: [], tornStart: 0 })
expect(() => scanZstdFrames(Buffer.alloc(4))).toThrow(/invalid frame magic/)
expect(() => scanZstdFrames(Buffer.concat([MAGIC, Buffer.from([0x08])]))).toThrow(/reserved frame-header bit/)
// Non-single-segment descriptor with no window descriptor.
expect(scanZstdFrames(Buffer.concat([MAGIC, Buffer.from([0x00])]))).toEqual({ frames: [], tornStart: 0 })
// Single-segment header followed by only two bytes of the three-byte block header.
expect(scanZstdFrames(Buffer.concat([MAGIC, Buffer.from([0x20, 0x00, 0x01, 0x00])]))).toEqual({
frames: [],
tornStart: 0,
})
const rawFiveBytes = Buffer.from([(5 << 3) | 1, 0, 0])
expect(scanZstdFrames(Buffer.concat([
MAGIC,
Buffer.from([0x20, 0x00]),
rawFiveBytes,
Buffer.from([0x01, 0x02]),
]))).toEqual({ frames: [], tornStart: 0 })
const reservedBlock = Buffer.concat([
MAGIC,
Buffer.from([0x20, 0x00, 0x07, 0x00, 0x00]),
])
expect(() => scanZstdFrames(reservedBlock)).toThrow(/reserved block type/)
})
it('covers standard header variants, RLE blocks, multiple blocks, and checksums', () => {
for (const descriptor of [0x00, 0x21, 0x42, 0x83, 0xE3]) {
const frame = emptyStructuralFrame(descriptor)
expect(scanZstdFrames(frame)).toEqual({ frames: [{ start: 0, end: frame.length }] })
}
const rle = Buffer.concat([
MAGIC,
Buffer.from([0x20, 0x01]),
Buffer.from([(1 << 3) | (1 << 1) | 1, 0, 0]),
Buffer.from([0x41]),
])
expect(scanZstdFrames(rle)).toEqual({ frames: [{ start: 0, end: rle.length }] })
const twoBlocks = Buffer.concat([
MAGIC,
Buffer.from([0x20, 0x00]),
Buffer.from([0, 0, 0]),
Buffer.from([1, 0, 0]),
])
expect(scanZstdFrames(twoBlocks)).toEqual({ frames: [{ start: 0, end: twoBlocks.length }] })
const checksummed = emptyStructuralFrame(0x24)
expect(scanZstdFrames(checksummed.subarray(0, -1))).toEqual({ frames: [], tornStart: 0 })
expect(scanZstdFrames(checksummed)).toEqual({ frames: [{ start: 0, end: checksummed.length }] })
})
})
describe('SessionPersistenceJsonl: default Zstandard encoding', () => {
it('writes .jsonl.zstd by default with one header frame and one first-batch frame', async () => {
const root = await freshRoot()
const ctx = await mount(root)
const header = meta('default-zstd', '/work')
await ctx.sessionPersistence.create(header)
await ctx.sessionPersistence.append(header.id, oneTurnLog())
const path = logPath(root, header.cwd, header.id, 'zstd')
const buffer = await readFile(path)
expect(buffer.subarray(0, 4)).toEqual(MAGIC)
await expect(stat(logPath(root, header.cwd, header.id, 'none'))).rejects.toThrow()
expect(ctx.sessionPersistence.locate(header)).toEqual({ kind: 'jsonl', path })
const scan = scanZstdFrames(buffer)
expect(scan.frames).toHaveLength(2)
const plaintext = await decodeCompleteFrames(buffer)
expect(plaintext.toString()).toBe([
JSON.stringify(toHeaderLine(header)),
...oneTurnLog().map(eventLine),
'',
].join('\n'))
expect((await ctx.sessionPersistence.load(header.id)).events).toEqual(oneTurnLog())
})
it('resolves the default when a programmatic wrapper bypasses Loader schema normalization', async () => {
const root = await freshRoot()
const ctx = new Context()
contexts.push(ctx)
await ctx.plugin(SessionStore)
let backend!: SessionPersistenceJsonl
await ctx.plugin(Object.assign((inner: Context) => {
backend = new SessionPersistenceJsonl(inner, { root })
}, { inject: ['sessions'] }))
const header = meta('direct-default')
expect(backend.locate(header)).toEqual({
kind: 'jsonl',
path: logPath(root, header.cwd, header.id, 'zstd'),
})
})
it('appends one frame per durable batch without rewriting prior bytes', async () => {
const root = await freshRoot()
const ctx = await mount(root)
const header = meta('append-frame')
await ctx.sessionPersistence.create(header)
await ctx.sessionPersistence.append(header.id, oneTurnLog())
const path = logPath(root, header.cwd, header.id, 'zstd')
const before = await readFile(path)
const secondTurn = [
{ 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[]
await ctx.sessionPersistence.append(header.id, secondTurn)
const after = await readFile(path)
expect(after.subarray(0, before.length)).toEqual(before)
expect(scanZstdFrames(after).frames).toHaveLength(3)
expect((await ctx.sessionPersistence.load(header.id)).events).toEqual([...oneTurnLog(), ...secondTurn])
})
it('lists from a multi-chunk header frame without decoding a corrupt event frame', async () => {
const root = await freshRoot()
const ctx = await mount(root)
const header = meta('large-header', `/work/${'x'.repeat(24_000)}`)
await ctx.sessionPersistence.create(header)
await ctx.sessionPersistence.append(header.id, oneTurnLog())
const path = logPath(root, header.cwd, header.id, 'zstd')
const buffer = Buffer.from(await readFile(path))
const eventFrame = scanZstdFrames(buffer).frames[1]!
buffer[eventFrame.end - 1] = buffer[eventFrame.end - 1]! ^ 0xFF
await writeFile(path, buffer)
expect((await ctx.sessionPersistence.list()).map(item => item.id)).toEqual([header.id])
await expect(ctx.sessionPersistence.load(header.id)).rejects.toThrow(/frame at byte .* failed validation/)
})
it('preserves complete records from a torn frame and re-encodes them with crash closers', async () => {
const root = await freshRoot()
const ctx = await mount(root)
const header = meta('recover-torn', '/proj')
await ctx.sessionPersistence.create(header)
await ctx.sessionPersistence.append(header.id, oneTurnLog())
const path = logPath(root, header.cwd, header.id, 'zstd')
const committed = await readFile(path)
const openTurn = [
{ type: 'turn/start', seq: 6, time: 7, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } },
{ type: 'step/start', seq: 7, time: 8, data: { turn: 2, step: 1 } },
{ type: 'assistant/chunk', seq: 8, time: 9, data: { turn: 2, step: 1, chunk: { type: 'text-delta', index: 0, text: deterministicNoise(300_000) } } },
] as SessionEvent[]
const plaintext = openTurn.map(eventLine).join('\n') + '\n'
const partial = await tornFrame(plaintext, (decoded) => {
const newlines = decoded.match(/\n/g)?.length ?? 0
return newlines >= 2 && !decoded.endsWith('\n')
})
await appendFile(path, partial)
const loaded = await ctx.sessionPersistence.load(header.id)
expect(loaded.events.map(event => event.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7, 8, 9])
expect(loaded.events[6]).toEqual(openTurn[0])
expect(loaded.events[7]).toEqual(openTurn[1])
expect(loaded.events.some(event => event.type === 'assistant/chunk' && event.seq === 8)).toBe(false)
expect(loaded.events[8]?.type).toBe('step/end')
expect(loaded.events[9]?.type).toBe('turn/end')
const repaired = await readFile(path)
expect(repaired.subarray(0, committed.length)).toEqual(committed)
expect(scanZstdFrames(repaired).tornStart).toBeUndefined()
expect(scanLog(await decodeCompleteFrames(repaired)).events).toEqual(loaded.events)
})
it('drops a frame torn in its header before it has produced plaintext', async () => {
const root = await freshRoot()
const ctx = await mount(root)
const header = meta('partial-magic')
await ctx.sessionPersistence.create(header)
await ctx.sessionPersistence.append(header.id, oneTurnLog())
const path = logPath(root, header.cwd, header.id, 'zstd')
const committed = await readFile(path)
await appendFile(path, MAGIC.subarray(0, 2))
expect((await ctx.sessionPersistence.load(header.id)).events).toEqual(oneTurnLog())
expect(await readFile(path)).toEqual(committed)
})
it('recovers complete events when EOF tears only the final frame checksum', async () => {
const root = await freshRoot()
const ctx = await mount(root)
const header = meta('partial-checksum')
await ctx.sessionPersistence.create(header)
await ctx.sessionPersistence.append(header.id, oneTurnLog())
const path = logPath(root, header.cwd, header.id, 'zstd')
const secondTurn = [
{ 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[]
const frame = await compressZstdFrame(secondTurn.map(eventLine).join('\n') + '\n')
await appendFile(path, frame.subarray(0, -1))
const loaded = await ctx.sessionPersistence.load(header.id)
expect(loaded.events).toEqual([...oneTurnLog(), ...secondTurn])
const repaired = await readFile(path)
expect(scanZstdFrames(repaired).tornStart).toBeUndefined()
expect(scanLog(await decodeCompleteFrames(repaired)).events).toEqual(loaded.events)
})
it('rejects a complete frame containing a torn JSONL record', async () => {
const root = await freshRoot()
const ctx = await mount(root)
const header = meta('complete-bad-jsonl')
await ctx.sessionPersistence.create(header)
await ctx.sessionPersistence.append(header.id, oneTurnLog())
await appendFile(
logPath(root, header.cwd, header.id, 'zstd'),
await compressZstdFrame('{"type":"turn/start"'),
)
await expect(ctx.sessionPersistence.load(header.id)).rejects.toThrow(/complete frame contains a torn JSONL record/)
})
it('rolls back a checksummed append frame when fsync fails', async () => {
const root = await freshRoot()
const ctx = await mount(root)
const header = meta('zstd-fsync-rollback')
await ctx.sessionPersistence.create(header)
await ctx.sessionPersistence.append(header.id, oneTurnLog())
const path = logPath(root, header.cwd, header.id, 'zstd')
const before = await readFile(path)
const handle = await open(path, 'r')
const prototype = Object.getPrototypeOf(handle) as { sync: () => Promise<void> }
await handle.close()
const realSync = prototype.sync
let failed = false
const spy = vi.spyOn(prototype, 'sync').mockImplementation(async function (this: FileHandle) {
if (!failed) {
failed = true
throw new Error('simulated Zstandard fsync failure')
}
return realSync.call(this)
})
const secondTurn = [
{ 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[]
await expect(ctx.sessionPersistence.append(header.id, secondTurn)).rejects.toThrow(/simulated Zstandard fsync failure/)
expect(await readFile(path)).toEqual(before)
spy.mockRestore()
await ctx.sessionPersistence.append(header.id, secondTurn)
expect((await ctx.sessionPersistence.load(header.id)).events).toEqual([...oneTurnLog(), ...secondTurn])
})
it('skips empty, incomplete, and non-header compressed artifacts while rejecting malformed header frames', async () => {
const root = await freshRoot()
const bucket = sessionDir(root, undefined)
await mkdir(bucket, { recursive: true })
await writeFile(join(bucket, 'empty.jsonl.zstd'), '')
await writeFile(join(bucket, 'partial.jsonl.zstd'), MAGIC)
await writeFile(join(bucket, 'not-header.jsonl.zstd'), await compressZstdFrame('{"type":"turn/start"}\n'))
const ctx = await mount(root)
expect(await ctx.sessionPersistence.list()).toEqual([])
await writeFile(join(bucket, 'two-lines.jsonl.zstd'), await compressZstdFrame([
JSON.stringify(toHeaderLine(meta('two-lines'))),
JSON.stringify({ type: 'turn/start' }),
'',
].join('\n')))
await expect(ctx.sessionPersistence.list()).rejects.toThrow(/first frame is not exactly one header line/)
await expect(ctx.sessionPersistence.load(SessionId('two-lines')))
.rejects.toThrow(/first frame is not exactly one header line/)
})
it('rejects missing, empty, and checksum-corrupt header frames on targeted reads', async () => {
const root = await freshRoot()
const bucket = sessionDir(root, undefined)
await mkdir(bucket, { recursive: true })
await writeFile(logPath(root, undefined, SessionId('partial-only'), 'zstd'), MAGIC)
await writeFile(logPath(root, undefined, SessionId('empty-header'), 'zstd'), await compressZstdFrame(''))
const corruptHeader = Buffer.from(await compressZstdFrame(`${JSON.stringify(toHeaderLine(meta('bad-checksum')))}\n`))
corruptHeader[corruptHeader.length - 1] = corruptHeader[corruptHeader.length - 1]! ^ 0xFF
await writeFile(logPath(root, undefined, SessionId('bad-checksum'), 'zstd'), corruptHeader)
const ctx = await mount(root)
await expect(ctx.sessionPersistence.load(SessionId('partial-only')))
.rejects.toThrow(/empty or header-less Zstandard session log/)
await expect(ctx.sessionPersistence.load(SessionId('empty-header')))
.rejects.toThrow(/first frame is not exactly one header line/)
await expect(ctx.sessionPersistence.list()).rejects.toThrow(/header frame failed validation/)
})
})
describe('SessionPersistenceJsonl: encoding selection', () => {
it('rejects roots owned by the opposite encoding in both directions', async () => {
const rawRoot = await freshRoot('dsh-jsonl-raw-mismatch-')
const raw = await mount(rawRoot, 'none')
const rawHeader = meta('raw-log')
await raw.sessionPersistence.create(rawHeader)
await raw.sessionPersistence.append(rawHeader.id, oneTurnLog())
const defaultBackend = await mount(rawRoot)
await expect(defaultBackend.sessionPersistence.list()).rejects.toThrow(/configured for compression "zstd"/)
const zstdRoot = await freshRoot('dsh-jsonl-zstd-mismatch-')
const zstd = await mount(zstdRoot)
const zstdHeader = meta('zstd-log')
await zstd.sessionPersistence.create(zstdHeader)
await zstd.sessionPersistence.append(zstdHeader.id, oneTurnLog())
const rawBackend = await mount(zstdRoot, 'none')
await expect(rawBackend.sessionPersistence.list()).rejects.toThrow(/configured for compression "none"/)
})
it('rechecks targeted artifacts and listing after an initially empty root', async () => {
const root = await freshRoot()
const ctx = await mount(root)
expect(await ctx.sessionPersistence.list()).toEqual([])
const loadHeader = meta('late-raw-load', '/late')
await mkdir(sessionDir(root, loadHeader.cwd), { recursive: true })
await writeFile(logPath(root, loadHeader.cwd, loadHeader.id, 'none'), [
JSON.stringify(toHeaderLine(loadHeader)),
...oneTurnLog().map(eventLine),
'',
].join('\n'))
await expect(ctx.sessionPersistence.load(loadHeader.id)).rejects.toThrow(/uses \.jsonl/)
await expect((ctx.sessionPersistence as SessionPersistenceJsonl).loadLive(loadHeader.id, loadHeader.cwd))
.rejects.toThrow(/uses \.jsonl/)
await expect(ctx.sessionPersistence.list()).rejects.toThrow(/uses \.jsonl/)
})
it('refuses materialization when an opposite artifact appears after create', async () => {
const root = await freshRoot()
const ctx = await mount(root)
await ctx.sessionPersistence.list()
const header = meta('late-raw-materialize', '/late')
await ctx.sessionPersistence.create(header)
await mkdir(sessionDir(root, header.cwd), { recursive: true })
await writeFile(logPath(root, header.cwd, header.id, 'none'), [
JSON.stringify(toHeaderLine(header)),
...oneTurnLog().map(eventLine),
'',
].join('\n'))
await expect(ctx.sessionPersistence.append(header.id, oneTurnLog())).rejects.toThrow(/uses \.jsonl/)
expect((await readdir(sessionDir(root, header.cwd))).some(name => name.endsWith('.jsonl.zstd'))).toBe(false)
})
})

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

@@ -253,14 +253,15 @@ export class SessionPersistenceSqlite extends SessionPersistence implements Pers
*/
private writeRow(meta: SessionHeader): void {
this.db.prepare(`
INSERT INTO sessions (id, version, created_at, cwd, parent_session, seed_length)
VALUES (?, ?, ?, ?, ?, ?)
INSERT INTO sessions (id, version, created_at, cwd, parent_session, seed_length, delegation_depth)
VALUES (?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(id) DO UPDATE SET
version = excluded.version,
created_at = excluded.created_at,
cwd = excluded.cwd,
parent_session = excluded.parent_session,
seed_length = excluded.seed_length
seed_length = excluded.seed_length,
delegation_depth = excluded.delegation_depth
`).run(
meta.id,
meta.version,
@@ -268,6 +269,7 @@ export class SessionPersistenceSqlite extends SessionPersistence implements Pers
meta.cwd ?? null,
meta.parentSession ?? null,
meta.seedLength ?? null,
meta.delegationDepth ?? null,
)
}
}

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

@@ -15,7 +15,7 @@ import type { SessionEvent, SessionId, SessionHeader, SurfaceOp } from '@deepsee
* layout; orthogonal to a session's own `version` (which versions the EVENT
* vocabulary, stored per session in the `sessions` row).
*/
export const SCHEMA_VERSION = 4
export const SCHEMA_VERSION = 5
/**
* A row of the `sessions` table — the out-of-log metadata ({@link SessionHeader}).
@@ -31,6 +31,7 @@ export interface SessionRow {
cwd: string | null
parent_session: string | null
seed_length: number | null
delegation_depth: number | null
}
/** An `events` table row: one `SessionEvent` mapped 1:1 (`data` is JSON text). */
@@ -83,9 +84,10 @@ export function openDatabase(path: string, journalMode: JournalMode): DatabaseSy
id TEXT PRIMARY KEY,
version INTEGER NOT NULL,
created_at INTEGER NOT NULL,
cwd TEXT,
parent_session TEXT,
seed_length INTEGER
cwd TEXT,
parent_session TEXT,
seed_length INTEGER,
delegation_depth INTEGER
) STRICT
`)
db.exec(`
@@ -116,6 +118,7 @@ export function rowToMeta(row: SessionRow): SessionHeader {
...row.cwd !== null ? { cwd: row.cwd } : {},
...row.parent_session !== null ? { parentSession: row.parent_session as SessionId } : {},
...row.seed_length !== null ? { seedLength: row.seed_length } : {},
...row.delegation_depth !== null ? { delegationDepth: row.delegation_depth } : {},
}
}

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

@@ -383,7 +383,7 @@ describe('SessionPersistenceSqlite: durability and crash semantics', () => {
})
it('exposes the schema version constant', () => {
expect(SCHEMA_VERSION).toBe(4)
expect(SCHEMA_VERSION).toBe(5)
})
})

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

@@ -2,7 +2,7 @@
The abstract durable session-persistence seam (`ctx.sessionPersistence`). Defines WHAT a persistence backend does — durably store, reload, and list sessions — without saying HOW. Mirrors the `dsh-bash` capability-seam template ([capability seams](../../../.agents/notes/implemented/architecture/2026-06-13-capability-seams.md)): an abstract service here, a concrete implementation in a sibling package, consumers that inject the interface.
The persisted unit IS the existing `SessionEvent` (event-sourced model — the log is the single source of truth), so there is no parallel "persisted message" type. Metadata that is NOT replayable conversation state (format version, cwd, lineage, seed boundary) travels separately as `SessionHeader`, owned by `dsh-session` and re-exported here.
The persisted unit IS the existing `SessionEvent` (event-sourced model — the log is the single source of truth), so there is no parallel "persisted message" type. Metadata that is NOT replayable conversation state (format version, cwd, lineage, seed boundary, delegation depth) travels separately as `SessionHeader`, owned by `dsh-session` and re-exported here.
## Service API (`ctx.sessionPersistence`)
@@ -51,7 +51,7 @@ Three backends run these suites: an in-memory reference (in `tests/`), `dsh-sess
## Metadata and location types
Re-exported from `dsh-session`: `SessionHeader` (immutable session metadata: `version`, `id`, `createdAt`, `cwd?`, `parentSession?`, `seedLength?`). `SessionLocation` is `{ readonly kind: string; readonly path: string }`; its path is an absolute backend target, not proof that the artifact exists or contains an unflushed turn.
Re-exported from `dsh-session`: `SessionHeader` (immutable session metadata: `version`, `id`, `createdAt`, `cwd?`, `parentSession?`, `seedLength?`, `delegationDepth?`). `SessionLocation` is `{ readonly kind: string; readonly path: string }`; its path is an absolute backend target, not proof that the artifact exists or contains an unflushed turn.
## Model Experience

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

@@ -107,6 +107,27 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
}
})
it('round-trips the delegation depth through persistence', async () => {
// A subagent child's recursion budget lives in its header; a reload that
// dropped it would reset the child to top-level and un-bound maxDepth
// (JSONL stores it in the header line; SQLite uses `delegation_depth`).
const fix = await makeFixture()
const { ctx, fiber } = await freshCtx(fix)
try {
const session = ctx.sessions.create(SessionId('delegated-child'), {
meta: { cwd: WORK, parentSession: SessionId('root'), delegationDepth: 2 },
})
send(session, oneTurnLog())
await ctx.parallel('session/flush', session)
const loaded = await ctx.sessionPersistence.load(SessionId('delegated-child'))
expect(loaded.meta.delegationDepth).toBe(2)
} finally {
await fiber.dispose()
await fix.cleanup()
}
})
it('source-frozen events cannot be mutated after buffering and persist unchanged', async () => {
const fix = await makeFixture()
const { ctx, fiber } = await freshCtx(fix)