fix(session): address restore review feedback

2026-08-06 05:13:26 +08:00
parent f0868d6a70
commit 2f98f29992
17 changed files with 233 additions and 38 deletions
@@ -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 .agents/notes/implemented/architecture/2026-08-05-session-preparation.md
-2026-08-05-session-preparation.md: a3dfb50c8484cfef0cadea1759125a86714d3cbc
-2026-08-05-session-preparation.zh.md: 031e96f0e4446f02946f8c29ddade3fb52fe396e
+2026-08-05-session-preparation.md: d6b6cc190779279d1cc008bca890f3818736e920
+2026-08-05-session-preparation.zh.md: 191ad84db0e7bba38149c90cd91b410187f08f20
@@ -43,7 +43,7 @@ Cold continuable-subagent access follows the same path. Descriptor authorization
 - The cache belongs to one persistence coordinator, not a process-global Session map. Live Sessions are owned by the existing stores and never occupy preparation capacity.
 - A fresh create never claims a cold persisted preparation with the same id. Persistence collisions continue to reject.
 - Third-party persistence implementations retain the abstract `prepare()` fallback through `load()`. They receive the same publication interface but gain exact-object reuse only when they override preparation.
- Revision validation establishes freshness at the reuse and repair-commit points; it does not add cross-process writer exclusion to a backend.
+- Revision validation establishes freshness at the reuse and repair-commit points; it does not add cross-process writer exclusion to a backend. Retries converge after the durable log remains unchanged for one read/check round trip, so continuous external writers can delay preparation.

 ## Verification

@@ -43,7 +43,7 @@ agent loop（智能体循环）通过同一条设置与发布流水线消费这
 - 缓存属于单个持久化协调器，而不是进程全局 Session map。实时 Session 由现有存储持有，绝不占用准备容量。
 - 新建流程绝不认领相同 id 的冷持久化准备对象。持久化冲突仍会被拒绝。
 - 第三方持久化实现继续获得通过 `load()` 实现的抽象 `prepare()` 回退。它们使用相同发布接口，但只有覆盖准备流程后才能复用精确对象。
- Revision 校验在复用点和修复提交点建立新鲜性，但不会为后端增加跨进程 writer 排他。
+- Revision 校验在复用点和修复提交点建立新鲜性，但不会为后端增加跨进程 writer 排他。持久日志在一次读取与复核往返内保持不变后，重试才能收敛，因此持续的外部写入可能延迟准备。

 ## 验证

@@ -1178,7 +1178,8 @@ abstract append(id: SessionId, events: readonly SessionEvent[]): Promise<void>
 * Prepare the exact unpublished Session used by resume. Implementations may
 * reuse object graphs retained by an earlier {@link inspect} after confirming
 * their durable revision is still current; disposal releases an unpublished
- * reservation.
+ * reservation. Revision retries require the durable log to remain unchanged
+ * for one read/check round trip; continuous external writers may delay completion.
 * @param id - persisted session to prepare.
 * @param signal - optional cancellation for preparation work.
 * @returns one owned unpublished Session preparation.
@@ -1194,6 +1195,7 @@ async prepare(id: SessionId, signal?: AbortSignal): Promise<SessionPreparation>
 * identity still bound to a live Session: a balanced live log may return as a
 * durable snapshot, while an open live turn rejects. Returned values may be
 * shared with immutable live or prepared state and must not be mutated.
+ * Revision-based implementations may wait for one stable read/check round trip.
 * @param id - the persisted session to reload.
 * @returns the header and a log ending on a balanced `turn/end`.
 */
@@ -1207,7 +1209,8 @@ abstract load(id: SessionId): Promise<SessionInspection>
 * open turn and its `session/end-seed` boundary. Coordinator-backed
 * implementations retain the exact cold unpublished Session for bounded
 * reuse by a later {@link prepare}, reloading it when its durable revision
- * changes; callers borrow only its immutable header and log.
+ * changes; callers borrow only its immutable header and log. Continuous
+ * external writers may delay revision convergence.
 * @param id - the persisted session to inspect.
 * @param signal - optional cancellation for queued and backend read work.
 * @returns the validated header and current logical event log.
@@ -1593,11 +1596,7 @@ Persistence is intentionally not implemented here — persistence plugins subscr
 * `dsh-agent-loop`'s creation transaction).
 *
 * @param id - the session id; omitted, the store mints `session-<n>`.
- * @param options - seed events and/or creation metadata for the header. With
- *   `seedSource: 'persistence'`, metadata and events must be fresh detached
- *   graphs whose ownership transfers to this call: they are validated and
- *   frozen in place through {@link Session.fromRestore}, so the caller must
- *   retain no mutable aliases.
+ * @param options - seed events and/or creation metadata for the header.
 * @returns the live session, already entered and announced.
 * @throws if a session with `id` already exists, metadata is not a plain
 *   lossless-JSON record with valid scalar fields, or `meta.cwd` is a
@@ -1615,7 +1614,11 @@ create(id?: SessionId, options?: CreateSessionOptions): Session
 * before the driver's closing events commit, dropping them.
 *
 * @param id - the session id; omitted, the store mints `session-<n>`.
- * @param options - seed events and/or creation metadata for the header.
+ * @param options - seed events and/or creation metadata for the header. With
+ *   `seedSource: 'persistence'`, metadata and events must be fresh detached
+ *   graphs whose ownership transfers to this call: they are validated and
+ *   frozen in place through {@link Session.fromRestore}, so the caller must
+ *   retain no mutable aliases.
 * @returns the constructed session, NOT yet in the store.
 * @throws if a session with `id` already exists, metadata is not a plain
 *   lossless-JSON record with valid scalar fields, or `meta.cwd` is a
@@ -582,15 +582,15 @@ export const SERVICE_API: readonly ServiceApiEntry[] = [
      },
      {
        signature: 'async prepare(id: SessionId, signal?: AbortSignal): Promise<SessionPreparation>',
-        jsDoc: '/**\n * Prepare the exact unpublished Session used by resume. Implementations may\n * reuse object graphs retained by an earlier {@link inspect} after confirming\n * their durable revision is still current; disposal releases an unpublished\n * reservation.\n * @param id - persisted session to prepare.\n * @param signal - optional cancellation for preparation work.\n * @returns one owned unpublished Session preparation.\n */',
+        jsDoc: '/**\n * Prepare the exact unpublished Session used by resume. Implementations may\n * reuse object graphs retained by an earlier {@link inspect} after confirming\n * their durable revision is still current; disposal releases an unpublished\n * reservation. Revision retries require the durable log to remain unchanged\n * for one read/check round trip; continuous external writers may delay completion.\n * @param id - persisted session to prepare.\n * @param signal - optional cancellation for preparation work.\n * @returns one owned unpublished Session preparation.\n */',
      },
      {
        signature: 'abstract load(id: SessionId): Promise<SessionInspection>',
-        jsDoc: '/**\n * Load an immutable balanced logical view and commit any required cold\n * recovery. A complete interrupted final turn is preserved and durably\n * closed with missing tool errors plus any open step and turn boundaries;\n * only a torn final record is discarded. Unknown versions and corruption in\n * the committed prefix reject. Implementations MUST NOT crash-repair an\n * identity still bound to a live Session: a balanced live log may return as a\n * durable snapshot, while an open live turn rejects. Returned values may be\n * shared with immutable live or prepared state and must not be mutated.\n * @param id - the persisted session to reload.\n * @returns the header and a log ending on a balanced `turn/end`.\n */',
+        jsDoc: '/**\n * Load an immutable balanced logical view and commit any required cold\n * recovery. A complete interrupted final turn is preserved and durably\n * closed with missing tool errors plus any open step and turn boundaries;\n * only a torn final record is discarded. Unknown versions and corruption in\n * the committed prefix reject. Implementations MUST NOT crash-repair an\n * identity still bound to a live Session: a balanced live log may return as a\n * durable snapshot, while an open live turn rejects. Returned values may be\n * shared with immutable live or prepared state and must not be mutated.\n * Revision-based implementations may wait for one stable read/check round trip.\n * @param id - the persisted session to reload.\n * @returns the header and a log ending on a balanced `turn/end`.\n */',
      },
      {
        signature: 'abstract inspect(id: SessionId, signal?: AbortSignal): Promise<SessionInspection>',
-        jsDoc: '/**\n * Inspect an immutable logical session without committing recovery or\n * publishing it. A cold complete interrupted turn receives synthetic closers\n * in memory and a torn physical tail remains untouched. An already-live\n * Session instead yields its current immutable snapshot, which may contain an\n * open turn and its `session/end-seed` boundary. Coordinator-backed\n * implementations retain the exact cold unpublished Session for bounded\n * reuse by a later {@link prepare}, reloading it when its durable revision\n * changes; callers borrow only its immutable header and log.\n * @param id - the persisted session to inspect.\n * @param signal - optional cancellation for queued and backend read work.\n * @returns the validated header and current logical event log.\n */',
+        jsDoc: '/**\n * Inspect an immutable logical session without committing recovery or\n * publishing it. A cold complete interrupted turn receives synthetic closers\n * in memory and a torn physical tail remains untouched. An already-live\n * Session instead yields its current immutable snapshot, which may contain an\n * open turn and its `session/end-seed` boundary. Coordinator-backed\n * implementations retain the exact cold unpublished Session for bounded\n * reuse by a later {@link prepare}, reloading it when its durable revision\n * changes; callers borrow only its immutable header and log. Continuous\n * external writers may delay revision convergence.\n * @param id - the persisted session to inspect.\n * @param signal - optional cancellation for queued and backend read work.\n * @returns the validated header and current logical event log.\n */',
      },
      {
        signature: 'abstract readFrom(id: SessionId, fromSeq: number, signal?: AbortSignal): Promise<{ meta: SessionHeader; events: SessionEvent[] }>',
@@ -740,11 +740,11 @@ export const SERVICE_API: readonly ServiceApiEntry[] = [
    methods: [
      {
        signature: 'create(id?: SessionId, options?: CreateSessionOptions): Session',
-        jsDoc: '/**\n * Create a session owned by the calling fiber: disposing that fiber stops\n * event notification and removes the session from the store. `options.seed`\n * populates the session with a copy of those events (replay/fork);\n * `options.meta` attaches creation metadata (validated absolute `cwd`, seed\n * and parent lineage, and delegation depth) as the immutable\n * {@link SessionHeader} (the store fills `version`/`id`/`createdAt`).\n *\n * For an agent whose session must be torn down IN ORDER with its loop (so the\n * loop\'s final events are published before the store attachment ends), do NOT use this\n * — fold the session lifecycle into the agent\'s own effect via\n * {@link prepare} + {@link enter} + {@link announce} (see\n * `dsh-agent-loop`\'s creation transaction).\n *\n * @param id - the session id; omitted, the store mints `session-<n>`.\n * @param options - seed events and/or creation metadata for the header. With\n *   `seedSource: \'persistence\'`, metadata and events must be fresh detached\n *   graphs whose ownership transfers to this call: they are validated and\n *   frozen in place through {@link Session.fromRestore}, so the caller must\n *   retain no mutable aliases.\n * @returns the live session, already entered and announced.\n * @throws if a session with `id` already exists, metadata is not a plain\n *   lossless-JSON record with valid scalar fields, or `meta.cwd` is a\n *   non-absolute path (storage backends key directories off it).\n */',
+        jsDoc: '/**\n * Create a session owned by the calling fiber: disposing that fiber stops\n * event notification and removes the session from the store. `options.seed`\n * populates the session with a copy of those events (replay/fork);\n * `options.meta` attaches creation metadata (validated absolute `cwd`, seed\n * and parent lineage, and delegation depth) as the immutable\n * {@link SessionHeader} (the store fills `version`/`id`/`createdAt`).\n *\n * For an agent whose session must be torn down IN ORDER with its loop (so the\n * loop\'s final events are published before the store attachment ends), do NOT use this\n * — fold the session lifecycle into the agent\'s own effect via\n * {@link prepare} + {@link enter} + {@link announce} (see\n * `dsh-agent-loop`\'s creation transaction).\n *\n * @param id - the session id; omitted, the store mints `session-<n>`.\n * @param options - seed events and/or creation metadata for the header.\n * @returns the live session, already entered and announced.\n * @throws if a session with `id` already exists, metadata is not a plain\n *   lossless-JSON record with valid scalar fields, or `meta.cwd` is a\n *   non-absolute path (storage backends key directories off it).\n */',
      },
      {
        signature: 'prepare(id?: SessionId, options?: PrepareSessionOptions): Session',
-        jsDoc: '/**\n * Build a session WITHOUT entering it into the store — validate the id/cwd and\n * construct the {@link Session} (with its immutable {@link SessionHeader}).\n * Pairs with {@link enter} + {@link announce}: a caller that owns a composite\n * `ctx.effect` (the agent factory) folds the session lifecycle into that ONE\n * effect so a fiber unload tears the session + agent down as a single ORDERED\n * chain rather than as racing sibling effects — which would remove the publication hooks\n * before the driver\'s closing events commit, dropping them.\n *\n * @param id - the session id; omitted, the store mints `session-<n>`.\n * @param options - seed events and/or creation metadata for the header.\n * @returns the constructed session, NOT yet in the store.\n * @throws if a session with `id` already exists, metadata is not a plain\n *   lossless-JSON record with valid scalar fields, or `meta.cwd` is a\n *   non-absolute path.\n */',
+        jsDoc: '/**\n * Build a session WITHOUT entering it into the store — validate the id/cwd and\n * construct the {@link Session} (with its immutable {@link SessionHeader}).\n * Pairs with {@link enter} + {@link announce}: a caller that owns a composite\n * `ctx.effect` (the agent factory) folds the session lifecycle into that ONE\n * effect so a fiber unload tears the session + agent down as a single ORDERED\n * chain rather than as racing sibling effects — which would remove the publication hooks\n * before the driver\'s closing events commit, dropping them.\n *\n * @param id - the session id; omitted, the store mints `session-<n>`.\n * @param options - seed events and/or creation metadata for the header. With\n *   `seedSource: \'persistence\'`, metadata and events must be fresh detached\n *   graphs whose ownership transfers to this call: they are validated and\n *   frozen in place through {@link Session.fromRestore}, so the caller must\n *   retain no mutable aliases.\n * @returns the constructed session, NOT yet in the store.\n * @throws if a session with `id` already exists, metadata is not a plain\n *   lossless-JSON record with valid scalar fields, or `meta.cwd` is a\n *   non-absolute path.\n */',
      },
      {
        signature: 'enter(session: Session): () => void',
@@ -821,11 +821,7 @@ export class SessionStore extends Service {
   * `dsh-agent-loop`'s creation transaction).
   *
   * @param id - the session id; omitted, the store mints `session-<n>`.
-   * @param options - seed events and/or creation metadata for the header. With
-   *   `seedSource: 'persistence'`, metadata and events must be fresh detached
-   *   graphs whose ownership transfers to this call: they are validated and
-   *   frozen in place through {@link Session.fromRestore}, so the caller must
-   *   retain no mutable aliases.
+   * @param options - seed events and/or creation metadata for the header.
   * @returns the live session, already entered and announced.
   * @throws if a session with `id` already exists, metadata is not a plain
   *   lossless-JSON record with valid scalar fields, or `meta.cwd` is a
@@ -854,7 +850,11 @@ export class SessionStore extends Service {
   * before the driver's closing events commit, dropping them.
   *
   * @param id - the session id; omitted, the store mints `session-<n>`.
-   * @param options - seed events and/or creation metadata for the header.
+   * @param options - seed events and/or creation metadata for the header. With
+   *   `seedSource: 'persistence'`, metadata and events must be fresh detached
+   *   graphs whose ownership transfers to this call: they are validated and
+   *   frozen in place through {@link Session.fromRestore}, so the caller must
+   *   retain no mutable aliases.
   * @returns the constructed session, NOT yet in the store.
   * @throws if a session with `id` already exists, metadata is not a plain
   *   lossless-JSON record with valid scalar fields, or `meta.cwd` is a
@@ -186,7 +186,10 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
    return this.readPrefix(path, id, signal)
  }

-  /** Read one log's stat-derived revision without loading its event bytes. */
+  /**
+   * Read one log's stat-derived revision without loading its event bytes.
+   * Resolving an id with unknown cwd still scans the project directories.
+   */
  async readStoredRevision(id: SessionId, signal?: AbortSignal): Promise<PersistenceRevision | undefined> {
    signal?.throwIfAborted()
    await this.ensureRootEncoding()
@@ -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: 327a61d1b8bf10239eab0ca3cb18ac865fc2dcce
-README.zh.md: 2cc0b94bfd46a9e75c7a3626554f2ab9e5949600
+README.md: b29ff5ba17f384e8d3b1700ed3ad6c80aeeb184c
+README.zh.md: 8ca4b6a7128383fdca706f35914a06eb1f26de02
@@ -54,7 +54,7 @@ The `PersistenceBackend<TornMarker>` hooks (the only seam between the coordinato
 | `list(signal?)` | List all stored metadata, observing optional cancellation. |
 | `close?()` | Optional lifecycle teardown (e.g. close a db handle), awaited after the dispose drain. |

-The coordinator asserts the stored id and compares stored/live cwd before repair or live adoption. Its `inspect()` path takes ownership of fresh backend values, validates and freezes them once, and retains at most the configured number of unpublished Sessions without calling `commitRepair`. A retained source is reused or repaired only when its revision still equals `readStoredRevision`; otherwise the coordinator reloads it. This freshness check does not add cross-process writer exclusion. The `tornMarker` is fully OPAQUE: the coordinator only tests `!== undefined` and round-trips it to `commitRepair`, never inspecting its value (the JSONL backend uses the byte offset to truncate to, the SQLite backend the seq to delete from). A third-party backend MAY implement the abstract service directly without the coordinator, but it must provide the same non-mutating inspection and trustworthy lightweight snapshot revisions. See [the write-coordinator Agent Note](../../../.agents/notes/implemented/architecture/2026-06-18-shared-persistence-write-coordinator.md).
+The coordinator asserts the stored id and compares stored/live cwd before repair or live adoption. Its `inspect()` path takes ownership of fresh backend values, validates and freezes them once, and retains at most the configured number of unpublished Sessions without calling `commitRepair`. A retained source is reused or repaired only when its revision still equals `readStoredRevision`; otherwise the coordinator reloads it. This freshness check does not add cross-process writer exclusion. Revision retries converge when the durable log remains unchanged for one read/check round trip; continuous external writers can delay `load`, `inspect`, or `prepare`. The `tornMarker` is fully OPAQUE: the coordinator only tests `!== undefined` and round-trips it to `commitRepair`, never inspecting its value (the JSONL backend uses the byte offset to truncate to, the SQLite backend the seq to delete from). A third-party backend MAY implement the abstract service directly without the coordinator, but it must provide the same non-mutating inspection and trustworthy lightweight snapshot revisions. See [the write-coordinator Agent Note](../../../.agents/notes/implemented/architecture/2026-06-18-shared-persistence-write-coordinator.md).

 ## Metadata and location types

@@ -54,7 +54,7 @@
 | `list(signal?)` | 列出全部已存储元数据，并遵循可选的取消信号。 |
 | `close?()` | 可选生命周期拆卸（例如关闭 db 句柄），在 dispose drain 后等待。 |

-协调器断言已存储 id，并在修复或实时接管前比较已存储/实时 cwd。其 `inspect()` 路径取得新鲜后端值的所有权，只验证和冻结一次，并在不调用 `commitRepair` 的情况下最多保留配置数量的未发布 Session。只有保留源的 revision 仍等于 `readStoredRevision` 时，系统才会复用或修复它；否则协调器会重新读取。该新鲜性校验不会增加跨进程写入排他。`tornMarker` 完全不透明：协调器只测试 `!== undefined`，并将其原样往返给 `commitRepair`，绝不检查值（JSONL 后端使用待截断字节偏移，SQLite 后端使用待删除 seq）。第三方后端可以不用协调器直接实现抽象服务，但必须提供相同非变更检查和可信轻量快照修订。详见[写入协调器 Agent Note](../../../.agents/notes/implemented/architecture/2026-06-18-shared-persistence-write-coordinator.md)。
+协调器断言已存储 id，并在修复或实时接管前比较已存储/实时 cwd。其 `inspect()` 路径取得新鲜后端值的所有权，只验证和冻结一次，并在不调用 `commitRepair` 的情况下最多保留配置数量的未发布 Session。只有保留源的 revision 仍等于 `readStoredRevision` 时，系统才会复用或修复它；否则协调器会重新读取。该新鲜性校验不会增加跨进程写入排他。持久日志在一次读取与复核往返内保持不变时，revision 重试才能收敛；持续的外部写入可能延迟 `load`、`inspect` 或 `prepare`。`tornMarker` 完全不透明：协调器只测试 `!== undefined`，并将其原样往返给 `commitRepair`，绝不检查值（JSONL 后端使用待截断字节偏移，SQLite 后端使用待删除 seq）。第三方后端可以不用协调器直接实现抽象服务，但必须提供相同非变更检查和可信轻量快照修订。详见[写入协调器 Agent Note](../../../.agents/notes/implemented/architecture/2026-06-18-shared-persistence-write-coordinator.md)。

 ## 元数据与位置类型

@@ -639,6 +639,8 @@ export class PersistenceCoordinator<TornMarker = unknown> {

  /**
   * Prepare and reserve the exact unpublished Session used by resume.
+   * Revision retries converge once the durable log remains unchanged for one
+   * read/check round trip; continuous external writers may delay completion.
   * @param id - persisted session to prepare.
   * @param signal - optional cancellation for reading and repair.
   * @returns an owned preparation released after publication or rollback.
@@ -674,6 +676,8 @@ export class PersistenceCoordinator<TornMarker = unknown> {

  /**
   * Commit recovery and return its immutable logical view without publication.
+   * Revision retries converge once the durable log remains unchanged for one
+   * read/check round trip; continuous external writers may delay completion.
   * @param id - persisted session to load.
   * @returns prepared header and balanced events.
   */
@@ -700,6 +704,9 @@ export class PersistenceCoordinator<TornMarker = unknown> {

  /**
   * Inspect a logical session without publishing it or committing recovery.
+   * Retained cold state is reloaded after its durable revision changes. Revision
+   * retries converge once the log is stable for one read/check round trip;
+   * continuous external writers may delay completion.
   * @param id - persisted session to inspect.
   * @param signal - optional cancellation for preparation work.
   * @returns immutable prepared metadata and events; a live view may have an open turn.
@@ -726,7 +733,9 @@ export class PersistenceCoordinator<TornMarker = unknown> {
        const published = this.ctx.sessions.get(id)
        if (published !== undefined) return this.inspectLive(published)
        if (current) return source.inspection
-        this.preparations.invalidate(id, source)
+        if (this.preparations.discardReady(id, source) === 'retained') {
+          return source.inspection
+        }
      } catch (error: unknown) {
        const attached = this.ctx.sessions.get(id)
        if (attached !== undefined) return this.inspectLive(attached)
@@ -104,7 +104,8 @@ export abstract class SessionPersistence extends Service {
   * Prepare the exact unpublished Session used by resume. Implementations may
   * reuse object graphs retained by an earlier {@link inspect} after confirming
   * their durable revision is still current; disposal releases an unpublished
-   * reservation.
+   * reservation. Revision retries require the durable log to remain unchanged
+   * for one read/check round trip; continuous external writers may delay completion.
   * @param id - persisted session to prepare.
   * @param signal - optional cancellation for preparation work.
   * @returns one owned unpublished Session preparation.
@@ -133,6 +134,7 @@ export abstract class SessionPersistence extends Service {
   * identity still bound to a live Session: a balanced live log may return as a
   * durable snapshot, while an open live turn rejects. Returned values may be
   * shared with immutable live or prepared state and must not be mutated.
+   * Revision-based implementations may wait for one stable read/check round trip.
   * @param id - the persisted session to reload.
   * @returns the header and a log ending on a balanced `turn/end`.
   */
@@ -146,7 +148,8 @@ export abstract class SessionPersistence extends Service {
   * open turn and its `session/end-seed` boundary. Coordinator-backed
   * implementations retain the exact cold unpublished Session for bounded
   * reuse by a later {@link prepare}, reloading it when its durable revision
-   * changes; callers borrow only its immutable header and log.
+   * changes; callers borrow only its immutable header and log. Continuous
+   * external writers may delay revision convergence.
   * @param id - the persisted session to inspect.
   * @param signal - optional cancellation for queued and backend read work.
   * @returns the validated header and current logical event log.
@@ -111,6 +111,7 @@ export class SessionPreparations<Source extends PreparedSource, CommitState> {
      this.makeReady(entry)
      throw error
    }
+    if (this.entries.get(id) !== entry) return undefined
    const reservation: SessionPreparationReservation<Source, CommitState> = {
      entry,
      source: committed.source,
@@ -190,6 +191,20 @@ export class SessionPreparations<Source extends PreparedSource, CommitState> {
    if (entry !== undefined && (expected === undefined || entry.source === expected)) this.remove(entry)
  }

+  /**
+   * Discard an exact stale ready source without disturbing an exclusive owner.
+   * @param id - changed session identity.
+   * @param expected - exact source observed before its revision check.
+   * @returns whether the source was discarded, retained by a reservation, or is absent.
+   */
+  discardReady(id: SessionId, expected: Source): 'discarded' | 'retained' | 'missing' {
+    const entry = this.entries.get(id)
+    if (entry === undefined || entry.source !== expected) return 'missing'
+    if (entry.phase !== 'ready') return 'retained'
+    this.remove(entry)
+    return 'discarded'
+  }
+
  /**
   * Reject writes while an unpublished Session exclusively reserves the id.
   * @param id - session identity to check.
@@ -208,10 +223,7 @@ export class SessionPreparations<Source extends PreparedSource, CommitState> {
   */
  takeReady(id: SessionId): Source | undefined {
    const entry = this.entries.get(id)
-    if (entry === undefined) return undefined
-    if (entry.phase !== 'ready' || entry.source === undefined) {
-      throw new Error(`cannot adopt session "${id}" while its preparation is pending`)
-    }
+    if (entry === undefined || entry.phase !== 'ready' || entry.source === undefined) return undefined
    this.remove(entry)
    return entry.source
  }
@@ -242,7 +254,7 @@ export class SessionPreparations<Source extends PreparedSource, CommitState> {
    void loading.then((source) => {
      if (this.entries.get(id) === entry) {
        entry.source = source
-        entry.phase = 'ready'
+        this.makeReady(entry)
      }
      deferred.resolve(source)
    }, (error: unknown) => {
@@ -682,6 +682,44 @@ describe('PersistenceCoordinator session preparations', () => {
    }
  })

+  it('retains a reserved preparation when inspection observes a newer external revision', async () => {
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    const backend = new ControlledBackend()
+    const id = SessionId('reserved-inspect-revision-race')
+    backend.store.set(id, { meta: meta(id), events: oneTurnLog() })
+    let coordinator!: PersistenceCoordinator<never>
+    const fiber = await ctx.plugin(Object.assign((inner: Context) => {
+      coordinator = new PersistenceCoordinator(inner, backend)
+    }, { inject: ['sessions'] }))
+    let preparation: Awaited<ReturnType<typeof coordinator.prepare>> | undefined
+    let detach: (() => void) | undefined
+
+    try {
+      const cached = await coordinator.inspect(id)
+      preparation = await coordinator.prepare(id)
+      backend.store.get(id)!.events.push(
+        { type: 'turn/start', seq: 6, time: 7, data: { turn: 2 } },
+        { type: 'turn/end', seq: 7, time: 8, data: { turn: 2, reason: { kind: 'completed' } } },
+      )
+
+      await expect(coordinator.inspect(id)).resolves.toBe(cached)
+      const preparations = (coordinator as unknown as {
+        preparations: { reservationFor: (session: Session) => object | undefined }
+      }).preparations
+      expect(preparations.reservationFor(preparation.session)).toBeDefined()
+
+      detach = ctx.sessions.enter(preparation.session)
+      expect(() => { ctx.sessions.announce(preparation!.session) }).not.toThrow()
+      expect(preparations.reservationFor(preparation.session)).toBeUndefined()
+    } finally {
+      detach?.()
+      preparation?.[Symbol.dispose]()
+      await fiber.dispose()
+      await ctx.fiber.dispose()
+    }
+  })
+
  it('queues a same-tick cold append behind preparation readiness', async () => {
    const ctx = new Context()
    await ctx.plugin(SessionStore)
@@ -715,6 +753,39 @@ describe('PersistenceCoordinator session preparations', () => {
    }
  })

+  it('allows a same-tick cold append to start before inspection', async () => {
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    const backend = new ControlledBackend()
+    const id = SessionId('cold-append-inspect-race')
+    backend.store.set(id, { meta: meta(id), events: oneTurnLog() })
+    let coordinator!: PersistenceCoordinator<never>
+    const fiber = await ctx.plugin(Object.assign((inner: Context) => {
+      coordinator = new PersistenceCoordinator(inner, backend)
+    }, { inject: ['sessions'] }))
+
+    try {
+      const append = coordinator.append(id, [{
+        type: 'turn/start',
+        seq: oneTurnLog().length,
+        time: 7,
+        data: { turn: 2 },
+      }])
+      const inspection = coordinator.inspect(id)
+
+      await expect(append).resolves.toBeUndefined()
+      await expect(inspection).resolves.toMatchObject({
+        meta: { id },
+        events: [...oneTurnLog(), { seq: 6 }, { seq: 7 }],
+      })
+      expect(backend.loadAttempts).toBe(2)
+      expect(backend.store.get(id)?.events).toHaveLength(oneTurnLog().length + 1)
+    } finally {
+      await fiber.dispose()
+      await ctx.fiber.dispose()
+    }
+  })
+
  it('retries cold append adoption when the prepared revision becomes stale', async () => {
    const ctx = new Context()
    await ctx.plugin(SessionStore)
@@ -60,6 +60,32 @@ describe('SessionPreparations inspection', () => {
    expect(load).toHaveBeenCalledOnce()
  })

+  it('evicts completed loads whose observers cancelled before readiness', async () => {
+    const preparations = new SessionPreparations<PreparedSource, string>(1)
+    const firstId = SessionId('cancelled-ready-first')
+    const secondId = SessionId('cancelled-ready-second')
+    const firstGate = Promise.withResolvers<PreparedSource>()
+    const secondGate = Promise.withResolvers<PreparedSource>()
+    const firstController = new AbortController()
+    const secondController = new AbortController()
+    const first = preparations.inspect(firstId, () => firstGate.promise, firstController.signal)
+    const second = preparations.inspect(secondId, () => secondGate.promise, secondController.signal)
+
+    firstController.abort(new Error('first observer cancelled'))
+    secondController.abort(new Error('second observer cancelled'))
+    await expect(first).rejects.toThrow('first observer cancelled')
+    await expect(second).rejects.toThrow('second observer cancelled')
+
+    firstGate.resolve(prepared(firstId))
+    await firstGate.promise
+    secondGate.resolve(prepared(secondId))
+    await secondGate.promise
+    await Promise.resolve()
+
+    expect(preparations.has(firstId)).toBe(false)
+    expect(preparations.has(secondId)).toBe(true)
+  })
+
  it('removes failed and invalidated in-flight loads without changing their observers', async () => {
    const preparations = new SessionPreparations<PreparedSource, string>(1)
    const failedId = SessionId('failed-inspection')
@@ -117,6 +143,23 @@ describe('SessionPreparations inspection', () => {
    preparations.discard(reservedB!)
    preparations.invalidate(SessionId('reserved-a'))
  })
+
+  it('discards only the exact ready source and retains exclusive reservations', async () => {
+    const preparations = new SessionPreparations<PreparedSource, string>(1)
+    const ready = prepared('discard-ready')
+    expect(preparations.discardReady(ready.session.id, ready)).toBe('missing')
+    await preparations.inspect(ready.session.id, () => Promise.resolve(ready))
+    expect(preparations.discardReady(ready.session.id, prepared('different'))).toBe('missing')
+    expect(preparations.discardReady(ready.session.id, ready)).toBe('discarded')
+
+    const reserved = await preparations.reserve(
+      ready.session.id,
+      () => Promise.resolve(ready),
+      committed,
+    )
+    expect(preparations.discardReady(ready.session.id, ready)).toBe('retained')
+    preparations.release(reserved!, false)
+  })
 })

 describe('SessionPreparations reservation', () => {
@@ -229,6 +272,26 @@ describe('SessionPreparations reservation', () => {
    expect(preparations.has(id)).toBe(false)
  })

+  it('does not reserve an entry invalidated while its commit succeeds', async () => {
+    const preparations = new SessionPreparations<PreparedSource, string>(1)
+    const id = SessionId('invalidated-successful-commit')
+    const source = prepared(id)
+    const commitStarted = Promise.withResolvers<undefined>()
+    const commitGate = Promise.withResolvers<undefined>()
+    const reservation = preparations.reserve(id, () => Promise.resolve(source), async (value) => {
+      commitStarted.resolve(undefined)
+      await commitGate.promise
+      return { source: value, state: value.label }
+    })
+
+    await commitStarted.promise
+    preparations.invalidate(id)
+    commitGate.resolve(undefined)
+
+    await expect(reservation).resolves.toBeUndefined()
+    expect(preparations.has(id)).toBe(false)
+  })
+
  it('returns undefined when a load is invalidated before reservation', async () => {
    const preparations = new SessionPreparations<PreparedSource, string>(1)
    const id = SessionId('invalidated-reservation')
@@ -239,12 +302,12 @@ describe('SessionPreparations reservation', () => {
    await expect(reservation).resolves.toBeUndefined()
  })

-  it('rejects pending adoption and accepts a ready source exactly once', async () => {
+  it('skips pending adoption and accepts a ready source exactly once', async () => {
    const preparations = new SessionPreparations<PreparedSource, string>(1)
    const id = SessionId('take-ready')
    const gate = Promise.withResolvers<PreparedSource>()
    const inspection = preparations.inspect(id, () => gate.promise)
-    expect(() => preparations.takeReady(id)).toThrow(/preparation is pending/)
+    expect(preparations.takeReady(id)).toBeUndefined()
    const source = prepared(id)
    gate.resolve(source)
    await inspection
@@ -705,6 +705,7 @@ export class SubagentContinuationManager {
    try {
      loaded = await persistence.inspect(childId, options.signal)
    } catch (error: unknown) {
+      options.signal.throwIfAborted()
      throw new SubagentError(`subagent "${childId}" is unavailable`, 'NOT_RESUMABLE', { cause: error })
    }
    options.signal.throwIfAborted()
@@ -568,6 +568,36 @@ describe('SubagentService.followup residency routing', () => {
      .rejects.toMatchObject({ code: 'NOT_RESUMABLE' })
  })

+  it('propagates cancellation while inspecting a cold child', async () => {
+    const { ctx, parent } = await setup([textResponse('first')])
+    const started = await ctx.subagents.startContinuable(startSpec(parent))
+    await waitNoActivation(ctx, started.childId)
+    const inspectStarted = Promise.withResolvers<undefined>()
+    const inspect = vi.spyOn(ctx.sessionPersistence, 'inspect').mockImplementation((_id, signal) => {
+      return new Promise<never>((_resolve, reject) => {
+        if (signal === undefined) {
+          reject(new Error('cold inspection must receive the followup signal'))
+          return
+        }
+        inspectStarted.resolve(undefined)
+        signal.addEventListener('abort', () => {
+          reject(reason)
+        }, { once: true })
+      })
+    })
+    const controller = new AbortController()
+    const reason = new Error('cold inspection cancelled')
+
+    try {
+      const delivery = followup(ctx, parent, started.childId, message('cancel me'), controller.signal)
+      await inspectStarted.promise
+      controller.abort(reason)
+      await expect(delivery).rejects.toBe(reason)
+    } finally {
+      inspect.mockRestore()
+    }
+  })
+
  it('preserves a SubagentError raised while cold-materializing a child', async () => {
    const { ctx, parent } = await setup([textResponse('first')])
    const started = await ctx.subagents.startContinuable(startSpec(parent))