From feb2c35cefb83155e86e88875842e5866c402d0c Mon Sep 17 00:00:00 2001 From: imccyu Date: Thu, 6 Aug 2026 02:53:48 +0800 Subject: [PATCH] fix(session-persistence): address preparation review feedback --- ...ed-persistence-write-coordinator.i18n.yaml | 4 +- ...18-shared-persistence-write-coordinator.md | 2 +- ...shared-persistence-write-coordinator.zh.md | 2 +- .../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 | 23 ++-- .../persistence.i18n.yaml | 4 +- docs/core-data-structures/persistence.md | 2 +- docs/core-data-structures/persistence.zh.md | 2 +- .../session-query.i18n.yaml | 4 +- docs/core-data-structures/session-query.md | 1 + docs/core-data-structures/session-query.zh.md | 1 + .../cordis/tool-cordis/src/api-catalog.ts | 4 +- packages/core/session/src/index.ts | 8 +- .../apiproxy/tests/api-proxy-cold.spec.ts | 61 ++++++++++ .../session-persistence-jsonl/src/index.ts | 4 +- .../session-persistence-sqlite/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 | 72 +++++++----- .../session-persistence/src/index.ts | 21 ++-- .../session-persistence/src/preparations.ts | 49 +++++--- .../tests/persistence.spec.ts | 105 ++++++++++++------ .../tests/preparations.spec.ts | 28 +++++ .../session-query/README.i18n.yaml | 4 +- .../session-query/session-query/README.md | 2 +- .../session-query/session-query/README.zh.md | 2 +- .../session-query/session-query/src/config.ts | 1 + .../session-query/session-query/src/corpus.ts | 9 +- .../src/service-boundary.ts | 4 + .../tests/sqlite-integration.spec.ts | 42 +++++++ .../subagent/subagent/src/list-children.ts | 2 + .../subagent/tests/list-children.spec.ts | 4 +- 35 files changed, 364 insertions(+), 124 deletions(-) diff --git a/.agents/notes/implemented/architecture/2026-06-18-shared-persistence-write-coordinator.i18n.yaml b/.agents/notes/implemented/architecture/2026-06-18-shared-persistence-write-coordinator.i18n.yaml index 8edf7aa98e..7216c38d38 100644 --- a/.agents/notes/implemented/architecture/2026-06-18-shared-persistence-write-coordinator.i18n.yaml +++ b/.agents/notes/implemented/architecture/2026-06-18-shared-persistence-write-coordinator.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-06-18-shared-persistence-write-coordinator.md -2026-06-18-shared-persistence-write-coordinator.md: b7a9f7306b7d24e2c924a623b02f7329634f242f -2026-06-18-shared-persistence-write-coordinator.zh.md: 9ceacb516bb29561dfdaa8222e9232c1bce10c8d +2026-06-18-shared-persistence-write-coordinator.md: 66b73b60ceec9497f1f1226747b8cebd831eb426 +2026-06-18-shared-persistence-write-coordinator.zh.md: 424ce6ec7384e8af7b979a29f58c31379a1d1850 diff --git a/.agents/notes/implemented/architecture/2026-06-18-shared-persistence-write-coordinator.md b/.agents/notes/implemented/architecture/2026-06-18-shared-persistence-write-coordinator.md index b7a9f7306b..66b73b60ce 100644 --- a/.agents/notes/implemented/architecture/2026-06-18-shared-persistence-write-coordinator.md +++ b/.agents/notes/implemented/architecture/2026-06-18-shared-persistence-write-coordinator.md @@ -35,7 +35,7 @@ The single design choice that keeps the seam clean: the crash-repair "where is t ## Testing -The shared `runPersistenceContract` (public-API contract) runs for every backend and proves that `inspect` balances an interrupted logical view without changing storage or revisions before `prepare` or `load` commits recovery. `runCoordinatorContract` (`tests/coordinator-contract.ts`) covers preparation reuse and reservation plus adoption, HMR, collision, session and backend disposal drains, and crash-tail repair through an in-memory reference, JSONL, and SQLite. Coordinator-specific tests cover bounded prepared-state eviction, eager follow-up batches, live-controller cleanup, same-id chain-tail races, failed-drain retry, and close ordering. The per-backend specs retain storage mechanics only. A through-coordinator torn-tail repair test per real backend keeps the opaque-marker branch covered because the contract crash case produces synthetic closers without a torn marker. +The shared `runPersistenceContract` (public-API contract) runs for every backend and proves that `inspect` balances an interrupted logical view without changing storage or revisions before `prepare` or `load` commits recovery. `runCoordinatorContract` (`tests/coordinator-contract.ts`) covers adoption, HMR, collision, session and backend disposal drains, and crash-tail repair through an in-memory reference, JSONL, and SQLite. `persistence.spec.ts` and `preparations.spec.ts` cover preparation reuse and reservation, bounded prepared-state eviction, eager follow-up batches, live-controller cleanup, same-id chain-tail races, failed-drain retry, and close ordering. The per-backend specs retain storage mechanics only. A through-coordinator torn-tail repair test per real backend keeps the opaque-marker branch covered because the contract crash case produces synthetic closers without a torn marker. ## Alternatives considered diff --git a/.agents/notes/implemented/architecture/2026-06-18-shared-persistence-write-coordinator.zh.md b/.agents/notes/implemented/architecture/2026-06-18-shared-persistence-write-coordinator.zh.md index 9ceacb516b..424ce6ec73 100644 --- a/.agents/notes/implemented/architecture/2026-06-18-shared-persistence-write-coordinator.zh.md +++ b/.agents/notes/implemented/architecture/2026-06-18-shared-persistence-write-coordinator.zh.md @@ -35,7 +35,7 @@ Status: implemented ## 测试 -共享的 `runPersistenceContract`（公开 API 契约）为每个后端运行，并证明 `inspect` 会配平被中断的逻辑视图但不改变存储或修订版本，随后由 `prepare` 或 `load` 提交恢复。`runCoordinatorContract`（`tests/coordinator-contract.ts`）通过内存参考实现、JSONL 与 SQLite 覆盖准备复用与预留，以及接管、HMR、碰撞、会话与后端 dispose 排空和崩溃尾部修复。协调器专属测试覆盖有界准备状态淘汰、立即执行的后续批次、存活控制器清理、同 id 链尾竞态、排空失败重试与关闭顺序。各后端自身的测试规格只保留存储机制。每个真实后端都有一个经由协调器的崩溃尾部修复测试，以覆盖不透明 marker 分支，因为契约中的崩溃用例会产生合成 closers，却不会产生 torn marker。 +共享的 `runPersistenceContract`（公开 API 契约）为每个后端运行，并证明 `inspect` 会配平被中断的逻辑视图但不改变存储或修订版本，随后由 `prepare` 或 `load` 提交恢复。`runCoordinatorContract`（`tests/coordinator-contract.ts`）通过内存参考实现、JSONL 与 SQLite 覆盖接管、HMR、碰撞、会话与后端 dispose 排空和崩溃尾部修复。`persistence.spec.ts` 与 `preparations.spec.ts` 覆盖准备复用与预留、有界准备状态淘汰、立即执行的后续批次、存活控制器清理、同 id 链尾竞态、排空失败重试与关闭顺序。各后端自身的测试规格只保留存储机制。每个真实后端都有一个经由协调器的崩溃尾部修复测试，以覆盖不透明 marker 分支，因为契约中的崩溃用例会产生合成 closers，却不会产生 torn marker。 ## 曾考虑的替代方案 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 ffd776ecfa..ec07467ab1 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: df9295fc63d71fe0346c7f54f474545cf3b570e9 -2026-08-05-session-preparation.zh.md: cf237c4b1e4b1470741047a3cc5b44a0c33b17d4 +2026-08-05-session-preparation.md: d0ec7e361c9499ec62c7a4860d7b1df9c3c1d449 +2026-08-05-session-preparation.zh.md: 089a6968eb5e42d146a74f324aa7e0f25c4d311a 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 df9295fc63..d0ec7e361c 100644 --- a/.agents/notes/implemented/architecture/2026-08-05-session-preparation.md +++ b/.agents/notes/implemented/architecture/2026-08-05-session-preparation.md @@ -44,7 +44,7 @@ Cold continuable-subagent access follows the same path. Descriptor authorization ## Verification -The shared persistence contract and coordinator tests pin same-id in-flight sharing, exact Session reuse across inspect and prepare, single repair commit, exclusive reservation, release after failed setup, ready-entry LRU eviction, append rejection during reservation, and publication of only the reserved Session. Agent-loop and continuable-subagent tests pin the common publication pipeline and inspection-to-resume path across cancellation and teardown. +The shared persistence contract pins non-mutating balanced cold inspection and later repair. `persistence.spec.ts` and `preparations.spec.ts` pin same-id in-flight sharing, exact Session reuse across inspect and prepare, single repair commit, exclusive reservation, release after failed setup, ready-entry LRU eviction, append rejection during reservation, and publication of only the reserved Session. Agent-loop and continuable-subagent tests pin the common publication pipeline and inspection-to-resume path across cancellation and teardown. ## Alternatives considered 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 cf237c4b1e..089a6968eb 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 @@ -44,7 +44,7 @@ agent loop（智能体循环）通过同一条设置与发布流水线消费这 ## 验证 -共享持久化契约和协调器测试覆盖同 id 进行中读取共享、检查与准备之间的精确 Session 复用、修复只提交一次、独占预留、设置失败后释放、就绪项 LRU 淘汰、预留期间拒绝 append，以及只允许发布预留 Session。agent loop 与 continuable subagent 测试覆盖统一发布流水线，以及取消和拆卸期间从检查到恢复的路径。 +共享持久化契约覆盖无变更且已配平的冷检查与后续修复。`persistence.spec.ts` 与 `preparations.spec.ts` 覆盖同 id 进行中读取共享、检查与准备之间的精确 Session 复用、修复只提交一次、独占预留、设置失败后释放、就绪项 LRU 淘汰、预留期间拒绝 append，以及只允许发布预留 Session。agent loop 与 continuable subagent 测试覆盖统一发布流水线，以及取消和拆卸期间从检查到恢复的路径。 ## 考虑过的替代方案 diff --git a/docs/cordis-catalog/services.md b/docs/cordis-catalog/services.md index f395fd1915..ce781fb004 100644 --- a/docs/cordis-catalog/services.md +++ b/docs/cordis-catalog/services.md @@ -1199,14 +1199,17 @@ async prepare(id: SessionId, signal?: AbortSignal): Promise abstract load(id: SessionId): Promise /** - * Inspect an immutable balanced logical session without committing recovery - * or publishing it. A complete interrupted turn receives synthetic closers - * in memory and a torn physical tail remains untouched. Coordinator-backed - * implementations retain the exact unpublished Session for bounded reuse by - * a later {@link prepare}; callers borrow only its immutable header and log. + * Inspect an immutable logical session without committing recovery or + * publishing it. A cold complete interrupted turn receives synthetic closers + * in memory and a torn physical tail remains untouched. An already-live + * Session instead yields its current immutable snapshot, which may contain an + * 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}; callers borrow only its immutable header + * and log. * @param id - the persisted session to inspect. * @param signal - optional cancellation for queued and backend read work. - * @returns the validated header and balanced logical event log. + * @returns the validated header and current logical event log. */ abstract inspect(id: SessionId, signal?: AbortSignal): Promise @@ -1252,7 +1255,7 @@ abstract listSnapshots(signal?: AbortSignal): Promise`. - * @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 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 diff --git a/docs/core-data-structures/persistence.i18n.yaml b/docs/core-data-structures/persistence.i18n.yaml index ee8bde8889..58321e0f22 100644 --- a/docs/core-data-structures/persistence.i18n.yaml +++ b/docs/core-data-structures/persistence.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 docs/core-data-structures/persistence.md -persistence.md: 4b4b1ed0c0809bb277cd0118bb14f893faa4d821 -persistence.zh.md: 226dbd03c514d2c76838d7ee64ecc699a7852cba +persistence.md: 0968496201defa869d94925e8e5ae3c5da1bbd37 +persistence.zh.md: efb01427b4355e531fb9b86922223cf27d3b3db0 diff --git a/docs/core-data-structures/persistence.md b/docs/core-data-structures/persistence.md index 4b4b1ed0c0..0968496201 100644 --- a/docs/core-data-structures/persistence.md +++ b/docs/core-data-structures/persistence.md @@ -16,7 +16,7 @@ A backend that reloads a log crashed mid-turn finds an open `turn/start` with no Repair applies only to cold sessions. For a live id, `SessionPersistence.load(id)` waits until the authoritative in-memory snapshot is durable and returns it only when balanced; an open live turn rejects rather than receiving synthetic interruption boundaries. HMR adopts a live prefix without closing its active turn. -`SessionPersistence.inspect(id)` constructs an immutable balanced logical Session without publishing it or writing recovery. Torn physical tails remain untouched, and synthetic interruption closers exist only in memory. Coordinator-backed implementations retain that exact unpublished Session in a bounded LRU, so repeated history reads and a later `prepare(id)` share one read, decompression, validation, freeze, and Session construction. `prepare(id)` reserves the Session, commits pending repair, and returns a disposable publication handle; `load(id)` uses the same machinery to commit repair without publication. The [Session preparation decision](../../.agents/notes/implemented/architecture/2026-08-05-session-preparation.md) owns this lifecycle. +`SessionPersistence.inspect(id)` constructs an immutable logical Session without publishing it or writing recovery. Cold inspection balances an interrupted turn in memory while leaving torn physical tails untouched; inspection of an already-live Session borrows its current immutable snapshot and may therefore contain an open turn. Coordinator-backed implementations retain the exact cold unpublished Session in a bounded LRU, so repeated history reads and a later `prepare(id)` share one read, decompression, validation, freeze, and Session construction. `prepare(id)` reserves the Session, commits pending repair, and returns a disposable publication handle; `load(id)` uses the same machinery to commit repair without publication. The [Session preparation decision](../../.agents/notes/implemented/architecture/2026-08-05-session-preparation.md) owns this lifecycle. ## `SessionLocation` — optional per-session artifact target diff --git a/docs/core-data-structures/persistence.zh.md b/docs/core-data-structures/persistence.zh.md index 226dbd03c5..efb01427b4 100644 --- a/docs/core-data-structures/persistence.zh.md +++ b/docs/core-data-structures/persistence.zh.md @@ -16,7 +16,7 @@ 修复仅适用于冷会话。对于活跃 id，`SessionPersistence.load(id)` 会等待权威内存快照完成持久化，并且只在日志平衡时返回；若活跃轮次仍未闭合，则拒绝操作，而不是添加合成的中断边界。HMR 会接管活跃前缀，而不会关闭其中正在进行的轮次。 -`SessionPersistence.inspect(id)` 会构造一个不可变且已配平的逻辑 Session，但不发布它，也不写入恢复内容。撕裂的物理尾部保持不变，合成的中断 closer 只存在于内存中。使用协调器的实现会在有界 LRU 中保留这个精确的未发布 Session，因此重复历史读取与后续 `prepare(id)` 可复用同一次读取、解压、验证、冻结及 Session 构造。`prepare(id)` 会预留该 Session、提交待处理修复并返回可 dispose 的发布句柄；`load(id)` 使用相同机制提交修复，但不会发布 Session。该生命周期由 [Session 准备阶段决策](../../.agents/notes/implemented/architecture/2026-08-05-session-preparation.md)定义。 +`SessionPersistence.inspect(id)` 会构造一个不可变的逻辑 Session，但不发布它，也不写入恢复内容。冷检查会在内存中配平中断的 turn，同时保持撕裂的物理尾部不变；检查已经实时存在的 Session 则借用其当前不可变快照，因此可能包含打开的 turn。使用协调器的实现会在有界 LRU 中保留这个精确的冷未发布 Session，因此重复历史读取与后续 `prepare(id)` 可复用同一次读取、解压、验证、冻结及 Session 构造。`prepare(id)` 会预留该 Session、提交待处理修复并返回可 dispose 的发布句柄；`load(id)` 使用相同机制提交修复，但不会发布 Session。该生命周期由 [Session 准备阶段决策](../../.agents/notes/implemented/architecture/2026-08-05-session-preparation.md)定义。 ## `SessionLocation`——可选的逐会话产物目标 diff --git a/docs/core-data-structures/session-query.i18n.yaml b/docs/core-data-structures/session-query.i18n.yaml index ad48bad616..eca4715ba9 100644 --- a/docs/core-data-structures/session-query.i18n.yaml +++ b/docs/core-data-structures/session-query.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 docs/core-data-structures/session-query.md -session-query.md: d92af4bac34f7d41457e9e193111c3a53fe8022e -session-query.zh.md: 8070dfda61a2945fca554939f65ae0f8b85078db +session-query.md: e7514dd6c3bc20a07395663bff40ce65e1363b78 +session-query.zh.md: 4c3dd4d435dbd8a20fbd4db5da1a7d649c2e6d0b diff --git a/docs/core-data-structures/session-query.md b/docs/core-data-structures/session-query.md index d92af4bac3..e7514dd6c3 100644 --- a/docs/core-data-structures/session-query.md +++ b/docs/core-data-structures/session-query.md @@ -338,6 +338,7 @@ The closed code union distinguishes request validation, missing targets, malform /** Stable machine-routable failure taxonomy for session reads, traces, and search. */ type SessionQueryErrorCode = | 'SESSION_QUERY_ABORTED' + | 'SESSION_QUERY_CORRUPT_SESSION' | 'SESSION_QUERY_EVENT_NOT_FOUND' | 'SESSION_QUERY_INDEX_FAILED' | 'SESSION_QUERY_INVALID_CONFIG' diff --git a/docs/core-data-structures/session-query.zh.md b/docs/core-data-structures/session-query.zh.md index 8070dfda61..4c3dd4d435 100644 --- a/docs/core-data-structures/session-query.zh.md +++ b/docs/core-data-structures/session-query.zh.md @@ -338,6 +338,7 @@ interface SessionEventTraceObservation extends SessionEventTrace { /** Stable machine-routable failure taxonomy for session reads, traces, and search. */ type SessionQueryErrorCode = | 'SESSION_QUERY_ABORTED' + | 'SESSION_QUERY_CORRUPT_SESSION' | 'SESSION_QUERY_EVENT_NOT_FOUND' | 'SESSION_QUERY_INDEX_FAILED' | 'SESSION_QUERY_INVALID_CONFIG' diff --git a/packages/cordis/tool-cordis/src/api-catalog.ts b/packages/cordis/tool-cordis/src/api-catalog.ts index 381f974e19..79589da591 100644 --- a/packages/cordis/tool-cordis/src/api-catalog.ts +++ b/packages/cordis/tool-cordis/src/api-catalog.ts @@ -590,7 +590,7 @@ export const SERVICE_API: readonly ServiceApiEntry[] = [ }, { signature: 'abstract inspect(id: SessionId, signal?: AbortSignal): Promise', - jsDoc: '/**\n * Inspect an immutable balanced logical session without committing recovery\n * or publishing it. A complete interrupted turn receives synthetic closers\n * in memory and a torn physical tail remains untouched. Coordinator-backed\n * implementations retain the exact unpublished Session for bounded reuse by\n * a later {@link prepare}; 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 balanced 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}; callers borrow only its immutable header\n * 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 */', }, { signature: 'abstract readFrom(id: SessionId, fromSeq: number, signal?: AbortSignal): Promise<{ meta: SessionHeader; events: SessionEvent[] }>', @@ -740,7 +740,7 @@ 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.\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. 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 */', }, { signature: 'prepare(id?: SessionId, options?: PrepareSessionOptions): Session', diff --git a/packages/core/session/src/index.ts b/packages/core/session/src/index.ts index ed53fcba29..645dc704ed 100644 --- a/packages/core/session/src/index.ts +++ b/packages/core/session/src/index.ts @@ -215,7 +215,7 @@ function assertSessionEventEnvelope(value: Record, index: numbe || !Object.hasOwn(event, 'seq') || typeof event['seq'] !== 'number' || !Number.isSafeInteger(event['seq']) || event['seq'] < 0 || !Object.hasOwn(event, 'time') || typeof event['time'] !== 'number' - || !Number.isSafeInteger(event['time']) || event['time'] < 0 + || !Number.isSafeInteger(event['time']) || !Object.hasOwn(event, 'data')) { throw new Error(`seed event at index ${index} has an invalid event envelope`) } @@ -821,7 +821,11 @@ 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. + * @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 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 diff --git a/packages/host/apiproxy/tests/api-proxy-cold.spec.ts b/packages/host/apiproxy/tests/api-proxy-cold.spec.ts index b8d867e1fa..5b3bebfcc5 100644 --- a/packages/host/apiproxy/tests/api-proxy-cold.spec.ts +++ b/packages/host/apiproxy/tests/api-proxy-cold.spec.ts @@ -15,6 +15,11 @@ import { MessageId } from '@deepseek-ai/dsh-llm' import type { Agent } from '@deepseek-ai/dsh-agent' import UserInteractionService from '@deepseek-ai/dsh-user-interaction' import type { SessionEvent, SessionHeader, SessionId } from '@deepseek-ai/dsh-session' +import { + PersistenceCoordinator, + type PersistenceBackend, + type StoredPrefix, +} from '@deepseek-ai/dsh-session-persistence' import type { RpcRequest } from '@deepseek-ai/dsh-host-apiproxy/api/rpc' import { RpcId } from '@deepseek-ai/dsh-host-apiproxy/api/rpc' import { createApiProxy } from '@deepseek-ai/dsh-host-apiproxy' @@ -114,6 +119,62 @@ describe('attached updatedAt excludes end-seed', () => { }) }) +describe('cold history recovery view', () => { + it('shows in-memory interruption repair without activating the session', async () => { + const ctx = new Context() + await ctx.plugin(SessionStore) + await ctx.plugin(UserInteractionService) + const sessionId = sid('session-interrupted') + const meta = header(sessionId, 1000) + const stored: StoredPrefix = { + meta, + events: [{ type: 'turn/start', seq: 0, time: 1, data: { turn: 1 } }], + } + const backend: PersistenceBackend = { + name: 'history-recovery-test', + loadStored: id => Promise.resolve(id === sessionId ? structuredClone(stored) : undefined), + appendBatch: () => Promise.resolve(), + commitRepair: () => Promise.resolve(), + list: () => Promise.resolve([structuredClone(meta)]), + } + const coordinator = new PersistenceCoordinator(ctx, backend) + ctx.provide('sessionPersistence', { + list: (signal?: AbortSignal) => backend.list(signal), + inspect: (id: SessionId, signal?: AbortSignal) => coordinator.inspect(id, signal), + locate: () => undefined, + } as never) + const api = createApiProxy(ctx, { provider: 'p', model: 'm', cwd: '/tmp', workspaceRoot: '/tmp' }) + + const history = await api.sessions.history(request({ sessionId, beforeSeq: 2, maxMessages: 10 })) + if (!history.result.ok) throw new Error('history failed') + expect(history.result.value.events.map(entry => entry.event)).toMatchInlineSnapshot(` + [ + { + "data": { + "turn": 1, + }, + "seq": 0, + "time": 1, + "type": "turn/start", + }, + { + "data": { + "reason": { + "kind": "interrupted", + }, + "turn": 1, + }, + "seq": 1, + "time": 1, + "type": "turn/end", + }, + ] + `) + expect(ctx.sessions.get(sessionId)).toBeUndefined() + await ctx.fiber.dispose() + }) +}) + describe('subagent ownership fence', () => { it('reads a cold child without an Agent and rejects generic resume or adoption', async () => { const ctx = new Context() diff --git a/packages/session-persistence/session-persistence-jsonl/src/index.ts b/packages/session-persistence/session-persistence-jsonl/src/index.ts index c153325139..8c7b92749a 100644 --- a/packages/session-persistence/session-persistence-jsonl/src/index.ts +++ b/packages/session-persistence/session-persistence-jsonl/src/index.ts @@ -105,11 +105,13 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi // Resolve once so later process.cwd() changes cannot split one backend across roots. this.root = resolve(config.root) // Programmatic wrappers may construct the backend without Schemastery normalization. + const preparedSessionCacheSize = config.preparedSessionCacheSize + ?? DEFAULT_PREPARED_SESSION_CACHE_SIZE this.packChunks = config.packChunks ?? DEFAULT_PACK_CHUNKS this.compression = config.compression ?? DEFAULT_COMPRESSION this.assertUsableRoot() this.coordinator = new PersistenceCoordinator(this.ctx, this, { - preparedSessionCacheSize: config.preparedSessionCacheSize ?? DEFAULT_PREPARED_SESSION_CACHE_SIZE, + preparedSessionCacheSize, }) } diff --git a/packages/session-persistence/session-persistence-sqlite/src/index.ts b/packages/session-persistence/session-persistence-sqlite/src/index.ts index 875926cdec..98a747a73b 100644 --- a/packages/session-persistence/session-persistence-sqlite/src/index.ts +++ b/packages/session-persistence/session-persistence-sqlite/src/index.ts @@ -105,11 +105,14 @@ export class SessionPersistenceSqlite extends SessionPersistence implements Pers constructor(ctx: Context, public config: Config) { super(ctx) + // Programmatic wrappers may construct the backend without Schemastery normalization. + const preparedSessionCacheSize = config.preparedSessionCacheSize + ?? DEFAULT_PREPARED_SESSION_CACHE_SIZE // Open asynchronously so directory creation does not block plugin apply; // every storage hook awaits the same readiness promise. this.ready = this.openDb(config.path, (config as Required).journalMode) this.coordinator = new PersistenceCoordinator(this.ctx, this, { - preparedSessionCacheSize: (config as Required).preparedSessionCacheSize, + preparedSessionCacheSize, }) } diff --git a/packages/session-persistence/session-persistence/README.i18n.yaml b/packages/session-persistence/session-persistence/README.i18n.yaml index 221ccfb30a..e94106d89c 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: 4c00c28184c0046d5507270980822529bfe3bc08 -README.zh.md: 32461cec5efd279a86bf3af40b23e7e54d33a89e +README.md: 675f59cebe338c7bbf9bed4436db26b4593ee473 +README.zh.md: ff6589c0f29e1a634b27a4118878dc7b503c9e98 diff --git a/packages/session-persistence/session-persistence/README.md b/packages/session-persistence/session-persistence/README.md index 4c00c28184..675f59cebe 100644 --- a/packages/session-persistence/session-persistence/README.md +++ b/packages/session-persistence/session-persistence/README.md @@ -15,7 +15,7 @@ The persisted unit IS the existing `SessionEvent` (event-sourced model — the l | `append(id, events): Promise` | Durably persist a batch. Append-only; first event `seq` == stored next-seq after any repair; rejects non-JSON-serializable data naming the offending type. | | `prepare(id, signal?): Promise` | Reserve the exact unpublished Session used by resume. A coordinator reuses an earlier inspection when available, commits pending recovery, and releases an unpublished reservation back to its bounded cache on disposal. | | `load(id): Promise<{ meta; events }>` | Return an immutable balanced logical log after supported same-version shape upgrades and commit cold recovery. A live load first flushes its snapshot and rejects while its turn is open; a cold load preserves an interrupted final turn and durably closes it with synthetic `tool/result`/`step/end?`/`turn/end {interrupted}` events. Only a torn tail fragment is dropped; committed corruption, malformed shapes, and unknown `version` reject. | -| `inspect(id, signal?): Promise<{ meta; events }>` | Return the same upgraded, validated, deeply frozen logical view without committing recovery or publishing a Session. Synthetic recovery closers exist only in memory, the physical torn tail remains untouched, and coordinator-backed implementations retain the exact unpublished Session in a bounded LRU for later `prepare`. Same-id inspections share an in-flight read. | +| `inspect(id, signal?): Promise<{ meta; events }>` | Return an upgraded, validated, deeply frozen logical view without committing recovery or publishing a Session. A cold view receives in-memory synthetic recovery closers while its physical torn tail remains untouched; an already-live view is its current immutable snapshot and may contain an open turn. Coordinator-backed implementations retain the exact cold unpublished Session in a bounded LRU for later `prepare`. Same-id inspections share an in-flight read. | | `readFrom(id, fromSeq, signal?): Promise<{ meta; events }>` | The detached physical-suffix primitive: return valid stored events with `seq >= fromSeq` without preparation caching, truncation, closers, or coordinator state. A `fromSeq` at or past the stored end returns an empty event list; a negative or non-safe-integer `fromSeq` rejects. Seek-capable backends (SQLite) read only the suffix unless a supported old shape requires prefix context for normalization; sequential backends (JSONL) parse the whole artifact and skip forward. Intended for checkpoint consumers that fold only the tail past a watermark. | | `list(signal?): Promise` | 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` | 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. | diff --git a/packages/session-persistence/session-persistence/README.zh.md b/packages/session-persistence/session-persistence/README.zh.md index 32461cec5e..ff6589c0f2 100644 --- a/packages/session-persistence/session-persistence/README.zh.md +++ b/packages/session-persistence/session-persistence/README.zh.md @@ -15,7 +15,7 @@ | `append(id, events): Promise` | 持久保存一个批次。仅追加；任何修复后，第一个事件 `seq` == 已存储 next-seq；非 JSON 可序列化数据会被拒绝，并命名违规类型。 | | `prepare(id, signal?): Promise` | 预留恢复使用的精确未发布 Session。协调器会尽可能复用之前的检查结果、提交待处理恢复，并在 dispose 时将未发布 reservation 释放回有界缓存。 | | `load(id): Promise<{ meta; events }>` | 在升级受支持的同版本形状后返回不可变、平衡的逻辑日志，并提交冷恢复。实时 load 先 flush 其快照，并在轮次开放时拒绝；冷 load 保留中断的最终轮次，并用合成 `tool/result`/`step/end?`/`turn/end {interrupted}` 事件持久关闭它。只丢弃撕裂尾部碎片；已提交损坏、格式错误的形状和未知 `version` 会被拒绝。 | -| `inspect(id, signal?): Promise<{ meta; events }>` | 返回同一份已经升级、验证和深度冻结的逻辑视图，但不提交恢复或发布 Session。合成恢复 closer 只存在于内存，物理撕裂尾部保持不变，基于协调器的实现会在有界 LRU 中保留精确的未发布 Session，供后续 `prepare` 使用。同 id 检查共享进行中的读取。 | +| `inspect(id, signal?): Promise<{ meta; events }>` | 返回已经升级、验证和深度冻结的逻辑视图，但不提交恢复或发布 Session。冷视图会获得仅存在于内存的合成恢复 closer，物理撕裂尾部保持不变；已经实时存在的视图则是当前不可变快照，可能包含打开的 turn。基于协调器的实现会在有界 LRU 中保留精确的冷未发布 Session，供后续 `prepare` 使用。同 id 检查共享进行中的读取。 | | `readFrom(id, fromSeq, signal?): Promise<{ meta; events }>` | 脱离的物理后缀原语：返回 `seq >= fromSeq` 的有效已存储事件，不进入 preparation 缓存、不截断、不合成 closer，也不发布协调器状态。`fromSeq` 达到或超过已存储末尾时返回空事件列表；负数或非安全整数 `fromSeq` 会被拒绝。可寻址后端（SQLite）只读后缀，除非受支持的旧形状需要前缀上下文才能完成规范化；顺序后端（JSONL）解析整个产物并向前跳过。用于只续折水位之后尾部的 checkpoint 消费方。 | | `list(signal?): Promise` | 从元数据轻量列出，不解析完整日志。可选信号取消后端列表工作。零事件延迟实体化会话不在 `list` 中。 | | `listSnapshots(signal?): Promise` | 返回轻量元数据和每份日志一个不透明、带品牌类型的修订值，不加载事件日志。日志及其后端存储不变时，修订保持相等；append 或变更性 load 修复后会改变；不会仅因两个存储使用相同本地计数器而冲突。可选信号请求取消后端发现工作；第一方后端会先等待所有已启动的列出工作结束，再予以拒绝，因此调用返回拒绝时，相关工作已完全停稳。 | diff --git a/packages/session-persistence/session-persistence/src/coordinator.ts b/packages/session-persistence/session-persistence/src/coordinator.ts index 40412776b1..711ed2ea48 100644 --- a/packages/session-persistence/session-persistence/src/coordinator.ts +++ b/packages/session-persistence/session-persistence/src/coordinator.ts @@ -22,6 +22,18 @@ import type { SessionPreparationReservation } from './preparations.ts' /** Default number of detached session preparations retained by a coordinator. */ export const DEFAULT_PREPARED_SESSION_CACHE_SIZE = 5 +/** Durable session contents failed validation after a successful backend read. */ +export class SessionPersistenceCorruptionError extends Error { + /** + * @param message - stable corruption context. + * @param options - original validation failure. + */ + constructor(message: string, options: ErrorOptions) { + super(message, options) + this.name = 'SessionPersistenceCorruptionError' + } +} + /** Coordinator policy supplied by a concrete persistence backend. */ export interface PersistenceCoordinatorOptions { /** Maximum completed unpublished preparations retained for reuse. */ @@ -623,7 +635,7 @@ export class PersistenceCoordinator { } const reservation = await this.preparations.reserve( id, - () => this.serialize(id, () => this.prepareCore(id, signal), signal), + () => this.serialize(id, () => this.prepareCore(id)), source => this.serialize(id, () => this.commitPrepared(source), signal), signal, ) @@ -674,16 +686,17 @@ export class PersistenceCoordinator { * Inspect a logical session without publishing it or committing recovery. * @param id - persisted session to inspect. * @param signal - optional cancellation for preparation work. - * @returns immutable prepared metadata and balanced events. + * @returns immutable prepared metadata and events; a live view may have an open turn. */ async inspect(id: SessionId, signal?: AbortSignal): Promise { - await this.waitForRetirement(id, signal) + signal?.throwIfAborted() + if (this.retirements.has(id)) await this.waitForRetirement(id, signal) const live = this.ctx.sessions.get(id) if (live !== undefined) return this.inspectLive(live) try { const source = await this.preparations.inspect( id, - () => this.serialize(id, () => this.prepareCore(id, signal), signal), + () => this.serialize(id, () => this.prepareCore(id)), signal, ) const attached = this.ctx.sessions.get(id) @@ -776,29 +789,36 @@ export class PersistenceCoordinator { } signal?.throwIfAborted() if (stored === undefined) throw new Error(`session "${id}" not found`) - const { meta, events, tornMarker } = stored - this.assertStoredId(id, meta) - this.assertVersion(meta) - const storedEvents = adoptStoredEvents(events, id) + try { + const { meta, events, tornMarker } = stored + this.assertStoredId(id, meta) + this.assertVersion(meta) + const storedEvents = adoptStoredEvents(events, id) - // Preserve complete interrupted events and synthesize only missing closers. - const closers = interruptedTurnClosers(storedEvents).map(adoptSessionEvent) - const balanced = [...storedEvents, ...closers] - const session = this.ctx.sessions.prepare(id, { - seed: balanced, - meta, - seedSource: 'persistence', - }) - const inspection: SessionInspection = Object.freeze({ - meta: session.header, - events: Object.freeze(balanced), - }) - return { - inspection, - session, - sessionLength: session.events.length, - tornMarker, - closers, + // Preserve complete interrupted events and synthesize only missing closers. + const closers = interruptedTurnClosers(storedEvents).map(adoptSessionEvent) + const balanced = [...storedEvents, ...closers] + const session = this.ctx.sessions.prepare(id, { + seed: balanced, + meta, + seedSource: 'persistence', + }) + const inspection: SessionInspection = Object.freeze({ + meta: session.header, + events: Object.freeze(balanced), + }) + return { + inspection, + session, + sessionLength: session.events.length, + tornMarker, + closers, + } + } catch (error: unknown) { + throw new SessionPersistenceCorruptionError( + `stored session "${id}" failed validation: ${String(error)}`, + { cause: error }, + ) } } diff --git a/packages/session-persistence/session-persistence/src/index.ts b/packages/session-persistence/session-persistence/src/index.ts index 35aa8bed2d..a30a636a89 100644 --- a/packages/session-persistence/session-persistence/src/index.ts +++ b/packages/session-persistence/session-persistence/src/index.ts @@ -31,7 +31,11 @@ export interface SessionInspection { } // The backend-agnostic write-path orchestration first-party backends compose. -export { DEFAULT_PREPARED_SESSION_CACHE_SIZE, PersistenceCoordinator } from './coordinator.ts' +export { + DEFAULT_PREPARED_SESSION_CACHE_SIZE, + PersistenceCoordinator, + SessionPersistenceCorruptionError, +} from './coordinator.ts' export type { PersistenceBackend, PersistenceCoordinatorOptions, @@ -134,14 +138,17 @@ export abstract class SessionPersistence extends Service { abstract load(id: SessionId): Promise /** - * Inspect an immutable balanced logical session without committing recovery - * or publishing it. A complete interrupted turn receives synthetic closers - * in memory and a torn physical tail remains untouched. Coordinator-backed - * implementations retain the exact unpublished Session for bounded reuse by - * a later {@link prepare}; callers borrow only its immutable header and log. + * Inspect an immutable logical session without committing recovery or + * publishing it. A cold complete interrupted turn receives synthetic closers + * in memory and a torn physical tail remains untouched. An already-live + * Session instead yields its current immutable snapshot, which may contain an + * 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}; callers borrow only its immutable header + * and log. * @param id - the persisted session to inspect. * @param signal - optional cancellation for queued and backend read work. - * @returns the validated header and balanced logical event log. + * @returns the validated header and current logical event log. */ abstract inspect(id: SessionId, signal?: AbortSignal): Promise diff --git a/packages/session-persistence/session-persistence/src/preparations.ts b/packages/session-persistence/session-persistence/src/preparations.ts index 5eb1593321..6cca26af5d 100644 --- a/packages/session-persistence/session-persistence/src/preparations.ts +++ b/packages/session-persistence/session-persistence/src/preparations.ts @@ -55,8 +55,8 @@ export class SessionPreparations { load: () => Promise, signal?: AbortSignal, ): Promise { - const { entry, created } = this.entryFor(id, load) - const loaded = signal === undefined || created + const entry = this.entryFor(id, load) + const loaded = signal === undefined ? await entry.result : await observeQueuedAbort(entry.result, signal) const source = entry.source ?? loaded @@ -78,10 +78,8 @@ export class SessionPreparations { commit: (source: Source) => Promise<{ source: Source; state: CommitState }>, signal?: AbortSignal, ): Promise | undefined> { - const { entry, created } = this.entryFor(id, load) - await (signal === undefined || created - ? entry.result - : observeQueuedAbort(entry.result, signal)) + const entry = this.entryFor(id, load) + await (signal === undefined ? entry.result : observeQueuedAbort(entry.result, signal)) while (this.entries.get(id) === entry && entry.phase !== 'ready') { const settled = entry.reservationSettled /* v8 ignore next -- committing/reserved transitions install this waiter synchronously. */ @@ -132,7 +130,7 @@ export class SessionPreparations { && entry.reservation !== undefined) { return entry.reservation } - throw new Error(`cannot publish session "${session.id}" while a persisted preparation exists`) + throw new Error(`cannot publish session "${session.id}": persisted state already owns this identity`) } /** @@ -216,20 +214,37 @@ export class SessionPreparations { private entryFor( id: SessionId, load: () => Promise, - ): { entry: PreparationEntry; created: boolean } { + ): PreparationEntry { const existing = this.entries.get(id) - if (existing !== undefined) return { entry: existing, created: false } - const result = Promise.resolve().then(load) - const entry: PreparationEntry = { id, result, phase: 'loading' } + if (existing !== undefined) return existing + const deferred = Promise.withResolvers() + const entry: PreparationEntry = { + id, + result: deferred.promise, + phase: 'loading', + } this.entries.set(id, entry) - void result.then((source) => { - if (this.entries.get(id) !== entry) return - entry.source = source - entry.phase = 'ready' - }, () => { + let loading: Promise + try { + // Start immediately so a same-tick serialized append queues behind this + // read. The deferred result settles only after the entry becomes ready. + loading = load() + } catch (error: unknown) { this.remove(entry) + deferred.reject(error) + return entry + } + void loading.then((source) => { + if (this.entries.get(id) === entry) { + entry.source = source + entry.phase = 'ready' + } + deferred.resolve(source) + }, (error: unknown) => { + this.remove(entry) + deferred.reject(error) }) - return { entry, created: true } + return entry } private makeReady(entry: PreparationEntry): void { diff --git a/packages/session-persistence/session-persistence/tests/persistence.spec.ts b/packages/session-persistence/session-persistence/tests/persistence.spec.ts index 31d80e3157..93d83499bd 100644 --- a/packages/session-persistence/session-persistence/tests/persistence.spec.ts +++ b/packages/session-persistence/session-persistence/tests/persistence.spec.ts @@ -353,7 +353,7 @@ describe('PersistenceCoordinator stored identity', () => { await expect(ctx.plugin(Object.assign((inner: Context) => { inner.sessions.create(id, { seed: [start], meta: header }) - }, { inject: ['sessions'] }))).rejects.toThrow(/persisted preparation exists/) + }, { inject: ['sessions'] }))).rejects.toThrow(/persisted state already owns this identity/) expect(ctx.sessions.get(id)).toBeUndefined() loadGate.resolve(true) @@ -590,6 +590,59 @@ describe('PersistenceCoordinator session preparations', () => { } }) + it('queues a same-tick cold append behind preparation readiness', async () => { + const ctx = new Context() + await ctx.plugin(SessionStore) + const backend = new ControlledBackend() + const id = SessionId('inspect-cold-append-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 inspection = coordinator.inspect(id) + const append = coordinator.append(id, [{ + type: 'turn/start', + seq: oneTurnLog().length, + time: 7, + data: { turn: 2 }, + }]) + + await expect(inspection).resolves.toMatchObject({ meta: { id } }) + await expect(append).resolves.toBeUndefined() + expect(backend.loadAttempts).toBe(1) + expect(backend.store.get(id)?.events).toHaveLength(oneTurnLog().length + 1) + } finally { + await fiber.dispose() + await ctx.fiber.dispose() + } + }) + + it('inspects an open live turn without balancing it', async () => { + const ctx = new Context() + await ctx.plugin(SessionStore) + const backend = new ControlledBackend() + let coordinator!: PersistenceCoordinator + const fiber = await ctx.plugin(Object.assign((inner: Context) => { + coordinator = new PersistenceCoordinator(inner, backend) + }, { inject: ['sessions'] })) + + try { + const session = ctx.sessions.create(SessionId('inspect-live-open-turn')) + session.append('turn/start', { turn: 1 }) + + const inspected = await coordinator.inspect(session.id) + expect(inspected.events).toBe(session.events) + expect(inspected.events.map(event => event.type)).toEqual(['turn/start']) + await expect(coordinator.load(session.id)).rejects.toThrow(/live turn is open/) + } finally { + await fiber.dispose() + await ctx.fiber.dispose() + } + }) + it('keeps synthetic recovery in memory during inspect and commits it only once on prepare', async () => { const ctx = new Context() await ctx.plugin(SessionStore) @@ -770,54 +823,36 @@ describe('PersistenceCoordinator observation cancellation', () => { } }) - it('waits for active cooperative inspection cleanup before rejecting cancellation', async () => { + it('keeps a shared cold read alive when its creating inspect is cancelled', async () => { const ctx = new Context() await ctx.plugin(SessionStore) const backend = new ControlledBackend() - const id = SessionId('active-inspect-cancellation') + const id = SessionId('creating-inspect-cancellation') backend.store.set(id, { meta: meta(id), events: oneTurnLog() }) - const cleanupGate = Promise.withResolvers() - let cleanupComplete = false - backend.beforeLoadStored = async (_attempt, signal) => { - await new Promise((resolve) => { - signal?.addEventListener('abort', () => { - void cleanupGate.promise.then(() => { - cleanupComplete = true - resolve() - }) - }, { once: true }) - }) - throw new Error('backend cancellation after cleanup') - } + const loadGate = Promise.withResolvers() + backend.beforeLoadStored = () => loadGate.promise.then(() => undefined) let coordinator!: PersistenceCoordinator const fiber = await ctx.plugin(Object.assign((inner: Context) => { coordinator = new PersistenceCoordinator(inner, backend) }, { inject: ['sessions'] })) + let prepared: Awaited> | undefined try { const controller = new AbortController() - const reason = new Error('active inspect cancelled') - const pending = coordinator.inspect(id, controller.signal) - let observedReason: unknown - const observed = pending.catch((error: unknown) => { - observedReason = error - }) + const reason = new Error('creating inspect cancelled') + const inspection = coordinator.inspect(id, controller.signal) await vi.waitFor(() => { expect(backend.loadAttempts).toBe(1) }) + const reservation = coordinator.prepare(id) controller.abort(reason) - await Promise.resolve() - - expect(observedReason).toBeUndefined() - expect(cleanupComplete).toBe(false) - cleanupGate.resolve(true) - await observed - expect(cleanupComplete).toBe(true) - expect(observedReason).toBe(reason) - const backendFailure = new Error('later inspection failure') - backend.beforeLoadStored = () => Promise.reject(backendFailure) - await expect(coordinator.inspect(id)).rejects.toBe(backendFailure) + await expect(inspection).rejects.toBe(reason) + loadGate.resolve(true) + prepared = await reservation + expect(prepared.session.id).toBe(id) + expect(backend.loadAttempts).toBe(1) } finally { - cleanupGate.resolve(true) + loadGate.resolve(true) + prepared?.[Symbol.dispose]() await fiber.dispose() await ctx.fiber.dispose() } @@ -1092,7 +1127,7 @@ describe('PersistenceCoordinator retirement', () => { await expect(ctx.plugin(Object.assign((inner: Context) => { inner.sessions.create(id) - }, { inject: ['sessions'] }))).rejects.toThrow(/persisted preparation exists/) + }, { inject: ['sessions'] }))).rejects.toThrow(/persisted state already owns this identity/) loadGate.resolve(true) await expect(coldLoad).resolves.toMatchObject({ diff --git a/packages/session-persistence/session-persistence/tests/preparations.spec.ts b/packages/session-persistence/session-persistence/tests/preparations.spec.ts index 8f3ace9aa8..7002a6d0ae 100644 --- a/packages/session-persistence/session-persistence/tests/preparations.spec.ts +++ b/packages/session-persistence/session-persistence/tests/preparations.spec.ts @@ -39,6 +39,25 @@ describe('SessionPreparations inspection', () => { expect(preparations.has(id)).toBe(false) }) + it('keeps a shared load alive when its first observer cancels', async () => { + const preparations = new SessionPreparations(1) + const id = SessionId('cancelled-first-observer') + const gate = Promise.withResolvers() + const load = vi.fn(() => gate.promise) + const controller = new AbortController() + const reason = new Error('first observer cancelled') + const first = preparations.inspect(id, load, controller.signal) + const joined = preparations.inspect(id, load) + + controller.abort(reason) + await expect(first).rejects.toBe(reason) + const source = prepared(id) + gate.resolve(source) + await expect(joined).resolves.toBe(source) + await expect(preparations.inspect(id, load)).resolves.toBe(source) + expect(load).toHaveBeenCalledOnce() + }) + it('removes failed and invalidated in-flight loads without changing their observers', async () => { const preparations = new SessionPreparations(1) const failedId = SessionId('failed-inspection') @@ -63,6 +82,15 @@ describe('SessionPreparations inspection', () => { await expect(rejected).rejects.toBe(failure) }) + it('removes a load that throws before returning its promise', async () => { + const preparations = new SessionPreparations(1) + const id = SessionId('synchronous-load-failure') + const failure = new Error('synchronous load failure') + + await expect(preparations.inspect(id, () => { throw failure })).rejects.toBe(failure) + expect(preparations.has(id)).toBe(false) + }) + it('evicts ready entries while leaving reserved entries alone', async () => { const preparations = new SessionPreparations(1) const reservedA = await preparations.reserve( diff --git a/packages/session-query/session-query/README.i18n.yaml b/packages/session-query/session-query/README.i18n.yaml index 470a0008bb..6d801d9d23 100644 --- a/packages/session-query/session-query/README.i18n.yaml +++ b/packages/session-query/session-query/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-query/session-query/README.md -README.md: df97333be3b2c2cf71dd8c9287959bcbd83a5063 -README.zh.md: 649d06cdc7c3a61f9f2459466bc9fdc3a42554e6 +README.md: b75c1f23264cfa7c9323970b3c1a77ddcfe69be0 +README.zh.md: e1b9727ff892047c56d006e02906f3270fb11293 diff --git a/packages/session-query/session-query/README.md b/packages/session-query/session-query/README.md index df97333be3..b75c1f2326 100644 --- a/packages/session-query/session-query/README.md +++ b/packages/session-query/session-query/README.md @@ -17,7 +17,7 @@ English | [中文](README.zh.md) - `traceSession(sessionId, signal?)` reads the corpus once and returns immediate-to-outward ancestors plus deterministic recursive descendant trees. `complete: false` identifies the first missing parent; a target-connected cycle fails with `SESSION_QUERY_INVALID_LINEAGE`. - `traceEvent(request, signal?)` loads the logical log once and returns its cloned source header with direct positional replacements and direct logged provenance. `replacementChain` follows positional replacers to the final replacement; provenance links remain non-transitive. -Persistence is optional and may mount or unmount dynamically. Cross-corpus listing and lineage tracing fail with `SESSION_QUERY_PERSISTENCE_FAILED` while mounted persistence is unreadable. A title read, event trace, or event read targeting a known live session does not consult persistence, so durable backend health cannot make current in-memory state unreadable. Persisted title and event operations list before loading and reject a metadata mismatch rather than combining inconsistent observations. Lineage-trace cancellation is passed to persisted listing; event-trace and event-read cancellation is passed to persisted listing and inspection. Each waits for the started backend call to settle, then rejects with the signal's exact reason even when the backend ignored that signal. A pre-aborted known-live title read, event trace, or event read rejects before folding or snapshotting without consulting persistence. A batch title observation performs one metadata listing, inspects its unique persisted ids with at most `persistedInspectConcurrency` workers, and preserves each title's own observed header for downstream authorization. Cancellation starts no queued inspections and rejects only after already-started workers settle. `listSessions()` remains lightweight and does not load logs or index titles. +Persistence is optional and may mount or unmount dynamically. Cross-corpus listing and lineage tracing fail with `SESSION_QUERY_PERSISTENCE_FAILED` while mounted persistence is unreadable; a successfully read durable record that fails Session validation reports `SESSION_QUERY_CORRUPT_SESSION` instead. A title read, event trace, or event read targeting a known live session does not consult persistence, so durable backend health cannot make current in-memory state unreadable. Persisted title and event operations list before loading and reject a metadata mismatch rather than combining inconsistent observations. Lineage-trace cancellation is passed to persisted listing; event-trace and event-read cancellation is passed to persisted listing and inspection. Each waits for the started backend call to settle, then rejects with the signal's exact reason even when the backend ignored that signal. A pre-aborted known-live title read, event trace, or event read rejects before folding or snapshotting without consulting persistence. A batch title observation performs one metadata listing, inspects its unique persisted ids with at most `persistedInspectConcurrency` workers, and preserves each title's own observed header for downstream authorization. Cancellation starts no queued inspections and rejects only after already-started workers settle. `listSessions()` remains lightweight and does not load logs or index titles. ## Filtering and extraction diff --git a/packages/session-query/session-query/README.zh.md b/packages/session-query/session-query/README.zh.md index 649d06cdc7..e1b9727ff8 100644 --- a/packages/session-query/session-query/README.zh.md +++ b/packages/session-query/session-query/README.zh.md @@ -17,7 +17,7 @@ - `traceSession(sessionId, signal?)` 只读取一次语料库，返回从直接父级向外的祖先，以及确定性的递归后代树。`complete: false` 标识第一个缺失父级；与目标相连的循环会以 `SESSION_QUERY_INVALID_LINEAGE` 失败。 - `traceEvent(request, signal?)` 只加载一次逻辑日志，返回其克隆源 header、直接位置替换和直接已记录来源信息。`replacementChain` 沿位置替换者跟踪到最终替换；来源链接仍不传递。 -持久化是可选的，可动态挂载或卸载。已挂载持久化无法读取时，跨语料库列表和血缘跟踪以 `SESSION_QUERY_PERSISTENCE_FAILED` 失败。针对已知实时会话的标题读取、事件跟踪或事件读取不会查询持久化，因此持久化后端的健康状态无法使当前内存状态变得不可读。持久化标题和事件操作在加载前先执行列表查询，并在元数据不匹配时拒绝，而不会组合不一致的观察。血缘跟踪的取消信号会传递给持久化列表查询；事件跟踪和事件读取的取消信号会传递给持久化列表查询和检查。每项操作都会等待已启动的后端调用结算，然后使用信号的精确原因拒绝，即使后端忽略了该信号。针对已知实时会话且预先中止的标题读取、事件跟踪或事件读取会在 fold 或快照之前拒绝，且不查询持久化。批量标题观察执行一次元数据列表查询，使用最多 `persistedInspectConcurrency` 个 worker 检查唯一持久化 id，并保留每个标题自己观察到的 header，供下游授权使用。取消不会启动已排队检查，且只在已启动 worker 结算后拒绝。`listSessions()` 仍保持轻量，不加载日志或索引标题。 +持久化是可选的，可动态挂载或卸载。已挂载持久化无法读取时，跨语料库列表和血缘跟踪以 `SESSION_QUERY_PERSISTENCE_FAILED` 失败；已经成功读取、但无法通过 Session 校验的持久化记录则以 `SESSION_QUERY_CORRUPT_SESSION` 失败。针对已知实时会话的标题读取、事件跟踪或事件读取不会查询持久化，因此持久化后端的健康状态无法使当前内存状态变得不可读。持久化标题和事件操作在加载前先执行列表查询，并在元数据不匹配时拒绝，而不会组合不一致的观察。血缘跟踪的取消信号会传递给持久化列表查询；事件跟踪和事件读取的取消信号会传递给持久化列表查询和检查。每项操作都会等待已启动的后端调用结算，然后使用信号的精确原因拒绝，即使后端忽略了该信号。针对已知实时会话且预先中止的标题读取、事件跟踪或事件读取会在 fold 或快照之前拒绝，且不查询持久化。批量标题观察执行一次元数据列表查询，使用最多 `persistedInspectConcurrency` 个 worker 检查唯一持久化 id，并保留每个标题自己观察到的 header，供下游授权使用。取消不会启动已排队检查，且只在已启动 worker 结算后拒绝。`listSessions()` 仍保持轻量，不加载日志或索引标题。 ## 过滤与提取 diff --git a/packages/session-query/session-query/src/config.ts b/packages/session-query/session-query/src/config.ts index 475cc55dbf..a449799193 100644 --- a/packages/session-query/session-query/src/config.ts +++ b/packages/session-query/session-query/src/config.ts @@ -19,6 +19,7 @@ export interface Config { /** Stable machine-routable failure taxonomy for session reads, traces, and search. */ export type SessionQueryErrorCode = | 'SESSION_QUERY_ABORTED' + | 'SESSION_QUERY_CORRUPT_SESSION' | 'SESSION_QUERY_EVENT_NOT_FOUND' | 'SESSION_QUERY_INDEX_FAILED' | 'SESSION_QUERY_INVALID_CONFIG' diff --git a/packages/session-query/session-query/src/corpus.ts b/packages/session-query/session-query/src/corpus.ts index 649a80965a..8711597b69 100644 --- a/packages/session-query/session-query/src/corpus.ts +++ b/packages/session-query/session-query/src/corpus.ts @@ -2,7 +2,7 @@ import type { Context, Fiber } from 'cordis' import type { Session, SessionEvent, SessionHeader, SessionId } from '@deepseek-ai/dsh-session' -import type SessionPersistence from '@deepseek-ai/dsh-session-persistence' +import SessionPersistence, { SessionPersistenceCorruptionError } from '@deepseek-ai/dsh-session-persistence' import type { SessionRecord } from './types.ts' import { SessionQueryError } from './config.ts' import { assertSessionHeadersCompatible } from './sources.ts' @@ -274,6 +274,13 @@ async function inspectPersisted( return await persistence.inspect(sessionId, signal) } catch (error: unknown) { if (signal?.aborted) signal.throwIfAborted() + if (error instanceof SessionPersistenceCorruptionError) { + throw new SessionQueryError( + `stored session "${sessionId}" is corrupt: ${errorMessage(error)}`, + 'SESSION_QUERY_CORRUPT_SESSION', + { cause: error }, + ) + } throw new SessionQueryError( `failed to inspect session "${sessionId}": ${errorMessage(error)}`, 'SESSION_QUERY_PERSISTENCE_FAILED', diff --git a/packages/session-query/tool-session-query/src/service-boundary.ts b/packages/session-query/tool-session-query/src/service-boundary.ts index bf1dbd24f4..495897fddd 100644 --- a/packages/session-query/tool-session-query/src/service-boundary.ts +++ b/packages/session-query/tool-session-query/src/service-boundary.ts @@ -23,6 +23,10 @@ const SAFE_SESSION_QUERY_FAILURES = { code: 'SESSION_QUERY_ABORTED', message: 'session query was cancelled', }, + SESSION_QUERY_CORRUPT_SESSION: { + code: 'SESSION_QUERY_CORRUPT_SESSION', + message: 'session event history is corrupt', + }, SESSION_QUERY_EVENT_NOT_FOUND: { code: 'SESSION_QUERY_EVENT_NOT_FOUND', message: 'session event was not found', diff --git a/packages/session-query/tool-session-query/tests/sqlite-integration.spec.ts b/packages/session-query/tool-session-query/tests/sqlite-integration.spec.ts index 05e8169e26..09c9c2e812 100644 --- a/packages/session-query/tool-session-query/tests/sqlite-integration.spec.ts +++ b/packages/session-query/tool-session-query/tests/sqlite-integration.spec.ts @@ -141,6 +141,26 @@ describe('tool-session-query with the real SQLite provider', () => { }), surfaceOp: 'append', }, + { + type: 'user/message', + seq: 2, + time: -124, + data: createUserMessage({ + content: [{ type: 'text', text: 'pre-epoch fractional needle' }], + source: { kind: 'user' }, + }), + surfaceOp: 'append', + }, + { + type: 'user/message', + seq: 3, + time: -123, + data: createUserMessage({ + content: [{ type: 'text', text: 'pre-epoch fractional needle' }], + source: { kind: 'user' }, + }), + surfaceOp: 'append', + }, ]) const caller = ctx.sessions.create(SessionId('fractional-caller'), { @@ -184,5 +204,27 @@ describe('tool-session-query with the real SQLite provider', () => { expect(emptySameMillisecond.isError).toBe(false) expect(emptySameMillisecond.content.map(block => block.type === 'text' ? block.text : '').join('\n')) .toContain('No prior event matches found.') + + const preEpochLower = await execute({ + session_id: persisted, + query: 'pre-epoch fractional needle', + time_from: '1969-12-31T23:59:59.87600001Z', + }) + expect(preEpochLower.isError).toBe(false) + const preEpochLowerText = preEpochLower.content + .map(block => block.type === 'text' ? block.text : '').join('\n') + expect(preEpochLowerText).toContain('seq 3') + expect(preEpochLowerText).not.toContain('seq 2') + + const preEpochUpper = await execute({ + session_id: persisted, + query: 'pre-epoch fractional needle', + time_to: '1969-12-31T19:59:59.8769999-04:00', + }) + expect(preEpochUpper.isError).toBe(false) + const preEpochUpperText = preEpochUpper.content + .map(block => block.type === 'text' ? block.text : '').join('\n') + expect(preEpochUpperText).toContain('seq 2') + expect(preEpochUpperText).not.toContain('seq 3') }) }) diff --git a/packages/subagent/subagent/src/list-children.ts b/packages/subagent/subagent/src/list-children.ts index c7d861c82f..5ec7d5279e 100644 --- a/packages/subagent/subagent/src/list-children.ts +++ b/packages/subagent/subagent/src/list-children.ts @@ -218,6 +218,8 @@ function perChildDiagnosticReason( ): 'corrupt' | 'unavailable' | undefined { if (!(error instanceof SessionQueryError)) return undefined switch (error.code) { + case 'SESSION_QUERY_CORRUPT_SESSION': + return 'corrupt' case 'SESSION_QUERY_SESSION_NOT_FOUND': case 'SESSION_QUERY_EVENT_NOT_FOUND': case 'SESSION_QUERY_PERSISTENCE_FAILED': diff --git a/packages/subagent/subagent/tests/list-children.spec.ts b/packages/subagent/subagent/tests/list-children.spec.ts index 07098d0387..04fdf986c8 100644 --- a/packages/subagent/subagent/tests/list-children.spec.ts +++ b/packages/subagent/subagent/tests/list-children.spec.ts @@ -268,7 +268,7 @@ describe('SubagentService.listChildren', () => { }) }) - it('maps a child rejected by persisted Session preparation to unavailable', async () => { + it('diagnoses a child rejected by persisted Session preparation as corrupt', async () => { const { ctx, parent } = await setup([]) // The surface-eligible user/message lacks its required surfaceOp. The // first-party persistence inspection rejects before session-query can fold it. @@ -285,7 +285,7 @@ describe('SubagentService.listChildren', () => { { type: 'subagent/descriptor', seq: 2, time: 3, data: descriptorPayload('broken surface') }, ] as SessionEvent[]) const entries = await ctx.subagents.listChildren(parent.id) - expect(entries).toEqual([{ kind: 'diagnostic', id: invalid, reason: 'unavailable' }]) + expect(entries).toEqual([{ kind: 'diagnostic', id: invalid, reason: 'corrupt' }]) }) it('diagnoses a malformed descriptor payload as corrupt', async () => {