From b10dad0fcbf36e18bbd8959ab7fb1d85065e00fd Mon Sep 17 00:00:00 2001 From: _Kerman Date: Tue, 28 Jul 2026 22:16:43 +0800 Subject: [PATCH] fix(persistence): restore pre-identity sessions --- .../2026-06-14-session-persistence.i18n.yaml | 6 +- .../2026-06-14-session-persistence.md | 2 +- .../2026-06-14-session-persistence.zh.md | 2 +- ...ad-pre-identity-session-messages.i18n.yaml | 6 + ...7-28-load-pre-identity-session-messages.md | 38 +++++ ...8-load-pre-identity-session-messages.zh.md | 38 +++++ docs/cordis-catalog/services.md | 9 +- .../cordis/tool-cordis/src/api-catalog.ts | 4 +- packages/core/agent-loop/tests/resume.spec.ts | 61 +++++++- packages/core/session/README.i18n.yaml | 4 +- packages/core/session/README.md | 2 +- packages/core/session/README.zh.md | 2 +- .../session-persistence/README.i18n.yaml | 4 +- .../session-persistence/README.md | 6 +- .../session-persistence/README.zh.md | 6 +- .../session-persistence/src/coordinator.ts | 146 +++++++++++++++++- .../session-persistence/src/index.ts | 9 +- .../tests/coordinator-contract.ts | 145 ++++++++++++++++- 18 files changed, 457 insertions(+), 33 deletions(-) create mode 100644 .agents/notes/implemented/bug-fix/2026-07-28-load-pre-identity-session-messages.i18n.yaml create mode 100644 .agents/notes/implemented/bug-fix/2026-07-28-load-pre-identity-session-messages.md create mode 100644 .agents/notes/implemented/bug-fix/2026-07-28-load-pre-identity-session-messages.zh.md diff --git a/.agents/notes/implemented/architecture/2026-06-14-session-persistence.i18n.yaml b/.agents/notes/implemented/architecture/2026-06-14-session-persistence.i18n.yaml index a29fa6e073..5b70e37ebe 100644 --- a/.agents/notes/implemented/architecture/2026-06-14-session-persistence.i18n.yaml +++ b/.agents/notes/implemented/architecture/2026-06-14-session-persistence.i18n.yaml @@ -1,6 +1,6 @@ # Bilingual-pair consistency record (docs/i18n/README.md): the git blob hash of each # 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 -2026-06-14-session-persistence.md: 75e13b860f621ed407849b3b4c62ff7287ab4812 -2026-06-14-session-persistence.zh.md: a6bd400a053779c742940236737447d1687622de +# pnpm run verify-translation-pairing --write .agents/notes/implemented/architecture/2026-06-14-session-persistence.md +2026-06-14-session-persistence.md: 137b2b01126214629952812f3dd3b71985a3acda +2026-06-14-session-persistence.zh.md: 2846ee92349c297fb3a173ba9dd3e2ff3cd9ee1a diff --git a/.agents/notes/implemented/architecture/2026-06-14-session-persistence.md b/.agents/notes/implemented/architecture/2026-06-14-session-persistence.md index 75e13b860f..137b2b0112 100644 --- a/.agents/notes/implemented/architecture/2026-06-14-session-persistence.md +++ b/.agents/notes/implemented/architecture/2026-06-14-session-persistence.md @@ -29,7 +29,7 @@ Key choices recorded here because they are durable, contested, and surprising: Each key choice above records its rejected alternative where the choice is stated: a **chunk-filtered canonical log** (Codex's `policy.rs` shape) — breaks the contiguous-seq contract; **truncating a crashed turn** — silently destroys a long autonomous run's real work; an **in-log `session/meta` event as line 0** — metadata is not replayable state; **finite fractional `createdAt` values** — have no producer and diverge from integer Unix-millisecond storage and query columns; **adopting a non-pristine unversioned SQLite file** — can overwrite unrelated objects or identity; **hard-injecting `sessionPersistence` into the loop** — would pend non-persistent demos forever. -Format versioning: the header carries a `version`; `load` rejects any non-current version (no migration — the pre-release session format is pinned at `SESSION_FORMAT_VERSION = 0` and absorbs shape churn, per the AGENTS.md pre-release stance). Stated honestly: append-only + flush is robust to partial trailing writes (tolerated on load) but not to fsync-less power loss mid-line; a DB/WAL backend is the stronger option later. +Format versioning: the header carries a `version`; `load` rejects any non-current version. The pre-release session format stays pinned at `SESSION_FORMAT_VERSION = 0` and carries no broad compatibility promise, while the coordinator may own an explicit narrow import upgrade when persisted user data requires it ([pre-identity message recovery](../bug-fix/2026-07-28-load-pre-identity-session-messages.md)). Stated honestly: append-only + flush is robust to partial trailing writes (tolerated on load) but not to fsync-less power loss mid-line; a DB/WAL backend is the stronger option later. ## Consequences diff --git a/.agents/notes/implemented/architecture/2026-06-14-session-persistence.zh.md b/.agents/notes/implemented/architecture/2026-06-14-session-persistence.zh.md index a6bd400a05..2846ee9234 100644 --- a/.agents/notes/implemented/architecture/2026-06-14-session-persistence.zh.md +++ b/.agents/notes/implemented/architecture/2026-06-14-session-persistence.zh.md @@ -29,7 +29,7 @@ Status: implemented 上述每个关键选择都在陈述处记录了被否决的替代方案：**过滤分片的规范日志**（Codex 的 `policy.rs` 形式）破坏连续 seq 契约；**截断崩溃的轮次**会静默销毁长时间自主运行中的真实工作；**日志内 `session/meta` 事件作为第 0 行**——元数据不是可回放状态；**有限的非整数 `createdAt` 值**没有生产方，且与整数 Unix 毫秒存储及查询列不一致；**接受非全新的未版本化 SQLite 文件**可能覆盖无关对象或应用标识；**将 `sessionPersistence` 硬注入循环**会让非持久化的演示永远挂起。 -格式版本控制：header 携带一个 `version`；`load` 拒绝任何非当前版本（不做迁移——预发布阶段的会话格式固定为 `SESSION_FORMAT_VERSION = 0` 并吸收形状变动，遵循 AGENTS.md 的预发布立场）。坦率地说：仅追加 + 刷写对部分尾部写入是健壮的（加载时容忍），但对行写入中途的无 fsync 断电不健壮；数据库/WAL 后端是后续更强的选项。 +格式版本控制：header 携带一个 `version`；`load` 拒绝任何非当前版本。预发布阶段的会话格式仍固定为 `SESSION_FORMAT_VERSION = 0`，不承诺广泛兼容；当持久化用户数据确有需要时，协调器可以负责显式且范围受限的导入升级（[消息标识机制引入前的消息恢复](../bug-fix/2026-07-28-load-pre-identity-session-messages.md)）。坦率地说：仅追加 + 刷写对部分尾部写入是健壮的（加载时容忍），但对行写入中途的无 fsync 断电不健壮；数据库/WAL 后端是后续更强的选项。 ## 后果 diff --git a/.agents/notes/implemented/bug-fix/2026-07-28-load-pre-identity-session-messages.i18n.yaml b/.agents/notes/implemented/bug-fix/2026-07-28-load-pre-identity-session-messages.i18n.yaml new file mode 100644 index 0000000000..56832b37e2 --- /dev/null +++ b/.agents/notes/implemented/bug-fix/2026-07-28-load-pre-identity-session-messages.i18n.yaml @@ -0,0 +1,6 @@ +# Bilingual-pair consistency record (docs/i18n/README.md): the git blob hash of each +# 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/bug-fix/2026-07-28-load-pre-identity-session-messages.md +2026-07-28-load-pre-identity-session-messages.md: 2901527658421b37576bdf5b49e66829104a3b41 +2026-07-28-load-pre-identity-session-messages.zh.md: 61d57ac9f3318299b63faa659b6d155e8e89fae3 diff --git a/.agents/notes/implemented/bug-fix/2026-07-28-load-pre-identity-session-messages.md b/.agents/notes/implemented/bug-fix/2026-07-28-load-pre-identity-session-messages.md new file mode 100644 index 0000000000..2901527658 --- /dev/null +++ b/.agents/notes/implemented/bug-fix/2026-07-28-load-pre-identity-session-messages.md @@ -0,0 +1,38 @@ +# Agent Note: Load sessions persisted before message identity + +Status: implemented + +English | [中文](2026-07-28-load-pre-identity-session-messages.zh.md) + +## Problem + +The identified immutable message change replaced four durable event payloads with complete message values. Existing v0 JSONL and SQLite sessions still held the immediately preceding shapes: direct `content`/`source` on user and steering events, `content`/`provenance` on assistant events, and `callId`/`content`/`isError` on tool results. Their headers still matched `SESSION_FORMAT_VERSION`, but current-shape validation rejected them before resume could construct a live `Session`. + +Changing the message representation without a version bump made those logs indistinguishable at the header level from current v0 logs. The runtime needs a narrow import rule that restores data created by the supported first-party backends without weakening validation for unrelated obsolete or malformed events. + +## Decision + +`PersistenceCoordinator` normalizes the four exact pre-identity message payloads after backend decoding and before current message validation. It wraps their existing semantic fields in the current role-specific message shape and assigns `legacy-message::` as the deterministic imported `MessageId`. A legacy `tool/result` content replacement inherits the imported id of its replacement target, preserving the current content-only rewrite invariant. + +The same normalization runs for `load`, `inspect`, an ownerless loaded state claiming its live session, and HMR prefix adoption. Prefix comparisons therefore compare the live current-shape seed with the same normalized stored view. Current-looking wrappers with missing or invalid fields are not repaired, and unsupported event vocabulary, request headers, versions, and surface relations retain their existing rejection paths. + +The upgrade is read-only. Stored legacy records remain unchanged; a resumed session appends only current-shape events after them. Deterministic identities make repeated loads and a mixed legacy/current log reproduce the same message ids without a backend-specific rewrite transaction. + +## Alternatives considered + +**Reject the logs under the pre-release compatibility stance.** This is the default for unrelated v0 churn, but it strands real first-party sessions even though every old field maps unambiguously to the current message representation. + +**Rewrite the complete stored log in place.** This would canonicalize the artifact but violate the append-only storage contract, require separate atomic replacement mechanisms for JSONL and SQLite, and expand a read compatibility fix into a migration system. + +**Mint random ids on each load.** The messages would satisfy the type shape but lose stable identity across inspect, resume, restart, and mixed legacy/current appends. + +## Consequences + +Pre-identity JSONL and SQLite sessions resume with their original message content, sources, provider provenance, tool correlation, errors, metadata, and surface replacements. The returned events are otherwise indistinguishable from current imported message snapshots and remain deeply frozen. + +This is one explicit same-version import exception, not a general v0 compatibility layer. Adding another exception requires another complete, unambiguous mapping at the persistence boundary; malformed current data continues to fail rather than being guessed into validity. The shared coordinator contract exercises the upgrade against the in-memory reference, JSONL, and SQLite backends, including deterministic reload and tool-result replacement identity. + +## Related + +- [Create every message as an identified immutable value](../architecture/2026-07-28-identified-immutable-message-values.md) — owns the current message identity and immutability contract. +- [Session persistence as an abstract service](../architecture/2026-06-14-session-persistence.md) — owns the append-only backend and resume boundary. diff --git a/.agents/notes/implemented/bug-fix/2026-07-28-load-pre-identity-session-messages.zh.md b/.agents/notes/implemented/bug-fix/2026-07-28-load-pre-identity-session-messages.zh.md new file mode 100644 index 0000000000..61d57ac9f3 --- /dev/null +++ b/.agents/notes/implemented/bug-fix/2026-07-28-load-pre-identity-session-messages.zh.md @@ -0,0 +1,38 @@ +# Agent Note: 加载消息标识机制引入前持久化的会话 + +Status: implemented + +[English](2026-07-28-load-pre-identity-session-messages.md) | 中文 + +## 问题 + +带标识的不可变消息变更将四种持久事件载荷替换为完整消息值。现有的 v0 JSONL 和 SQLite 会话仍保留紧邻该变更之前的形状：用户事件和 steering（中途引导）事件直接携带 `content`/`source`，assistant 事件携带 `content`/`provenance`，工具结果则携带 `callId`/`content`/`isError`。这些会话的 header 仍与 `SESSION_FORMAT_VERSION` 匹配，但当前形状验证会拒绝它们，导致恢复流程无法构造实时 `Session`。 + +消息表示改变时没有提升版本，导致这些日志无法仅凭 header 与当前的 v0 日志区分。运行时需要一条范围受限的导入规则，既能恢复受支持的第一方后端所创建的数据，又不削弱对无关过时事件或格式错误事件的验证。 + +## 决策 + +`PersistenceCoordinator` 会在后端解码之后、当前消息验证之前，规范化消息标识机制引入前的四种特定消息载荷。它将载荷现有的语义字段包装进当前按角色区分的消息形状，并为其分配确定性的导入 `MessageId`：`legacy-message::`。旧版 `tool/result` 的内容替换会继承替换目标导入后的 id，从而保持当前仅改写内容的不变量。 + +同一项规范化也用于 `load`、`inspect`、无 owner 的已加载状态认领其实时会话，以及 HMR（热模块替换）前缀接管。因此，前缀比较会将实时的当前形状 seed 与同一份规范化存储视图进行比较。看似当前形状、但字段缺失或无效的包装层不会被修复；不受支持的事件词汇、请求 header、版本和 surface 关系仍沿用现有拒绝路径。 + +这项升级只发生在读取时。存储中的旧版记录保持不变；会话恢复后，只会在其后追加当前形状的事件。确定性标识使重复加载以及新旧形状混合的日志无需执行后端专用的重写事务，也能复现相同的消息 id。 + +## 考虑过的替代方案 + +**按照预发布兼容性立场拒绝这些日志。** 这是处理其他 v0 形状变动的默认方式，但即使每个旧字段都能明确映射到当前消息表示，它仍会导致真实的第一方会话无法恢复。 + +**就地重写完整的存储日志。** 这会使产物规范化，但违反仅追加存储契约，还需要为 JSONL 和 SQLite 分别实现原子替换机制，并将一次读取兼容性修复扩大为迁移系统。 + +**每次加载时随机生成 id。** 这些消息会满足类型形状，却无法在检查、恢复、重启以及新旧形状混合追加之间保持稳定标识。 + +## 后果 + +消息标识机制引入前的 JSONL 和 SQLite 会话可以恢复，并保留原始的消息内容、来源、提供方溯源信息、工具关联、错误、元数据和 surface 替换。除此之外，返回事件与当前导入的消息快照无法区分，并且仍然经过深度冻结。 + +这是一个显式的同版本导入例外，而非通用的 v0 兼容层。若要增加另一个例外，必须在持久化边界提供另一套完整且无歧义的映射；当前数据若格式错误，系统仍会拒绝，而不会猜测如何将其变成有效数据。共享协调器契约会针对内存参考实现、JSONL 和 SQLite 后端验证这项升级，包括重新加载时的确定性，以及工具结果替换时的标识继承。 + +## 相关 + +- [将每条消息创建为带标识的不可变值](../architecture/2026-07-28-identified-immutable-message-values.md)：该记录负责当前的消息标识与不可变性契约。 +- [会话持久化作为抽象服务](../architecture/2026-06-14-session-persistence.md)：该记录负责仅追加后端与恢复边界。 diff --git a/docs/cordis-catalog/services.md b/docs/cordis-catalog/services.md index ac556aaf08..7bb03af778 100644 --- a/docs/cordis-catalog/services.md +++ b/docs/cordis-catalog/services.md @@ -1042,7 +1042,9 @@ abstract append(id: SessionId, events: readonly SessionEvent[]): Promise * A coordinator-backed cold load reserves the identity across storage awaits, * so concurrent publication of a same-id live Session rejects. * Returned events are detached, and every identified message is deeply - * frozen; malformed identified messages reject before any stored event is returned. + * frozen. Coordinator-backed implementations upgrade supported pre-identity + * message events before validation; other malformed messages reject before + * any stored event is returned. * @param id - the persisted session to reload. * @returns the header and a log ending on a balanced `turn/end`. */ @@ -1052,8 +1054,9 @@ abstract load(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEven * Inspect a header and its valid contiguous stored prefix without repairing * a torn tail, closing an interrupted turn, or publishing coordinator state. * This read is serialized with writes for the same id and returns detached - * values with deeply frozen identified messages, so observers cannot mutate message - * identity/content or backend-owned state. Malformed identified messages reject. + * values with upgraded, deeply frozen identified messages, so observers + * cannot mutate message identity/content or backend-owned state. Other + * malformed messages reject. * @param id - the persisted session to inspect. * @param signal - optional cancellation for queued and backend read work. * @returns the header and valid stored event prefix exactly as observed. diff --git a/packages/cordis/tool-cordis/src/api-catalog.ts b/packages/cordis/tool-cordis/src/api-catalog.ts index 311f7fc592..30ff0534ca 100644 --- a/packages/cordis/tool-cordis/src/api-catalog.ts +++ b/packages/cordis/tool-cordis/src/api-catalog.ts @@ -518,11 +518,11 @@ export const SERVICE_API: readonly ServiceApiEntry[] = [ }, { signature: 'abstract load(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }>', - jsDoc: '/**\n * Load a header and balanced contiguous log. A complete interrupted final\n * turn is preserved and durably closed with missing tool errors plus any open\n * step and turn boundaries; only a torn final record is discarded. Unknown\n * versions and corruption in the committed prefix reject. Implementations\n * MUST NOT crash-repair an identity still bound to a live Session: a balanced\n * live log may return with its stored header as a durable snapshot, while an\n * open live turn rejects.\n * A coordinator-backed cold load reserves the identity across storage awaits,\n * so concurrent publication of a same-id live Session rejects.\n * Returned events are detached, and every identified message is deeply\n * frozen; malformed identified messages reject before any stored event is returned.\n * @param id - the persisted session to reload.\n * @returns the header and a log ending on a balanced `turn/end`.\n */', + jsDoc: '/**\n * Load a header and balanced contiguous log. A complete interrupted final\n * turn is preserved and durably closed with missing tool errors plus any open\n * step and turn boundaries; only a torn final record is discarded. Unknown\n * versions and corruption in the committed prefix reject. Implementations\n * MUST NOT crash-repair an identity still bound to a live Session: a balanced\n * live log may return with its stored header as a durable snapshot, while an\n * open live turn rejects.\n * A coordinator-backed cold load reserves the identity across storage awaits,\n * so concurrent publication of a same-id live Session rejects.\n * Returned events are detached, and every identified message is deeply\n * frozen. Coordinator-backed implementations upgrade supported pre-identity\n * message events before validation; other malformed messages reject before\n * any stored event is returned.\n * @param id - the persisted session to reload.\n * @returns the header and a log ending on a balanced `turn/end`.\n */', }, { signature: 'abstract inspect(id: SessionId, signal?: AbortSignal): Promise<{ meta: SessionHeader; events: SessionEvent[] }>', - jsDoc: '/**\n * Inspect a header and its valid contiguous stored prefix without repairing\n * a torn tail, closing an interrupted turn, or publishing coordinator state.\n * This read is serialized with writes for the same id and returns detached\n * values with deeply frozen identified messages, so observers cannot mutate message\n * identity/content or backend-owned state. Malformed identified messages reject.\n * @param id - the persisted session to inspect.\n * @param signal - optional cancellation for queued and backend read work.\n * @returns the header and valid stored event prefix exactly as observed.\n */', + jsDoc: '/**\n * Inspect a header and its valid contiguous stored prefix without repairing\n * a torn tail, closing an interrupted turn, or publishing coordinator state.\n * This read is serialized with writes for the same id and returns detached\n * values with upgraded, deeply frozen identified messages, so observers\n * cannot mutate message identity/content or backend-owned state. Other\n * malformed messages reject.\n * @param id - the persisted session to inspect.\n * @param signal - optional cancellation for queued and backend read work.\n * @returns the header and valid stored event prefix exactly as observed.\n */', }, { signature: 'abstract list(signal?: AbortSignal): Promise', diff --git a/packages/core/agent-loop/tests/resume.spec.ts b/packages/core/agent-loop/tests/resume.spec.ts index bf3e91b078..f133f1979c 100644 --- a/packages/core/agent-loop/tests/resume.spec.ts +++ b/packages/core/agent-loop/tests/resume.spec.ts @@ -5,7 +5,7 @@ import { mkdtemp, rm } from 'node:fs/promises' import { tmpdir } from 'node:os' import { join } from 'node:path' import LlmService from '@deepseek-ai/dsh-llm' -import SessionStore, { Session, SessionId } from '@deepseek-ai/dsh-session' +import SessionStore, { SESSION_FORMAT_VERSION, Session, SessionId } from '@deepseek-ai/dsh-session' import type { SessionEvent } from '@deepseek-ai/dsh-session' import SystemPrompt from '@deepseek-ai/dsh-system-prompt' import ToolRegistry from '@deepseek-ai/dsh-tools' @@ -77,6 +77,65 @@ function throwUnknown(value: unknown): never { } describe('the session-persistence Agent Note: AgentLoop factory create/resume', () => { + it('resumes a session persisted before messages gained identities', async () => { + const sessionId = SessionId('pre-identity-resume') + const first = await persistentHarness(new MockAdapter([])) + await first.ctx.sessionPersistence.create({ + version: SESSION_FORMAT_VERSION, + id: sessionId, + createdAt: 1, + }) + await first.ctx.sessionPersistence.append(sessionId, [ + { type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } }, + { + type: 'user/message', + seq: 1, + time: 2, + data: { content: [{ type: 'text', text: 'old question' }], source: { kind: 'user' } }, + surfaceOp: 'append', + }, + { type: 'step/start', seq: 2, time: 3, data: { turn: 1, step: 1 } }, + { + type: 'assistant/message', + seq: 3, + time: 4, + data: { + turn: 1, + step: 1, + content: [{ type: 'text', text: 'old answer' }], + provenance: { provider: 'mock', model: 'mock' }, + }, + surfaceOp: 'append', + }, + { type: 'step/end', seq: 4, time: 5, data: { turn: 1, step: 1 } }, + { type: 'turn/end', seq: 5, time: 6, data: { turn: 1, reason: { kind: 'completed' } } }, + ] as unknown as SessionEvent[]) + await first.ctx.fiber.dispose() + + const ctx = await mountPersistentHarness(first.root, new MockAdapter([textResponse('new answer')])) + const handle = await ctx.agents.resume({ + resumeSessionId: sessionId, + agentOptions: { provider: 'mock', model: 'mock' }, + }) + expect(handle.agent.session.deriveMessages()).toMatchObject([ + { id: `legacy-message:${sessionId}:1`, role: 'user' }, + { id: `legacy-message:${sessionId}:3`, role: 'assistant' }, + ]) + + handle.agent.followup(createUserMessage({ + content: [{ type: 'text', text: 'new question' }], + source: { kind: 'user' }, + })) + await waitForIdle(ctx, handle.agent) + expect(handle.agent.session.deriveMessages()).toHaveLength(4) + expect(handle.agent.session.events.at(-1)).toMatchObject({ + type: 'turn/end', + data: { reason: { kind: 'completed' } }, + }) + await handle.dispose() + await ctx.fiber.dispose() + }) + it('normalizes a non-Error resume publication failure for rollback and rethrows it', async () => { const sessionId = SessionId('unknown-resume-failure-s') const root = await persistSession(sessionId) diff --git a/packages/core/session/README.i18n.yaml b/packages/core/session/README.i18n.yaml index c9bf568a38..9fb097d9fb 100644 --- a/packages/core/session/README.i18n.yaml +++ b/packages/core/session/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/core/session/README.md -README.md: 40516d12180de9c30efd40fdffa873da20ddacb3 -README.zh.md: 43842643a3434c741f219f7b6c26622cddfae8e7 +README.md: a9b6905dcf2b8ef1f75595e567273f7a3150a412 +README.zh.md: f1a5e97e32d1ad1abcd6ad96e6c621af9972e989 diff --git a/packages/core/session/README.md b/packages/core/session/README.md index 40516d1218..a9b6905dcf 100644 --- a/packages/core/session/README.md +++ b/packages/core/session/README.md @@ -142,5 +142,5 @@ Logging causes no invalidation, and exact reconstruction preserves request-prefi - **Session branching/tree** (pi-style entry tree) — deferred unless needed beyond boundary-based `fork()`. - **`fork()` cuts only at stable boundaries of live sessions** — the selected prefix must end outside an open turn and the source must be in the store; forking a persisted-but-unloaded session is excluded from the [fork API](../../../.agents/notes/implemented/feature/2026-06-30-session-store-fork-api.md). -- **`SESSION_FORMAT_VERSION` stays pinned at `0`** — pre-release, no compatibility implied: a backend rejects any other version, and no migration path exists until the first release ([policy](../../../AGENTS.md)). +- **`SESSION_FORMAT_VERSION` stays pinned at `0`** — pre-release, no broad compatibility implied: `Session` accepts only current seed shapes and a backend rejects any other version. Narrow storage import upgrades belong to the persistence boundary ([policy](../../../AGENTS.md), [pre-identity message recovery](../../../.agents/notes/implemented/bug-fix/2026-07-28-load-pre-identity-session-messages.md)). - **`TurnEndReasonMap` omits the ACP-named `refusal` / `max_turn_requests` variants** — producer-gated: they land when an adapter or the loop first emits them. diff --git a/packages/core/session/README.zh.md b/packages/core/session/README.zh.md index 43842643a3..f1a5e97e32 100644 --- a/packages/core/session/README.zh.md +++ b/packages/core/session/README.zh.md @@ -142,5 +142,5 @@ - **会话分支／树**（pi 风格条目树）：除非需要超越基于边界的 `fork()` 能力，否则暂缓。 - **`fork()` 仅在实时会话的稳定边界处切分**：所选前缀结束时不得有开放轮次，且源会话必须位于存储中；[fork API](../../../.agents/notes/implemented/feature/2026-06-30-session-store-fork-api.md) 不支持对已持久化但未加载的会话进行 fork。 -- **`SESSION_FORMAT_VERSION` 固定为 `0`**：预发布阶段不承诺兼容性；后端会拒绝其他任何版本，首次发布前不提供迁移路径（[政策](../../../AGENTS.md)）。 +- **`SESSION_FORMAT_VERSION` 固定为 `0`**：预发布阶段不承诺广泛兼容性；`Session` 只接受当前 seed 形状，后端会拒绝其他任何版本。范围受限的存储导入升级应由持久化边界负责（[政策](../../../AGENTS.md)、[消息标识机制引入前的消息恢复](../../../.agents/notes/implemented/bug-fix/2026-07-28-load-pre-identity-session-messages.md)）。 - **`TurnEndReasonMap` 不含 ACP（Agent Client Protocol）命名的 `refusal`／`max_turn_requests` 变体**：受生产方约束；只有当适配器或循环首次产生这些变体时才加入。 diff --git a/packages/session-persistence/session-persistence/README.i18n.yaml b/packages/session-persistence/session-persistence/README.i18n.yaml index 0e2cecaada..184f29e3e8 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: 08d8adac8040747a6dac01dbc41525073f17060c -README.zh.md: 7676f27a1aa934eb3472e1b32b9ecd55d460fb63 +README.md: 3617305d0343ab4c0d9d802669a3c4f964271dc7 +README.zh.md: ffa86b0093331306d524a590364fac527a2e5071 diff --git a/packages/session-persistence/session-persistence/README.md b/packages/session-persistence/session-persistence/README.md index 08d8adac80..3617305d03 100644 --- a/packages/session-persistence/session-persistence/README.md +++ b/packages/session-persistence/session-persistence/README.md @@ -13,8 +13,8 @@ The persisted unit IS the existing `SessionEvent` (event-sourced model — the l | `locate(meta): SessionLocation \| undefined` | Resolve an absolute per-session artifact target without I/O or materialization. Backends without an independent local artifact return `undefined`. | | `create(meta): Promise` | Register a new session's metadata. MAY defer the physical write until the first `append` (lazy materialization). | | `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. | -| `load(id): Promise<{ meta; events }>` | Return a stored header plus a balanced contiguous log whose events are detached and validated and whose identified messages are deeply frozen. A live load first flushes its snapshot and rejects while its turn is open; a cold load preserves an interrupted final turn and closes it with synthetic `tool/result`/`step/end?`/`turn/end {interrupted}` events. Only a torn tail fragment is dropped; committed corruption, malformed messages, and unknown `version` reject. | -| `inspect(id, signal?): Promise<{ meta; events }>` | Return a detached valid stored prefix with validated, deeply frozen identified messages, without truncating a torn tail, synthesizing recovery closers, or publishing coordinator state. Serialized with same-id writes; the optional signal promptly rejects a queued caller, prevents that queued backend read from starting, and cancels active backend read work. Intended for read models and other observers that must never recover a log. | +| `load(id): Promise<{ meta; events }>` | Return a stored header plus a balanced contiguous log whose events are detached and validated and whose identified messages are deeply frozen. The coordinator upgrades the four pre-identity message event shapes into current wrappers in the returned snapshot; all other obsolete or malformed shapes still reject. A live load first flushes its snapshot and rejects while its turn is open; a cold load preserves an interrupted final turn and closes it with synthetic `tool/result`/`step/end?`/`turn/end {interrupted}` events. Only a torn tail fragment is dropped; committed corruption and unknown `version` reject. | +| `inspect(id, signal?): Promise<{ meta; events }>` | Return a detached valid stored prefix with upgraded, validated, deeply frozen identified messages, without truncating a torn tail, synthesizing recovery closers, or publishing coordinator state. Serialized with same-id writes; the optional signal promptly rejects a queued caller, prevents that queued backend read from starting, and cancels active backend read work. Intended for read models and other observers that must never recover a log. | | `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,6 +33,8 @@ Each `session/event` copies its event into the session controller and starts an Crash repair is cold-only. For a live id, `load(id)` snapshots the authoritative in-memory log, waits for that snapshot to become durable, and returns it with the coordinator's stored header only when balanced; an open live turn rejects instead of receiving synthetic interruption closers. A cold load reserves its id across backend reads and repair writes, so concurrent publication of a same-id live `Session` rejects and rolls back. HMR adoption reads through `loadStored`, applies the coordinator's cwd check, and never closes the active turn. +Backend reads normalize pre-identity `user/message`, `assistant/message`, `tool/result`, and `steering/message` payloads before current-shape validation. Each imported message receives the deterministic id `legacy-message::`; a tool-result content replacement inherits its target's imported id. The coordinator uses the same normalized view for `load`, `inspect`, ownerless-state claims, and HMR prefix adoption, so resumed sessions can append current events without a false prefix collision. Storage remains append-only: the read does not rewrite old records, and every later append uses the current shape. This is the narrow import exception from the [pre-identity message recovery decision](../../../.agents/notes/implemented/bug-fix/2026-07-28-load-pre-identity-session-messages.md), not a general v0 migration promise. + When a live session emits `session/disposed`, the coordinator waits for its controller, serializes a final drain, then releases state owned by that exact `Session` object. Failed retirement leaves the controller in the live-session map, so backend teardown can retry it. Backend teardown stops event admission first, flushes every remaining controller, awaits per-id operations, and only then closes the storage handle. The side-effect-free `locate` and lightweight `listSnapshots` queries remain backend-owned because they describe storage topology and revision identity rather than write orchestration. `listSnapshots(signal?)` passes the caller's exact signal into backend discovery so observers can cancel that work without detaching it. diff --git a/packages/session-persistence/session-persistence/README.zh.md b/packages/session-persistence/session-persistence/README.zh.md index 7676f27a1a..ffa86b0093 100644 --- a/packages/session-persistence/session-persistence/README.zh.md +++ b/packages/session-persistence/session-persistence/README.zh.md @@ -13,8 +13,8 @@ | `locate(meta): SessionLocation \| undefined` | 在不执行 I/O 或实体化的情况下解析绝对的每会话产物目标。没有独立本地产物的后端返回 `undefined`。 | | `create(meta): Promise` | 注册新会话元数据。可以将物理写入延迟到第一次 `append`（延迟实体化）。 | | `append(id, events): Promise` | 持久保存一个批次。仅追加；任何修复后，第一个事件 `seq` == 已存储 next-seq；非 JSON 可序列化数据会被拒绝，并命名违规类型。 | -| `load(id): Promise<{ meta; events }>` | 返回已存储 header 和平衡、连续的日志，其中事件已脱离并验证，带标识的消息已深度冻结。实时 load 先 flush 其快照，并在轮次开放时拒绝；冷 load 保留中断的最终轮次，并用合成 `tool/result`/`step/end?`/`turn/end {interrupted}` 事件关闭它。只丢弃撕裂尾部碎片；已提交损坏、格式错误的消息和未知 `version` 会被拒绝。 | -| `inspect(id, signal?): Promise<{ meta; events }>` | 返回脱离的有效已存储前缀，其中带标识的消息已经验证并深度冻结；不截断撕裂尾部、合成恢复 closer 或发布协调器状态。它与同 id 写入串行化；可选信号会迅速拒绝已排队调用方，阻止该后端读取启动，并取消活动后端读取工作。用于绝不应恢复日志的读模型和其他观察者。 | +| `load(id): Promise<{ meta; events }>` | 返回已存储 header 和平衡、连续的日志，其中事件已脱离并验证，带标识的消息已深度冻结。协调器会在返回快照中，将消息标识机制引入前的四种消息事件形状升级为当前包装层；其余过时或格式错误的形状仍会被拒绝。实时 load 先 flush 其快照，并在轮次开放时拒绝；冷 load 保留中断的最终轮次，并用合成 `tool/result`/`step/end?`/`turn/end {interrupted}` 事件关闭它。只丢弃撕裂尾部碎片；已提交损坏和未知 `version` 会被拒绝。 | +| `inspect(id, signal?): Promise<{ meta; events }>` | 返回脱离的有效已存储前缀，其中带标识的消息已经升级、验证并深度冻结；不截断撕裂尾部、合成恢复 closer 或发布协调器状态。它与同 id 写入串行化；可选信号会迅速拒绝已排队调用方，阻止该后端读取启动，并取消活动后端读取工作。用于绝不应恢复日志的读模型和其他观察者。 | | `list(signal?): Promise` | 从元数据轻量列出，不解析完整日志。可选信号取消后端列表工作。零事件延迟实体化会话不在 `list` 中。 | | `listSnapshots(signal?): Promise` | 返回轻量元数据和不透明品牌化每日志修订，不加载事件日志。日志及其后端存储不变时，修订保持相等；append 或变更性 load 修复后会改变；不会仅因两个存储使用相同本地计数器而冲突。可选信号请求取消后端发现工作；第一方后端在拒绝前结算已启动列表工作，使已等待调用完全停稳。 | @@ -33,6 +33,8 @@ 崩溃修复只适用于冷状态。对于实时 id，`load(id)` 为权威内存日志制作快照，等待该快照持久，并只在平衡时将其与协调器已存储 header 一起返回；开放实时轮次会被拒绝，而不会收到合成中断 closer。冷 load 在后端读取和修复写入期间保留 id，因此同 id 实时 `Session` 的并发发布会拒绝并回滚。HMR 接管通过 `loadStored` 读取，应用协调器 cwd 检查，并绝不关闭活动轮次。 +后端读取会在当前形状验证前，规范化消息标识机制引入前的 `user/message`、`assistant/message`、`tool/result` 以及 steering（中途引导）对应的 `steering/message` 载荷。每条导入消息都会获得确定性的 id `legacy-message::`；工具结果的内容替换会继承其目标导入后的 id。协调器对 `load`、`inspect`、无 owner 状态的认领和 HMR 前缀接管使用同一份规范化视图，因此恢复后的会话可以追加当前事件，不会被误判为发生前缀冲突。存储仍然仅追加：读取不会重写旧记录，此后追加的每个事件都使用当前形状。这是[消息标识机制引入前的消息恢复决策](../../../.agents/notes/implemented/bug-fix/2026-07-28-load-pre-identity-session-messages.md)所规定的范围受限的导入例外，并不构成通用的 v0 迁移承诺。 + 实时会话发出 `session/disposed` 时，协调器等待其 controller，串行化最终 drain，然后释放该精确 `Session` 对象拥有的状态。失败退役会将 controller 保留在实时会话 map 中，使后端拆卸可重试。后端拆卸先停止事件接纳，flush 每个剩余 controller，等待每 id 操作，最后才关闭存储句柄。 无副作用 `locate` 和轻量 `listSnapshots` 查询仍由后端负责，因为它们描述存储拓扑和修订身份，而非写入编排。`listSnapshots(signal?)` 将调用方的精确信号传入后端发现，使观察者可在不脱离该工作的情况下取消。 diff --git a/packages/session-persistence/session-persistence/src/coordinator.ts b/packages/session-persistence/session-persistence/src/coordinator.ts index 878dddcb42..2d4d0290a9 100644 --- a/packages/session-persistence/session-persistence/src/coordinator.ts +++ b/packages/session-persistence/session-persistence/src/coordinator.ts @@ -146,10 +146,142 @@ function assertSupportedEvents(events: readonly SessionEvent[], id: SessionId): } } -/** Materialize stored events as validated snapshots with immutable messages. */ +/** Return an object record without widening arrays into message payloads. */ +function asRecord(value: unknown): Record | undefined { + return typeof value === 'object' && value !== null && !Array.isArray(value) + ? value as Record + : undefined +} + +type PersistedMessageId = SessionEvent<'user/message'>['data']['id'] + +/** Mint the stable import identity for a message persisted before identities existed. */ +function legacyMessageId(id: SessionId, seq: number): PersistedMessageId { + return `legacy-message:${id}:${seq}` as PersistedMessageId +} + +/** Read a replacement target while leaving malformed surface metadata to the session validator. */ +function replacementStart(event: SessionEvent): number | undefined { + const op = asRecord((event as SessionEvent & { surfaceOp?: unknown }).surfaceOp) + return op?.['op'] === 'replace' && typeof op['start'] === 'number' + ? op['start'] + : undefined +} + +/** + * Upgrade one pre-identity message event into the current wrapper shape. + * Current-looking malformed events remain untouched so validation rejects them + * instead of disguising corruption as legacy data. + */ +function migrateLegacyMessageEvent( + event: SessionEvent, + id: SessionId, + messageIds: ReadonlyMap, +): SessionEvent { + const data = asRecord(event.data) + if (data === undefined) return event + switch (event.type) { + case 'user/message': { + if (Object.hasOwn(data, 'id') || Object.hasOwn(data, 'role') + || Object.hasOwn(data, 'message') + || !Object.hasOwn(data, 'content') || !Object.hasOwn(data, 'source')) return event + return { + ...event, + data: { + ...data, + id: legacyMessageId(id, event.seq), + role: 'user', + }, + } as SessionEvent + } + case 'assistant/message': { + if (Object.hasOwn(data, 'message') + || !Object.hasOwn(data, 'content') || !Object.hasOwn(data, 'provenance')) return event + const { content, provenance, ...eventData } = data + return { + ...event, + data: { + ...eventData, + message: { + id: legacyMessageId(id, event.seq), + role: 'assistant', + content, + source: { + ...asRecord(provenance), + kind: 'model', + }, + }, + }, + } as SessionEvent + } + case 'tool/result': { + if (Object.hasOwn(data, 'message') + || !Object.hasOwn(data, 'callId') || !Object.hasOwn(data, 'content') + || !Object.hasOwn(data, 'isError')) return event + const { callId, content, isError, ...eventData } = data + const inheritedId = replacementStart(event) + return { + ...event, + data: { + ...eventData, + message: { + id: inheritedId === undefined + ? legacyMessageId(id, event.seq) + : messageIds.get(inheritedId), + role: 'user', + content: [{ + type: 'tool-result', + toolCallId: callId, + content, + isError, + }], + source: { + kind: 'tool', + callId, + }, + }, + }, + } as SessionEvent + } + case 'steering/message': { + if (Object.hasOwn(data, 'message') + || !Object.hasOwn(data, 'content') || !Object.hasOwn(data, 'source')) return event + const { content, source, ...eventData } = data + return { + ...event, + data: { + ...eventData, + message: { + id: legacyMessageId(id, event.seq), + role: 'user', + content, + source, + }, + }, + } as SessionEvent + } + default: + return event + } +} + +/** Read the identified message carried by one validated current event. */ +function eventMessageId(event: SessionEvent): PersistedMessageId | undefined { + const data = asRecord(event.data) + const message = event.type === 'user/message' ? data : asRecord(data?.['message']) + return typeof message?.['id'] === 'string' ? message['id'] as PersistedMessageId : undefined +} + +/** Materialize stored events as upgraded, validated snapshots with immutable messages. */ function snapshotStoredEvents(events: readonly SessionEvent[], id: SessionId): SessionEvent[] { assertSupportedEvents(events, id) - return events.map(snapshotSessionEvent) + const messageIds = new Map() + return events.map((event) => { + const snapshot = snapshotSessionEvent(migrateLegacyMessageEvent(event, id, messageIds)) + const messageId = eventMessageId(snapshot) + if (messageId !== undefined) messageIds.set(snapshot.seq, messageId) + return snapshot + }) } /** @@ -526,7 +658,7 @@ export class PersistenceCoordinator { /* v8 ignore next -- a cursor > 0 means the session was materialized, so it exists */ if (stored === undefined) return false this.assertStoredId(id, stored.meta) - return seedCoversPrefix(seed, stored.events.slice(0, cursor)) + return seedCoversPrefix(seed, snapshotStoredEvents(stored.events, id).slice(0, cursor)) } /** @@ -614,19 +746,19 @@ export class PersistenceCoordinator { throw new Error(`session "${session.header.id}" is already persisted at a different cwd (persisted: ${String(meta.cwd)}, live: ${String(session.header.cwd)}) (id collision)`) } this.assertVersion(meta) - assertSupportedEvents(events, session.header.id) - if (!seedCoversPrefix(seed, events)) { + const storedEvents = snapshotStoredEvents(events, session.header.id) + if (!seedCoversPrefix(seed, storedEvents)) { throw new Error(`session "${session.header.id}" already has a persisted log on disk that does not match this live session (id collision)`) } // Truncate-only repair (no closers): the open turn is NOT closed here. if (tornMarker !== undefined) await this.backend.commitRepair(meta, tornMarker, []) this.states.set(session.header.id, { meta: { ...meta }, - cursor: events.length, + cursor: storedEvents.length, materialized: true, owner: session, }) - const suffix = seed.slice(events.length) + const suffix = seed.slice(storedEvents.length) if (suffix.length > 0) await this.appendCore(session.header.id, suffix) } diff --git a/packages/session-persistence/session-persistence/src/index.ts b/packages/session-persistence/session-persistence/src/index.ts index 261b26d364..0af538194a 100644 --- a/packages/session-persistence/session-persistence/src/index.ts +++ b/packages/session-persistence/session-persistence/src/index.ts @@ -93,7 +93,9 @@ export abstract class SessionPersistence extends Service { * A coordinator-backed cold load reserves the identity across storage awaits, * so concurrent publication of a same-id live Session rejects. * Returned events are detached, and every identified message is deeply - * frozen; malformed identified messages reject before any stored event is returned. + * frozen. Coordinator-backed implementations upgrade supported pre-identity + * message events before validation; other malformed messages reject before + * any stored event is returned. * @param id - the persisted session to reload. * @returns the header and a log ending on a balanced `turn/end`. */ @@ -103,8 +105,9 @@ export abstract class SessionPersistence extends Service { * Inspect a header and its valid contiguous stored prefix without repairing * a torn tail, closing an interrupted turn, or publishing coordinator state. * This read is serialized with writes for the same id and returns detached - * values with deeply frozen identified messages, so observers cannot mutate message - * identity/content or backend-owned state. Malformed identified messages reject. + * values with upgraded, deeply frozen identified messages, so observers + * cannot mutate message identity/content or backend-owned state. Other + * malformed messages reject. * @param id - the persisted session to inspect. * @param signal - optional cancellation for queued and backend read work. * @returns the header and valid stored event prefix exactly as observed. diff --git a/packages/session-persistence/session-persistence/tests/coordinator-contract.ts b/packages/session-persistence/session-persistence/tests/coordinator-contract.ts index d63cb857a2..1de20528b2 100644 --- a/packages/session-persistence/session-persistence/tests/coordinator-contract.ts +++ b/packages/session-persistence/session-persistence/tests/coordinator-contract.ts @@ -13,8 +13,8 @@ import { createUserMessage } from '@deepseek-ai/dsh-llm' import { describe, expect, it, vi } from 'vitest' import { Context, type Fiber } from 'cordis' import { scopeTarget } from '@deepseek-ai/dsh-scope' -import SessionStore, { SESSION_FORMAT_VERSION, SessionId } from '@deepseek-ai/dsh-session' -import type { Session, SessionEvent } from '@deepseek-ai/dsh-session' +import SessionStore, { SESSION_FORMAT_VERSION, Session, SessionId } from '@deepseek-ai/dsh-session' +import type { SessionEvent } from '@deepseek-ai/dsh-session' import { meta, oneTurnLog, appendLog } from './contract.ts' /** @@ -45,6 +45,80 @@ function send(session: Session, events: readonly SessionEvent[]): void { appendLog(session, events) } +/** A valid persisted log from immediately before messages gained wrappers and identities. */ +function legacyMessageLog(): SessionEvent[] { + return [ + { type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } }, + { + type: 'user/message', + seq: 1, + time: 2, + data: { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }, + surfaceOp: 'append', + }, + { type: 'step/start', seq: 2, time: 3, data: { turn: 1, step: 1 } }, + { + type: 'assistant/message', + seq: 3, + time: 4, + data: { + turn: 1, + step: 1, + content: [{ type: 'tool-call', id: 'call-1', name: 'read', arguments: '{}' }], + provenance: { provider: 'mock', model: 'mock' }, + }, + surfaceOp: 'append', + }, + { + type: 'tool/call', + seq: 4, + time: 5, + data: { turn: 1, step: 1, callId: 'call-1', name: 'read', arguments: '{}' }, + }, + { + type: 'tool/result', + seq: 5, + time: 6, + data: { + turn: 1, + step: 1, + callId: 'call-1', + content: [{ type: 'text', text: 'full result' }], + isError: false, + }, + sourceEventSeqs: [4], + surfaceOp: 'append', + }, + { + type: 'steering/message', + seq: 6, + time: 7, + data: { + turn: 1, + content: [{ type: 'text', text: 'continue' }], + source: { kind: 'plugin', plugin: 'test' }, + }, + surfaceOp: 'append', + }, + { + type: 'tool/result', + seq: 7, + time: 8, + data: { + turn: 1, + step: 1, + callId: 'call-1', + content: [{ type: 'text', text: 'pruned' }], + isError: false, + }, + sourceEventSeqs: [5], + surfaceOp: { op: 'replace', start: 5, end: 5 }, + }, + { type: 'step/end', seq: 8, time: 9, data: { turn: 1, step: 1 } }, + { type: 'turn/end', seq: 9, time: 10, data: { turn: 1, reason: { kind: 'completed' } } }, + ] as unknown as SessionEvent[] +} + /** A live session created inside its OWN fiber, so it survives a backend reload. */ async function liveSessionInFiber( ctx: Context, id: string, cwd: string | undefined, @@ -269,6 +343,48 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise< } }) + it('loads pre-identity message logs into resumable current sessions', async () => { + const fix = await makeFixture() + const { ctx, fiber } = await freshCtx(fix) + try { + const id = SessionId('legacy-message-load') + await ctx.sessionPersistence.create(meta(id, WORK)) + await ctx.sessionPersistence.append(id, legacyMessageLog()) + + for (const snapshot of [ + await ctx.sessionPersistence.inspect(id), + await ctx.sessionPersistence.load(id), + ]) { + const messages = snapshot.events.flatMap((event) => { + if (event.type === 'user/message') return [event.data] + if (event.type === 'assistant/message' + || event.type === 'tool/result' + || event.type === 'steering/message') return [event.data.message] + return [] + }) + expect(messages.map(message => message.id)).toEqual([ + `legacy-message:${id}:1`, + `legacy-message:${id}:3`, + `legacy-message:${id}:5`, + `legacy-message:${id}:6`, + `legacy-message:${id}:5`, + ]) + expect(messages.every(message => Object.isFrozen(message))).toBe(true) + + const resumed = new Session(id, snapshot.events, snapshot.meta) + expect(resumed.deriveMessages().map(message => message.id)).toEqual([ + `legacy-message:${id}:1`, + `legacy-message:${id}:3`, + `legacy-message:${id}:5`, + `legacy-message:${id}:6`, + ]) + } + } finally { + await fiber.dispose() + await fix.cleanup() + } + }) + it('rejects malformed persisted message events before returning them', async () => { const fix = await makeFixture() const { ctx, fiber } = await freshCtx(fix) @@ -292,6 +408,31 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise< .rejects.toThrow('message must have role "user"') await expect(ctx.sessionPersistence.load(id)) .rejects.toThrow('message must have role "user"') + + for (const type of ['tool/result', 'steering/message'] as const) { + const malformedId = SessionId(`invalid-${type}`) + await ctx.sessionPersistence.create(meta(malformedId, WORK)) + await ctx.sessionPersistence.append(malformedId, [{ + type, + seq: 0, + time: 1, + surfaceOp: 'append', + data: { message: null }, + } as unknown as SessionEvent]) + await expect(ctx.sessionPersistence.inspect(malformedId)) + .rejects.toThrow('lacks an identified message') + } + + const pluginId = SessionId('non-object-plugin-event') + await ctx.sessionPersistence.create(meta(pluginId, WORK)) + await ctx.sessionPersistence.append(pluginId, [{ + type: 'plugin/test', + seq: 0, + time: 1, + data: null, + } as unknown as SessionEvent]) + await expect(ctx.sessionPersistence.inspect(pluginId)) + .resolves.toMatchObject({ events: [{ type: 'plugin/test', data: null }] }) } finally { await fiber.dispose() await fix.cleanup()