Merge remote-tracking branch 'origin/master' into feat/read-image-context

2026-08-11 11:54:22 +08:00
parent d8abd97b1c f2b3a7382e
commit 2a755f2863
65 changed files with 2004 additions and 226 deletions
--- a/packages/core/session/README.i18n.yaml
+++ b/packages/core/session/README.i18n.yaml
@@ -2,5 +2,5 @@
 # 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 packages/core/session/README.md
-README.md: db477d94037d3463870fc8e66ea35d5e607fb6fe
-README.zh.md: 1ce1e823a7e0fdbcf7b6898764a89c52b74adf6a
+README.md: 57569e9c0dbfa7cb696e3a561a9ff108c2ac981f
+README.zh.md: 16629dc70c79ca838ba7088aeafcc5b38b124f87
--- a/packages/core/session/README.md
+++ b/packages/core/session/README.md
@@ -76,10 +76,11 @@ Also defines `TurnEndReasonMap`, the merge-extensible `kind`-tagged sum type for

 An interrupted live turn ends with `{ kind: 'aborted', reason: AgentCancelCause }`, preserving the typed cancellation cause in the durable transcript. Persistence imports the coarse aborted outcome from the supported older format as `{ kind: 'aborted', reason: { kind: 'legacy' } }`, because that record did not retain its caller. A turn failure carries `{ kind: 'error', error }`; crash recovery alone synthesizes `{ kind: 'interrupted' }`.

-Every `SessionEvent` carries two optional top-level fields (structural metadata):
+Every `SessionEvent` carries three optional top-level fields (structural metadata):

 - `sourceEventSeqs?: number[]` — seq numbers of earlier events cited as sources (e.g., the `assistant/chunk` seqs behind an `assistant/message`, or the shadowed entries behind a compaction replacement entry). On `assistant/message`, a present `[]` records a known empty provider stream, while omission means a legacy or foreign event did not record the source stream; other surface events require a non-empty list when this field is present.
 - `surfaceOp?: SurfaceOp` — how this event entered the surface. Absent for non-surface events (boundaries, chunks, usage, errors).
+- `ignorable?: true` — marks an event a reader may safely skip when it does not recognize the type; absent means required, so an unknown-type event refuses session reconstruction ([mechanism](../../../.agents/notes/implemented/architecture/2026-08-10-session-log-version-mechanism.md)).

 ### Metadata types (`types.ts`)

@@ -139,5 +140,5 @@ Logging causes no invalidation, and exact reconstruction preserves request-prefi

 - **Session branching/tree** (pi-style entry tree) — deferred unless needed beyond boundary-based `fork()`.
 - **`fork()` cuts only at stable boundaries of live sessions** — the selected prefix must end outside an open turn and the source must be in the store; forking a persisted-but-unloaded session is excluded from the [fork API](../../../.agents/notes/implemented/feature/2026-06-30-session-store-fork-api.md).
- **`SESSION_FORMAT_VERSION` stays pinned at `0`** — pre-release, no broad compatibility implied: `Session` accepts only current seed shapes and a backend rejects any other version. Narrow storage import upgrades belong to the persistence boundary ([policy](../../../AGENTS.md), [pre-identity message recovery](../../../.agents/notes/implemented/bug-fix/2026-07-28-load-pre-identity-session-messages.md)).
+- **`SESSION_FORMAT_VERSION` stays pinned at `0`** — pre-release, no broad compatibility implied: `Session` accepts only current seed shapes, and a backend refuses any other version naming the direction (newer: "written by a newer harness — upgrade"; older: no upgrade path ships yet). Unknown event types refuse the same way unless marked `ignorable` in the envelope; the versioning mechanism is the [session-log-version-mechanism note](../../../.agents/notes/implemented/architecture/2026-08-10-session-log-version-mechanism.md). Narrow storage import upgrades belong to the persistence boundary ([policy](../../../AGENTS.md), [pre-identity message recovery](../../../.agents/notes/implemented/bug-fix/2026-07-28-load-pre-identity-session-messages.md)).
 - **`TurnEndReasonMap` omits the ACP-named `refusal` / `max_turn_requests` variants** — producer-gated: they land when an adapter or the loop first emits them.
--- a/packages/core/session/README.zh.md
+++ b/packages/core/session/README.zh.md
@@ -76,10 +76,11 @@

 被中断的实时轮次以 `{ kind: 'aborted', reason: AgentCancelCause }` 结束，在持久 transcript（文本记录）中保留类型化取消原因。持久化会将受支持旧格式中的粗粒度中止结果导入为 `{ kind: 'aborted', reason: { kind: 'legacy' } }`，因为该记录没有保留调用方。轮次失败携带 `{ kind: 'error', error }`；只有崩溃恢复会合成 `{ kind: 'interrupted' }`。

-每个 `SessionEvent` 都有两个可选顶层字段（结构元数据）：
+每个 `SessionEvent` 都有三个可选顶层字段（结构元数据）：

 - `sourceEventSeqs?: number[]`：被引用为来源的较早事件 seq（例如 `assistant/message` 引用的 `assistant/chunk` seq，或压缩替换条目引用的已遮蔽条目）。对于 `assistant/message`，存在的 `[]` 表示已知提供方流为空；省略则表示旧版或外部事件没有记录源流。其他 surface 事件若有此字段，则要求非空列表。
 - `surfaceOp?: SurfaceOp`：事件进入 surface 的方式。非 surface 事件（边界、分片、用量、错误）不含该字段。
+- `ignorable?: true`：标记读取器在不认识事件类型时可以安全跳过该事件；缺失表示必需，不认识的事件类型会使会话重建被拒绝（[机制](../../../.agents/notes/implemented/architecture/2026-08-10-session-log-version-mechanism.md)）。

 ### 元数据类型（`types.ts`）

@@ -139,5 +140,5 @@

 - **会话分支／树**（pi 风格条目树）：除非需要超越基于边界的 `fork()` 能力，否则暂缓。
 - **`fork()` 仅在实时会话的稳定边界处切分**：所选前缀结束时不得有开放轮次，且源会话必须位于存储中；[fork API](../../../.agents/notes/implemented/feature/2026-06-30-session-store-fork-api.md) 不支持对已持久化但未加载的会话进行 fork。
- **`SESSION_FORMAT_VERSION` 固定为 `0`**：预发布阶段不承诺广泛兼容性；`Session` 只接受当前 seed 形状，后端会拒绝其他任何版本。范围受限的存储导入升级应由持久化边界负责（[政策](../../../AGENTS.md)、[消息标识机制引入前的消息恢复](../../../.agents/notes/implemented/bug-fix/2026-07-28-load-pre-identity-session-messages.md)）。
+- **`SESSION_FORMAT_VERSION` 固定为 `0`**：预发布阶段不承诺广泛兼容性；`Session` 只接受当前 seed 形状，后端拒绝其他任何版本并说明方向（更新的版本提示"由更新的 harness 写入，请升级"；更旧的版本说明尚无升级路径）。不认识的事件类型同样被拒绝，除非信封带 `ignorable` 标记；版本机制见 [session-log 版本机制 Agent Note](../../../.agents/notes/implemented/architecture/2026-08-10-session-log-version-mechanism.md)。范围受限的存储导入升级应由持久化边界负责（[政策](../../../AGENTS.md)、[消息标识机制引入前的消息恢复](../../../.agents/notes/implemented/bug-fix/2026-07-28-load-pre-identity-session-messages.md)）。
 - **`TurnEndReasonMap` 不含 ACP（Agent Client Protocol）命名的 `refusal`／`max_turn_requests` 变体**：受生产方约束；只有当适配器或循环首次产生这些变体时才加入。
--- a/packages/core/session/src/index.ts
+++ b/packages/core/session/src/index.ts
@@ -32,6 +32,7 @@ export type { ChunkRow, StorageRecord } from './chunk-rows.ts'
 export type { SessionSurface, SurfaceFoldReplacement, SurfaceFoldResult } from './surface.ts'
 export { deriveEventMessage, foldSurface, isAppendSurfaceEvent, isReplacementSurfaceEvent, isSurfaceEvent, isSurfaceEligibleType } from './surface.ts'
 export { canonicalHeader, foldRequestHeader, headerEquals } from './request-header.ts'
+export { KNOWN_SESSION_EVENT_TYPES } from './known-event-types.ts'

 /**
 * Find the latest closed turn that entered at least one model step, ignoring
@@ -243,6 +244,7 @@ function assertSessionEventEnvelope(value: Record<string, unknown>, index: numbe
      case 'data':
      case 'surfaceOp':
      case 'sourceEventSeqs':
+      case 'ignorable':
        break
      default:
        throw new Error(`seed event at index ${index} has an invalid event envelope`)
@@ -254,7 +256,8 @@ function assertSessionEventEnvelope(value: Record<string, unknown>, index: numbe
  if (typeof type !== 'string'
    || typeof seq !== 'number' || !Number.isSafeInteger(seq) || seq < 0
    || typeof time !== 'number' || !Number.isSafeInteger(time)
-    || event['data'] === undefined) {
+    || event['data'] === undefined
+    || (event['ignorable'] !== undefined && event['ignorable'] !== true)) {
    throw new Error(`seed event at index ${index} has an invalid event envelope`)
  }
  switch (type) {
--- a/packages/core/session/src/known-event-types.ts
+++ b/packages/core/session/src/known-event-types.ts
@@ -0,0 +1,59 @@
+/**
+ * GENERATED by `scripts/gen-persistence-catalog.ts` — do not edit by hand; run
+ * `pnpm run gen-persistence-catalog` to regenerate (verified fresh by
+ * `pnpm run verify-persistence-catalog`, part of `doc-sync`).
+ * @module @deepseek-ai/dsh-session/known-event-types
+ */
+
+/**
+ * Every `SessionEventMap` member declared in this repository — the event
+ * vocabulary this build understands. The persistence read path refuses to
+ * interpret a log containing a type outside this set unless the event
+ * carries the envelope's `ignorable` marker (see `SessionEvent.ignorable`
+ * in `./types.ts`): such a log was likely written by a newer harness, and
+ * silently skipping a required event would reconstruct a wrong session.
+ * Downstream (out-of-repo) plugin events are outside this list by
+ * construction; a registration surface for them is deferred until such a
+ * consumer exists.
+ */
+export const KNOWN_SESSION_EVENT_TYPES: ReadonlySet<string> = new Set([
+  'agent-preset/selected',
+  'agent/inbox/spliced',
+  'approval/asked',
+  'approval/decided',
+  'approval/policy',
+  'assistant/chunk',
+  'assistant/message',
+  'command/done',
+  'command/run',
+  'compact/end',
+  'compact/prune',
+  'compact/start',
+  'compact/summary',
+  'feedback/record',
+  'goal/change',
+  'hook/invoked',
+  'hook/result',
+  'llm/retry',
+  'llm/retry-started',
+  'permission/preset',
+  'plan/mode',
+  'request/context',
+  'request/header',
+  'sandbox/mode',
+  'session/end-seed',
+  'session/title',
+  'session/title-llm-request',
+  'step/end',
+  'step/start',
+  'subagent/descriptor',
+  'todo/write',
+  'tool/call',
+  'tool/code-dispatch',
+  'tool/code-dispatch-start',
+  'tool/result',
+  'turn/end',
+  'turn/start',
+  'user/message',
+  'web/deepseek-search-llm-request',
+])
--- a/packages/core/session/src/types.ts
+++ b/packages/core/session/src/types.ts
@@ -30,8 +30,23 @@ export function SessionId(id: string): SessionId {
 * and enforced by every persistence backend on load. The single source of truth for the
 * version — write sites and the load-time check all read it.
 * While the harness is unreleased it is pinned at `0`: no compatibility is
- * implied, incompatible logs are rejected, and no migration is provided. A
- * monotonic version policy starts with the first tagged release.
+ * implied, incompatible logs are rejected, and no migration is provided.
+ *
+ * The version is a single monotonic integer with no major/minor split. Whether
+ * a bump is needed is decided by what the WRITER emits, never by what a newer
+ * reader can accept: bump exactly when an older runtime could no longer handle
+ * a new log with full semantic correctness ("parses without error" is not
+ * correctness — silently skipping content that shapes reconstruction is a
+ * wrong read). Only structural changes reach that bar: the header shape, the
+ * {@link SessionEvent} envelope, core event semantics, or the surface
+ * mechanism (the {@link SurfaceEventType} set and {@link SurfaceOp} variants).
+ * Adding an ordinary event type does not bump — the per-event
+ * {@link SessionEvent.ignorable} guard covers vocabulary growth instead. When
+ * in doubt, bump: a near-identity upgrade step is almost free, a missed bump
+ * makes older runtimes read new logs wrong silently. The full mechanism
+ * (upgrade-step chain, in-memory view conversion, migrate-on-continue) is
+ * recorded in the session-log-version-mechanism Agent Note
+ * (`.agents/notes/implemented/architecture/2026-08-10-session-log-version-mechanism.md`).
 */
 export const SESSION_FORMAT_VERSION = 0

@@ -389,6 +404,17 @@ export type SessionEvent<T extends SessionEventType = SessionEventType> = {
    /** Unix epoch milliseconds. */
    time: number
    data: SessionEventMap[K]
+    /**
+     * Marks an event a reader may safely skip when it does not recognize
+     * `type`. Absent means required: a reader meeting an unrecognized type
+     * without this marker MUST refuse to reconstruct the session instead of
+     * silently dropping the event, because an unrecognized required event may
+     * change how the rest of the log is interpreted. A writer sets `true` only
+     * on purely informational records whose loss cannot affect reconstruction;
+     * defaulting to required means a forgotten marker over-refuses (an
+     * inconvenience) rather than silently resuming a gutted session.
+     */
+    ignorable?: true
  } & (K extends SurfaceEventType ? {
    /**
     * Seq numbers of earlier events that this event cites as sources
--- a/packages/core/session/tests/session.spec.ts
+++ b/packages/core/session/tests/session.spec.ts
@@ -1090,12 +1090,20 @@ describe('Session', () => {
      { ...base, time: '1' },
      { ...base, time: 0.5 },
      { type: base.type, seq: base.seq, time: base.time },
+      { ...base, ignorable: false },
+      { ...base, ignorable: 'yes' },
    ]

    for (const [index, event] of cases.entries()) {
      expect(() => Session.create(SessionId(`bad-envelope-${index}`), [event as SessionEvent]))
        .toThrow(/invalid event envelope/)
    }
+
+    // `ignorable: true` is the one accepted marker value (unknown-type skip contract).
+    const marked = Session.create(SessionId('ignorable-envelope'), [
+      { ...base, ignorable: true } as SessionEvent,
+    ])
+    expect(marked.events[0]?.ignorable).toBe(true)
  })
 })

--- a/packages/host/apiproxy/src/api/sessions.schema.ts
+++ b/packages/host/apiproxy/src/api/sessions.schema.ts
@@ -45,6 +45,7 @@ export const sessionEventSchema = z.object({
  data: z.unknown(),
  sourceEventSeqs: z.array(z.number()).optional(),
  surfaceOp: z.unknown().optional(),
+  ignorable: z.literal(true).optional(),
 }) as unknown as z.ZodType<SessionEvent>

 /** SessionSummary row of session.list (`projections` reuses the history block's shape and schema). */
--- a/packages/mcp/mcp-client/README.i18n.yaml
+++ b/packages/mcp/mcp-client/README.i18n.yaml
@@ -2,5 +2,5 @@
 # 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 packages/mcp/mcp-client/README.md
-README.md: 76d1271f6f7a3e9c959bdcf5e969906f25563c56
-README.zh.md: b2da1119af3a8d52761a5040059e7a9d922aa567
+README.md: 97ac9c173fc2b848f524e8c0fdd93eca072567af
+README.zh.md: 1b5b5c523e0a477db30f97a748651dbe7e6992ea
--- a/packages/mcp/mcp-client/README.md
+++ b/packages/mcp/mcp-client/README.md
@@ -45,6 +45,10 @@ The model sees `mcp__github__create_issue`, `mcp__web__search`, … — the same
 | `headers` | http | no | Extra headers (e.g. auth tokens) |
 | `toolCallTimeoutMs` | both | no | Timeout per `callTool` invocation (default 60000) |
 | `failOnStartupError` | both | no | Reject plugin activation when initial connection or tool synchronization fails (default `false`) |
+| `reconnect.enabled` | both | no | Reconnect automatically after a lost connection (default `true`) |
+| `reconnect.initialDelayMs` | both | no | First reconnect delay in ms; doubles per consecutive failed attempt (default 500) |
+| `reconnect.maxDelayMs` | both | no | Backoff ceiling in ms; also the uptime after which the attempt budget resets (default 30000) |
+| `reconnect.maxAttempts` | both | no | Consecutive failed attempts per outage before giving up for good (default 10) |

 ## Tool naming

@@ -62,7 +66,9 @@ Every MCP tool has two names: the raw MCP name (sent on the wire in `tools/call`
 - Tool execute: `client.callTool({ name: rawName, arguments }, { signal })` with timeout + abort support—the public name is never sent to the server.
 - Canonical success is `{ content: JsonValue[], structuredContent? }`; complete JSON MCP blocks survive for programmatic callers. A supported advertised `outputSchema` validates `structuredContent`; unsupported schema vocabulary falls back to unconstrained `JsonValue`.
 - Native/model rendering keeps the existing text projection: text blocks join with newlines while image, audio, resource, and unsupported blocks become placeholders.
- On disconnect/crash: no auto-reconnect. Registered tools remain until plugin disposal or a successful re-sync, and calls can fail against the closed transport; reload with HMR or restart the Host to reconnect.
+- On disconnect/crash: the supervisor restarts the original server config with exponential backoff (`reconnect.initialDelayMs` doubling up to `reconnect.maxDelayMs`) and re-runs discovery on success — the recovered generation replaces the previous one, so tools neither duplicate nor leak. During the outage the last good generation stays registered; calls against it fail until recovery.
+- Reconnection is budgeted per outage: after `reconnect.maxAttempts` consecutive failures the server's tools are unregistered and reconnection stops until an HMR reload or Host restart. A connection that survives past `maxDelayMs` resets the budget, so an occasionally-crashing server recovers indefinitely while a crash-looping one — even with briefly successful connects — still exhausts the cap instead of restarting forever.
+- Reconnect states are user-visible in logs: reconnecting (warn, with attempt count and delay), recovered (info), final failure and disabled-loss (error). Disposal cancels any pending reconnect. With `reconnect.enabled: false`, a lost connection keeps tools registered but failing until a reload — the manual-recovery behavior.

 ## Services consumed

@@ -76,7 +82,7 @@ Every MCP tool has two names: the raw MCP name (sent on the wire in `tools/call`

 #### What the model sees

-After initial discovery succeeds, each advertised MCP tool appears as a native tool named `mcp__<serverName>__<rawName>` (or its deterministic normalized form), with the server-provided description and input schema. A successful re-sync replaces the generation; plugin disposal removes it.
+After initial discovery succeeds, each advertised MCP tool appears as a native tool named `mcp__<serverName>__<rawName>` (or its deterministic normalized form), with the server-provided description and input schema. A successful re-sync — including the one after an automatic reconnect — replaces the generation; plugin disposal or an exhausted reconnect budget removes it.

 #### Token effect

@@ -84,7 +90,7 @@ Data-dependent schema cost is paid on every request while the tools are register

 #### KV Cache effect

-Prefix-stable while the discovered tool set and schemas are unchanged. A re-sync that adds, removes, renames, or changes a tool replaces definitions and may invalidate reuse from the first changed schema token.
+Prefix-stable while the discovered tool set and schemas are unchanged. A re-sync that adds, removes, renames, or changes a tool replaces definitions and may invalidate reuse from the first changed schema token; a reconnect that recovers an unchanged list reproduces identical definitions and stays prefix-stable.

 ### Tool-call history and results

@@ -104,6 +110,6 @@ Append-only; newly visible content follows the reusable request prefix and does

 - **Tools are the only bridged MCP capability** — Resources and Prompts have no harness consumption surface and are deferred.
 - **Startup timeout is inherited from the MCP SDK** — DSH does not yet expose a connection/discovery timeout. Each initialize or paginated `tools/list` request uses the SDK's 60-second default, so an unresponsive server or cursor chain can delay both activation and teardown while the initial synchronization settles.
- **Crash recovery is manual** — transport closure does not auto-reconnect; registered tools can remain visible but fail against the closed transport until an HMR reload or Host restart.
+- **Reconnect triggers on transport close** — a crashed stdio child fires it; Streamable HTTP failures surface per request and through the SDK transport's own SSE-stream recovery, so an unreachable HTTP server is retried per call rather than respawned by the supervisor.
 - **Native non-text rendering is lossy** — image, audio, and resource payloads become placeholders in model context even though the execution-local canonical value preserves their JSON blocks. Richer Native multimedia projection is deferred.
 - **Unsupported MCP output schemas are not enforced** — `structuredContent` falls back to `JsonValue` when the advertised schema uses vocabulary outside the harness subset.
--- a/packages/mcp/mcp-client/README.zh.md
+++ b/packages/mcp/mcp-client/README.zh.md
@@ -45,6 +45,10 @@ MCP 客户端桥接插件：连接外部 [Model Context Protocol](https://modelc
 | `headers` | http | 否 | 额外标头（例如认证 token） |
 | `toolCallTimeoutMs` | 两者 | 否 | 每次 `callTool` 调用的超时（默认 60000） |
 | `failOnStartupError` | 两者 | 否 | 初始连接或工具同步失败时拒绝插件激活（默认 `false`） |
+| `reconnect.enabled` | 两者 | 否 | 连接丢失后自动重新连接（默认 `true`） |
+| `reconnect.initialDelayMs` | 两者 | 否 | 首次重连延迟（毫秒）；每次连续失败尝试翻倍（默认 500） |
+| `reconnect.maxDelayMs` | 两者 | 否 | 退避上限（毫秒）；同时也是重置尝试预算所需的正常运行时长（默认 30000） |
+| `reconnect.maxAttempts` | 两者 | 否 | 每次中断期间连续失败尝试次数上限，超出后彻底放弃（默认 10） |

 ## 工具命名

@@ -62,7 +66,9 @@ MCP 客户端桥接插件：连接外部 [Model Context Protocol](https://modelc
 - 工具执行：`client.callTool({ name: rawName, arguments }, { signal })`，支持超时 + 中止；公开名称绝不会发给服务器。
 - 规范成功值是 `{ content: JsonValue[], structuredContent? }`；完整的 JSON MCP 块会保留给编程调用方。受支持且已声明的 `outputSchema` 会验证 `structuredContent`；不受支持的 schema 词汇会回退为不受约束的 `JsonValue`。
 - Native／模型渲染保留现有文本投影：文本块以换行连接，图片、音频、资源和不受支持的块会变成占位符。
- 断开／崩溃时：不自动重新连接。已注册工具会一直保留到对插件执行 dispose（资源释放）或成功重新同步，针对已关闭传输的调用可能失败；请通过 HMR 重新加载或重启 Host 来重新连接。
+- 断开／崩溃时：supervisor 以指数退避（`reconnect.initialDelayMs` 逐次翻倍，上限 `reconnect.maxDelayMs`）重启原始服务器配置，成功后重新执行发现——恢复的世代会替换前一个，因此工具既不会重复也不会泄漏。中断期间最后一个正常世代保持注册；针对它的调用在恢复前会失败。
+- 重连按中断预算控制：连续失败达到 `reconnect.maxAttempts` 次后，该服务器的工具会被注销，重连停止，直到 HMR 重载或重启 Host。连接存活超过 `maxDelayMs` 会重置预算，因此偶尔崩溃的服务器可以无限恢复，而崩溃循环的服务器——即使短暂连接成功——仍会耗尽上限而非永远重启。
+- 重连状态在日志中对用户可见：reconnecting（warn，含尝试次数和延迟）、recovered（info）、最终失败和 disabled-loss（error）。dispose（资源释放）会取消任何待执行的重连。设置 `reconnect.enabled: false` 时，连接丢失后工具保持注册但调用失败，直到重载——即手动恢复行为。

 ## 消费的服务

@@ -76,7 +82,7 @@ MCP 客户端桥接插件：连接外部 [Model Context Protocol](https://modelc

 #### 模型看到的内容

-初始发现成功后，每个已声明的 MCP 工具都会显示为名为 `mcp__<serverName>__<rawName>`（或其确定性规范化形式）的原生工具，并携带服务器提供的描述和输入 schema。成功的重新同步会替换整个世代；对插件执行 dispose 会移除该世代。
+初始发现成功后，每个已声明的 MCP 工具都会显示为名为 `mcp__<serverName>__<rawName>`（或其确定性规范化形式）的原生工具，并携带服务器提供的描述和输入 schema。成功的重新同步——包括自动重连后的同步——会替换整个世代；对插件执行 dispose（资源释放）或重连预算耗尽会移除该世代。

 #### Token 影响

@@ -84,7 +90,7 @@ MCP 客户端桥接插件：连接外部 [Model Context Protocol](https://modelc

 #### KV Cache 影响

-只要已发现工具集合及其 schema 不变，前缀就保持稳定。增加、移除、重命名或更改工具的重新同步会替换定义，并可能使从第一个变化的 schema token 起的复用失效。
+只要已发现工具集合及其 schema 不变，前缀就保持稳定。增加、移除、重命名或更改工具的重新同步会替换定义，并可能使从第一个变化的 schema token 起的复用失效；恢复了未变列表的重连会生成完全相同的定义，前缀保持稳定。

 ### 工具调用历史与结果

@@ -104,6 +110,6 @@ MCP 客户端桥接插件：连接外部 [Model Context Protocol](https://modelc

 - **只桥接 MCP 的工具能力**：资源和提示词没有 harness 消费接口，暂缓实现。
 - **启动超时继承自 MCP SDK**：DSH 尚未公开连接／发现超时。每次 initialize 请求或分页 `tools/list` 请求都使用 SDK 默认的 60 秒，因此在初始同步完成期间，无响应的 server 或 cursor chain 可能同时延迟激活与 teardown。
- **崩溃恢复需要手动触发**：传输关闭后不会自动重新连接；已注册工具可能仍然可见，但会因传输已关闭而调用失败，直到 HMR 重载或重启 Host。
+- **重连在传输关闭时触发**：崩溃的 stdio 子进程会触发重连；Streamable HTTP 失败通过每次请求以及 SDK 传输自身的 SSE（Server-Sent Events）流恢复机制暴露，因此不可达的 HTTP 服务器会按调用重试，而非由 supervisor 重新 spawn。
 - **Native 非文本渲染有损**：图片、音频与资源载荷在模型上下文中会变成占位符，即使执行局部的规范值保留了其 JSON 块。更丰富的 Native 多媒体投影暂缓实现。
 - **不强制执行不受支持的 MCP 输出 schema**：已声明 schema 使用 harness 子集之外的词汇时，`structuredContent` 会回退到 `JsonValue`。
--- a/packages/mcp/mcp-client/package.json
+++ b/packages/mcp/mcp-client/package.json
@@ -35,6 +35,7 @@
    "@deepseek-ai/dsh-invariants": "workspace:^",
    "@deepseek-ai/dsh-llm": "workspace:^",
    "@deepseek-ai/dsh-subprocess": "workspace:^",
+    "@deepseek-ai/dsh-timeout": "workspace:^",
    "@deepseek-ai/dsh-tools": "workspace:^",
    "@deepseek-ai/cordis": "workspace:^"
  },
@@ -47,6 +48,7 @@
    "@deepseek-ai/dsh-invariants": "workspace:^",
    "@deepseek-ai/dsh-llm": "workspace:^",
    "@deepseek-ai/dsh-subprocess": "workspace:^",
+    "@deepseek-ai/dsh-timeout": "workspace:^",
    "@deepseek-ai/dsh-tools": "workspace:^",
    "@modelcontextprotocol/server-everything": "^2026.7.4",
    "@modelcontextprotocol/server-filesystem": "^2026.7.4",
--- a/packages/mcp/mcp-client/src/connection.ts
+++ b/packages/mcp/mcp-client/src/connection.ts
@@ -0,0 +1,351 @@
+/**
+ * Connection supervisor: owns the MCP client/transport generations for one
+ * plugin instance, keeps the harness tool registry in sync with the live
+ * generation, and — when the connection drops — restarts the configured
+ * server with bounded exponential backoff.
+ *
+ * One outage shares one attempt budget (`maxAttempts` consecutive failed
+ * attempts, delays doubling from `initialDelayMs` up to `maxDelayMs`). A
+ * connection that stays up past the stability window closes the outage, so
+ * the next disconnect starts a fresh budget while a crash-looping server —
+ * even one whose connects briefly succeed — still exhausts the cap instead of
+ * restarting forever. Exhaustion unregisters the server's tools and stops;
+ * disposal (including HMR) is the only way back from that state.
+ *
+ * @module
+ */
+
+import { Client } from '@modelcontextprotocol/sdk/client/index.js'
+import { ToolListChangedNotificationSchema } from '@modelcontextprotocol/sdk/types.js'
+import type { Context } from '@deepseek-ai/cordis'
+import { MAX_TIMER_DELAY_MS } from '@deepseek-ai/dsh-timeout'
+import { createTransport } from './transport.ts'
+import { syncTools } from './tools.ts'
+import type { ToolBridgeOptions, ToolDisposers } from './tools.ts'
+import type { Config } from './index.ts'
+
+/** Automatic reconnect policy for one MCP server connection. */
+export interface ReconnectConfig {
+  /** Reconnect automatically after a lost connection (default true). */
+  enabled?: boolean
+  /** First reconnect delay in milliseconds; doubles per consecutive failed attempt (default 500). */
+  initialDelayMs?: number
+  /** Backoff ceiling in milliseconds; also the uptime after which the attempt budget resets (default 30000). */
+  maxDelayMs?: number
+  /** Consecutive failed attempts per outage before giving up for good (default 10). */
+  maxAttempts?: number
+}
+
+/** Defaults shared by the Config schema and {@link resolveReconnectPolicy}. */
+export const RECONNECT_DEFAULTS: Required<ReconnectConfig> = Object.freeze({
+  enabled: true,
+  initialDelayMs: 500,
+  maxDelayMs: 30_000,
+  maxAttempts: 10,
+})
+
+// The SDK's stdio transport owns two two-second termination grace periods.
+// Keep one additional second for the process-close event that proves the old
+// generation is gone; timing out fails closed instead of overlapping children.
+const GENERATION_CLOSE_TIMEOUT_MS = 5_000
+
+/** Fully resolved reconnect policy captured at plugin load. */
+export type ResolvedReconnectPolicy = Readonly<Required<ReconnectConfig>>
+
+/**
+ * The one explicit resolve step from raw reconnect config to the policy the
+ * supervisor runs. Programmatic construction may bypass Schemastery
+ * normalization, so every default and bound is re-judged here — misconfiguration
+ * fails the plugin instance at load.
+ *
+ * @param config - Raw `reconnect` config; omission uses the defaults.
+ * @param path - Diagnostic prefix naming the config location in thrown messages.
+ * @returns The frozen resolved policy.
+ */
+export function resolveReconnectPolicy(config: ReconnectConfig | undefined, path: string): ResolvedReconnectPolicy {
+  if (config !== undefined) {
+    for (const key of Object.keys(config)) {
+      if (!Object.hasOwn(RECONNECT_DEFAULTS, key)) throw new Error(`${path}.${key} is not a reconnect option`)
+    }
+  }
+  const enabled = config?.enabled ?? RECONNECT_DEFAULTS.enabled
+  const initialDelayMs = config?.initialDelayMs ?? RECONNECT_DEFAULTS.initialDelayMs
+  const maxDelayMs = config?.maxDelayMs ?? RECONNECT_DEFAULTS.maxDelayMs
+  const maxAttempts = config?.maxAttempts ?? RECONNECT_DEFAULTS.maxAttempts
+  /* jscpd:ignore-start — domain-specific delay validation parallels llm retry-policy; not extractable */
+  if (!Number.isFinite(initialDelayMs) || initialDelayMs <= 0 || initialDelayMs > MAX_TIMER_DELAY_MS) {
+    throw new Error(`${path}.initialDelayMs must be a positive finite number no greater than ${MAX_TIMER_DELAY_MS}`)
+  }
+  if (!Number.isFinite(maxDelayMs) || maxDelayMs <= 0 || maxDelayMs > MAX_TIMER_DELAY_MS) {
+    throw new Error(`${path}.maxDelayMs must be a positive finite number no greater than ${MAX_TIMER_DELAY_MS}`)
+  }
+  if (initialDelayMs > maxDelayMs) {
+    throw new Error(`${path}.initialDelayMs must be less than or equal to maxDelayMs`)
+  }
+  if (!Number.isInteger(maxAttempts) || maxAttempts < 1) {
+    throw new Error(`${path}.maxAttempts must be a positive integer`)
+  }
+  /* jscpd:ignore-end */
+  return Object.freeze({ enabled, initialDelayMs, maxDelayMs, maxAttempts })
+}
+
+/** Result from the initial connection attempt, for startup-await semantics. */
+export interface ConnectionOutcome {
+  /** If the initial connection or tool sync failed, the error; otherwise absent. */
+  error?: unknown
+}
+
+/** Handle for one plugin instance's supervised connection. */
+export interface ConnectionHandle {
+  /**
+   * Settles when the first connection attempt completes (success or failure).
+   * The supervisor enters its reconnect loop regardless; the caller decides
+   * whether a failed startup is fatal via `failOnStartupError`.
+   */
+  ready: Promise<ConnectionOutcome>
+  /**
+   * Stop reconnection, close the live client, wait for the in-flight attempt
+   * and queued tool syncs to quiesce, then unregister every tool this server
+   * still owns.
+   */
+  dispose(): Promise<void>
+}
+
+/**
+ * Start the supervised connection for one MCP server and keep it alive per
+ * the reconnect policy.
+ *
+ * @param ctx - Cordis context providing the `tools` registry and logger.
+ * @param config - Resolved plugin config selecting the transport and server identity.
+ * @param policy - Resolved reconnect policy from {@link resolveReconnectPolicy}.
+ * @returns Handle with a `ready` promise for startup-await and a `dispose` for teardown.
+ */
+export function startConnection(ctx: Context, config: Config, policy: ResolvedReconnectPolicy): ConnectionHandle {
+  const label = `mcp-client(${config.serverName})`
+  const opts: ToolBridgeOptions = {
+    registrationFailure: 'contain',
+    serverName: config.serverName,
+    toolCallTimeoutMs: config.toolCallTimeoutMs,
+  }
+  // The initial sync uses 'throw' when failOnStartupError is configured, so
+  // a registration conflict propagates to the startup-await path. Re-syncs
+  // and reconnect syncs always contain conflicts.
+  const startupOpts: ToolBridgeOptions = config.failOnStartupError
+    ? { ...opts, registrationFailure: 'throw' }
+    : opts
+
+  let disposed = false
+  /** Current generation: the connecting or connected client; undefined during backoff waits and after final failure. */
+  let client: Client | undefined
+  /** Close signal paired with {@link client}; captured by dispose before current ownership is cleared. */
+  let clientClosed: Promise<void> | undefined
+  /** Live tool registrations owned by this server; only {@link enqueueSync} and dispose swap it. */
+  let disposers: ToolDisposers = new Map()
+  let reconnectTimer: NodeJS.Timeout | undefined
+  /** Consecutive failed connection attempts within the current outage. */
+  let failedAttempts = 0
+  /** When the current generation finished connect + initial sync; undefined while down. */
+  let connectedAt: number | undefined
+  /** The real error from the first connection attempt, for startup-await diagnostics. */
+  let firstAttemptError: unknown
+
+  /** A generation may act only while it is the current one on a live plugin. */
+  const isCurrent = (generation: Client): boolean => !disposed && client === generation
+
+  /**
+   * Serializes every syncTools call — initial syncs and notification re-syncs
+   * across all generations — so two syncs can never interleave their
+   * dispose-previous/register-next swap (which would double-dispose one
+   * generation and leak another).
+   */
+  let syncChain: Promise<void> = Promise.resolve()
+  function enqueueSync(generation: Client, syncOpts: ToolBridgeOptions = opts): Promise<void> {
+    const run = syncChain.then(async () => {
+      if (!isCurrent(generation)) return
+      disposers = await syncTools(generation, ctx, syncOpts, disposers)
+    })
+    // The chain tail must survive a failed sync; the enqueuing caller owns reporting.
+    syncChain = run.catch(() => {})
+    return run
+  }
+
+  /** One disconnect decision per generation: the isCurrent guard makes racing close/error signals idempotent. */
+  function generationDown(generation: Client): void {
+    if (!isCurrent(generation)) return
+    client = undefined
+    clientClosed = undefined
+    scheduleReconnect()
+  }
+
+  /** Wait for the transport-owned close signal without letting a broken transport wedge teardown forever. */
+  function waitForClose(closed: Promise<void>): Promise<boolean> {
+    return new Promise((resolve) => {
+      const timeout = setTimeout(() => { resolve(false) }, GENERATION_CLOSE_TIMEOUT_MS)
+      timeout.unref()
+      void closed.then(() => {
+        clearTimeout(timeout)
+        resolve(true)
+      })
+    })
+  }
+
+  function scheduleReconnect(): void {
+    const lostEstablishedConnection = connectedAt !== undefined
+    if (!policy.enabled) {
+      const message = lostEstablishedConnection
+        ? 'connection lost and reconnect is disabled — registered tools will fail until an HMR reload or Host restart'
+        : 'connection failed and reconnect is disabled — no tools were registered; reload the plugin or restart the Host to connect'
+      ctx.logger.error(`${label}: ${message}`)
+      return
+    }
+    // A connection that stayed up past the stability window (= maxDelayMs, the
+    // longest backoff spacing) ended the previous outage: start a fresh budget.
+    if (connectedAt !== undefined && Date.now() - connectedAt >= policy.maxDelayMs) failedAttempts = 0
+    connectedAt = undefined
+    failedAttempts += 1
+    if (failedAttempts > policy.maxAttempts) {
+      // Enqueue the give-up disposal so it cannot race an in-flight sync's
+      // phase-2 swap (which checks isCurrent inside the queue).
+      syncChain = syncChain.then(() => {
+        for (const dispose of disposers.values()) dispose()
+        disposers = new Map()
+      })
+      ctx.logger.error(`${label}: giving up after ${policy.maxAttempts} consecutive failed reconnect attempts — tools unregistered; reload the plugin or restart the Host to reconnect`)
+      return
+    }
+    const delayMs = Math.min(policy.maxDelayMs, policy.initialDelayMs * 2 ** (failedAttempts - 1))
+    const action = lostEstablishedConnection ? 'connection lost; reconnecting' : 'connection failed; retrying'
+    ctx.logger.warn(`${label}: ${action} in ${delayMs}ms (attempt ${failedAttempts}/${policy.maxAttempts})`)
+    reconnectTimer = setTimeout(() => {
+      reconnectTimer = undefined
+      settling = connectGeneration(false)
+    }, delayMs)
+    // An armed reconnect timer must never hold the process open on its own.
+    reconnectTimer.unref()
+  }
+
+  /**
+   * One connection attempt: fresh transport + client (the MCP SDK binds a
+   * Protocol to one transport for life), connect, then queue the initial tool
+   * sync. The startup flag belongs to the attempt rather than the shared sync
+   * queue, so an early notification cannot consume strict startup semantics.
+   * Every failure funnels through {@link generationDown}; success arms the
+   * onclose-driven disconnect path. Never rejects.
+   *
+   * @param startup - Whether this is the plugin's activation attempt.
+   */
+  async function connectGeneration(startup: boolean): Promise<void> {
+    const generation = new Client(
+      { name: 'dsh-mcp-client', version: '0.0.1' },
+      { capabilities: {} },
+    )
+    const closed: PromiseWithResolvers<void> = Promise.withResolvers()
+    let attemptSettled = false
+    let closeObserved = false
+    const hasClosed = (): boolean => closeObserved
+    client = generation
+    clientClosed = closed.promise
+    generation.onclose = () => {
+      closeObserved = true
+      closed.resolve()
+      // A failed connect owns its close barrier in the catch path below. An
+      // established generation can transition down directly from this signal.
+      if (attemptSettled) generationDown(generation)
+    }
+    // Registered before connect so a list change during the initial sync is
+    // queued behind it rather than dropped.
+    generation.setNotificationHandler(
+      ToolListChangedNotificationSchema,
+      async () => {
+        if (!isCurrent(generation)) return
+        ctx.logger.info(`${label}: tool list changed, re-syncing`)
+        try {
+          await enqueueSync(generation)
+        } catch (error) {
+          // Fetch-phase failure: the previous generation is still registered
+          // and `disposers` still owns it — keep serving the last good list.
+          if (!disposed) ctx.logger.error(`${label}: tool re-sync failed: ${String(error)}`)
+        }
+      },
+    )
+    try {
+      await generation.connect(createTransport(config))
+      if (hasClosed()) {
+        attemptSettled = true
+        generationDown(generation)
+        return
+      }
+      await enqueueSync(generation, startup ? startupOpts : opts)
+    } catch (error) {
+      if (firstAttemptError === undefined) firstAttemptError = error
+      // Disposal clears current ownership before it closes the generation, so
+      // only a live supervisor reports an attempt failure.
+      if (isCurrent(generation)) ctx.logger.warn(`${label}: connection attempt failed: ${String(error)}`)
+      try { await generation.close() } catch { /* transport already gone */ }
+      const quiesced = hasClosed() || await waitForClose(closed.promise)
+      attemptSettled = true
+      if (!isCurrent(generation)) return
+      if (!quiesced) {
+        client = undefined
+        clientClosed = undefined
+        ctx.logger.error(`${label}: failed generation did not close within ${GENERATION_CLOSE_TIMEOUT_MS}ms — reconnect stopped to avoid overlapping server processes; reload the plugin or restart the Host to retry`)
+        return
+      }
+      generationDown(generation)
+      return
+    }
+    attemptSettled = true
+    if (hasClosed()) {
+      generationDown(generation)
+      return
+    }
+    if (!isCurrent(generation)) return
+    connectedAt = Date.now()
+    if (failedAttempts > 0) ctx.logger.info(`${label}: reconnected and re-synced tools (attempt ${failedAttempts}/${policy.maxAttempts})`)
+  }
+
+  /** The in-flight (or last settled) connection attempt; dispose awaits it for quiescence. */
+  let settling = connectGeneration(true)
+
+  // The ready promise settles when the first attempt finishes (regardless of
+  // success). If the first attempt fails and reconnect is enabled, the
+  // supervisor is already scheduling a retry — ready just reports the outcome.
+  const ready: Promise<ConnectionOutcome> = settling.then(() => {
+    // After settling: if client is set the initial connect+sync succeeded.
+    // If not, the supervisor either scheduled a retry (error logged) or gave
+    // up (error logged). Either way the outcome is reported with the real error.
+    // Note: settling.then() is a microtask; stdio onclose is a macrotask — so
+    // a server that crashes AFTER a successful initial sync cannot flip client
+    // to undefined before this continuation runs.
+    if (client !== undefined) return {}
+    /* v8 ignore next -- defensive: firstAttemptError is always set when connect/sync fails */
+    return { error: firstAttemptError ?? new Error(`${label}: initial connection failed`) }
+  })
+
+  return {
+    ready,
+    async dispose(): Promise<void> {
+      disposed = true
+      if (reconnectTimer !== undefined) {
+        clearTimeout(reconnectTimer)
+        reconnectTimer = undefined
+      }
+      const current = client
+      const currentClosed = clientClosed
+      client = undefined
+      clientClosed = undefined
+      if (current !== undefined) {
+        try { await current.close() } catch { /* transport already gone */ }
+        if (currentClosed !== undefined && !await waitForClose(currentClosed)) {
+          ctx.logger.error(`${label}: generation did not close within ${GENERATION_CLOSE_TIMEOUT_MS}ms during disposal — server shutdown may be incomplete`)
+        }
+      }
+      // Quiesce, don't just request it: the in-flight attempt enqueues its
+      // sync before settling, so awaiting both leaves `disposers` final.
+      await settling
+      await syncChain
+      for (const dispose of disposers.values()) dispose()
+      disposers = new Map()
+    },
+  }
+}
--- a/packages/mcp/mcp-client/src/index.ts
+++ b/packages/mcp/mcp-client/src/index.ts
@@ -15,14 +15,14 @@

 import type { Context } from '@deepseek-ai/cordis'
 import z from '@deepseek-ai/schemastery'
-import { Client } from '@modelcontextprotocol/sdk/client/index.js'
-import { ToolListChangedNotificationSchema } from '@modelcontextprotocol/sdk/types.js'
-import { createTransport } from './transport.ts'
-import { syncTools } from './tools.ts'
+import { MAX_TIMER_DELAY_MS } from '@deepseek-ai/dsh-timeout'
+import { RECONNECT_DEFAULTS, resolveReconnectPolicy, startConnection } from './connection.ts'
+import type { ReconnectConfig } from './connection.ts'
 // Side-effect type import: declaration-merges `ctx.tools` onto Context.
 import type {} from '@deepseek-ai/dsh-tools'

 export type { McpResult } from './tools.ts'
+export type { ReconnectConfig, ResolvedReconnectPolicy } from './connection.ts'

 /** Cordis plugin name used by loader diagnostics. */
 export const name = 'mcp-client'
@@ -68,6 +68,8 @@ export interface StdioConfig {
  toolCallTimeoutMs: number
  /** Fail plugin activation when the initial connection or tool synchronization fails. */
  failOnStartupError: boolean
+  /** Automatic reconnect policy after a lost connection; omission uses the defaults. */
+  reconnect?: ReconnectConfig
 }

 /** Config for connecting to an MCP server over Streamable HTTP (SSE). */
@@ -88,11 +90,20 @@ export interface StreamableHttpConfig {
  toolCallTimeoutMs: number
  /** Fail plugin activation when the initial connection or tool synchronization fails. */
  failOnStartupError: boolean
+  /** Automatic reconnect policy after a lost connection; omission uses the defaults. */
+  reconnect?: ReconnectConfig
 }

 /** Configuration for one stdio or Streamable HTTP MCP server. */
 export type Config = StdioConfig | StreamableHttpConfig

+const Reconnect: z<ReconnectConfig> = z.object({
+  enabled: z.boolean().default(RECONNECT_DEFAULTS.enabled),
+  initialDelayMs: z.number().min(1).max(MAX_TIMER_DELAY_MS).default(RECONNECT_DEFAULTS.initialDelayMs),
+  maxDelayMs: z.number().min(1).max(MAX_TIMER_DELAY_MS).default(RECONNECT_DEFAULTS.maxDelayMs),
+  maxAttempts: z.number().step(1).min(1).max(Number.MAX_SAFE_INTEGER).default(RECONNECT_DEFAULTS.maxAttempts),
+})
+
 export const Config = z.union([
  z.object({
    transport: z.const('stdio'),
@@ -103,6 +114,7 @@ export const Config = z.union([
    cwd: z.string().default(''),
    toolCallTimeoutMs: z.number().default(DEFAULT_TOOL_CALL_TIMEOUT_MS),
    failOnStartupError: z.boolean().default(false),
+    reconnect: Reconnect,
  }),
  z.object({
    transport: z.const('streamable-http'),
@@ -111,6 +123,7 @@ export const Config = z.union([
    headers: z.dict(String).default({}),
    toolCallTimeoutMs: z.number().default(DEFAULT_TOOL_CALL_TIMEOUT_MS),
    failOnStartupError: z.boolean().default(false),
+    reconnect: Reconnect,
  }),
 ]) as unknown as z<Config>

@@ -125,7 +138,12 @@ export const Config = z.union([
 * @returns startup readiness after connection and initial tool discovery settle.
 */
 export async function apply(ctx: Context, config: Config): Promise<void> {
-  // Reserve the namespace first: a duplicate `serverName` fails THIS instance
+  // Fail loud at load: reconnect misconfiguration (including programmatic
+  // construction that bypassed Schemastery) rejects THIS instance before any
+  // effect registers.
+  const reconnect = resolveReconnectPolicy(config.reconnect, `mcp-client(${config.serverName}): reconnect`)
+
+  // Reserve the namespace next: a duplicate `serverName` fails THIS instance
  // at load with an actionable error and leaves the earlier instance intact.
  ctx.effect(() => {
    let names = activeServerNames.get(ctx.root)
@@ -142,58 +160,22 @@ export async function apply(ctx: Context, config: Config): Promise<void> {
    return () => void names.delete(config.serverName)
  }, 'mcp-client.serverName')

-  const transport = createTransport(config)
-  const client = new Client(
-    { name: 'dsh-mcp-client', version: '0.0.1' },
-    { capabilities: {} },
-  )
+  // The supervisor owns the client/transport generations, the reconnect
+  // loop, and the live tool registrations; disposal stops reconnection,
+  // quiesces in-flight work, and unregisters the current generation.
+  const connection = startConnection(ctx, config, reconnect)

-  const opts = {
-    registrationFailure: 'contain' as const,
-    serverName: config.serverName,
-    toolCallTimeoutMs: config.toolCallTimeoutMs,
-  }
-
-  // Connect and set up tools. `ready` always settles to an outcome so rollback
-  // can close a partially opened client even when strict startup later rejects.
-  // Its accessor returns the CURRENT disposer generation, so disposal always
-  // unregisters the live set, not the first one.
-  const ready = (async () => {
-    await client.connect(transport)
-
-    let disposers = await syncTools(client, ctx, {
-      ...opts,
-      registrationFailure: config.failOnStartupError ? 'throw' : 'contain',
-    }, new Map())
-
-    client.setNotificationHandler(
-      ToolListChangedNotificationSchema,
-      async () => {
-        ctx.logger.info(`mcp-client(${config.serverName}): tool list changed, re-syncing`)
-        try {
-          disposers = await syncTools(client, ctx, opts, disposers)
-        } catch (error) {
-          // Fetch-phase failure: the previous generation is still registered
-          // and `disposers` still owns it — keep serving the last good list.
-          ctx.logger.error(`mcp-client(${config.serverName}): tool re-sync failed: ${String(error)}`)
-        }
-      },
-    )
-
-    return { getDisposers: () => disposers }
-  })().catch((error: unknown) => {
-    ctx.logger.error(`mcp-client(${config.serverName}): startup failed: ${String(error)}`)
-    return { getDisposers: () => new Map<string, () => void>(), error }
-  })
-
-  ctx.effect(() => async () => {
-    const outcome = await ready
-    for (const dispose of outcome.getDisposers().values()) dispose()
-    try { await client.close() } catch { /* transport already gone */ }
+  ctx.effect(() => {
+    return () => connection.dispose()
  }, 'mcp-client.connection')

-  const outcome = await ready
-  if ('error' in outcome && config.failOnStartupError) {
+  // Block plugin activation on the initial connection + tool discovery so
+  // Cordis consumers observe the tools immediately after the fiber activates.
+  // When failOnStartupError is true, a failed initial attempt rejects the
+  // fiber (Cordis rolls it back); otherwise the error is logged and the
+  // supervisor enters its reconnect loop.
+  const outcome = await connection.ready
+  if (outcome.error !== undefined && config.failOnStartupError) {
    throw new Error(`mcp-client(${config.serverName}): initial connection or tool synchronization failed`, { cause: outcome.error })
  }
 }
--- a/packages/mcp/mcp-client/tests/apply.spec.ts
+++ b/packages/mcp/mcp-client/tests/apply.spec.ts
@@ -124,6 +124,33 @@ describe('mcp-client plugin module exports', () => {
    } as never)
    expect(resolved.serverName).toBe('github-prod_1')
  })
+
+  it('Config schema materializes reconnect defaults and merges partial overrides', () => {
+    const omitted = ConfigSchema({
+      transport: 'stdio',
+      serverName: 'srv',
+      command: 'echo',
+    } as never)
+    expect(omitted.reconnect).toEqual({ enabled: true, initialDelayMs: 500, maxDelayMs: 30_000, maxAttempts: 10 })
+
+    const partial = ConfigSchema({
+      transport: 'stdio',
+      serverName: 'srv',
+      command: 'echo',
+      reconnect: { initialDelayMs: 100 },
+    } as never)
+    expect(partial.reconnect).toEqual({ enabled: true, initialDelayMs: 100, maxDelayMs: 30_000, maxAttempts: 10 })
+  })
+
+  it('Config schema rejects an invalid reconnect block', () => {
+    // schemastery unions wrap branch errors, so assert the throw only.
+    expect(() => ConfigSchema({
+      transport: 'stdio',
+      serverName: 'srv',
+      command: 'echo',
+      reconnect: { maxAttempts: 0 },
+    } as never)).toThrow()
+  })
 })

 describe('apply (plugin lifecycle)', () => {
@@ -132,7 +159,10 @@ describe('apply (plugin lifecycle)', () => {
  beforeEach(async () => {
    vi.clearAllMocks()
    mockConnect.mockResolvedValue(undefined)
-    mockClose.mockResolvedValue(undefined)
+    mockClose.mockImplementation(function (this: { onclose?: () => void }) {
+      this.onclose?.()
+      return Promise.resolve()
+    })
    mockListTools.mockResolvedValue({
      tools: [{ name: 'remote', description: 'A remote tool', inputSchema: { type: 'object' } }],
      nextCursor: undefined,
@@ -216,19 +246,23 @@ describe('apply (plugin lifecycle)', () => {
    expect(mockListTools).not.toHaveBeenCalled()
    expect(ctx.tools.get('mcp__srv__remote')).toBeUndefined()

-    // Disposal exercises the empty fallback accessor: nothing to unregister,
-    // close still attempted, no throw.
+    // Disposal cancels the scheduled reconnect attempt: nothing to
+    // unregister, close already attempted by the failed attempt, no throw.
    await ctx.fiber.dispose()
    await sleep(50)
    expect(mockClose).toHaveBeenCalled()
  })

  it('rejects activation and still closes the client when startup failure is configured as fatal', async () => {
-    mockConnect.mockRejectedValue(new Error('connection refused'))
+    const cause = new Error('connection refused')
+    mockConnect.mockRejectedValue(cause)
    await expect(apply(ctx, {
      ...stdioConfig,
      failOnStartupError: true,
-    })).rejects.toThrow('initial connection or tool synchronization failed')
+    })).rejects.toMatchObject({
+      message: 'mcp-client(srv): initial connection or tool synchronization failed',
+      cause,
+    })

    expect(mockListTools).not.toHaveBeenCalled()
    expect(ctx.tools.get('mcp__srv__remote')).toBeUndefined()
@@ -258,6 +292,32 @@ describe('apply (plugin lifecycle)', () => {
    expect(mockClose).toHaveBeenCalled()
  })

+  it('preserves strict startup registration when list_changed arrives before connect resolves', async () => {
+    ctx.tools.register({
+      name: 'mcp__srv__remote',
+      description: 'Foreign squatter',
+      parameters: { type: 'object' },
+      output: {
+        schema: { type: 'string' },
+        render: (_args, value) => [{ type: 'text', text: value as string }],
+      },
+      execute: async () => 'foreign',
+    })
+    mockConnect.mockImplementation(async () => {
+      const handler = mockSetNotificationHandler.mock.calls[0]![1] as () => Promise<void>
+      await handler()
+    })
+
+    await expect(apply(ctx, {
+      ...stdioConfig,
+      failOnStartupError: true,
+    })).rejects.toThrow('initial connection or tool synchronization failed')
+
+    expect(mockListTools).toHaveBeenCalledTimes(2)
+    expect(ctx.tools.get('mcp__srv__remote')?.description).toBe('Foreign squatter')
+    await ctx.fiber.dispose()
+  })
+
  it('re-syncs tools on ToolListChanged notification', async () => {
    await apply(ctx, stdioConfig)

@@ -311,7 +371,10 @@ describe('apply (plugin lifecycle)', () => {
  })

  it('effect disposer handles client.close failure gracefully', async () => {
-    mockClose.mockRejectedValue(new Error('already closed'))
+    mockClose.mockImplementation(function (this: { onclose?: () => void }) {
+      this.onclose?.()
+      return Promise.reject(new Error('already closed'))
+    })

    await apply(ctx, stdioConfig)

--- a/packages/mcp/mcp-client/tests/fixture-server.ts
+++ b/packages/mcp/mcp-client/tests/fixture-server.ts
@@ -51,6 +51,17 @@ server.registerTool('image', {
  ],
 }))

+server.registerTool('crash', {
+  title: 'Crash Tool',
+  description: 'Replies, then exits the server process (crash-recovery test).',
+  inputSchema: {},
+}, async () => {
+  // Exit AFTER the response flushes so the caller observes a clean result
+  // followed by a transport close, like a real post-reply crash.
+  setTimeout(() => process.exit(7), 25)
+  return { content: [{ type: 'text', text: 'crashing' }] }
+})
+
 // Dotted name: legal in MCP, illegal in the DeepSeek function-name contract.
 // Exercises the bridge's normalize-and-hash public-name path end to end.
 server.registerTool('admin.reset', {
--- a/packages/mcp/mcp-client/tests/mcp-client.e2e.ts
+++ b/packages/mcp/mcp-client/tests/mcp-client.e2e.ts
@@ -13,7 +13,7 @@ import { mkdtemp, rm, writeFile, readFile } from 'node:fs/promises'
 import { tmpdir } from 'node:os'
 import { join } from 'node:path'
 import { fileURLToPath } from 'node:url'
-import { afterAll, beforeAll, describe, expect, it } from 'vitest'
+import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest'
 import { Context } from '@deepseek-ai/cordis'
 import { McpServer } from '@modelcontextprotocol/sdk/server/mcp.js'
 import { StreamableHTTPServerTransport } from '@modelcontextprotocol/sdk/server/streamableHttp.js'
@@ -200,6 +200,87 @@ describe('fixture server — disposal', () => {
  }, 30_000)
 })

+describe('fixture server — crash recovery', () => {
+  function crashConfig(serverName: string, reconnect: NonNullable<Config['reconnect']>): Config {
+    return {
+      transport: 'stdio',
+      serverName,
+      command: process.execPath,
+      args: [fixtureServerPath],
+      env: {},
+      cwd: packageDir,
+      toolCallTimeoutMs: 15_000,
+      failOnStartupError: false,
+      reconnect,
+    }
+  }
+
+  it('auto-reconnects after a stdio crash and serves tool calls again', async () => {
+    const ctx = await mountRegistry()
+    await apply(ctx, crashConfig('crashy', { initialDelayMs: 50, maxDelayMs: 500, maxAttempts: 40 }))
+
+    const before = await ctx.tools.execute({
+      signal: testToolSignal,
+      callId: nextCallId(), name: 'mcp__crashy__add', arguments: { a: 2, b: 3 },
+    })
+    expect(textOf(before.content[0])).toBe('5')
+
+    // The crash tool replies, then kills the real child process.
+    const crash = await ctx.tools.execute({
+      signal: testToolSignal,
+      callId: nextCallId(), name: 'mcp__crashy__crash', arguments: {},
+    })
+    expect(crash.isError).toBe(false)
+
+    // Recovery is proven by the world: a post-crash call round-trips through
+    // the respawned server process.
+    await vi.waitFor(async () => {
+      const after = await ctx.tools.execute({
+        signal: testToolSignal,
+        callId: nextCallId(), name: 'mcp__crashy__add', arguments: { a: 20, b: 22 },
+      })
+      expect(after.isError).toBe(false)
+      expect(textOf(after.content[0])).toBe('42')
+    }, { timeout: 15_000, interval: 250 })
+
+    // The recovered generation replaced the dead one: no duplicates, no leak.
+    const addEntries = ctx.tools.schemas().map(s => s.name).filter(name => name === 'mcp__crashy__add')
+    expect(addEntries).toHaveLength(1)
+
+    await ctx.fiber.dispose()
+    await sleep(200)
+  }, 30_000)
+
+  it('plugin unload during an outage stops reconnection and unregisters tools', async () => {
+    const ctx = await mountRegistry()
+    const fiber = ctx.plugin(
+      { name: 'mcp-client', inject: ['tools'], apply },
+      crashConfig('ephemeral', { initialDelayMs: 8_000, maxDelayMs: 8_000, maxAttempts: 5 }),
+    )
+    // Cordis awaits async apply() as startup work; wait for it.
+    await vi.waitFor(() => { expect(ctx.tools.get('mcp__ephemeral__add')).toBeDefined() }, { timeout: 20_000 })
+
+    const crash = await ctx.tools.execute({
+      signal: testToolSignal,
+      callId: nextCallId(), name: 'mcp__ephemeral__crash', arguments: {},
+    })
+    expect(crash.isError).toBe(false)
+
+    // Give the transport close a moment to land the supervisor in its 8s
+    // backoff wait, then unload: disposal must not sit out the backoff.
+    await sleep(300)
+    const started = Date.now()
+    await fiber.dispose()
+    expect(Date.now() - started).toBeLessThan(4_000)
+
+    expect(ctx.tools.get('mcp__ephemeral__add')).toBeUndefined()
+    await sleep(200)
+    expect(ctx.tools.get('mcp__ephemeral__add')).toBeUndefined()
+
+    await ctx.fiber.dispose()
+  }, 30_000)
+})
+
 // ---- @modelcontextprotocol/server-everything ----

 describe('server-everything — official test server', () => {
--- a/packages/mcp/mcp-client/tests/reconnect.spec.ts
+++ b/packages/mcp/mcp-client/tests/reconnect.spec.ts
@@ -0,0 +1,521 @@
+/**
+ * Tests for the mcp-client connection supervisor: crash-driven reconnection
+ * with bounded backoff, generation-safe tool re-registration, the failure
+ * cap, the stability-window budget reset, and disposal stopping reconnection.
+ * Isolated file so vi.mock of the MCP SDK doesn't pollute other test suites.
+ */
+import { describe, expect, it, vi, beforeEach } from 'vitest'
+import { Context } from '@deepseek-ai/cordis'
+import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
+import ToolRegistry from '@deepseek-ai/dsh-tools'
+import { CallId } from '@deepseek-ai/dsh-llm'
+import type { Config } from '@deepseek-ai/dsh-mcp-client'
+
+// ---- Mock MCP SDK ----
+
+// vi.mock factories are hoisted above every import/const, so the mock fns and
+// class must be created inside vi.hoisted to exist when the factories run.
+const { mockConnect, mockClose, mockListTools, mockCallTool, mockSetNotificationHandler, MockClient, instances } = vi.hoisted(() => {
+  const mockConnect = vi.fn<() => Promise<void>>()
+  const mockClose = vi.fn<() => Promise<void>>()
+  const mockListTools = vi.fn<(_params?: Record<string, unknown>) => Promise<unknown>>()
+  const mockCallTool = vi.fn<(
+    _params?: Record<string, unknown>, _compatibilitySchema?: unknown, _options?: unknown,
+  ) => Promise<unknown>>()
+  const mockSetNotificationHandler = vi.fn()
+  const mockRequest = vi.fn(async (
+    request: { method: string; params?: Record<string, unknown> },
+    _schema: unknown,
+    options?: unknown,
+  ): Promise<unknown> => {
+    if (request.method === 'tools/list') return await mockListTools(request.params)
+    if (request.method === 'tools/call') return await mockCallTool(request.params, undefined, options)
+    throw new Error(`unexpected MCP request: ${request.method}`)
+  })
+  class MockClient {
+    onclose: (() => void) | undefined
+    connect = mockConnect
+    close = mockClose
+    request = mockRequest
+    setNotificationHandler = mockSetNotificationHandler
+    constructor() { instances.push(this) }
+  }
+  const instances: MockClient[] = []
+  return { mockConnect, mockClose, mockListTools, mockCallTool, mockSetNotificationHandler, MockClient, instances }
+})
+
+vi.mock('@modelcontextprotocol/sdk/client/index.js', () => ({
+  Client: MockClient,
+}))
+
+vi.mock('@modelcontextprotocol/sdk/client/stdio.js', () => ({
+  StdioClientTransport: vi.fn(),
+}))
+
+vi.mock('@modelcontextprotocol/sdk/client/streamableHttp.js', () => ({
+  StreamableHTTPClientTransport: vi.fn(),
+}))
+
+// vi.mock is hoisted above static imports, so the modules under test see the
+// mocked SDK even through a static import.
+import { apply } from '@deepseek-ai/dsh-mcp-client/src/index.ts'
+import { RECONNECT_DEFAULTS, resolveReconnectPolicy, startConnection } from '@deepseek-ai/dsh-mcp-client/src/connection.ts'
+
+// ---- Helpers ----
+
+const testToolSignal = new AbortController().signal
+
+async function mountRegistry(): Promise<Context> {
+  const ctx = new Context()
+  await ctx.plugin(SystemPrompt)
+  await ctx.plugin(ToolRegistry)
+  return ctx
+}
+
+function sleep(ms: number): Promise<void> {
+  // Annotated binding (not withResolvers<void>()): the tests lint layer runs
+  // no-invalid-void-type with default options, which rejects the explicit
+  // type argument in call position but accepts the inferred form.
+  const gate: PromiseWithResolvers<void> = Promise.withResolvers()
+  setTimeout(gate.resolve, ms)
+  return gate.promise
+}
+
+/** Capture the supervisor's logger lines by level on one context. */
+function captureLogs(ctx: Context): { warns: string[]; errors: string[]; infos: string[] } {
+  const warns: string[] = []
+  const errors: string[] = []
+  const infos: string[] = []
+  ctx.logger.warn = ((message: unknown) => { warns.push(String(message)) }) as typeof ctx.logger.warn
+  ctx.logger.error = ((message: unknown) => { errors.push(String(message)) }) as typeof ctx.logger.error
+  ctx.logger.info = ((message: unknown) => { infos.push(String(message)) }) as typeof ctx.logger.info
+  return { warns, errors, infos }
+}
+
+function stdioConfig(reconnect?: Config['reconnect']): Config {
+  return {
+    transport: 'stdio',
+    serverName: 'srv',
+    command: 'echo',
+    args: [],
+    env: {},
+    cwd: '',
+    toolCallTimeoutMs: 60_000,
+    failOnStartupError: false,
+    ...reconnect === undefined ? {} : { reconnect },
+  }
+}
+
+/** The tool list the mock server advertises after a successful (re)connect. */
+function listing(...names: string[]): { tools: { name: string; inputSchema: { type: string } }[]; nextCursor: undefined } {
+  return {
+    tools: names.map(name => ({ name, inputSchema: { type: 'object' } })),
+    nextCursor: undefined,
+  }
+}
+
+let callSeq = 0
+function nextCallId(): CallId {
+  return CallId(`reconnect-${++callSeq}`)
+}
+
+// ---- Tests ----
+
+describe('reconnect supervisor', () => {
+  let ctx: Context
+
+  beforeEach(async () => {
+    vi.clearAllMocks()
+    instances.length = 0
+    mockConnect.mockResolvedValue(undefined)
+    mockClose.mockImplementation(function (this: { onclose?: () => void }) {
+      this.onclose?.()
+      return Promise.resolve()
+    })
+    mockListTools.mockResolvedValue(listing('remote'))
+    mockCallTool.mockResolvedValue({ content: [{ type: 'text', text: 'ok' }] })
+    ctx = await mountRegistry()
+  })
+
+  it('reconnects after a transport close, re-syncs tools through the new generation, and serves calls', async () => {
+    const { warns, infos } = captureLogs(ctx)
+    await apply(ctx, stdioConfig({ initialDelayMs: 5, maxDelayMs: 40, maxAttempts: 5 }))
+    await vi.waitFor(() => { expect(ctx.tools.get('mcp__srv__remote')).toBeDefined() })
+    expect(instances).toHaveLength(1)
+
+    // The recovered server advertises a different list: the swap must neither
+    // duplicate nor leak the pre-crash generation.
+    mockListTools.mockResolvedValue(listing('revived'))
+    instances[0]!.onclose?.()
+
+    await vi.waitFor(() => { expect(ctx.tools.get('mcp__srv__revived')).toBeDefined() })
+    expect(ctx.tools.get('mcp__srv__remote')).toBeUndefined()
+    expect(instances).toHaveLength(2)
+    expect(mockConnect).toHaveBeenCalledTimes(2)
+
+    // Post-recovery calls execute through the re-registered definition.
+    const result = await ctx.tools.execute({
+      signal: testToolSignal,
+      callId: nextCallId(), name: 'mcp__srv__revived', arguments: {},
+    })
+    expect(result.isError).toBe(false)
+
+    // User-visible state: reconnecting and recovered are distinct lines.
+    expect(warns.some(line => line.includes('reconnecting in 5ms (attempt 1/5)'))).toBe(true)
+    expect(infos.some(line => line.includes('reconnected and re-synced tools'))).toBe(true)
+
+    // A late close signal from the replaced generation is ignored.
+    instances[0]!.onclose?.()
+    await sleep(30)
+    expect(instances).toHaveLength(2)
+  })
+
+  it('stops at the failure cap, unregisters the tools, and reports final failure', async () => {
+    const { warns, errors } = captureLogs(ctx)
+    await apply(ctx, stdioConfig({ initialDelayMs: 2, maxDelayMs: 8, maxAttempts: 2 }))
+    await vi.waitFor(() => { expect(ctx.tools.get('mcp__srv__remote')).toBeDefined() })
+
+    mockConnect.mockRejectedValue(new Error('server gone'))
+    // A failing close on the failed attempt's cleanup must not break the loop.
+    mockClose.mockImplementation(function (this: { onclose?: () => void }) {
+      this.onclose?.()
+      return Promise.reject(new Error('already closed'))
+    })
+    instances[0]!.onclose?.()
+
+    await vi.waitFor(() => {
+      expect(errors.some(line => line.includes('giving up after 2 consecutive failed reconnect attempts'))).toBe(true)
+    })
+    // Stale tools do not leak past final failure.
+    expect(ctx.tools.get('mcp__srv__remote')).toBeUndefined()
+    // Initial connect + exactly maxAttempts reconnect attempts.
+    expect(mockConnect).toHaveBeenCalledTimes(3)
+    expect(warns.some(line => line.includes('connection attempt failed: Error: server gone'))).toBe(true)
+    expect(warns.some(line => line.includes('connection failed; retrying in 4ms (attempt 2/2)'))).toBe(true)
+    await sleep(30)
+    expect(mockConnect).toHaveBeenCalledTimes(3)
+  })
+
+  it('gives up behind an in-flight re-sync and removes the generation it publishes', async () => {
+    const { errors } = captureLogs(ctx)
+    await apply(ctx, stdioConfig({ initialDelayMs: 2, maxDelayMs: 8, maxAttempts: 1 }))
+    await vi.waitFor(() => { expect(ctx.tools.get('mcp__srv__remote')).toBeDefined() })
+
+    const gate: PromiseWithResolvers<unknown> = Promise.withResolvers()
+    mockListTools.mockImplementation(() => gate.promise)
+    const handler = mockSetNotificationHandler.mock.calls[0]![1] as () => Promise<void>
+    const resync = handler()
+    await vi.waitFor(() => { expect(mockListTools).toHaveBeenCalledTimes(2) })
+
+    mockConnect.mockRejectedValue(new Error('server gone'))
+    instances[0]!.onclose?.()
+    await vi.waitFor(() => {
+      expect(errors.some(line => line.includes('giving up after 1 consecutive failed reconnect attempts'))).toBe(true)
+    })
+
+    gate.resolve(listing('late'))
+    await resync
+    await vi.waitFor(() => {
+      expect(ctx.tools.get('mcp__srv__remote')).toBeUndefined()
+      expect(ctx.tools.get('mcp__srv__late')).toBeUndefined()
+    })
+    expect(mockConnect).toHaveBeenCalledTimes(2)
+  })
+
+  it('does not start a replacement until a failed generation reports that it closed', async () => {
+    const { warns } = captureLogs(ctx)
+    mockConnect.mockRejectedValueOnce(new Error('initialize failed'))
+    // Model the SDK's fire-and-forget close after initialize fails: the
+    // harness's second close call returns, but the child has not exited yet.
+    mockClose.mockResolvedValue(undefined)
+
+    const applying = apply(ctx, stdioConfig({ initialDelayMs: 2, maxDelayMs: 8, maxAttempts: 2 }))
+    await vi.waitFor(() => { expect(mockClose).toHaveBeenCalled() })
+    await sleep(30)
+    expect(instances).toHaveLength(1)
+
+    instances[0]!.onclose?.()
+    await applying
+    await vi.waitFor(() => { expect(instances).toHaveLength(2) })
+    expect(warns.some(line => line.includes('connection failed; retrying in 2ms (attempt 1/2)'))).toBe(true)
+  })
+
+  it('stops reconnecting when a failed generation never reports that it closed', async () => {
+    vi.useFakeTimers()
+    try {
+      const { errors } = captureLogs(ctx)
+      mockConnect.mockRejectedValue(new Error('initialize failed'))
+      mockClose.mockResolvedValue(undefined)
+
+      const applying = apply(ctx, stdioConfig({ initialDelayMs: 2, maxDelayMs: 8, maxAttempts: 2 }))
+      await vi.advanceTimersByTimeAsync(5_000)
+      await applying
+
+      expect(instances).toHaveLength(1)
+      expect(errors.some(line => line.includes('reconnect stopped to avoid overlapping server processes'))).toBe(true)
+    } finally {
+      vi.useRealTimers()
+    }
+  })
+
+  it('suppresses retry reporting when disposal owns a pending connect rejection', async () => {
+    const { warns } = captureLogs(ctx)
+    const gate: PromiseWithResolvers<void> = Promise.withResolvers()
+    mockConnect.mockImplementation(() => gate.promise)
+    const handle = startConnection(ctx, stdioConfig(), resolveReconnectPolicy(undefined, 'reconnect'))
+    await vi.waitFor(() => { expect(instances).toHaveLength(1) })
+
+    const disposing = handle.dispose()
+    gate.reject(new Error('disposed connect'))
+    await disposing
+    await handle.ready
+
+    expect(warns.some(line => line.includes('connection attempt failed'))).toBe(false)
+    expect(instances).toHaveLength(1)
+  })
+
+  it('bounds disposal while a resolving generation never reports that it closed', async () => {
+    vi.useFakeTimers()
+    try {
+      const { errors } = captureLogs(ctx)
+      const gate: PromiseWithResolvers<void> = Promise.withResolvers()
+      mockConnect.mockImplementation(() => gate.promise)
+      mockClose.mockResolvedValue(undefined)
+      const handle = startConnection(ctx, stdioConfig(), resolveReconnectPolicy(undefined, 'reconnect'))
+      await vi.advanceTimersByTimeAsync(0)
+
+      const disposing = handle.dispose()
+      await vi.advanceTimersByTimeAsync(5_000)
+      gate.resolve()
+      await disposing
+
+      expect(mockListTools).not.toHaveBeenCalled()
+      expect(errors.some(line => line.includes('server shutdown may be incomplete'))).toBe(true)
+    } finally {
+      vi.useRealTimers()
+    }
+  })
+
+  it('dispose during the backoff wait cancels the pending reconnect', async () => {
+    await apply(ctx, stdioConfig({ initialDelayMs: 60_000, maxDelayMs: 60_000, maxAttempts: 5 }))
+    await vi.waitFor(() => { expect(ctx.tools.get('mcp__srv__remote')).toBeDefined() })
+
+    instances[0]!.onclose?.()
+    // Now waiting out a 60s backoff; disposal must return promptly anyway.
+    await ctx.fiber.dispose()
+    await sleep(30)
+    expect(mockConnect).toHaveBeenCalledTimes(1)
+    expect(instances).toHaveLength(1)
+  })
+
+  it('a transport close after dispose schedules nothing', async () => {
+    const fiber = ctx.plugin({ name: 'mcp-client', inject: ['tools'], apply }, stdioConfig())
+    await vi.waitFor(() => { expect(ctx.tools.get('mcp__srv__remote')).toBeDefined() })
+
+    await fiber.dispose()
+    expect(ctx.tools.get('mcp__srv__remote')).toBeUndefined()
+
+    // The disposer's client.close() fires onclose in the real SDK.
+    instances[0]!.onclose?.()
+    await sleep(30)
+    expect(instances).toHaveLength(1)
+    expect(mockConnect).toHaveBeenCalledTimes(1)
+  })
+
+  it('reconnect disabled keeps the registered tools and reports manual recovery', async () => {
+    const { errors } = captureLogs(ctx)
+    await apply(ctx, stdioConfig({ enabled: false }))
+    await vi.waitFor(() => { expect(ctx.tools.get('mcp__srv__remote')).toBeDefined() })
+
+    instances[0]!.onclose?.()
+    await sleep(30)
+    expect(mockConnect).toHaveBeenCalledTimes(1)
+    // Pre-reconnect contract: the generation stays registered until disposal.
+    expect(ctx.tools.get('mcp__srv__remote')).toBeDefined()
+    expect(errors.some(line => line.includes('connection lost and reconnect is disabled'))).toBe(true)
+  })
+  it('reconnect disabled after a failed initial connect reports no registered tools', async () => {
+    const { errors } = captureLogs(ctx)
+    mockConnect.mockRejectedValue(new Error('refused'))
+    await apply(ctx, stdioConfig({ enabled: false }))
+    await sleep(30)
+    expect(ctx.tools.get('mcp__srv__remote')).toBeUndefined()
+    expect(errors.some(line => line.includes('connection failed and reconnect is disabled'))).toBe(true)
+    expect(errors.some(line => line.includes('no tools were registered'))).toBe(true)
+  })
+
+  it('an uptime past the stability window resets the attempt budget', async () => {
+    const { errors } = captureLogs(ctx)
+    await apply(ctx, stdioConfig({ initialDelayMs: 2, maxDelayMs: 30, maxAttempts: 1 }))
+    await vi.waitFor(() => { expect(ctx.tools.get('mcp__srv__remote')).toBeDefined() })
+
+    instances[0]!.onclose?.()
+    await vi.waitFor(() => { expect(instances).toHaveLength(2) })
+    await vi.waitFor(() => { expect(ctx.tools.get('mcp__srv__remote')).toBeDefined() })
+
+    // Outlive the stability window (= maxDelayMs), then crash again: the
+    // budget restarts at attempt 1 instead of exceeding maxAttempts.
+    await sleep(40)
+    instances[1]!.onclose?.()
+    await vi.waitFor(() => { expect(instances).toHaveLength(3) })
+    await vi.waitFor(() => { expect(ctx.tools.get('mcp__srv__remote')).toBeDefined() })
+    expect(errors).toHaveLength(0)
+  })
+
+  it('a crash loop with briefly successful connects still exhausts the cap', async () => {
+    const { errors } = captureLogs(ctx)
+    await apply(ctx, stdioConfig({ initialDelayMs: 2, maxDelayMs: 10_000, maxAttempts: 1 }))
+    await vi.waitFor(() => { expect(ctx.tools.get('mcp__srv__remote')).toBeDefined() })
+
+    // Crash, recover (attempt 1 of 1), crash again well inside the stability
+    // window: the successful connect must not launder the budget.
+    instances[0]!.onclose?.()
+    await vi.waitFor(() => { expect(instances).toHaveLength(2) })
+    await vi.waitFor(() => { expect(ctx.tools.get('mcp__srv__remote')).toBeDefined() })
+    instances[1]!.onclose?.()
+
+    await vi.waitFor(() => {
+      expect(errors.some(line => line.includes('giving up after 1 consecutive failed reconnect attempts'))).toBe(true)
+    })
+    expect(instances).toHaveLength(2)
+    expect(ctx.tools.get('mcp__srv__remote')).toBeUndefined()
+  })
+
+  it('a connect rejection racing its own transport close schedules exactly one retry per attempt', async () => {
+    const { errors } = captureLogs(ctx)
+    await apply(ctx, stdioConfig({ initialDelayMs: 2, maxDelayMs: 8, maxAttempts: 3 }))
+    await vi.waitFor(() => { expect(ctx.tools.get('mcp__srv__remote')).toBeDefined() })
+
+    // Each reconnect attempt sees the stdio transport die (onclose) AND its
+    // connect() reject — the real SDK emits both for a spawn failure.
+    mockConnect.mockImplementation(async () => {
+      instances.at(-1)!.onclose?.()
+      throw new Error('spawn failed')
+    })
+    instances[0]!.onclose?.()
+
+    await vi.waitFor(() => {
+      expect(errors.some(line => line.includes('giving up after 3 consecutive failed reconnect attempts'))).toBe(true)
+    })
+    // Initial generation + exactly one generation per budgeted attempt: a
+    // double-scheduled retry would create more.
+    expect(instances).toHaveLength(4)
+    expect(errors.filter(line => line.includes('giving up')).length).toBe(1)
+  })
+
+  it('a transport that closes during a resolving connect registers nothing from the dead generation', async () => {
+    await apply(ctx, stdioConfig({ initialDelayMs: 2, maxDelayMs: 8, maxAttempts: 2 }))
+    await vi.waitFor(() => { expect(ctx.tools.get('mcp__srv__remote')).toBeDefined() })
+    expect(mockListTools).toHaveBeenCalledTimes(1)
+
+    mockConnect.mockImplementation(async () => {
+      instances.at(-1)!.onclose?.()
+    })
+    instances[0]!.onclose?.()
+
+    await vi.waitFor(() => { expect(ctx.tools.get('mcp__srv__remote')).toBeUndefined() })
+    // The dead generations never reached tool discovery.
+    expect(mockListTools).toHaveBeenCalledTimes(1)
+  })
+
+  it('dispose during an in-flight initial sync quiesces without leaking tools', async () => {
+    const fiber = ctx.plugin({ name: 'mcp-client', inject: ['tools'], apply }, stdioConfig({ initialDelayMs: 2, maxDelayMs: 8, maxAttempts: 5 }))
+    await vi.waitFor(() => { expect(ctx.tools.get('mcp__srv__remote')).toBeDefined() })
+
+    // Block the reconnect attempt's tool discovery until after dispose starts.
+    const gate: PromiseWithResolvers<unknown> = Promise.withResolvers()
+    mockListTools.mockImplementation(() => gate.promise)
+    instances[0]!.onclose?.()
+    await vi.waitFor(() => { expect(mockListTools).toHaveBeenCalledTimes(2) })
+
+    const disposing = fiber.dispose()
+    await sleep(10)
+    gate.resolve(listing('late'))
+    await disposing
+
+    // The late sync's swap ran, then disposal unregistered its result: no
+    // generation survives the plugin.
+    expect(ctx.tools.get('mcp__srv__remote')).toBeUndefined()
+    expect(ctx.tools.get('mcp__srv__late')).toBeUndefined()
+  })
+
+  it('a re-sync failing because dispose closed the transport stays silent', async () => {
+    const { errors } = captureLogs(ctx)
+    const fiber = ctx.plugin({ name: 'mcp-client', inject: ['tools'], apply }, stdioConfig())
+    await vi.waitFor(() => { expect(ctx.tools.get('mcp__srv__remote')).toBeDefined() })
+
+    const gate: PromiseWithResolvers<unknown> = Promise.withResolvers()
+    mockListTools.mockImplementation(() => gate.promise)
+    const handler = mockSetNotificationHandler.mock.calls[0]![1] as () => Promise<void>
+    const resync = handler()
+    await vi.waitFor(() => { expect(mockListTools).toHaveBeenCalledTimes(2) })
+
+    const disposing = fiber.dispose()
+    await sleep(10)
+    gate.reject(new Error('Connection closed'))
+    await disposing
+    await resync
+
+    expect(errors.some(line => line.includes('tool re-sync failed'))).toBe(false)
+  })
+
+  it('a stale notification handler from a replaced generation is ignored', async () => {
+    await apply(ctx, stdioConfig({ initialDelayMs: 2, maxDelayMs: 8, maxAttempts: 5 }))
+    await vi.waitFor(() => { expect(ctx.tools.get('mcp__srv__remote')).toBeDefined() })
+
+    instances[0]!.onclose?.()
+    await vi.waitFor(() => { expect(instances).toHaveLength(2) })
+    await vi.waitFor(() => { expect(ctx.tools.get('mcp__srv__remote')).toBeDefined() })
+    const listCalls = mockListTools.mock.calls.length
+
+    const staleHandler = mockSetNotificationHandler.mock.calls[0]![1] as () => Promise<void>
+    await staleHandler()
+    expect(mockListTools).toHaveBeenCalledTimes(listCalls)
+  })
+})
+
+// ---- Policy resolution ----
+
+describe('resolveReconnectPolicy', () => {
+  const path = 'mcp-client(srv): reconnect'
+
+  it('resolves omission to the defaults, frozen', () => {
+    const policy = resolveReconnectPolicy(undefined, path)
+    expect(policy).toEqual(RECONNECT_DEFAULTS)
+    expect(Object.isFrozen(policy)).toBe(true)
+  })
+
+  it('keeps explicit values', () => {
+    expect(resolveReconnectPolicy(
+      { enabled: false, initialDelayMs: 1, maxDelayMs: 2, maxAttempts: 7 },
+      path,
+    )).toEqual({ enabled: false, initialDelayMs: 1, maxDelayMs: 2, maxAttempts: 7 })
+  })
+
+  it('rejects unknown keys', () => {
+    expect(() => resolveReconnectPolicy({ jitterRatio: 0.5 } as never, path))
+      .toThrow(/reconnect\.jitterRatio is not a reconnect option/)
+  })
+
+  it('rejects out-of-range delays', () => {
+    expect(() => resolveReconnectPolicy({ initialDelayMs: 0 }, path)).toThrow(/initialDelayMs must be a positive finite number/)
+    expect(() => resolveReconnectPolicy({ initialDelayMs: Number.POSITIVE_INFINITY }, path)).toThrow(/initialDelayMs/)
+    expect(() => resolveReconnectPolicy({ maxDelayMs: -1 }, path)).toThrow(/maxDelayMs must be a positive finite number/)
+  })
+
+  it('rejects an initial delay above the ceiling', () => {
+    expect(() => resolveReconnectPolicy({ initialDelayMs: 100, maxDelayMs: 5 }, path))
+      .toThrow(/initialDelayMs must be less than or equal to maxDelayMs/)
+  })
+
+  it('rejects non-positive-integer attempt caps', () => {
+    expect(() => resolveReconnectPolicy({ maxAttempts: 0 }, path)).toThrow(/maxAttempts must be a positive integer/)
+    expect(() => resolveReconnectPolicy({ maxAttempts: 1.5 }, path)).toThrow(/maxAttempts must be a positive integer/)
+  })
+
+  it('apply fails loud at load on a misconfigured reconnect', async () => {
+    const ctx = await mountRegistry()
+    await expect(apply(ctx, stdioConfig({ initialDelayMs: 100, maxDelayMs: 5 })))
+      .rejects.toThrow(/initialDelayMs must be less than or equal to maxDelayMs/)
+  })
+})
--- a/packages/mcp/mcp-client/tsconfig.json
+++ b/packages/mcp/mcp-client/tsconfig.json
@@ -26,6 +26,9 @@
    },
    {
      "path": "../../support/invariants"
+    },
+    {
+      "path": "../../util/timeout"
    }
  ]
 }
--- a/packages/self-modification/tool-cordis/src/api-catalog.ts
+++ b/packages/self-modification/tool-cordis/src/api-catalog.ts
@@ -2637,7 +2637,7 @@ export const TYPE_API: readonly TypeApiEntry[] = [
  },
  {
    name: 'SessionEvent',
-    declaration: 'export type SessionEvent<T extends SessionEventType = SessionEventType> = {\n    [K in SessionEventType]: {\n        type: K;\n        seq: number;\n        time: number;\n        data: SessionEventMap[K];\n    } & (K extends SurfaceEventType ? {\n        sourceEventSeqs?: number[];\n        surfaceOp?: SurfaceOp;\n    } : object);\n}[T];',
+    declaration: 'export type SessionEvent<T extends SessionEventType = SessionEventType> = {\n    [K in SessionEventType]: {\n        type: K;\n        seq: number;\n        time: number;\n        data: SessionEventMap[K];\n        ignorable?: true;\n    } & (K extends SurfaceEventType ? {\n        sourceEventSeqs?: number[];\n        surfaceOp?: SurfaceOp;\n    } : object);\n}[T];',
  },
  {
    name: 'SessionEventMap',
--- a/packages/session/session-persistence-jsonl/src/format.ts
+++ b/packages/session/session-persistence-jsonl/src/format.ts
@@ -9,8 +9,9 @@
 */

 import { join } from 'node:path'
-import { decodeStorageRecord, packChunkRuns } from '@deepseek-ai/dsh-session'
+import { decodeStorageRecord, packChunkRuns, SESSION_FORMAT_VERSION } from '@deepseek-ai/dsh-session'
 import type { SessionEvent, SessionHeader, SessionId, StorageRecord } from '@deepseek-ai/dsh-session'
+import { SessionFormatUnsupportedError, sessionFormatVersionRefusal } from '@deepseek-ai/dsh-session-persistence'

 /** Physical encoding selected for JSONL session artifacts. */
 export type JsonlCompression = 'zstd' | 'none'
@@ -229,6 +230,22 @@ interface SessionLogScan {
 }

 /** Parse one complete header record supplied independently from event rows. */
+/**
+ * Refuse a header carrying a format version this build does not read BEFORE
+ * validating the current header shape or decoding any event row: a future
+ * format need not satisfy today's structural checks at all, and its user must
+ * see "upgrade the harness", never "corrupt session log".
+ * @param parsed - the JSON-parsed first line of a session artifact.
+ */
+function refuseForeignFormatVersion(parsed: unknown): void {
+  if (typeof parsed !== 'object' || parsed === null) return
+  const { version, id } = parsed as { version?: unknown; id?: unknown }
+  if (typeof version !== 'number' || version === SESSION_FORMAT_VERSION) return
+  throw new SessionFormatUnsupportedError(
+    sessionFormatVersionRefusal(typeof id === 'string' ? id : String(id), version),
+  )
+}
+
 function parseHeaderRecord(record: Buffer): SessionHeader {
  if (record.length === 0 || record.at(-1) !== 0x0A || record.indexOf(0x0A) !== record.length - 1) {
    throw new Error('empty or header-less session log')
@@ -239,6 +256,7 @@ function parseHeaderRecord(record: Buffer): SessionHeader {
  } catch {
    throw new Error('corrupt session log: header line is not valid JSON')
  }
+  refuseForeignFormatVersion(parsed)
  if (!isHeaderLine(parsed)) {
    throw new Error('corrupt session log: first line is not a session header')
  }
--- a/packages/session/session-persistence-jsonl/src/index.ts
+++ b/packages/session/session-persistence-jsonl/src/index.ts
@@ -16,7 +16,7 @@ import { scheduler } from 'node:timers/promises'
 import { randomBytes } from 'node:crypto'
 import {
  DEFAULT_PREPARED_SESSION_CACHE_SIZE, DEFAULT_WRITE_BATCH_MAX_DELAY_MS, MAX_WRITE_BATCH_DELAY_MS,
-  SessionPersistence, SessionPersistenceRevision, PersistenceCoordinator,
+  SessionPersistence, SessionPersistenceRevision, PersistenceCoordinator, SessionFormatUnsupportedError,
  type PersistenceBackend, type SessionLocation, type SessionPersistenceSnapshot,
  type SessionInspection, type SessionPersistenceRevision as PersistenceRevision, type StoredPrefix,
 } from '@deepseek-ai/dsh-session-persistence'
@@ -256,19 +256,29 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
      }
    }
    let prefix: Omit<StoredPrefix<JsonlTornMarker>, 'revision'>
-    if (this.compression === 'zstd') {
-      prefix = await this.readZstdPrefix(buffer, signal)
-    } else {
-      signal?.throwIfAborted()
-      const { meta, events, committedBytes } = scanLog(buffer)
-      signal?.throwIfAborted()
-      prefix = {
-        meta,
-        events,
-        ...committedBytes < buffer.byteLength
-          ? { tornMarker: { truncateTo: committedBytes, recoveredEvents: [] } }
-          : {},
+    try {
+      if (this.compression === 'zstd') {
+        prefix = await this.readZstdPrefix(buffer, signal)
+      } else {
+        signal?.throwIfAborted()
+        const { meta, events, committedBytes } = scanLog(buffer)
+        signal?.throwIfAborted()
+        prefix = {
+          meta,
+          events,
+          ...committedBytes < buffer.byteLength
+            ? { tornMarker: { truncateTo: committedBytes, recoveredEvents: [] } }
+            : {},
+        }
      }
+    } catch (error: unknown) {
+      // A parse-time format refusal predates any SessionHeader, so the
+      // coordinator's locate-based enrichment cannot run; attach the artifact
+      // this read actually refused.
+      if (error instanceof SessionFormatUnsupportedError && error.location === undefined) {
+        throw new SessionFormatUnsupportedError(`${error.message} (raw log: ${path})`, { kind: 'jsonl', path })
+      }
+      throw error
    }
    signal?.throwIfAborted()
    await this.assertStoredIdentity(path, prefix.meta, expectedId, signal)
--- a/packages/session/session-persistence-jsonl/tests/jsonl.spec.ts
+++ b/packages/session/session-persistence-jsonl/tests/jsonl.spec.ts
@@ -3,7 +3,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
 import { Context } from '@deepseek-ai/cordis'
 import { appendFile, mkdtemp, mkdir, rm, readFile, writeFile, readdir, stat, symlink } from 'node:fs/promises'
 import { tmpdir } from 'node:os'
-import { isAbsolute, join, relative, resolve } from 'node:path'
+import { dirname, isAbsolute, join, relative, resolve } from 'node:path'
 import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
 import type { Session, SessionEvent, SessionHeader } from '@deepseek-ai/dsh-session'
 import SessionPersistenceJsonl from '@deepseek-ai/dsh-session-persistence-jsonl'
@@ -186,6 +186,76 @@ describe('SessionPersistenceJsonl: format helpers', () => {
    })
    await fiber.dispose()
  })
+
+  it('refuses a structurally foreign future header as unsupported, not corrupt', async () => {
+    const absoluteRoot = await freshRoot()
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    const fiber = await ctx.plugin(SessionPersistenceJsonl, { root: absoluteRoot, compression: 'none' })
+    // A future format need not satisfy today's header shape at all (no
+    // createdAt, unknown fields): the version must be refused before shape
+    // validation, so the user sees the upgrade direction.
+    const id = SessionId('future-shape')
+    const path = rawLogPath(resolve(absoluteRoot), '/work', id)
+    await mkdir(dirname(path), { recursive: true })
+    await writeFile(path, `${JSON.stringify({ type: 'session', version: 42, id, futureOnly: true })}\n{"future":"row"}\n`)
+    const failure = await ctx.sessionPersistence.load(id).then(() => undefined, (error: unknown) => error as Error)
+    expect(failure?.name).toBe('SessionFormatUnsupportedError')
+    expect(failure?.message).toMatch(/written by a newer harness.*upgrade the harness/)
+    expect(failure?.message).toContain(`(raw log: ${path})`)
+    await fiber.dispose()
+  })
+
+  it('keeps a non-object header line a corruption, not a format refusal', async () => {
+    const absoluteRoot = await freshRoot()
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    const fiber = await ctx.plugin(SessionPersistenceJsonl, { root: absoluteRoot, compression: 'none' })
+    // Valid JSON that is no object carries no version to compare, so the
+    // version guard must pass it through to the corruption diagnostics.
+    const id = SessionId('scalar-header')
+    const path = rawLogPath(resolve(absoluteRoot), '/work', id)
+    await mkdir(dirname(path), { recursive: true })
+    await writeFile(path, '42\n')
+    const failure = await ctx.sessionPersistence.load(id).then(() => undefined, (error: unknown) => error as Error)
+    expect(failure?.name).not.toBe('SessionFormatUnsupportedError')
+    expect(failure?.message).toContain('first line is not a session header')
+    await fiber.dispose()
+  })
+
+  it('names a foreign-version header by its stringified non-string id', async () => {
+    const absoluteRoot = await freshRoot()
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    const fiber = await ctx.plugin(SessionPersistenceJsonl, { root: absoluteRoot, compression: 'none' })
+    // A future header's id field is as untrusted as the rest of its shape:
+    // the refusal must still name the session it read, not crash on the type.
+    const id = SessionId('numeric-id')
+    const path = rawLogPath(resolve(absoluteRoot), '/work', id)
+    await mkdir(dirname(path), { recursive: true })
+    await writeFile(path, `${JSON.stringify({ type: 'session', version: 42, id: 123 })}\n`)
+    const failure = await ctx.sessionPersistence.load(id).then(() => undefined, (error: unknown) => error as Error)
+    expect(failure?.name).toBe('SessionFormatUnsupportedError')
+    expect(failure?.message).toContain('session "123" uses log format v42')
+    await fiber.dispose()
+  })
+
+  it('points a format refusal at the raw log path', async () => {
+    const absoluteRoot = await freshRoot()
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    const fiber = await ctx.plugin(SessionPersistenceJsonl, { root: absoluteRoot, compression: 'none' })
+    const m = { ...meta('newer-format', '/work'), version: 7 }
+    await ctx.sessionPersistence.create(m)
+    await ctx.sessionPersistence.append(m.id, [
+      { type: 'turn/start', seq: 0, time: 1, data: { turn: 1 } },
+      { type: 'turn/end', seq: 1, time: 2, data: { turn: 1, reason: { kind: 'completed' } } },
+    ])
+    const failure = await ctx.sessionPersistence.load(m.id).then(() => undefined, (error: unknown) => error as Error)
+    expect(failure?.name).toBe('SessionFormatUnsupportedError')
+    expect(failure?.message).toContain(`(raw log: ${rawLogPath(resolve(absoluteRoot), '/work', m.id)})`)
+    await fiber.dispose()
+  })
 })

 describe('SessionPersistenceJsonl: durability and crash semantics', () => {
--- a/packages/session/session-persistence-sqlite/src/index.ts
+++ b/packages/session/session-persistence-sqlite/src/index.ts
@@ -28,15 +28,18 @@ import {
 export { SCHEMA_VERSION } from './schema.ts'

 /**
- * Serialize an event's surface-metadata fields for SQL binding. Both fields are
- * nullable TEXT columns — null when the event has no surface metadata (non-surface
- * events, events written before surface support).
+ * Serialize an event's optional envelope fields for SQL binding. The surface
+ * fields are nullable TEXT columns — null when the event has no surface
+ * metadata (non-surface events, events written before surface support); the
+ * ignorable marker is a nullable INTEGER column — `1` iff the envelope carries
+ * `ignorable: true`.
 */
-function surfaceBindings(event: SessionEvent): [string | null, string | null] {
+function envelopeBindings(event: SessionEvent): [string | null, string | null, number | null] {
  const se = event as SessionEvent<SurfaceEventType>
  return [
    se.sourceEventSeqs ? JSON.stringify(se.sourceEventSeqs) : null,
    se.surfaceOp !== undefined ? JSON.stringify(se.surfaceOp) : null,
+    event.ignorable === true ? 1 : null,
  ]
 }

@@ -225,7 +228,7 @@ export class SessionPersistenceSqlite extends SessionPersistence implements Pers
    if (row === undefined) return undefined
    const meta = rowToMeta(row)
    const eventRows = this.db
-      .prepare('SELECT seq, type, time, data, source_event_seqs, surface_op FROM events WHERE session_id = ? AND seq >= ? ORDER BY seq')
+      .prepare('SELECT seq, type, time, data, source_event_seqs, surface_op, ignorable FROM events WHERE session_id = ? AND seq >= ? ORDER BY seq')
      .all(id, fromSeq) as unknown as EventRow[]
    signal?.throwIfAborted()
    const { preserved } = scanRows(eventRows, fromSeq)
@@ -247,7 +250,7 @@ export class SessionPersistenceSqlite extends SessionPersistence implements Pers
      const row = this.rowFor(id)
      if (row !== undefined) {
        const eventRows = this.db
-          .prepare('SELECT seq, type, time, data, source_event_seqs, surface_op FROM events WHERE session_id = ? ORDER BY seq')
+          .prepare('SELECT seq, type, time, data, source_event_seqs, surface_op, ignorable FROM events WHERE session_id = ? ORDER BY seq')
          .all(id) as unknown as EventRow[]
        snapshot = { row, eventRows }
      }
@@ -279,14 +282,14 @@ export class SessionPersistenceSqlite extends SessionPersistence implements Pers
  async appendBatch(meta: SessionHeader, events: readonly SessionEvent[], isMaterialized: boolean): Promise<void> {
    await this.ready
    const insertEvent = this.db.prepare(
-      'INSERT INTO events (session_id, seq, type, time, data, source_event_seqs, surface_op) VALUES (?, ?, ?, ?, ?, ?, ?)',
+      'INSERT INTO events (session_id, seq, type, time, data, source_event_seqs, surface_op, ignorable) VALUES (?, ?, ?, ?, ?, ?, ?, ?)',
    )
    this.db.exec('BEGIN')
    try {
      if (!isMaterialized) this.writeRow(meta)
      for (const event of events) {
-        const [surfaceSeqs, surfaceOp] = surfaceBindings(event)
-        insertEvent.run(meta.id, event.seq, event.type, event.time, JSON.stringify(event.data), surfaceSeqs, surfaceOp)
+        const [surfaceSeqs, surfaceOp, ignorable] = envelopeBindings(event)
+        insertEvent.run(meta.id, event.seq, event.type, event.time, JSON.stringify(event.data), surfaceSeqs, surfaceOp, ignorable)
      }
      this.db.prepare('UPDATE sessions SET revision = revision + 1 WHERE id = ?').run(meta.id)
      this.db.exec('COMMIT')
@@ -310,11 +313,11 @@ export class SessionPersistenceSqlite extends SessionPersistence implements Pers
      }
      if (closers.length > 0) {
        const insertEvent = this.db.prepare(
-          'INSERT INTO events (session_id, seq, type, time, data, source_event_seqs, surface_op) VALUES (?, ?, ?, ?, ?, ?, ?)',
+          'INSERT INTO events (session_id, seq, type, time, data, source_event_seqs, surface_op, ignorable) VALUES (?, ?, ?, ?, ?, ?, ?, ?)',
        )
        for (const event of closers) {
-          const [surfaceSeqs, surfaceOp] = surfaceBindings(event)
-          insertEvent.run(meta.id, event.seq, event.type, event.time, JSON.stringify(event.data), surfaceSeqs, surfaceOp)
+          const [surfaceSeqs, surfaceOp, ignorable] = envelopeBindings(event)
+          insertEvent.run(meta.id, event.seq, event.type, event.time, JSON.stringify(event.data), surfaceSeqs, surfaceOp, ignorable)
        }
      }
      if (tornMarker !== undefined || closers.length > 0) {
--- a/packages/session/session-persistence-sqlite/src/schema.ts
+++ b/packages/session/session-persistence-sqlite/src/schema.ts
@@ -17,7 +17,7 @@ import type { SessionEvent, SessionId, SessionHeader, SurfaceOp } from '@deepsee
 * layout; orthogonal to a session's own `version` (which versions the EVENT
 * vocabulary, stored per session in the `sessions` row).
 */
-export const SCHEMA_VERSION = 14
+export const SCHEMA_VERSION = 15

 /** SQLite application id protecting unrelated databases from persistence writes. */
 export const SESSION_PERSISTENCE_SQLITE_APPLICATION_ID = 0x44534850
@@ -55,6 +55,8 @@ export interface EventRow {
  source_event_seqs: string | null
  /** JSON-encoded `SurfaceOp` — how the event entered the surface, or null. */
  surface_op: string | null
+  /** `1` iff the event carries the envelope's `ignorable: true` marker, else null. */
+  ignorable: number | null
 }

 /**
@@ -139,6 +141,7 @@ function configureDatabase(db: DatabaseSync, path: string, journalMode: JournalM
        data              TEXT NOT NULL,
        source_event_seqs TEXT,
        surface_op        TEXT,
+        ignorable         INTEGER,
        PRIMARY KEY (session_id, seq)
      ) STRICT
    `)
@@ -203,12 +206,14 @@ export function rowToEvent(row: EventRow): SessionEvent {
    ...row.source_event_seqs !== null ? { sourceEventSeqs: JSON.parse(row.source_event_seqs) as number[] } : {},
    ...row.surface_op !== null ? { surfaceOp: JSON.parse(row.surface_op) as SurfaceOp } : {},
  }
+  const ignorableField = row.ignorable === 1 ? { ignorable: true as const } : {}
  return {
    type: row.type as SessionEvent['type'],
    seq: row.seq,
    time: row.time,
    data: JSON.parse(row.data) as SessionEvent['data'],
    ...surfaceFields,
+    ...ignorableField,
  } as SessionEvent
 }

--- a/packages/session/session-persistence-sqlite/tests/sqlite.spec.ts
+++ b/packages/session/session-persistence-sqlite/tests/sqlite.spec.ts
@@ -92,6 +92,7 @@ describe('scanRows', () => {
        seq: e.seq, type: e.type, time: e.time, data: JSON.stringify(e.data),
        source_event_seqs: se.sourceEventSeqs !== undefined ? JSON.stringify(se.sourceEventSeqs) : null,
        surface_op: se.surfaceOp !== undefined ? JSON.stringify(se.surfaceOp) : null,
+        ignorable: e.ignorable === true ? 1 : null,
      }
    })

@@ -142,8 +143,8 @@ describe('scanRows', () => {

  it('throws on an unparsable row inside the committed region', () => {
    const withCorruptCommitted: EventRow[] = [
-      { seq: 0, type: 'turn/start', time: 1, data: '{not json', source_event_seqs: null, surface_op: null }, // corrupt, sits before a turn/end
-      { seq: 1, type: 'turn/end', time: 2, data: JSON.stringify({ turn: 1, reason: { kind: 'completed' } }), source_event_seqs: null, surface_op: null },
+      { seq: 0, type: 'turn/start', time: 1, data: '{not json', source_event_seqs: null, surface_op: null, ignorable: null }, // corrupt, sits before a turn/end
+      { seq: 1, type: 'turn/end', time: 2, data: JSON.stringify({ turn: 1, reason: { kind: 'completed' } }), source_event_seqs: null, surface_op: null, ignorable: null },
    ]
    expect(() => scanRows(withCorruptCommitted)).toThrow(/unparsable committed event/)
  })
@@ -151,7 +152,7 @@ describe('scanRows', () => {
  it('tolerates an unparsable torn-tail row after the last turn/end', () => {
    const withCorruptTail: EventRow[] = [
      ...rows(oneTurnLog()),
-      { seq: 6, type: 'turn/start', time: 7, data: '{not json', source_event_seqs: null, surface_op: null }, // torn fragment, no committed turn/end after
+      { seq: 6, type: 'turn/start', time: 7, data: '{not json', source_event_seqs: null, surface_op: null, ignorable: null }, // torn fragment, no committed turn/end after
    ]
    const { preserved, tornFrom } = scanRows(withCorruptTail)
    expect(preserved).toEqual(oneTurnLog())
@@ -658,7 +659,7 @@ describe('SessionPersistenceSqlite: durability and crash semantics', () => {
  })

  it('exposes the schema version constant', () => {
-    expect(SCHEMA_VERSION).toBe(14)
+    expect(SCHEMA_VERSION).toBe(15)
  })

  it('keeps the revision stable for an empty repair hook', async () => {
@@ -857,6 +858,7 @@ describe('surface field round-trip', () => {
      data: JSON.stringify({ turn: 1, step: 1, content: [] }),
      source_event_seqs: JSON.stringify([3, 5]),
      surface_op: JSON.stringify('append'),
+      ignorable: null,
    }
    const event = rowToEvent(row)
    expect((event as SurfaceEvent).sourceEventSeqs).toEqual([3, 5])
@@ -869,6 +871,7 @@ describe('surface field round-trip', () => {
      data: JSON.stringify({ turn: 1, step: 1, content: [] }),
      source_event_seqs: JSON.stringify([0, 1]),
      surface_op: JSON.stringify({ op: 'replace', start: 0, end: 1 }),
+      ignorable: null,
    }
    const event = rowToEvent(row)
    expect((event as SurfaceEvent).sourceEventSeqs).toEqual([0, 1])
@@ -879,10 +882,10 @@ describe('surface field round-trip', () => {
    const rows: EventRow[] = [
      { seq: 0, type: 'user/message', time: 1,
        data: JSON.stringify({ content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }),
-        source_event_seqs: null, surface_op: '{"op":"replace","start":0,"end":0}' },
+        source_event_seqs: null, surface_op: '{"op":"replace","start":0,"end":0}', ignorable: null },
      { seq: 1, type: 'turn/end', time: 2,
        data: JSON.stringify({ turn: 1, reason: { kind: 'completed' } }),
-        source_event_seqs: null, surface_op: null },
+        source_event_seqs: null, surface_op: null, ignorable: 1 },
    ]
    const { preserved } = scanRows(rows)
    expect(preserved).toHaveLength(2)
--- a/packages/session/session-persistence/README.i18n.yaml
+++ b/packages/session/session-persistence/README.i18n.yaml
@@ -2,5 +2,5 @@
 # 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 packages/session/session-persistence/README.md
-README.md: 391548b1b896dca14cbe4f4ae55cf4180c4e0ac2
-README.zh.md: 7213e1ee71ba418ffacc3685df371dcba33588a7
+README.md: 324c00b3202bd136566137e1bd398b29d2ea4b82
+README.zh.md: 2ef5e9a90f0323f8edf8fdc4f936c41ca7e08c70
--- a/packages/session/session-persistence/README.md
+++ b/packages/session/session-persistence/README.md
@@ -14,9 +14,9 @@ The persisted unit IS the existing `SessionEvent` (event-sourced model — the l
 | `create(meta): Promise<void>` | Register a new session's metadata. MAY defer the physical write until the first `append` (lazy materialization). |
 | `append(id, events): Promise<void>` | Durably persist a batch. Append-only; first event `seq` == stored next-seq after any repair; rejects non-JSON-serializable data naming the offending type. |
 | `prepare(id, signal?): Promise<SessionPreparation>` | Reserve the exact unpublished Session used by resume. A coordinator reuses an earlier inspection when available, commits pending recovery, and releases an unpublished reservation back to its bounded cache on disposal. |
-| `load(id): Promise<{ meta; events }>` | Return an immutable balanced logical log after converting supported older records from the same format version and committing cold recovery. A live load first flushes its snapshot and rejects while its turn is open; a cold load preserves an interrupted final turn and durably closes it with synthetic `tool/result`/`step/end?`/`turn/end {interrupted}` events. Only a torn tail fragment is dropped; committed corruption, malformed records, and unknown `version` reject. |
+| `load(id): Promise<{ meta; events }>` | Return an immutable balanced logical log after converting supported older records from the same format version and committing cold recovery. A live load first flushes its snapshot and rejects while its turn is open; a cold load preserves an interrupted final turn and durably closes it with synthetic `tool/result`/`step/end?`/`turn/end {interrupted}` events. Only a torn tail fragment is dropped; committed corruption and malformed records reject as `SessionPersistenceCorruptionError`, while an unsupported format `version` or an event type unknown to this build (without the envelope's `ignorable` marker) refuses as `SessionFormatUnsupportedError`, naming the refusal direction and the raw log path when the backend keeps one artifact per session. |
 | `inspect(id, signal?): Promise<{ meta; events }>` | Return an upgraded, validated, deeply frozen logical view without committing recovery or publishing a Session. A cold view receives in-memory synthetic recovery closers while its physical torn tail remains untouched; an already-live view is its current immutable snapshot and may contain an open turn. Coordinator-backed implementations retain the exact cold unpublished Session in a bounded LRU for later `prepare`, but discard and reload it when the stored revision changes. Same-id inspections share an in-flight read. |
-| `readFrom(id, fromSeq, signal?): Promise<{ meta; events }>` | Return valid stored events with `seq >= fromSeq` without preparation caching, truncation, closers, or coordinator state. A `fromSeq` at or past the stored end returns an empty event list; a negative or non-safe-integer `fromSeq` rejects. Seek-capable backends (SQLite) read only the suffix unless converting a supported older record requires earlier records; sequential backends (JSONL) parse the whole artifact and skip forward. Intended for checkpoint consumers that apply only events after a stored sequence number. |
+| `readFrom(id, fromSeq, signal?): Promise<{ meta; events }>` | Return valid stored events with `seq >= fromSeq` without preparation caching, truncation, closers, or coordinator state. A `fromSeq` at or past the stored end returns an empty event list; a negative or non-safe-integer `fromSeq` rejects. Seek-capable backends (SQLite) read only the suffix unless converting a supported older record requires earlier records; sequential backends (JSONL) parse the whole artifact and skip forward. Unknown-type refusal follows that access pattern: a seek read checks only the returned suffix, while the sequential fallback also refuses on an unknown required event below the window. Intended for checkpoint consumers that apply only events after a stored sequence number. |
 | `list(signal?): Promise<SessionHeader[]>` | Lightweight listing from metadata, no full-log parse. The optional signal cancels backend listing work. A zero-event lazily-materialized session is absent from `list`. |
 | `listSnapshots(signal?): Promise<SessionPersistenceSnapshot[]>` | Lightweight metadata plus an opaque branded per-log revision, without loading event logs. A revision stays equal while that log and its backing store are unchanged, changes after append or mutating load repair, and cannot collide solely because two stores use the same local counter. The optional signal requests cancellation of backend discovery work; first-party backends settle any started listing work before rejecting so an awaited call is quiescent. |

--- a/packages/session/session-persistence/README.zh.md
+++ b/packages/session/session-persistence/README.zh.md
@@ -14,9 +14,9 @@
 | `create(meta): Promise<void>` | 注册新会话元数据。可以将物理写入延迟到第一次 `append`（延迟实体化）。 |
 | `append(id, events): Promise<void>` | 持久保存一个批次。仅追加；任何修复后，第一个事件 `seq` == 已存储 next-seq；非 JSON 可序列化数据会被拒绝，并命名违规类型。 |
 | `prepare(id, signal?): Promise<SessionPreparation>` | 预留恢复所使用的那个未发布 Session。协调器会尽可能复用之前的检查结果、提交待处理恢复，并在 dispose 时将未发布 reservation 释放回有界缓存。 |
-| `load(id): Promise<{ meta; events }>` | 转换同一格式版本中受支持的旧记录后，返回不可变、平衡的逻辑日志，并提交冷恢复。实时 load 先 flush 其快照，并在轮次开放时拒绝；冷 load 保留中断的最终轮次，并用合成 `tool/result`/`step/end?`/`turn/end {interrupted}` 事件持久关闭它。只丢弃撕裂尾部碎片；已提交损坏、格式错误的记录和未知 `version` 会被拒绝。 |
+| `load(id): Promise<{ meta; events }>` | 转换同一格式版本中受支持的旧记录后，返回不可变、平衡的逻辑日志，并提交冷恢复。实时 load 先 flush 其快照，并在轮次开放时拒绝；冷 load 保留中断的最终轮次，并用合成 `tool/result`/`step/end?`/`turn/end {interrupted}` 事件持久关闭它。只丢弃撕裂尾部碎片；已提交损坏和格式错误的记录以 `SessionPersistenceCorruptionError` 拒绝，不支持的格式 `version` 或本构建不认识且信封未带 `ignorable` 标记的事件类型以 `SessionFormatUnsupportedError` 拒绝，消息说明拒绝方向，并在后端为每个会话保留独立文件时给出原始日志路径。 |
 | `inspect(id, signal?): Promise<{ meta; events }>` | 返回已经升级、验证和深度冻结的逻辑视图，但不提交恢复或发布 Session。冷视图会获得仅存在于内存的合成恢复 closer，物理撕裂尾部保持不变；实时状态下的视图则是当前不可变快照，可能包含开放的轮次。基于协调器的实现会在有界 LRU 中保留该冷状态下未发布的 Session 本身，供后续 `prepare` 使用，但已存储修订值变化后会丢弃并重新读取。同 id 检查共享进行中的读取。 |
-| `readFrom(id, fromSeq, signal?): Promise<{ meta; events }>` | 返回 `seq >= fromSeq` 的有效已存储事件，不进入 preparation 缓存、不截断、不合成 closer，也不发布协调器状态。`fromSeq` 达到或超过已存储末尾时返回空事件列表；负数或非安全整数 `fromSeq` 会被拒绝。可寻址后端（SQLite）只读后缀，除非转换受支持的旧记录需要读取更早的记录；顺序后端（JSONL）解析整个产物并向前跳过。供 checkpoint 消费方只应用已存序号之后的事件。 |
+| `readFrom(id, fromSeq, signal?): Promise<{ meta; events }>` | 返回 `seq >= fromSeq` 的有效已存储事件，不进入 preparation 缓存、不截断、不合成 closer，也不发布协调器状态。`fromSeq` 达到或超过已存储末尾时返回空事件列表；负数或非安全整数 `fromSeq` 会被拒绝。可寻址后端（SQLite）只读后缀，除非转换受支持的旧记录需要读取更早的记录；顺序后端（JSONL）解析整个产物并向前跳过。未知类型拒绝遵循同一读取方式：寻址读取只检查返回的后缀，顺序回退路径还会拒绝窗口以下的未知必需事件。供 checkpoint 消费方只应用已存序号之后的事件。 |
 | `list(signal?): Promise<SessionHeader[]>` | 从元数据轻量列出，不解析完整日志。可选信号取消后端列表工作。零事件延迟实体化会话不在 `list` 中。 |
 | `listSnapshots(signal?): Promise<SessionPersistenceSnapshot[]>` | 返回轻量元数据和每份日志一个不透明、带品牌类型的修订值，不加载事件日志。日志及其后端存储不变时，修订保持相等；append 或变更性 load 修复后会改变；不会仅因两个存储使用相同本地计数器而冲突。可选信号请求取消后端发现工作；第一方后端会先等待所有已启动的列出工作结束，再予以拒绝，因此调用返回拒绝时，相关工作已完全停稳。 |

--- a/packages/session/session-persistence/src/coordinator.ts
+++ b/packages/session/session-persistence/src/coordinator.ts
@@ -9,6 +9,7 @@ import { Context } from '@deepseek-ai/cordis'
 import {
  adoptSessionEvent,
  interruptedTurnClosers,
+  KNOWN_SESSION_EVENT_TYPES,
  SESSION_FORMAT_VERSION,
  SessionPreparation,
  snapshotJsonValue,
@@ -16,7 +17,7 @@ import {
 } from '@deepseek-ai/dsh-session'
 import type { Session, SessionEvent, SessionId, SessionHeader } from '@deepseek-ai/dsh-session'
 import { MAX_TIMER_DELAY_MS } from '@deepseek-ai/dsh-timeout'
-import type { SessionInspection } from './index.ts'
+import type { SessionInspection, SessionLocation } from './index.ts'
 import type { SessionPersistenceRevision } from './revision.ts'
 import { observeQueuedAbort, SessionPreparations } from './preparations.ts'
 import type { SessionPreparationReservation } from './preparations.ts'
@@ -43,6 +44,42 @@ export class SessionPersistenceCorruptionError extends Error {
  }
 }

+/**
+ * The stored log is intact but this runtime cannot faithfully interpret it:
+ * the header carries an unsupported format version, or an event's type is
+ * unknown to this build and the event is not marked ignorable. Distinct from
+ * {@link SessionPersistenceCorruptionError} — nothing is damaged; the raw log
+ * remains readable at {@link location} when the backend keeps one artifact
+ * per session.
+ */
+export class SessionFormatUnsupportedError extends Error {
+  /**
+   * @param message - stable reason the log cannot be interpreted, already
+   *   including the raw-log path when one exists.
+   * @param location - the backend's artifact location, when one exists.
+   */
+  constructor(message: string, readonly location?: SessionLocation) {
+    super(message)
+    this.name = 'SessionFormatUnsupportedError'
+  }
+}
+
+/**
+ * Direction-aware refusal text for a stored session whose format version this
+ * build does not read. Shared by the coordinator's load-time check and by
+ * backends that must refuse BEFORE decoding version-dependent structure (a
+ * future format may not satisfy today's structural checks at all, and the
+ * user must see "upgrade the harness", never "corrupt").
+ * @param id - the stored session id, for message context.
+ * @param version - the stored format version.
+ * @returns the stable refusal text, without a raw-log path suffix.
+ */
+export function sessionFormatVersionRefusal(id: string, version: number): string {
+  return version > SESSION_FORMAT_VERSION
+    ? `session "${id}" uses log format v${version}, but this harness reads only v${SESSION_FORMAT_VERSION}: the log was written by a newer harness — upgrade the harness to open it`
+    : `session "${id}" uses log format v${version}, older than the supported v${SESSION_FORMAT_VERSION}, and this build ships no upgrade path for it`
+}
+
 /** Coordinator policy supplied by a concrete persistence backend. */
 export interface PersistenceCoordinatorOptions {
  /** Maximum completed unpublished preparations retained for reuse. */
@@ -126,6 +163,11 @@ export interface PersistenceBackend<TornMarker = unknown> {
   * contains a supported legacy shape whose normalization needs earlier
   * message-identity facts, in which case the coordinator falls back
   * to the complete stored prefix.
+   * Unknown-type refusal follows the same suffix scope: a seek-capable
+   * backend's `readFrom` checks only the returned suffix, while the
+   * sequential fallback parses the whole artifact and refuses on an unknown
+   * required event anywhere in it — over-refusal on the sequential side is
+   * accepted rather than widening the seek read.
   * @param id - persisted session id to resolve.
   * @param fromSeq - first event seq to include (non-negative safe integer,
   *   validated by the coordinator before this hook runs).
@@ -156,6 +198,14 @@ export interface PersistenceBackend<TornMarker = unknown> {
   */
  list(signal?: AbortSignal): Promise<SessionHeader[]>

+  /**
+   * Optional side-effect-free artifact locator, used to point refusal
+   * diagnostics ({@link SessionFormatUnsupportedError}) at the raw log.
+   * Backends without one artifact per session omit it or return `undefined`.
+   * @param meta - the header whose artifact is requested.
+   */
+  locate?(meta: SessionHeader): SessionLocation | undefined
+
  /**
   * Optional lifecycle teardown (e.g. close a database handle). Awaited by the
   * coordinator's dispose effect AFTER the quiescence drain. A stateless file
@@ -631,9 +681,13 @@ export class PersistenceCoordinator<TornMarker = unknown> {

  private async appendCore(id: SessionId, events: readonly SessionEvent[]): Promise<void> {
    // Every append route converges here: the public service, live write-behind
-    // drains, and HMR seed/suffix adoption. Keep vocabulary rejection at that
-    // shared boundary so a stale JavaScript plugin cannot persist an event that
-    // this same backend will refuse to load.
+    // drains, and HMR seed/suffix adoption. Legacy-shape rejection stays at
+    // this shared boundary so a stale JavaScript plugin cannot persist a
+    // retired shape this backend refuses to load. The unknown-type guard is
+    // deliberately read-side only: an append-time refusal would stall a live
+    // session's durability mid-flight, which costs more than a loud refusal at
+    // the log's next load (trade-off owned by the session-log-version-mechanism
+    // Agent Note).
    assertSupportedEvents(events, id)
    if (events.length === 0) return
    this.preparations.assertWritable(id)
@@ -806,7 +860,9 @@ export class PersistenceCoordinator<TornMarker = unknown> {
        const whole = await this.readStoredPrefix(id, signal)
        return { meta: whole.meta, events: whole.events.filter(event => event.seq >= fromSeq) }
      }
-      return { meta: structuredClone(suffix.meta), events: snapshotStoredEvents(suffix.events, id) }
+      const events = snapshotStoredEvents(suffix.events, id)
+      this.assertEventsSupported(suffix.meta, events)
+      return { meta: structuredClone(suffix.meta), events }
    }
    const whole = await this.readStoredPrefix(id, signal)
    // Sequential fallback: contiguous seqs from 0 make the suffix an index slice.
@@ -824,9 +880,11 @@ export class PersistenceCoordinator<TornMarker = unknown> {
    if (stored === undefined) throw new Error(`session "${id}" not found`)
    this.assertStoredId(id, stored.meta)
    this.assertVersion(stored.meta)
+    const events = snapshotStoredEvents(stored.events, id)
+    this.assertEventsSupported(stored.meta, events)
    return {
      meta: structuredClone(stored.meta),
-      events: snapshotStoredEvents(stored.events, id),
+      events,
    }
  }

@@ -839,6 +897,7 @@ export class PersistenceCoordinator<TornMarker = unknown> {
      this.assertStoredId(id, meta)
      this.assertVersion(meta)
      const storedEvents = adoptStoredEvents(events, id)
+      this.assertEventsSupported(meta, storedEvents)

      // Preserve complete interrupted events and synthesize only missing closers.
      const closers = interruptedTurnClosers(storedEvents).map(adoptSessionEvent)
@@ -861,6 +920,9 @@ export class PersistenceCoordinator<TornMarker = unknown> {
        closers,
      }
    } catch (error: unknown) {
+      // An unsupported format is a refusal over an intact log, not damage —
+      // surface it unwrapped so callers can point at the raw artifact.
+      if (error instanceof SessionFormatUnsupportedError) throw error
      throw new SessionPersistenceCorruptionError(
        `stored session "${id}" failed validation: ${String(error)}`,
        { cause: error },
@@ -982,11 +1044,36 @@ export class PersistenceCoordinator<TornMarker = unknown> {
  }

  private assertVersion(meta: SessionHeader): void {
-    if (meta.version !== SESSION_FORMAT_VERSION) {
-      throw new Error(`unsupported session format version ${meta.version} for "${meta.id}" (only v${SESSION_FORMAT_VERSION} is supported)`)
+    if (meta.version === SESSION_FORMAT_VERSION) return
+    throw this.unsupported(meta, sessionFormatVersionRefusal(meta.id, meta.version))
+  }
+
+  /**
+   * Refuse a log containing an event type this build does not know, unless the
+   * writer marked the event ignorable: an unrecognized required event may
+   * change how the rest of the log must be interpreted, so silently skipping
+   * it would reconstruct a wrong session (the envelope contract on
+   * `SessionEvent.ignorable`). Runs on NORMALIZED events — after
+   * `snapshotStoredEvents`/`adoptStoredEvents` has upgraded the legacy shapes
+   * this build still reads and rejected the ones it does not, so those keep
+   * their specific diagnostics.
+   */
+  private assertEventsSupported(meta: SessionHeader, events: readonly SessionEvent[]): void {
+    for (const event of events) {
+      if (KNOWN_SESSION_EVENT_TYPES.has(event.type) || event.ignorable === true) continue
+      throw this.unsupported(meta, `session "${meta.id}" contains event type "${event.type}" (seq ${event.seq}) unknown to this harness and not marked ignorable; refusing to interpret the log — it was likely written by a newer harness`)
    }
  }

+  /** Build a format refusal that points at the raw artifact when the backend has one. */
+  private unsupported(meta: SessionHeader, reason: string): SessionFormatUnsupportedError {
+    const location = this.backend.locate?.(meta)
+    return new SessionFormatUnsupportedError(
+      location === undefined ? reason : `${reason} (raw log: ${location.path})`,
+      location,
+    )
+  }
+
  /** Reject backend metadata that is not bound to the requested session id. */
  private assertStoredId(id: SessionId, meta: SessionHeader): void {
    if (meta.id !== id) {
@@ -1219,6 +1306,7 @@ export class PersistenceCoordinator<TornMarker = unknown> {
    }
    this.assertVersion(meta)
    const storedEvents = snapshotStoredEvents(events, session.header.id)
+    this.assertEventsSupported(meta, storedEvents)
    if (!seedCoversPrefix(seed, storedEvents)) {
      throw new Error(`session "${session.header.id}" already has a persisted log on disk that does not match this live session (id collision)`)
    }
--- a/packages/session/session-persistence/src/index.ts
+++ b/packages/session/session-persistence/src/index.ts
@@ -36,7 +36,9 @@ export {
  DEFAULT_WRITE_BATCH_MAX_DELAY_MS,
  MAX_WRITE_BATCH_DELAY_MS,
  PersistenceCoordinator,
+  SessionFormatUnsupportedError,
  SessionPersistenceCorruptionError,
+  sessionFormatVersionRefusal,
 } from './coordinator.ts'
 export type {
  PersistenceBackend,
--- a/packages/session/session-persistence/tests/coordinator-contract.ts
+++ b/packages/session/session-persistence/tests/coordinator-contract.ts
@@ -706,6 +706,9 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
            .rejects.toThrow('lacks an identified message')
        }

+        // An out-of-repo event type passes only with the envelope's ignorable
+        // marker (unknown-type refusal otherwise), and its non-object data is
+        // not message-validated.
        const pluginId = SessionId('non-object-plugin-event')
        await ctx.sessionPersistence.create(meta(pluginId, WORK))
        await ctx.sessionPersistence.append(pluginId, [{
@@ -713,11 +716,12 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
          seq: 0,
          time: 1,
          data: null,
+          ignorable: true,
        } as unknown as SessionEvent])
        await expect(ctx.sessionPersistence.inspect(pluginId))
-          .resolves.toMatchObject({ events: [{ type: 'plugin/test', data: null }] })
+          .resolves.toMatchObject({ events: [{ type: 'plugin/test', data: null, ignorable: true }] })
        await expect(ctx.sessionPersistence.readFrom(pluginId, 0))
-          .resolves.toMatchObject({ events: [{ type: 'plugin/test', data: null }] })
+          .resolves.toMatchObject({ events: [{ type: 'plugin/test', data: null, ignorable: true }] })

        for (const type of ['user/message', 'assistant/message'] as const) {
          const missingContentId = SessionId(`invalid-${type}-without-content`)
@@ -1321,14 +1325,60 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
      }
    })

-    it('rejects an unknown format version on load (assertVersion)', async () => {
+    it('rejects a newer format version on load, naming the upgrade direction', async () => {
      const fix = await makeFixture()
      const { ctx, fiber } = await freshCtx(fix)
      try {
        const m = { version: 99, id: SessionId('v99'), createdAt: 1, cwd: WORK }
        await ctx.sessionPersistence.create(m)
        await ctx.sessionPersistence.append(m.id, oneTurnLog())
-        await expect(ctx.sessionPersistence.load(m.id)).rejects.toThrow(/version/)
+        const failure = await ctx.sessionPersistence.load(m.id).then(() => undefined, (error: unknown) => error as Error)
+        expect(failure?.name).toBe('SessionFormatUnsupportedError')
+        expect(failure?.message).toMatch(/written by a newer harness.*upgrade the harness/)
+      } finally {
+        await fiber.dispose()
+        await fix.cleanup()
+      }
+    })
+
+    it('rejects an older format version on load without claiming an upgrade path', async () => {
+      const fix = await makeFixture()
+      const { ctx, fiber } = await freshCtx(fix)
+      try {
+        const m = { version: -1, id: SessionId('v-older'), createdAt: 1, cwd: WORK }
+        await ctx.sessionPersistence.create(m)
+        await ctx.sessionPersistence.append(m.id, oneTurnLog())
+        const failure = await ctx.sessionPersistence.load(m.id).then(() => undefined, (error: unknown) => error as Error)
+        expect(failure?.name).toBe('SessionFormatUnsupportedError')
+        expect(failure?.message).toMatch(/older than the supported v0.*no upgrade path/)
+      } finally {
+        await fiber.dispose()
+        await fix.cleanup()
+      }
+    })
+
+    it('rejects an unknown event type on load unless the event is marked ignorable', async () => {
+      const fix = await makeFixture()
+      const { ctx, fiber } = await freshCtx(fix)
+      try {
+        const required = meta('unknown-required', WORK)
+        await ctx.sessionPersistence.create(required)
+        await ctx.sessionPersistence.append(required.id, [
+          ...oneTurnLog(),
+          { type: 'future/event', seq: oneTurnLog().length, time: 99, data: { payload: 1 } } as unknown as SessionEvent,
+        ])
+        const failure = await ctx.sessionPersistence.load(required.id).then(() => undefined, (error: unknown) => error as Error)
+        expect(failure?.name).toBe('SessionFormatUnsupportedError')
+        expect(failure?.message).toMatch(/event type "future\/event".*not marked ignorable/)
+
+        const skippable = meta('unknown-ignorable', WORK)
+        await ctx.sessionPersistence.create(skippable)
+        await ctx.sessionPersistence.append(skippable.id, [
+          ...oneTurnLog(),
+          { type: 'future/event', seq: oneTurnLog().length, time: 99, data: { payload: 1 }, ignorable: true } as unknown as SessionEvent,
+        ])
+        const loaded = await ctx.sessionPersistence.load(skippable.id)
+        expect(loaded.events.some(event => (event.type as string) === 'future/event')).toBe(true)
      } finally {
        await fiber.dispose()
        await fix.cleanup()