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 ec07467ab1..ebb80b31aa 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: d0ec7e361c9499ec62c7a4860d7b1df9c3c1d449 -2026-08-05-session-preparation.zh.md: 089a6968eb5e42d146a74f324aa7e0f25c4d311a +2026-08-05-session-preparation.md: a3dfb50c8484cfef0cadea1759125a86714d3cbc +2026-08-05-session-preparation.zh.md: 031e96f0e4446f02946f8c29ddade3fb52fe396e 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 d0ec7e361c..a3dfb50c84 100644 --- a/.agents/notes/implemented/architecture/2026-08-05-session-preparation.md +++ b/.agents/notes/implemented/architecture/2026-08-05-session-preparation.md @@ -20,11 +20,11 @@ This refines the publication boundary from the [Agent lifecycle and ownership de ## Persisted preparation lifecycle -A coordinator-backed persistence implementation loads one cold source into a prepared Session. The backend transfers fresh, mutually unaliased metadata and events; the Session restore path validates and freezes those graphs in place instead of cloning them. The coordinator computes interrupted-turn closers and constructs the exact unpublished Session once. Its immutable header and balanced logical event log form the `SessionInspection` borrowed by readers. +A coordinator-backed persistence implementation loads one cold source into a prepared Session. The backend transfers fresh, mutually unaliased metadata and events together with the source-qualified revision that identifies those exact values; the Session restore path validates and freezes the graphs in place instead of cloning them. The coordinator computes interrupted-turn closers and constructs the exact unpublished Session once. Its immutable header and balanced logical event log form the `SessionInspection` borrowed by readers, while the revision remains internal to persistence. -`inspect(id, signal?)` does not mutate storage. Synthetic closers exist only in the prepared in-memory view, and a torn physical tail remains untouched. Same-id callers share an in-flight cold read. Once ready, the preparation may remain in a per-coordinator LRU whose capacity defaults to five and is configurable by first-party backends. +`inspect(id, signal?)` does not mutate storage. Synthetic closers exist only in the prepared in-memory view, and a torn physical tail remains untouched. Same-id callers share an in-flight cold read. Once ready, the preparation may remain in a per-coordinator LRU whose capacity defaults to five and is configurable by first-party backends. Before reusing a retained source, the coordinator reads that id's current revision; a mismatch evicts the source and repeats the cold materialization. -`prepare(id, signal?)` exclusively reserves the prepared Session. It commits any torn-tail and interrupted-turn repair, establishes the durable cursor, then returns a disposable preparation. Another same-id preparation waits until the reservation is published or released. Publication accepts only the exact reserved Session and attaches the committed cursor without rebuilding its history. Failed setup or cancellation returns an unchanged unpublished Session to the LRU; mutation or attachment consumes the reservation. +`prepare(id, signal?)` exclusively reserves the prepared Session. It confirms the retained revision before committing any torn-tail and interrupted-turn repair, establishes the durable cursor, then returns a disposable preparation. A stale source is discarded and reloaded instead of being repaired or published. Another same-id preparation waits until the reservation is published or released. Publication accepts only the exact reserved Session and attaches the committed cursor without rebuilding its history. Failed setup or cancellation returns an unchanged unpublished Session to the LRU; mutation or attachment consumes the reservation. The legacy `load(id)` API uses the same preparation and repair machinery, then discards its reservation and returns the immutable logical view. It remains a compatibility API, not the history-to-resume reuse path. This lifecycle extends the [shared persistence coordinator](2026-06-18-shared-persistence-write-coordinator.md) while preserving the storage and recovery rules owned by the [session persistence decision](2026-06-14-session-persistence.md). @@ -32,6 +32,8 @@ The legacy `load(id)` API uses the same preparation and repair machinery, then d History reads use `inspect()`, so repeated pages borrow the same immutable prepared state without activating an Agent. A later resume uses `prepare()` and receives the exact Session retained by inspection; it does not read, decompress, parse, clone, validate, or freeze the complete log again. +If the durable log changes after inspection, its revision changes. The next history read or resume discards the retained Session and materializes the new log, so an old event graph cannot be associated with a newer snapshot revision. + Cold continuable-subagent access follows the same path. Descriptor authorization first inspects the child, then `ctx.agents.resume()` reserves and publishes the retained Session. This preserves the lifecycle and authorization rules in the [continuable subagent conversation decision](../feature/2026-07-28-continuable-subagent-conversations.md) while removing its duplicate cold read. ## Boundaries @@ -41,10 +43,11 @@ 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. ## Verification -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. +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, revision-triggered refresh before history and resume, single repair commit, exclusive reservation, release after failed setup, ready-entry LRU eviction, append rejection during reservation, and publication of only the reserved Session. Backend tests pin that full and lightweight reads use the same revision identity. 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 089a6968eb..031e96f0e4 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 @@ -20,11 +20,11 @@ agent loop（智能体循环）通过同一条设置与发布流水线消费这 ## 持久化准备生命周期 -使用协调器的持久化实现会将一个冷源加载为准备完成的 Session。后端转移新鲜、彼此无别名的元数据和事件；Session 恢复路径直接验证并冻结这些对象图，不再复制。协调器计算中断轮次的 closer，并且只构造一次精确的未发布 Session。其不可变 header 与平衡逻辑事件日志构成读取方借用的 `SessionInspection`。 +使用协调器的持久化实现会将一个冷源加载为准备完成的 Session。后端转移新鲜、彼此无别名的元数据和事件，以及标识这些精确值的来源限定 revision；Session 恢复路径直接验证并冻结这些对象图，不再复制。协调器计算中断轮次的 closer，并且只构造一次精确的未发布 Session。其不可变 header 与平衡逻辑事件日志构成读取方借用的 `SessionInspection`，revision 则保留在持久化内部。 -`inspect(id, signal?)` 不修改存储。合成 closer 只存在于准备完成的内存视图中，撕裂的物理尾部保持不变。同 id 调用方共享进行中的冷读。准备完成后，该对象可以进入每个协调器自己的 LRU；第一方后端可配置容量，默认保留五个。 +`inspect(id, signal?)` 不修改存储。合成 closer 只存在于准备完成的内存视图中，撕裂的物理尾部保持不变。同 id 调用方共享进行中的冷读。准备完成后，该对象可以进入每个协调器自己的 LRU；第一方后端可配置容量，默认保留五个。协调器复用保留源之前会读取该 id 的当前 revision；如果不匹配，就淘汰旧源并重新完成冷实体化。 -`prepare(id, signal?)` 独占预留准备完成的 Session。它先提交撕裂尾部和中断轮次修复并建立持久游标，再返回可 dispose 的准备对象。同 id 的另一个准备请求会等待当前预留发布或释放。发布只接受精确的预留 Session，并直接附接已提交游标，无需重建历史。设置失败或取消时，未发生变化的未发布 Session 会返回 LRU；发生变更或完成附接后，系统会消费该预留。 +`prepare(id, signal?)` 独占预留准备完成的 Session。它先确认保留的 revision，再提交撕裂尾部和中断轮次修复、建立持久游标，最后返回可 dispose 的准备对象。过期源会被丢弃并重新读取，不会参与修复或发布。同 id 的另一个准备请求会等待当前预留发布或释放。发布只接受精确的预留 Session，并直接附接已提交游标，无需重建历史。设置失败或取消时，未发生变化的未发布 Session 会返回 LRU；发生变更或完成附接后，系统会消费该预留。 存量 `load(id)` API 使用相同的准备和修复机制，随后丢弃其预留并返回不可变逻辑视图。它保留为兼容 API，不承担历史到恢复的复用路径。该生命周期扩展了[共享持久化协调器](2026-06-18-shared-persistence-write-coordinator.md)，同时继续遵循[会话持久化决策](2026-06-14-session-persistence.md)所规定的存储与恢复规则。 @@ -32,6 +32,8 @@ agent loop（智能体循环）通过同一条设置与发布流水线消费这 历史读取使用 `inspect()`，因此重复分页可以借用同一份不可变准备状态，而不会激活 agent。后续恢复调用 `prepare()`，直接取得检查阶段保留的精确 Session；系统不会再次完整读取、解压、解析、复制、验证或冻结日志。 +如果持久日志在检查后发生变化，其 revision 也会变化。下一次历史读取或恢复会丢弃保留的 Session，并实体化新日志，因此旧事件对象图不会被关联到较新的快照 revision。 + 冷 continuable subagent 访问沿用同一路径。系统先检查子会话并完成 descriptor 授权，再由 `ctx.agents.resume()` 预留并发布保留的 Session。这样既遵循 [continuable subagent 会话决策](../feature/2026-07-28-continuable-subagent-conversations.md)中的生命周期与授权规则，也消除了重复冷读。 ## 边界 @@ -41,10 +43,11 @@ agent loop（智能体循环）通过同一条设置与发布流水线消费这 - 缓存属于单个持久化协调器，而不是进程全局 Session map。实时 Session 由现有存储持有，绝不占用准备容量。 - 新建流程绝不认领相同 id 的冷持久化准备对象。持久化冲突仍会被拒绝。 - 第三方持久化实现继续获得通过 `load()` 实现的抽象 `prepare()` 回退。它们使用相同发布接口，但只有覆盖准备流程后才能复用精确对象。 +- Revision 校验在复用点和修复提交点建立新鲜性，但不会为后端增加跨进程 writer 排他。 ## 验证 -共享持久化契约覆盖无变更且已配平的冷检查与后续修复。`persistence.spec.ts` 与 `preparations.spec.ts` 覆盖同 id 进行中读取共享、检查与准备之间的精确 Session 复用、修复只提交一次、独占预留、设置失败后释放、就绪项 LRU 淘汰、预留期间拒绝 append，以及只允许发布预留 Session。agent loop 与 continuable subagent 测试覆盖统一发布流水线，以及取消和拆卸期间从检查到恢复的路径。 +共享持久化契约覆盖无变更且已配平的冷检查与后续修复。`persistence.spec.ts` 与 `preparations.spec.ts` 覆盖同 id 进行中读取共享、检查与准备之间的精确 Session 复用、在历史读取与恢复前由 revision 触发刷新、修复只提交一次、独占预留、设置失败后释放、就绪项 LRU 淘汰、预留期间拒绝 append，以及只允许发布预留 Session。后端测试覆盖完整读取与轻量读取使用同一 revision 身份。agent loop 与 continuable subagent 测试覆盖统一发布流水线，以及取消和拆卸期间从检查到恢复的路径。 ## 考虑过的替代方案 diff --git a/docs/config-catalog.md b/docs/config-catalog.md index 9099f99b85..55b2c0d9d3 100644 --- a/docs/config-catalog.md +++ b/docs/config-catalog.md @@ -1184,7 +1184,7 @@ export interface Config { export type JournalMode = 'wal' | 'delete' | 'truncate' | 'persist' ``` -Source: [`packages/session-persistence/session-persistence-sqlite/src/index.ts:58`](../packages/session-persistence/session-persistence-sqlite/src/index.ts) +Source: [`packages/session-persistence/session-persistence-sqlite/src/index.ts:66`](../packages/session-persistence/session-persistence-sqlite/src/index.ts) ## `@deepseek-ai/dsh-session-projection-cache` diff --git a/docs/cordis-catalog/services.md b/docs/cordis-catalog/services.md index ce781fb004..1d2335df3c 100644 --- a/docs/cordis-catalog/services.md +++ b/docs/cordis-catalog/services.md @@ -1176,8 +1176,9 @@ 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}; disposal - * releases an unpublished reservation. + * reuse object graphs retained by an earlier {@link inspect} after confirming + * their durable revision is still current; disposal releases an unpublished + * reservation. * @param id - persisted session to prepare. * @param signal - optional cancellation for preparation work. * @returns one owned unpublished Session preparation. @@ -1205,8 +1206,8 @@ abstract load(id: SessionId): Promise * 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. + * reuse by a later {@link prepare}, reloading it when its durable revision + * changes; 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 current logical event log. diff --git a/packages/cordis/tool-cordis/src/api-catalog.ts b/packages/cordis/tool-cordis/src/api-catalog.ts index 79589da591..f366c16c57 100644 --- a/packages/cordis/tool-cordis/src/api-catalog.ts +++ b/packages/cordis/tool-cordis/src/api-catalog.ts @@ -582,7 +582,7 @@ 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}; disposal\n * releases an unpublished 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.\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', @@ -590,7 +590,7 @@ export const SERVICE_API: readonly ServiceApiEntry[] = [ }, { 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}; 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 */', + 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 */', }, { signature: 'abstract readFrom(id: SessionId, fromSeq: number, signal?: AbortSignal): Promise<{ meta: SessionHeader; events: SessionEvent[] }>', diff --git a/packages/host/apiproxy/tests/api-proxy-cold.spec.ts b/packages/host/apiproxy/tests/api-proxy-cold.spec.ts index 5b3bebfcc5..8b01e9005a 100644 --- a/packages/host/apiproxy/tests/api-proxy-cold.spec.ts +++ b/packages/host/apiproxy/tests/api-proxy-cold.spec.ts @@ -17,6 +17,7 @@ import UserInteractionService from '@deepseek-ai/dsh-user-interaction' import type { SessionEvent, SessionHeader, SessionId } from '@deepseek-ai/dsh-session' import { PersistenceCoordinator, + SessionPersistenceRevision, type PersistenceBackend, type StoredPrefix, } from '@deepseek-ai/dsh-session-persistence' @@ -129,10 +130,14 @@ describe('cold history recovery view', () => { const stored: StoredPrefix = { meta, events: [{ type: 'turn/start', seq: 0, time: 1, data: { turn: 1 } }], + revision: SessionPersistenceRevision('history-recovery-test:1'), } const backend: PersistenceBackend = { name: 'history-recovery-test', loadStored: id => Promise.resolve(id === sessionId ? structuredClone(stored) : undefined), + readStoredRevision: id => Promise.resolve( + id === sessionId ? SessionPersistenceRevision('history-recovery-test:1') : undefined, + ), appendBatch: () => Promise.resolve(), commitRepair: () => Promise.resolve(), list: () => Promise.resolve([structuredClone(meta)]), diff --git a/packages/session-persistence/session-persistence-jsonl/README.i18n.yaml b/packages/session-persistence/session-persistence-jsonl/README.i18n.yaml index fe7c9523f8..763b55a1a4 100644 --- a/packages/session-persistence/session-persistence-jsonl/README.i18n.yaml +++ b/packages/session-persistence/session-persistence-jsonl/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-jsonl/README.md -README.md: 55b802c57c27c771e0259451924501e5fb367911 -README.zh.md: a5e71f27990938576d5a77bf182dbd6919c047f1 +README.md: cd087539bde2433fcdb70b2c511ff30880a877e1 +README.zh.md: 144b404e04f8a5fd3623d9329e4fbcafd328524d diff --git a/packages/session-persistence/session-persistence-jsonl/README.md b/packages/session-persistence/session-persistence-jsonl/README.md index 55b802c57c..cd087539bd 100644 --- a/packages/session-persistence/session-persistence-jsonl/README.md +++ b/packages/session-persistence/session-persistence-jsonl/README.md @@ -44,7 +44,7 @@ A root belongs to one encoding. Startup discovery and targeted lookup reject the - **Crash recovery — preserve valid tail work.** `load` validates every complete compressed frame and scans their decompressed JSONL. If the last frame is structurally incomplete, the reader keeps its complete decoded records, truncates from that frame's start, and re-encodes those records with the synthetic tool, step, and turn closers required by the shared [persistence contract](../../../.agents/notes/implemented/architecture/2026-06-14-session-persistence.md). Raw mode truncates from its first incomplete line. A checksum/decompression failure in a complete frame, or a defect at or before the last committed `turn/end`, is corruption and rejects. - **Non-mutating inspection.** `inspect()` returns an immutable balanced logical view and may synthesize recovery closers in memory, without truncating an incomplete tail or changing the lightweight revision. - **Contiguous-seq.** `append` rejects a batch whose first `seq` does not continue the stored log, and rejects non-JSON-serializable `event.data` naming the offending event type. -- **Lightweight revisions.** `listSnapshots(signal?)` identifies a log by its device, inode, size, and nanosecond timestamps, avoiding a full-log parse while changing after append, repair, replacement, or store changes. It forwards the exact signal through artifact discovery and checks cancellation around every `stat`; because filesystem `stat` is not interruptible, cancellation waits for the active call to settle, then rejects without starting another. +- **Lightweight revisions.** `listSnapshots(signal?)` identifies a log by its device, inode, size, and nanosecond timestamps, avoiding a full-log parse while changing after append, repair, replacement, or store changes. A full-prefix read requires the same identity before and after reading the bytes, and `readStoredRevision()` uses that identity to validate retained preparations without loading the log. Snapshot listing forwards the exact signal through artifact discovery and checks cancellation around every `stat`; because filesystem `stat` is not interruptible, cancellation waits for the active call to settle, then rejects without starting another. ## Write path diff --git a/packages/session-persistence/session-persistence-jsonl/README.zh.md b/packages/session-persistence/session-persistence-jsonl/README.zh.md index a5e71f2799..144b404e04 100644 --- a/packages/session-persistence/session-persistence-jsonl/README.zh.md +++ b/packages/session-persistence/session-persistence-jsonl/README.zh.md @@ -44,7 +44,7 @@ JSONL 持久会话存储后端：`SessionPersistence` 的一个具体实现（`d - **崩溃恢复：保留有效尾部工作。**`load` 验证每个完整压缩 frame，并扫描解压 JSONL。最后 frame 结构不完整时，读取器保留其完整解码记录，从 frame 开头截断，并使用共享[持久化契约](../../../.agents/notes/implemented/architecture/2026-06-14-session-persistence.md) 需要的合成工具、步骤和轮次 closer 重新编码这些记录。原始 mode 从第一个不完整行截断。完整 frame 中的 checksum/解压失败，或位于最后已提交的 `turn/end` 处或之前的缺陷属于损坏，会被拒绝。 - **非变更检查。**`inspect()` 返回不可变、平衡的逻辑视图，并可在内存中合成恢复 closer，但不会截断不完整尾部或更改轻量修订。 - **连续 seq。**`append` 拒绝第一个 `seq` 不继续已存储日志的批次，并拒绝非 JSON 可序列化 `event.data`，同时命名违规事件类型。 -- **轻量修订。**`listSnapshots(signal?)` 使用 device、inode、size 和纳秒时间戳标识日志，避免解析完整日志；该标识会在 append、修复、替换或存储变更后改变。它通过产物发现转发精确信号，并在每个 `stat` 前后检查取消；由于文件系统 `stat` 不可中断，取消会等待活动调用完成，然后在不启动另一次调用的情况下拒绝。 +- **轻量修订。**`listSnapshots(signal?)` 使用 device、inode、size 和纳秒时间戳标识日志，避免解析完整日志；该标识会在 append、修复、替换或存储变更后改变。完整前缀读取要求读取字节前后的身份一致，`readStoredRevision()` 使用同一身份校验保留的 preparation，而不加载日志。快照列表通过产物发现转发精确信号，并在每个 `stat` 前后检查取消；由于文件系统 `stat` 不可中断，取消会等待活动调用完成，然后在不启动另一次调用的情况下拒绝。 ## 写入路径 diff --git a/packages/session-persistence/session-persistence-jsonl/src/index.ts b/packages/session-persistence/session-persistence-jsonl/src/index.ts index 8c7b92749a..bbce17b7db 100644 --- a/packages/session-persistence/session-persistence-jsonl/src/index.ts +++ b/packages/session-persistence/session-persistence-jsonl/src/index.ts @@ -15,7 +15,7 @@ import { randomBytes } from 'node:crypto' import { DEFAULT_PREPARED_SESSION_CACHE_SIZE, SessionPersistence, SessionPersistenceRevision, PersistenceCoordinator, type PersistenceBackend, type SessionLocation, type SessionPersistenceSnapshot, - type SessionInspection, type StoredPrefix, + type SessionInspection, type SessionPersistenceRevision as PersistenceRevision, type StoredPrefix, } from '@deepseek-ai/dsh-session-persistence' import type { SessionEvent, SessionId, SessionHeader, SessionPreparation } from '@deepseek-ai/dsh-session' import { @@ -66,6 +66,25 @@ interface JsonlTornMarker { recoveredEvents: SessionEvent[] } +interface FileRevisionIdentity { + readonly dev: bigint + readonly ino: bigint + readonly size: bigint + readonly mtimeNs: bigint + readonly ctimeNs: bigint +} + +/** Build the source-qualified revision shared by full and lightweight reads. */ +function fileRevision(identity: FileRevisionIdentity): PersistenceRevision { + return SessionPersistenceRevision([ + identity.dev, + identity.ino, + identity.size, + identity.mtimeNs, + identity.ctimeNs, + ].join(':')) +} + /** Whether a filesystem error means absence; every non-ENOENT failure must surface. */ function isENOENT(error: unknown): boolean { return (error as NodeJS.ErrnoException | null)?.code === 'ENOENT' @@ -167,6 +186,24 @@ 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. */ + async readStoredRevision(id: SessionId, signal?: AbortSignal): Promise { + signal?.throwIfAborted() + await this.ensureRootEncoding() + signal?.throwIfAborted() + const path = await this.findLog(id, signal) + if (path === undefined) return undefined + try { + const identity = await stat(path, { bigint: true }) + signal?.throwIfAborted() + return fileRevision(identity) + } catch (error: unknown) { + signal?.throwIfAborted() + if (isENOENT(error)) return undefined + throw error + } + } + /** * Read a stored prefix and convert torn-tail state to the opaque marker the * coordinator can round-trip without knowing the physical encoding. @@ -176,9 +213,20 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi expectedId?: SessionId, signal?: AbortSignal, ): Promise> { - const buffer = await readFile(path, { signal }) - signal?.throwIfAborted() - let prefix: StoredPrefix + let buffer: Buffer + let revision: PersistenceRevision + for (;;) { + signal?.throwIfAborted() + const before = fileRevision(await stat(path, { bigint: true })) + buffer = await readFile(path, { signal }) + signal?.throwIfAborted() + const after = fileRevision(await stat(path, { bigint: true })) + if (before === after) { + revision = after + break + } + } + let prefix: Omit, 'revision'> if (this.compression === 'zstd') { prefix = await this.readZstdPrefix(buffer, signal) } else { @@ -196,14 +244,14 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi signal?.throwIfAborted() await this.assertStoredIdentity(path, prefix.meta, expectedId, signal) signal?.throwIfAborted() - return prefix + return { ...prefix, revision } } /** Decode complete frames and retain complete JSONL records from a torn final frame. */ private async readZstdPrefix( buffer: Buffer, signal?: AbortSignal, - ): Promise> { + ): Promise, 'revision'>> { signal?.throwIfAborted() const { frames, tornStart } = scanZstdFrames(buffer) signal?.throwIfAborted() @@ -307,13 +355,7 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi signal?.throwIfAborted() snapshots.push({ header: artifact.header, - revision: SessionPersistenceRevision([ - identity.dev, - identity.ino, - identity.size, - identity.mtimeNs, - identity.ctimeNs, - ].join(':')), + revision: fileRevision(identity), }) } catch (error: unknown) { signal?.throwIfAborted() diff --git a/packages/session-persistence/session-persistence-jsonl/tests/jsonl.spec.ts b/packages/session-persistence/session-persistence-jsonl/tests/jsonl.spec.ts index cda6d577d8..3cc11da3b3 100644 --- a/packages/session-persistence/session-persistence-jsonl/tests/jsonl.spec.ts +++ b/packages/session-persistence/session-persistence-jsonl/tests/jsonl.spec.ts @@ -255,6 +255,17 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => { await otherCtx.fiber.dispose() }) + it('binds a full stored prefix to the same revision as a lightweight read', async () => { + const m = meta('stored-prefix-revision') + await ctx.sessionPersistence.create(m) + await ctx.sessionPersistence.append(m.id, oneTurnLog()) + const persistence = ctx.sessionPersistence as SessionPersistenceJsonl + + const stored = await persistence.loadStored(m.id) + expect(stored?.revision).toBe(await persistence.readStoredRevision(m.id)) + expect(await persistence.readStoredRevision(SessionId('missing-revision'))).toBeUndefined() + }) + it('omits a snapshot artifact removed after discovery', async () => { const m = meta('vanishing-snapshot') await ctx.sessionPersistence.create(m) diff --git a/packages/session-persistence/session-persistence-sqlite/README.i18n.yaml b/packages/session-persistence/session-persistence-sqlite/README.i18n.yaml index 2be033ceee..3dbbf87a2d 100644 --- a/packages/session-persistence/session-persistence-sqlite/README.i18n.yaml +++ b/packages/session-persistence/session-persistence-sqlite/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-sqlite/README.md -README.md: de70eb80559611f0409e05d3b2a6f69777a781a6 -README.zh.md: 53df3ef835ebc01b4a1ef7109190c4b64863a5ba +README.md: d01ba6ebfa1f59a9e4d58f3032bbe1d016970290 +README.zh.md: c11bef5467a3401948b58d5fd8e3301b72ff3d03 diff --git a/packages/session-persistence/session-persistence-sqlite/README.md b/packages/session-persistence/session-persistence-sqlite/README.md index de70eb8055..d01ba6ebfa 100644 --- a/packages/session-persistence/session-persistence-sqlite/README.md +++ b/packages/session-persistence/session-persistence-sqlite/README.md @@ -22,7 +22,7 @@ On filesystems with POSIX modes, the backend requests mode `0700` for missing di - **Lazy materialization.** `create()` records intent in memory only — no row is written until the first `append`. A created-but-never-appended session has no `sessions` row, so it is absent from `list()` (which reports exactly the sessions that have a row). - **Interrupted-turn close on load.** `load()` implements the shared [crash-recovery contract](../../../.agents/notes/implemented/architecture/2026-06-14-session-persistence.md): preserve the valid interrupted turn, append its synthetic closing events in one transaction, and remove only a torn tail row. Committed parse errors or sequence gaps make the session unloadable. Because recovery mutates stored rows, the next append starts from a balanced log and accurate cursor. - **Non-mutating inspection.** `inspect()` returns an immutable balanced logical view and may synthesize recovery closers in memory, without deleting a torn tail row, appending recovery rows, or changing the lightweight revision. -- **Lightweight revisions.** `listSnapshots(signal?)` combines the immutable store and database-file identity, a per-materialization incarnation id, and a per-session counter incremented in each mutating transaction. This keeps unchanged observations stable without parsing event rows and distinguishes independent stores and recreated same-id logs. It checks cancellation before and after shared readiness and the synchronous metadata query; the query itself is non-preemptible. +- **Lightweight revisions.** `listSnapshots(signal?)` combines the immutable store and database-file identity, a per-materialization incarnation id, and a per-session counter incremented in each mutating transaction. A full-prefix read captures that revision and its event rows in one read transaction, while `readStoredRevision()` queries only the session row to validate retained preparations. This keeps unchanged observations stable without parsing event rows and distinguishes independent stores and recreated same-id logs. It checks cancellation before and after shared readiness and the synchronous metadata query; the query itself is non-preemptible. ## Configuration (schemastery) diff --git a/packages/session-persistence/session-persistence-sqlite/README.zh.md b/packages/session-persistence/session-persistence-sqlite/README.zh.md index 53df3ef835..c11bef5467 100644 --- a/packages/session-persistence/session-persistence-sqlite/README.zh.md +++ b/packages/session-persistence/session-persistence-sqlite/README.zh.md @@ -22,7 +22,7 @@ SQLite 持久会话存储后端：第二个 `SessionPersistence` 实现（见[ - **延迟实体化。**`create()` 只在内存记录意图，第一次 `append` 前不写行。从未 append 的会话没有 `sessions` 行，因此不在 `list()` 中（它精确报告有行的会话）。 - **在 load 时关闭中断轮次。**`load()` 实现共享[崩溃恢复契约](../../../.agents/notes/implemented/architecture/2026-06-14-session-persistence.md)：保留有效中断轮次，在一个事务中追加合成关闭事件，并只移除撕裂尾部行。已提交解析错误或序列缺口使会话无法加载。恢复会变更已存储行，因此下一次 append 从平衡日志和准确游标开始。 - **非变更检查。**`inspect()` 返回不可变、平衡的逻辑视图，并可在内存中合成恢复 closer，但不会删除撕裂尾部行、追加恢复行或更改轻量修订。 -- **轻量修订。**`listSnapshots(signal?)` 组合不可变存储与数据库文件身份、每实体化 incarnation id，以及在每个变更事务中递增的每会话计数器。它在不解析事件行的情况下保持未变观察稳定，并区分独立存储和重建的同 id 日志。它在共享就绪和同步元数据查询前后检查取消；查询本身不可抢占。 +- **轻量修订。**`listSnapshots(signal?)` 组合不可变存储与数据库文件身份、每实体化 incarnation id，以及在每个变更事务中递增的每会话计数器。完整前缀读取在同一个读事务中捕获该 revision 及其事件行，`readStoredRevision()` 则只查询 session 行来校验保留的 preparation。它在不解析事件行的情况下保持未变观察稳定，并区分独立存储和重建的同 id 日志。它在共享就绪和同步元数据查询前后检查取消；查询本身不可抢占。 ## 配置（schemastery） diff --git a/packages/session-persistence/session-persistence-sqlite/src/index.ts b/packages/session-persistence/session-persistence-sqlite/src/index.ts index 98a747a73b..173d1bf857 100644 --- a/packages/session-persistence/session-persistence-sqlite/src/index.ts +++ b/packages/session-persistence/session-persistence-sqlite/src/index.ts @@ -16,7 +16,8 @@ import { dirname, resolve } from 'node:path' import { DEFAULT_PREPARED_SESSION_CACHE_SIZE, SessionPersistence, SessionPersistenceRevision, PersistenceCoordinator, type PersistenceBackend, type SessionLocation, type SessionPersistenceSnapshot, - type SessionInspection, type StoredPrefix, type StoredSuffix, + type SessionInspection, type SessionPersistenceRevision as PersistenceRevision, + type StoredPrefix, type StoredSuffix, } from '@deepseek-ai/dsh-session-persistence' import type { SessionEvent, SurfaceEventType, SessionId, SessionHeader, SessionPreparation } from '@deepseek-ai/dsh-session' import { @@ -38,6 +39,13 @@ function surfaceBindings(event: SessionEvent): [string | null, string | null] { ] } +/** Build the source-qualified revision shared by full and lightweight reads. */ +function sqliteRevision(storeIdentity: string, row: SessionRow): PersistenceRevision { + return SessionPersistenceRevision( + `${storeIdentity}:incarnation:${row.incarnation}:revision:${row.revision}`, + ) +} + /** * Exclusively create a missing database file with owner-only permissions. * Existing files retain their modes, and errors other than `EEXIST` propagate. @@ -187,6 +195,15 @@ export class SessionPersistenceSqlite extends SessionPersistence implements Pers return this.readPrefix(id, signal) } + /** Read one row's revision without loading its events. */ + async readStoredRevision(id: SessionId, signal?: AbortSignal): Promise { + signal?.throwIfAborted() + await this.ready + signal?.throwIfAborted() + const row = this.rowFor(id) + return row === undefined ? undefined : sqliteRevision(this.storeIdentity, row) + } + /** * Seek-capable suffix read: SQL selects `seq >= fromSeq` directly, so the * read scales with the suffix, not the log. Torn rows past the preserved @@ -216,15 +233,33 @@ export class SessionPersistenceSqlite extends SessionPersistence implements Pers signal?.throwIfAborted() await this.ready signal?.throwIfAborted() - const row = this.rowFor(id) - if (row === undefined) return undefined - const meta = rowToMeta(row) - const eventRows = this.db - .prepare('SELECT seq, type, time, data, source_event_seqs, surface_op FROM events WHERE session_id = ? ORDER BY seq') - .all(id) as unknown as EventRow[] + this.db.exec('BEGIN') + let snapshot: { row: SessionRow; eventRows: EventRow[] } | undefined + try { + const row = this.rowFor(id) + if (row !== undefined) { + const eventRows = this.db + .prepare('SELECT seq, type, time, data, source_event_seqs, surface_op FROM events WHERE session_id = ? ORDER BY seq') + .all(id) as unknown as EventRow[] + snapshot = { row, eventRows } + } + this.db.exec('COMMIT') + } catch (error: unknown) { + /* v8 ignore start -- synchronous read failures only need transaction cleanup before propagation. */ + this.db.exec('ROLLBACK') + throw error + /* v8 ignore stop */ + } signal?.throwIfAborted() + if (snapshot === undefined) return undefined + const { row, eventRows } = snapshot const { preserved, tornFrom } = scanRows(eventRows) - return { meta, events: preserved, ...tornFrom !== undefined ? { tornMarker: tornFrom } : {} } + return { + meta: rowToMeta(row), + events: preserved, + revision: sqliteRevision(this.storeIdentity, row), + ...tornFrom !== undefined ? { tornMarker: tornFrom } : {}, + } } /** diff --git a/packages/session-persistence/session-persistence-sqlite/tests/sqlite.spec.ts b/packages/session-persistence/session-persistence-sqlite/tests/sqlite.spec.ts index ca8c1cd76a..50e3a593e0 100644 --- a/packages/session-persistence/session-persistence-sqlite/tests/sqlite.spec.ts +++ b/packages/session-persistence/session-persistence-sqlite/tests/sqlite.spec.ts @@ -576,6 +576,19 @@ describe('SessionPersistenceSqlite: durability and crash semantics', () => { await b.dispose() }) + it('binds a full stored prefix to the same revision as a lightweight read', async () => { + const b = await backend() + const m = meta('stored-prefix-revision') + await b.ctx.sessionPersistence.create(m) + await b.ctx.sessionPersistence.append(m.id, oneTurnLog()) + const persistence = b.ctx.sessionPersistence as SessionPersistenceSqlite + + const stored = await persistence.loadStored(m.id) + expect(stored?.revision).toBe(await persistence.readStoredRevision(m.id)) + expect(await persistence.readStoredRevision(SessionId('missing-revision'))).toBeUndefined() + await b.dispose() + }) + it('changes revisions when a deleted session id is materialized again in the same database', async () => { const path = await freshDbPath() const m = meta('recreated-revision') diff --git a/packages/session-persistence/session-persistence/README.i18n.yaml b/packages/session-persistence/session-persistence/README.i18n.yaml index e94106d89c..a4e98264f3 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: 675f59cebe338c7bbf9bed4436db26b4593ee473 -README.zh.md: ff6589c0f29e1a634b27a4118878dc7b503c9e98 +README.md: 327a61d1b8bf10239eab0ca3cb18ac865fc2dcce +README.zh.md: 2cc0b94bfd46a9e75c7a3626554f2ab9e5949600 diff --git a/packages/session-persistence/session-persistence/README.md b/packages/session-persistence/session-persistence/README.md index 675f59cebe..327a61d1b8 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 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. | +| `inspect(id, signal?): Promise<{ meta; events }>` | Return an upgraded, validated, deeply frozen logical view without committing recovery or publishing a Session. A cold view receives in-memory synthetic recovery closers while its physical torn tail remains untouched; an already-live view is its current immutable snapshot and may contain an open turn. Coordinator-backed implementations retain the exact cold unpublished Session in a bounded LRU for later `prepare`, but discard and reload it when the stored revision changes. Same-id inspections share an in-flight read. | | `readFrom(id, fromSeq, signal?): Promise<{ meta; events }>` | 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. | @@ -33,27 +33,28 @@ The persisted unit IS the existing `SessionEvent` (event-sourced model — the l Each `session/event` copies its event into the session controller and starts an eager drain without blocking the producer. Concurrent notifications share the current drain; events admitted during a write remain pending and trigger the next batch. `session/flush` is an observation barrier that waits until the controller has no current or pending batch. An eager failure is logged and retains the batch; the next explicit flush or backend teardown retries it and surfaces failure to its caller. -Crash repair is cold-only. For a live id, `load(id)` snapshots the authoritative in-memory log, waits for that snapshot to become durable, and returns it only when balanced; an open live turn rejects instead of receiving synthetic interruption closers. For a cold id, inspection reads, validates, freezes, and constructs one unpublished Session; repeated inspection reuses that object graph. `prepare(id)` reserves the exact Session, commits any pending torn-tail/interrupted-turn repair, and returns it for publication. HMR adoption reads through `loadStored`, applies the coordinator's cwd check, and never closes the active turn. +Crash repair is cold-only. For a live id, `load(id)` snapshots the authoritative in-memory log, waits for that snapshot to become durable, and returns it only when balanced; an open live turn rejects instead of receiving synthetic interruption closers. For a cold id, inspection reads, validates, freezes, and constructs one unpublished Session; repeated inspection reuses that object graph only while its source revision remains current. `prepare(id)` performs the same check before repair, reserves the exact Session, commits any pending torn-tail/interrupted-turn repair, and returns it for publication. HMR adoption reads through `loadStored`, applies the coordinator's cwd check, and never closes the active turn. Backend reads normalize the exact supported same-version shapes before current-shape validation. Pre-identity messages receive the deterministic id `legacy-message::`; a tool-result content replacement inherits its target's imported id. A pre-react-loop `turn/start` loses its obsolete trigger, a removed `steering/message` becomes the same identified `user/message`, and an older `turn/end` maps its terminal reason without inventing unavailable cancellation provenance. The coordinator uses the same normalized view for `load`, `inspect`, `readFrom`, ownerless-state claims, and HMR prefix adoption. Storage remains append-only: reads do not rewrite old records, and later appends use the current shape. These are narrow import exceptions from the [pre-identity message](../../../.agents/notes/implemented/bug-fix/2026-07-28-load-pre-identity-session-messages.md) and [pre-react-loop session](../../../.agents/notes/implemented/bug-fix/2026-08-04-load-pre-react-loop-sessions.md) decisions, not a general v0 migration promise. When a live session emits `session/disposed`, the coordinator waits for its controller, serializes a final drain, then releases state owned by that exact `Session` object. Failed retirement leaves the controller in the live-session map, so backend teardown can retry it. Backend teardown stops event admission first, flushes every remaining controller, awaits per-id operations, and only then closes the storage handle. -The side-effect-free `locate` and lightweight `listSnapshots` queries remain backend-owned because they describe storage topology and revision identity rather than write orchestration. `listSnapshots(signal?)` passes the caller's exact signal into backend discovery so observers can cancel that work without detaching it. +The side-effect-free `locate`, lightweight `listSnapshots`, and per-id `readStoredRevision` queries remain backend-owned because they describe storage topology and revision identity rather than write orchestration. `listSnapshots(signal?)` passes the caller's exact signal into backend discovery so observers can cancel that work without detaching it. The `PersistenceBackend` hooks (the only seam between the coordinator and storage): | Hook | Role | |---|---| | `name` | Backend label for the dispose-failure `AggregateError`. | -| `loadStored(id, signal?)` | Read a stored prefix by id across every storage scope. Used by resume/load, non-mutating inspect, live adoption, and the create-collision probe. The optional signal belongs to observation-only reads. Returned metadata identifies `id`; an opaque `tornMarker` is present iff a torn tail must be truncated. | +| `loadStored(id, signal?)` | Read a stored prefix by id across every storage scope. Used by resume/load, non-mutating inspect, live adoption, and the create-collision probe. The optional signal belongs to observation-only reads. Returned metadata identifies `id`; `revision` identifies exactly the returned header and events; an opaque `tornMarker` is present iff a torn tail must be truncated. | +| `readStoredRevision(id, signal?)` | Read the current source-qualified revision for one id without loading its event log. It uses the same revision representation as `loadStored` and returns `undefined` when the id is absent. | | `loadStoredFrom?(id, fromSeq, signal?)` | Optional seek-capable suffix read behind the service's `readFrom`: the header plus stored events with `seq >= fromSeq`, non-mutating, no torn marker. SQLite implements it (`WHERE seq >= ?`); a backend that omits it gets the coordinator's fallback — `loadStored` plus a forward skip. | | `appendBatch(meta, events, isMaterialized)` | Durably append a contiguous batch, lazily materializing ATOMICALLY when not yet materialized. | | `commitRepair(meta, tornMarker, closers)` | Make a crash repair durable: truncate the torn tail (iff `tornMarker !== undefined` — a marker may be falsy, e.g. seq/offset `0`) and append `closers`. NOT required to be atomic. Used by load (truncate + closers) and live-adoption (truncate only). | | `list(signal?)` | List all stored metadata, observing optional cancellation. | | `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`. 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. 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 ff6589c0f2..2cc0b94bfd 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，物理撕裂尾部保持不变；已经实时存在的视图则是当前不可变快照，可能包含打开的 turn。基于协调器的实现会在有界 LRU 中保留精确的冷未发布 Session，供后续 `prepare` 使用。同 id 检查共享进行中的读取。 | +| `inspect(id, signal?): Promise<{ meta; events }>` | 返回已经升级、验证和深度冻结的逻辑视图，但不提交恢复或发布 Session。冷视图会获得仅存在于内存的合成恢复 closer，物理撕裂尾部保持不变；已经实时存在的视图则是当前不可变快照，可能包含打开的 turn。基于协调器的实现会在有界 LRU 中保留精确的冷未发布 Session，供后续 `prepare` 使用，但已存储 revision 变化后会丢弃并重新读取。同 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 修复后会改变；不会仅因两个存储使用相同本地计数器而冲突。可选信号请求取消后端发现工作；第一方后端会先等待所有已启动的列出工作结束，再予以拒绝，因此调用返回拒绝时，相关工作已完全停稳。 | @@ -33,27 +33,28 @@ 每个 `session/event` 将事件复制到会话 controller，并在不阻塞生产者的情况下立即启动 drain。并发通知共享当前 drain；写入期间接纳的事件保持 pending，并触发下一批。`session/flush` 是观察屏障，会等待 controller 无当前或 pending 批次。即时写入失败会记录日志并保留批次；下一次显式 flush 或后端拆卸会重试该批次，并将失败返回给调用方。 -崩溃修复只适用于冷状态。对于实时 id，`load(id)` 为权威内存日志制作快照，等待该快照持久，并只在平衡时返回；开放实时轮次会被拒绝，而不会收到合成中断 closer。对于冷 id，检查只读取、验证、冻结并构造一次未发布 Session；重复检查会复用该对象图。`prepare(id)` 预留精确 Session，提交任何待处理的撕裂尾部或中断轮次修复，并将其返回用于发布。HMR 接管通过 `loadStored` 读取，应用协调器 cwd 检查，并绝不关闭活动轮次。 +崩溃修复只适用于冷状态。对于实时 id，`load(id)` 为权威内存日志制作快照，等待该快照持久，并只在平衡时返回；开放实时轮次会被拒绝，而不会收到合成中断 closer。对于冷 id，检查只读取、验证、冻结并构造一次未发布 Session；只有来源 revision 仍然是当前值时，重复检查才会复用该对象图。`prepare(id)` 在修复前执行相同校验，预留精确 Session，提交任何待处理的撕裂尾部或中断轮次修复，并将其返回用于发布。HMR 接管通过 `loadStored` 读取，应用协调器 cwd 检查，并绝不关闭活动轮次。 后端读取会在当前形状验证前，规范化明确受支持的同版本形状。消息标识机制引入前的消息会获得确定性的 id `legacy-message::`；工具结果的内容替换会继承其目标导入后的 id。react-loop 重构前的 `turn/start` 会移除过时的 trigger，已移除的 steering（中途引导）事件 `steering/message` 会转换为同一条带标识的 `user/message`；旧版 `turn/end` 会在不虚构无法获得的取消来源的前提下映射终止原因。协调器对 `load`、`inspect`、`readFrom`、无 owner 状态的认领和 HMR 前缀接管使用同一份规范化视图。存储仍然仅追加：读取不会重写旧记录，此后追加的事件使用当前形状。这些是[消息标识机制引入前的消息](../../../.agents/notes/implemented/bug-fix/2026-07-28-load-pre-identity-session-messages.md)与 [react-loop 重构前会话](../../../.agents/notes/implemented/bug-fix/2026-08-04-load-pre-react-loop-sessions.md)决策所规定的范围受限的导入例外，并不构成通用的 v0 迁移承诺。 实时会话发出 `session/disposed` 时，协调器等待其 controller，串行化最终 drain，然后释放该精确 `Session` 对象拥有的状态。失败退役会将 controller 保留在实时会话 map 中，使后端拆卸可重试。后端拆卸先停止事件接纳，flush 每个剩余 controller，等待每 id 操作，最后才关闭存储句柄。 -无副作用 `locate` 和轻量 `listSnapshots` 查询仍由后端负责，因为它们描述存储拓扑和修订身份，而非写入编排。`listSnapshots(signal?)` 将调用方传入的同一个信号传给后端发现流程，使观察者可在不脱离该工作的情况下取消。 +无副作用 `locate`、轻量 `listSnapshots` 和按 id 查询的 `readStoredRevision` 仍由后端负责，因为它们描述存储拓扑和 revision 身份，而非写入编排。`listSnapshots(signal?)` 将调用方传入的同一个信号传给后端发现流程，使观察者可在不脱离该工作的情况下取消。 `PersistenceBackend` 钩子（协调器与存储之间的唯一 seam）： | 钩子 | 职责 | |---|---| | `name` | dispose 失败 `AggregateError` 的后端标签。 | -| `loadStored(id, signal?)` | 在全部存储范围中按 id 读取已存储前缀。用于恢复／加载、非修改式 inspect、实时接管和 create 冲突探测。可选信号属于仅观察读取。返回元数据标识 `id`；当且仅当必须截断撕裂尾部时才存在不透明 `tornMarker`。 | +| `loadStored(id, signal?)` | 在全部存储范围中按 id 读取已存储前缀。用于恢复／加载、非修改式 inspect、实时接管和 create 冲突探测。可选信号属于仅观察读取。返回元数据标识 `id`；`revision` 精确标识返回的 header 和事件；当且仅当必须截断撕裂尾部时才存在不透明 `tornMarker`。 | +| `readStoredRevision(id, signal?)` | 在不加载事件日志的情况下读取一个 id 当前的来源限定 revision。它使用与 `loadStored` 相同的 revision 表示；id 不存在时返回 `undefined`。 | | `loadStoredFrom?(id, fromSeq, signal?)` | 服务 `readFrom` 背后的可选可寻址后缀读取：返回 header 和 `seq >= fromSeq` 的已存储事件，非变更、无撕裂标记。SQLite 实现它（`WHERE seq >= ?`）；不实现的后端使用协调器回退——`loadStored` 加向前跳过。 | | `appendBatch(meta, events, isMaterialized)` | 持久追加连续批次；尚未实体化时以原子方式延迟实体化。 | | `commitRepair(meta, tornMarker, closers)` | 使崩溃修复持久：截断撕裂尾部（当且仅当 `tornMarker !== undefined`；标记可为 falsy，例如 seq/offset `0`），并追加 `closers`。不要求原子性。由 load（截断 + closer）和实时接管（仅截断）使用。 | | `list(signal?)` | 列出全部已存储元数据，并遵循可选的取消信号。 | | `close?()` | 可选生命周期拆卸（例如关闭 db 句柄），在 dispose drain 后等待。 | -协调器断言已存储 id，并在修复或实时接管前比较已存储/实时 cwd。其 `inspect()` 路径取得新鲜后端值的所有权，只验证和冻结一次，并在不调用 `commitRepair` 的情况下最多保留配置数量的未发布 Session。`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` 时，系统才会复用或修复它；否则协调器会重新读取。该新鲜性校验不会增加跨进程写入排他。`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 711ed2ea48..3baa7dcce7 100644 --- a/packages/session-persistence/session-persistence/src/coordinator.ts +++ b/packages/session-persistence/session-persistence/src/coordinator.ts @@ -16,6 +16,7 @@ import { } from '@deepseek-ai/dsh-session' import type { Session, SessionEvent, SessionId, SessionHeader } from '@deepseek-ai/dsh-session' import type { SessionInspection } from './index.ts' +import type { SessionPersistenceRevision } from './revision.ts' import { observeQueuedAbort, SessionPreparations } from './preparations.ts' import type { SessionPreparationReservation } from './preparations.ts' @@ -41,13 +42,17 @@ export interface PersistenceCoordinatorOptions { } /** - * A stored session's header, valid contiguous event prefix, and optional opaque - * torn-tail marker. The coordinator only checks marker presence and returns its - * value to {@link PersistenceBackend.commitRepair}; each backend owns the type. + * A stored session's header, valid contiguous event prefix, source-qualified + * revision, and optional opaque torn-tail marker. The revision identifies the + * exact detached prefix. The coordinator only checks marker presence and + * returns its value to {@link PersistenceBackend.commitRepair}; each backend + * owns the marker type. */ export interface StoredPrefix { meta: SessionHeader events: SessionEvent[] + /** Revision observed for exactly this detached prefix. */ + revision: SessionPersistenceRevision tornMarker?: TornMarker } @@ -83,12 +88,22 @@ export interface PersistenceBackend { * and — via `!== undefined` — the create-collision probe. The returned * `tornMarker` is present iff there is a torn tail to truncate. Every header * and event graph must be fresh, mutually unaliased, and unretained by the - * backend because preparation freezes and publishes them in place. + * backend because preparation freezes and publishes them in place. The + * returned revision must identify exactly those values and use the same + * representation as {@link readStoredRevision}. * @param id - persisted session id to resolve. * @param signal - optional cancellation for backend read work. */ loadStored(id: SessionId, signal?: AbortSignal): Promise | undefined> + /** + * Read the current source-qualified revision for one stored session without + * loading its event log. Returns `undefined` when the identity is absent. + * @param id - persisted session id to observe. + * @param signal - optional cancellation for backend read work. + */ + readStoredRevision(id: SessionId, signal?: AbortSignal): Promise + /** * Optional seek-capable suffix read behind the service's `readFrom`: return * the header plus the stored events with `seq >= fromSeq` without reading @@ -170,6 +185,7 @@ interface LiveSessionState { interface PreparedSessionSource { readonly inspection: SessionInspection readonly session: Session + readonly revision: SessionPersistenceRevision /** Session length after constructor-owned seed markers were appended. */ readonly sessionLength: number readonly tornMarker: TornMarker | undefined @@ -689,22 +705,33 @@ export class PersistenceCoordinator { * @returns immutable prepared metadata and events; a live view may have an open turn. */ async inspect(id: SessionId, signal?: AbortSignal): Promise { - 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, - ) - const attached = this.ctx.sessions.get(id) - return attached === undefined ? source.inspection : this.inspectLive(attached) - } catch (error: unknown) { - const attached = this.ctx.sessions.get(id) - if (attached !== undefined) return this.inspectLive(attached) - throw error + for (;;) { + 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, + ) + const attached = this.ctx.sessions.get(id) + if (attached !== undefined) return this.inspectLive(attached) + const current = await this.serialize( + id, + () => this.isPreparedSourceCurrent(source, signal), + signal, + ) + const published = this.ctx.sessions.get(id) + if (published !== undefined) return this.inspectLive(published) + if (current) return source.inspection + this.preparations.invalidate(id, source) + } catch (error: unknown) { + const attached = this.ctx.sessions.get(id) + if (attached !== undefined) return this.inspectLive(attached) + throw error + } } } @@ -790,7 +817,7 @@ export class PersistenceCoordinator { signal?.throwIfAborted() if (stored === undefined) throw new Error(`session "${id}" not found`) try { - const { meta, events, tornMarker } = stored + const { meta, events, revision, tornMarker } = stored this.assertStoredId(id, meta) this.assertVersion(meta) const storedEvents = adoptStoredEvents(events, id) @@ -810,6 +837,7 @@ export class PersistenceCoordinator { return { inspection, session, + revision, sessionLength: session.events.length, tornMarker, closers, @@ -825,15 +853,22 @@ export class PersistenceCoordinator { /** Commit one prepared repair and establish its ownerless durable cursor. */ private async commitPrepared( source: PreparedSessionSource, - ): Promise<{ source: PreparedSessionSource; state: SessionState }> { + ): Promise<{ source: PreparedSessionSource; state: SessionState } | undefined> { const id = source.inspection.meta.id const cursor = source.inspection.events.length const existing = this.states.get(id) if (existing?.owner !== undefined) { throw new Error(`session "${id}" already has a live persistence owner`) } + if (!await this.isPreparedSourceCurrent(source)) return undefined + let committedSource = source if (source.tornMarker !== undefined || source.closers.length > 0) { await this.backend.commitRepair(source.inspection.meta, source.tornMarker, source.closers) + const revision = await this.backend.readStoredRevision(id) + if (revision === undefined) { + throw new Error(`session "${id}" disappeared after persistence repair`) + } + committedSource = { ...source, revision, tornMarker: undefined, closers: [] } } const state = existing ?? { meta: source.inspection.meta, @@ -845,13 +880,19 @@ export class PersistenceCoordinator { state.materialized = true this.states.set(id, state) return { - source: source.tornMarker === undefined && source.closers.length === 0 - ? source - : { ...source, tornMarker: undefined, closers: [] }, + source: committedSource, state, } } + /** Whether one cached source still names the current durable log revision. */ + private async isPreparedSourceCurrent( + source: PreparedSessionSource, + signal?: AbortSignal, + ): Promise { + return await this.backend.readStoredRevision(source.inspection.meta.id, signal) === source.revision + } + /** Return one durable immutable view of an already-live Session. */ private async loadLiveSnapshot(session: Session): Promise { const events = session.events @@ -918,8 +959,11 @@ export class PersistenceCoordinator { private async adopt(id: SessionId): Promise { // This runs inside the id's serialization chain, so it uses core helpers // instead of re-entering through public prepare/load methods. - const source = this.preparations.takeReady(id) ?? await this.prepareCore(id) - return (await this.commitPrepared(source)).state + for (;;) { + const source = this.preparations.takeReady(id) ?? await this.prepareCore(id) + const committed = await this.commitPrepared(source) + if (committed !== undefined) return committed.state + } } private assertVersion(meta: SessionHeader): void { diff --git a/packages/session-persistence/session-persistence/src/index.ts b/packages/session-persistence/session-persistence/src/index.ts index a30a636a89..4ee46600cd 100644 --- a/packages/session-persistence/session-persistence/src/index.ts +++ b/packages/session-persistence/session-persistence/src/index.ts @@ -102,8 +102,9 @@ 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}; disposal - * releases an unpublished reservation. + * reuse object graphs retained by an earlier {@link inspect} after confirming + * their durable revision is still current; disposal releases an unpublished + * reservation. * @param id - persisted session to prepare. * @param signal - optional cancellation for preparation work. * @returns one owned unpublished Session preparation. @@ -144,8 +145,8 @@ export abstract class SessionPersistence extends Service { * 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. + * reuse by a later {@link prepare}, reloading it when its durable revision + * changes; 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 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 6cca26af5d..a4a88b3ef8 100644 --- a/packages/session-persistence/session-persistence/src/preparations.ts +++ b/packages/session-persistence/session-persistence/src/preparations.ts @@ -75,7 +75,7 @@ export class SessionPreparations { async reserve( id: SessionId, load: () => Promise, - commit: (source: Source) => Promise<{ source: Source; state: CommitState }>, + commit: (source: Source) => Promise<{ source: Source; state: CommitState } | undefined>, signal?: AbortSignal, ): Promise | undefined> { const entry = this.entryFor(id, load) @@ -93,13 +93,17 @@ export class SessionPreparations { entry.phase = 'committing' entry.reservationSettled = reservationSettled.promise entry.settleReservation = reservationSettled.resolve - let committed: { source: Source; state: CommitState } + let committed: { source: Source; state: CommitState } | undefined try { committed = await commit(source) } catch (error: unknown) { this.remove(entry) throw error } + if (committed === undefined) { + this.remove(entry) + return undefined + } entry.source = committed.source try { signal?.throwIfAborted() @@ -179,10 +183,11 @@ export class SessionPreparations { /** * Discard a prepared view after the durable log changes. * @param id - changed session identity. + * @param expected - when supplied, invalidate only that exact source. */ - invalidate(id: SessionId): void { + invalidate(id: SessionId, expected?: Source): void { const entry = this.entries.get(id) - if (entry !== undefined) this.remove(entry) + if (entry !== undefined && (expected === undefined || entry.source === expected)) this.remove(entry) } /** diff --git a/packages/session-persistence/session-persistence/tests/persistence.spec.ts b/packages/session-persistence/session-persistence/tests/persistence.spec.ts index 93d83499bd..d8bff0bcc4 100644 --- a/packages/session-persistence/session-persistence/tests/persistence.spec.ts +++ b/packages/session-persistence/session-persistence/tests/persistence.spec.ts @@ -4,7 +4,7 @@ import SessionStore, { Session, SessionId, isJsonValue } from '@deepseek-ai/dsh- import type { SessionEvent, SessionHeader } from '@deepseek-ai/dsh-session' import { SessionPersistence, SessionPersistenceRevision, PersistenceCoordinator, - type PersistenceBackend, type SessionPersistenceSnapshot, type StoredPrefix, + type PersistenceBackend, type SessionPersistenceSnapshot, type StoredPrefix, type StoredSuffix, } from '../src/index.ts' import { runPersistenceContract, meta, oneTurnLog } from './contract.ts' import { runCoordinatorContract, type CoordinatorFixture } from './coordinator-contract.ts' @@ -12,6 +12,11 @@ import { runCoordinatorContract, type CoordinatorFixture } from './coordinator-c /** The durable store shape: materialized sessions only (no lazy entries). */ type MemoryStore = Map +/** Test-store revision that changes for any metadata or event mutation. */ +function memoryRevision(entry: { meta: SessionHeader; events: SessionEvent[] }): SessionPersistenceRevision { + return SessionPersistenceRevision(JSON.stringify(entry)) +} + /** An obsolete event fixture that emulates an untyped pre-change producer. */ function legacyHeaderDelta(seq = 0): SessionEvent { return { @@ -114,7 +119,16 @@ class MemoryPersistence extends SessionPersistence implements PersistenceBackend async loadStored(id: SessionId): Promise | undefined> { const entry = this.store.get(id) if (!entry) return undefined - return { meta: structuredClone(entry.meta), events: structuredClone(entry.events) } + return { + meta: structuredClone(entry.meta), + events: structuredClone(entry.events), + revision: memoryRevision(entry), + } + } + + async readStoredRevision(id: SessionId): Promise { + const entry = this.store.get(id) + return entry === undefined ? undefined : memoryRevision(entry) } async appendBatch(m: SessionHeader, events: readonly SessionEvent[], _isMaterialized: boolean): Promise { @@ -151,7 +165,7 @@ class MemoryPersistence extends SessionPersistence implements PersistenceBackend signal?.throwIfAborted() return [...this.store.values()].map(entry => ({ header: structuredClone(entry.meta), - revision: SessionPersistenceRevision(`events:${entry.events.length}`), + revision: memoryRevision(entry), })) } } @@ -167,9 +181,9 @@ class ControlledBackend implements PersistenceBackend { beforeAppend?: (attempt: number) => Promise beforeLoadStored?: (attempt: number, signal?: AbortSignal) => Promise /** When set, the declared seek hook delegates here so readFrom exercises it; unset throws (tests set it first). */ - seekHook?: (id: SessionId, fromSeq: number, signal?: AbortSignal) => Promise | undefined> + seekHook?: (id: SessionId, fromSeq: number, signal?: AbortSignal) => Promise - loadStoredFrom(id: SessionId, fromSeq: number, signal?: AbortSignal): Promise | undefined> { + loadStoredFrom(id: SessionId, fromSeq: number, signal?: AbortSignal): Promise { if (this.seekHook === undefined) throw new Error('seekHook not configured for this test') return this.seekHook(id, fromSeq, signal) } @@ -179,7 +193,17 @@ class ControlledBackend implements PersistenceBackend { await this.beforeLoadStored?.(attempt, signal) const entry = this.store.get(id) if (entry === undefined) return undefined - return { meta: structuredClone(entry.meta), events: structuredClone(entry.events) } + return { + meta: structuredClone(entry.meta), + events: structuredClone(entry.events), + revision: memoryRevision(entry), + } + } + + async readStoredRevision(id: SessionId, signal?: AbortSignal): Promise { + signal?.throwIfAborted() + const entry = this.store.get(id) + return entry === undefined ? undefined : memoryRevision(entry) } async appendBatch(m: SessionHeader, events: readonly SessionEvent[], _isMaterialized: boolean): Promise { @@ -590,6 +614,64 @@ describe('PersistenceCoordinator session preparations', () => { } }) + it('reloads a cached inspection after the durable revision changes', async () => { + const ctx = new Context() + await ctx.plugin(SessionStore) + const backend = new ControlledBackend() + const id = SessionId('inspect-revision-refresh') + 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 first = await coordinator.inspect(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' } } }, + ) + + const refreshed = await coordinator.inspect(id) + expect(refreshed.events).toHaveLength(8) + expect(refreshed.events[0]).not.toBe(first.events[0]) + expect(backend.loadAttempts).toBe(2) + } finally { + await fiber.dispose() + await ctx.fiber.dispose() + } + }) + + it('does not restore from a cached inspection after the durable revision changes', async () => { + const ctx = new Context() + await ctx.plugin(SessionStore) + const backend = new ControlledBackend() + const id = SessionId('prepare-revision-refresh') + 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 + + try { + const inspected = await coordinator.inspect(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' } } }, + ) + + preparation = await coordinator.prepare(id) + expect(preparation.session.events).toHaveLength(9) + expect(preparation.session.events[0]).not.toBe(inspected.events[0]) + expect(backend.loadAttempts).toBe(2) + } finally { + 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) @@ -610,9 +692,12 @@ describe('PersistenceCoordinator session preparations', () => { data: { turn: 2 }, }]) - await expect(inspection).resolves.toMatchObject({ meta: { id } }) + await expect(inspection).resolves.toMatchObject({ + meta: { id }, + events: [...oneTurnLog(), { seq: 6 }, { seq: 7 }], + }) await expect(append).resolves.toBeUndefined() - expect(backend.loadAttempts).toBe(1) + expect(backend.loadAttempts).toBe(2) expect(backend.store.get(id)?.events).toHaveLength(oneTurnLog().length + 1) } finally { await fiber.dispose() diff --git a/packages/session-persistence/session-persistence/tests/preparations.spec.ts b/packages/session-persistence/session-persistence/tests/preparations.spec.ts index 7002a6d0ae..cec0ac370b 100644 --- a/packages/session-persistence/session-persistence/tests/preparations.spec.ts +++ b/packages/session-persistence/session-persistence/tests/preparations.spec.ts @@ -34,6 +34,8 @@ describe('SessionPreparations inspection', () => { await expect(preparations.inspect(id, load)).resolves.toBe(source) expect(load).toHaveBeenCalledOnce() + preparations.invalidate(id, prepared('different-source')) + expect(preparations.has(id)).toBe(true) preparations.invalidate(id) preparations.invalidate(id) expect(preparations.has(id)).toBe(false)