From 2f98f299928792477b1f4a4ee46c5770d4c0040e Mon Sep 17 00:00:00 2001 From: imccyu Date: Thu, 6 Aug 2026 05:13:26 +0800 Subject: [PATCH] fix(session): address restore review feedback --- .../2026-08-05-session-preparation.i18n.yaml | 4 +- .../2026-08-05-session-preparation.md | 2 +- .../2026-08-05-session-preparation.zh.md | 2 +- docs/cordis-catalog/services.md | 19 ++--- .../cordis/tool-cordis/src/api-catalog.ts | 10 +-- packages/core/session/src/index.ts | 12 ++-- .../session-persistence-jsonl/src/index.ts | 5 +- .../session-persistence/README.i18n.yaml | 4 +- .../session-persistence/README.md | 2 +- .../session-persistence/README.zh.md | 2 +- .../session-persistence/src/coordinator.ts | 11 ++- .../session-persistence/src/index.ts | 7 +- .../session-persistence/src/preparations.ts | 22 ++++-- .../tests/persistence.spec.ts | 71 +++++++++++++++++++ .../tests/preparations.spec.ts | 67 ++++++++++++++++- .../subagent/subagent/src/continuation.ts | 1 + .../subagent/tests/continuation.spec.ts | 30 ++++++++ 17 files changed, 233 insertions(+), 38 deletions(-) diff --git a/.agents/notes/implemented/architecture/2026-08-05-session-preparation.i18n.yaml b/.agents/notes/implemented/architecture/2026-08-05-session-preparation.i18n.yaml index ebb80b31aa..2e5125fa4c 100644 --- a/.agents/notes/implemented/architecture/2026-08-05-session-preparation.i18n.yaml +++ b/.agents/notes/implemented/architecture/2026-08-05-session-preparation.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 .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 diff --git a/.agents/notes/implemented/architecture/2026-08-05-session-preparation.md b/.agents/notes/implemented/architecture/2026-08-05-session-preparation.md index a3dfb50c84..d6b6cc1907 100644 --- a/.agents/notes/implemented/architecture/2026-08-05-session-preparation.md +++ b/.agents/notes/implemented/architecture/2026-08-05-session-preparation.md @@ -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 diff --git a/.agents/notes/implemented/architecture/2026-08-05-session-preparation.zh.md b/.agents/notes/implemented/architecture/2026-08-05-session-preparation.zh.md index 031e96f0e4..191ad84db0 100644 --- a/.agents/notes/implemented/architecture/2026-08-05-session-preparation.zh.md +++ b/.agents/notes/implemented/architecture/2026-08-05-session-preparation.zh.md @@ -43,7 +43,7 @@ agent loop(智能体循环)通过同一条设置与发布流水线消费这 - 缓存属于单个持久化协调器,而不是进程全局 Session map。实时 Session 由现有存储持有,绝不占用准备容量。 - 新建流程绝不认领相同 id 的冷持久化准备对象。持久化冲突仍会被拒绝。 - 第三方持久化实现继续获得通过 `load()` 实现的抽象 `prepare()` 回退。它们使用相同发布接口,但只有覆盖准备流程后才能复用精确对象。 -- Revision 校验在复用点和修复提交点建立新鲜性,但不会为后端增加跨进程 writer 排他。 +- Revision 校验在复用点和修复提交点建立新鲜性,但不会为后端增加跨进程 writer 排他。持久日志在一次读取与复核往返内保持不变后,重试才能收敛,因此持续的外部写入可能延迟准备。 ## 验证 diff --git a/docs/cordis-catalog/services.md b/docs/cordis-catalog/services.md index 1d2335df3c..db1322a275 100644 --- a/docs/cordis-catalog/services.md +++ b/docs/cordis-catalog/services.md @@ -1178,7 +1178,8 @@ abstract append(id: SessionId, events: readonly SessionEvent[]): Promise * 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 * 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 * 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-`. - * @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-`. - * @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 diff --git a/packages/cordis/tool-cordis/src/api-catalog.ts b/packages/cordis/tool-cordis/src/api-catalog.ts index f366c16c57..d00d545068 100644 --- a/packages/cordis/tool-cordis/src/api-catalog.ts +++ b/packages/cordis/tool-cordis/src/api-catalog.ts @@ -582,15 +582,15 @@ export const SERVICE_API: readonly ServiceApiEntry[] = [ }, { signature: 'async prepare(id: SessionId, signal?: AbortSignal): Promise', - 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', - 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', - 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 * @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 * @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 * @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 * @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', diff --git a/packages/core/session/src/index.ts b/packages/core/session/src/index.ts index 645dc704ed..c745638b24 100644 --- a/packages/core/session/src/index.ts +++ b/packages/core/session/src/index.ts @@ -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-`. - * @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-`. - * @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 diff --git a/packages/session-persistence/session-persistence-jsonl/src/index.ts b/packages/session-persistence/session-persistence-jsonl/src/index.ts index bbce17b7db..8523a5ffdd 100644 --- a/packages/session-persistence/session-persistence-jsonl/src/index.ts +++ b/packages/session-persistence/session-persistence-jsonl/src/index.ts @@ -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 { signal?.throwIfAborted() await this.ensureRootEncoding() diff --git a/packages/session-persistence/session-persistence/README.i18n.yaml b/packages/session-persistence/session-persistence/README.i18n.yaml index a4e98264f3..7c96b1a541 100644 --- a/packages/session-persistence/session-persistence/README.i18n.yaml +++ b/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: 327a61d1b8bf10239eab0ca3cb18ac865fc2dcce -README.zh.md: 2cc0b94bfd46a9e75c7a3626554f2ab9e5949600 +README.md: b29ff5ba17f384e8d3b1700ed3ad6c80aeeb184c +README.zh.md: 8ca4b6a7128383fdca706f35914a06eb1f26de02 diff --git a/packages/session-persistence/session-persistence/README.md b/packages/session-persistence/session-persistence/README.md index 327a61d1b8..b29ff5ba17 100644 --- a/packages/session-persistence/session-persistence/README.md +++ b/packages/session-persistence/session-persistence/README.md @@ -54,7 +54,7 @@ The `PersistenceBackend` 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 diff --git a/packages/session-persistence/session-persistence/README.zh.md b/packages/session-persistence/session-persistence/README.zh.md index 2cc0b94bfd..8ca4b6a712 100644 --- a/packages/session-persistence/session-persistence/README.zh.md +++ b/packages/session-persistence/session-persistence/README.zh.md @@ -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)。 ## 元数据与位置类型 diff --git a/packages/session-persistence/session-persistence/src/coordinator.ts b/packages/session-persistence/session-persistence/src/coordinator.ts index 7cf0d50d48..876fd5f818 100644 --- a/packages/session-persistence/session-persistence/src/coordinator.ts +++ b/packages/session-persistence/session-persistence/src/coordinator.ts @@ -639,6 +639,8 @@ export class PersistenceCoordinator { /** * 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 { /** * 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 { /** * 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 { 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) diff --git a/packages/session-persistence/session-persistence/src/index.ts b/packages/session-persistence/session-persistence/src/index.ts index 4ee46600cd..c7955c34e3 100644 --- a/packages/session-persistence/session-persistence/src/index.ts +++ b/packages/session-persistence/session-persistence/src/index.ts @@ -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. diff --git a/packages/session-persistence/session-persistence/src/preparations.ts b/packages/session-persistence/session-persistence/src/preparations.ts index a4a88b3ef8..31f6fb3d31 100644 --- a/packages/session-persistence/session-persistence/src/preparations.ts +++ b/packages/session-persistence/session-persistence/src/preparations.ts @@ -111,6 +111,7 @@ export class SessionPreparations { this.makeReady(entry) throw error } + if (this.entries.get(id) !== entry) return undefined const reservation: SessionPreparationReservation = { entry, source: committed.source, @@ -190,6 +191,20 @@ export class SessionPreparations { 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 { */ 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 { void loading.then((source) => { if (this.entries.get(id) === entry) { entry.source = source - entry.phase = 'ready' + this.makeReady(entry) } deferred.resolve(source) }, (error: unknown) => { diff --git a/packages/session-persistence/session-persistence/tests/persistence.spec.ts b/packages/session-persistence/session-persistence/tests/persistence.spec.ts index a0f936a6a2..3d365cd59d 100644 --- a/packages/session-persistence/session-persistence/tests/persistence.spec.ts +++ b/packages/session-persistence/session-persistence/tests/persistence.spec.ts @@ -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 + const fiber = await ctx.plugin(Object.assign((inner: Context) => { + coordinator = new PersistenceCoordinator(inner, backend) + }, { inject: ['sessions'] })) + let preparation: Awaited> | 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 + 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) diff --git a/packages/session-persistence/session-persistence/tests/preparations.spec.ts b/packages/session-persistence/session-persistence/tests/preparations.spec.ts index cec0ac370b..b59a1590d5 100644 --- a/packages/session-persistence/session-persistence/tests/preparations.spec.ts +++ b/packages/session-persistence/session-persistence/tests/preparations.spec.ts @@ -60,6 +60,32 @@ describe('SessionPreparations inspection', () => { expect(load).toHaveBeenCalledOnce() }) + it('evicts completed loads whose observers cancelled before readiness', async () => { + const preparations = new SessionPreparations(1) + const firstId = SessionId('cancelled-ready-first') + const secondId = SessionId('cancelled-ready-second') + const firstGate = Promise.withResolvers() + const secondGate = Promise.withResolvers() + 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(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(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(1) + const id = SessionId('invalidated-successful-commit') + const source = prepared(id) + const commitStarted = Promise.withResolvers() + const commitGate = Promise.withResolvers() + 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(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(1) const id = SessionId('take-ready') const gate = Promise.withResolvers() 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 diff --git a/packages/subagent/subagent/src/continuation.ts b/packages/subagent/subagent/src/continuation.ts index 869656a550..3644180056 100644 --- a/packages/subagent/subagent/src/continuation.ts +++ b/packages/subagent/subagent/src/continuation.ts @@ -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() diff --git a/packages/subagent/subagent/tests/continuation.spec.ts b/packages/subagent/subagent/tests/continuation.spec.ts index 38cb358fc5..7b7a2ab541 100644 --- a/packages/subagent/subagent/tests/continuation.spec.ts +++ b/packages/subagent/subagent/tests/continuation.spec.ts @@ -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() + const inspect = vi.spyOn(ctx.sessionPersistence, 'inspect').mockImplementation((_id, signal) => { + return new Promise((_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))