Merge branch 'split/agent-factory' into split/session-persistence-sqlite

# Conflicts:
#	docs/adr/0018-session-persistence.md
#	docs/architecture.md

packages/agent-loop/src/index.ts

@@ -17,7 +17,7 @@ import { SessionId } from '@deepseek-ai/dsh-session'
 import type { Session } from '@deepseek-ai/dsh-session'
 import type {} from '@deepseek-ai/dsh-system-prompt'
 import type {} from '@deepseek-ai/dsh-tools'
-import type {} from '@deepseek-ai/dsh-session-persistence'
+import type { SessionPersistence } from '@deepseek-ai/dsh-session-persistence'
 import { LoopAgent } from './agent.ts'
 
 export { LoopAgent } from './agent.ts'
@@ -32,7 +32,19 @@ declare module 'cordis' {
 
 export interface Config {
   /** Agents created from configuration at startup. */
-  agents: (AgentOptions & { id: string })[]
+  agents: (AgentOptions & {
+    id: string
+    /**
+     * If set, the config agent RESUMES this persisted session id instead of
+     * starting a fresh `${id}-session-<uuid>`. Sourced from an env var in
+     * cordis.yml (`resumeSessionId: !!js process.env.RESUME_SESSION_ID`), so a
+     * demo can continue a prior conversation without code changes. Requires a
+     * `dsh-session-persistence` backend; the resume is deferred until that
+     * service is available (via `ctx.inject`) and the loaded session's events
+     * seed the live session so history continues.
+     */
+    resumeSessionId?: string
+  })[]
 }
 
 /**
@@ -53,6 +65,7 @@ export class AgentLoop extends Service implements AgentFactory {
       id: z.string().required(),
       model: z.string(),
       systemPrompt: z.string(),
+      resumeSessionId: z.string(),
     })).default([]),
   })
 
@@ -61,8 +74,27 @@ export class AgentLoop extends Service implements AgentFactory {
     // Provide the agent-creation factory to the registry (effect-scoped: the
     // slot is cleared on dispose).
     ctx.effect(() => this.ctx.agents.setFactory(this), 'agentLoop.setFactory()')
-    for (const { id, ...options } of config.agents) {
-      this.create(id, options)
+    for (const { id, resumeSessionId, ...options } of config.agents) {
+      if (resumeSessionId !== undefined && resumeSessionId !== '') {
+        // Resume a prior session instead of starting fresh. resume() needs
+        // `ctx.sessionPersistence`, which may load AFTER this plugin (cordis.yml
+        // lists the backend later). `ctx.inject(['sessionPersistence'], cb)`
+        // runs `cb` with a child ctx once the service exists; the child reads
+        // the persistence and hands it to resumeWith (which uses this.ctx — the
+        // parent — for sessions/registry, all in AgentLoop's static inject). A
+        // failed resume is contained + logged: startup must not crash.
+        ctx.effect(() => {
+          const fiber = this.ctx.inject(['sessionPersistence'], (childCtx: Context) => {
+            void this.resumeWith(childCtx.sessionPersistence, { agentId: id, resumeSessionId, agentOptions: options })
+              .catch((error: unknown) => {
+                this.ctx.logger.warn(`agent "${id}": config-driven resume of "${resumeSessionId}" failed: ${String(error)}`)
+              })
+          })
+          return () => void fiber.dispose()
+        }, `agentLoop.resume(${id})`)
+      } else {
+        this.create(id, options)
+      }
     }
   }
 
@@ -120,7 +152,6 @@ export class AgentLoop extends Service implements AgentFactory {
    * by the time this runs the service exists.
    */
   async resume(options: ResumeAgentOptions): Promise<Agent> {
-    this.assertAgentIdFree(options.agentId)
     const persistence = this.ctx.sessionPersistence
     // `sessionPersistence` is declaration-merged onto Context as non-optional,
     // but the service is only present when a backend plugin is loaded — and
@@ -130,6 +161,20 @@ export class AgentLoop extends Service implements AgentFactory {
     if (persistence === undefined) {
       throw new Error('cannot resume: session persistence is not configured (load a dsh-session-persistence backend)')
     }
+    return this.resumeWith(persistence, options)
+  }
+
+  /**
+   * Resume against an EXPLICIT persistence handle. Factored out of {@link resume}
+   * so the config-driven path can pass the handle it obtained from a
+   * `ctx.inject(['sessionPersistence'], …)` child context: `this.ctx` (the
+   * service's own fiber) did not inject `sessionPersistence`, so reading it
+   * there from inside the inject child trips the cordis inject guard. The
+   * sessions store + registry are still read through `this.ctx` (both are in
+   * AgentLoop's static inject, so they resolve fine).
+   */
+  private async resumeWith(persistence: SessionPersistence, options: ResumeAgentOptions): Promise<Agent> {
+    this.assertAgentIdFree(options.agentId)
     const { meta, events } = await persistence.load(SessionId(options.resumeSessionId))
     // Re-check the agent id AFTER the await: the pre-load check above can go
     // stale while load() is pending (a concurrent resume/create may register the

packages/agent-loop/tests/config-session-id.spec.ts

@@ -1,4 +1,4 @@
-import { afterEach, describe, expect, it } from 'vitest'
+import { afterEach, describe, expect, it, vi } from 'vitest'
 import { Context } from 'cordis'
 import { mkdtemp, rm } from 'node:fs/promises'
 import { tmpdir } from 'node:os'
@@ -62,4 +62,76 @@ describe('config-driven session id', () => {
     await waitForIdle(ctx2, a2)
     await ctx2.fiber.dispose()
   })
+
+  it('config-driven resumeSessionId continues a persisted session (env-var resume)', async () => {
+    const root = await mkdtemp(join(tmpdir(), 'dsh-cfg-resume-'))
+    dirs.push(root)
+
+    // Run 1: a programmatically-created agent on a KNOWN session id persists a
+    // completed turn, so run 2 has a concrete id to resume.
+    const ctx1 = new Context()
+    await ctx1.plugin(LlmService)
+    await ctx1.plugin(SessionStore)
+    await ctx1.plugin(SystemPrompt)
+    await ctx1.plugin(ToolRegistry)
+    await ctx1.plugin(AgentRegistry)
+    await ctx1.plugin(AgentLoop, { agents: [] })
+    await ctx1.plugin(SessionPersistenceJsonl, { root })
+    ctx1.llm.registerAdapter(['mock'], new MockAdapter([textResponse('first')]))
+    const a1 = ctx1.agents.create({ agentId: 'main', sessionId: 'sticky-1' }) as LoopAgent
+    a1.send([{ type: 'text', text: 'remember me' }], { source: { kind: 'user' } })
+    await waitForIdle(ctx1, a1)
+    await ctx1.fiber.dispose()
+
+    // Run 2: a CONFIG agent with resumeSessionId continues that session. The
+    // resume is deferred until sessionPersistence loads (ctx.inject), so wait
+    // for the agent to appear, then assert it is on the resumed id with history.
+    const ctx2 = new Context()
+    await ctx2.plugin(LlmService)
+    await ctx2.plugin(SessionStore)
+    await ctx2.plugin(SystemPrompt)
+    await ctx2.plugin(ToolRegistry)
+    await ctx2.plugin(AgentRegistry)
+    await ctx2.plugin(AgentLoop, { agents: [{ id: 'main', model: 'mock', systemPrompt: '', resumeSessionId: 'sticky-1' }] })
+    await ctx2.plugin(SessionPersistenceJsonl, { root })
+    ctx2.llm.registerAdapter(['mock'], new MockAdapter([textResponse('second')]))
+
+    // The deferred resume runs on a microtask after the backend is available.
+    let resumed: LoopAgent | undefined
+    for (let i = 0; i < 50 && !resumed; i++) {
+      await new Promise(r => setTimeout(r, 5))
+      resumed = ctx2.agents.get('main') as LoopAgent | undefined
+    }
+    expect(resumed).toBeDefined()
+    // The live session id IS the resumed id (NOT a fresh ${id}-session-<uuid>),
+    // and the prior turn's user message is in the derived history.
+    expect(resumed!.session.id).toBe('sticky-1')
+    const derived = resumed!.session.deriveMessages()
+    expect(JSON.stringify(derived)).toContain('remember me')
+    await ctx2.fiber.dispose()
+  })
+
+  it('config-driven resume of a missing session is contained: logs a warning, no agent, no crash', async () => {
+    const root = await mkdtemp(join(tmpdir(), 'dsh-cfg-resume-miss-'))
+    dirs.push(root)
+    const ctx = new Context()
+    await ctx.plugin(LlmService)
+    await ctx.plugin(SessionStore)
+    await ctx.plugin(SystemPrompt)
+    await ctx.plugin(ToolRegistry)
+    await ctx.plugin(AgentRegistry)
+    await ctx.plugin(AgentLoop, { agents: [{ id: 'main', model: 'mock', systemPrompt: '', resumeSessionId: 'does-not-exist' }] })
+    const warn = vi.spyOn((ctx.agentLoop as unknown as { ctx: { logger: { warn: (...a: unknown[]) => void } } }).ctx.logger, 'warn')
+      .mockImplementation(() => undefined)
+    await ctx.plugin(SessionPersistenceJsonl, { root })
+    ctx.llm.registerAdapter(['mock'], new MockAdapter([textResponse('x')]))
+
+    // The deferred resume fails (no such session on disk). It must be contained:
+    // a warning is logged, no 'main' agent is registered, and the app stays up.
+    await new Promise(r => setTimeout(r, 200))
+    expect(ctx.agents.get('main')).toBeUndefined()
+    expect(warn).toHaveBeenCalledWith(expect.stringContaining('config-driven resume of "does-not-exist" failed'))
+    warn.mockRestore()
+    await ctx.fiber.dispose()
+  })
 })

packages/session-persistence-jsonl/README.md

@@ -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.
 

packages/session-persistence-jsonl/src/format.ts

@@ -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). */

packages/session-persistence-jsonl/src/index.ts

@@ -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[]> {

packages/session-persistence-jsonl/tests/jsonl.spec.ts

@@ -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 () => {

packages/session-persistence/README.md

@@ -10,14 +10,14 @@ The persisted unit IS the existing `SessionEvent` (event-sourced model — the l
 |---|---|
 | `create(meta): Promise<void>` | Register a new session's metadata. MAY defer the physical write until the first `append` (lazy materialization). |
 | `append(id, events): Promise<void>` | Durably persist a batch (from the `session/flush` drain). Append-only; first event `seq` == stored next-seq after any repair; rejects non-JSON-serializable data naming the offending type. |
-| `load(id): Promise<{ meta; events }>` | Reload meta + log up to the last complete `turn/end`; events contiguous (`events[i].seq === i`); rejects a mid-log gap/parse error or unknown `version`. |
+| `load(id): Promise<{ meta; events }>` | Reload meta + log. Preserves an interrupted (unclosed) final turn and closes it with synthetic `step/end?`+`turn/end {interrupted}` (a turn can be huge — never truncated); only a torn tail fragment is dropped. Events contiguous (`events[i].seq === i`); rejects a committed-region gap/parse error or unknown `version`. |
 | `list(): Promise<SessionMeta[]>` | Lightweight listing from metadata, no full-log parse. |
 | `has(id)` / `delete(id)` | Existence / removal. A zero-event lazily-materialized session is absent from `has`/`list`. |
 | `update(id, summary): Promise<void>` | Update mutable `SessionSummary` fields without touching the append-only log. |
 
 ## Invariants every backend must honor
 
-- **Append-only.** Committed events (at or below a flushed `turn/end`) are never rewritten. The only exception is the one-time truncation-repair of a never-committed crash tail on the first `append` after a `load`.
+- **Append-only; a crashed turn is closed, not truncated.** Committed events (at or below a flushed `turn/end`) are never rewritten. A crash can leave an unclosed final turn whose events are real and possibly large; `load` preserves them and durably appends synthetic closers (`step/end?`+`turn/end {interrupted}`) to balance the log. Only a never-fully-written torn tail fragment is discarded.
 - **Contiguous seq.** `load` rejects a `seq` gap/parse error in the MIDDLE of the log; `append`'s first `seq` must equal the stored next-seq.
 - **JSON-serializable data.** `append` rejects non-serializable `event.data`; backends snapshot each event when buffering (the live `session.events` object is mutable).
 - **Durability.** `append` returns only once the batch is durable.

packages/session-persistence/src/index.ts

@@ -42,14 +42,16 @@ declare module 'cordis' {
  * Contracts every implementation MUST honor (a DB backend asserts them inside
  * a transaction; a file backend appends at EOF):
  *
- * - **Append-only.** Committed events — those at or below a flushed `turn/end`
- *   — are never rewritten. The ONLY exception is the one-time truncation-repair
- *   of a never-committed crash tail on the first {@link append} after a
- *   {@link load} (see {@link load}).
+ * - **Append-only; a crashed turn is closed, not truncated.** Committed events
+ *   — those at or below a flushed `turn/end` — are never rewritten. A crash can
+ *   leave an unclosed final turn whose events are real (and possibly large);
+ *   {@link load} preserves them and closes the orphaned turn with synthetic
+ *   boundary events (see {@link load}). Only a never-fully-written torn tail
+ *   fragment is discarded.
  * - **Contiguous seq.** A persisted log is contiguous: `events[i].seq === i`.
- *   {@link load} rejects a parse error or a `seq` gap in the MIDDLE of the log
+ *   {@link load} rejects a parse error or a `seq` gap in the COMMITTED region
  *   (unloadable); {@link append}'s first event `seq` MUST equal the backend's
- *   stored next-seq after any repair.
+ *   stored next-seq (after `load` has balanced any interrupted turn).
  * - **JSON-serializable data.** `SessionEventMap` is merge-extensible and
  *   `event.data` is typed only as `SessionEventMap[K]`, so {@link append}
  *   REJECTS non-JSON-serializable data with an error naming the offending
@@ -75,9 +77,9 @@ export abstract class SessionPersistence extends Service {
   /**
    * Durably persist a batch of events (called from the write-behind drain at
    * the `session/flush` checkpoint). Honors the append-only and contiguous-seq
-   * contracts: the first event's `seq` MUST equal the stored next-seq after
-   * any truncation-repair of a crash tail. Rejects non-JSON-serializable
-   * `event.data` with an error naming the offending event type.
+   * contracts: the first event's `seq` MUST equal the stored next-seq (after
+   * `load` has durably closed any interrupted turn). Rejects non-JSON-
+   * serializable `event.data` with an error naming the offending event type.
    */
   abstract append(id: SessionId, events: readonly SessionEvent[]): Promise<void>
 
@@ -86,13 +88,19 @@ export abstract class SessionPersistence extends Service {
    * durable checkpoint. Returns `meta` AND `events` so the live session is
    * reconstructed with its `cwd`/lineage, not just its log.
    *
-   * The loop only flushes at `turn/end`, so a crash can leave a half-written
-   * final turn below the last committed checkpoint. `load` returns events only
-   * up to the **last complete `turn/end`**; a subsequent {@link append} runs
-   * the one-time truncation-repair that physically discards the orphaned tail
-   * before writing. Returned events are contiguous (`events[i].seq === i`); a
-   * parse error or a `seq` gap in the MIDDLE of the log makes the session
-   * unloadable (reject). Rejects an unknown format `version`.
+   * The loop only flushes at `turn/end`, so a crash can leave a durable log
+   * whose final turn never closed: real, fully-written events sit after the last
+   * `turn/end`. Those events are PRESERVED — a single turn can be huge in a
+   * long-horizon task, so truncating it would destroy real work — and `load`
+   * CLOSES the orphaned turn by durably appending the minimal synthetic boundary
+   * events (a `step/end` if a step was open, then a `turn/end` carrying the
+   * `{ kind: 'interrupted' }` reason). The returned `events` therefore end on a
+   * balanced `turn/end` and are immediately usable as a session seed. Only a
+   * never-fully-written TORN tail fragment (a half-written final record) is
+   * discarded. Returned events are contiguous (`events[i].seq === i`); a parse
+   * error or a `seq` gap in the COMMITTED region (at or before the last real
+   * `turn/end`) makes the session unloadable (reject). Rejects an unknown format
+   * `version`. See ADR 0018 for the crash-recovery contract.
    */
   abstract load(id: SessionId): Promise<{ meta: SessionMeta; events: SessionEvent[] }>
 

packages/session-persistence/tests/contract.ts

@@ -64,6 +64,44 @@ export function runPersistenceContract(name: string, make: () => Promise<Contrac
       }
     })
 
+    it('crash recovery: load preserves an interrupted (unclosed) turn and closes it with turn/end {interrupted}', async () => {
+      const { persistence, dispose } = await make()
+      try {
+        const m = meta('interrupted')
+        await persistence.create(m)
+        await persistence.append(m.id, oneTurnLog()) // turn 1, committed (seqs 0..5)
+        // A second turn that crashed mid-flight: turn/start + step/start were
+        // durably written, but no step/end / turn/end ever arrived.
+        await persistence.append(m.id, [
+          { 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 } },
+        ])
+
+        // load PRESERVES the interrupted turn's events (a turn can be huge — they
+        // must not be truncated) and closes the orphaned turn with synthetic
+        // boundary events: step/end (the step was open) then turn/end {interrupted}.
+        const loaded = await persistence.load(m.id)
+        expect(loaded.events.map(e => e.type)).toEqual([
+          'turn/start', 'user/message', 'step/start', 'assistant/message', 'step/end', 'turn/end', // turn 1
+          'turn/start', 'step/start', 'step/end', 'turn/end', // turn 2: real events + synthetic closers
+        ])
+        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' })
+
+        // The closed log is durable and continuable: a fresh append continues at
+        // the balanced length (seq 10), and a reload round-trips identically.
+        await persistence.append(m.id, [
+          { type: 'turn/start', seq: 10, time: 9, data: { turn: 3, trigger: { kind: 'message', source: { kind: 'user' } } } },
+          { type: 'turn/end', seq: 11, time: 10, data: { turn: 3, reason: { kind: 'completed' } } },
+        ])
+        const reloaded = await persistence.load(m.id)
+        expect(reloaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11])
+      } finally {
+        await dispose()
+      }
+    })
+
     it('has()/list() exclude a created-but-never-appended (zero-event) session', async () => {
       const { persistence, dispose } = await make()
       try {

packages/session-persistence/tests/persistence.spec.ts

@@ -1,6 +1,6 @@
 import { describe, expect, it } from 'vitest'
 import { Context } from 'cordis'
-import { SessionId, isJsonValue } from '@deepseek-ai/dsh-session'
+import { SessionId, isJsonValue, interruptedTurnClosers } from '@deepseek-ai/dsh-session'
 import type { SessionEvent, SessionMeta, SessionSummary } from '@deepseek-ai/dsh-session'
 import { SessionPersistence } from '../src/index.ts'
 import { runPersistenceContract, meta, oneTurnLog } from './contract.ts'
@@ -46,6 +46,11 @@ class MemoryPersistence extends SessionPersistence {
   async load(id: SessionId): Promise<{ meta: SessionMeta; events: SessionEvent[] }> {
     const entry = this.store.get(id)
     if (!entry) throw new Error(`session "${id}" not found`)
+    // Honor the crash-recovery contract: if the stored log ends mid-turn, close
+    // the orphaned turn durably with synthetic boundary events and continue from
+    // the balanced length.
+    const closers = interruptedTurnClosers(entry.events)
+    if (closers.length > 0) entry.events.push(...structuredClone(closers))
     return { meta: structuredClone(entry.meta), events: structuredClone(entry.events) }
   }
 

packages/session/src/index.ts

@@ -15,6 +15,7 @@ import { isJsonValue } from './json.ts'
 
 export * from './types.ts'
 export { isJsonValue } from './json.ts'
+export { interruptedTurnClosers } from './repair.ts'
 
 declare module 'cordis' {
   interface Context {

packages/session/src/repair.ts

@@ -0,0 +1,79 @@
+/**
+ * Crash-recovery repair for an interrupted session log.
+ *
+ * A persistence backend flushes only at `turn/end`, so a crash can leave a
+ * durable log whose final turn never closed: real, fully-written events sit
+ * after the last `turn/end` with no closing boundary. A single turn can be huge
+ * in a long-horizon task (many steps, large tool output), so those events MUST
+ * be preserved — truncating the turn would silently destroy real work. Instead,
+ * on reload the backend CLOSES the orphaned turn by appending the minimal
+ * synthetic boundary events (a `step/end` if a step was still open, then a
+ * `turn/end` carrying the merge-extensible `{ kind: 'interrupted' }` reason).
+ * The marker records that the turn was cut short by a crash, not completed by
+ * the model. See ADR 0018.
+ *
+ * This module computes those synthetic closers from an event list; the backend
+ * returns them inline from `load` (so the reconstructed session is balanced and
+ * immediately usable) and persists them on the first post-load `append`.
+ *
+ * @module @deepseek-ai/dsh-session/repair
+ */
+
+import type { SessionEvent } from './types.ts'
+
+/**
+ * Scan `events` for an open turn/step at the tail and return the synthetic
+ * boundary events that close them, with `seq` continuing the log and `time`
+ * copied from the last real event (the closers stand in for the crash moment;
+ * reusing the last timestamp keeps them deterministic and never invents a
+ * "future" time). Returns an empty array when the log is already balanced
+ * (ends on a `turn/end`, or is empty) — the common, non-crash case.
+ *
+ * Only the LAST turn can be open: the invariants plugin guarantees a `turn/end`
+ * before any later `turn/start`, so an interior open turn is impossible in a
+ * valid committed log. Likewise at most one step is open within that turn.
+ */
+export function interruptedTurnClosers(events: readonly SessionEvent[]): SessionEvent[] {
+  let openTurn: number | null = null
+  let openStep: number | null = null
+  for (const event of events) {
+    switch (event.type) {
+      case 'turn/start':
+        openTurn = event.data.turn
+        break
+      case 'turn/end':
+        openTurn = null
+        openStep = null
+        break
+      case 'step/start':
+        openStep = event.data.step
+        break
+      case 'step/end':
+        openStep = null
+        break
+      // Other event types do not move the turn/step boundary cursor.
+      default:
+        break
+    }
+  }
+
+  // Balanced log (no crash mid-turn): nothing to close. An open turn implies
+  // `events` is non-empty (its turn/start was logged), so `last` exists.
+  const last = events.at(-1)
+  if (openTurn === null || last === undefined) return []
+
+  // The last real event supplies the seq base and the timestamp for the
+  // synthetic closers (reusing the last timestamp keeps them deterministic and
+  // never invents a "future" time).
+  let seq = last.seq + 1
+  const time = last.time
+  const closers: SessionEvent[] = []
+
+  // Close an open step first — a turn/end while a step is open is an invariant
+  // violation, so the step's boundary must be synthesized before the turn's.
+  if (openStep !== null) {
+    closers.push({ type: 'step/end', seq: seq++, time, data: { turn: openTurn, step: openStep } })
+  }
+  closers.push({ type: 'turn/end', seq: seq++, time, data: { turn: openTurn, reason: { kind: 'interrupted' } } })
+  return closers
+}

packages/session/src/types.ts

@@ -99,6 +99,17 @@ export interface TurnEndReasonMap {
   aborted: { kind: 'aborted'; reason?: string }
   error: { kind: 'error'; message: string; code?: string }
   disposed: { kind: 'disposed' }
+  /**
+   * The turn never ended on its own: the process crashed mid-turn and a
+   * persistence backend later closed the orphaned (open) turn on reload so the
+   * log stays balanced. SYNTHESIZED by the backend's crash-recovery repair — no
+   * loop ever emits this. Its events are real (they were durably appended before
+   * the crash) and are PRESERVED, not discarded: a single turn can be huge in a
+   * long-horizon task (many steps, large tool output), so truncating it would
+   * lose real work. The marker records that the turn was cut short, not that the
+   * model completed it. See ADR 0018.
+   */
+  interrupted: { kind: 'interrupted' }
 }
 
 export type TurnEndReason = TurnEndReasonMap[keyof TurnEndReasonMap]