Merge remote-tracking branch 'origin/master' into codex/remove-stdio-agent

# Conflicts:
#	AGENTS.md
#	docs/config-catalog.md
#	examples/echo-agent/README.md
#	packages/examples/stdio-demo/README.md
#	packages/examples/stdio-demo/src/index.ts
#	packages/examples/stdio-demo/tests/built-bin.e2e.ts
#	packages/examples/stdio-demo/tests/stdio-agent.spec.ts
#	scripts/run-gates.ts

.agents/notes/implemented/architecture/2026-06-14-session-persistence.md

@@ -13,7 +13,7 @@ The [event-sourced model](2026-06-11-event-sourced-sessions.md) makes the append
 Persistence is an abstract **capability seam** ([capability seams](2026-06-13-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`. 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`**).
+2. **Implementation** (`dsh-session-persistence-jsonl`) — an append-only logical JSONL log per session (a `SessionHeader` line then one `SessionEvent` per line, verbatim **including `assistant/chunk`**), encoded as [checksummed Zstandard frames by default](2026-07-19-zstandard-jsonl-session-logs.md) or raw lines by configuration.
 
 Key choices recorded here because they are durable, contested, and surprising:
 

.agents/notes/implemented/architecture/2026-06-18-shared-persistence-write-coordinator.md

@@ -28,7 +28,7 @@ Six methods (five required + an optional lifecycle hook) — the only seam betwe
 
 ### The opaque torn marker
 
-The single design choice that keeps the seam clean: the crash-repair "where is the torn tail" token is OPAQUE to the coordinator. The coordinator computes the synthetic closers (it owns `interruptedTurnClosers` from `dsh-session`), but it only ever tests `tornMarker !== undefined` and passes the value straight back to `commitRepair` — it never inspects it. Each backend picks its own marker type: JSONL uses the byte offset to truncate to, SQLite the seq to delete from (both happen to be `number`). The JSONL backend folds its `committedBytes < buffer.byteLength` comparison INSIDE the hook so the returned marker is already `number | undefined`; without that fold the coordinator would have to know about byte lengths.
+The single design choice that keeps the seam clean: the crash-repair "where is the torn tail" token is OPAQUE to the coordinator. The coordinator computes the synthetic closers (it owns `interruptedTurnClosers` from `dsh-session`), but it only ever tests `tornMarker !== undefined` and passes the value straight back to `commitRepair` — it never inspects it. Each backend picks its own marker type: JSONL carries the byte offset to truncate to plus any complete events decoded from an incomplete final frame, while SQLite carries the seq to delete from. The coordinator therefore knows neither byte lengths nor frame recovery state.
 
 ## Testing
 

.agents/notes/implemented/architecture/2026-07-19-zstandard-jsonl-session-logs.i18n.yaml

@@ -0,0 +1,6 @@
+# Bilingual-pair consistency record (docs/i18n/README.md): the git blob hash of each
+# side as of the last confirmed-consistent state. Both languages carry equal authority;
+# after editing either side, bring the other along and re-record with:
+#   pnpm run verify-translation-pairing --write
+2026-07-19-zstandard-jsonl-session-logs.md: 09d30594fe31eed138a128dabc1947b15857808d
+2026-07-19-zstandard-jsonl-session-logs.zh.md: 131531d9dba7cb01407191bf937f8b0ee3c6860a

.agents/notes/implemented/architecture/2026-07-19-zstandard-jsonl-session-logs.md

@@ -0,0 +1,57 @@
+# Agent Note: Zstandard JSONL session logs
+
+Status: implemented
+
+English | [中文](2026-07-19-zstandard-jsonl-session-logs.zh.md)
+
+## Problem
+
+The JSONL persistence backend keeps every `SessionEvent` verbatim, including high-volume `assistant/chunk` records. Raw text makes logs inspectable but spends storage and I/O on repeated JSON keys and model text. Compression must retain the existing append/fsync commit boundary, collision-safe first materialization, crash repair, and metadata-only listing; rewriting a whole compressed file after every turn would discard those properties.
+
+The encoding also has to remain explicit at the deployment boundary. Snapshot fixtures and external line readers require raw JSONL, while a backend cannot safely guess between compressed and raw artifacts in one root or silently migrate pre-release session data.
+
+## Decision
+
+### Configuration and suffix ownership
+
+`dsh-session-persistence-jsonl` accepts `compression?: 'zstd' | 'none'` and explicitly resolves omission to `'zstd'`. Zstandard artifacts end in `.jsonl.zstd`; `'none'` retains the original newline-delimited UTF-8 `.jsonl` representation. `SessionLocation.kind` remains `'jsonl'`, because both encodings carry the same logical record format, and `SESSION_FORMAT_VERSION` remains `0` under the repository's pre-release reject-without-migration policy.
+
+Each persistence root belongs to one encoding. A one-time discovery preflight rejects any opposite suffix, and targeted load, live-adoption, listing, and materialization paths repeat the relevant suffix check after an initially empty preflight. The error names the incompatible artifact and directs the deployment to the matching configuration or a separate root. There is no migration, dual read, dual write, or extension-based fallback.
+
+### Frame and write path
+
+The compressed artifact is a standard concatenation of independent [Zstandard frames](https://datatracker.ietf.org/doc/html/rfc8878): one checksummed frame containing exactly the header line, followed by one checksummed frame for every durable append batch. Normal loop batches are turn commits, so frame boundaries preserve the existing persistence checkpoint without making the storage layer depend on turn event types.
+
+Compression uses Node's built-in [`zstdCompress` and `zstdDecompress`](https://nodejs.org/download/release/v22.19.0/docs/api/zlib.html), available at the repository's Node 22.19 floor. The backend enables `ZSTD_c_checksumFlag`, otherwise accepts Node's defaults, and exposes neither a compression-level knob nor a new dependency. The API is marked experimental by Node, so the Node 22.19, 24, and 26 compatibility gate exercises the exact helper.
+
+First materialization compresses the two initial frames before opening the temporary file, then keeps the existing write, file `fsync`, collision-safe hard-link publication, and directory `fsync` sequence. Later batches are compressed before opening the destination and appended at EOF. A caught write or file-sync failure truncates to the prior byte length, syncs the rollback, and rethrows so the coordinator can retry the unchanged batch.
+
+### Read, listing, and crash recovery
+
+A frame-boundary scanner reads the standard magic, variable header fields, block headers and payload sizes, and optional checksum trailer. It does not interpret compressed blocks. Complete frames are decompressed independently and sequentially, which validates their checksums, and their plaintext is passed to the existing JSONL scanner. A checksum/decompression failure in any complete frame, a malformed complete-frame JSONL tail, or invalid frame structure is corruption and rejects.
+
+Listing reads in bounded chunks only until the first complete frame is available, validates and decompresses that header frame, and never reads an event frame. The dedicated header frame therefore preserves metadata-only listing even for very large session logs.
+
+EOF inside the final frame is a recoverable torn tail. Node's decoder is given the available frame prefix; every complete newline-terminated event it emits is retained. Repair truncates from that frame's starting byte and appends one new checksummed frame containing the recovered complete events followed by the coordinator's synthetic tool, step, and turn closers. If the tear occurs before any complete event is decodable, repair drops the partial frame and retains all prior complete frames.
+
+### Consumers and verification
+
+The CLI, ACP, and stdio app bundles expose symmetric `persistenceCompression` pass-through configuration. Snapshot recording and replay compositions select `'none'` explicitly because committed fixtures are raw JSONL inputs to replay and normalization; ordinary runtime compositions use the compressed default.
+
+The shared persistence and coordinator contracts run against both encodings. Backend tests cover standard framing and checksum interoperability, header-only listing, append rollback, encoding mismatch rejection, complete-frame corruption, and final-frame tears through headers, blocks, and checksum trailers. Default runtime, built-bin, headless, ACP, and Python smokes assert the compressed suffix and Zstandard magic or decode the header; raw-content tests opt out explicitly.
+
+## Alternatives considered
+
+- **One frame per JSONL record** — rejected because it multiplies frame headers and checksums for high-volume chunk events and makes a physical boundary unrelated to the durable append batch.
+- **Rewrite one whole compressed stream after every append** — rejected because cost grows with log size and replacement would give up append/fsync rollback and the established collision-safe materialization mechanics.
+- **Use a streaming compressor across appends** — rejected because an interrupted encoder state does not leave independently checksummed append units, complicating bounded listing and frame-start repair.
+- **Add an external native Zstandard dependency** — rejected because the supported Node floor already provides the required codec; another native artifact would enlarge installation and executable-packaging risk without adding a required behavior.
+- **Expose compression level or keep raw JSONL as the default** — rejected because there is no deployment evidence for a second tuning policy, while `'none'` preserves the line-readable path for fixtures and integrations that need it.
+
+## Consequences
+
+- Ordinary session roots store `.jsonl.zstd` and retain append-only, fsync, rollback, and interrupted-turn recovery semantics.
+- Raw JSONL remains a deliberate configuration, but changing encoding requires a fresh/separate root or selecting the mode that matches existing artifacts.
+- One frame per durable batch adds bounded framing/checksum overhead and allows header-only listing plus repair from an exact append boundary.
+- External tools must understand concatenated Zstandard frames or consume raw-mode artifacts; generic one-shot Node decompression reads only the first independent frame, so backend reads walk frames explicitly.
+- The implementation depends on Node's experimental built-in Zstandard API without an npm dependency; the supported-version compatibility gate makes drift visible.

.agents/notes/implemented/architecture/2026-07-19-zstandard-jsonl-session-logs.zh.md

@@ -0,0 +1,57 @@
+# Agent Note: Zstandard JSONL 会话日志
+
+Status: implemented
+
+[English](2026-07-19-zstandard-jsonl-session-logs.md) | 中文
+
+## 问题
+
+JSONL 持久化后端会逐字保留每个 `SessionEvent`，其中包括数量庞大的 `assistant/chunk` 记录。原始文本便于检查，但重复的 JSON 键和模型文本会增加存储与 I/O 开销。压缩编码必须保留既有的 append/fsync 提交边界、首次物化时的无冲突发布、崩溃修复以及仅元数据列举；如果每轮都重写整个压缩文件，就会失去这些属性。
+
+编码还必须在部署边界上保持显式。快照 fixture 与外部逐行读取器需要原始 JSONL，而后端无法在同一根目录中安全猜测压缩产物与原始产物，也不能静默迁移预发布会话数据。
+
+## 决策
+
+### 配置与后缀归属
+
+`dsh-session-persistence-jsonl` 接受 `compression?: 'zstd' | 'none'`，并将省略值显式解析为 `'zstd'`。Zstandard 产物使用 `.jsonl.zstd` 后缀；`'none'` 保留原有的换行分隔 UTF-8 `.jsonl` 表示。`SessionLocation.kind` 仍为 `'jsonl'`，因为两种编码承载同一逻辑记录格式；按照仓库的预发布拒绝且不迁移策略，`SESSION_FORMAT_VERSION` 仍为 `0`。
+
+每个持久化根目录只归属于一种编码。一次性的发现预检会拒绝任何相反后缀，而针对性的加载、活跃采用、列举与物化路径会在最初空目录预检之后再次执行对应后缀检查。错误会指出不兼容产物，并要求部署选择匹配配置或单独根目录。系统不提供迁移、双重读取、双重写入或基于扩展名的兜底。
+
+### 帧与写入路径
+
+压缩产物是标准独立 [Zstandard 帧](https://datatracker.ietf.org/doc/html/rfc8878)的串联：第一个带校验和的帧只包含头部行，后续每个持久追加批次各占一个带校验和的帧。正常 agent loop 批次就是轮次提交，因此帧边界保留既有持久化检查点，同时不让存储层依赖轮次事件类型。
+
+压缩使用 Node 内置的 [`zstdCompress` 与 `zstdDecompress`](https://nodejs.org/download/release/v22.19.0/docs/api/zlib.html)，仓库最低支持的 Node 22.19 已提供这些 API。后端启用 `ZSTD_c_checksumFlag`，其余采用 Node 默认值，不公开压缩级别调节项，也不增加依赖。Node 将该 API 标记为实验性，因此 Node 22.19、24 与 26 兼容性门禁会执行同一个辅助实现。
+
+首次物化会在打开临时文件之前压缩两个初始帧，然后保留既有的写入、文件 `fsync`、避免冲突的硬链接发布与目录 `fsync` 顺序。后续批次也会先压缩，再打开目标并在 EOF 追加。捕获到写入或文件同步失败时，后端会截断到原有字节长度，同步回滚结果，再重新抛出错误，让协调器重试未变化的批次。
+
+### 读取、列举与崩溃恢复
+
+帧边界扫描器会读取标准魔数、可变头字段、块头与负载长度，以及可选校验和尾部，但不会解释压缩块。后端独立且按顺序解压完整帧，由此验证各帧校验和，再把明文交给既有 JSONL 扫描器。任何完整帧的校验和或解压失败、完整帧中畸形的 JSONL 尾部，或者无效帧结构都属于损坏并拒绝加载。
+
+列举只按有界分片读取到第一个完整帧可用为止，验证并解压该头部帧，绝不读取事件帧。因此，即使会话日志很大，专用头部帧仍能维持仅元数据列举。
+
+最终帧内部遇到 EOF 属于可恢复的撕裂尾部。后端把已有帧前缀交给 Node 解码器，并保留其产出的每个完整、以换行结束的事件。修复从该帧起始字节截断，再追加一个新的带校验和帧，其中依次包含恢复出的完整事件，以及协调器生成的工具、步骤与轮次闭合事件。如果撕裂位置尚不足以解码任何完整事件，修复会丢弃该不完整帧并保留此前全部完整帧。
+
+### 消费方与验证
+
+CLI、ACP 与 stdio 应用包公开对称的 `persistenceCompression` 透传配置。快照录制与回放组合显式选择 `'none'`，因为提交的 fixture 是回放与规范化过程使用的原始 JSONL 输入；普通运行时组合使用压缩默认值。
+
+共享持久化契约与协调器契约会针对两种编码运行。后端测试覆盖标准帧与校验和互操作性、仅头部列举、追加回滚、编码不匹配拒绝、完整帧损坏，以及横跨头部、块和校验和尾部的最终帧撕裂。默认运行时、构建后二进制、headless、ACP 与 Python 冒烟测试会断言压缩后缀与 Zstandard 魔数，或解码头部；读取原始内容的测试则显式退出压缩。
+
+## 考虑过的替代方案
+
+- **每条 JSONL 记录一个帧**——不予采纳，因为它会让大量分片事件各自承担帧头与校验和开销，并让物理边界脱离持久追加批次。
+- **每次追加都重写一个完整压缩流**——不予采纳，因为成本会随日志大小增长，而且替换操作会放弃追加/fsync 回滚和既有的无冲突物化机制。
+- **跨追加使用流式压缩器**——不予采纳，因为编码器状态中断后不会留下可独立校验的追加单元，从而使有界列举与按帧起点修复更复杂。
+- **增加外部原生 Zstandard 依赖**——不予采纳，因为受支持的 Node 最低版本已经提供所需编解码器；另一个原生产物会增加安装与可执行文件打包风险，却不增加必需行为。
+- **公开压缩级别或继续默认使用原始 JSONL**——不予采纳，因为没有部署证据支持第二种调节策略，而 `'none'` 已为需要逐行读取的 fixture 与集成保留路径。
+
+## 后果
+
+- 普通会话根目录存储 `.jsonl.zstd`，并保留仅追加、fsync、回滚与中断轮次恢复语义。
+- 原始 JSONL 仍是显式配置，但切换编码需要使用全新或单独根目录，或者选择与既有产物匹配的模式。
+- 每个持久批次一个帧会增加有界的帧与校验和开销，同时支持仅头部列举和从精确追加边界开始修复。
+- 外部工具必须理解串联的 Zstandard 帧，或者消费原始模式产物；Node 通用的一次性解压只读取第一个独立帧，因此后端读取会显式遍历各帧。
+- 实现依赖 Node 的实验性内置 Zstandard API，但不增加 NPM 依赖；受支持版本兼容性门禁会暴露 API 漂移。

.agents/notes/implemented/testing/2026-06-19-acp-snapshot-tests.md

@@ -40,7 +40,7 @@ Replay is positional and therefore permits only one in-flight model stream per s
 
 ### Recording harvests the log; keyless replay needs a providerless config
 
-Recording runs the scenario with the real `llm-deepseek` adapter and the JSONL persistence backend, then copies the produced `.jsonl` into the scenario dir. Per-event appends are durable, but the harness shuts the subprocess down gracefully (close stdin → `await ctx.dispose()`) before harvesting so the final events are flushed. `llm-replay` itself does no recording — it is replay-only.
+Recording runs the scenario with the real `llm-deepseek` adapter and the JSONL persistence backend configured with `persistenceCompression: 'none'`, then copies the produced `.jsonl` into the scenario dir. The explicit raw mode keeps committed replay fixtures line-readable while ordinary deployments use the backend's compressed default. Per-event appends are durable, but the harness shuts the subprocess down gracefully (close stdin → `await ctx.dispose()`) before harvesting so the final events are flushed. `llm-replay` itself does no recording — it is replay-only.
 
 Replay uses a `cordis.snapshot.yml` overlay that replaces the real adapter with `llm-replay` while retaining the live composition. Recording uses the ordinary config and a harness-supplied persistence root. Replay mode skips `.env` loading, so a stray API key cannot trigger a live call. See the [single-source config Agent Note](2026-07-04-single-source-acp-replay-config.md).
 

docs/architecture.md

@@ -137,7 +137,7 @@ The session log is the source of truth. `deriveMessages()` projects session even
 
 **Model-visible ⟺ logged**: the log reconstructs every request — messages at `step/start` fronted by the header's session prefix, headers by folding `request/header` — and dev invariants assert this ([reconstructability](../.agents/notes/implemented/architecture/2026-07-05-reconstructable-requests.md)).
 
-Durability is a plugin concern. Persistence backends buffer synchronous `session/event` notifications and the loop awaits a turn-end checkpoint before moving on. The `SessionPersistence` seam stores `SessionEvent` directly, with metadata in `SessionHeader`; JSONL and SQLite share one contract suite.
+Durability is a plugin concern. Backends buffer synchronous `session/event` notifications; the loop awaits a turn-end checkpoint. `SessionPersistence` stores `SessionEvent` directly and metadata in `SessionHeader`; JSONL defaults to checksummed Zstandard, with SQLite under one contract.
 
 ### Model Content
 

docs/config-catalog.md

@@ -58,6 +58,8 @@ export interface Config {
   dshHome?: string
   /** Directory the JSONL session backend writes under. Defaults to `./.sessions`. */
   persistenceRoot?: string
+  /** JSONL artifact encoding; defaults to checksummed Zstandard frames. */
+  persistenceCompression?: JsonlCompression
   /** Controls automatic AGENTS.md/CLAUDE.md loading; configure a byte budget or set `false`. */
   workspaceContext: agentCore.Config['workspaceContext']
   /** Skill registry, local-provider, and model-facing consumer config forwarded to agent-spine-demo. */
@@ -69,9 +71,9 @@ export interface Config {
 }
 ```
 
-Depends on: [`agentCore`](../packages/examples/agent-spine-demo/src/index.ts) · [`ToolsConfig`](#deepseek-aidsh-tools)
+Depends on: [`agentCore`](../packages/examples/agent-spine-demo/src/index.ts) · [`JsonlCompression`](../packages/session-persistence/session-persistence-jsonl/src/index.ts) · [`ToolsConfig`](#deepseek-aidsh-tools)
 
-Source: [`packages/examples/acp-demo/src/index.ts:33`](../packages/examples/acp-demo/src/index.ts)
+Source: [`packages/examples/acp-demo/src/index.ts:36`](../packages/examples/acp-demo/src/index.ts)
 
 ## `@deepseek-ai/dsh-agent-loop`
 
@@ -224,6 +226,8 @@ export interface Config {
   dshHome?: string
   /** Directory the JSONL session backend writes under. Defaults to `./.sessions`. */
   persistenceRoot?: string
+  /** JSONL artifact encoding; defaults to checksummed Zstandard frames. */
+  persistenceCompression?: JsonlCompression
   /** Skill registry, local-provider, and model-facing consumer config. */
   skills?: agentCore.SkillConfig
   /** Model-facing bash tool config forwarded through agent-spine-demo. */
@@ -235,9 +239,9 @@ export interface Config {
 }
 ```
 
-Depends on: [`agentCore`](../packages/examples/agent-spine-demo/src/index.ts) · [`ToolsConfig`](#deepseek-aidsh-tools)
+Depends on: [`agentCore`](../packages/examples/agent-spine-demo/src/index.ts) · [`JsonlCompression`](../packages/session-persistence/session-persistence-jsonl/src/index.ts) · [`ToolsConfig`](#deepseek-aidsh-tools)
 
-Source: [`packages/examples/cli-demo/src/index.ts:22`](../packages/examples/cli-demo/src/index.ts)
+Source: [`packages/examples/cli-demo/src/index.ts:25`](../packages/examples/cli-demo/src/index.ts)
 
 ## `@deepseek-ai/dsh-code-runtime-worker`
 
@@ -738,10 +742,15 @@ export interface Config {
    * (bash calls, subprocesses). Sessions group under per-cwd subdirectories.
    */
   root: string
+  /** Physical encoding; defaults to checksummed Zstandard frames. */
+  compression?: JsonlCompression
 }
+
+/** Physical encoding selected for JSONL session artifacts. */
+export type JsonlCompression = 'zstd' | 'none'
 ```
 
-Source: [`packages/session-persistence/session-persistence-jsonl/src/index.ts:24`](../packages/session-persistence/session-persistence-jsonl/src/index.ts)
+Source: [`packages/session-persistence/session-persistence-jsonl/src/index.ts:36`](../packages/session-persistence/session-persistence-jsonl/src/index.ts)
 
 ## `@deepseek-ai/dsh-session-persistence-sqlite`
 
@@ -1263,6 +1272,8 @@ export interface Config {
   dshHome?: string
   /** Directory the JSONL session backend writes under. Defaults to `./.sessions`. */
   persistenceRoot?: string
+  /** JSONL artifact encoding; defaults to checksummed Zstandard frames. */
+  persistenceCompression?: JsonlCompression
   /** TUI subtitle rendered on start. Defaults to `ready.`. */
   welcome?: string
   /** Full-screen TUI presentation settings. */
@@ -1280,9 +1291,9 @@ export interface Config {
 }
 ```
 
-Depends on: [`agentCore`](../packages/examples/agent-spine-demo/src/index.ts) · [`ToolsConfig`](#deepseek-aidsh-tools) · [`uiTui`](../packages/ui/tui/src/index.ts)
+Depends on: [`agentCore`](../packages/examples/agent-spine-demo/src/index.ts) · [`JsonlCompression`](../packages/session-persistence/session-persistence-jsonl/src/index.ts) · [`ToolsConfig`](#deepseek-aidsh-tools) · [`uiTui`](../packages/ui/tui/src/index.ts)
 
-Source: [`packages/examples/tui-demo/src/index.ts:28`](../packages/examples/tui-demo/src/index.ts)
+Source: [`packages/examples/tui-demo/src/index.ts:31`](../packages/examples/tui-demo/src/index.ts)
 
 ## `@deepseek-ai/dsh-user-approval`
 

docs/core-data-structures/persistence.md

@@ -102,7 +102,7 @@ Replay/fork is therefore `ctx.sessions.create(id, { seed: seedEvents })`; resumi
 
 Both implement the same abstract `SessionPersistence` (locate/create/append/load/list over `SessionEvent`) and pass `runPersistenceContract`, proving the seam is genuinely backend-agnostic:
 
-- **[dsh-session-persistence-jsonl](../../packages/session-persistence/session-persistence-jsonl)** — an append-only JSONL log per session with crash-safe atomic writes, the interrupted-turn crash recovery above, and a read/replay path.
+- **[dsh-session-persistence-jsonl](../../packages/session-persistence/session-persistence-jsonl)** — an append-only logical JSONL log per session, stored as checksummed concatenated Zstandard frames by default or raw lines by configuration, with crash-safe atomic writes, interrupted-turn recovery, and a read/replay path.
 - **[dsh-session-persistence-sqlite](../../packages/session-persistence/session-persistence-sqlite)** — `node:sqlite`, one row per `SessionEvent`. The row shape `(session_id, seq, type, time, data, source_event_seqs, surface_op)` maps 1:1 onto the event, including optional surface metadata, so there is no parallel persisted schema to keep in sync.
 
 Multiple backends sharing one on-disk session coordinate writes through the [shared persistence write-coordinator](../../.agents/notes/implemented/architecture/2026-06-18-shared-persistence-write-coordinator.md).

docs/i18n/terminology.md

@@ -65,6 +65,7 @@
 | waterfall | waterfall | waterfall（瀑布式事件） | | |
 | wheel | wheel 包 | | | Python 打包格式 |
 | worktree | worktree | | | git 工作区概念 |
+| Zstandard | Zstandard | | | RFC 8878 compression format; `zstd` remains a code value. |
 
 ## 双语类（中英文文本各自使用中英文）
 

examples/acp-agent/advanced.cordis.snapshot.yml

@@ -13,6 +13,7 @@
           provider: deepseek
           model: deepseek-v4-flash
           persistenceRoot: !!js process.env.DSH_SNAPSHOT_SESSIONS_ROOT ?? './.sessions'
+          persistenceCompression: 'none'
           workspaceContext:
             maxBytes: 65536
           tools:

examples/acp-agent/advanced.cordis.yml

@@ -11,6 +11,7 @@
           provider: deepseek
           model: deepseek-v4-flash
           persistenceRoot: !!js process.env.DSH_SNAPSHOT_SESSIONS_ROOT ?? './.sessions'
+          persistenceCompression: !!js "process.env.DSH_SNAPSHOT === undefined ? 'zstd' : 'none'"
           workspaceContext:
             maxBytes: 65536
           tools:

examples/acp-agent/both-mode.cordis.snapshot.yml

@@ -15,6 +15,7 @@
           provider: deepseek
           model: deepseek-v4-flash
           persistenceRoot: !!js process.env.DSH_SNAPSHOT_SESSIONS_ROOT ?? './.sessions'
+          persistenceCompression: 'none'
           workspaceContext:
             maxBytes: 65536
           tools:

examples/acp-agent/both-mode.cordis.yml

@@ -13,6 +13,7 @@
           provider: deepseek
           model: deepseek-v4-flash
           persistenceRoot: !!js process.env.DSH_SNAPSHOT_SESSIONS_ROOT ?? './.sessions'
+          persistenceCompression: !!js "process.env.DSH_SNAPSHOT === undefined ? 'zstd' : 'none'"
           workspaceContext:
             maxBytes: 65536
           tools:

examples/acp-agent/code-mode-workspace-context.cordis.snapshot.yml

@@ -14,6 +14,7 @@
           provider: deepseek
           model: deepseek-v4-flash
           persistenceRoot: !!js process.env.DSH_SNAPSHOT_SESSIONS_ROOT ?? './.sessions'
+          persistenceCompression: 'none'
           workspaceContext:
             maxBytes: 65536
           tools:

examples/acp-agent/code-mode-workspace-context.cordis.yml

@@ -11,6 +11,7 @@
           provider: deepseek
           model: deepseek-v4-flash
           persistenceRoot: !!js process.env.DSH_SNAPSHOT_SESSIONS_ROOT ?? './.sessions'
+          persistenceCompression: !!js "process.env.DSH_SNAPSHOT === undefined ? 'zstd' : 'none'"
           workspaceContext:
             maxBytes: 65536
           tools:

examples/acp-agent/code-mode.cordis.snapshot.yml

@@ -15,6 +15,7 @@
           provider: deepseek
           model: deepseek-v4-flash
           persistenceRoot: !!js process.env.DSH_SNAPSHOT_SESSIONS_ROOT ?? './.sessions'
+          persistenceCompression: 'none'
           workspaceContext:
             maxBytes: 65536
           tools:

examples/acp-agent/code-mode.cordis.yml

@@ -14,6 +14,7 @@
           provider: deepseek
           model: deepseek-v4-flash
           persistenceRoot: !!js process.env.DSH_SNAPSHOT_SESSIONS_ROOT ?? './.sessions'
+          persistenceCompression: !!js "process.env.DSH_SNAPSHOT === undefined ? 'zstd' : 'none'"
           workspaceContext:
             maxBytes: 65536
           tools:

examples/acp-agent/cordis.yml

@@ -41,12 +41,14 @@
 # The ACP server app: the agent-spine-demo spine + JSONL persistence + the ACP bridge.
 # Persistence root: $DSH_SNAPSHOT_SESSIONS_ROOT when the snapshot harness sets it
 # (so it can harvest / isolate the log), else ./.sessions for the demo.
+# Snapshot modes use raw JSONL fixtures; ordinary runs keep the compressed default.
 - id: acp-agent
   name: '@deepseek-ai/dsh-acp-demo'
   config:
     provider: deepseek
     model: deepseek-v4-flash
     persistenceRoot: !!js process.env.DSH_SNAPSHOT_SESSIONS_ROOT ?? './.sessions'
+    persistenceCompression: !!js "process.env.DSH_SNAPSHOT === undefined ? 'zstd' : 'none'"
     workspaceContext:
       maxBytes: 65536
     # Keep the persona to identity and behavior; tool plugins own tool guidance.

examples/acp-agent/workspace-context.cordis.snapshot.yml

@@ -15,6 +15,7 @@
           provider: deepseek
           model: deepseek-v4-flash
           persistenceRoot: !!js process.env.DSH_SNAPSHOT_SESSIONS_ROOT ?? './.sessions'
+          persistenceCompression: 'none'
           workspaceContext:
             maxBytes: 65536
             dshHome: !!js process.cwd() + '/.dsh'

examples/acp-agent/workspace-context.cordis.yml

@@ -12,6 +12,7 @@
           provider: deepseek
           model: deepseek-v4-flash
           persistenceRoot: !!js process.env.DSH_SNAPSHOT_SESSIONS_ROOT ?? './.sessions'
+          persistenceCompression: !!js "process.env.DSH_SNAPSHOT === undefined ? 'zstd' : 'none'"
           workspaceContext:
             maxBytes: 65536
             dshHome: !!js process.cwd() + '/.dsh'

examples/headless-agent/advanced.cordis.yml

@@ -10,6 +10,7 @@
           provider: deepseek
           model: deepseek-v4-flash
           persistenceRoot: './.sessions'
+          persistenceCompression: !!js "process.env.DSH_SNAPSHOT === undefined ? 'zstd' : 'none'"
           workspaceContext:
             maxBytes: 65536
           tools:

examples/headless-agent/cordis.yml

@@ -25,6 +25,7 @@
     provider: deepseek
     model: deepseek-v4-flash
     persistenceRoot: './.sessions'
+    persistenceCompression: !!js "process.env.DSH_SNAPSHOT === undefined ? 'zstd' : 'none'"
     workspaceContext:
       maxBytes: 65536
     persona: |

examples/headless-agent/tests/fixtures/time-context.cordis.yml

@@ -15,4 +15,5 @@
     model: time-context-mock
     persona: 'Test the time-context plugin.'
     persistenceRoot: './.sessions'
+    persistenceCompression: 'none'
     workspaceContext: false

examples/headless-agent/tests/keyless-smoke.e2e.ts

@@ -1,4 +1,7 @@
-import { readdir } from 'node:fs/promises'
+import { readFile, readdir } from 'node:fs/promises'
+import { zstdDecompress } from 'node:zlib'
+import { promisify } from 'node:util'
+import { join } from 'node:path'
 import { fileURLToPath } from 'node:url'
 import { describe, expect, it } from 'vitest'
 import { LOADER_SMOKE_TEST_TIMEOUT_MS, runLoaderSmoke } from '@deepseek-ai/dsh-loader-smoke'
@@ -7,10 +10,11 @@ import type { SessionEvent } from '@deepseek-ai/dsh-session'
 const binScript = fileURLToPath(new URL('../../../packages/examples/cli-demo/src/bin.ts', import.meta.url))
 const configPath = fileURLToPath(new URL('./fixtures/cli.cordis.yml', import.meta.url))
 const tsconfigPath = fileURLToPath(new URL('../../../tsconfig.json', import.meta.url))
+const decompress = promisify(zstdDecompress)
 
 describe('headless-agent keyless smoke', () => {
   it('boots the real Loader tree, runs a real bash tool round trip, and persists the turn', async () => {
-    let persisted = false
+    let persistedHeader: Record<string, unknown> | undefined
     const { stdout, stderr } = await runLoaderSmoke({
       label: 'headless-agent',
       tempDirPrefix: 'headless-agent-smoke-',
@@ -20,7 +24,11 @@ describe('headless-agent keyless smoke', () => {
       tsconfigPath,
       inspect: async (cwd) => {
         const files = await readdir(cwd, { recursive: true })
-        persisted = files.some(file => file.endsWith('.jsonl'))
+        const relativePath = files.find(file => file.endsWith('.jsonl.zstd'))
+        if (relativePath === undefined) return
+        const compressed = await readFile(join(cwd, relativePath))
+        expect(compressed.subarray(0, 4).toString('hex')).toBe('28b52ffd')
+        persistedHeader = JSON.parse((await decompress(compressed)).toString()) as Record<string, unknown>
       },
     })
     const lines = stdout.trimEnd().split('\n').map(line => JSON.parse(line) as Record<string, unknown>)
@@ -38,6 +46,6 @@ describe('headless-agent keyless smoke', () => {
       usage: { inputTokens: 18, outputTokens: 8, cacheReadTokens: 2, reasoningTokens: 1 },
     })
     expect(String(result?.['result'])).toContain('CLI_TOOL_ROUND_TRIP')
-    expect(persisted).toBe(true)
+    expect(persistedHeader).toMatchObject({ type: 'session' })
   }, LOADER_SMOKE_TEST_TIMEOUT_MS)
 })

examples/jsonrpc-agent/tests/keyless-smoke.e2e.ts

@@ -1,14 +1,17 @@
 import { spawn } from 'node:child_process'
 import { createServer } from 'node:http'
-import { mkdtemp, rm } from 'node:fs/promises'
+import { mkdtemp, readFile, readdir, rm } from 'node:fs/promises'
 import { tmpdir } from 'node:os'
 import { join } from 'node:path'
 import { fileURLToPath } from 'node:url'
+import { promisify } from 'node:util'
+import { zstdDecompress } from 'node:zlib'
 import { describe, expect, it } from 'vitest'
 
 const binScript = fileURLToPath(new URL('../../../packages/examples/jsonrpc-demo/src/bin.ts', import.meta.url))
 const configPath = fileURLToPath(new URL('../cordis.yml', import.meta.url))
 const repoRoot = fileURLToPath(new URL('../../..', import.meta.url))
+const decompress = promisify(zstdDecompress)
 
 function waitForLine(
   lines: string[],
@@ -152,6 +155,13 @@ describe('jsonrpc-agent keyless smoke', () => {
       } else {
         expect(child.exitCode, stderr).toBe(0)
       }
+      const sessionsRoot = join(root, '.sessions')
+      const files = await readdir(sessionsRoot, { recursive: true })
+      const log = files.find(file => file.endsWith('.jsonl.zstd'))
+      expect(log).toBeDefined()
+      const compressed = await readFile(join(sessionsRoot, log!))
+      expect(compressed.subarray(0, 4).toString('hex')).toBe('28b52ffd')
+      expect(JSON.parse((await decompress(compressed)).toString())).toMatchObject({ type: 'session', id: 'main' })
     } finally {
       if (child.exitCode === null) child.kill('SIGKILL')
       await new Promise<void>(resolve => modelServer.close(() => { resolve() }))

packages/bash/tool-bash/tests/integration.spec.ts

@@ -23,7 +23,9 @@ import { MockAdapter, textResponse, toolCallResponse } from '../../../core/agent
 async function harness(adapter: MockAdapter, sessionRoot?: string, dshHome?: string) {
   const ctx = new Context()
   await mountAgentLoopTestDependencies(ctx)
-  if (sessionRoot !== undefined) await ctx.plugin(SessionPersistenceJsonl, { root: sessionRoot })
+  if (sessionRoot !== undefined) {
+    await ctx.plugin(SessionPersistenceJsonl, { root: sessionRoot, compression: 'none' })
+  }
   await ctx.plugin(AgentLoop, { agents: [] })
   await ctx.plugin(TaskService)
   await ctx.plugin(ToolTasks)

packages/examples/acp-demo/README.md

@@ -37,6 +37,7 @@ Because the package wires no logger entry, an ACP leaf has **nothing to get wron
 | `toolBash` | owner defaults | model-facing bash config routed through `dsh-agent-spine-demo`, including bash's producer-local `enableRunInBackground` |
 | `toolTasks` | owner defaults | generic `task_output` wait bounds routed through `dsh-agent-spine-demo` |
 | `persistenceRoot` | `./.sessions` | the JSONL backend's root directory |
+| `persistenceCompression` | `'zstd'` | JSONL artifact encoding (`'zstd'` or raw `'none'`) |
 
 The leaf supplies the swappable backends: an LLM adapter (`llm-deepseek` for the real model, `llm-replay` for keyless snapshot replay), a bash executor, and optionally a `ctx.fs` provider. Workspace context becomes a no-op without `ctx.fs`; the shipped [`examples/acp-agent/cordis.yml`](../../../examples/acp-agent/cordis.yml) selects `dsh-sandbox-policy`, `dsh-fs-sandbox`, `dsh-fs-policy`, and `dsh-tool-fs` so baseline instructions and model-facing `read`/`write`/`edit` share one provider, sandbox mode, workspace root, and observed-version policy.
 

packages/examples/acp-demo/src/index.ts

@@ -15,7 +15,10 @@ import * as acp from '@deepseek-ai/dsh-acp'
 import * as agentCore from '@deepseek-ai/dsh-agent-spine-demo'
 import * as workspaceContext from '@deepseek-ai/dsh-workspace-context'
 import ToolRegistry, { type Config as ToolsConfig } from '@deepseek-ai/dsh-tools'
-import SessionPersistenceJsonl from '@deepseek-ai/dsh-session-persistence-jsonl'
+import SessionPersistenceJsonl, {
+  JsonlCompressionSchema,
+  type JsonlCompression,
+} from '@deepseek-ai/dsh-session-persistence-jsonl'
 import UserInteractionService from '@deepseek-ai/dsh-user-interaction'
 
 export const name = 'acp-demo'
@@ -47,6 +50,8 @@ export interface Config {
   dshHome?: string
   /** Directory the JSONL session backend writes under. Defaults to `./.sessions`. */
   persistenceRoot?: string
+  /** JSONL artifact encoding; defaults to checksummed Zstandard frames. */
+  persistenceCompression?: JsonlCompression
   /** Controls automatic AGENTS.md/CLAUDE.md loading; configure a byte budget or set `false`. */
   workspaceContext: agentCore.Config['workspaceContext']
   /** Skill registry, local-provider, and model-facing consumer config forwarded to agent-spine-demo. */
@@ -72,6 +77,7 @@ export const Config: z<Config> = z.object({
   tools: ToolRegistry.Config,
   dshHome: z.string(),
   persistenceRoot: z.string().default(DEFAULT_PERSISTENCE_ROOT),
+  persistenceCompression: JsonlCompressionSchema,
   workspaceContext: z.union([z.const(false), workspaceContext.Config]).required(),
   skills: agentCore.SkillConfigSchema,
   toolBash: agentCore.ToolBashConfigSchema,
@@ -89,6 +95,9 @@ export const Config: z<Config> = z.object({
 export function apply(ctx: Context, config: Config): void {
   ctx.plugin(agentCore, agentCore.pickSpineConfig(config))
   ctx.plugin(UserInteractionService)
-  ctx.plugin(SessionPersistenceJsonl, { root: config.persistenceRoot ?? DEFAULT_PERSISTENCE_ROOT })
+  ctx.plugin(SessionPersistenceJsonl, {
+    root: config.persistenceRoot ?? DEFAULT_PERSISTENCE_ROOT,
+    ...(config.persistenceCompression === undefined ? {} : { compression: config.persistenceCompression }),
+  })
   ctx.plugin(acp, { provider: config.provider, model: config.model })
 }

packages/examples/acp-demo/tests/acp-agent.spec.ts

@@ -70,10 +70,19 @@ async function withIsolatedSkillHomes<T>(run: () => Promise<T>): Promise<T> {
 
 describe('dsh-acp-demo composition', () => {
   it('brings up the spine + persistence + the ACP bridge', async () => {
-    const ctx = await mount({ provider: 'mock', model: 'mock', persona: 'hi', persistenceRoot: '/tmp/dsh-acp-demo-test', skills: await isolatedSkillsConfig(), workspaceContext: false })
+    const ctx = await mount({
+      provider: 'mock',
+      model: 'mock',
+      persona: 'hi',
+      persistenceRoot: '/tmp/dsh-acp-demo-test',
+      persistenceCompression: 'none',
+      skills: await isolatedSkillsConfig(),
+      workspaceContext: false,
+    })
     expect(ctx.get('agents')).toBeDefined()
     expect(ctx.get('sessions')).toBeDefined()
     expect(ctx.get('sessionPersistence')).toBeDefined()
+    expect((ctx.get('sessionPersistence') as unknown as { config: { compression?: string } }).config.compression).toBe('none')
     expect(ctx.get('agentLoop')).toBeDefined()
     expect(ctx.get('userInteraction')).toBeDefined()
     expect(ctx.get('tools')?.get('ask_user_question')).toBeUndefined()

packages/examples/acp-demo/tests/built-bin.e2e.ts

@@ -1,5 +1,5 @@
 import { spawn } from 'node:child_process'
-import { mkdtemp, mkdir, rm, symlink, writeFile, readFile } from 'node:fs/promises'
+import { mkdtemp, mkdir, readdir, rm, symlink, writeFile, readFile } from 'node:fs/promises'
 import { existsSync } from 'node:fs'
 import { tmpdir } from 'node:os'
 import { dirname, join } from 'node:path'
@@ -15,21 +15,24 @@ import {
   type SessionNotification,
 } from '@agentclientprotocol/sdk'
 import { Readable, Writable } from 'node:stream'
+import { promisify } from 'node:util'
+import { zstdDecompress } from 'node:zlib'
 import { afterEach, describe, expect, it } from 'vitest'
 
 /**
  * Published-entry smoke: run `lib/bin.js` under plain Node in a symlinked external consumer and
- * require a valid initialize response. This catches built-only settle races and stdout protocol
- * leaks that the tsx source-path smoke cannot. It skips before build; initialize is keyless, with a
- * dummy key used only to boot the adapter. `--expose-internals` enables Cordis bare-plugin loading.
+ * complete a mock-backed turn. This catches built-only settle races, stdout protocol leaks, and
+ * published persistence behavior that the tsx source-path smoke cannot. It skips before build;
+ * `--expose-internals` enables Cordis bare-plugin loading.
  */
 
 const repoRoot = fileURLToPath(new URL('../../../../', import.meta.url))
 const acpBin = join(repoRoot, 'packages/examples/acp-demo/lib/bin.js')
+const decompress = promisify(zstdDecompress)
 
 const dshPackages = [
   'examples/agent-spine-demo', 'core/agent', 'core/session', 'core/system-prompt',
-  'core/tools', 'core/agent-loop', 'llm/llm', 'llm/llm-deepseek', 'bash/bash',
+  'core/tools', 'core/agent-loop', 'llm/llm', 'bash/bash',
   'bash/bash-local', 'bash/tool-bash', 'context/workspace-context', 'support/invariants', 'ui/app-boot',
   'session-persistence/session-persistence',
   'session-persistence/session-persistence-jsonl', 'ui/acp', 'examples/acp-demo', 'util/paths',
@@ -73,18 +76,31 @@ async function makeConsumer(): Promise<string> {
     const resolved = fileURLToPath(import.meta.resolve(`${dep}/package.json`, fromAcp))
     await link(dirname(resolved), dep, nm)
   }
+  await writeFile(join(dir, 'mock-llm.mjs'), [
+    "import { LlmAdapter } from '@deepseek-ai/dsh-llm'",
+    'class Mock extends LlmAdapter {',
+    '  async * stream() {',
+    "    yield { type: 'block-start', index: 0, blockType: 'text' }",
+    "    yield { type: 'text-delta', index: 0, text: 'ACP BUILT OK' }",
+    "    yield { type: 'block-end', index: 0, block: { type: 'text', text: 'ACP BUILT OK' } }",
+    "    yield { type: 'finish', reason: { kind: 'stop' } }",
+    '  }',
+    '}',
+    "export const name = 'built-acp-mock'",
+    "export const inject = ['llm']",
+    "export function apply(ctx) { ctx.llm.registerAdapter(['built-acp-mock'], new Mock()) }",
+    '',
+  ].join('\n'))
   await writeFile(join(dir, 'cordis.yml'), [
-    '- id: llm-deepseek',
-    '  name: \'@deepseek-ai/dsh-llm-deepseek\'',
-    '  config:',
-    '    apiKey: !!js process.env.DEEPSEEK_API_KEY',
+    '- id: mock-llm',
+    '  name: \'./mock-llm.mjs\'',
     '- id: bash',
     '  name: \'@deepseek-ai/dsh-bash-local\'',
     '- id: acp-agent',
     '  name: \'@deepseek-ai/dsh-acp-demo\'',
     '  config:',
-    '    provider: deepseek',
-    '    model: deepseek-v4-flash',
+    '    provider: built-acp-mock',
+    '    model: built-acp-mock',
     '    persona: \'test agent\'',
     '    workspaceContext: false',
     '',
@@ -113,14 +129,12 @@ afterEach(async () => {
 })
 
 describe.skipIf(!existsSync(acpBin))('dsh-acp-demo BUILT bin (node lib/bin.js, no tsx)', () => {
-  it('boots the published bin and answers an initialize JSON-RPC frame on stdout', async () => {
+  it('boots the published bin, completes a turn, and writes default Zstandard persistence', async () => {
     consumer = await makeConsumer()
     child = spawn(process.execPath, ['--expose-internals', acpBin, '--config', './cordis.yml'], {
       cwd: consumer,
-      // Dummy key: initialize never reaches the model, so it is never used.
       env: {
         ...process.env,
-        DEEPSEEK_API_KEY: process.env.DEEPSEEK_API_KEY ?? 'sk-dummy-for-boot',
         DSH_HOME: join(consumer, '.dsh'),
         DSH_AGENTS_HOME: join(consumer, '.agents'),
       },
@@ -151,6 +165,18 @@ describe.skipIf(!existsSync(acpBin))('dsh-acp-demo BUILT bin (node lib/bin.js, n
     // regression would exit before answering); loadSession proves the real app
     // mounted, not a collapsed export shape.
     expect(init.agentCapabilities?.loadSession).toBe(true)
+    const { sessionId } = await client.newSession({ cwd: consumer, mcpServers: [] })
+    const result = await client.prompt({ sessionId, prompt: [{ type: 'text', text: 'reply' }] })
+    expect(result.stopReason).toBe('end_turn')
+    const sessionsRoot = join(consumer, '.sessions')
+    let log: string | undefined
+    await expect.poll(async () => {
+      log = (await readdir(sessionsRoot, { recursive: true })).find(file => file.endsWith('.jsonl.zstd'))
+      return log
+    }).toBeTypeOf('string')
+    const compressed = await readFile(join(sessionsRoot, log!))
+    expect(compressed.subarray(0, 4).toString('hex')).toBe('28b52ffd')
+    expect(JSON.parse((await decompress(compressed)).toString())).toMatchObject({ type: 'session', id: sessionId })
     expect(stderr.join('')).not.toContain('without inject')
     // stdout purity: every emitted line is a JSON-RPC frame, no logger leak.
     for (const line of rawOut.join('').split('\n').filter(l => l.trim().length > 0)) {
@@ -182,7 +208,6 @@ function runBinExpectingExit(configArg: string, cwd: string = tmpdir()): Promise
       cwd,
       env: {
         ...process.env,
-        DEEPSEEK_API_KEY: process.env.DEEPSEEK_API_KEY ?? 'sk-dummy-for-boot',
         DSH_HOME: join(cwd, '.dsh'),
         DSH_AGENTS_HOME: join(cwd, '.agents'),
       },

packages/examples/cli-demo/README.md

@@ -19,6 +19,7 @@ The package mounts no console logger, interactive UI, user-interaction service,
 | `toolBash` | owner defaults | model-facing bash config, including this producer's background opt-in |
 | `toolTasks` | owner defaults | generic `task_output` wait bounds |
 | `persistenceRoot` | `./.sessions` | JSONL session root |
+| `persistenceCompression` | `'zstd'` | JSONL artifact encoding (`'zstd'` or raw `'none'`) |
 | `workspaceContext` | required | workspace-instruction byte budget, or `false` to disable loading |
 
 ## CLI contract

packages/examples/cli-demo/src/index.ts

@@ -11,7 +11,10 @@ import z from 'schemastery'
 import { SessionId } from '@deepseek-ai/dsh-session'
 import ToolRegistry, { type Config as ToolsConfig } from '@deepseek-ai/dsh-tools'
 import * as agentCore from '@deepseek-ai/dsh-agent-spine-demo'
-import SessionPersistenceJsonl from '@deepseek-ai/dsh-session-persistence-jsonl'
+import SessionPersistenceJsonl, {
+  JsonlCompressionSchema,
+  type JsonlCompression,
+} from '@deepseek-ai/dsh-session-persistence-jsonl'
 import * as workspaceContext from '@deepseek-ai/dsh-workspace-context'
 
 const DEFAULT_PERSISTENCE_ROOT = './.sessions'
@@ -36,6 +39,8 @@ export interface Config {
   dshHome?: string
   /** Directory the JSONL session backend writes under. Defaults to `./.sessions`. */
   persistenceRoot?: string
+  /** JSONL artifact encoding; defaults to checksummed Zstandard frames. */
+  persistenceCompression?: JsonlCompression
   /** Skill registry, local-provider, and model-facing consumer config. */
   skills?: agentCore.SkillConfig
   /** Model-facing bash tool config forwarded through agent-spine-demo. */
@@ -54,6 +59,7 @@ export const Config: z<Config> = z.object({
   model: z.string().required(),
   maxParallelToolCalls: z.number().step(1).min(1),
   persistenceRoot: z.string().default(DEFAULT_PERSISTENCE_ROOT),
+  persistenceCompression: JsonlCompressionSchema,
   persona: z.string(),
   dshHome: z.string(),
   skills: agentCore.SkillConfigSchema,
@@ -78,5 +84,8 @@ export function apply(ctx: Context, config: Config): void {
     ...agentCore.pickSpineConfig(config),
     agents: [{ id: SessionId('main'), provider: config.provider, model: config.model, cwd: process.cwd() }],
   })
-  ctx.plugin(SessionPersistenceJsonl, { root: config.persistenceRoot ?? DEFAULT_PERSISTENCE_ROOT })
+  ctx.plugin(SessionPersistenceJsonl, {
+    root: config.persistenceRoot ?? DEFAULT_PERSISTENCE_ROOT,
+    ...(config.persistenceCompression === undefined ? {} : { compression: config.persistenceCompression }),
+  })
 }

packages/examples/cli-demo/tests/built-bin.e2e.ts

@@ -3,11 +3,14 @@ import { existsSync } from 'node:fs'
 import { mkdtemp, mkdir, readFile, readdir, rm, symlink, writeFile } from 'node:fs/promises'
 import { tmpdir } from 'node:os'
 import { dirname, join } from 'node:path'
+import { promisify } from 'node:util'
 import { fileURLToPath } from 'node:url'
+import { zstdDecompress } from 'node:zlib'
 import { afterEach, describe, expect, it } from 'vitest'
 
 const repoRoot = fileURLToPath(new URL('../../../../', import.meta.url))
 const cliBin = join(repoRoot, 'packages/examples/cli-demo/lib/bin.js')
+const decompress = promisify(zstdDecompress)
 const dshPackages = [
   'examples/agent-spine-demo', 'examples/cli-demo', 'core/agent', 'core/session',
   'core/system-prompt', 'core/tools', 'core/agent-loop', 'llm/llm', 'bash/bash',
@@ -140,8 +143,13 @@ describe.skipIf(!existsSync(cliBin))('dsh-cli-demo BUILT bin', () => {
     const lines = stream.stdout.trimEnd().split('\n').map(line => JSON.parse(line) as Record<string, unknown>)
     expect(lines[0]).toMatchObject({ type: 'session_event', event: { type: 'turn/start' } })
     expect(lines.at(-1)).toMatchObject({ type: 'result', success: true, result: 'BUILT: stream task' })
-    const files = await readdir(join(consumer, '.sessions'), { recursive: true })
-    expect(files.filter(file => file.endsWith('.jsonl'))).toHaveLength(3)
+    const sessionsRoot = join(consumer, '.sessions')
+    const files = await readdir(sessionsRoot, { recursive: true })
+    const logs = files.filter(file => file.endsWith('.jsonl.zstd'))
+    expect(logs).toHaveLength(3)
+    const compressed = await readFile(join(sessionsRoot, logs[0]!))
+    expect(compressed.subarray(0, 4).toString('hex')).toBe('28b52ffd')
+    expect(JSON.parse((await decompress(compressed)).toString())).toMatchObject({ type: 'session' })
   }, 30_000)
 
   it('keeps stdout empty for invalid argv and missing config', async () => {

packages/examples/cli-demo/tests/cli-demo.spec.ts

@@ -51,12 +51,14 @@ describe('dsh-cli-demo app composition', () => {
       persona: 'Headless.',
       tools: { mode: 'native' },
       persistenceRoot: root,
+      persistenceCompression: 'none',
       skills: await skillConfig(),
       workspaceContext: false,
     })
     const [agent] = ctx.get('agents')?.roots() ?? []
     expect(ctx.get('agentLoop')).toBeDefined()
     expect(ctx.get('sessionPersistence')).toBeDefined()
+    expect((ctx.get('sessionPersistence') as unknown as { config: { compression?: string } }).config.compression).toBe('none')
     expect(agent?.session.header.cwd).toBe(process.cwd())
     expect(ctx.get('userInteraction')).toBeUndefined()
     expect(ctx.get('tools')?.get('ask_user_question')).toBeUndefined()

packages/examples/cli-demo/tests/cli.spec.ts

@@ -303,7 +303,7 @@ describe('runOneShot and executeCli', () => {
     expect(output).toEqual({ code: 0, stdout: 'final answer\n', stderr: '' })
     expect(agent.status).toBe('disposed')
     const files = await readdir(persistenceRoot, { recursive: true })
-    expect(files.some(file => file.endsWith('.jsonl'))).toBe(true)
+    expect(files.some(file => file.endsWith('.jsonl.zstd'))).toBe(true)
   })
 
   it('sums usage across tool steps and selects the last text-bearing assistant message', async () => {

packages/examples/tui-demo/README.md

@@ -32,6 +32,7 @@ Swappable LLM, bash, filesystem, and other capability providers remain in the le
 | `toolTasks` | owner defaults | Background-task control-tool config, or `false` |
 | `workspaceContext` | required | Workspace-instruction config, or `false` |
 | `persistenceRoot` | `./.sessions` | JSONL persistence root |
+| `persistenceCompression` | `'zstd'` | JSONL artifact encoding (`'zstd'` or raw `'none'`) |
 | `welcome` | `ready.` | TUI subtitle |
 | `ui` | owner defaults | TUI presentation settings such as reasoning, color, and card height |
 | `resumeSessionId` | — | Exact persisted session to resume |

packages/examples/tui-demo/src/index.ts

@@ -15,7 +15,10 @@ import { SessionId } from '@deepseek-ai/dsh-session'
 import ToolRegistry, { type Config as ToolsConfig } from '@deepseek-ai/dsh-tools'
 import * as agentCore from '@deepseek-ai/dsh-agent-spine-demo'
 import * as workspaceContext from '@deepseek-ai/dsh-workspace-context'
-import SessionPersistenceJsonl from '@deepseek-ai/dsh-session-persistence-jsonl'
+import SessionPersistenceJsonl, {
+  JsonlCompressionSchema,
+  type JsonlCompression,
+} from '@deepseek-ai/dsh-session-persistence-jsonl'
 import UserInteractionService from '@deepseek-ai/dsh-user-interaction'
 import * as toolAskUser from '@deepseek-ai/dsh-tool-ask-user'
 import * as uiTui from '@deepseek-ai/dsh-tui'
@@ -42,6 +45,8 @@ export interface Config {
   dshHome?: string
   /** Directory the JSONL session backend writes under. Defaults to `./.sessions`. */
   persistenceRoot?: string
+  /** JSONL artifact encoding; defaults to checksummed Zstandard frames. */
+  persistenceCompression?: JsonlCompression
   /** TUI subtitle rendered on start. Defaults to `ready.`. */
   welcome?: string
   /** Full-screen TUI presentation settings. */
@@ -71,6 +76,7 @@ export const Config: z<Config> = z.object({
   tools: ToolRegistry.Config,
   dshHome: z.string(),
   persistenceRoot: z.string().default(DEFAULT_PERSISTENCE_ROOT),
+  persistenceCompression: JsonlCompressionSchema,
   welcome: z.string().default(DEFAULT_WELCOME),
   ui: uiTui.TuiConfigSchema,
   skills: agentCore.SkillConfigSchema,
@@ -91,7 +97,10 @@ export const Config: z<Config> = z.object({
 export function composeTuiApp(ctx: Context, config: Config): void {
   const resumeSessionId = config.resumeSessionId === '' ? undefined : config.resumeSessionId
   const sessionId = SessionId(resumeSessionId ?? `main-session-${randomUUID()}`)
-  ctx.plugin(SessionPersistenceJsonl, { root: config.persistenceRoot ?? DEFAULT_PERSISTENCE_ROOT })
+  ctx.plugin(SessionPersistenceJsonl, {
+    root: config.persistenceRoot ?? DEFAULT_PERSISTENCE_ROOT,
+    ...(config.persistenceCompression === undefined ? {} : { compression: config.persistenceCompression }),
+  })
   ctx.plugin(UserInteractionService)
   ctx.plugin(uiTui, {
     ...config.ui,

packages/examples/tui-demo/tests/tui-agent.spec.ts

@@ -31,6 +31,7 @@ describe('dsh-tui-demo app', () => {
       tools: { mode: 'code' },
       dshHome: '/tmp/dsh-home',
       persistenceRoot: '/tmp/tui-sessions',
+      persistenceCompression: 'none',
       welcome: 'TUI ready',
       ui: { color: false, maxToolOutputLines: 3 },
       skills: { tool: { catalogDescriptionMaxLength: 8 } },
@@ -46,7 +47,7 @@ describe('dsh-tui-demo app', () => {
       'agent-spine-demo',
       'tool-ask-user',
     ])
-    expect(calls[0]?.config).toEqual({ root: '/tmp/tui-sessions' })
+    expect(calls[0]?.config).toEqual({ root: '/tmp/tui-sessions', compression: 'none' })
     const tuiConfig = calls[2]?.config as { sessionId: string }
     expect(tuiConfig).toMatchObject({ welcome: 'TUI ready', color: false, maxToolOutputLines: 3 })
     expect(tuiConfig.sessionId).toMatch(/^main-session-[0-9a-f-]{36}$/)

packages/session-persistence/session-persistence-jsonl/README.md

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

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

@@ -12,8 +12,20 @@ import { createHash } from 'node:crypto'
 import { join } from 'node:path'
 import type { SessionEvent, SessionHeader, SessionId } from '@deepseek-ai/dsh-session'
 
+/** Physical encoding selected for JSONL session artifacts. */
+export type JsonlCompression = 'zstd' | 'none'
+
+/**
+ * Return the artifact suffix for one physical encoding.
+ * @param compression - configured JSONL artifact encoding.
+ * @returns `.jsonl.zstd` for Zstandard or `.jsonl` for plaintext.
+ */
+export function logSuffix(compression: JsonlCompression): '.jsonl.zstd' | '.jsonl' {
+  return compression === 'zstd' ? '.jsonl.zstd' : '.jsonl'
+}
+
 /**
- * The first line of a session's `.jsonl` file: the immutable
+ * The first JSONL record of a session artifact: the immutable
  * {@link SessionHeader} tagged as a `session` record so a reader can tell it
  * apart from an event line.
  */
@@ -126,10 +138,16 @@ export function sessionDir(root: string, cwd: string | undefined): string {
  * @param root - the backend's session root directory.
  * @param cwd - the session's project directory (picks the per-cwd bucket; `undefined` → `_no-cwd`).
  * @param id - the session id, path-encoded via {@link encodeSegment} before filesystem use.
- * @returns the session's `.jsonl` log file path.
+ * @param compression - physical artifact encoding and filename suffix.
+ * @returns the session's configured JSONL artifact path.
  */
-export function logPath(root: string, cwd: string | undefined, id: SessionId): string {
-  return join(sessionDir(root, cwd), `${encodeSegment(id)}.jsonl`)
+export function logPath(
+  root: string,
+  cwd: string | undefined,
+  id: SessionId,
+  compression: JsonlCompression,
+): string {
+  return join(sessionDir(root, cwd), `${encodeSegment(id)}${logSuffix(compression)}`)
 }
 
 /**

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

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

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

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

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

@@ -40,6 +40,10 @@ async function freshRoot(): Promise<string> {
   return dir
 }
 
+function rawLogPath(root: string, cwd: string | undefined, id: SessionId): string {
+  return logPath(root, cwd, id, 'none')
+}
+
 afterEach(async () => {
   vi.restoreAllMocks()
   for (const d of dirs.splice(0)) await rm(d, { recursive: true, force: true })
@@ -70,11 +74,11 @@ function appendClosedTurn(session: Session): void {
 }
 
 // Run the shared backend contract against the real JSONL backend.
-runPersistenceContract('jsonl', async () => {
+runPersistenceContract('jsonl-none', async () => {
   const dir = await mkdtemp(join(tmpdir(), 'dsh-jsonl-'))
   const ctx = new Context()
   await ctx.plugin(SessionStore)
-  const fiber = await ctx.plugin(SessionPersistenceJsonl, { root: dir })
+  const fiber = await ctx.plugin(SessionPersistenceJsonl, { root: dir, compression: 'none' })
   return {
     persistence: ctx.sessionPersistence,
     dispose: async () => {
@@ -86,18 +90,18 @@ runPersistenceContract('jsonl', async () => {
 
 // Two mounts share this temp root to exercise reload. `corruptTail` appends a partial,
 // newline-less fragment past the committed region so coordinator repair runs on real file bytes.
-runCoordinatorContract('jsonl', async (): Promise<CoordinatorFixture> => {
+runCoordinatorContract('jsonl-none', async (): Promise<CoordinatorFixture> => {
   const dir = await mkdtemp(join(tmpdir(), 'dsh-jsonl-coord-'))
   return {
     mount: async (ctx) => {
-      const fiber = await ctx.plugin(SessionPersistenceJsonl, { root: dir })
+      const fiber = await ctx.plugin(SessionPersistenceJsonl, { root: dir, compression: 'none' })
       return fiber
     },
     corruptTail: async (id, cwd) => {
       // A half-written record with no trailing newline: scanLog treats it as an
       // uncommitted crash fragment and reports committedBytes < byteLength, so
       // the coordinator sees a tornMarker to truncate.
-      await appendFile(logPath(dir, cwd, id), '{"type":"assistant/chunk","seq":8,"ti')
+      await appendFile(rawLogPath(dir, cwd, id), '{"type":"assistant/chunk","seq":8,"ti')
     },
     cleanup: async () => { await rm(dir, { recursive: true, force: true }) },
   }
@@ -134,11 +138,14 @@ describe('SessionPersistenceJsonl: format helpers', () => {
     const absoluteRoot = await freshRoot()
     const ctx = new Context()
     await ctx.plugin(SessionStore)
-    const fiber = await ctx.plugin(SessionPersistenceJsonl, { root: relative(process.cwd(), absoluteRoot) })
+    const fiber = await ctx.plugin(SessionPersistenceJsonl, {
+      root: relative(process.cwd(), absoluteRoot),
+      compression: 'none',
+    })
     const m = meta('relative-location', '/work')
     expect(ctx.sessionPersistence.locate(m)).toEqual({
       kind: 'jsonl',
-      path: logPath(resolve(absoluteRoot), '/work', m.id),
+      path: rawLogPath(resolve(absoluteRoot), '/work', m.id),
     })
     await fiber.dispose()
   })
@@ -150,26 +157,26 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => {
     root = await freshRoot()
     ctx = new Context()
     await ctx.plugin(SessionStore)
-    await ctx.plugin(SessionPersistenceJsonl, { root })
+    await ctx.plugin(SessionPersistenceJsonl, { root, compression: 'none' })
   })
   afterEach(async () => { await ctx.fiber.dispose() })
 
   it('lazy materialization: create() writes no file until the first append', async () => {
     const m = meta('lazy', '/work')
     const location = ctx.sessionPersistence.locate(m)
-    expect(location).toEqual({ kind: 'jsonl', path: logPath(root, '/work', m.id) })
+    expect(location).toEqual({ kind: 'jsonl', path: rawLogPath(root, '/work', m.id) })
     expect(isAbsolute(location!.path)).toBe(true)
 
     await ctx.sessionPersistence.create(m)
     // locate() is a pure target-path calculation: neither it nor create()
     // materializes a file before the first append.
     const dir = sessionDir(root, '/work')
-    await expect(stat(logPath(root, '/work', m.id))).rejects.toThrow()
+    await expect(stat(rawLogPath(root, '/work', m.id))).rejects.toThrow()
     expect((await ctx.sessionPersistence.list()).map(h => h.id)).not.toContain(m.id)
 
     await ctx.sessionPersistence.append(m.id, oneTurnLog())
     // now materialized
-    expect((await stat(logPath(root, '/work', m.id))).isFile()).toBe(true)
+    expect((await stat(rawLogPath(root, '/work', m.id))).isFile()).toBe(true)
     expect((await ctx.sessionPersistence.list()).map(h => h.id)).toContain(m.id)
     void dir
   })
@@ -191,7 +198,7 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => {
     }
     const childLocation = ctx.sessionPersistence.locate(child)
     expect(childLocation?.path).not.toBe(parentLocation?.path)
-    expect(childLocation).toEqual({ kind: 'jsonl', path: logPath(root, '/work', child.id) })
+    expect(childLocation).toEqual({ kind: 'jsonl', path: rawLogPath(root, '/work', child.id) })
   })
 
   it('round-trip is byte-identical (incl. assistant/chunk verbatim)', async () => {
@@ -213,7 +220,7 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => {
 
   it('rejects a stored v0 log containing a legacy request/header-delta event', async () => {
     const m = meta('legacy-header-delta', '/legacy')
-    const path = logPath(root, m.cwd, m.id)
+    const path = rawLogPath(root, m.cwd, m.id)
     await mkdir(sessionDir(root, m.cwd), { recursive: true })
     await writeFile(path, [
       JSON.stringify(toHeaderLine(m)),
@@ -228,7 +235,7 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => {
 
   it('rejects a stored v0 full header carrying the legacy fallback reason', async () => {
     const m = meta('legacy-header-fallback', '/legacy')
-    const path = logPath(root, m.cwd, m.id)
+    const path = rawLogPath(root, m.cwd, m.id)
     await mkdir(sessionDir(root, m.cwd), { recursive: true })
     await writeFile(path, [
       JSON.stringify(toHeaderLine(m)),
@@ -270,7 +277,7 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => {
     // Simulate a crash mid-second-turn: append raw lines that are NOT closed by
     // a turn/end (turn/start + step/start are fully written), plus a final
     // partial line with no newline (a torn fragment never fully flushed).
-    const path = logPath(root, '/proj', m.id)
+    const path = rawLogPath(root, '/proj', m.id)
     await writeFile(path, [
       JSON.stringify({ type: 'turn/start', seq: 6, time: 8, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } }),
       JSON.stringify({ type: 'step/start', seq: 7, time: 9, data: { turn: 2, step: 1 } }),
@@ -303,17 +310,17 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => {
     const m = meta('append-only')
     await ctx.sessionPersistence.create(m)
     await ctx.sessionPersistence.append(m.id, oneTurnLog())
-    const before = await readFile(logPath(root, undefined, m.id), 'utf8')
+    const before = await readFile(rawLogPath(root, undefined, m.id), 'utf8')
     const committedPrefix = before // the whole committed log
 
     // A crash tail then a repair-append.
-    await writeFile(logPath(root, undefined, m.id), '\n{"partial', { flag: 'a' })
+    await writeFile(rawLogPath(root, undefined, m.id), '\n{"partial', { flag: 'a' })
     await ctx.sessionPersistence.load(m.id)
     await ctx.sessionPersistence.append(m.id, [
       { type: 'turn/start', seq: 6, time: 9, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } },
       { type: 'turn/end', seq: 7, time: 10, data: { turn: 2, reason: { kind: 'completed' } } },
     ] as SessionEvent[])
-    const after = await readFile(logPath(root, undefined, m.id), 'utf8')
+    const after = await readFile(rawLogPath(root, undefined, m.id), 'utf8')
     // the committed prefix is byte-for-byte intact at the head of the file
     expect(after.startsWith(committedPrefix)).toBe(true)
   })
@@ -322,12 +329,12 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => {
     const m = meta('truncate-retry')
     await ctx.sessionPersistence.create(m)
     await ctx.sessionPersistence.append(m.id, oneTurnLog()) // materialized, seqs 0..5
-    const sizeBefore = (await stat(logPath(root, undefined, m.id))).size
+    const sizeBefore = (await stat(rawLogPath(root, undefined, m.id))).size
 
     // Force the NEXT fsync (inside appendLines) to fail once, AFTER writeFile
     // has already put bytes on disk — simulating an ENOSPC/fsync error
     // mid-append. The recovery truncate() also fsyncs, so allow that one.
-    const handle = await (await import('node:fs/promises')).open(logPath(root, undefined, m.id), 'r')
+    const handle = await (await import('node:fs/promises')).open(rawLogPath(root, undefined, m.id), 'r')
     const proto = Object.getPrototypeOf(handle) as { sync: () => Promise<void> }
     await handle.close()
     const realSync = proto.sync
@@ -344,7 +351,7 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => {
     // The append rejects, but the partial bytes are truncated back: the file is
     // its pre-append size and the cursor is unchanged.
     await expect(ctx.sessionPersistence.append(m.id, turn2)).rejects.toThrow(/ENOSPC/)
-    expect((await stat(logPath(root, undefined, m.id))).size).toBe(sizeBefore)
+    expect((await stat(rawLogPath(root, undefined, m.id))).size).toBe(sizeBefore)
     spy.mockRestore()
 
     // The retry now succeeds with NO seq gap — the log is contiguous 0..7.
@@ -425,7 +432,7 @@ describe('SessionPersistenceJsonl: write path (session/event → flush)', () =>
     root = await freshRoot()
     const ctx = new Context()
     await ctx.plugin(SessionStore)
-    await ctx.plugin(SessionPersistenceJsonl, { root })
+    await ctx.plugin(SessionPersistenceJsonl, { root, compression: 'none' })
 
     const a = ctx.sessions.create(SessionId('sa'))
     const b = ctx.sessions.create(SessionId('sb'))
@@ -552,7 +559,7 @@ describe('SessionPersistenceJsonl: edge cases', () => {
     root = await freshRoot()
     ctx = new Context()
     await ctx.plugin(SessionStore)
-    await ctx.plugin(SessionPersistenceJsonl, { root })
+    await ctx.plugin(SessionPersistenceJsonl, { root, compression: 'none' })
   })
   afterEach(async () => { await ctx.fiber.dispose() })
 
@@ -572,8 +579,8 @@ describe('SessionPersistenceJsonl: edge cases', () => {
     await p
     await ctx.sessionPersistence.append(SessionId('create-snap'), oneTurnLog())
     // The log materialized under the ORIGINAL cwd, not the mutated one.
-    expect((await stat(logPath(root, '/orig', SessionId('create-snap')))).isFile()).toBe(true)
-    await expect(stat(logPath(root, '/mutated', SessionId('create-snap')))).rejects.toThrow()
+    expect((await stat(rawLogPath(root, '/orig', SessionId('create-snap')))).isFile()).toBe(true)
+    await expect(stat(rawLogPath(root, '/mutated', SessionId('create-snap')))).rejects.toThrow()
   })
 
   it('list discovers sessions across multiple cwd buckets', async () => {
@@ -654,7 +661,7 @@ describe('SessionPersistenceJsonl: edge cases', () => {
     // of grafting no-cwd events onto a log with mismatched cwd.
     const ctx2 = new Context()
     await ctx2.plugin(SessionStore)
-    await ctx2.plugin(SessionPersistenceJsonl, { root })
+    await ctx2.plugin(SessionPersistenceJsonl, { root, compression: 'none' })
     let b!: Session
     await ctx2.plugin(Object.assign((inner: Context) => {
       b = inner.sessions.create(SessionId('x')) // no cwd
@@ -663,10 +670,10 @@ describe('SessionPersistenceJsonl: edge cases', () => {
 
     // The "/w" log is untouched — no no-cwd events were grafted onto it, and no
     // `_no-cwd` log for "x" was created.
-    const inW = scanLog(await readFile(logPath(root, '/w', SessionId('x'))))
+    const inW = scanLog(await readFile(rawLogPath(root, '/w', SessionId('x'))))
     expect(inW.meta.cwd).toBe('/w')
     expect(inW.events).toHaveLength(6)
-    await expect(stat(logPath(root, undefined, SessionId('x')))).rejects.toThrow()
+    await expect(stat(rawLogPath(root, undefined, SessionId('x')))).rejects.toThrow()
     await ctx2.fiber.dispose()
   })
 
@@ -710,7 +717,10 @@ describe('SessionPersistenceJsonl: edge cases', () => {
   it('list returns nothing when the root directory does not exist', async () => {
     const ctx2 = new Context()
     await ctx2.plugin(SessionStore)
-    await ctx2.plugin(SessionPersistenceJsonl, { root: join(root, 'does-not-exist-yet') })
+    await ctx2.plugin(SessionPersistenceJsonl, {
+      root: join(root, 'does-not-exist-yet'),
+      compression: 'none',
+    })
     expect(await ctx2.sessionPersistence.list()).toEqual([])
     await ctx2.fiber.dispose()
   })
@@ -722,7 +732,7 @@ describe('SessionPersistenceJsonl: edge cases', () => {
     await writeFile(filePath, 'x')
     const ctx2 = new Context()
     await ctx2.plugin(SessionStore)
-    await ctx2.plugin(SessionPersistenceJsonl, { root: filePath })
+    await ctx2.plugin(SessionPersistenceJsonl, { root: filePath, compression: 'none' })
     await expect(ctx2.sessionPersistence.list()).rejects.toThrow(/ENOTDIR/)
     await ctx2.fiber.dispose()
   })
@@ -733,7 +743,7 @@ describe('SessionPersistenceJsonl: edge cases', () => {
     const cwd = '/x'
     const ctx2 = new Context()
     await ctx2.plugin(SessionStore)
-    await ctx2.plugin(SessionPersistenceJsonl, { root })
+    await ctx2.plugin(SessionPersistenceJsonl, { root, compression: 'none' })
     await writeFile(sessionDir(root, cwd), 'x') // bucket path is now a FILE
     let s!: Session
     await ctx2.plugin(Object.assign((inner: Context) => {
@@ -748,14 +758,14 @@ describe('SessionPersistenceJsonl: edge cases', () => {
     const m = meta('disk-append', '/d')
     await ctx.sessionPersistence.create(m)
     await ctx.sessionPersistence.append(m.id, oneTurnLog())
-    await writeFile(logPath(root, '/d', m.id), '\n{"partial crash', { flag: 'a' })
+    await writeFile(rawLogPath(root, '/d', m.id), '\n{"partial crash', { flag: 'a' })
 
     // A FRESH backend with no in-memory state: append directly (no prior load)
     // → append must adopt from disk, and the adopt's load schedules a repair
     // that the same append then performs before writing.
     const ctx2 = new Context()
     await ctx2.plugin(SessionStore)
-    await ctx2.plugin(SessionPersistenceJsonl, { root })
+    await ctx2.plugin(SessionPersistenceJsonl, { root, compression: 'none' })
     await ctx2.sessionPersistence.append(m.id, [
       { type: 'turn/start', seq: 6, time: 9, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } },
       { type: 'turn/end', seq: 7, time: 10, data: { turn: 2, reason: { kind: 'completed' } } },
@@ -791,7 +801,7 @@ describe('SessionPersistenceJsonl: edge cases', () => {
     // nondeterministic. create scans every bucket, not just meta.cwd's.
     const ctx2 = new Context()
     await ctx2.plugin(SessionStore)
-    await ctx2.plugin(SessionPersistenceJsonl, { root })
+    await ctx2.plugin(SessionPersistenceJsonl, { root, compression: 'none' })
     await expect(ctx2.sessionPersistence.create(meta('dup-id', '/projB')))
       .rejects.toThrow(/already has a persisted log on disk/)
     await ctx2.fiber.dispose()
@@ -801,7 +811,7 @@ describe('SessionPersistenceJsonl: edge cases', () => {
     root = await freshRoot()
     const ctx2 = new Context()
     await ctx2.plugin(SessionStore)
-    await ctx2.plugin(SessionPersistenceJsonl, { root })
+    await ctx2.plugin(SessionPersistenceJsonl, { root, compression: 'none' })
     const session = ctx2.sessions.create(SessionId('flush-fail'))
     // A full turn lands in the write-behind buffer.
     session.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }, { surfaceOp: 'append' })

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

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

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

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

packages/support/acp-snapshot/README.md

@@ -52,5 +52,5 @@ None; this package neither assembles nor sends a provider request.
 
 ## Known Limitations and Deferred Work
 
-- **Session harvest is JSONL-only** — `runScenario` collects persisted `.jsonl` logs, so an example composed over the SQLite persistence backend has no snapshot path.
+- **Session harvest requires raw JSONL mode** — `runScenario` collects persisted `.jsonl` logs, so snapshot configs set `persistenceCompression: 'none'`; compressed JSONL and SQLite compositions have no snapshot-harvest path.
 - **The subprocess boots the unbuilt tsx/Loader path only** — the built-bin artifact is guarded by the separate `built-bin` e2e smokes, never by this tier.

packages/support/acp-snapshot/src/harness.ts

@@ -395,11 +395,11 @@ async function runStep(
  * header line, and return them ordered primary-first: the top-level session (no
  * `parentSession`) leads, then each subagent child by ascending `createdAt`.
  *
- * The JSONL backend lays sessions out as `<root>/<cwd-bucket>/<encoded-id>.jsonl`
- * (one bucket per cwd), so a parent and its same-cwd in-process child land in
- * the SAME bucket — collecting all files across all buckets catches both (a
- * first-match short-circuit would silently drop the child). Returns `[]` if no
- * log was produced (a no-session scenario).
+ * Snapshot configs select the JSONL backend's raw mode, which lays sessions
+ * out as `<root>/<cwd-bucket>/<encoded-id>.jsonl` (one bucket per cwd). A
+ * parent and its same-cwd in-process child land in the SAME bucket, so
+ * collecting all files across all buckets catches both. Returns `[]` if no log
+ * was produced (a no-session scenario).
  */
 async function harvestSessionLogs(root: string): Promise<HarvestedLog[]> {
   let cwdDirs: string[]

python/sdk/tests/manual_sdk_agent_smoke.py

@@ -83,15 +83,12 @@ def run_smoke(repo_root: Path, keep_sessions: bool) -> None:
         assert request["authorization"] == "Bearer sdk-smoke-key"
         assert request["body"]["model"] == "sdk-smoke-model"
 
-        jsonl_files = sorted(session_root.rglob("*.jsonl"))
-        assert jsonl_files, f"no jsonl sessions were written under {session_root}"
-        print("session_jsonl_files:")
+        jsonl_files = sorted(session_root.rglob("*.jsonl.zstd"))
+        assert jsonl_files, f"no Zstandard JSONL sessions were written under {session_root}"
+        print("session_jsonl_zstd_files:")
         for path in jsonl_files:
             print(f"  {path} bytes={path.stat().st_size}")
-            with path.open("r", encoding="utf-8") as handle:
-                first_line = handle.readline().strip()
-            if first_line:
-                print(f"  first_line={first_line[:500]}")
+            assert path.read_bytes().startswith(bytes.fromhex("28b52ffd"))
     finally:
         server.shutdown()
         server.server_close()

scripts/run-gates.ts

@@ -180,6 +180,11 @@ function gatesForMode(selected: Mode): Gate[] {
           'run',
           'packages/workflow/workflow-workerthread/tests/source-worker.compat.spec.ts',
         ], { label: 'source worker smoke' }),
+        pnpmExec('jsonl-zstd-smoke', [
+          'vitest',
+          'run',
+          'packages/session-persistence/session-persistence-jsonl/tests/zstd.compat.spec.ts',
+        ], { label: 'JSONL Zstandard smoke' }),
       ]
     case 'pre-push':
       return [

scripts/smoke-python-runtime.py

@@ -64,6 +64,7 @@ CUSTOM_CORDIS = """\
   name: '@deepseek-ai/dsh-session-persistence-jsonl'
   config:
     root: !!js process.env.DSH_SESSION_ROOT
+    compression: 'none'
 - id: bash
   name: '@deepseek-ai/dsh-bash-local'
   config:
@@ -391,7 +392,7 @@ def smoke_sdk_default(base_url: str) -> None:
             result = harness.run("reply with the smoke text", session_id="default-smoke")
         assert result.status == "ok", result
         assert result.final_response == EXPECTED_TEXT, result.final_response
-        assert_session_log(sessions, root, EXPECTED_TEXT)
+        assert_zstd_session_log(sessions)
 
 
 def smoke_sdk_custom(base_url: str, executable: Path) -> None:
@@ -585,6 +586,14 @@ def assert_session_log(sessions: Path, cwd: Path, *expected_texts: str) -> None:
             raise AssertionError(f"session log has no {expected!r} response: {logs[0]}")
 
 
+def assert_zstd_session_log(sessions: Path) -> None:
+    logs = list(sessions.rglob("*.jsonl.zstd"))
+    if len(logs) != 1:
+        raise AssertionError(f"expected one Zstandard JSONL session log under {sessions}, found {logs}")
+    if not logs[0].read_bytes().startswith(bytes.fromhex("28b52ffd")):
+        raise AssertionError(f"session log has no Zstandard magic: {logs[0]}")
+
+
 def read_session_logs(sessions: Path) -> dict[str, list[dict[str, object]]]:
     """Parse every persisted JSONL session into a map keyed by header id."""
     logs: dict[str, list[dict[str, object]]] = {}