Merge pull request #33 from deepseek-ai/split/session-persistence

feat(session-persistence): abstract seam + JSONL backend + wiring (split 3/5)

.github/workflows/ci.yml

@@ -75,5 +75,7 @@ jobs:
           echo "$out"
           echo "$out" | grep -q '\[tool call\] echo({"text":"ci smoke"})'
           echo "$out" | grep -q '\[tool result\] ECHO: CI SMOKE'
-          test -f examples/echo-agent/main-session.jsonl
+          # The JSONL backend (root ./.sessions, no cwd → _no-cwd bucket) writes a
-          rm -f examples/echo-agent/*.jsonl
+          # per-run session log named main-session-<uuid>.jsonl. Assert one exists.
+          ls .sessions/_no-cwd/main-session-*.jsonl >/dev/null
+          rm -rf .sessions

.gitignore

@@ -6,6 +6,8 @@ lib/
 pnpm-debug.log
 .pnpm-store/
 examples/*/*.jsonl
+.sessions/
+examples/*/.sessions/
 coverage/
 .doc-typecheck-*/
 .vscode/

docs/adr/0017-turn-enclosure-invariant.md

@@ -4,7 +4,7 @@ Status: accepted (2026-06-15)
 
 ## Context
 
-A durable session-persistence backend (added in a companion change) uses the **turn** as its crash-recovery boundary: `load` returns events only up to the last complete `turn/end`, and the first post-load `append` truncates whatever follows as a never-committed crash tail. This is safe only if nothing *legitimately* durable can sit after the last `turn/end`.
+A durable session-persistence backend (added in a companion change) uses the **turn** as its crash-recovery boundary: a crash can leave an unclosed final turn, which `load` closes with a synthetic `turn/end {kind:'interrupted'}` while preserving the turn's real events (see [ADR 0018](0018-session-persistence.md)). This recovery is only well-defined if nothing *legitimately* durable sits OUTSIDE a turn — between the last `turn/end` and the next `turn/start` — since such an event would be swept into the next turn's interrupted close.
 
 That assumption did not hold. Two paths recorded events outside any turn:
 

docs/adr/0018-session-persistence.md

@@ -0,0 +1,30 @@
+# ADR 0018: Session persistence as an abstract service over the existing `SessionEvent`
+
+Status: accepted (2026-06-15)
+
+## Context
+
+Sessions lived only in memory. The example `session-jsonl.ts` plugin (duplicated byte-for-byte in both examples) was write-only telemetry: it buffered `session/event` and appended JSON lines, with no read/replay path, no crash-safety (no fsync, no atomic write, a fire-and-forget dispose drain), no listing, and no format versioning. Nothing could rehydrate a past session from disk into a live agent, so durable resume ("continue yesterday's task"), durable forking, and the ACP `session/load` method (RFC 010) were all impossible.
+
+The [event-sourced model](0003-event-sourced-sessions.md) makes the append-only log the single source of truth and derives LLM history from it. Persistence had to stay faithful to that: persist the existing `SessionEvent` directly, with no parallel "persisted message" type that the log is converted to and from. The backend also had to be swappable — a file store now, a database store later — behind one interface.
+
+## Decision
+
+Persistence is an abstract **capability seam** ([ADR 0009](0009-capability-seams.md), the `dsh-bash` template), not loop or core logic:
+
+1. **Interface** (`dsh-session-persistence`, `ctx.sessionPersistence`) — an abstract `SessionPersistence` service: `create`/`append`/`load`/`list`/`has`/`delete`/`update`. Its persisted unit IS the existing `SessionEvent` (`{ type, seq, time, data }`), reused verbatim — no conversion type.
+2. **Implementation** (`dsh-session-persistence-jsonl`) — an append-only JSONL log per session (a `SessionHeader` line then one `SessionEvent` per line, verbatim **including `assistant/chunk`**) plus an atomic `.summary.json` sidecar for the mutable `SessionSummary`.
+
+Key choices recorded here because they are durable, contested, and surprising:
+
+- **The canonical durable log persists every `SessionEvent` verbatim, including `assistant/chunk`.** `deriveMessages()` skips chunks, and a chunk-filtered rollout (Codex's `policy.rs`) is tempting — but `seq = log.length` and the load-validation `events[i].seq === i` require a *contiguous* log; filtering chunks out would leave holes and break both the contract and resume. A chunk-filtered projection is possible later as a derived view with its own renumbering, but it is NOT the canonical log.
+- **Append-only; a crashed turn is closed, never truncated.** Committed events — those at or below a flushed `turn/end` — are never rewritten. 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`. **A single turn can be huge in a long-horizon task** (many steps, large tool output spanning a long autonomous run), so discarding the interrupted turn would silently destroy a large amount of real work — truncating a turn is wrong. Instead, on reload `load` PRESERVES those events and CLOSES the orphaned turn by durably appending the minimal synthetic boundary events: an error `tool/result` for every `tool-call` the crash left unanswered, then a `step/end` if a step was still open, then a `turn/end` carrying the merge-extensible `{ kind: 'interrupted' }` reason (a marker that records the turn was cut short by a crash, not completed by the model — no loop ever emits it). The synthetic tool results matter for resume correctness: the loop logs the `assistant/message` (carrying the `tool-call` blocks) BEFORE running the tools, so a crash mid-tool leaves calls without results; `deriveMessages()` would then replay a dangling assistant tool-call, which every provider rejects as an invalid transcript on the next request. Answering each orphaned call with an error result keeps the rehydrated history valid. `load` returns the balanced log, so a resumed session is immediately usable. The ONLY thing discarded is a never-fully-written **torn tail fragment** — a final record whose bytes (JSONL) or row were never completely flushed; that fragment is not a valid event and is dropped before the synthetic closers are written. A parse error or `seq` gap in the COMMITTED region (at or before the last real `turn/end`) is genuine corruption and makes the session unloadable.
+- **File backend canonical, DB backend a drop-in.** `SessionEvent` maps 1:1 onto a row `(session_id, seq, type, time, data)` — `append` is INSERT (in a transaction asserting the contiguous-seq contract), `load` is SELECT … ORDER BY seq. A future `dsh-session-persistence-sqlite` is a `SessionPersistence` subclass with no interface change (opencode runs this exact shape on SQLite/WAL).
+- **Metadata is out-of-log.** Format version, cwd, and lineage are storage concerns, not replayable conversation state, so they live in a `SessionMeta` (`SessionHeader & SessionSummary`) owned by `dsh-session` and attached to a `Session` via a new readonly `session.header` — never in `SessionEventMap`, never reaching `deriveMessages()`. The alternative (a merge-extensible `session/meta` event as log line 0) was rejected: an in-log event would ride along with a seeded/forked session for free, but metadata is not replayable state, so the explicit out-of-log header seam is the cleaner cost.
+- **`load` returns a resumable event log, not just bytes.** `load(sessionId)` yields the `SessionMeta` plus the committed `SessionEvent[]` (through the last complete `turn/end`), shaped so a caller can reconstruct a live session with the loaded events as seed (so `lastTurnNumber`/`deriveMessages` continue) on the SAME session id. The agent-facing create/resume factory that consumes this is a separate seam (a follow-up on `ctx.agents`); the persistence layer deliberately stops at the `load` primitive and does NOT reach into the loop. The agent-loop does NOT hard-inject `sessionPersistence` (that would pend non-persistent demos forever), so any resume path built on this rejects with a clear error when the backend is absent.
+
+Format versioning: the header carries a `version`; `load` rejects an unknown version (no v1 migration). Stated honestly: append-only + flush is robust to partial trailing writes (tolerated on load) but not to fsync-less power loss mid-line; a DB/WAL backend is the stronger option later.
+
+## Consequences
+
+Two new packages and the metadata seam in `dsh-session` (`session.header`, the `create(id?, options?)` signature). Bought: durable resume/fork, a read/replay path, crash tolerance, and the foundation RFC 010's `session/load` needs — all over the existing event-sourced log, with the backend swappable behind one interface. The reusable `runPersistenceContract` suite holds every backend to the same append-only / contiguous-seq / lazy-materialization / serializability semantics. This completes [ADR 0003](0003-event-sourced-sessions.md)'s deferred "real persistence backend" and resolves its `TODO(review)` on the event vocabulary: persisting the log freezes its shape, and the `assistant/chunk` fidelity question is answered above (persist verbatim).

docs/adr/README.md

@@ -29,3 +29,4 @@ Do NOT write an ADR for: a mechanical or local choice (a variable name, a one-fi
 | [0015](0015-structured-error-taxonomy.md) | Structured error taxonomy (HarnessError base) | accepted |
 | [0016](0016-pnpm-over-yarn.md) | pnpm as the package manager instead of Yarn 4 | accepted |
 | [0017](0017-turn-enclosure-invariant.md) | Every session event is enclosed in a turn | accepted |
+| [0018](0018-session-persistence.md) | Session persistence as an abstract service over `SessionEvent` | accepted |

docs/architecture.md

@@ -22,11 +22,13 @@ Requirement context: [Coding Harness MVP 需求分析][mvp-doc].
 │  @deepseek-ai/dsh-agent-loop      (the ONE concrete plugin)  │
 │  @deepseek-ai/dsh-bash-local      (bash impl)                │
 │  @deepseek-ai/dsh-tool-bash       (bash tool schemas)        │
+│  @deepseek-ai/dsh-session-persistence-jsonl (persistence impl)│
 ├─────────────────────────────────────────────────────────────┤
 │  @deepseek-ai/dsh-agent           (vocabulary + registry)    │
 │  @deepseek-ai/dsh-tools           (registry + exec waterfall)│
 │  @deepseek-ai/dsh-system-prompt   (assembly registry)        │
 │  @deepseek-ai/dsh-session         (event-sourced log)        │
+│  @deepseek-ai/dsh-session-persistence (persistence seam)     │
 │  @deepseek-ai/dsh-llm             (abstract model service)   │
 │  @deepseek-ai/dsh-bash            (abstract bash executor)   │
 ├─────────────────────────────────────────────────────────────┤
@@ -43,6 +45,7 @@ Dependency rule: plugins depend on interface packages, never on `dsh-agent-loop`
 |---|---|---|---|
 | `ctx.llm` | `LlmService` | dsh-llm | adapter registry; `stream()` / `streamBlocks()` / `generate()` |
 | `ctx.sessions` | `SessionStore` | dsh-session | creates/holds event-sourced `Session`s |
+| `ctx.sessionPersistence` | `SessionPersistence` (abstract) | dsh-session-persistence | durable persistence seam: create/append/load/list/update sessions |
 | `ctx.systemPrompt` | `SystemPrompt` | dsh-system-prompt | ordered sections + tool schemas → `assemble()` |
 | `ctx.tools` | `ToolRegistry` | dsh-tools | tool definitions; `execute()` through waterfall |
 | `ctx.agents` | `AgentRegistry` | dsh-agent | live `Agent` handles |
@@ -82,7 +85,7 @@ A `Session` is an append-only log of typed `SessionEvent`s — the single source
 
 Replay/fork = `ctx.sessions.create(id, { seed: seedEvents })`. Trace/telemetry = listen to `session/event`.
 
-**Durability seam**: `session/event` is a synchronous notification; persistence plugins buffer (write-behind) and drain at the awaited `session/flush` checkpoint the loop fires at every turn end (see `examples/echo-agent/src/session-jsonl.ts` for the pattern). **TODO**: real persistence backends (JSONL per session dir, sqlite) are a future phase.
+**Durability seam**: `session/event` is a synchronous notification; persistence plugins buffer (write-behind) and drain at the awaited `session/flush` checkpoint the loop fires at every turn end. The durable backend is a real **capability seam**: the abstract `SessionPersistence` service (`dsh-session-persistence`, `ctx.sessionPersistence`) defines create/append/load/list/update over the existing `SessionEvent` (no parallel persisted type), and `dsh-session-persistence-jsonl` is the first implementation — an append-only JSONL log per session with crash-safe atomic writes, crash recovery that PRESERVES an interrupted turn (closing it with a synthetic `turn/end {interrupted}` rather than truncating — a turn can be huge), and a read/replay path. Session metadata (format version, cwd, lineage) travels separately as `SessionMeta`, attached to a `Session` via `session.header`. `ctx.sessionPersistence.load(sessionId)` returns the committed event log so a caller can reconstruct a live session and continue it. A SQLite/WAL backend is a future drop-in `SessionPersistence` subclass (the row shape `(session_id, seq, type, time, data)` maps 1:1 onto `SessionEvent`).
 
 ## Prompt assembly (dsh-system-prompt)
 
@@ -152,7 +155,7 @@ forever:
 
 Error containment: a throwing `agent/turn-continuation` listener or a broken step ends the **turn** with an `error` event (appended INSIDE the turn, before `turn/end`) — never the driver loop. An adapter that ends its stream with a `finish {kind:'error'}` or `{kind:'aborted'}` chunk (the in-band error path, for adapters that can't throw mid-stream) is likewise translated into a step error, so the turn ends `error`/`aborted` instead of logging a normal `completed` assistant message. `abort()` is honored mid-stream **and** between tool calls; disposal mid-turn ends the turn with reason `disposed` and emits `agent/status('disposed')`.
 
-A failure that happens once the turn is already closed has no in-turn position for a session `error` event (appending one after `turn/end` would put it past a persistence backend's commit boundary, where it is dropped as a crash tail — ADR 0017). So a rejecting `session/flush` (the post-`turn/end` durability checkpoint) and a throwing `agent/turn-end` listener are reported via `agent/error` + the logger only, NOT as a session event; the turn stays balanced and the backend keeps its buffered events for the next flush.
+A failure that happens once the turn is already closed has no in-turn position for a session `error` event (appending one after `turn/end` would put it past the persistence commit boundary, where it is dropped as a crash tail — ADR 0017). So a rejecting `session/flush` (the post-`turn/end` durability checkpoint) and a throwing `agent/turn-end` listener are reported via `agent/error` + the logger only, NOT as a session event; the turn stays balanced and the persistence backend keeps its buffered events for the next flush.
 
 **Turn-enclosure invariant**: every session event lives inside a turn (between a `turn/start` and its `turn/end`). The loop appends queued `user/message` events *after* `turn/start`, and an idle `agent.inject()` wraps its `context/message` in a one-shot `injection` turn. This makes the turn the single durability/replay boundary: a persistence backend can treat anything after the last `turn/end` as an interrupted-crash tail without risking the loss of legitimately-recorded between-turn context. The `dsh-invariants` plugin enforces it in dev (a message event outside an open turn throws). See ADR 0017.
 
@@ -228,8 +231,7 @@ Code skeletons for the three plugin shapes (tool, hook/permission-gate, UI) and
 Tracked here deliberately — each is designed-for but not implemented:
 
 - **Sub-agent spawn/fork semantics** (seam: `AgentLoop.create()`); inter-agent channels beyond `send`/`steer`/events.
-- **Persistence backends** (JSONL session dirs, sqlite) on the `session/event` + `session/flush` seam.
+- **SQLite/WAL persistence backend** — a drop-in `SessionPersistence` subclass (the abstract seam + the JSONL backend landed; see the durability-seam paragraph).
 - **Compaction implementation** (auto thresholds, summarization prompts) on the `agent/request` seam, with its session-event types added by declaration merging.
 - **Parallel tool execution** (concurrency-safety hints on ToolDefinition).
 - **Session branching/tree** (pi-style entry tree) if needed beyond seed-based forking.
-- **Session event vocabulary review** once the loop and a persistence plugin coexist (`TODO(review)` in dsh-session).

docs/module-graph.md

@@ -14,9 +14,12 @@ graph TD
   system-prompt --> llm
   agent --> llm
   agent --> session
+  session-persistence --> session
   invariants --> agent
   invariants --> llm
   invariants --> session
+  session-persistence-jsonl --> session
+  session-persistence-jsonl --> session-persistence
   tools --> agent
   tools --> llm
   tools --> system-prompt
@@ -41,7 +44,9 @@ graph TD
 | `session` | `llm` |
 | `system-prompt` | `llm` |
 | `agent` | `llm`, `session` |
+| `session-persistence` | `session` |
 | `invariants` | `agent`, `llm`, `session` |
+| `session-persistence-jsonl` | `session`, `session-persistence` |
 | `tools` | `agent`, `llm`, `system-prompt` |
 | `agent-loop` | `agent`, `llm`, `session`, `system-prompt`, `tools` |
 | `tool-bash` | `agent`, `bash`, `llm`, `tools` |

docs/rfc/009-session-persistence-and-resumability.md

@@ -1,6 +1,6 @@
 # RFC 009: Durable session persistence — an abstract, append-only, event-based store
 
-Status: proposed
+Status: implemented (see [ADR 0018](../adr/0018-session-persistence.md))
 
 ## Problem
 

docs/rfc/013-typed-event-schemas.md

@@ -0,0 +1,65 @@
+# RFC 013: Runtime schemas for the event vocabulary (Zod vs the merge-extensible-map pattern)
+
+Status: proposed
+
+## Problem
+
+The harness models its core vocabulary — content blocks, message sources, finish reasons, turn triggers, turn-end reasons, and session events — as **merge-extensible maps**: a TypeScript `interface` (e.g. `SessionEventMap`, `ContentBlockMap`) that plugins augment via declaration merging, with the public union derived as `Map[keyof Map]`. This is the repo's universal extension pattern, documented in [docs/architecture.md](../architecture.md) ("The same merge-extensible-map pattern is used for `MessageSource`, `FinishReason`, `TurnTrigger`, and `TurnEndReason`") and relied on by the `defineTool` `InferArgs` DSL and the `assertNever` exhaustiveness convention.
+
+The pattern is **compile-time only**. The types vanish at runtime: there is no schema object to validate an incoming value against, parse untrusted input with, or enumerate at runtime. Two concrete consequences surfaced in review of the session-persistence work (#33):
+
+1. **Persistence treats `event.data` as opaque JSON.** The JSONL/SQLite backends `JSON.stringify`/`JSON.parse` each event verbatim; the only runtime guard is `isJsonValue` (round-trip serializability — rejects BigInt, functions, cycles, non-finite numbers, …), NOT structural validation. A corrupted-but-still-JSON event datum (wrong field types, missing fields) round-trips silently and is only caught later, if at all, by a consumer's `switch`.
+2. **No runtime contract for plugin-added variants.** A plugin that declaration-merges a new `SessionEventMap` key gets compile-time typing for its own code, but nothing validates that the values it produces match the shape it declared — at the producer, at the persistence boundary, or on reload.
+
+A reviewer asked whether the project should move "all the JSON serialization/deserialization" — and ultimately the event vocabulary itself — to **Zod** (or a similar runtime-schema library), so the durable boundary and the plugin extension points are backed by runtime schemas rather than erased types.
+
+This RFC scopes that question. It does **not** propose an implementation; it records the tradeoff so the decision is made deliberately rather than incrementally inside a persistence PR.
+
+## Why this is not a persistence change
+
+It is tempting to read "use Zod for serialization" as a local change to `dsh-session-persistence-jsonl/src/format.ts`. It is not, for one structural reason: **a plugin cannot declaration-merge a Zod schema.** Declaration merging is a TypeScript compile-time mechanism; a Zod schema is a runtime value. To validate events with Zod you need a **runtime registry** that every event-producing package contributes its schema to (e.g. `ctx.sessionEvents.register('compaction/marker', z.object({…}))`), and every consumer reads from. That registry — not the persistence backend — becomes the source of truth for the vocabulary, replacing the merge-extensible interface.
+
+So the real proposal is: **replace the compile-time merge-extensible-map pattern with a runtime schema registry, repo-wide.** That is a core-vocabulary redesign.
+
+## Blast radius (measured)
+
+A migration of the event/vocabulary surface to runtime schemas touches, at minimum:
+
+- **Six merge-extensible maps** (~370 LOC of core types): `ContentBlockMap`, `MessageSourceMap`, `FinishReasonMap` (in `dsh-llm`); `TurnTriggerMap`, `TurnEndReasonMap`, `SessionEventMap` (in `dsh-session`).
+- **~10 `declare module` augmentation sites** across `dsh-agent`, `dsh-agent-loop`, `dsh-bash`, `dsh-llm`, `dsh-session`, `dsh-session-persistence`, `dsh-system-prompt`, `dsh-tools` — each would move from declaration merging to a runtime `register()` call.
+- **The event producers** — 16 `session.append(...)` call sites in the loop — unchanged in shape but now validated at the boundary.
+- **~7 switch-consumers** that branch on these unions: `deriveMessages` (`dsh-session`), `BlockAssembler` (`dsh-llm`), the `dsh-invariants` plugin, both LLM adapters (`dsh-llm-deepseek`, `dsh-llm-pi-ai`), and the tool schema layer (`dsh-tools`). The `assertNever`-on-closed-unions vs fall-through-on-extensible-unions convention (a documented lint rule) would need rethinking — runtime variants are not statically exhaustive.
+- **The `defineTool` `InferArgs` DSL** (`dsh-tools`), which derives zero-cast `execute` arg types from a compile-time schema spec — the showcase of the current approach.
+- **Docs**: architecture.md (the pattern is described as foundational), ADR 0012 (dev-invariants), and any ADR/RFC that references the pattern.
+
+This is a HUGE change. It is not in scope for the RFC-009 session-persistence work and must not be smuggled in through it.
+
+## Options
+
+### A. Status quo — merge-extensible types + `isJsonValue` at the durable boundary
+Keep the compile-time pattern. Persistence stays opaque-JSON + serializability guard. Plugins extend via declaration merging; correctness of event *shape* is the producer's responsibility, enforced by TypeScript at compile time and by the `dsh-invariants` plugin's structural checks in dev.
+
+- **Pros**: zero churn; plugin extension is a one-line `interface` augmentation with full type inference and no runtime registration ceremony; no new runtime dependency; the `defineTool` DSL and `assertNever` exhaustiveness keep working.
+- **Cons**: no runtime structural validation at the persistence boundary or at plugin seams; a malformed-but-JSON datum is caught late.
+
+### B. Header/closed-shape validation only (schemastery), events stay opaque
+Tighten only the genuinely-closed shapes that already have hand-rolled type guards — e.g. the JSONL `HeaderLine` guard (`isHeaderLine`) — using **schemastery** (the repo's existing schema library, already used for every plugin `static Config`). Leave the merge-extensible event union as-is.
+
+- **Pros**: small, fits the existing convention (schemastery, not a new lib); replaces hand-rolled guards on closed shapes with declarative schemas; no core redesign.
+- **Cons**: does not address event-data validation (the thing the reviewer actually asked about); only helps the fixed metadata records.
+
+### C. Runtime schema registry for the whole vocabulary (Zod or schemastery)
+Replace the merge-extensible maps with a runtime registry the producers contribute to and the persistence/consumer paths validate against.
+
+- **Pros**: real runtime validation at the durable boundary and at plugin seams; one source of truth; enables generic tooling (auto-generated docs, fuzzing, wire-format checks).
+- **Cons**: the full blast radius above; **Zod is not currently a direct dependency** (only a transitive dep of `@earendil-works/pi-ai`) and the repo's chosen schema lib is **schemastery** — adopting Zod broadly is itself a dependency decision; declaration-merge ergonomics (one-line plugin extension, full inference) are replaced by runtime registration + manual type wiring; the `assertNever` exhaustiveness guarantee weakens (runtime variants aren't statically exhaustive).
+
+## Recommendation
+
+Defer. Do **not** change #33. If runtime validation is wanted at the durable boundary in the near term, **Option B** (schemastery on the closed header/metadata shapes) is the proportionate step and stays within the existing convention. **Option C** is a genuine architecture decision that should be evaluated on its own merits — including whether the chosen library is Zod or schemastery — and, if accepted, land as its own change with its own ADR, not as a side effect of persistence serialization.
+
+## Open questions
+
+- If a registry is adopted, is the library **schemastery** (already in the tree, already the config schema lib) or **Zod** (richer ecosystem, currently only transitive)? Adopting two schema libraries is a cost in itself.
+- Can a hybrid keep compile-time inference (so `defineTool` and plugin DX survive) while adding an *optional* runtime schema per variant, validated only at the persistence/wire boundary rather than on every in-process append?
+- Does the `dsh-invariants` plugin already cover enough of the runtime-shape gap in dev that boundary validation is only needed for genuinely untrusted input (reload of an externally-modified log)?

docs/rfc/README.md

@@ -16,3 +16,4 @@ Proposals for substantial future work — reviewed before implementation, unlike
 | [010](010-acp-agent-client-protocol.md) | Agent Client Protocol (ACP) support for external editors | proposed |
 | [011](011-acp-multi-session.md) | Multiplex concurrent ACP sessions over one connection | proposed |
 | [012](012-optional-code-mode.md) | Optional Code Mode — model writes TypeScript against an SDK of all tools | proposed |
+| [013](013-typed-event-schemas.md) | Runtime schemas for the event vocabulary (Zod vs the merge-extensible-map pattern) | proposed |

examples/coding-agent/README.md

@@ -29,7 +29,7 @@ Type a coding task. The agent's only tools are `bash` (+ `bash_output` / `bash_k
 | `llm-deepseek` | real `LlmAdapter` via config (`!!js process.env.…` secrets); swap one line to `@deepseek-ai/dsh-llm-pi-ai` for the library-backed twin |
 | `bash` (`dsh-bash-local`) + `tool-bash` | the executor seam + tool schemas as separate plugins |
 | `agent-loop` | agent created from config with a coding system prompt |
-| `src/session-jsonl.ts` | write-behind persistence on `session/event` + `session/flush` (copied from echo-agent) |
+| `session-persistence` (`dsh-session-persistence-jsonl`) | durable JSONL persistence (`root: ./.sessions`): append-only event log per session, crash-safe atomic writes — the shared backend, no per-example file |
 | `src/stdio-chat.ts` | UI as a plugin; copied from echo-agent with reasoning-dimming and an exit-on-idle close handler for piped stdin. Example-local on purpose — extract a shared UI package when a third example needs it |
 
 ## End-to-end tests (`pnpm run test:e2e`, key-gated)

examples/coding-agent/cordis.yml

@@ -74,8 +74,10 @@
           failures before moving on. Verify your work by running the code or
           tests. Keep answers brief and factual.
 
-- id: session-jsonl
+- id: session-persistence
-  name: './src/session-jsonl.ts'
+  name: '@deepseek-ai/dsh-session-persistence-jsonl'
+  config:
+    root: './.sessions'
 
 - id: stdio-chat
   name: './src/stdio-chat.ts'

examples/coding-agent/src/session-jsonl.ts

@@ -1,36 +0,0 @@
-import { appendFile } from 'node:fs/promises'
-import { join } from 'node:path'
-import type { Context } from 'cordis'
-import type { Session, SessionEvent } from '@deepseek-ai/dsh-session'
-
-export const name = 'session-jsonl'
-export const inject = ['sessions']
-
-/**
- * Minimal persistence plugin: buffers session events (write-behind) and
- * drains to a JSONL file at every `session/flush` checkpoint — the pattern a
- * real JSONL/sqlite persistence plugin would follow.
- */
-export function apply(ctx: Context) {
-  const buffers = new Map<Session, SessionEvent[]>()
-  const path = (session: Session) => join(import.meta.dirname, '..', `${session.id}.jsonl`)
-
-  ctx.on('session/event', (session, event) => {
-    let buffer = buffers.get(session)
-    if (!buffer) buffers.set(session, buffer = [])
-    buffer.push(event)
-  })
-
-  const flush = async (session: Session) => {
-    const buffer = buffers.get(session)
-    if (!buffer?.length) return
-    const lines = buffer.splice(0).map(event => JSON.stringify(event) + '\n').join('')
-    await appendFile(path(session), lines)
-  }
-
-  ctx.on('session/flush', flush)
-  ctx.effect(() => () => {
-    // drain remaining buffers on dispose
-    for (const session of buffers.keys()) void flush(session)
-  }, 'session-jsonl')
-}

examples/echo-agent/README.md

@@ -7,7 +7,7 @@ Runnable demo: stdin chat with a scripted mock model and an echo tool.
 - A complete Cordis app loaded from `cordis.yml` — the standard "stack of plugins" pattern
 - `mock-llm.ts` — a mock `LlmAdapter` that streams scripted responses and calls the `echo` tool when the user types "echo <something>"
 - `echo-tool.ts` — a tool registered via `ctx.tools.register()` that echoes text back uppercased
-- `session-jsonl.ts` — a minimal persistence plugin: write-behind buffering of `session/event` notifications, drained to a JSONL file at `session/flush`
+- `@deepseek-ai/dsh-session-persistence-jsonl` — the durable JSONL persistence backend (loaded from `cordis.yml`, `root: ./.sessions`): append-only event log per session with crash-safe atomic writes, replacing the old write-only example plugin
 - `stdio-chat.ts` — a minimal UI plugin: reads stdin lines and `send`/`steer`s the agent, renders stream deltas, tool calls, and tool results
 
 ## Plugin files
@@ -16,10 +16,11 @@ Runnable demo: stdin chat with a scripted mock model and an echo tool.
 |---|---|---|
 | `mock-llm.ts` | `LlmAdapter` registration | `ctx.llm.registerAdapter(['mock-echo'], …)`, streaming chunks with proper `block-start`/`block-end` protocol |
 | `echo-tool.ts` | Tool registration | `ctx.tools.register(defineTool(…))` with typed `execute` args, tool execution returning `ContentBlock[]` |
-| `session-jsonl.ts` | Persistence | `session/event` listener + `session/flush` drain, fiber-dispose cleanup |
 | `stdio-chat.ts` | UI | `agent/stream-chunk`, `session/event` (tool/*), stdin→send/steer |
 | `start.ts` | Bootstrap | `Context` + `Loader` + `plugin-include` wired to `cordis.yml` |
 
+Persistence is the shared `@deepseek-ai/dsh-session-persistence-jsonl` plugin (not a per-example file).
+
 ## Run
 
 ```sh
@@ -30,4 +31,4 @@ node --expose-internals --import tsx examples/echo-agent/start.ts
 
 Type a message and press Enter. "echo <text>" triggers a tool call round-trip (the mock model requests the `echo` tool, which echoes the text uppercased, and the next model step acknowledges it).
 
-The session is persisted to `<session-id>.jsonl` in the `examples/echo-agent/` directory. Clean up with: `rm -f examples/echo-agent/*.jsonl`
+The session is persisted under `.sessions/` relative to the directory you launch the demo from. `pnpm run demo:echo` runs from the repo root, so the logs land in `<repo-root>/.sessions/` (a session with no cwd goes in the `_no-cwd/` bucket, one `.jsonl` log per session). Clean up with: `rm -rf .sessions`

examples/echo-agent/cordis.yml

@@ -46,8 +46,10 @@
 - id: echo-tool
   name: './src/echo-tool.ts'
 
-- id: session-jsonl
+- id: session-persistence
-  name: './src/session-jsonl.ts'
+  name: '@deepseek-ai/dsh-session-persistence-jsonl'
+  config:
+    root: './.sessions'
 
 - id: stdio-chat
   name: './src/stdio-chat.ts'

examples/echo-agent/src/session-jsonl.ts

@@ -1,36 +0,0 @@
-import { appendFile } from 'node:fs/promises'
-import { join } from 'node:path'
-import type { Context } from 'cordis'
-import type { Session, SessionEvent } from '@deepseek-ai/dsh-session'
-
-export const name = 'session-jsonl'
-export const inject = ['sessions']
-
-/**
- * Minimal persistence plugin: buffers session events (write-behind) and
- * drains to a JSONL file at every `session/flush` checkpoint — the pattern a
- * real JSONL/sqlite persistence plugin would follow.
- */
-export function apply(ctx: Context) {
-  const buffers = new Map<Session, SessionEvent[]>()
-  const path = (session: Session) => join(import.meta.dirname, '..', `${session.id}.jsonl`)
-
-  ctx.on('session/event', (session, event) => {
-    let buffer = buffers.get(session)
-    if (!buffer) buffers.set(session, buffer = [])
-    buffer.push(event)
-  })
-
-  const flush = async (session: Session) => {
-    const buffer = buffers.get(session)
-    if (!buffer?.length) return
-    const lines = buffer.splice(0).map(event => JSON.stringify(event) + '\n').join('')
-    await appendFile(path(session), lines)
-  }
-
-  ctx.on('session/flush', flush)
-  ctx.effect(() => () => {
-    // drain remaining buffers on dispose
-    for (const session of buffers.keys()) void flush(session)
-  }, 'session-jsonl')
-}

packages/agent-loop/README.md

@@ -8,7 +8,7 @@ This is the only package in the harness that contains concrete loop logic. Every
 
 ### Public API
 
-- `ctx.agentLoop.create(id: string, options?: AgentOptions): LoopAgent` Create an agent, start its loop, and register it in `ctx.agents`. Disposed with the calling fiber.
+- `ctx.agentLoop.create(id: string, options?: AgentOptions): LoopAgent` — create an agent on a fresh per-run session id `${id}-session-<uuid>`, start its loop, and register it in `ctx.agents`. The per-run uuid avoids colliding with the on-disk log a prior run materialized once a durable persistence backend is loaded; each run is a new session (a deliberate demo simplification — a real resume-or-create policy is a TODO). Disposed with the calling fiber.
 
 ### Injected services
 

packages/agent-loop/package.json

@@ -35,6 +35,7 @@
     "@deepseek-ai/dsh-invariants": "workspace:^",
     "@deepseek-ai/dsh-llm": "workspace:^",
     "@deepseek-ai/dsh-session": "workspace:^",
+    "@deepseek-ai/dsh-session-persistence-jsonl": "workspace:^",
     "@deepseek-ai/dsh-system-prompt": "workspace:^",
     "@deepseek-ai/dsh-tools": "workspace:^",
     "cordis": "^4.0.0-rc.6"

packages/agent-loop/src/index.ts

@@ -8,6 +8,7 @@
  */
 
 import { Context, Service } from 'cordis'
+import { randomUUID } from 'node:crypto'
 import z from 'schemastery'
 import { AgentId } from '@deepseek-ai/dsh-agent'
 import type { AgentOptions } from '@deepseek-ai/dsh-agent'
@@ -62,12 +63,24 @@ export class AgentLoop extends Service {
    * Create an agent, start its loop, and register it. Returns the agent.
    * Disposed with the calling fiber.
    *
+   * The session id is per-run (`${id}-session-<uuid>`, no fixed name): once a
+   * durable persistence backend is loaded, a fixed `${id}-session` collides on
+   * the second run — the backend refuses to re-create an id whose log already
+   * exists on disk (the SessionId is the identity). A fresh id means each run
+   * is a new session.
+   *
+   * TODO(demo): each run starting a brand-new session is fine for demos but is
+   * NOT real conversation continuity. A production config-driven agent needs a
+   * deliberate resume-or-create policy (resume the prior session if one exists,
+   * else start fresh) or an explicit caller-chosen session id — revisit when the
+   * UI/ACP path owns session selection.
+   *
    * TODO(sub-agents): spawn/fork land here — accept a parent agent reference;
    * fork seeds the new Session with the parent's event log, spawn starts
    * fresh; the child is returned as a regular Agent handle.
    */
   create(id: string, options: AgentOptions = {}): LoopAgent {
-    const session = this.ctx.sessions.create(`${id}-session`)
+    const session = this.ctx.sessions.create(`${id}-session-${randomUUID()}`, { meta: {} })
     const agent = new LoopAgent(this.ctx, AgentId(id), options, session)
     // Generator effect: stop and unregister are independent disposables
     // (LIFO), so a throwing stop() cannot leak the registry entry.

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

@@ -0,0 +1,65 @@
+import { afterEach, describe, expect, it } from 'vitest'
+import { Context } from 'cordis'
+import { mkdtemp, rm } from 'node:fs/promises'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import LlmService from '@deepseek-ai/dsh-llm'
+import SessionStore from '@deepseek-ai/dsh-session'
+import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
+import ToolRegistry from '@deepseek-ai/dsh-tools'
+import AgentRegistry from '@deepseek-ai/dsh-agent'
+import SessionPersistenceJsonl from '@deepseek-ai/dsh-session-persistence-jsonl'
+import AgentLoop, { LoopAgent } from '@deepseek-ai/dsh-agent-loop'
+import { MockAdapter, textResponse } from './mock-adapter.ts'
+
+const dirs: string[] = []
+afterEach(async () => { for (const d of dirs.splice(0)) await rm(d, { recursive: true, force: true }) })
+
+function waitForIdle(ctx: Context, agent: LoopAgent): Promise<void> {
+  return new Promise((resolve) => {
+    const dispose = ctx.on('agent/status', (subject, status) => {
+      if (subject === agent && status === 'idle') { dispose(); resolve() }
+    })
+  })
+}
+
+describe('config-driven session id', () => {
+  it('config-driven create uses a fresh ${id}-session-<uuid> per run (restart-safe)', async () => {
+    const root = await mkdtemp(join(tmpdir(), 'dsh-cfg-session-'))
+    dirs.push(root)
+    const idPattern = /^cfg-session-[0-9a-f-]{36}$/
+    // Run 1: a config agent persists a turn under a generated session id.
+    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: [{ id: 'cfg', model: 'mock', systemPrompt: '' }] })
+    await ctx1.plugin(SessionPersistenceJsonl, { root })
+    ctx1.llm.registerAdapter(['mock'], new MockAdapter([textResponse('cfg')]))
+    const a1 = ctx1.agents.get('cfg') as LoopAgent
+    expect(a1.session.id).toMatch(idPattern)
+    a1.send([{ type: 'text', text: 'q' }], { source: { kind: 'user' } })
+    await waitForIdle(ctx1, a1)
+    await ctx1.fiber.dispose()
+
+    // Run 2 over the SAME root: a fresh id means no on-disk collision (a fixed
+    // ${id}-session would crash here with "already has a persisted log").
+    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: 'cfg', model: 'mock', systemPrompt: '' }] })
+    await ctx2.plugin(SessionPersistenceJsonl, { root })
+    ctx2.llm.registerAdapter(['mock'], new MockAdapter([textResponse('cfg2')]))
+    const a2 = ctx2.agents.get('cfg') as LoopAgent
+    expect(a2.session.id).toMatch(idPattern)
+    expect(a2.session.id).not.toBe(a1.session.id)
+    a2.send([{ type: 'text', text: 'q2' }], { source: { kind: 'user' } })
+    await waitForIdle(ctx2, a2)
+    await ctx2.fiber.dispose()
+  })
+})

packages/session-persistence-jsonl/README.md

@@ -0,0 +1,33 @@
+# @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 plus a small atomic `.summary.json` sidecar for mutable metadata.
+
+## 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>.summary.json    # mutable SessionSummary (atomic temp-write + rename)
+```
+
+- The first `.jsonl` line is the immutable `SessionHeader` tagged `{ type: 'session', version, id, cwd?, createdAt, parentSession? }`; 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).
+
+## 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). |
+
+## Durability and crash semantics
+
+- **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`.
+- **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: an error `tool/result` for every `tool-call` the crash left unanswered (the loop logs the assistant message before running the tools, so a mid-tool crash leaves dangling calls — and `deriveMessages()` would replay an assistant tool-call with no result, which providers reject), then 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.
+
+## Write path
+
+The plugin generalizes the example `session-jsonl.ts`: it subscribes to `session/created` (capture the header; persist a fork's seed once), `session/event` (snapshot each event when buffering — the live `session.events` object is mutable), and `session/flush`/dispose (drain the write-behind buffer through `append`). A per-session write cursor means a resumed session never re-appends already-stored events. Existing live sessions are seeded on plugin apply (HMR does not replay `session/created`). All backend operations for one session are serialized, and disposal awaits quiescence (every init + final drain) before returning, so no write lands after teardown.

packages/session-persistence-jsonl/package.json

@@ -0,0 +1,35 @@
+{
+  "name": "@deepseek-ai/dsh-session-persistence-jsonl",
+  "description": "JSONL durable session persistence backend for the DeepSeek Harness",
+  "version": "0.0.1",
+  "private": true,
+  "type": "module",
+  "main": "lib/index.js",
+  "types": "lib/index.d.ts",
+  "exports": {
+    ".": {
+      "types": "./lib/index.d.ts",
+      "default": "./lib/index.js"
+    },
+    "./src/*": "./src/*",
+    "./package.json": "./package.json"
+  },
+  "files": [
+    "lib",
+    "src"
+  ],
+  "license": "BSD-3-Clause",
+  "peerDependencies": {
+    "@deepseek-ai/dsh-session": "^0.0.1",
+    "@deepseek-ai/dsh-session-persistence": "^0.0.1",
+    "cordis": "^4.0.0-rc.6"
+  },
+  "dependencies": {
+    "schemastery": "^3.18.0"
+  },
+  "devDependencies": {
+    "@deepseek-ai/dsh-session": "workspace:^",
+    "@deepseek-ai/dsh-session-persistence": "workspace:^",
+    "cordis": "^4.0.0-rc.6"
+  }
+}

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

@@ -0,0 +1,264 @@
+/**
+ * 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)
+}
+
+/**
+ * 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): 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')
+  // 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` — 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 }
+  }
+
+  // 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) {
+      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) {
+      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
+    }
+    preserved.push(p.event)
+  }
+
+  // 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). */
+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)
+}

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

@@ -0,0 +1,879 @@
+/**
+ * JSONL durable session-persistence backend (`@deepseek-ai/dsh-session-persistence-jsonl`).
+ *
+ * Two concerns in one plugin:
+ *
+ * 1. **The backend** — a concrete {@link SessionPersistence}: one append-only
+ *    `.jsonl` event log per session (a header line then one `SessionEvent` per
+ *    line, verbatim including `assistant/chunk` so `seq` stays contiguous) plus
+ *    a small atomic `.summary.json` sidecar for the mutable `SessionSummary`.
+ *    Lazy materialization (no file until the first `append`), atomic first
+ *    write, and truncation-repair of a never-committed crash tail on the first
+ *    `append` after a `load`.
+ *
+ * 2. **The write path** — the `session/event` → buffer → `session/flush` drain
+ *    that generalizes the example `session-jsonl.ts`: snapshot each event when
+ *    it is buffered (the live `session.events` object is mutable), persist
+ *    forks once on `session/created`, maintain a per-session write cursor so a
+ *    resumed session never re-appends already-stored events, and seed existing
+ *    live sessions on plugin apply (HMR does not replay `session/created`).
+ *
+ * @module @deepseek-ai/dsh-session-persistence-jsonl
+ */
+
+import { Context } from 'cordis'
+import z from 'schemastery'
+import { open, mkdir, readFile, readdir, rename, link, rm, truncate } from 'node:fs/promises'
+import { resolve } from 'node:path'
+import { randomBytes } from 'node:crypto'
+import { SessionPersistence } from '@deepseek-ai/dsh-session-persistence'
+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,
+} from './format.ts'
+
+export interface Config {
+  /**
+   * Root directory for all session files. Required (no default): a default of
+   * `process.cwd()` would scatter session files as the process's cwd changes
+   * (bash calls, subprocesses). Sessions group under per-cwd subdirectories.
+   */
+  root: string
+}
+
+/** Per-session write state held by the backend's in-memory bookkeeping. */
+interface SessionState {
+  meta: SessionMeta
+  /** The next seq the backend expects to append (the stored log length). */
+  cursor: number
+  /** Whether the `.jsonl` file has been physically materialized. */
+  materialized: boolean
+  /**
+   * The live Session this state was bound to via `onCreated`, if any. Used to
+   * detect a DIFFERENT live session reusing a tracked id (a collision): state
+   * created through the public `create()`/`load()` API has no owner, but state
+   * bound to a live session lets `onCreated` reject a second, unrelated session
+   * object on the same id instead of silently no-opping (which would leave the
+   * new session's events to be dropped against the old cursor).
+   */
+  owner?: Session
+}
+
+/**
+ * Whether a live session's `seed` reproduces a persisted `prefix` exactly — the
+ * prefix is no longer than the seed, and each prefix event DEEP-equals the seed
+ * event at the same index. Used to tell a session legitimately continuing a
+ * persisted log (HMR re-seeing its own session, or a resume) from a different
+ * session that merely reuses the id: the latter would have its already-counted
+ * seq 0..prefix-1 events filtered out on flush and its conversation silently
+ * grafted onto the old log.
+ *
+ * The comparison is a full structural equality (via canonical JSON) of each
+ * event INCLUDING its `data` payload, not just `seq`/`type`/`time` — a session
+ * built from loaded events but with mutated message/tool payloads (same seq/
+ * type/time) must NOT be accepted, or the live history and durable log diverge.
+ * Both sides are JSON-serializable by contract (Session.append enforces it), so
+ * JSON.stringify is a sound canonical form here.
+ */
+function seedCoversPrefix(seed: readonly SessionEvent[], prefix: readonly SessionEvent[]): boolean {
+  return prefix.length <= seed.length
+    && prefix.every((e, i) => {
+      const s = seed[i]
+      return s !== undefined && JSON.stringify(s) === JSON.stringify(e)
+    })
+}
+
+/**
+ * Reject non-JSON-serializable `event.data`, naming the offending type. Used on
+ * the backend's `append(events)` entry point (replay/fork paths that bypass a
+ * live `Session`); events that flow through `Session.append` are already
+ * validated at the source, so the live write path never needs this.
+ */
+function assertSerializable(events: readonly SessionEvent[]): void {
+  for (const event of events) {
+    if (!isJsonValue(event.data)) {
+      throw new Error(`event "${event.type}" carries non-JSON-serializable data (seq ${event.seq})`)
+    }
+  }
+}
+
+/**
+ * Whether `error` is a "no such file/directory" (`ENOENT`) failure — the ONLY
+ * filesystem error that legitimately means "this session/root is absent" for a
+ * durable backend. Any OTHER error (`EACCES`, `ENOTDIR`, transient I/O) must
+ * surface rather than be silently reported as absence: masking it would let
+ * `list()` report no sessions, `load()` report "not found", and collision
+ * checks proceed under a false absence assumption — all unsafe for durable
+ * persistence. (A NodeJS filesystem rejection carries a string `code`.)
+ */
+function isENOENT(error: unknown): boolean {
+  return (error as NodeJS.ErrnoException | null)?.code === 'ENOENT'
+}
+
+/**
+ * The JSONL persistence backend. Load as a plugin; it registers as
+ * `ctx.sessionPersistence` and installs the write-path listeners.
+ */
+export class SessionPersistenceJsonl extends SessionPersistence {
+  static inject = ['sessions']
+
+  static Config: z<Config> = z.object({
+    root: z.string().required(),
+  })
+
+  private root: string
+  /** Backend bookkeeping keyed by session id (NOT the live Session object). */
+  private states = new Map<string, SessionState>()
+  /** Write-behind buffers keyed by the live Session (write path). */
+  private buffers = new Map<Session, SessionEvent[]>()
+  /**
+   * Per-session serialization: every backend operation chains onto the prior
+   * one for the same id, so concurrent flushes / a flush racing onCreated never
+   * interleave file writes or read a half-built state. Keyed by session id.
+   */
+  private chains = new Map<string, Promise<unknown>>()
+  /**
+   * Per-session init promise (onCreated). Keyed by the LIVE Session OBJECT, not
+   * its id: a disposed fiber's session can be replaced by a different live
+   * Session reusing the same id (HMR, an ACP reconnect), and an id-keyed cache
+   * would hand the new object the old object's init promise — skipping
+   * onCreated for the new session, so its events start at seq 0 while flush
+   * filters against the stale cursor and silently drops them. Keying by object
+   * gives each live Session its own init. flush awaits it before appending.
+   */
+  private inits = new Map<Session, Promise<void>>()
+
+  constructor(ctx: Context, public config: Config) {
+    super(ctx)
+    // Resolve the configured root to an ABSOLUTE path ONCE, here. A relative
+    // root (the examples use `./.sessions`) would otherwise re-resolve against
+    // `process.cwd()` at every later readdir/open — so if any plugin or test
+    // changed cwd between create, append, and load, one session's files could
+    // split across directories. Pinning it at construction makes all paths
+    // stable regardless of later cwd changes.
+    this.root = resolve(config.root)
+    this.installWritePath()
+  }
+
+  // --- SessionPersistence backend surface (all serialized per session id) ---
+
+  create(meta: SessionMeta): Promise<void> {
+    // Snapshot the metadata at call time: the op runs later (behind the
+    // per-session chain) and the snapshot is also stored as the lazy state, so
+    // keeping the caller's object by reference would let a later mutation of
+    // `id`/`cwd` register under one key but materialize under a different
+    // path/header. A shallow copy is enough — SessionMeta is a flat record.
+    const snapshot: SessionMeta = { ...meta }
+    return this.serialize(snapshot.id, () => this.createCore(snapshot))
+  }
+
+  private async createCore(meta: SessionMeta): Promise<void> {
+    // Do NOT clobber an existing session. If we already track it, or a log
+    // exists on disk under this id, refuse — the SessionId IS the identity, and
+    // silently resetting state (cursor 0, materialized false) over committed
+    // data would let the next append rename over the existing log.
+    if (this.states.has(meta.id)) {
+      throw new Error(`session "${meta.id}" already exists in this backend`)
+    }
+    // Scan ALL cwd buckets (pass undefined), not just meta.cwd's: load/has/adopt
+    // identify a session by id alone and search every bucket, so an id already
+    // persisted under a DIFFERENT cwd must still block creation here. Probing
+    // only meta.cwd's bucket would let two logs share one id and make resume
+    // (which picks the first matching bucket) nondeterministic.
+    if (await this.findLog(meta.id, undefined) !== undefined) {
+      throw new Error(`session "${meta.id}" already has a persisted log on disk; load/resume it instead of creating`)
+    }
+    // Pure lazy: record intent only. No file until the first append, so an
+    // abandoned (never-appended) session leaves nothing on disk and stays
+    // absent from has()/list().
+    this.states.set(meta.id, { meta, cursor: 0, materialized: false })
+  }
+
+  /**
+   * Run `op` after any in-flight operation for the same session id, so writes
+   * for one session never interleave (two flushes, a flush racing a load, an
+   * update racing an append). Errors do not poison the chain — the next op
+   * still runs. NOTE: serialized public methods must NOT call each other (that
+   * would deadlock on the same chain); they call the unserialized `*Core`
+   * helpers instead.
+   */
+  private serialize<T>(id: SessionId, op: () => Promise<T>): Promise<T> {
+    const prior = this.chains.get(id) ?? Promise.resolve()
+    const next = prior.then(op, op)
+    // Keep the chain alive but swallow this op's rejection for the NEXT waiter
+    // (the caller still sees the real rejection via `next`).
+    this.chains.set(id, next.then(() => undefined, () => undefined))
+    return next
+  }
+
+  // `async` so the synchronous validate/clone below reject (not throw) per the
+  // Promise<void> contract — callers use `await expect(...).rejects`.
+  async append(id: SessionId, events: readonly SessionEvent[]): Promise<void> {
+    // Validate serializability BEFORE cloning, so a bad event surfaces the typed
+    // "non-JSON-serializable" error rather than an opaque DataCloneError from
+    // structuredClone below. (In an async method this throw becomes a rejection,
+    // honoring the Promise<void> contract rather than throwing synchronously.)
+    assertSerializable(events)
+    // Deep-snapshot the batch here, BEFORE the op waits behind the per-session
+    // chain: the op may await before serializing, so a caller that passes a live
+    // array (e.g. session.events) and mutates it — OR mutates an event object
+    // inside it — before the op runs would otherwise have those changes
+    // persisted, or advance the cursor past what was actually written.
+    // structuredClone covers both the array and the event objects (safe now that
+    // serializability is checked above). The clone happens synchronously (before
+    // the first await), so it is taken at call time.
+    const batch = events.map(e => structuredClone(e))
+    return this.serialize(id, () => this.appendCore(id, batch))
+  }
+
+  private async appendCore(id: SessionId, events: readonly SessionEvent[]): Promise<void> {
+    if (events.length === 0) return
+    assertSerializable(events)
+    let state = this.states.get(id)
+    if (state === undefined) state = await this.adopt(id) // calls loadCore, not load
+
+    // Contiguity contract: each event's seq must continue the stored log.
+    for (const [i, event] of events.entries()) {
+      if (event.seq !== state.cursor + i) {
+        throw new Error(`append seq mismatch for "${id}": expected ${state.cursor + i} at index ${i}, got ${event.seq}`)
+      }
+    }
+
+    if (!state.materialized) {
+      await this.materialize(state, events)
+    } else {
+      await this.appendLines(state, events)
+    }
+    // The durable event log is the transaction: advance the cursor as soon as
+    // the log write commits. The sidecar (mutable summary) is best-effort here
+    // — a failed sidecar write must NOT reject an append whose log already
+    // landed (that would desync the cursor and let a retry duplicate seqs).
+    state.cursor += events.length
+    await this.touchSummary(state).catch(() => { /* sidecar is recoverable metadata; log is durable */ })
+  }
+
+  load(id: SessionId): Promise<{ meta: SessionMeta; events: SessionEvent[] }> {
+    return this.serialize(id, () => this.loadCore(id))
+  }
+
+  private async loadCore(id: SessionId): Promise<{ meta: SessionMeta; events: SessionEvent[] }> {
+    const cwd = this.states.get(id)?.meta.cwd
+    const file = await this.findLog(id, cwd)
+    if (file === undefined) throw new Error(`session "${id}" not found`)
+    const buffer = await readFile(file.path)
+    const { meta, events, committedBytes } = scanLog(buffer)
+    this.assertVersion(meta)
+
+    const summary = await this.readSidecar(id, meta.cwd)
+    const fullMeta: SessionMeta = { ...meta, ...summary }
+
+    // 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,
+    }
+    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[]> {
+    const metas: SessionMeta[] = []
+    for (const dir of await this.listCwdDirs()) {
+      for (const name of await this.listJsonl(dir)) {
+        // Read ONLY the header line, not the whole log: a session picker must
+        // scale with the number of sessions, not the total size of every
+        // conversation (the log persists every assistant/chunk verbatim, so a
+        // full scanLog here would be O(total history)).
+        const first = 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
+        const summary = await this.readSidecar(meta.id, meta.cwd)
+        metas.push({ ...meta, ...summary })
+      }
+    }
+    return metas
+  }
+
+  /**
+   * Read the first newline-terminated line of a file without loading the whole
+   * file. Returns undefined if the file is empty or has no complete first line
+   * (a half-written log). Reads in bounded chunks so a huge log costs only the
+   * header read.
+   */
+  private async readFirstLine(path: string): Promise<string | undefined> {
+    const handle = await open(path, 'r')
+    try {
+      const chunks: Buffer[] = []
+      const buf = Buffer.alloc(8192)
+      for (;;) {
+        const { bytesRead } = await handle.read(buf, 0, buf.length, null)
+        if (bytesRead === 0) return undefined // EOF with no newline → no complete line
+        const slice = buf.subarray(0, bytesRead)
+        const nl = slice.indexOf(0x0a)
+        if (nl !== -1) {
+          chunks.push(slice.subarray(0, nl))
+          return Buffer.concat(chunks).toString('utf8')
+        }
+        chunks.push(Buffer.from(slice))
+      }
+    } finally {
+      await handle.close()
+    }
+  }
+
+  async has(id: SessionId): Promise<boolean> {
+    const state = this.states.get(id)
+    if (state?.materialized) return true
+    const cwd = state?.meta.cwd
+    return (await this.findLog(id, cwd)) !== undefined
+  }
+
+  delete(id: SessionId): Promise<void> {
+    return this.serialize(id, () => this.deleteCore(id))
+  }
+
+  private async deleteCore(id: SessionId): Promise<void> {
+    const cwd = this.states.get(id)?.meta.cwd
+    const file = await this.findLog(id, cwd)
+    if (file) await rm(file.path, { force: true })
+    // Remove the sidecar too. A lazy session (update() before the first
+    // append()) has a `.summary.json` sidecar but NO `.jsonl` log, and after a
+    // restart the in-memory cwd is gone — so keying sidecar removal off the log
+    // or the in-memory cwd would leak its possibly-sensitive title/firstPrompt.
+    // Scan every cwd bucket for the sidecar by its (sanitized) filename.
+    await this.removeSidecars(id)
+    this.states.delete(id)
+  }
+
+  /** Remove a session's summary sidecar from EVERY cwd bucket (id is unique). */
+  private async removeSidecars(id: SessionId): Promise<void> {
+    const target = `${encodeSegment(id)}.summary.json`
+    for (const dir of await this.listCwdDirs()) {
+      await rm(`${dir}/${target}`, { force: true })
+    }
+  }
+
+  update(id: SessionId, summary: Partial<SessionSummary>): Promise<void> {
+    return this.serialize(id, () => this.updateCore(id, summary))
+  }
+
+  private async updateCore(id: SessionId, summary: Partial<SessionSummary>): Promise<void> {
+    let state = this.states.get(id)
+    if (state === undefined) state = await this.adopt(id)
+    // Build the NEXT meta separately and commit it to in-memory state only AFTER
+    // the sidecar write succeeds. update's only durable effect is the sidecar,
+    // so a failure DOES reject (unlike append, whose log is the transaction and
+    // sidecar is best-effort) — but if we mutated state.meta first, a later
+    // touchSummary() on a successful append would persist the rejected
+    // title/firstPrompt, making a failed update durable after the fact.
+    const nextMeta: SessionMeta = { ...state.meta, ...summary }
+    await this.writeSidecar(nextMeta)
+    state.meta = nextMeta
+  }
+
+  // --- materialization / append / repair ---
+
+  /** Atomically write the header line + first batch (temp-write, fsync, rename). */
+  private async materialize(state: SessionState, events: readonly SessionEvent[]): Promise<void> {
+    const dir = sessionDir(this.root, state.meta.cwd)
+    await mkdir(dir, { recursive: true, mode: 0o700 })
+    const finalPath = logPath(this.root, state.meta.cwd, state.meta.id)
+    // Never rename over an existing committed log: materialize is the FIRST
+    // write of a session the backend believes is new. A file here means a
+    // different session shares this id on disk — reject loudly rather than
+    // clobber committed data. (createCore already guards the create path before
+    // this point, so this is unreachable-in-practice defense-in-depth against a
+    // TOCTOU/fork race; ignored for coverage.)
+    /* v8 ignore next 3 -- createCore guards collisions before materialize; this is a TOCTOU backstop */
+    if (await this.exists(finalPath)) {
+      throw new Error(`refusing to materialize "${state.meta.id}": a log already exists on disk (load/resume it instead)`)
+    }
+    const header = JSON.stringify(toHeaderLine(state.meta))
+    const body = events.map(eventLine).join('\n')
+    const content = header + '\n' + body + '\n'
+
+    const tmp = `${finalPath}.${randomBytes(6).toString('hex')}.tmp`
+    const handle = await open(tmp, 'wx', 0o600)
+    try {
+      await handle.writeFile(content)
+      await handle.sync()
+    } finally {
+      await handle.close()
+    }
+    // Publish via link()+unlink(), NOT rename(): link fails with EEXIST if the
+    // final path already exists, so two processes materializing the same id
+    // concurrently cannot clobber each other (both could pass the exists() check
+    // above, but only one link() wins). rename() would silently overwrite the
+    // log the other process just committed.
+    let linked = false
+    try {
+      await link(tmp, finalPath)
+      linked = true
+    } finally {
+      // If link FAILED (EEXIST on a race, or any I/O error), the temp is the
+      // only reference and must be removed before the original error propagates.
+      // If link SUCCEEDED, the temp cleanup is deferred to AFTER the publish is
+      // durable (below) so a temp-rm failure can never reject a session whose
+      // log already published — that would leave state.materialized false and
+      // wedge every retry on the exists() backstop above.
+      /* v8 ignore next -- link failure is the TOCTOU/IO race guarded above; not reachable in test */
+      if (!linked) await rm(tmp, { force: true })
+    }
+    // link() succeeded — the log is published. fsync the directory so the new
+    // entry survives a power loss: on POSIX filesystems the new link is not
+    // crash-durable until the parent directory's metadata is synced. The seam
+    // contract is "append returns once durable", and materialize is the first
+    // append's write — so the directory entry must be durable before we return.
+    await this.syncDir(dir)
+    state.materialized = true
+    // Best-effort temp cleanup: the log is already published and durable, so a
+    // failure to remove the (now-redundant) temp hard link must NOT reject the
+    // append. A leftover `*.tmp` is harmless — it is never read, and the next
+    // materialize of this id is guarded by exists()/link(). Swallow only the
+    // rm failure; nothing else of consequence runs in the try.
+    try {
+      await rm(tmp, { force: true })
+    } catch {
+      /* v8 ignore next -- redundant temp link; publish already durable, rm failure is an unreachable IO edge */
+    }
+  }
+
+  /** fsync a directory so a just-created/renamed entry inside it is crash-durable. */
+  private async syncDir(dir: string): Promise<void> {
+    const handle = await open(dir, 'r')
+    try {
+      await handle.sync()
+    } finally {
+      await handle.close()
+    }
+  }
+
+  /**
+   * Append event lines at EOF and fsync. On a write/sync failure AFTER the
+   * kernel accepted some bytes (ENOSPC, an fsync error), truncate the file back
+   * to its pre-append size before rethrowing: `cursor` is unchanged, so the
+   * batch will be retried, and without this rollback the retry would append
+   * AFTER the partial bytes — producing duplicate seqs that make `scanLog` see a
+   * gap and render the session unloadable.
+   */
+  private async appendLines(state: SessionState, events: readonly SessionEvent[]): Promise<void> {
+    const path = logPath(this.root, state.meta.cwd, state.meta.id)
+    const handle = await open(path, 'a')
+    try {
+      const { size: before } = await handle.stat()
+      try {
+        await handle.writeFile(events.map(eventLine).join('\n') + '\n')
+        await handle.sync()
+      } catch (error) {
+        // Roll back whatever bytes landed so a retry starts from a clean EOF.
+        await handle.truncate(before)
+        await handle.sync()
+        throw error
+      }
+    } finally {
+      await handle.close()
+    }
+  }
+
+  /** Truncate the log file to `offset` bytes and fsync (discard the crash tail). */
+  private async repair(state: SessionState, offset: number): Promise<void> {
+    const path = logPath(this.root, state.meta.cwd, state.meta.id)
+    await truncate(path, offset)
+    const handle = await open(path, 'r+')
+    try {
+      await handle.sync()
+    } finally {
+      await handle.close()
+    }
+  }
+
+  // --- sidecar (mutable summary) ---
+
+  private async touchSummary(state: SessionState): Promise<void> {
+    state.meta = { ...state.meta, updatedAt: Date.now() }
+    await this.writeSidecar(state.meta)
+  }
+
+  /**
+   * Atomic sidecar write (temp-write + rename), summary fields only.
+   *
+   * Deliberately NOT directory-fsynced (unlike {@link materialize}): the
+   * sidecar holds mutable, recoverable summary metadata (updatedAt, title,
+   * firstPrompt), not source-of-truth log data. The rename is atomic so a
+   * reader never sees a torn file, but a power loss may lose the most recent
+   * summary — acceptable because it is re-derivable and the durable log (the
+   * transaction) is independently synced. Strict crash-durability is reserved
+   * for the event log.
+   */
+  private async writeSidecar(meta: SessionMeta): Promise<void> {
+    const dir = sessionDir(this.root, meta.cwd)
+    await mkdir(dir, { recursive: true, mode: 0o700 })
+    const path = sidecarPath(this.root, meta.cwd, meta.id)
+    const summary: SessionSummary = {
+      updatedAt: meta.updatedAt,
+      ...meta.title !== undefined ? { title: meta.title } : {},
+      ...meta.firstPrompt !== undefined ? { firstPrompt: meta.firstPrompt } : {},
+    }
+    const tmp = `${path}.${randomBytes(6).toString('hex')}.tmp`
+    // Exclusive owner-only create ('wx', 0o600), matching the log-materialization
+    // temp write: the sidecar can carry user data (title/firstPrompt), so a
+    // predictable/pre-existing temp path must never be silently truncated and
+    // followed (symlink race / disclosure). The random suffix already makes a
+    // collision unlikely; 'wx' makes reuse an error rather than a clobber.
+    const handle = await open(tmp, 'wx', 0o600)
+    try {
+      await handle.writeFile(JSON.stringify(summary))
+    } finally {
+      await handle.close()
+    }
+    await rename(tmp, path)
+  }
+
+  /**
+   * Read the mutable-summary sidecar, or `undefined` if it is absent/unreadable
+   * (a session that has never been `update()`d, or a failed sidecar write). The
+   * caller keeps the header-derived `updatedAt` (the session's createdAt) in
+   * that case rather than overlaying `0` — reporting an active session as
+   * updated at the Unix epoch would be wrong.
+   */
+  private async readSidecar(id: SessionId, cwd: string | undefined): Promise<SessionSummary | undefined> {
+    try {
+      const raw = await readFile(sidecarPath(this.root, cwd, id), 'utf8')
+      return JSON.parse(raw) as SessionSummary
+    } catch {
+      return undefined
+    }
+  }
+
+  // --- discovery helpers ---
+
+  /** Find a session's log file across cwd buckets (when cwd is unknown). */
+  private async findLog(id: SessionId, cwd: string | undefined): Promise<{ path: string; cwd: string | undefined } | undefined> {
+    if (cwd !== undefined) {
+      const path = logPath(this.root, cwd, id)
+      return (await this.exists(path)) ? { path, cwd } : undefined
+    }
+    // Unknown cwd: scan buckets for a matching file name.
+    const target = encodeSegment(id) + '.jsonl'
+    for (const dir of await this.listCwdDirs()) {
+      const path = `${dir}/${target}`
+      if (await this.exists(path)) {
+        // Recover cwd from the header for accurate sidecar pathing.
+        const { meta } = scanLog(await readFile(path))
+        return { path, cwd: meta.cwd }
+      }
+    }
+    return undefined
+  }
+
+  /** The cwd-bucket directories under the root (absolute paths). */
+  private async listCwdDirs(): Promise<string[]> {
+    try {
+      const entries = await readdir(this.root, { withFileTypes: true })
+      return entries.filter(e => e.isDirectory()).map(e => `${this.root}/${e.name}`)
+    } catch (error) {
+      // ENOENT = the root has not been created yet → genuinely no sessions.
+      // Any other error (EACCES, ENOTDIR, transient I/O) must NOT be reported
+      // as "no sessions" — a durable backend cannot silently pretend persisted
+      // state is absent on a storage fault.
+      if (isENOENT(error)) return []
+      throw error
+    }
+  }
+
+  private async listJsonl(dir: string): Promise<string[]> {
+    const entries = await readdir(dir)
+    return entries.filter(n => n.endsWith('.jsonl'))
+  }
+
+  private async exists(path: string): Promise<boolean> {
+    try {
+      const handle = await open(path, 'r')
+      await handle.close()
+      return true
+    } catch (error) {
+      // Only ENOENT means absent. A permission/I/O error must surface, not be
+      // collapsed to `false` — otherwise load() reports "not found" and
+      // collision checks proceed under a false absence assumption.
+      if (isENOENT(error)) return false
+      throw error
+    }
+  }
+
+  /** Build a state for a session discovered on disk but not yet in memory. */
+  private async adopt(id: SessionId): Promise<SessionState> {
+    // loadCore (NOT load) — adopt runs inside an already-serialized op, so
+    // re-entering the chain via the public load() would deadlock.
+    await this.loadCore(id)
+    const state = this.states.get(id)
+    /* v8 ignore next -- loadCore always sets the state for the id */
+    if (!state) throw new Error(`failed to adopt session "${id}"`)
+    return state
+  }
+
+  private assertVersion(meta: SessionMeta): void {
+    if (meta.version !== 1) {
+      throw new Error(`unsupported session format version ${meta.version} for "${meta.id}" (only v1 is supported)`)
+    }
+  }
+
+  // --- write path (session/event → flush drain) ---
+
+  private installWritePath(): void {
+    const ctx = this.ctx
+
+    // Capture the header on creation; persist a fork's seed once. Record the
+    // init promise so flush/dispose can await it (onCreated is async).
+    ctx.on('session/created', (session) => { void this.initFor(session) })
+
+    // Snapshot + buffer every event (the live object is mutable; clone so a
+    // later in-place mutation of session.events cannot rewrite a buffered
+    // event). Serializability is guaranteed at the source — `Session.append`
+    // rejects non-JSON-serializable data before the event ever enters the log
+    // or this emit — so structuredClone here can never hit a non-cloneable
+    // value, and the durable log can never diverge from session.events.
+    ctx.on('session/event', (session, event) => {
+      let buffer = this.buffers.get(session)
+      if (!buffer) this.buffers.set(session, buffer = [])
+      buffer.push(structuredClone(event))
+    })
+
+    // Drain to the backend at the durability checkpoint.
+    ctx.on('session/flush', session => this.flush(session))
+
+    // Dispose must reach quiescence: await every session's init + final drain
+    // BEFORE returning, so no write lands after teardown (orphan rename/ENOENT).
+    ctx.effect(() => async () => {
+      await Promise.allSettled([...this.inits.values()])
+      await Promise.allSettled([...this.buffers.keys()].map(s => this.flush(s)))
+      await Promise.allSettled([...this.chains.values()])
+    }, 'session-persistence-jsonl write path')
+
+    // HMR: a hot reload does not replay session/created, so seed existing live
+    // sessions (mirrors dsh-invariants).
+    for (const session of ctx.sessions.list()) void this.initFor(session)
+  }
+
+  /** Start (once) the async init for a session and remember its promise. */
+  private initFor(session: Session): Promise<void> {
+    const existing = this.inits.get(session)
+    if (existing) return existing
+    // Snapshot the seed SYNCHRONOUSLY here — initFor runs inside the
+    // `session/created` emit, before any later `append` adds non-seed events.
+    // A clone freezes it against later mutation of the live event objects.
+    const seed = session.events.map(e => structuredClone(e))
+    const p = this.onCreated(session, seed)
+    // Attach a no-op rejection handler so a failing init (e.g. an id collision)
+    // does not surface as an unhandled rejection if no flush observes `p` before
+    // it rejects. The REAL error is still delivered: flush/dispose await the
+    // same `p` from the map and see the rejection there.
+    p.catch(() => { /* observed by flush/dispose via the stored promise */ })
+    this.inits.set(session, p)
+    return p
+  }
+
+  /**
+   * Whether a live `session`'s `seed` reproduces the first `cursor` persisted
+   * events. Reads the on-disk committed prefix and compares. A `cursor` of 0
+   * (nothing persisted yet) trivially matches. Used when a live session claims
+   * ownerless state left by a prior `load()`/`create()` — to reject a fresh,
+   * unrelated session that reuses the id and would otherwise have its seq
+   * 0..cursor-1 events filtered as already-written.
+   */
+  private async seedMatchesPersisted(session: Session, seed: readonly SessionEvent[], cursor: number): Promise<boolean> {
+    if (cursor === 0) return true
+    const onDisk = await this.findLog(session.header.id, session.header.cwd)
+    /* v8 ignore next -- a cursor > 0 means the log was materialized, so it exists */
+    if (onDisk === undefined) return false
+    const { events: diskEvents } = scanLog(await readFile(onDisk.path))
+    return seedCoversPrefix(seed, diskEvents.slice(0, cursor))
+  }
+
+  /**
+   * On session/created: sync the backend's in-memory state to a live Session.
+   *
+   * Cases, by whether this backend tracks the id and whether a log is on disk:
+   *   1. Already in `states` (created here, or a prior load/resume) → no-op.
+   *   2. Not tracked, a log EXISTS on disk, and it is a seq-aligned PREFIX of the
+   *      live session's current events → ADOPT it (HMR/reload): a fresh backend
+   *      instance (empty `states`) meets a live session whose log a previous
+   *      instance materialized; the live object already carries that history (it
+   *      is the source of truth this run), so we continue from the stored length
+   *      instead of re-creating. This keeps persistence alive across hot reload.
+   *   3. Not tracked, a log EXISTS on disk, but it is NOT a prefix of the live
+   *      session's events → REJECT: a different session collides on the id. The
+   *      SessionId is the identity, so two unrelated sessions sharing one is a
+   *      bug, not a resume — fail loudly rather than clobber committed data.
+   *   4. Not tracked and NO log on disk → a genuinely new session: register its
+   *      meta (lazy) and persist its `seed` once.
+   *
+   * The public `create(meta)` API is stricter still (rejects ANY on-disk id):
+   * there the caller asserts "brand new", so even a prefix match is a bug.
+   *
+   * The seed events were copied into the Session by its constructor WITHOUT
+   * emitting session/event, so the write-behind buffer never sees them — the
+   * one explicit `append(seed)` below is the only persistence of the seed.
+   * Events appended AFTER creation flow through the session/event buffer and
+   * are persisted by flush (filtered by the write cursor), never here.
+   */
+  private async onCreated(session: Session, seed: readonly SessionEvent[]): Promise<void> {
+    const id = session.header.id
+    const tracked = this.states.get(id)
+    if (tracked !== undefined) {
+      // case 1: already tracked.
+      // (owner === session is a defensive same-object guard: initFor dedupes by
+      // session object, so onCreated never actually runs twice for one session.)
+      /* v8 ignore next -- initFor dedupes per session object; same-object re-entry can't occur */
+      if (tracked.owner === session) return
+      if (tracked.owner === undefined) {
+        // Ownerless state was created via the public create()/load() API. The
+        // FIRST live session to arrive claims it — but ONLY if its seed is the
+        // already-persisted prefix. A load() for preview leaves cursor at the
+        // persisted length; a fresh, unrelated session reusing that id has a
+        // seed shorter than (or not matching) that prefix, so flush would filter
+        // its seq 0..cursor-1 events as already-written and silently graft the
+        // new conversation onto the old log. Verify the seed covers the cursor.
+        if (!await this.seedMatchesPersisted(session, seed, tracked.cursor)) {
+          throw new Error(`session "${id}" is already persisted with ${tracked.cursor} event(s) that do not match this live session (id collision)`)
+        }
+        tracked.owner = session
+        // Persist the live seed SUFFIX beyond the persisted prefix. Constructor
+        // seed events (from sessions.create(id, { seed })) never emit
+        // session/event, so the write-behind buffer never sees them — without
+        // this they would be lost and a later flush would seq-mismatch. (cursor
+        // is 0 for a public create(), so this covers the whole seed there.)
+        const suffix = seed.slice(tracked.cursor)
+        if (suffix.length > 0) await this.append(id, suffix)
+        return
+      }
+      // The state is owned by a DIFFERENT live session. We may reclaim the id
+      // ONLY if that owner left nothing behind: never materialized a log (cursor
+      // 0, not materialized) AND has no write-behind buffer still pending. A
+      // session that appended events but was disposed before its first flush is
+      // NOT materialized yet but DOES have buffered events — reclaiming then
+      // would let that stale buffer drain against the new session's state
+      // (persisting old events under the new id, or dropping the new session's
+      // seq-0 events). Such an owner, and any materialized owner, is a real
+      // collision and rejects; only a truly-abandoned (artifact-free) id is
+      // freed, honoring lazy materialization's "leaves nothing behind" promise.
+      const ownerBuffer = this.buffers.get(tracked.owner)
+      if (!tracked.materialized && !ownerBuffer?.length) {
+        this.states.delete(id)
+      } else {
+        throw new Error(`session "${id}" is already bound to a different live session in this backend (id collision)`)
+      }
+    }
+
+    const onDisk = await this.findLog(id, session.header.cwd)
+    if (onDisk !== undefined) {
+      // Read the committed on-disk events and check they are a seq-aligned
+      // prefix of the live session (HMR re-seeing its own session) vs. an
+      // unrelated session colliding on the id.
+      const { events: diskEvents } = scanLog(await readFile(onDisk.path))
+      if (!seedCoversPrefix(seed, diskEvents)) {
+        // case 3: genuine collision — fail loudly rather than clobber.
+        throw new Error(`session "${id}" already has a persisted log on disk that does not match this live session (id collision)`)
+      }
+      // case 2: adopt. loadCore sets the state (cursor = committed length,
+      // repair offset if a crash tail exists).
+      await this.serialize(id, () => this.loadCore(id))
+      const adopted = this.states.get(id)
+      /* v8 ignore next -- loadCore always sets the state for the id */
+      if (adopted !== undefined) adopted.owner = session
+      // Persist the live SUFFIX beyond the on-disk prefix. These events live
+      // ONLY in `seed` (the live session was ahead of disk — mid-turn at
+      // reload, or events appended while the previous backend was disposed);
+      // this backend never buffered them via session/event, so without this
+      // they would be lost and the next flush (starting at a later seq) would
+      // mismatch or skip them.
+      const suffix = seed.slice(diskEvents.length)
+      if (suffix.length > 0) await this.append(id, suffix)
+      return
+    }
+
+    // case 4: a genuinely new session. Register its meta (lazy), then persist
+    // its seed (events present at creation time) once.
+    const meta: SessionMeta = { ...session.header, updatedAt: Date.now() }
+    await this.create(meta)
+    // Bind this state to the live session so a later DIFFERENT session reusing
+    // the id is detected as a collision (case 1) rather than silently no-opped.
+    const created = this.states.get(id)
+    /* v8 ignore next -- create() always sets the state for the id */
+    if (created !== undefined) created.owner = session
+    if (seed.length > 0) {
+      await this.append(id, seed)
+    }
+  }
+
+  private async flush(session: Session): Promise<void> {
+    // Wait for the session's init (onCreated) to finish so the state/cursor and
+    // any fork-seed persistence are in place before we drain. Awaiting the same
+    // promise initFor stored also surfaces an init failure (e.g. an id
+    // collision) here, where the caller of session/flush observes it.
+    await this.inits.get(session)
+    // Serialize the WHOLE drain (read cursor → append → splice) on the
+    // per-session chain. Two concurrent flushes (e.g. an idle inject()'s
+    // fire-and-forget flush racing an explicit checkpoint) would otherwise both
+    // read the same cursor, both compute the same `fresh` slice, and the second
+    // append would seq-mismatch against the cursor the first already advanced.
+    await this.serialize(session.header.id, () => this.drain(session))
+  }
+
+  /** Drain a session's write buffer to disk. Caller serializes this per id. */
+  private async drain(session: Session): Promise<void> {
+    const buffer = this.buffers.get(session)
+    if (!buffer?.length) return
+    // Copy WITHOUT removing: the buffer is the only durable-pending copy of
+    // these events (session/event does not re-emit). Splicing before the append
+    // means a failed append (disk error, or a seq mismatch after a dropped bad
+    // event) permanently loses a completed turn. Drain the buffer only AFTER
+    // the append commits; events pushed during the await sit past batch.length
+    // and survive the prefix splice, so a retry/dispose re-drains the rest.
+    const batch = buffer.slice()
+    const state = this.states.get(session.header.id)
+    // Only append events at or beyond the write cursor (a resumed session's
+    // seed is already on disk; the cursor was set to the loaded length). flush
+    // awaits the init above, which always sets state, so the `?? 0` fallback is
+    // a defensive guard that never fires in practice.
+    /* v8 ignore next -- state is always set by the awaited init before flush */
+    const cursor = state?.cursor ?? 0
+    const fresh = batch.filter(e => e.seq >= cursor)
+    // appendCore (NOT the serialized append) — drain already runs inside the
+    // per-session chain, so re-entering it via append() would deadlock.
+    if (fresh.length > 0) await this.appendCore(session.header.id, fresh)
+    buffer.splice(0, batch.length)
+  }
+}
+
+export default SessionPersistenceJsonl

packages/session-persistence-jsonl/tsconfig.json

@@ -0,0 +1,15 @@
+{
+  "extends": "../../tsconfig.base.json",
+  "compilerOptions": {
+    "rootDir": "src",
+    "outDir": "lib"
+  },
+  "include": ["src"],
+  "references": [
+    { "path": "../../vendor/cosmokit" },
+    { "path": "../../vendor/cordis" },
+    { "path": "../../vendor/schemastery" },
+    { "path": "../session" },
+    { "path": "../session-persistence" }
+  ]
+}

packages/session-persistence/README.md

@@ -0,0 +1,33 @@
+# @deepseek-ai/dsh-session-persistence
+
+The abstract durable session-persistence seam (`ctx.sessionPersistence`). Defines WHAT a persistence backend does — durably store, reload, list, and update sessions — without saying HOW. Mirrors the `dsh-bash` capability-seam template ([ADR 0009](../../docs/adr/0009-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) travels separately as `SessionMeta`, owned by `dsh-session` and re-exported here.
+
+## Service API (`ctx.sessionPersistence`)
+
+| Method | Contract |
+|---|---|
+| `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. Preserves an interrupted (unclosed) final turn and closes it with synthetic closers — an error `tool/result` per unanswered `tool-call`, then `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; 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 (an error `tool/result` per unanswered `tool-call`, then `step/end?`+`turn/end {interrupted}`) to balance the log and keep the rehydrated history a valid provider transcript. 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.
+
+## Testing backends
+
+Import `runPersistenceContract` from `tests/contract.ts` and call it with a factory that yields a fresh, empty backend plus a teardown. Every backend is held to the same append-only / contiguous-seq / lazy-materialization / serializability semantics; a backend's own spec adds implementation-specific tests (crash repair, path sanitization) on top.
+
+> **TODO (validate the abstraction with a second backend):** `dsh-session-persistence-jsonl` is currently the only implementation, so the interface and `runPersistenceContract` are only proven against one storage model. A second backend — a SQLite implementation (`dsh-session-persistence-sqlite`), where each `SessionEvent` maps 1:1 onto a row `(session_id, seq, type, time, data)` — would run the SAME `runPersistenceContract` suite and so prove the seam is genuinely backend-agnostic (lazy materialization, crash-tail-on-load, contiguous-seq all expressed against a transactional store rather than an append-only file).
+
+## Metadata types
+
+Re-exported from `dsh-session`: `SessionHeader` (immutable: `version`, `id`, `createdAt`, `cwd?`, `parentSession?`), `SessionSummary` (mutable: `updatedAt`, `title?`, `firstPrompt?`), `SessionMeta` (their intersection).

packages/session-persistence/package.json

@@ -0,0 +1,30 @@
+{
+  "name": "@deepseek-ai/dsh-session-persistence",
+  "description": "Abstract durable session persistence seam (ctx.sessionPersistence) for the DeepSeek Harness",
+  "version": "0.0.1",
+  "private": true,
+  "type": "module",
+  "main": "lib/index.js",
+  "types": "lib/index.d.ts",
+  "exports": {
+    ".": {
+      "types": "./lib/index.d.ts",
+      "default": "./lib/index.js"
+    },
+    "./src/*": "./src/*",
+    "./package.json": "./package.json"
+  },
+  "files": [
+    "lib",
+    "src"
+  ],
+  "license": "BSD-3-Clause",
+  "peerDependencies": {
+    "@deepseek-ai/dsh-session": "^0.0.1",
+    "cordis": "^4.0.0-rc.6"
+  },
+  "devDependencies": {
+    "@deepseek-ai/dsh-session": "workspace:^",
+    "cordis": "^4.0.0-rc.6"
+  }
+}

packages/session-persistence/src/index.ts

@@ -0,0 +1,125 @@
+/**
+ * The durable session-persistence seam (`ctx.sessionPersistence`): an abstract
+ * service defining WHAT a persistence backend does — durably store, reload,
+ * list, and update sessions — without saying HOW. Implementations subclass
+ * {@link SessionPersistence} and register themselves as the
+ * `sessionPersistence` service; `@deepseek-ai/dsh-session-persistence-jsonl`
+ * (an append-only JSONL log per session) is the first. Future backends swap in
+ * SQLite/WAL, an object store, or a remote service without touching the
+ * consumers (the write-path plugin, the agent-loop resume seam).
+ *
+ * The persisted unit IS the existing {@link SessionEvent} — there is no
+ * parallel "persisted message" type the log must be converted to and from
+ * (faithful to the event-sourced model: the log is the single source of
+ * truth). Metadata that is NOT replayable conversation state (format version,
+ * cwd, lineage) travels separately as {@link SessionMeta}, which is owned by
+ * `dsh-session` and re-exported here.
+ *
+ * @module @deepseek-ai/dsh-session-persistence
+ */
+
+import { Context, Service } from 'cordis'
+import type { SessionEvent, SessionId, SessionMeta, SessionSummary } from '@deepseek-ai/dsh-session'
+
+// Re-export the metadata vocabulary so consumers import it from the seam.
+export type { SessionHeader, SessionSummary, SessionMeta } from '@deepseek-ai/dsh-session'
+
+declare module 'cordis' {
+  interface Context {
+    sessionPersistence: SessionPersistence
+  }
+}
+
+/**
+ * Abstract durable session-persistence service. Subclass, implement the
+ * abstract methods, and load the subclass as a plugin — it registers as
+ * `ctx.sessionPersistence` (one implementation per context; loading a second
+ * throws, cordis' standard duplicate-service behavior).
+ *
+ * Contracts every implementation MUST honor (a DB backend asserts them inside
+ * a transaction; a file backend appends at EOF):
+ *
+ * - **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 COMMITTED region
+ *   (unloadable); {@link append}'s first event `seq` MUST equal the backend's
+ *   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
+ *   event type. A backend snapshots (serializes/clones) each event when it
+ *   buffers, since `session.events` hands out the live mutable object.
+ * - **Durability.** {@link append} returns only once the batch is durable
+ *   (the file backend fsyncs; a DB commits). {@link create} MAY defer the
+ *   physical write until the first {@link append} (lazy materialization).
+ */
+export abstract class SessionPersistence extends Service {
+  constructor(ctx: Context) {
+    super(ctx, 'sessionPersistence')
+  }
+
+  /**
+   * Register a new session's metadata. A backend MAY defer the physical write
+   * until the first {@link append} (lazy materialization), in which case a
+   * created-but-never-appended session is absent from {@link has}/{@link list}
+   * — abandoned sessions leave nothing behind.
+   */
+  abstract create(meta: SessionMeta): Promise<void>
+
+  /**
+   * 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
+   * `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>
+
+  /**
+   * Reload a session: its {@link SessionMeta} plus the event log up to the last
+   * 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 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: an error `tool/result` for every `tool-call` the crash left
+   * unanswered (so the rehydrated history is a valid provider transcript — a
+   * dangling assistant tool-call is otherwise rejected), then 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[] }>
+
+  /** Lightweight listing from metadata, without a full-log parse. */
+  abstract list(): Promise<SessionMeta[]>
+
+  /** Whether a session is durably present (materialized). */
+  abstract has(id: SessionId): Promise<boolean>
+
+  /** Remove a session and all its persisted artifacts. */
+  abstract delete(id: SessionId): Promise<void>
+
+  /**
+   * Update mutable metadata ({@link SessionSummary}: `updatedAt`, `title`,
+   * `firstPrompt`) WITHOUT touching the append-only event log. A backend
+   * stores the summary beside the log (a sidecar file, a header row) and
+   * rewrites only it.
+   */
+  abstract update(id: SessionId, summary: Partial<SessionSummary>): Promise<void>
+}
+
+export default SessionPersistence

packages/session-persistence/tests/contract.ts

@@ -0,0 +1,264 @@
+/**
+ * Reusable contract test for any {@link SessionPersistence} backend. A backend
+ * package imports {@link runPersistenceContract} and calls it with a factory
+ * that yields a fresh, empty backend (and a teardown), so every backend is held
+ * to the same append-only / contiguous-seq / lazy-materialization / crash
+ * semantics. The JSONL backend's own spec adds file-specific tests on top.
+ *
+ * @module @deepseek-ai/dsh-session-persistence/tests/contract
+ */
+
+import { describe, expect, it } from 'vitest'
+import { SessionId } from '@deepseek-ai/dsh-session'
+import type { SessionEvent, SessionMeta } from '@deepseek-ai/dsh-session'
+import { CallId } from '@deepseek-ai/dsh-llm'
+import type { SessionPersistence } from '../src/index.ts'
+
+/** A backend under test plus its teardown. */
+export interface ContractBackend {
+  persistence: SessionPersistence
+  dispose: () => Promise<void>
+}
+
+/** Build a minimal {@link SessionMeta} for a session id. */
+export function meta(id: string, cwd?: string): SessionMeta {
+  return {
+    version: 1,
+    id: SessionId(id),
+    createdAt: 1000,
+    updatedAt: 1000,
+    ...cwd !== undefined ? { cwd } : {},
+  }
+}
+
+/** A well-formed one-turn event log (contiguous seqs from 0). */
+export function oneTurnLog(): SessionEvent[] {
+  return [
+    { type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } },
+    { type: 'user/message', seq: 1, time: 2, data: { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } } },
+    { type: 'step/start', seq: 2, time: 3, data: { turn: 1, step: 1 } },
+    { type: 'assistant/message', seq: 3, time: 4, data: { turn: 1, step: 1, content: [{ type: 'text', text: 'hello' }] } },
+    { type: 'step/end', seq: 4, time: 5, data: { turn: 1, step: 1 } },
+    { type: 'turn/end', seq: 5, time: 6, data: { turn: 1, reason: { kind: 'completed' } } },
+  ]
+}
+
+/**
+ * Run the backend-agnostic contract suite. `make()` MUST return a fresh, empty
+ * backend each call.
+ */
+export function runPersistenceContract(name: string, make: () => Promise<ContractBackend>): void {
+  describe(`SessionPersistence contract: ${name}`, () => {
+    it('round-trips a session: create + append → load returns identical meta and byte-identical events', async () => {
+      const { persistence, dispose } = await make()
+      try {
+        const m = meta('s1', '/work')
+        const log = oneTurnLog()
+        await persistence.create(m)
+        await persistence.append(m.id, log)
+
+        const loaded = await persistence.load(m.id)
+        expect(loaded.meta).toMatchObject({ version: 1, id: m.id, cwd: '/work' })
+        expect(loaded.events).toEqual(log)
+      } finally {
+        await dispose()
+      }
+    })
+
+    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('crash recovery: an interrupted tool call gets a synthetic error result so resume is a valid transcript', async () => {
+      const { persistence, dispose } = await make()
+      try {
+        const m = meta('interrupted-toolcall')
+        await persistence.create(m)
+        await persistence.append(m.id, oneTurnLog()) // turn 1, committed (seqs 0..5)
+        // Turn 2 crashed AFTER the assistant message asked for a tool call but
+        // BEFORE the tool/result was written (the loop runs tools after logging
+        // the assistant message — a process killed mid-tool lands exactly here).
+        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 } },
+          { type: 'assistant/message', seq: 8, time: 9, data: { turn: 2, step: 1, content: [
+            { type: 'tool-call', id: CallId('call-x'), name: 'bash', arguments: '{}' },
+          ] } },
+        ])
+
+        const loaded = await persistence.load(m.id)
+        // The orphaned call is answered by a synthetic error tool/result BEFORE
+        // step/end + turn/end {interrupted}, so the step (and turn) are balanced
+        // and a resumed session derives a valid transcript (no dangling call).
+        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', 'assistant/message', 'tool/result', 'step/end', 'turn/end', // turn 2
+        ])
+        const synthetic = loaded.events.find(e => e.type === 'tool/result')
+        expect(synthetic?.type === 'tool/result' && synthetic.data).toMatchObject({
+          callId: CallId('call-x'), isError: true, error: { code: 'interrupted' },
+        })
+        // The synthetic result carries the SAME callId as the orphaned tool-call,
+        // so deriveMessages() pairs them — no provider-invalid dangling call.
+        const call = loaded.events.findLast(e => e.type === 'assistant/message')
+        const callId = call?.type === 'assistant/message'
+          && call.data.content.find(b => b.type === 'tool-call')
+        expect(callId && callId.type === 'tool-call' && callId.id).toBe(CallId('call-x'))
+      } finally {
+        await dispose()
+      }
+    })
+
+    it('has()/list() exclude a created-but-never-appended (zero-event) session', async () => {
+      const { persistence, dispose } = await make()
+      try {
+        await persistence.create(meta('empty'))
+        expect(await persistence.has(SessionId('empty'))).toBe(false)
+        expect((await persistence.list()).map(m => m.id)).not.toContain(SessionId('empty'))
+      } finally {
+        await dispose()
+      }
+    })
+
+    it('has()/list() include a session once it has events', async () => {
+      const { persistence, dispose } = await make()
+      try {
+        const m = meta('s2')
+        await persistence.create(m)
+        await persistence.append(m.id, oneTurnLog())
+        expect(await persistence.has(m.id)).toBe(true)
+        expect((await persistence.list()).map(x => x.id)).toContain(m.id)
+      } finally {
+        await dispose()
+      }
+    })
+
+    it('append rejects a batch whose first seq does not match the stored next-seq', async () => {
+      const { persistence, dispose } = await make()
+      try {
+        const m = meta('s3')
+        await persistence.create(m)
+        await persistence.append(m.id, oneTurnLog()) // seqs 0..5, next-seq = 6
+        // A re-append of an already-stored seq must be rejected, not duplicated.
+        const restated = oneTurnLog()
+        await expect(persistence.append(m.id, restated)).rejects.toThrow()
+      } finally {
+        await dispose()
+      }
+    })
+
+    it('append rejects a mid-batch seq gap', async () => {
+      const { persistence, dispose } = await make()
+      try {
+        const m = meta('s4')
+        await persistence.create(m)
+        const gapped: SessionEvent[] = [
+          { type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } },
+          { type: 'step/start', seq: 2, time: 2, data: { turn: 1, step: 1 } }, // gap: missing seq 1
+        ]
+        await expect(persistence.append(m.id, gapped)).rejects.toThrow()
+      } finally {
+        await dispose()
+      }
+    })
+
+    it('append rejects non-JSON-serializable event data, naming the event type', async () => {
+      const { persistence, dispose } = await make()
+      try {
+        // Every value `isJsonValue` rejects must be rejected by the backend, not
+        // just BigInt — otherwise a backend could pass this contract while still
+        // accepting values that corrupt the durable round-trip. Each is a
+        // plugin-added `extra` field on a single user/message (seq 0).
+        const cyclic: Record<string, unknown> = { type: 'text', text: 'x' }
+        cyclic['self'] = cyclic
+        const badValues: unknown[] = [
+          1n,                  // BigInt
+          undefined,           // dropped by JSON.stringify
+          Infinity,            // → null
+          () => 0,             // function
+          Symbol('s'),         // symbol
+          new Map(),           // exotic object
+          cyclic,              // circular ref
+        ]
+        for (const [i, bad] of badValues.entries()) {
+          // A fresh session per value isolates each rejection (a rejected append
+          // must leave no state behind, but isolating keeps the assertion clean).
+          const mi = meta(`s5-${i}`)
+          await persistence.create(mi)
+          const events = [
+            { type: 'user/message', seq: 0, time: 1, data: { content: [{ type: 'text', text: 'x' }], source: { kind: 'user' }, extra: bad } },
+          ] as unknown as SessionEvent[]
+          await expect(persistence.append(mi.id, events)).rejects.toThrow(/user\/message/)
+        }
+      } finally {
+        await dispose()
+      }
+    })
+
+    it('delete removes a session', async () => {
+      const { persistence, dispose } = await make()
+      try {
+        const m = meta('s6')
+        await persistence.create(m)
+        await persistence.append(m.id, oneTurnLog())
+        expect(await persistence.has(m.id)).toBe(true)
+        await persistence.delete(m.id)
+        expect(await persistence.has(m.id)).toBe(false)
+      } finally {
+        await dispose()
+      }
+    })
+
+    it('update mutates summary fields without touching the event log', async () => {
+      const { persistence, dispose } = await make()
+      try {
+        const m = meta('s7')
+        const log = oneTurnLog()
+        await persistence.create(m)
+        await persistence.append(m.id, log)
+        await persistence.update(m.id, { title: 'My session', firstPrompt: 'hi' })
+
+        const loaded = await persistence.load(m.id)
+        expect(loaded.meta.title).toBe('My session')
+        expect(loaded.meta.firstPrompt).toBe('hi')
+        expect(loaded.events).toEqual(log) // log untouched
+      } finally {
+        await dispose()
+      }
+    })
+  })
+}

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

@@ -0,0 +1,106 @@
+import { describe, expect, it } from 'vitest'
+import { Context } from 'cordis'
+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'
+
+/**
+ * A minimal in-memory {@link SessionPersistence} used to (a) cover the abstract
+ * base's constructor + service registration and (b) validate the reusable
+ * contract suite itself. The real durable backend is
+ * `@deepseek-ai/dsh-session-persistence-jsonl`.
+ */
+class MemoryPersistence extends SessionPersistence {
+  private store = new Map<string, { meta: SessionMeta; events: SessionEvent[] }>()
+  private pending = new Map<string, SessionMeta>()
+
+  async create(m: SessionMeta): Promise<void> {
+    // Lazy: record the intended meta, but stay absent from has/list until the
+    // first append materializes the session.
+    this.pending.set(m.id, m)
+  }
+
+  async append(id: SessionId, events: readonly SessionEvent[]): Promise<void> {
+    const existing = this.store.get(id)
+    const nextSeq = existing ? existing.events.length : 0
+    if (events.length > 0 && events[0]!.seq !== nextSeq) {
+      throw new Error(`append seq mismatch for "${id}": expected ${nextSeq}, got ${events[0]!.seq}`)
+    }
+    for (let i = 0; i < events.length; i++) {
+      const e = events[i]!
+      if (e.seq !== nextSeq + i) throw new Error(`non-contiguous seq in batch for "${id}" at index ${i}`)
+      if (!isJsonValue(e.data)) {
+        throw new Error(`event "${e.type}" carries non-JSON-serializable data`)
+      }
+    }
+    if (!existing) {
+      const m = this.pending.get(id)
+      if (!m) throw new Error(`append before create for "${id}"`)
+      this.store.set(id, { meta: m, events: structuredClone(events) as SessionEvent[] })
+    } else {
+      existing.events.push(...structuredClone(events) as SessionEvent[])
+    }
+  }
+
+  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) }
+  }
+
+  async list(): Promise<SessionMeta[]> {
+    return [...this.store.values()].map(e => structuredClone(e.meta))
+  }
+
+  async has(id: SessionId): Promise<boolean> {
+    return this.store.has(id)
+  }
+
+  async delete(id: SessionId): Promise<void> {
+    this.store.delete(id)
+    this.pending.delete(id)
+  }
+
+  async update(id: SessionId, summary: Partial<SessionSummary>): Promise<void> {
+    const entry = this.store.get(id)
+    if (entry) Object.assign(entry.meta, summary)
+  }
+}
+
+// Run the shared contract against the in-memory backend.
+runPersistenceContract('memory', async () => {
+  const ctx = new Context()
+  const fiber = await ctx.plugin(MemoryPersistence)
+  return {
+    persistence: ctx.sessionPersistence,
+    dispose: async () => { await fiber.dispose() },
+  }
+})
+
+describe('SessionPersistence service registration', () => {
+  it('registers as ctx.sessionPersistence and is removed on fiber dispose (HMR safety)', async () => {
+    const ctx = new Context()
+    const fiber = await ctx.plugin(MemoryPersistence)
+    expect(ctx.sessionPersistence).toBeInstanceOf(SessionPersistence)
+
+    await fiber.dispose()
+    expect(ctx.sessionPersistence).toBeUndefined()
+  })
+
+  it('round-trips through the registered service instance', async () => {
+    const ctx = new Context()
+    const fiber = await ctx.plugin(MemoryPersistence)
+    const m = meta('reg')
+    await ctx.sessionPersistence.create(m)
+    await ctx.sessionPersistence.append(m.id, oneTurnLog())
+    const loaded = await ctx.sessionPersistence.load(m.id)
+    expect(loaded.events).toHaveLength(6)
+    await fiber.dispose()
+  })
+})

packages/session-persistence/tsconfig.json

@@ -0,0 +1,13 @@
+{
+  "extends": "../../tsconfig.base.json",
+  "compilerOptions": {
+    "rootDir": "src",
+    "outDir": "lib"
+  },
+  "include": ["src"],
+  "references": [
+    { "path": "../../vendor/cosmokit" },
+    { "path": "../../vendor/cordis" },
+    { "path": "../session" }
+  ]
+}

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,138 @@
+/**
+ * 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:
+ *
+ *   1. an error `tool/result` for every `tool-call` in the interrupted turn that
+ *      never got its matching `tool/result` (so the rehydrated history is a
+ *      VALID provider transcript — see below),
+ *   2. a `step/end` if a step was still open, then
+ *   3. 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.
+ *
+ * Why the synthetic tool results matter: `deriveMessages()` renders the
+ * `tool-call` blocks inside a durable `assistant/message` but only emits a
+ * matching tool-result when a `tool/result` EVENT exists. A crash between the
+ * assistant message and its tool results (the loop runs the tools AFTER logging
+ * the assistant message, so a process killed mid-tool leaves the calls without
+ * results) would otherwise reload a history with a dangling assistant tool-call
+ * — which every provider rejects as an invalid transcript on the next request.
+ * Synthesizing an error result per orphaned call keeps resume safe.
+ *
+ * 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 { CallId } from '@deepseek-ai/dsh-llm'
+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.
+ *
+ * The closers, in order: an error `tool/result` for each unmatched `tool-call`
+ * in the interrupted turn, then a `step/end` if a step is open, then the
+ * `turn/end {interrupted}`. The tool-results come first so a step that issued
+ * tool calls is balanced (every call has a result) before its `step/end`.
+ *
+ * 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
+  // Track tool calls vs. their results WITHIN the currently-open turn only: a
+  // call is "pending" until its matching tool/result arrives. Reset at every
+  // turn boundary so a committed earlier turn (already balanced) never leaks a
+  // phantom pending call into the interrupted-turn repair.
+  const pendingCalls = new Map<CallId, { step: number }>()
+  for (const event of events) {
+    switch (event.type) {
+      case 'turn/start':
+        openTurn = event.data.turn
+        openStep = null
+        pendingCalls.clear()
+        break
+      case 'turn/end':
+        openTurn = null
+        openStep = null
+        pendingCalls.clear()
+        break
+      case 'step/start':
+        openStep = event.data.step
+        break
+      case 'step/end':
+        openStep = null
+        break
+      case 'assistant/message':
+        // The assistant message carries the tool-call blocks; each is pending
+        // until a tool/result event with the same callId is logged.
+        for (const block of event.data.content) {
+          if (block.type === 'tool-call') pendingCalls.set(block.id, { step: event.data.step })
+        }
+        break
+      case 'tool/result':
+        pendingCalls.delete(event.data.callId)
+        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[] = []
+
+  // Synthesize an error tool/result for each tool-call left unanswered by the
+  // crash, so deriveMessages() yields a valid provider transcript on resume (a
+  // dangling assistant tool-call is rejected by every provider). Insertion
+  // order follows the Map (insertion = log order of the assistant messages).
+  for (const [callId, { step }] of pendingCalls) {
+    closers.push({
+      type: 'tool/result',
+      seq: seq++,
+      time,
+      data: {
+        turn: openTurn,
+        step,
+        callId,
+        content: [{ type: 'text', text: 'Tool call interrupted by a crash; no result was recorded.' }],
+        isError: true,
+        error: { name: 'InterruptedError', code: 'interrupted' },
+      },
+    })
+  }
+
+  // Close an open step next — 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]

packages/session/tests/repair.spec.ts

@@ -0,0 +1,125 @@
+import { describe, expect, it } from 'vitest'
+import { CallId } from '@deepseek-ai/dsh-llm'
+import { interruptedTurnClosers } from '../src/index.ts'
+import type { SessionEvent } from '../src/index.ts'
+
+/**
+ * Unit coverage for the crash-recovery closer synthesis. The persistence
+ * contract exercises it end-to-end through both backends; these tests pin the
+ * pure function's branches directly — especially the synthetic error
+ * `tool/result` for a tool call the crash left unanswered (without it a
+ * resumed session replays a dangling assistant tool-call and the provider
+ * rejects the transcript).
+ */
+
+const userTurnStart = (turn: number, seq: number): SessionEvent =>
+  ({ type: 'turn/start', seq, time: seq, data: { turn, trigger: { kind: 'message', source: { kind: 'user' } } } })
+
+describe('interruptedTurnClosers', () => {
+  it('returns nothing for a balanced log (ends on turn/end)', () => {
+    const balanced: SessionEvent[] = [
+      userTurnStart(1, 0),
+      { type: 'turn/end', seq: 1, time: 1, data: { turn: 1, reason: { kind: 'completed' } } },
+    ]
+    expect(interruptedTurnClosers(balanced)).toEqual([])
+  })
+
+  it('returns nothing for an empty log', () => {
+    expect(interruptedTurnClosers([])).toEqual([])
+  })
+
+  it('closes an open turn with no open step (turn/end {interrupted} only)', () => {
+    const events: SessionEvent[] = [userTurnStart(1, 0)]
+    const closers = interruptedTurnClosers(events)
+    expect(closers.map(e => e.type)).toEqual(['turn/end'])
+    const end = closers[0]!
+    expect(end.seq).toBe(1)
+    expect(end.type === 'turn/end' && end.data.reason).toEqual({ kind: 'interrupted' })
+  })
+
+  it('closes an open step before the turn (step/end then turn/end)', () => {
+    const events: SessionEvent[] = [
+      userTurnStart(1, 0),
+      { type: 'step/start', seq: 1, time: 1, data: { turn: 1, step: 1 } },
+    ]
+    const closers = interruptedTurnClosers(events)
+    expect(closers.map(e => e.type)).toEqual(['step/end', 'turn/end'])
+    expect(closers.map(e => e.seq)).toEqual([2, 3])
+  })
+
+  it('synthesizes an error tool/result for a tool-call the crash left unanswered', () => {
+    // A step issued one tool call (in the assistant message) but crashed before
+    // the tool/result was logged — the classic mid-tool crash.
+    const events: SessionEvent[] = [
+      userTurnStart(2, 0),
+      { type: 'step/start', seq: 1, time: 1, data: { turn: 2, step: 1 } },
+      { type: 'assistant/message', seq: 2, time: 2, data: { turn: 2, step: 1, content: [
+        { type: 'text', text: 'calling a tool' },
+        { type: 'tool-call', id: CallId('call-1'), name: 'bash', arguments: '{}' },
+      ] } },
+    ]
+    const closers = interruptedTurnClosers(events)
+    // tool/result (for the orphaned call) → step/end → turn/end, contiguous seqs.
+    expect(closers.map(e => e.type)).toEqual(['tool/result', 'step/end', 'turn/end'])
+    expect(closers.map(e => e.seq)).toEqual([3, 4, 5])
+    const result = closers[0]!
+    expect(result.type === 'tool/result' && result.data).toMatchObject({
+      turn: 2, step: 1, callId: CallId('call-1'), isError: true, error: { code: 'interrupted' },
+    })
+  })
+
+  it('does NOT synthesize a result for a tool-call that already has one', () => {
+    const events: SessionEvent[] = [
+      userTurnStart(2, 0),
+      { type: 'step/start', seq: 1, time: 1, data: { turn: 2, step: 1 } },
+      { type: 'assistant/message', seq: 2, time: 2, data: { turn: 2, step: 1, content: [
+        { type: 'tool-call', id: CallId('call-1'), name: 'bash', arguments: '{}' },
+      ] } },
+      { type: 'tool/result', seq: 3, time: 3, data: { turn: 2, step: 1, callId: CallId('call-1'), content: [{ type: 'text', text: 'ok' }], isError: false } },
+    ]
+    // The call is answered, so only the open step + turn need closing.
+    const closers = interruptedTurnClosers(events)
+    expect(closers.map(e => e.type)).toEqual(['step/end', 'turn/end'])
+  })
+
+  it('synthesizes results only for the still-open turn, not a committed earlier turn', () => {
+    // Turn 1 completed with its own tool call+result (balanced). Turn 2 crashed
+    // with an unanswered call. Only turn 2's call must get a synthetic result.
+    const events: SessionEvent[] = [
+      userTurnStart(1, 0),
+      { type: 'step/start', seq: 1, time: 1, data: { turn: 1, step: 1 } },
+      { type: 'assistant/message', seq: 2, time: 2, data: { turn: 1, step: 1, content: [
+        { type: 'tool-call', id: CallId('old-call'), name: 'bash', arguments: '{}' },
+      ] } },
+      { type: 'tool/result', seq: 3, time: 3, data: { turn: 1, step: 1, callId: CallId('old-call'), content: [], isError: false } },
+      { type: 'step/end', seq: 4, time: 4, data: { turn: 1, step: 1 } },
+      { type: 'turn/end', seq: 5, time: 5, data: { turn: 1, reason: { kind: 'completed' } } },
+      userTurnStart(2, 6),
+      { type: 'step/start', seq: 7, time: 7, data: { turn: 2, step: 1 } },
+      { type: 'assistant/message', seq: 8, time: 8, data: { turn: 2, step: 1, content: [
+        { type: 'tool-call', id: CallId('new-call'), name: 'bash', arguments: '{}' },
+      ] } },
+    ]
+    const closers = interruptedTurnClosers(events)
+    expect(closers.map(e => e.type)).toEqual(['tool/result', 'step/end', 'turn/end'])
+    const result = closers[0]!
+    expect(result.type === 'tool/result' && result.data.callId).toBe('new-call')
+  })
+
+  it('synthesizes a result for each of multiple unanswered calls, in log order', () => {
+    const events: SessionEvent[] = [
+      userTurnStart(1, 0),
+      { type: 'step/start', seq: 1, time: 1, data: { turn: 1, step: 1 } },
+      { type: 'assistant/message', seq: 2, time: 2, data: { turn: 1, step: 1, content: [
+        { type: 'tool-call', id: CallId('call-a'), name: 'bash', arguments: '{}' },
+        { type: 'tool-call', id: CallId('call-b'), name: 'bash', arguments: '{}' },
+      ] } },
+      // call-a got answered before the crash; call-b did not.
+      { type: 'tool/result', seq: 3, time: 3, data: { turn: 1, step: 1, callId: CallId('call-a'), content: [], isError: false } },
+    ]
+    const closers = interruptedTurnClosers(events)
+    expect(closers.map(e => e.type)).toEqual(['tool/result', 'step/end', 'turn/end'])
+    const result = closers[0]!
+    expect(result.type === 'tool/result' && result.data.callId).toBe('call-b')
+  })
+})

pnpm-lock.yaml

@@ -93,6 +93,9 @@ importers:
       '@deepseek-ai/dsh-session':
         specifier: workspace:^
         version: link:../session
+      '@deepseek-ai/dsh-session-persistence-jsonl':
+        specifier: workspace:^
+        version: link:../session-persistence-jsonl
       '@deepseek-ai/dsh-system-prompt':
         specifier: workspace:^
         version: link:../system-prompt
@@ -184,6 +187,31 @@ importers:
         specifier: ^4.0.0-rc.6
         version: 4.0.0-rc.6(@cordisjs/plugin-include@1.0.4)(@cordisjs/plugin-loader@1.0.0-rc.4)
 
+  packages/session-persistence:
+    devDependencies:
+      '@deepseek-ai/dsh-session':
+        specifier: workspace:^
+        version: link:../session
+      cordis:
+        specifier: ^4.0.0-rc.6
+        version: 4.0.0-rc.6(@cordisjs/plugin-include@1.0.4)(@cordisjs/plugin-loader@1.0.0-rc.4)
+
+  packages/session-persistence-jsonl:
+    dependencies:
+      schemastery:
+        specifier: ^3.18.0
+        version: 3.18.0
+    devDependencies:
+      '@deepseek-ai/dsh-session':
+        specifier: workspace:^
+        version: link:../session
+      '@deepseek-ai/dsh-session-persistence':
+        specifier: workspace:^
+        version: link:../session-persistence
+      cordis:
+        specifier: ^4.0.0-rc.6
+        version: 4.0.0-rc.6(@cordisjs/plugin-include@1.0.4)(@cordisjs/plugin-loader@1.0.0-rc.4)
+
   packages/system-prompt:
     devDependencies:
       '@deepseek-ai/dsh-llm':

scripts/publint-all.ts

@@ -6,6 +6,8 @@ import { resolve } from 'node:path'
 const packages = [
   'packages/llm',
   'packages/session',
+  'packages/session-persistence',
+  'packages/session-persistence-jsonl',
   'packages/system-prompt',
   'packages/tools',
   'packages/agent',

tsconfig.base.json

@@ -36,6 +36,8 @@
       "@cordisjs/plugin-logger-console": ["./vendor/logger-console/src"],
       "@deepseek-ai/dsh-llm": ["./packages/llm/src"],
       "@deepseek-ai/dsh-session": ["./packages/session/src"],
+      "@deepseek-ai/dsh-session-persistence": ["./packages/session-persistence/src"],
+      "@deepseek-ai/dsh-session-persistence-jsonl": ["./packages/session-persistence-jsonl/src"],
       "@deepseek-ai/dsh-system-prompt": ["./packages/system-prompt/src"],
       "@deepseek-ai/dsh-tools": ["./packages/tools/src"],
       "@deepseek-ai/dsh-agent": ["./packages/agent/src"],

tsconfig.build.json

@@ -12,6 +12,8 @@
     { "path": "./vendor/logger-console" },
     { "path": "./packages/llm" },
     { "path": "./packages/session" },
+    { "path": "./packages/session-persistence" },
+    { "path": "./packages/session-persistence-jsonl" },
     { "path": "./packages/system-prompt" },
     { "path": "./packages/agent" },
     { "path": "./packages/tools" },

tsconfig.typecheck.json

@@ -18,6 +18,8 @@
       "@cordisjs/plugin-logger-console": ["./vendor/logger-console/lib/shared"],
       "@deepseek-ai/dsh-llm": ["./packages/llm/src"],
       "@deepseek-ai/dsh-session": ["./packages/session/src"],
+      "@deepseek-ai/dsh-session-persistence": ["./packages/session-persistence/src"],
+      "@deepseek-ai/dsh-session-persistence-jsonl": ["./packages/session-persistence-jsonl/src"],
       "@deepseek-ai/dsh-system-prompt": ["./packages/system-prompt/src"],
       "@deepseek-ai/dsh-tools": ["./packages/tools/src"],
       "@deepseek-ai/dsh-agent": ["./packages/agent/src"],