Files
deepseek-harness/packages/session-persistence-jsonl/src/format.ts
Tianyi Cui df4b7d3d9a feat(session-persistence): abstract seam + JSONL backend + wiring
Add the durable session-persistence capability seam (ADR 0016): an
abstract SessionPersistence service (dsh-session-persistence,
ctx.sessionPersistence) defining create/append/load/list/has/delete/
update over the existing SessionEvent — no parallel persisted type — and
a first implementation (dsh-session-persistence-jsonl): an append-only
JSONL log per session with crash-safe atomic writes, truncation-repair
of a never-committed crash tail, and a read/replay path. SessionMeta
(format version, cwd, lineage) travels out-of-log via session.header.

A shared runPersistenceContract suite holds every backend to the same
append-only / contiguous-seq / lazy-materialization / serializability
semantics.

Config-driven create() now uses a per-run ${id}-session-<uuid> session
id so a fixed name no longer collides with an on-disk log once a durable
backend is loaded; each run is a new session (a demo simplification). The
examples drop their hand-rolled session-jsonl.ts and load the JSONL
backend via cordis.yml; CI smoke-loads it too.

The agent-facing create/resume factory that consumes load() is a
separate seam, deferred to a follow-up; this change stops at the load
primitive and does not reach into the loop.
2026-06-15 21:05:46 +08:00

250 lines
10 KiB
TypeScript

/**
* On-disk format helpers for the JSONL session-persistence backend: path
* sanitization (a {@link SessionId} is an unvalidated branded string, so it
* MUST be encoded before use in a path — no traversal, no collision), the
* per-cwd directory layout, header-line (de)serialization, the atomic sidecar
* for mutable summary fields, and the truncation-repair offset computation.
*
* @module dsh-session-persistence-jsonl/format
*/
import { createHash } from 'node:crypto'
import { join } from 'node:path'
import type { SessionEvent, SessionHeader, SessionId, SessionMeta } from '@deepseek-ai/dsh-session'
/**
* The first line of a session's `.jsonl` file: the immutable
* {@link SessionHeader} tagged as a `session` record so a reader can tell it
* apart from an event line.
*/
export interface HeaderLine {
type: 'session'
version: number
id: SessionId
createdAt: number
cwd?: string
parentSession?: SessionId
}
/** Build the header line object from a {@link SessionHeader}. */
export function toHeaderLine(header: SessionHeader): HeaderLine {
return {
type: 'session',
version: header.version,
id: header.id,
createdAt: header.createdAt,
...header.cwd !== undefined ? { cwd: header.cwd } : {},
...header.parentSession !== undefined ? { parentSession: header.parentSession } : {},
}
}
/** Parse a header line back into a {@link SessionHeader}. */
export function fromHeaderLine(line: HeaderLine): SessionHeader {
return {
version: line.version,
id: line.id,
createdAt: line.createdAt,
...line.cwd !== undefined ? { cwd: line.cwd } : {},
...line.parentSession !== undefined ? { parentSession: line.parentSession } : {},
}
}
/** Type guard: a parsed first line is a well-formed session header. */
function isHeaderLine(value: unknown): value is HeaderLine {
return (
typeof value === 'object' && value !== null
&& (value as { type?: unknown }).type === 'session'
&& typeof (value as { version?: unknown }).version === 'number'
&& typeof (value as { id?: unknown }).id === 'string'
&& typeof (value as { createdAt?: unknown }).createdAt === 'number'
)
}
/**
* Encode an arbitrary string as a single safe path segment, injectively over
* ALL JS (UTF-16) strings — including lone surrogates. A {@link SessionId} is
* an unvalidated branded string, so this neutralizes `../`, absolute paths,
* NUL, and separators before any filesystem use.
*
* Each UTF-16 code unit is either kept literal (the safe set `[A-Za-z0-9_-]`)
* or escaped as `~XXXX` (its 4-hex-digit code unit). `~` is itself escaped, so
* the mapping is injective and reversible: distinct inputs never collide. We
* iterate code UNITS (`charCodeAt`), not code points, so a lone surrogate
* escapes to a distinct `~XXXX` instead of being normalized to U+FFFD (which
* `Buffer.from(…, 'utf8')` would do, breaking injectivity). `.` is in the safe
* set for readability but the whole-segment tokens `.`/`..` are escaped so they
* can never traverse.
*/
export function encodeSegment(raw: string): string {
if (raw.length === 0) throw new Error('cannot encode an empty path segment')
if (raw === '.') return '~002E'
if (raw === '..') return '~002E~002E'
let out = ''
for (let i = 0; i < raw.length; i++) {
const code = raw.charCodeAt(i)
const ch = String.fromCharCode(code)
if (ch !== '~' && /^[A-Za-z0-9._-]$/.test(ch)) {
out += ch
} else {
out += '~' + code.toString(16).toUpperCase().padStart(4, '0')
}
}
return out
}
/**
* The directory a session's files live in: the configured root, then a per-cwd
* subdirectory so sessions group by project. The cwd subdir is a stable hash
* (short, collision-resistant, filesystem-safe) plus an encoded suffix for
* readability; sessions without a cwd go in a shared `_no-cwd` bucket.
*/
export function sessionDir(root: string, cwd: string | undefined): string {
if (cwd === undefined) return join(root, '_no-cwd')
const hash = createHash('sha256').update(cwd).digest('hex').slice(0, 12)
return join(root, `cwd-${hash}`)
}
/** The append-only event-log file path for a session. */
export function logPath(root: string, cwd: string | undefined, id: SessionId): string {
return join(sessionDir(root, cwd), `${encodeSegment(id)}.jsonl`)
}
/** The mutable-summary sidecar path for a session (beside its log). */
export function sidecarPath(root: string, cwd: string | undefined, id: SessionId): string {
return join(sessionDir(root, cwd), `${encodeSegment(id)}.summary.json`)
}
/** Serialize one event as a JSONL line (no trailing newline). */
export function eventLine(event: SessionEvent): string {
return JSON.stringify(event)
}
/**
* Compute the byte offset of the END of the last complete `turn/end` line in a
* JSONL log buffer (the header line is index 0). Returns the offset to which a
* crash tail should be truncated, and the contiguous events up to and including
* that `turn/end`. A parse error or a `seq` gap in the MIDDLE (at or before the
* last `turn/end`) makes the session unloadable and throws; trailing garbage
* AFTER the last `turn/end` is the tolerated crash tail and is excluded.
*
* This relies on the session-log invariant that every event lives inside a turn
* (`Session.append` enforces it): the last `turn/end` is therefore the last
* durable boundary, and nothing committed can sit outside a completed turn.
*/
export function scanLog(buffer: Buffer): { meta: SessionMeta; events: SessionEvent[]; committedBytes: number } {
const text = buffer.toString('utf8')
// Split into complete (newline-terminated) lines, tracking the byte offset of
// each line's end so the truncation point is exact (multi-byte chars make the
// char offset differ from the byte offset). A trailing line with no newline is
// an uncommitted crash fragment and is ignored — it is below the last
// turn/end by construction (the loop only flushes whole lines).
//
// Track the byte offset with a RUNNING accumulator (`endByte`), adding each
// line's byte length as we go. Recomputing `Buffer.byteLength(text.slice(0, i))`
// per newline would rescan the whole prefix every time — O(n²) over a long
// log (one assistant/chunk line per token makes that pathological).
const lines: { text: string; endByte: number }[] = []
let start = 0
let byteOffset = 0
for (let i = 0; i < text.length; i++) {
if (text[i] === '\n') {
const lineText = text.slice(start, i)
byteOffset += Buffer.byteLength(lineText, 'utf8') + 1 // +1 for the '\n' (a 1-byte char)
lines.push({ text: lineText, endByte: byteOffset })
start = i + 1
}
}
const [headerEntry, ...eventEntries] = lines
if (headerEntry === undefined) throw new Error('empty or header-less session log')
// Line 0 is the header.
let parsedHeader: unknown
try {
parsedHeader = JSON.parse(headerEntry.text)
} catch {
throw new Error('corrupt session log: header line is not valid JSON')
}
if (!isHeaderLine(parsedHeader)) {
throw new Error('corrupt session log: first line is not a session header')
}
const headerLine = parsedHeader
// Find the committed region: the prefix up to and including the LAST complete
// `turn/end` in the WHOLE log. Two passes so a crash tail after the last
// turn/end is tolerated, but corruption/gaps AT OR BEFORE the last committed
// turn/end make the log unloadable (committed data must never be silently
// dropped).
//
// Pass 1: parse every line that parses, recording (parsedOk, seq, isTurnEnd,
// endByte) per line index. Lines that fail to parse are holes.
interface Parsed { ok: boolean; event?: SessionEvent; endByte: number }
const parsed: Parsed[] = eventEntries.map((entry) => {
try {
return { ok: true, event: JSON.parse(entry.text) as SessionEvent, endByte: entry.endByte }
} catch {
return { ok: false, endByte: entry.endByte }
}
})
// The last index (into eventEntries) that is a valid `turn/end`.
let lastTurnEnd = -1
for (let i = parsed.length - 1; i >= 0; i--) {
const p = parsed[i]
if (p?.ok && p.event?.type === 'turn/end') { lastTurnEnd = i; break }
}
// No committed turn/end anywhere: nothing is committed. The whole event
// region is an uncommitted (first-turn) tail — committedBytes is the header.
if (lastTurnEnd < 0) {
const meta = metaFrom(headerLine)
return { meta, events: [], committedBytes: headerEntry.endByte }
}
// Pass 2: the committed prefix [0..lastTurnEnd] must be fully intact and
// contiguous (line i is a parsed event with seq === i). A hole or seq gap in
// the committed region means committed data was damaged → unloadable.
const committed: SessionEvent[] = []
for (let i = 0; i <= lastTurnEnd; i++) {
const p = parsed[i]
if (!p?.ok || p.event === undefined) {
throw new Error(`corrupt session log: unparsable committed event at line ${i + 1}`)
}
if (p.event.seq !== i) {
throw new Error(`corrupt session log: seq gap in committed region at line ${i + 1} (expected ${i}, got ${p.event.seq})`)
}
committed.push(p.event)
}
const lastEntry = parsed[lastTurnEnd]
/* v8 ignore next -- lastTurnEnd indexes a parsed entry by construction */
const committedBytes = lastEntry ? lastEntry.endByte : headerEntry.endByte
return { meta: metaFrom(headerLine), events: committed, committedBytes }
}
/** Build the load-time {@link SessionMeta} from a header line (summary overlaid later). */
function metaFrom(headerLine: HeaderLine): SessionMeta {
return {
...fromHeaderLine(headerLine),
updatedAt: headerLine.createdAt, // overlaid by the sidecar in load()
}
}
/**
* Parse just the header line of a log into load-time {@link SessionMeta}, or
* `undefined` if it is missing/not a header. Used by `list()` to read session
* metadata WITHOUT parsing the whole log: a session picker scales with the
* number of sessions, not the total size of every conversation. The summary
* sidecar is overlaid by the caller; `updatedAt` here mirrors `createdAt` until
* then (same as {@link scanLog}'s load-time meta).
*/
export function parseHeaderMeta(firstLine: string): SessionMeta | undefined {
let parsed: unknown
try {
parsed = JSON.parse(firstLine)
} catch {
return undefined
}
if (!isHeaderLine(parsed)) return undefined
return metaFrom(parsed)
}