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

feat(session-persistence): preserve interrupted turns on crash; don't truncate (review #33)

Browse Source 
A crash can leave a durable log whose final turn never closed. The old
behavior truncated everything after the last turn/end as a "crash tail".
But a single turn can be HUGE in a long-horizon task (many steps, large
tool output), so truncating it silently destroys real, durably-written
work — truncating a turn is wrong.

New crash recovery (ADR 0018): load() PRESERVES the interrupted turn's
events and CLOSES the orphaned turn by durably appending synthetic
boundary events — a step/end if a step was open, then a turn/end carrying
the new merge-extensible TurnEndReason {kind:'interrupted'}. load()
returns the balanced log, so a resumed session is immediately usable. Only
a never-fully-written TORN tail fragment is discarded; corruption in the
committed region is still unloadable.

- dsh-session: TurnEndReason {kind:'interrupted'} + shared
  interruptedTurnClosers() repair helper.
- JSONL backend: scanLog preserves the longest contiguous prefix
  (including a partial final turn); loadCore truncates a torn fragment and
  durably writes the closers, returning the balanced log.
- runPersistenceContract gains a crash-recovery test (both backends + mock).
- Docs: ADR 0018/0017, architecture.md, package READMEs.

Also (review #33): RFC 013 records the "move event vocabulary to Zod"
question (merge-extensible maps → runtime schema registry) + blast radius;
deferred, not done here.
This commit is contained in:
Tianyi Cui
2026-06-16 21:27:50 +08:00
parent 96331432b8
commit efee449cfe
16 changed files with 361 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -24,7 +24,7 @@ The JSONL durable session-persistence backend — a concrete `SessionPersistence
- **Lazy materialization.** `create(meta)` writes nothing; the `.jsonl` (header + first batch) is written atomically (temp-write + `fsync` + rename) on the first `append`. A created-but-never-appended session leaves nothing on disk and is absent from `has`/`list`.
- **Append-only.** Committed events (at or below a flushed `turn/end`) are never rewritten. Subsequent appends are line appends at EOF + `fsync`.
- **Truncation-repair.** `load` returns events only up to the last complete `turn/end` and records the byte offset of any never-committed crash tail; the first post-load `append` `ftruncate`s to that offset (+ `fsync`) before writing, atomically discarding only the uncommitted tail.
- **Crash recovery — close, don't truncate.** A crash can leave a log whose final turn never closed (real events after the last `turn/end`). `load` PRESERVES those events (a turn can be huge — they are real work) and closes the orphaned turn by durably appending synthetic boundary events (a `step/end` if a step was open, then `turn/end {kind:'interrupted'}`), returning a balanced log. Only a never-fully-written **torn tail fragment** (a final line with no newline / unparseable) is `ftruncate`d away before the closers are written. See ADR 0018.
- **Contiguous-seq.** `load` rejects a mid-log parse error or `seq` gap (unloadable); `append` rejects a batch whose first `seq` does not continue the stored log, and rejects non-JSON-serializable `event.data` naming the offending event type.
- **Format version.** Only v1 is supported; `load` rejects an unknown version. A future format change requires a version bump + migration.

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

@@ -120,16 +120,23 @@ export function eventLine(event: SessionEvent): string {
}
/**
* 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.
* Parse a JSONL log buffer into its preserved event prefix (the header is line
* 0). Returns the longest prefix of complete, seq-contiguous events plus the
* byte offset of the end of the last preserved line (`committedBytes`).
*
* A crash can leave a durable log whose final turn never closed: real,
* fully-written events sit after the last `turn/end`. Those are PRESERVED (a
* single turn can be huge in a long-horizon task — truncating it would destroy
* real work); the backend closes the orphaned open turn with a synthetic
* `turn/end {kind:'interrupted'}` on reload (ADR 0018). Only a TORN trailing
* fragment — a final line never fully flushed (no newline, unparseable, or a
* seq gap) — is excluded; it bounds the preserved region. A parse error or seq
* gap AT OR BEFORE the last committed `turn/end` is committed-data corruption
* and makes the session unloadable (throws).
*
* This relies on the session-log invariant that every event lives inside a turn
* (`Session.append` enforces it): the last `turn/end` is therefore the last
* durable boundary, and nothing committed can sit outside a completed turn.
* (`Session.append` enforces it): only the final turn can be open, so the
* preserved tail is at most one unclosed turn.
*/
export function scanLog(buffer: Buffer): { meta: SessionMeta; events: SessionEvent[]; committedBytes: number } {
const text = buffer.toString('utf8')
@@ -187,38 +194,46 @@ export function scanLog(buffer: Buffer): { meta: SessionMeta; events: SessionEve
}
})
// The last index (into eventEntries) that is a valid `turn/end`.
// The last index (into eventEntries) that is a valid `turn/end` — the last
// fully-committed boundary (the loop flushes only at turn/end).
let lastTurnEnd = -1
for (let i = parsed.length - 1; i >= 0; i--) {
const p = parsed[i]
if (p?.ok && p.event?.type === 'turn/end') { lastTurnEnd = i; break }
}
// 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++) {
// Walk the longest PREFIX of complete, seq-contiguous, parseable event lines
// (line i is a parsed event with seq === i). This is the preservable region:
// it includes any fully-written events of an interrupted final turn AFTER the
// last turn/end — those are real, durably-written work and must NOT be
// truncated (a single turn can be huge in a long-horizon task; the orphaned
// open turn is closed with a synthetic turn/end on reload, not discarded —
// ADR 0018). The walk stops at the first hole (unparseable line or seq gap):
// - if that hole is AT OR BEFORE the last committed turn/end, committed data
// was damaged → the session is unloadable (throw);
// - if it is AFTER (or there is no committed turn/end yet), it is the
// tolerated crash boundary — a torn final line never fully flushed — and
// it simply bounds the preserved tail.
const preserved: SessionEvent[] = []
for (let i = 0; i < parsed.length; i++) {
const p = parsed[i]
if (!p?.ok || p.event === undefined) {
throw new Error(`corrupt session log: unparsable committed event at line ${i + 1}`)
if (i <= lastTurnEnd) throw new Error(`corrupt session log: unparsable committed event at line ${i + 1}`)
break // torn tail fragment after the last turn/end — stop, tolerate
}
if (p.event.seq !== i) {
throw new Error(`corrupt session log: seq gap in committed region at line ${i + 1} (expected ${i}, got ${p.event.seq})`)
if (i <= lastTurnEnd) throw new Error(`corrupt session log: seq gap in committed region at line ${i + 1} (expected ${i}, got ${p.event.seq})`)
break // gap after the last turn/end — torn tail, stop
}
committed.push(p.event)
preserved.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 }
// committedBytes = end of the last PRESERVED line (header if none): the next
// append truncates any torn bytes past this point before writing the
// synthetic closers + new events.
const lastPreserved = parsed[preserved.length - 1]
const committedBytes = preserved.length > 0 && lastPreserved ? lastPreserved.endByte : headerEntry.endByte
return { meta: metaFrom(headerLine), events: preserved, committedBytes }
}
/** Build the load-time {@link SessionMeta} from a header line (summary overlaid later). */

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

@@ -27,7 +27,7 @@ import { open, mkdir, readFile, readdir, rename, link, rm, truncate } from 'node
import { resolve } from 'node:path'
import { randomBytes } from 'node:crypto'
import { SessionPersistence } from '@deepseek-ai/dsh-session-persistence'
import { isJsonValue } from '@deepseek-ai/dsh-session'
import { isJsonValue, interruptedTurnClosers } from '@deepseek-ai/dsh-session'
import type { Session, SessionEvent, SessionId, SessionMeta, SessionSummary } from '@deepseek-ai/dsh-session'
import {
encodeSegment, eventLine, logPath, parseHeaderMeta, scanLog, sessionDir, sidecarPath, toHeaderLine,
@@ -58,11 +58,6 @@ interface SessionState {
* new session's events to be dropped against the old cursor).
*/
owner?: Session
/**
* If a load truncation-repair is pending, the byte offset to truncate the
* file to before the next append (discards the never-committed crash tail).
*/
repairTo?: number
}
/**
@@ -238,13 +233,6 @@ export class SessionPersistenceJsonl extends SessionPersistence {
let state = this.states.get(id)
if (state === undefined) state = await this.adopt(id) // calls loadCore, not load
// Truncation-repair: on the first append after a load that found a crash
// tail, physically discard the orphaned bytes before writing.
if (state.repairTo !== undefined) {
await this.repair(state, state.repairTo)
delete state.repairTo
}
// Contiguity contract: each event's seq must continue the stored log.
for (const [i, event] of events.entries()) {
if (event.seq !== state.cursor + i) {
@@ -280,19 +268,42 @@ export class SessionPersistenceJsonl extends SessionPersistence {
const summary = await this.readSidecar(id, meta.cwd)
const fullMeta: SessionMeta = { ...meta, ...summary }
// Record the state so the next append repairs the crash tail (if any) and
// continues at the committed length. The state keeps its OWN copy of the
// meta; the value returned to the caller is a SEPARATE copy so a consumer
// mutating `loaded.meta` (e.g. `cwd`) cannot corrupt the backend's pathing
// metadata and send later reads/writes to the wrong log.
const needsRepair = committedBytes < buffer.byteLength
this.states.set(id, {
// 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 — ADR 0018).
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: { ...fullMeta },
cursor: events.length,
materialized: true,
...needsRepair ? { repairTo: committedBytes } : {},
})
return { meta: fullMeta, events }
}
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. No sidecar touch here: load is not a
// summary-changing op (the closers carry no new title/firstPrompt), and
// the next real append bumps `updatedAt` — keeping the summary write off
// the recovery path avoids a second best-effort failure mode.
await this.appendLines(state, closers)
state.cursor = balanced.length
}
return { meta: fullMeta, events: balanced }
}
async list(): Promise<SessionMeta[]> {

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

@@ -107,36 +107,41 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => {
expect(loaded.events).toEqual(log) // chunks preserved, contiguous seqs
})
it('crash tolerance: load truncates an uncommitted final turn back to the last turn/end', async () => {
it('crash recovery: load preserves the interrupted turn and closes it with a synthetic turn/end {interrupted}', async () => {
const m = meta('crash', '/proj')
await ctx.sessionPersistence.create(m)
await ctx.sessionPersistence.append(m.id, oneTurnLog()) // seqs 0..5, turn/end at 5
// Simulate a crash mid-second-turn: append raw lines that are NOT closed by
// a turn/end (and a final partial line with no newline).
// 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 tail = [
await writeFile(path, [
JSON.stringify({ type: 'turn/start', seq: 6, time: 8, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } }),
JSON.stringify({ type: 'step/start', seq: 7, time: 9, data: { turn: 2, step: 1 } }),
'{"type":"assistant/chunk","seq":8,"ti', // truncated partial line
].join('\n')
await writeFile(path, tail, { flag: 'a' })
'{"type":"assistant/chunk","seq":8,"ti', // truncated partial line (no newline)
].join('\n'), { flag: 'a' })
// load returns only the committed first turn.
// load PRESERVES the interrupted turn's real events (turn/start 6, step/start
// 7) — a turn can be huge, so they must not be truncated — and durably closes
// the orphaned turn with synthetic step/end (8) + turn/end {interrupted} (9).
const loaded = await ctx.sessionPersistence.load(m.id)
expect(loaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5])
expect(loaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7, 8, 9])
const last = loaded.events.at(-1)!
expect(last.type === 'turn/end' && last.data.reason).toEqual({ kind: 'interrupted' })
const stepEnd = loaded.events[8]!
expect(stepEnd.type).toBe('step/end')
// the torn seq-8 chunk fragment did not survive
expect(loaded.events.some(e => e.type === 'assistant/chunk' && e.seq === 8)).toBe(false)
// The next append repairs the file (discarding the crash tail) and resumes
// at seq 6.
const turn2 = [
{ type: 'turn/start', seq: 6, time: 10, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } },
{ type: 'turn/end', seq: 7, time: 11, data: { turn: 2, reason: { kind: 'completed' } } },
// The next append continues at seq 10 (the balanced length).
const turn3 = [
{ type: 'turn/start', seq: 10, time: 11, data: { turn: 3, trigger: { kind: 'message', source: { kind: 'user' } } } },
{ type: 'turn/end', seq: 11, time: 12, data: { turn: 3, reason: { kind: 'completed' } } },
] as SessionEvent[]
await ctx.sessionPersistence.append(m.id, turn2)
await ctx.sessionPersistence.append(m.id, turn3)
const reloaded = await ctx.sessionPersistence.load(m.id)
expect(reloaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7])
// and no orphaned seq-8 chunk survived
expect(reloaded.events.some(e => e.seq === 8)).toBe(false)
expect(reloaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11])
})
it('committed events are never rewritten: only the crash tail is repaired', async () => {
@@ -490,15 +495,17 @@ describe('SessionPersistenceJsonl: scanLog unit', () => {
expect(() => scanLog(Buffer.from('{"type":"event"}\n'))).toThrow(/session header/)
})
it('a seq gap with NO committed turn/end yields zero committed events (uncommitted tail)', () => {
it('a seq gap after the last turn/end bounds the preserved tail (torn fragment tolerated)', () => {
const log = [
JSON.stringify({ type: 'session', version: 1, id: 'g', createdAt: 1 }),
JSON.stringify({ type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } }),
JSON.stringify({ type: 'step/start', seq: 2, time: 2, data: { turn: 1, step: 1 } }), // gap: missing seq 1
].join('\n') + '\n'
// Nothing reached a turn/end, so nothing is committed — the whole region is
// an uncommitted (crash) tail. Safe to load as empty, NOT a corruption.
expect(scanLog(Buffer.from(log)).events).toEqual([])
// No committed turn/end, so the gap is a tolerated crash boundary: scanLog
// PRESERVES the contiguous prefix (turn/start seq 0) — real interrupted-turn
// work, not discarded — and stops at the gap. The orphaned open turn is
// closed by loadCore's synthetic turn/end, not here.
expect(scanLog(Buffer.from(log)).events.map(e => e.seq)).toEqual([0])
})
it('rejects a seq gap BEFORE a later committed turn/end (committed data damaged)', () => {
@@ -522,13 +529,23 @@ describe('SessionPersistenceJsonl: scanLog unit', () => {
expect(() => scanLog(Buffer.from(log))).toThrow(/unparsable committed event/)
})
it('a corrupt line with NO committed turn/end yields zero committed events', () => {
it('a header-only log (no event lines at all) preserves nothing — committedBytes is the header', () => {
const log = JSON.stringify({ type: 'session', version: 1, id: 'h0', createdAt: 1 }) + '\n'
const scanned = scanLog(Buffer.from(log))
expect(scanned.events).toEqual([])
// committedBytes falls back to the header line's end (no preserved events).
expect(scanned.committedBytes).toBe(Buffer.byteLength(log, 'utf8'))
})
it('a corrupt line after the last turn/end bounds the preserved tail', () => {
const log = [
JSON.stringify({ type: 'session', version: 1, id: 'c2', createdAt: 1 }),
JSON.stringify({ type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } }),
'{not json', // corrupt crash fragment, no turn/end committed
].join('\n') + '\n'
expect(scanLog(Buffer.from(log)).events).toEqual([])
// The contiguous prefix (turn/start seq 0) is preserved; the corrupt
// fragment after it is the tolerated crash boundary.
expect(scanLog(Buffer.from(log)).events.map(e => e.seq)).toEqual([0])
})
it('tolerates a seq gap AFTER a turn/end (uncommitted tail)', () => {
@@ -1066,13 +1083,19 @@ describe('SessionPersistenceJsonl: edge cases', () => {
await ctx2.fiber.dispose()
})
it('a header-only log (no turn/end) loads as zero committed events', () => {
const log = [
JSON.stringify({ type: 'session', version: 1, id: 'open', createdAt: 1 }),
JSON.stringify({ type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } }),
].join('\n') + '\n'
const { events } = scanLog(Buffer.from(log))
expect(events).toEqual([]) // nothing committed (no turn/end)
it('a header-only log (open turn, no turn/end) preserves the open turn on load and closes it', async () => {
// A session whose only durable content is an unclosed first turn. scanLog
// preserves the turn/start; loadCore closes it with a synthetic
// turn/end {interrupted} so the returned log is balanced.
const m = meta('open-turn', '/h')
await ctx.sessionPersistence.create(m)
await ctx.sessionPersistence.append(m.id, [
{ type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } },
] as SessionEvent[])
const { events } = await ctx.sessionPersistence.load(m.id)
expect(events.map(e => e.type)).toEqual(['turn/start', 'turn/end'])
const end = events[1]!
expect(end.type === 'turn/end' && end.data.reason).toEqual({ kind: 'interrupted' })
})
it('initFor is idempotent: a re-seeded existing session is not re-initialized', async () => {