Merge latest master into subagent policy inheritance

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

@@ -22,6 +22,9 @@ class TestPersistence extends SessionPersistence {
   inspect(_id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }> {
     return Promise.reject(new Error('not used'))
   }
+  readFrom(_id: SessionId, _fromSeq: number): Promise<{ meta: SessionHeader; events: SessionEvent[] }> {
+    return Promise.reject(new Error('not used'))
+  }
   list(): Promise<SessionHeader[]> { return Promise.resolve([]) }
   listSnapshots(): Promise<never[]> { return Promise.resolve([]) }
 }

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

@@ -134,6 +134,12 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
     return this.coordinator.inspect(id, signal)
   }
 
+  // JSONL is sequential media: no loadStoredFrom hook, so the coordinator
+  // parses the stored prefix (both encodings) and skips forward to fromSeq.
+  readFrom(id: SessionId, fromSeq: number, signal?: AbortSignal): Promise<{ meta: SessionHeader; events: SessionEvent[] }> {
+    return this.coordinator.readFrom(id, fromSeq, signal)
+  }
+
   // One method serves both public `list` and the backend hook; delegating it to
   // the coordinator would call this hook recursively.
 

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

@@ -16,7 +16,7 @@ import { dirname, resolve } from 'node:path'
 import {
   SessionPersistence, SessionPersistenceRevision, PersistenceCoordinator,
   type PersistenceBackend, type SessionLocation, type SessionPersistenceSnapshot,
-  type StoredPrefix,
+  type StoredPrefix, type StoredSuffix,
 } from '@deepseek-ai/dsh-session-persistence'
 import type { SessionEvent, SurfaceEventType, SessionId, SessionHeader } from '@deepseek-ai/dsh-session'
 import {
@@ -161,6 +161,10 @@ export class SessionPersistenceSqlite extends SessionPersistence implements Pers
     return this.coordinator.inspect(id, signal)
   }
 
+  readFrom(id: SessionId, fromSeq: number, signal?: AbortSignal): Promise<{ meta: SessionHeader; events: SessionEvent[] }> {
+    return this.coordinator.readFrom(id, fromSeq, signal)
+  }
+
   // One method serves both public `list` and the backend hook; delegating it to
   // the coordinator would call this hook recursively.
 
@@ -171,6 +175,26 @@ export class SessionPersistenceSqlite extends SessionPersistence implements Pers
     return this.readPrefix(id, signal)
   }
 
+  /**
+   * Seek-capable suffix read: SQL selects `seq >= fromSeq` directly, so the
+   * read scales with the suffix, not the log. Torn rows past the preserved
+   * region are dropped, never repaired (non-mutating read).
+   */
+  async loadStoredFrom(id: SessionId, fromSeq: number, signal?: AbortSignal): Promise<StoredSuffix | undefined> {
+    signal?.throwIfAborted()
+    await this.ready
+    signal?.throwIfAborted()
+    const row = this.rowFor(id)
+    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')
+      .all(id, fromSeq) as unknown as EventRow[]
+    signal?.throwIfAborted()
+    const { preserved } = scanRows(eventRows, fromSeq)
+    return { meta, events: preserved }
+  }
+
   /**
    * Read a session's row + ordered events into a {@link StoredPrefix}. The
    * torn-tail marker is the seq from which a never-committed tail must be deleted

packages/session-persistence/session-persistence-sqlite/src/schema.ts

@@ -213,10 +213,12 @@ export function rowToEvent(row: EventRow): SessionEvent {
  * the committed region rejects.
  *
  * @param rows - one session's event rows, ordered by seq ascending.
+ * @param base - the seq the first row is expected to carry; `0` for a whole
+ *   log, the requested `fromSeq` for a suffix read (`loadStoredFrom`).
  * @returns the preserved event prefix, plus `tornFrom` — the seq the physical
  *   delete starts at — when a torn tail exists.
  */
-export function scanRows(rows: readonly EventRow[]): { preserved: SessionEvent[]; tornFrom?: number } {
+export function scanRows(rows: readonly EventRow[], base = 0): { preserved: SessionEvent[]; tornFrom?: number } {
   // Pass 1: parse each row's data; a row whose data is not valid JSON is a hole.
   // (The seq/type COLUMNS are always present even when `data` is corrupt.)
   interface Parsed { ok: boolean; event?: SessionEvent }
@@ -244,8 +246,8 @@ export function scanRows(rows: readonly EventRow[]): { preserved: SessionEvent[]
       if (i <= lastTurnEnd) throw new Error(`corrupt session log: unparsable committed event at seq ${rows[i]?.seq}`)
       break // torn tail fragment after the last turn/end — stop, tolerate
     }
-    if (p.event.seq !== i) {
-      if (i <= lastTurnEnd) throw new Error(`corrupt session log: seq gap in committed region (expected ${i}, got ${p.event.seq})`)
+    if (p.event.seq !== base + i) {
+      if (i <= lastTurnEnd) throw new Error(`corrupt session log: seq gap in committed region (expected ${base + i}, got ${p.event.seq})`)
       break // gap after the last turn/end — torn tail, stop
     }
     preserved.push(p.event)
@@ -253,5 +255,5 @@ export function scanRows(rows: readonly EventRow[]): { preserved: SessionEvent[]
 
   // Any rows past the preserved prefix are a never-committed torn tail; their
   // first seq is the deletion point for load's physical repair.
-  return preserved.length < rows.length ? { preserved, tornFrom: preserved.length } : { preserved }
+  return preserved.length < rows.length ? { preserved, tornFrom: base + preserved.length } : { preserved }
 }

packages/session-persistence/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-persistence/session-persistence/README.md
-README.md: 08d8adac8040747a6dac01dbc41525073f17060c
-README.zh.md: 7676f27a1aa934eb3472e1b32b9ecd55d460fb63
+README.md: a8a4f14c8613a7e51bcf467e816b7f7bdb7ea80b
+README.zh.md: 369e8a01b8ac411ed9acfbac86b9b34db8037c1f

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

@@ -13,8 +13,9 @@ The persisted unit IS the existing `SessionEvent` (event-sourced model — the l
 | `locate(meta): SessionLocation \| undefined` | Resolve an absolute per-session artifact target without I/O or materialization. Backends without an independent local artifact return `undefined`. |
 | `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. |
-| `load(id): Promise<{ meta; events }>` | Return a stored header plus a balanced contiguous log whose events are detached and validated and whose identified messages are deeply frozen. A live load first flushes its snapshot and rejects while its turn is open; a cold load preserves an interrupted final turn and closes it with synthetic `tool/result`/`step/end?`/`turn/end {interrupted}` events. Only a torn tail fragment is dropped; committed corruption, malformed messages, and unknown `version` reject. |
-| `inspect(id, signal?): Promise<{ meta; events }>` | Return a detached valid stored prefix with validated, deeply frozen identified messages, without truncating a torn tail, synthesizing recovery closers, or publishing coordinator state. Serialized with same-id writes; the optional signal promptly rejects a queued caller, prevents that queued backend read from starting, and cancels active backend read work. Intended for read models and other observers that must never recover a log. |
+| `load(id): Promise<{ meta; events }>` | Return a stored header plus a balanced contiguous log whose events are detached and validated and whose identified messages are deeply frozen. The coordinator upgrades the four pre-identity message event shapes into current wrappers in the returned snapshot; all other obsolete or malformed shapes still reject. A live load first flushes its snapshot and rejects while its turn is open; a cold load preserves an interrupted final turn and closes it with synthetic `tool/result`/`step/end?`/`turn/end {interrupted}` events. Only a torn tail fragment is dropped; committed corruption and unknown `version` reject. |
+| `inspect(id, signal?): Promise<{ meta; events }>` | Return a detached valid stored prefix with upgraded, validated, deeply frozen identified messages, without truncating a torn tail, synthesizing recovery closers, or publishing coordinator state. Serialized with same-id writes; the optional signal promptly rejects a queued caller, prevents that queued backend read from starting, and cancels active backend read work. Intended for read models and other observers that must never recover a log. |
+| `readFrom(id, fromSeq, signal?): Promise<{ meta; events }>` | The read-from-seq primitive: return the header plus the valid stored events with `seq >= fromSeq`, detached and non-mutating like `inspect` (no truncation, no closers, no 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; sequential backends (JSONL) still parse the whole artifact and skip forward — the primitive bounds what is returned and refolded, not every backend's physical read. Intended for checkpoint consumers (e.g. the persisted projection cache) that fold only the tail past a watermark. |
 | `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. |
 
@@ -33,6 +34,8 @@ Each `session/event` copies its event into the session controller and starts an
 
 Crash repair is cold-only. For a live id, `load(id)` snapshots the authoritative in-memory log, waits for that snapshot to become durable, and returns it with the coordinator's stored header only when balanced; an open live turn rejects instead of receiving synthetic interruption closers. A cold load reserves its id across backend reads and repair writes, so concurrent publication of a same-id live `Session` rejects and rolls back. HMR adoption reads through `loadStored`, applies the coordinator's cwd check, and never closes the active turn.
 
+Backend reads normalize pre-identity `user/message`, `assistant/message`, `tool/result`, and `steering/message` payloads before current-shape validation. Each imported message receives the deterministic id `legacy-message:<session-id>:<event-seq>`; a tool-result content replacement inherits its target's imported id. The coordinator uses the same normalized view for `load`, `inspect`, ownerless-state claims, and HMR prefix adoption, so resumed sessions can append current events without a false prefix collision. Storage remains append-only: the read does not rewrite old records, and every later append uses the current shape. This is the narrow import exception from the [pre-identity message recovery decision](../../../.agents/notes/implemented/bug-fix/2026-07-28-load-pre-identity-session-messages.md), not a general v0 migration promise.
+
 When a live session emits `session/disposed`, the coordinator waits for its controller, serializes a final drain, then releases state owned by that exact `Session` object. Failed retirement leaves the controller in the live-session map, so backend teardown can retry it. Backend teardown stops event admission first, flushes every remaining controller, awaits per-id operations, and only then closes the storage handle.
 
 The side-effect-free `locate` and lightweight `listSnapshots` queries remain backend-owned because they describe storage topology and revision identity rather than write orchestration. `listSnapshots(signal?)` passes the caller's exact signal into backend discovery so observers can cancel that work without detaching it.
@@ -43,6 +46,7 @@ The `PersistenceBackend<TornMarker>` hooks (the only seam between the coordinato
 |---|---|
 | `name` | Backend label for the dispose-failure `AggregateError`. |
 | `loadStored(id, signal?)` | Read a stored prefix by id across every storage scope. Used by resume/load, non-mutating inspect, live adoption, and the create-collision probe. The optional signal belongs to observation-only reads. Returned metadata identifies `id`; an opaque `tornMarker` is present iff a torn tail must be truncated. |
+| `loadStoredFrom?(id, fromSeq, signal?)` | Optional seek-capable suffix read behind the service's `readFrom`: the header plus stored events with `seq >= fromSeq`, non-mutating, no torn marker. SQLite implements it (`WHERE seq >= ?`); a backend that omits it gets the coordinator's fallback — `loadStored` plus a forward skip. |
 | `appendBatch(meta, events, isMaterialized)` | Durably append a contiguous batch, lazily materializing ATOMICALLY when not yet materialized. |
 | `commitRepair(meta, tornMarker, closers)` | Make a crash repair durable: truncate the torn tail (iff `tornMarker !== undefined` — a marker may be falsy, e.g. seq/offset `0`) and append `closers`. NOT required to be atomic. Used by load (truncate + closers) and live-adoption (truncate only). |
 | `list(signal?)` | List all stored metadata, observing optional cancellation. |

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

@@ -13,8 +13,9 @@
 | `locate(meta): SessionLocation \| undefined` | 在不执行 I/O 或实体化的情况下解析绝对的每会话产物目标。没有独立本地产物的后端返回 `undefined`。 |
 | `create(meta): Promise<void>` | 注册新会话元数据。可以将物理写入延迟到第一次 `append`（延迟实体化）。 |
 | `append(id, events): Promise<void>` | 持久保存一个批次。仅追加；任何修复后，第一个事件 `seq` == 已存储 next-seq；非 JSON 可序列化数据会被拒绝，并命名违规类型。 |
-| `load(id): Promise<{ meta; events }>` | 返回已存储 header 和平衡、连续的日志，其中事件已脱离并验证，带标识的消息已深度冻结。实时 load 先 flush 其快照，并在轮次开放时拒绝；冷 load 保留中断的最终轮次，并用合成 `tool/result`/`step/end?`/`turn/end {interrupted}` 事件关闭它。只丢弃撕裂尾部碎片；已提交损坏、格式错误的消息和未知 `version` 会被拒绝。 |
-| `inspect(id, signal?): Promise<{ meta; events }>` | 返回脱离的有效已存储前缀，其中带标识的消息已经验证并深度冻结；不截断撕裂尾部、合成恢复 closer 或发布协调器状态。它与同 id 写入串行化；可选信号会迅速拒绝已排队调用方，阻止该后端读取启动，并取消活动后端读取工作。用于绝不应恢复日志的读模型和其他观察者。 |
+| `load(id): Promise<{ meta; events }>` | 返回已存储 header 和平衡、连续的日志，其中事件已脱离并验证，带标识的消息已深度冻结。协调器会在返回快照中，将消息标识机制引入前的四种消息事件形状升级为当前包装层；其余过时或格式错误的形状仍会被拒绝。实时 load 先 flush 其快照，并在轮次开放时拒绝；冷 load 保留中断的最终轮次，并用合成 `tool/result`/`step/end?`/`turn/end {interrupted}` 事件关闭它。只丢弃撕裂尾部碎片；已提交损坏和未知 `version` 会被拒绝。 |
+| `inspect(id, signal?): Promise<{ meta; events }>` | 返回脱离的有效已存储前缀，其中带标识的消息已经升级、验证并深度冻结；不截断撕裂尾部、合成恢复 closer 或发布协调器状态。它与同 id 写入串行化；可选信号会迅速拒绝已排队调用方，阻止该后端读取启动，并取消活动后端读取工作。用于绝不应恢复日志的读模型和其他观察者。 |
+| `readFrom(id, fromSeq, signal?): Promise<{ meta; events }>` | read-from-seq 原语：返回 header 和 `seq >= fromSeq` 的有效已存储事件，与 `inspect` 同样脱离且非变更（不截断、不合成 closer、不发布协调器状态）。`fromSeq` 达到或超过已存储末尾时返回空事件列表；负数或非安全整数 `fromSeq` 会被拒绝。可寻址后端（SQLite）只读后缀；顺序后端（JSONL）仍解析整个产物并向前跳过——原语约束的是返回和重折叠的量，不是每个后端的物理读取。用于从水位续折尾部的 checkpoint 消费者（例如持久投影缓存）。 |
 | `list(signal?): Promise<SessionHeader[]>` | 从元数据轻量列出，不解析完整日志。可选信号取消后端列表工作。零事件延迟实体化会话不在 `list` 中。 |
 | `listSnapshots(signal?): Promise<SessionPersistenceSnapshot[]>` | 返回轻量元数据和不透明品牌化每日志修订，不加载事件日志。日志及其后端存储不变时，修订保持相等；append 或变更性 load 修复后会改变；不会仅因两个存储使用相同本地计数器而冲突。可选信号请求取消后端发现工作；第一方后端在拒绝前结算已启动列表工作，使已等待调用完全停稳。 |
 
@@ -33,6 +34,8 @@
 
 崩溃修复只适用于冷状态。对于实时 id，`load(id)` 为权威内存日志制作快照，等待该快照持久，并只在平衡时将其与协调器已存储 header 一起返回；开放实时轮次会被拒绝，而不会收到合成中断 closer。冷 load 在后端读取和修复写入期间保留 id，因此同 id 实时 `Session` 的并发发布会拒绝并回滚。HMR 接管通过 `loadStored` 读取，应用协调器 cwd 检查，并绝不关闭活动轮次。
 
+后端读取会在当前形状验证前，规范化消息标识机制引入前的 `user/message`、`assistant/message`、`tool/result` 以及 steering（中途引导）对应的 `steering/message` 载荷。每条导入消息都会获得确定性的 id `legacy-message:<session-id>:<event-seq>`；工具结果的内容替换会继承其目标导入后的 id。协调器对 `load`、`inspect`、无 owner 状态的认领和 HMR 前缀接管使用同一份规范化视图，因此恢复后的会话可以追加当前事件，不会被误判为发生前缀冲突。存储仍然仅追加：读取不会重写旧记录，此后追加的每个事件都使用当前形状。这是[消息标识机制引入前的消息恢复决策](../../../.agents/notes/implemented/bug-fix/2026-07-28-load-pre-identity-session-messages.md)所规定的范围受限的导入例外，并不构成通用的 v0 迁移承诺。
+
 实时会话发出 `session/disposed` 时，协调器等待其 controller，串行化最终 drain，然后释放该精确 `Session` 对象拥有的状态。失败退役会将 controller 保留在实时会话 map 中，使后端拆卸可重试。后端拆卸先停止事件接纳，flush 每个剩余 controller，等待每 id 操作，最后才关闭存储句柄。
 
 无副作用 `locate` 和轻量 `listSnapshots` 查询仍由后端负责，因为它们描述存储拓扑和修订身份，而非写入编排。`listSnapshots(signal?)` 将调用方的精确信号传入后端发现，使观察者可在不脱离该工作的情况下取消。
@@ -43,6 +46,7 @@
 |---|---|
 | `name` | dispose 失败 `AggregateError` 的后端标签。 |
 | `loadStored(id, signal?)` | 在全部存储范围中按 id 读取已存储前缀。用于 resume/load、非变更 inspect、实时接管和 create 冲突探测。可选信号属于仅观察读取。返回元数据标识 `id`；当且仅当必须截断撕裂尾部时才存在不透明 `tornMarker`。 |
+| `loadStoredFrom?(id, fromSeq, signal?)` | 服务 `readFrom` 背后的可选可寻址后缀读取：返回 header 和 `seq >= fromSeq` 的已存储事件，非变更、无撕裂标记。SQLite 实现它（`WHERE seq >= ?`）；不实现的后端使用协调器回退——`loadStored` 加向前跳过。 |
 | `appendBatch(meta, events, isMaterialized)` | 持久追加连续批次；尚未实体化时以原子方式延迟实体化。 |
 | `commitRepair(meta, tornMarker, closers)` | 使崩溃修复持久：截断撕裂尾部（当且仅当 `tornMarker !== undefined`；标记可为 falsy，例如 seq/offset `0`），并追加 `closers`。不要求原子性。由 load（截断 + closer）和实时接管（仅截断）使用。 |
 | `list(signal?)` | 列出全部已存储元数据，观察可选取消。 |

packages/session-persistence/session-persistence/src/coordinator.ts

@@ -25,6 +25,17 @@ export interface StoredPrefix<TornMarker = unknown> {
   tornMarker?: TornMarker
 }
 
+/**
+ * A stored session's header plus the events at or past a requested seq — the
+ * return shape of the optional seek-capable
+ * {@link PersistenceBackend.loadStoredFrom} hook. Non-mutating reads carry no
+ * torn marker: there is nothing to repair.
+ */
+export interface StoredSuffix {
+  meta: SessionHeader
+  events: SessionEvent[]
+}
+
 /**
  * The storage seam between {@link PersistenceCoordinator} and a concrete
  * backend: the minimal set of durable primitives the orchestration calls. A
@@ -50,6 +61,22 @@ export interface PersistenceBackend<TornMarker = unknown> {
    */
   loadStored(id: SessionId, signal?: AbortSignal): Promise<StoredPrefix<TornMarker> | undefined>
 
+  /**
+   * Optional seek-capable suffix read behind the service's `readFrom`: return
+   * the header plus the stored events with `seq >= fromSeq` without reading
+   * the whole log. A backend whose medium can address events by seq (SQLite)
+   * implements this so `readFrom` scales with the suffix; sequential backends
+   * omit it and the coordinator falls back to {@link loadStored} plus a
+   * forward skip. Non-mutating (no truncation, no closers). Validation of the
+   * region strictly below `fromSeq` is limited to seq contiguity — the
+   * service contract scopes this read to the suffix.
+   * @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).
+   * @param signal - optional cancellation for backend read work.
+   */
+  loadStoredFrom?(id: SessionId, fromSeq: number, signal?: AbortSignal): Promise<StoredSuffix | undefined>
+
   /**
    * Durably append a CONTIGUOUS batch, lazily materializing the session first
    * when `!isMaterialized`. The materialize-write and the first event batch MUST
@@ -146,10 +173,142 @@ function assertSupportedEvents(events: readonly SessionEvent[], id: SessionId):
   }
 }
 
-/** Materialize stored events as validated snapshots with immutable messages. */
+/** Return an object record without widening arrays into message payloads. */
+function asRecord(value: unknown): Record<string, unknown> | undefined {
+  return typeof value === 'object' && value !== null && !Array.isArray(value)
+    ? value as Record<string, unknown>
+    : undefined
+}
+
+type PersistedMessageId = SessionEvent<'user/message'>['data']['id']
+
+/** Mint the stable import identity for a message persisted before identities existed. */
+function legacyMessageId(id: SessionId, seq: number): PersistedMessageId {
+  return `legacy-message:${id}:${seq}` as PersistedMessageId
+}
+
+/** Read a replacement target while leaving malformed surface metadata to the session validator. */
+function replacementStart(event: SessionEvent): number | undefined {
+  const op = asRecord((event as SessionEvent & { surfaceOp?: unknown }).surfaceOp)
+  return op?.['op'] === 'replace' && typeof op['start'] === 'number'
+    ? op['start']
+    : undefined
+}
+
+/**
+ * Upgrade one pre-identity message event into the current wrapper shape.
+ * Current-looking malformed events remain untouched so validation rejects them
+ * instead of disguising corruption as legacy data.
+ */
+function migrateLegacyMessageEvent(
+  event: SessionEvent,
+  id: SessionId,
+  messageIds: ReadonlyMap<number, PersistedMessageId>,
+): SessionEvent {
+  const data = asRecord(event.data)
+  if (data === undefined) return event
+  switch (event.type) {
+    case 'user/message': {
+      if (Object.hasOwn(data, 'id') || Object.hasOwn(data, 'role')
+        || Object.hasOwn(data, 'message')
+        || !Object.hasOwn(data, 'content') || !Object.hasOwn(data, 'source')) return event
+      return {
+        ...event,
+        data: {
+          ...data,
+          id: legacyMessageId(id, event.seq),
+          role: 'user',
+        },
+      } as SessionEvent
+    }
+    case 'assistant/message': {
+      if (Object.hasOwn(data, 'message')
+        || !Object.hasOwn(data, 'content') || !Object.hasOwn(data, 'provenance')) return event
+      const { content, provenance, ...eventData } = data
+      return {
+        ...event,
+        data: {
+          ...eventData,
+          message: {
+            id: legacyMessageId(id, event.seq),
+            role: 'assistant',
+            content,
+            source: {
+              ...asRecord(provenance),
+              kind: 'model',
+            },
+          },
+        },
+      } as SessionEvent
+    }
+    case 'tool/result': {
+      if (Object.hasOwn(data, 'message')
+        || !Object.hasOwn(data, 'callId') || !Object.hasOwn(data, 'content')
+        || !Object.hasOwn(data, 'isError')) return event
+      const { callId, content, isError, ...eventData } = data
+      const inheritedId = replacementStart(event)
+      return {
+        ...event,
+        data: {
+          ...eventData,
+          message: {
+            id: inheritedId === undefined
+              ? legacyMessageId(id, event.seq)
+              : messageIds.get(inheritedId),
+            role: 'user',
+            content: [{
+              type: 'tool-result',
+              toolCallId: callId,
+              content,
+              isError,
+            }],
+            source: {
+              kind: 'tool',
+              callId,
+            },
+          },
+        },
+      } as SessionEvent
+    }
+    case 'steering/message': {
+      if (Object.hasOwn(data, 'message')
+        || !Object.hasOwn(data, 'content') || !Object.hasOwn(data, 'source')) return event
+      const { content, source, ...eventData } = data
+      return {
+        ...event,
+        data: {
+          ...eventData,
+          message: {
+            id: legacyMessageId(id, event.seq),
+            role: 'user',
+            content,
+            source,
+          },
+        },
+      } as SessionEvent
+    }
+    default:
+      return event
+  }
+}
+
+/** Read the identified message carried by one validated current event. */
+function eventMessageId(event: SessionEvent): PersistedMessageId | undefined {
+  const data = asRecord(event.data)
+  const message = event.type === 'user/message' ? data : asRecord(data?.['message'])
+  return typeof message?.['id'] === 'string' ? message['id'] as PersistedMessageId : undefined
+}
+
+/** Materialize stored events as upgraded, validated snapshots with immutable messages. */
 function snapshotStoredEvents(events: readonly SessionEvent[], id: SessionId): SessionEvent[] {
   assertSupportedEvents(events, id)
-  return events.map(snapshotSessionEvent)
+  const messageIds = new Map<number, PersistedMessageId>()
+  return events.map((event) => {
+    const snapshot = snapshotSessionEvent(migrateLegacyMessageEvent(event, id, messageIds))
+    const messageId = eventMessageId(snapshot)
+    if (messageId !== undefined) messageIds.set(snapshot.seq, messageId)
+    return snapshot
+  })
 }
 
 /**
@@ -325,6 +484,52 @@ export class PersistenceCoordinator<TornMarker = unknown> {
     }
   }
 
+  /**
+   * Read the stored events from `fromSeq` onward, detached and non-mutating
+   * (the read-from-seq primitive behind the service's `readFrom`). Runs on
+   * the same per-id chain as writes; a backend with the seek-capable
+   * {@link PersistenceBackend.loadStoredFrom} hook reads only the suffix,
+   * every other backend reads its stored prefix and skips forward here.
+   * @param id - persisted session to read.
+   * @param fromSeq - first event seq to include; a non-negative safe integer.
+   * @param signal - optional cancellation for queued and backend read work.
+   * @returns stored header and the valid stored events with `seq >= fromSeq`.
+   */
+  readFrom(id: SessionId, fromSeq: number, signal?: AbortSignal): Promise<{ meta: SessionHeader; events: SessionEvent[] }> {
+    if (!Number.isSafeInteger(fromSeq) || fromSeq < 0) {
+      return Promise.reject(new TypeError(`readFrom fromSeq must be a non-negative safe integer, got ${String(fromSeq)}`))
+    }
+    const retired = Promise.resolve(this.retirements.get(id))
+    const waited = signal === undefined ? retired : observeQueuedAbort(retired, signal, () => false)
+    return waited.then(() => this.serialize(id, () => this.readFromCore(id, fromSeq, signal), signal))
+  }
+
+  private async readFromCore(
+    id: SessionId,
+    fromSeq: number,
+    signal?: AbortSignal,
+  ): Promise<{ meta: SessionHeader; events: SessionEvent[] }> {
+    signal?.throwIfAborted()
+    if (this.backend.loadStoredFrom !== undefined) {
+      let suffix: StoredSuffix | undefined
+      try {
+        suffix = await this.backend.loadStoredFrom(id, fromSeq, signal)
+      } catch (error: unknown) {
+        if (signal?.aborted) signal.throwIfAborted()
+        throw error
+      }
+      signal?.throwIfAborted()
+      if (suffix === undefined) throw new Error(`session "${id}" not found`)
+      this.assertStoredId(id, suffix.meta)
+      this.assertVersion(suffix.meta)
+      assertSupportedEvents(suffix.events, id)
+      return { meta: structuredClone(suffix.meta), events: structuredClone(suffix.events) }
+    }
+    const whole = await this.inspectCore(id, signal)
+    // Sequential fallback: contiguous seqs from 0 make the suffix an index slice.
+    return { meta: whole.meta, events: whole.events.slice(fromSeq) }
+  }
+
   private async loadCore(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }> {
     const stored = await this.backend.loadStored(id)
     if (stored === undefined) throw new Error(`session "${id}" not found`)
@@ -526,7 +731,7 @@ export class PersistenceCoordinator<TornMarker = unknown> {
     /* v8 ignore next -- a cursor > 0 means the session was materialized, so it exists */
     if (stored === undefined) return false
     this.assertStoredId(id, stored.meta)
-    return seedCoversPrefix(seed, stored.events.slice(0, cursor))
+    return seedCoversPrefix(seed, snapshotStoredEvents(stored.events, id).slice(0, cursor))
   }
 
   /**
@@ -614,19 +819,19 @@ export class PersistenceCoordinator<TornMarker = unknown> {
       throw new Error(`session "${session.header.id}" is already persisted at a different cwd (persisted: ${String(meta.cwd)}, live: ${String(session.header.cwd)}) (id collision)`)
     }
     this.assertVersion(meta)
-    assertSupportedEvents(events, session.header.id)
-    if (!seedCoversPrefix(seed, events)) {
+    const storedEvents = snapshotStoredEvents(events, session.header.id)
+    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)`)
     }
     // Truncate-only repair (no closers): the open turn is NOT closed here.
     if (tornMarker !== undefined) await this.backend.commitRepair(meta, tornMarker, [])
     this.states.set(session.header.id, {
       meta: { ...meta },
-      cursor: events.length,
+      cursor: storedEvents.length,
       materialized: true,
       owner: session,
     })
-    const suffix = seed.slice(events.length)
+    const suffix = seed.slice(storedEvents.length)
     if (suffix.length > 0) await this.appendCore(session.header.id, suffix)
   }
 

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

@@ -23,7 +23,7 @@ export interface SessionPersistenceSnapshot {
 
 // The backend-agnostic write-path orchestration first-party backends compose.
 export { PersistenceCoordinator } from './coordinator.ts'
-export type { PersistenceBackend, StoredPrefix } from './coordinator.ts'
+export type { PersistenceBackend, StoredPrefix, StoredSuffix } from './coordinator.ts'
 
 declare module 'cordis' {
   interface Context {
@@ -93,7 +93,9 @@ export abstract class SessionPersistence extends Service {
    * A coordinator-backed cold load reserves the identity across storage awaits,
    * so concurrent publication of a same-id live Session rejects.
    * Returned events are detached, and every identified message is deeply
-   * frozen; malformed identified messages reject before any stored event is returned.
+   * frozen. Coordinator-backed implementations upgrade supported pre-identity
+   * message events before validation; other malformed messages reject before
+   * any stored event is returned.
    * @param id - the persisted session to reload.
    * @returns the header and a log ending on a balanced `turn/end`.
    */
@@ -103,14 +105,35 @@ export abstract class SessionPersistence extends Service {
    * Inspect a header and its valid contiguous stored prefix without repairing
    * a torn tail, closing an interrupted turn, or publishing coordinator state.
    * This read is serialized with writes for the same id and returns detached
-   * values with deeply frozen identified messages, so observers cannot mutate message
-   * identity/content or backend-owned state. Malformed identified messages reject.
+   * values with upgraded, deeply frozen identified messages, so observers
+   * cannot mutate message identity/content or backend-owned state. Other
+   * malformed messages reject.
    * @param id - the persisted session to inspect.
    * @param signal - optional cancellation for queued and backend read work.
    * @returns the header and valid stored event prefix exactly as observed.
    */
   abstract inspect(id: SessionId, signal?: AbortSignal): Promise<{ meta: SessionHeader; events: SessionEvent[] }>
 
+  /**
+   * Read the stored events from `fromSeq` onward — the read-from-seq
+   * primitive for read models that resume from a watermark (e.g. a persisted
+   * projection cache folding only the tail past its checkpoint). Like
+   * {@link inspect} it is non-mutating and detached: no torn-tail truncation,
+   * no synthetic closers, no coordinator-state publication; only events from
+   * the valid contiguous stored prefix are returned, so a torn fragment never
+   * reaches the caller. `fromSeq` at or beyond the stored prefix returns an
+   * empty event list (never an error). Backends whose medium can seek by seq
+   * (SQLite) read only the suffix; sequential media (JSONL, both encodings)
+   * still parse the whole artifact and skip forward — the primitive bounds
+   * what is RETURNED and refolded, not every backend's physical read.
+   * @param id - the persisted session to read.
+   * @param fromSeq - first event seq to include; a non-negative safe integer.
+   * @param signal - optional cancellation for queued and backend read work.
+   * @returns the header and the stored events with `seq >= fromSeq`.
+   */
+  abstract readFrom(id: SessionId, fromSeq: number, signal?: AbortSignal):
+  Promise<{ meta: SessionHeader; events: SessionEvent[] }>
+
   /**
    * Lightweight listing from metadata, without a full-log parse.
    * @param signal - optional cancellation for backend listing work.

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

@@ -289,6 +289,43 @@ export function runPersistenceContract(name: string, make: () => Promise<Contrac
         await expect(persistence.listSnapshots(controller.signal)).rejects.toBe(reason)
         await expect(persistence.inspect(SessionId('cancelled-inspect'), controller.signal))
           .rejects.toBe(reason)
+        await expect(persistence.readFrom(SessionId('cancelled-read-from'), 0, controller.signal))
+          .rejects.toBe(reason)
+      } finally {
+        await dispose()
+      }
+    })
+
+    it('readFrom returns exactly the stored suffix from the requested seq, without mutating the log', async () => {
+      const { persistence, dispose } = await make()
+      try {
+        const m = meta('read-from', '/work')
+        const log = oneTurnLog()
+        await persistence.create(m)
+        await persistence.append(m.id, log)
+
+        const whole = await persistence.readFrom(m.id, 0)
+        expect(whole.meta).toMatchObject({ id: m.id, cwd: '/work' })
+        expect(whole.events).toEqual(log)
+
+        const suffix = await persistence.readFrom(m.id, 3)
+        expect(suffix.events).toEqual(log.slice(3))
+        expect(suffix.events[0]?.seq).toBe(3)
+
+        // At/past the stored end: an empty tail, never an error.
+        await expect(persistence.readFrom(m.id, log.length)).resolves.toMatchObject({ events: [] })
+        await expect(persistence.readFrom(m.id, log.length + 100)).resolves.toMatchObject({ events: [] })
+
+        // Non-mutating: an interrupted-turn log is served as stored, no closers.
+        await persistence.append(m.id, [
+          { type: 'turn/start', seq: 6, time: 7, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } },
+        ])
+        const tail = await persistence.readFrom(m.id, 6)
+        expect(tail.events.map(event => event.type)).toEqual(['turn/start'])
+
+        await expect(persistence.readFrom(SessionId('absent-read-from'), 0)).rejects.toThrow('not found')
+        await expect(persistence.readFrom(m.id, -1)).rejects.toThrow('non-negative safe integer')
+        await expect(persistence.readFrom(m.id, 1.5)).rejects.toThrow('non-negative safe integer')
       } finally {
         await dispose()
       }

packages/session-persistence/session-persistence/tests/coordinator-contract.ts

@@ -13,8 +13,8 @@ import { createUserMessage } from '@deepseek-ai/dsh-llm'
 import { describe, expect, it, vi } from 'vitest'
 import { Context, type Fiber } from 'cordis'
 import { scopeTarget } from '@deepseek-ai/dsh-scope'
-import SessionStore, { SESSION_FORMAT_VERSION, SessionId } from '@deepseek-ai/dsh-session'
-import type { Session, SessionEvent } from '@deepseek-ai/dsh-session'
+import SessionStore, { SESSION_FORMAT_VERSION, Session, SessionId } from '@deepseek-ai/dsh-session'
+import type { SessionEvent } from '@deepseek-ai/dsh-session'
 import { meta, oneTurnLog, appendLog } from './contract.ts'
 
 /**
@@ -45,6 +45,80 @@ function send(session: Session, events: readonly SessionEvent[]): void {
   appendLog(session, events)
 }
 
+/** A valid persisted log from immediately before messages gained wrappers and identities. */
+function legacyMessageLog(): SessionEvent[] {
+  return [
+    { type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } },
+    {
+      type: 'user/message',
+      seq: 1,
+      time: 2,
+      data: { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } },
+      surfaceOp: 'append',
+    },
+    { type: 'step/start', seq: 2, time: 3, data: { turn: 1, step: 1 } },
+    {
+      type: 'assistant/message',
+      seq: 3,
+      time: 4,
+      data: {
+        turn: 1,
+        step: 1,
+        content: [{ type: 'tool-call', id: 'call-1', name: 'read', arguments: '{}' }],
+        provenance: { provider: 'mock', model: 'mock' },
+      },
+      surfaceOp: 'append',
+    },
+    {
+      type: 'tool/call',
+      seq: 4,
+      time: 5,
+      data: { turn: 1, step: 1, callId: 'call-1', name: 'read', arguments: '{}' },
+    },
+    {
+      type: 'tool/result',
+      seq: 5,
+      time: 6,
+      data: {
+        turn: 1,
+        step: 1,
+        callId: 'call-1',
+        content: [{ type: 'text', text: 'full result' }],
+        isError: false,
+      },
+      sourceEventSeqs: [4],
+      surfaceOp: 'append',
+    },
+    {
+      type: 'steering/message',
+      seq: 6,
+      time: 7,
+      data: {
+        turn: 1,
+        content: [{ type: 'text', text: 'continue' }],
+        source: { kind: 'plugin', plugin: 'test' },
+      },
+      surfaceOp: 'append',
+    },
+    {
+      type: 'tool/result',
+      seq: 7,
+      time: 8,
+      data: {
+        turn: 1,
+        step: 1,
+        callId: 'call-1',
+        content: [{ type: 'text', text: 'pruned' }],
+        isError: false,
+      },
+      sourceEventSeqs: [5],
+      surfaceOp: { op: 'replace', start: 5, end: 5 },
+    },
+    { type: 'step/end', seq: 8, time: 9, data: { turn: 1, step: 1 } },
+    { type: 'turn/end', seq: 9, time: 10, data: { turn: 1, reason: { kind: 'completed' } } },
+  ] as unknown as SessionEvent[]
+}
+
 /** A live session created inside its OWN fiber, so it survives a backend reload. */
 async function liveSessionInFiber(
   ctx: Context, id: string, cwd: string | undefined,
@@ -269,6 +343,48 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
       }
     })
 
+    it('loads pre-identity message logs into resumable current sessions', async () => {
+      const fix = await makeFixture()
+      const { ctx, fiber } = await freshCtx(fix)
+      try {
+        const id = SessionId('legacy-message-load')
+        await ctx.sessionPersistence.create(meta(id, WORK))
+        await ctx.sessionPersistence.append(id, legacyMessageLog())
+
+        for (const snapshot of [
+          await ctx.sessionPersistence.inspect(id),
+          await ctx.sessionPersistence.load(id),
+        ]) {
+          const messages = snapshot.events.flatMap((event) => {
+            if (event.type === 'user/message') return [event.data]
+            if (event.type === 'assistant/message'
+              || event.type === 'tool/result'
+              || event.type === 'steering/message') return [event.data.message]
+            return []
+          })
+          expect(messages.map(message => message.id)).toEqual([
+            `legacy-message:${id}:1`,
+            `legacy-message:${id}:3`,
+            `legacy-message:${id}:5`,
+            `legacy-message:${id}:6`,
+            `legacy-message:${id}:5`,
+          ])
+          expect(messages.every(message => Object.isFrozen(message))).toBe(true)
+
+          const resumed = new Session(id, snapshot.events, snapshot.meta)
+          expect(resumed.deriveMessages().map(message => message.id)).toEqual([
+            `legacy-message:${id}:1`,
+            `legacy-message:${id}:3`,
+            `legacy-message:${id}:5`,
+            `legacy-message:${id}:6`,
+          ])
+        }
+      } finally {
+        await fiber.dispose()
+        await fix.cleanup()
+      }
+    })
+
     it('rejects malformed persisted message events before returning them', async () => {
       const fix = await makeFixture()
       const { ctx, fiber } = await freshCtx(fix)
@@ -292,6 +408,31 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
           .rejects.toThrow('message must have role "user"')
         await expect(ctx.sessionPersistence.load(id))
           .rejects.toThrow('message must have role "user"')
+
+        for (const type of ['tool/result', 'steering/message'] as const) {
+          const malformedId = SessionId(`invalid-${type}`)
+          await ctx.sessionPersistence.create(meta(malformedId, WORK))
+          await ctx.sessionPersistence.append(malformedId, [{
+            type,
+            seq: 0,
+            time: 1,
+            surfaceOp: 'append',
+            data: { message: null },
+          } as unknown as SessionEvent])
+          await expect(ctx.sessionPersistence.inspect(malformedId))
+            .rejects.toThrow('lacks an identified message')
+        }
+
+        const pluginId = SessionId('non-object-plugin-event')
+        await ctx.sessionPersistence.create(meta(pluginId, WORK))
+        await ctx.sessionPersistence.append(pluginId, [{
+          type: 'plugin/test',
+          seq: 0,
+          time: 1,
+          data: null,
+        } as unknown as SessionEvent])
+        await expect(ctx.sessionPersistence.inspect(pluginId))
+          .resolves.toMatchObject({ events: [{ type: 'plugin/test', data: null }] })
       } finally {
         await fiber.dispose()
         await fix.cleanup()

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

@@ -99,6 +99,10 @@ class MemoryPersistence extends SessionPersistence implements PersistenceBackend
     return this.coordinator.inspect(id, signal)
   }
 
+  readFrom(id: SessionId, fromSeq: number, signal?: AbortSignal): Promise<{ meta: SessionHeader; events: SessionEvent[] }> {
+    return this.coordinator.readFrom(id, fromSeq, signal)
+  }
+
   // --- PersistenceBackend hooks (the Map storage primitives) ---
 
   // A Map-backed store has no torn tails, so `tornMarker` is never set.
@@ -157,6 +161,13 @@ class ControlledBackend implements PersistenceBackend<never> {
   repairAttempts = 0
   beforeAppend?: (attempt: number) => Promise<void>
   beforeLoadStored?: (attempt: number, signal?: AbortSignal) => Promise<void>
+  /** When set, the declared seek hook delegates here so readFrom exercises it; unset throws (tests set it first). */
+  seekHook?: (id: SessionId, fromSeq: number, signal?: AbortSignal) => Promise<StoredPrefix<never> | undefined>
+
+  loadStoredFrom(id: SessionId, fromSeq: number, signal?: AbortSignal): Promise<StoredPrefix<never> | undefined> {
+    if (this.seekHook === undefined) throw new Error('seekHook not configured for this test')
+    return this.seekHook(id, fromSeq, signal)
+  }
 
   async loadStored(id: SessionId, signal?: AbortSignal): Promise<StoredPrefix<never> | undefined> {
     await this.beforeLoadStored?.(++this.loadAttempts, signal)
@@ -453,6 +464,58 @@ describe('PersistenceCoordinator observation cancellation', () => {
     }
   })
 
+  it('readFrom via the seek hook: serves the suffix, maps undefined to not-found, and relays hook failures by abort state', async () => {
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    const backend = new ControlledBackend()
+    const id = SessionId('seek-read-from')
+    const log = oneTurnLog()
+    backend.store.set(id, { meta: meta(id), events: log })
+    let coordinator!: PersistenceCoordinator<never>
+    const fiber = await ctx.plugin(Object.assign((inner: Context) => {
+      coordinator = new PersistenceCoordinator(inner, backend)
+    }, { inject: ['sessions'] }))
+
+    try {
+      // Happy path through the hook: only the suffix comes back, detached.
+      backend.seekHook = async (hookId, fromSeq) => {
+        const entry = backend.store.get(hookId)
+        if (entry === undefined) return undefined
+        return { meta: structuredClone(entry.meta), events: entry.events.filter(e => e.seq >= fromSeq) }
+      }
+      const suffix = await coordinator.readFrom(id, 3)
+      expect(suffix.events).toEqual(log.slice(3))
+      // The hook's undefined is the seam's not-found.
+      await expect(coordinator.readFrom(SessionId('missing-seek'), 0)).rejects.toThrow('not found')
+
+      // A hook failure with no cancellation in play propagates as-is.
+      const hookFailure = new Error('seek backend exploded')
+      backend.seekHook = () => Promise.reject(hookFailure)
+      await expect(coordinator.readFrom(id, 0)).rejects.toBe(hookFailure)
+
+      // A hook failure after cancellation surfaces the caller's abort reason,
+      // not the backend's internal teardown error. The abort fires only once
+      // the hook is provably entered, so the failure exercises the catch (not
+      // the pre-invocation throwIfAborted).
+      const controller = new AbortController()
+      const reason = new Error('read-from cancelled mid-hook')
+      let hookEntered = false
+      backend.seekHook = async (_hookId, _fromSeq, signal) => {
+        hookEntered = true
+        await new Promise<void>((resolve) => { signal?.addEventListener('abort', () => { resolve() }, { once: true }) })
+        throw new Error('backend teardown after abort')
+      }
+      const pending = coordinator.readFrom(id, 0, controller.signal)
+      const observed = pending.catch((error: unknown) => error)
+      await vi.waitFor(() => { expect(hookEntered).toBe(true) })
+      controller.abort(reason)
+      expect(await observed).toBe(reason)
+    } finally {
+      await fiber.dispose()
+      await ctx.fiber.dispose()
+    }
+  })
+
   it('rejects a cancelled inspect while an in-flight retirement drain is still pending', async () => {
     const ctx = new Context()
     await ctx.plugin(SessionStore)
@@ -537,6 +600,66 @@ describe('PersistenceCoordinator retirement', () => {
     }
   })
 
+  it('a superseded retirement leaves the successor lifecycle\'s pending drain in place', async () => {
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    const backend = new ControlledBackend()
+    let coordinator!: PersistenceCoordinator<never>
+    const backendFiber = await ctx.plugin(Object.assign((inner: Context) => {
+      coordinator = new PersistenceCoordinator(inner, backend)
+    }, { inject: ['sessions'] }))
+    const internals = coordinator as unknown as CoordinatorInternals
+    const readGate = Promise.withResolvers<boolean>()
+
+    try {
+      const id = SessionId('superseded-retirement')
+      // First lifecycle: unmaterialized (zero events), so a same-id successor
+      // may legally reclaim the abandoned id later.
+      let first!: Session
+      const firstFiber = await ctx.plugin(Object.assign((inner: Context) => {
+        first = inner.sessions.create(id)
+      }, { inject: ['sessions'] }))
+      await ctx.sessions.flush(first)
+
+      // Occupy the per-id serialize chain with a gated read: everything the
+      // two retirements queue stays pending behind it. (Attempt counting
+      // starts here — an absent beforeLoadStored short-circuits the optional
+      // call without evaluating its ++ argument.)
+      backend.beforeLoadStored = async (attempt) => {
+        if (attempt === 1) await readGate.promise
+      }
+      const parked = coordinator.inspect(id).catch((error: unknown) => error)
+      await vi.waitFor(() => { expect(backend.loadAttempts).toBe(1) })
+
+      // First retirement queues behind the gate and stays pending.
+      await firstFiber.dispose()
+      await vi.waitFor(() => { expect(internals.retirements.has(id)).toBe(true) })
+      const firstRetirement = internals.retirements.get(id)
+
+      // Successor lifecycle retires while the first drain is still in flight:
+      // retire() replaces the map entry synchronously.
+      const secondFiber = await ctx.plugin(Object.assign((inner: Context) => {
+        inner.sessions.create(id)
+      }, { inject: ['sessions'] }))
+      await secondFiber.dispose()
+      await vi.waitFor(() => {
+        expect(internals.retirements.get(id)).not.toBe(firstRetirement)
+      })
+
+      // Release the chain: the first drain settles and its forget() must not
+      // delete the successor's entry (exact-entry guard); the successor's own
+      // forget() then clears the map.
+      readGate.resolve(true)
+      expect(await parked).toBeInstanceOf(Error) // the parked inspect (not found) is observed
+      await firstRetirement
+      await vi.waitFor(() => { expect(internals.retirements.has(id)).toBe(false) })
+    } finally {
+      readGate.resolve(true)
+      await backendFiber.dispose()
+      await ctx.fiber.dispose()
+    }
+  })
+
   it('a replacement queued before retirement cleanup still collides with the live owner', async () => {
     const ctx = new Context()
     await ctx.plugin(SessionStore)