From 15b87d7aed56621585384ae77fa8b74b530a9a9d Mon Sep 17 00:00:00 2001 From: shiquan_zhan Date: Thu, 9 Jul 2026 16:24:45 +0800 Subject: [PATCH 1/2] docs: add issue 89 replay consistency design --- ...7-09-issue-89-replay-consistency-design.md | 325 ++++++++++++++++++ 1 file changed, 325 insertions(+) create mode 100644 docs/superpowers/specs/2026-07-09-issue-89-replay-consistency-design.md diff --git a/docs/superpowers/specs/2026-07-09-issue-89-replay-consistency-design.md b/docs/superpowers/specs/2026-07-09-issue-89-replay-consistency-design.md new file mode 100644 index 00000000..98adea69 --- /dev/null +++ b/docs/superpowers/specs/2026-07-09-issue-89-replay-consistency-design.md @@ -0,0 +1,325 @@ +# Issue #89 Session / Memory 多后端回放一致性框架设计 + +## 1. 目标与边界 + +本设计为 tRPC-Agent-Python 建立可重复、可诊断、可扩展的 Session / Memory / Summary 回放一致性测试框架，回答三个相互独立的问题： + +1. 相同操作和显式配置下，不同后端的业务行为是否一致； +2. SQLite 关闭并重新创建服务后，持久化投影是否保持不变； +3. 重试、响应丢失或持久化失败后，SDK 当前表现为何，比较框架能否准确识别。 + +框架不新增生产 API，也不把测试框架观测字段伪装成 SDK 契约。当前 SDK 没有持久化的 Summary version，因此 `observed_generation_ordinal` 只能作为 harness observation；`summary.persisted_version` 在维护者确认前必须报告为 unsupported。 + +## 2. 测试层次 + +### 2.1 Replay Consistency + +使用相同轨迹和相同 `SessionServiceConfig` 驱动 InMemory 与 SQLite，并在集成模式下选择性加入 Redis。严格比较 Session、事件、State、Memory 和 Summary 的当前公开契约。 + +### 2.2 Persistence Recovery + +在同一组 SQLite Session/Memory 数据库文件上执行： + +```text +replay -> warm snapshot -> close -> reopen -> cold snapshot +``` + +Warm/Cold 只严格比较持久化投影。重新打开的 SQLite 必须使用新的 `SummarizerSessionManager`，其运行时 cache 为空。 + +### 2.3 Mutation / Capability Detection + +- Snapshot mutation 只验证比较器对人为差异的检出能力； +- 操作级故障注入用于识别当前幂等性和原子恢复能力； +- 未由 SDK 保证的理想行为不作为强制契约； +- 无法唯一分类的状态必须使测试失败。 + +## 3. 运行模式 + +统一使用 `REPLAY_MODE`： + +| 模式 | 必需后端 | 内容 | +| --- | --- | --- | +| `inmemory` | InMemory | 10 条 replay、业务不变量、10 条 mutation、运行时 Summary、性能记录 | +| `contract` | InMemory + SQLite | 跨后端比较、SQLite Warm/Cold、Default Profile、恢复能力分类 | +| `integration` | InMemory + SQLite；Redis 可选 | Contract 全部内容及 Redis 比较 | + +默认模式为 `contract`。Contract/Integration 中 SQLite 构造失败必须失败，不能 skip 或退化为单后端。只有 Redis 可以因未配置、缺少可选依赖或外部服务不可达而明确 skip。 + +InMemory 轻量模式不创建 SQLite Engine。30 秒预算默认记录，通过 `REPLAY_ENFORCE_BUDGET=1` 在专用验收中强制。 + +## 4. 工程结构 + +```text +tests/sessions/ +├── replay_consistency/ +│ ├── __init__.py +│ ├── __main__.py +│ ├── model.py +│ ├── backends.py +│ ├── replay.py +│ ├── snapshot.py +│ ├── compare.py +│ ├── replay_cases.jsonl +│ ├── report.schema.json +│ ├── example_report.json +│ └── README.md +├── test_replay_consistency.py +└── test_replay_recovery.py +``` + +职责如下： + +- `model.py`：case、operation、required observation、diff 和 evaluation 模型，JSONL loader； +- `backends.py`：运行模式、后端构造、生命周期及 SQLite reopen； +- `replay.py`：操作执行、真实 Event factory、确定性 Summary、故障 wrapper； +- `snapshot.py`：Session、Memory、Summary 分层快照及 canonical 表示； +- `compare.py`：字段策略、path-aware diff、allowed diff、mutation、报告组装； +- `__main__.py`：显式报告生成 CLI； +- 两个测试入口分别负责一致性与恢复能力。 + +## 5. 后端与 Summary Stack + +后端使用当前真实构造参数： + +```python +InMemorySessionService( + summarizer_manager=manager, + session_config=session_config, +) + +SqlSessionService( + db_url=f"sqlite:///{session_db_path}", + summarizer_manager=manager, + session_config=session_config, + is_async=False, +) + +SqlMemoryService( + db_url=f"sqlite:///{memory_db_path}", + is_async=False, +) +``` + +每个后端、并行 case 和 SQLite reopen 实例必须拥有独立的 Summary stack。不得共享 Manager 或运行时 cache。 + +确定性 Summarizer 覆盖真实调用点 `_compress_session_to_summary()`，根据固定 Event ID、author 和 canonical content 生成 SHA-256 摘要，不调用外部模型。它配置始终返回真的 checker 和 `auto_summarize=True`，但 replay 必须通过公开 API 触发： + +```python +await session_service.create_session_summary(session) +``` + +## 6. Profile + +### 6.1 Contract Profile + +所有参与比较的后端显式使用： + +```python +SessionServiceConfig(store_historical_events=True) +``` + +只有该 Profile 计算跨后端一致率和误报率。 + +### 6.2 Default Profile + +分别使用后端默认构造，并断言： + +1. SQL 默认保存 historical events； +2. InMemory 行为与其默认配置一致。 + +Default Profile 不参与跨后端一致率。 + +## 7. Replay Case + +公开 JSONL 至少包含以下 10 条轨迹： + +1. 单轮 user/assistant 文本； +2. 多轮连续对话； +3. 真实 `FunctionCall` 与 `FunctionResponse`； +4. State 多次写入、覆盖和临时状态持久化边界； +5. `store_session()` / `search_memory()` Memory 存取； +6. 首次 Summary； +7. Summary 更新； +8. Summary 与事件截断； +9. Memory 重复 `store_session()`； +10. 异常或重复操作轨迹。 + +每条 case 声明 required observations，包括最少 active/historical events、Memory 数量、Summary anchor、FunctionCall 和 FunctionResponse 数量。缺少必需观察结果必须失败，防止后端、Memory 或 Summary 空跑。 + +Event ID、invocation ID、request ID 和业务输入固定。Event timestamp 使用固定基准加严格递增偏移，避免同时间戳导致 SQL 排序不稳定。 + +## 8. Snapshot 契约 + +### 8.1 Session 与 Event + +严格比较 Session 作用域、事件数量和顺序、author、文本、工具调用参数和响应、State、active/historical 划分及业务可注入 ID。 + +### 8.2 Memory + +Memory 使用真实 `store_session()` 和 `search_memory()`。内容与数量通过 canonical multiset（`Counter`）比较，不能使用会吞掉重复项的 set。原始顺序保留在 diagnostics；只有 SDK 明确承诺排序时才作为严格契约。 + +### 8.3 Summary + +Summary 快照分为三层： + +- `runtime_contract`：`session_id`、文本、原始/压缩事件数、`summary_timestamp`、metadata，仅在 Manager 存活时存在； +- `persisted_projection`：Summary anchor、文本、Session 归属、active/historical 覆盖关系和摘要后新事件，Warm/Cold 严格比较； +- `harness_observations`：生成序号、操作 ID 和 lane，不属于 SDK 契约。 + +Cold reopen 不要求恢复运行时 Summary cache、计数或 runtime timestamp。 + +## 9. Summary ID 与时间策略 + +不同后端独立生成的 Summary anchor ID 只验证 UUID 结构、非空、唯一性、位置和 Summary flag；SQLite Warm/Cold 指向同一持久化事件，anchor ID 必须严格相等。 + +Runtime Summary timestamp 使用每个测试独立安装的无限确定性时钟： + +```python +ticks = count(start=1_700_001_000, step=10) +``` + +每个后端内部必须满足： + +```text +v1 timestamp 是有限正数 +v2 timestamp > v1 timestamp +v2 summary text != v1 summary text +``` + +不同后端的 runtime timestamp 不要求绝对值相等。 + +Summary anchor timestamp 是持久化 Event 字段： + +- InMemory vs SQLite：验证有限正数、位置和结构，不比较绝对值； +- SQLite Warm vs Cold：必须严格相等。 + +## 10. Normalizer、Comparator 与 Allowed Diff + +Normalizer 只统一 datetime、Pydantic/dataclass、字典键序、tuple/list 和浮点表现形式，不负责忽略差异。 + +Comparator 必须严格检查类型，并按 `ComparisonContext`（backend pair、profile、lane）选择字段策略。禁止 `.*id.*`、`.*timestamp.*` 等宽泛规则。 + +Allowed diff 必须包含受控路径、比较模式、原因、容差及适用 backend pair。报告区分： + +```text +actual_diff +allowed_diff +unsupported_contract +harness_observation +diagnostic +``` + +误报率采用 case-level 公式： + +```text +存在 unexpected actual diff 的正常 Contract case 数 +÷ 正常 Contract case 总数 +``` + +同一 case 的跨后端和 Warm/Cold 比较最多计一次。InMemory-only 模式误报率为不适用。 + +## 11. Mutation + +对真实 baseline snapshot 的副本执行： + +1. `drop_event` +2. `duplicate_event` +3. `swap_event_order` +4. `change_tool_argument` +5. `change_state_value` +6. `drop_memory` +7. `duplicate_memory` +8. `drop_summary` +9. `stale_summary_overwrite` +10. `wrong_summary_session` + +每个 mutation 必须产生符合预期路径的 diff，mutation score 必须为 10/10。后三条分别验证 Summary 丢失、覆盖错误和串 Session 的 100% 检出。 + +## 12. 操作级恢复能力 + +### 12.1 Append 响应丢失后重试 + +实际完成 append 后 wrapper 抛出模拟响应丢失，再使用同一 Event ID 重试。结果按证据互斥分类为： + +- `IDEMPOTENT`：无重试异常，目标 Event 恰好一个； +- `DUPLICATE_EVENT`：目标 ID 至少两个，并产生准确 duplicate diff； +- `RETRY_REJECTED`：重试异常存在，首次写入仍存在且目标 Event 恰好一个。 + +其他状态必须失败。 + +### 12.2 Memory 重复保存 + +重复调用 `store_session()`，通过 multiset 验证内容数量；已明确保证的幂等行为作为严格契约，否则按证据分类。 + +### 12.3 Summary 持久化失败 + +生成 v1，追加事件并生成 v2，在 `update_session()` 注入失败，分别捕获 runtime 与重新加载的 persisted projection。 + +先计算 `mixed`： + +```text +同时包含 v1/v2 投影 +或多个不同 anchor +或事件覆盖部分更新 +或 active/historical 重叠 +``` + +再进行互斥分类： + +- `OLD_SUMMARY_PRESERVED`：`not mixed`、runtime=v1、persisted=v1、anchor=1； +- `RUNTIME_PERSISTED_DIVERGENCE`：`not mixed`、runtime=v2、persisted=v1、anchor=1； +- `PARTIAL_PERSISTENCE`：`mixed`。 + +命中数量不是 1 时必须失败，不能使用兜底枚举掩盖未知状态。 + +## 13. 报告与 CLI + +普通 pytest 只向 `tmp_path` 写完整报告。仓库提交稳定的 JSON Schema 与去除时间、主机名、绝对路径、动态耗时和随机标识的示例报告。 + +Diff 至少包含： + +```text +case_id +session_id +event_index 或 summary_anchor_event_id +field_path +left/right backend +left/right value +kind +allowed diff reason +``` + +CLI： + +```bash +python -m tests.sessions.replay_consistency \ + --mode contract \ + --output replay-report.json +``` + +稳定示例只能通过显式 `--write-example-report` 更新。 + +## 14. 验收映射 + +- 默认支持 InMemory 与 SQLite，Redis 环境变量开启； +- InMemory-only 模式不依赖 SQLite、Redis 或外部模型； +- 10 条公开 replay case； +- 10 条真实 snapshot mutation 100% 检出； +- Summary 丢失、覆盖错误、串 Session 100% 检出； +- 正常 Contract case 误报率不超过 5%； +- Diff 定位到 Session、事件/Summary 和字段路径； +- SQLite Warm/Cold 使用同一数据库文件； +- 轻量模式目标耗时不超过 30 秒； +- `summary.persisted_version` 在维护者确认前显式标为 unsupported。 + +## 15. 150–300 字设计说明 + +本框架使用统一 replay case 驱动 Session、Memory 与 Summary 后端，支持 InMemory 轻量模式、InMemory/SQLite 契约模式及可选 Redis 集成模式。业务 ID 与事件时间采用确定性注入；时间精度、后端生成标识等差异通过字段级 allowed diff 说明，禁止通配忽略。Memory 使用 canonical multiset 比较，既消除无契约顺序差异，也保留重复项检测。Summary 分为运行时契约、持久化投影和框架观测三层；SQLite 同库重启严格比较摘要锚点、文本、Session 归属及事件覆盖关系。Snapshot mutation 验证差异检测率，操作级故障注入则基于互斥证据分类记录当前 SDK 的幂等和原子恢复能力。 + +## 16. 外部决策点 + +编码不受阻塞，但提交 PR 前应在 Issue #89 中向维护者确认： + +> 当前 SDK 的 `SessionSummary` 未持久化 version。是否接受使用 harness 的 `observed_generation_ordinal` 验证摘要更新顺序，并通过 Summary anchor 与事件覆盖关系验证替换语义，同时将 `summary.persisted_version` 标记为当前 SDK 不支持？ + +未得到确认前，PR 不得声称完整验证了持久化 Summary version。 From 8f506ee20b9dc11aa776d9e66046a95237ecd873 Mon Sep 17 00:00:00 2001 From: shiquan_zhan Date: Fri, 10 Jul 2026 23:32:29 +0800 Subject: [PATCH 2/2] test(sessions): add replay consistency harness --- docs/issue_89_replay_consistency_design.md | 9 + ...7-09-issue-89-replay-consistency-design.md | 325 ------- session_memory_summary_diff_report.json | 265 ++++++ .../replay_cases/issue_89_replay_cases.jsonl | 10 + tests/sessions/test_replay_consistency.py | 802 ++++++++++++++++++ 5 files changed, 1086 insertions(+), 325 deletions(-) create mode 100644 docs/issue_89_replay_consistency_design.md delete mode 100644 docs/superpowers/specs/2026-07-09-issue-89-replay-consistency-design.md create mode 100644 session_memory_summary_diff_report.json create mode 100644 tests/sessions/replay_cases/issue_89_replay_cases.jsonl create mode 100644 tests/sessions/test_replay_consistency.py diff --git a/docs/issue_89_replay_consistency_design.md b/docs/issue_89_replay_consistency_design.md new file mode 100644 index 00000000..43274470 --- /dev/null +++ b/docs/issue_89_replay_consistency_design.md @@ -0,0 +1,9 @@ +# Issue 89 Replay Consistency Design + +The replay harness loads public JSONL traces and applies the same operations to InMemory and SQLite-backed Session and Memory services. Each run writes user and assistant events, tool call and tool response parts, state deltas, memory store/search operations, and summary creation through the SDK public service APIs. The resulting backend state is converted into a normalized snapshot before comparison. + +Normalization removes non-business noise: SDK-generated event ids, raw timestamps, and JSON field ordering are not compared directly. Event order, authors, invocation ids, content parts, state deltas, memory search results, summary anchors, and summary coverage are still compared. Backend-specific tolerances are recorded as `allowed_diffs` rather than silently ignored. + +Summary comparison is split into text, metadata, and coverage checks. Summary text is compared after deterministic generation. Session ownership, anchor presence, observable revision state, replacement behavior, timestamp validity, and summary-plus-recent-event context are checked separately. The current SDK does not expose a dedicated persisted `summary.version` field, so the harness interprets version as observable revision state: v1/v2 generation order, current persisted summary text, anchor ownership, and coverage/projection consistency. + +SQLite is the default persistent backend for light mode, so no external database is required. Redis integration is environment-gated by `TRPC_AGENT_REPLAY_REDIS_URL` and skipped otherwise. The report `session_memory_summary_diff_report.json` records each case, backend pair, field path, session id, event index or summary id, and both compared values. diff --git a/docs/superpowers/specs/2026-07-09-issue-89-replay-consistency-design.md b/docs/superpowers/specs/2026-07-09-issue-89-replay-consistency-design.md deleted file mode 100644 index 98adea69..00000000 --- a/docs/superpowers/specs/2026-07-09-issue-89-replay-consistency-design.md +++ /dev/null @@ -1,325 +0,0 @@ -# Issue #89 Session / Memory 多后端回放一致性框架设计 - -## 1. 目标与边界 - -本设计为 tRPC-Agent-Python 建立可重复、可诊断、可扩展的 Session / Memory / Summary 回放一致性测试框架，回答三个相互独立的问题： - -1. 相同操作和显式配置下，不同后端的业务行为是否一致； -2. SQLite 关闭并重新创建服务后，持久化投影是否保持不变； -3. 重试、响应丢失或持久化失败后，SDK 当前表现为何，比较框架能否准确识别。 - -框架不新增生产 API，也不把测试框架观测字段伪装成 SDK 契约。当前 SDK 没有持久化的 Summary version，因此 `observed_generation_ordinal` 只能作为 harness observation；`summary.persisted_version` 在维护者确认前必须报告为 unsupported。 - -## 2. 测试层次 - -### 2.1 Replay Consistency - -使用相同轨迹和相同 `SessionServiceConfig` 驱动 InMemory 与 SQLite，并在集成模式下选择性加入 Redis。严格比较 Session、事件、State、Memory 和 Summary 的当前公开契约。 - -### 2.2 Persistence Recovery - -在同一组 SQLite Session/Memory 数据库文件上执行： - -```text -replay -> warm snapshot -> close -> reopen -> cold snapshot -``` - -Warm/Cold 只严格比较持久化投影。重新打开的 SQLite 必须使用新的 `SummarizerSessionManager`，其运行时 cache 为空。 - -### 2.3 Mutation / Capability Detection - -- Snapshot mutation 只验证比较器对人为差异的检出能力； -- 操作级故障注入用于识别当前幂等性和原子恢复能力； -- 未由 SDK 保证的理想行为不作为强制契约； -- 无法唯一分类的状态必须使测试失败。 - -## 3. 运行模式 - -统一使用 `REPLAY_MODE`： - -| 模式 | 必需后端 | 内容 | -| --- | --- | --- | -| `inmemory` | InMemory | 10 条 replay、业务不变量、10 条 mutation、运行时 Summary、性能记录 | -| `contract` | InMemory + SQLite | 跨后端比较、SQLite Warm/Cold、Default Profile、恢复能力分类 | -| `integration` | InMemory + SQLite；Redis 可选 | Contract 全部内容及 Redis 比较 | - -默认模式为 `contract`。Contract/Integration 中 SQLite 构造失败必须失败，不能 skip 或退化为单后端。只有 Redis 可以因未配置、缺少可选依赖或外部服务不可达而明确 skip。 - -InMemory 轻量模式不创建 SQLite Engine。30 秒预算默认记录，通过 `REPLAY_ENFORCE_BUDGET=1` 在专用验收中强制。 - -## 4. 工程结构 - -```text -tests/sessions/ -├── replay_consistency/ -│ ├── __init__.py -│ ├── __main__.py -│ ├── model.py -│ ├── backends.py -│ ├── replay.py -│ ├── snapshot.py -│ ├── compare.py -│ ├── replay_cases.jsonl -│ ├── report.schema.json -│ ├── example_report.json -│ └── README.md -├── test_replay_consistency.py -└── test_replay_recovery.py -``` - -职责如下： - -- `model.py`：case、operation、required observation、diff 和 evaluation 模型，JSONL loader； -- `backends.py`：运行模式、后端构造、生命周期及 SQLite reopen； -- `replay.py`：操作执行、真实 Event factory、确定性 Summary、故障 wrapper； -- `snapshot.py`：Session、Memory、Summary 分层快照及 canonical 表示； -- `compare.py`：字段策略、path-aware diff、allowed diff、mutation、报告组装； -- `__main__.py`：显式报告生成 CLI； -- 两个测试入口分别负责一致性与恢复能力。 - -## 5. 后端与 Summary Stack - -后端使用当前真实构造参数： - -```python -InMemorySessionService( - summarizer_manager=manager, - session_config=session_config, -) - -SqlSessionService( - db_url=f"sqlite:///{session_db_path}", - summarizer_manager=manager, - session_config=session_config, - is_async=False, -) - -SqlMemoryService( - db_url=f"sqlite:///{memory_db_path}", - is_async=False, -) -``` - -每个后端、并行 case 和 SQLite reopen 实例必须拥有独立的 Summary stack。不得共享 Manager 或运行时 cache。 - -确定性 Summarizer 覆盖真实调用点 `_compress_session_to_summary()`，根据固定 Event ID、author 和 canonical content 生成 SHA-256 摘要，不调用外部模型。它配置始终返回真的 checker 和 `auto_summarize=True`，但 replay 必须通过公开 API 触发： - -```python -await session_service.create_session_summary(session) -``` - -## 6. Profile - -### 6.1 Contract Profile - -所有参与比较的后端显式使用： - -```python -SessionServiceConfig(store_historical_events=True) -``` - -只有该 Profile 计算跨后端一致率和误报率。 - -### 6.2 Default Profile - -分别使用后端默认构造，并断言： - -1. SQL 默认保存 historical events； -2. InMemory 行为与其默认配置一致。 - -Default Profile 不参与跨后端一致率。 - -## 7. Replay Case - -公开 JSONL 至少包含以下 10 条轨迹： - -1. 单轮 user/assistant 文本； -2. 多轮连续对话； -3. 真实 `FunctionCall` 与 `FunctionResponse`； -4. State 多次写入、覆盖和临时状态持久化边界； -5. `store_session()` / `search_memory()` Memory 存取； -6. 首次 Summary； -7. Summary 更新； -8. Summary 与事件截断； -9. Memory 重复 `store_session()`； -10. 异常或重复操作轨迹。 - -每条 case 声明 required observations，包括最少 active/historical events、Memory 数量、Summary anchor、FunctionCall 和 FunctionResponse 数量。缺少必需观察结果必须失败，防止后端、Memory 或 Summary 空跑。 - -Event ID、invocation ID、request ID 和业务输入固定。Event timestamp 使用固定基准加严格递增偏移，避免同时间戳导致 SQL 排序不稳定。 - -## 8. Snapshot 契约 - -### 8.1 Session 与 Event - -严格比较 Session 作用域、事件数量和顺序、author、文本、工具调用参数和响应、State、active/historical 划分及业务可注入 ID。 - -### 8.2 Memory - -Memory 使用真实 `store_session()` 和 `search_memory()`。内容与数量通过 canonical multiset（`Counter`）比较，不能使用会吞掉重复项的 set。原始顺序保留在 diagnostics；只有 SDK 明确承诺排序时才作为严格契约。 - -### 8.3 Summary - -Summary 快照分为三层： - -- `runtime_contract`：`session_id`、文本、原始/压缩事件数、`summary_timestamp`、metadata，仅在 Manager 存活时存在； -- `persisted_projection`：Summary anchor、文本、Session 归属、active/historical 覆盖关系和摘要后新事件，Warm/Cold 严格比较； -- `harness_observations`：生成序号、操作 ID 和 lane，不属于 SDK 契约。 - -Cold reopen 不要求恢复运行时 Summary cache、计数或 runtime timestamp。 - -## 9. Summary ID 与时间策略 - -不同后端独立生成的 Summary anchor ID 只验证 UUID 结构、非空、唯一性、位置和 Summary flag；SQLite Warm/Cold 指向同一持久化事件，anchor ID 必须严格相等。 - -Runtime Summary timestamp 使用每个测试独立安装的无限确定性时钟： - -```python -ticks = count(start=1_700_001_000, step=10) -``` - -每个后端内部必须满足： - -```text -v1 timestamp 是有限正数 -v2 timestamp > v1 timestamp -v2 summary text != v1 summary text -``` - -不同后端的 runtime timestamp 不要求绝对值相等。 - -Summary anchor timestamp 是持久化 Event 字段： - -- InMemory vs SQLite：验证有限正数、位置和结构，不比较绝对值； -- SQLite Warm vs Cold：必须严格相等。 - -## 10. Normalizer、Comparator 与 Allowed Diff - -Normalizer 只统一 datetime、Pydantic/dataclass、字典键序、tuple/list 和浮点表现形式，不负责忽略差异。 - -Comparator 必须严格检查类型，并按 `ComparisonContext`（backend pair、profile、lane）选择字段策略。禁止 `.*id.*`、`.*timestamp.*` 等宽泛规则。 - -Allowed diff 必须包含受控路径、比较模式、原因、容差及适用 backend pair。报告区分： - -```text -actual_diff -allowed_diff -unsupported_contract -harness_observation -diagnostic -``` - -误报率采用 case-level 公式： - -```text -存在 unexpected actual diff 的正常 Contract case 数 -÷ 正常 Contract case 总数 -``` - -同一 case 的跨后端和 Warm/Cold 比较最多计一次。InMemory-only 模式误报率为不适用。 - -## 11. Mutation - -对真实 baseline snapshot 的副本执行： - -1. `drop_event` -2. `duplicate_event` -3. `swap_event_order` -4. `change_tool_argument` -5. `change_state_value` -6. `drop_memory` -7. `duplicate_memory` -8. `drop_summary` -9. `stale_summary_overwrite` -10. `wrong_summary_session` - -每个 mutation 必须产生符合预期路径的 diff，mutation score 必须为 10/10。后三条分别验证 Summary 丢失、覆盖错误和串 Session 的 100% 检出。 - -## 12. 操作级恢复能力 - -### 12.1 Append 响应丢失后重试 - -实际完成 append 后 wrapper 抛出模拟响应丢失，再使用同一 Event ID 重试。结果按证据互斥分类为： - -- `IDEMPOTENT`：无重试异常，目标 Event 恰好一个； -- `DUPLICATE_EVENT`：目标 ID 至少两个，并产生准确 duplicate diff； -- `RETRY_REJECTED`：重试异常存在，首次写入仍存在且目标 Event 恰好一个。 - -其他状态必须失败。 - -### 12.2 Memory 重复保存 - -重复调用 `store_session()`，通过 multiset 验证内容数量；已明确保证的幂等行为作为严格契约，否则按证据分类。 - -### 12.3 Summary 持久化失败 - -生成 v1，追加事件并生成 v2，在 `update_session()` 注入失败，分别捕获 runtime 与重新加载的 persisted projection。 - -先计算 `mixed`： - -```text -同时包含 v1/v2 投影 -或多个不同 anchor -或事件覆盖部分更新 -或 active/historical 重叠 -``` - -再进行互斥分类： - -- `OLD_SUMMARY_PRESERVED`：`not mixed`、runtime=v1、persisted=v1、anchor=1； -- `RUNTIME_PERSISTED_DIVERGENCE`：`not mixed`、runtime=v2、persisted=v1、anchor=1； -- `PARTIAL_PERSISTENCE`：`mixed`。 - -命中数量不是 1 时必须失败，不能使用兜底枚举掩盖未知状态。 - -## 13. 报告与 CLI - -普通 pytest 只向 `tmp_path` 写完整报告。仓库提交稳定的 JSON Schema 与去除时间、主机名、绝对路径、动态耗时和随机标识的示例报告。 - -Diff 至少包含： - -```text -case_id -session_id -event_index 或 summary_anchor_event_id -field_path -left/right backend -left/right value -kind -allowed diff reason -``` - -CLI： - -```bash -python -m tests.sessions.replay_consistency \ - --mode contract \ - --output replay-report.json -``` - -稳定示例只能通过显式 `--write-example-report` 更新。 - -## 14. 验收映射 - -- 默认支持 InMemory 与 SQLite，Redis 环境变量开启； -- InMemory-only 模式不依赖 SQLite、Redis 或外部模型； -- 10 条公开 replay case； -- 10 条真实 snapshot mutation 100% 检出； -- Summary 丢失、覆盖错误、串 Session 100% 检出； -- 正常 Contract case 误报率不超过 5%； -- Diff 定位到 Session、事件/Summary 和字段路径； -- SQLite Warm/Cold 使用同一数据库文件； -- 轻量模式目标耗时不超过 30 秒； -- `summary.persisted_version` 在维护者确认前显式标为 unsupported。 - -## 15. 150–300 字设计说明 - -本框架使用统一 replay case 驱动 Session、Memory 与 Summary 后端，支持 InMemory 轻量模式、InMemory/SQLite 契约模式及可选 Redis 集成模式。业务 ID 与事件时间采用确定性注入；时间精度、后端生成标识等差异通过字段级 allowed diff 说明，禁止通配忽略。Memory 使用 canonical multiset 比较，既消除无契约顺序差异，也保留重复项检测。Summary 分为运行时契约、持久化投影和框架观测三层；SQLite 同库重启严格比较摘要锚点、文本、Session 归属及事件覆盖关系。Snapshot mutation 验证差异检测率，操作级故障注入则基于互斥证据分类记录当前 SDK 的幂等和原子恢复能力。 - -## 16. 外部决策点 - -编码不受阻塞，但提交 PR 前应在 Issue #89 中向维护者确认： - -> 当前 SDK 的 `SessionSummary` 未持久化 version。是否接受使用 harness 的 `observed_generation_ordinal` 验证摘要更新顺序，并通过 Summary anchor 与事件覆盖关系验证替换语义，同时将 `summary.persisted_version` 标记为当前 SDK 不支持？ - -未得到确认前，PR 不得声称完整验证了持久化 Summary version。 diff --git a/session_memory_summary_diff_report.json b/session_memory_summary_diff_report.json new file mode 100644 index 00000000..606120c0 --- /dev/null +++ b/session_memory_summary_diff_report.json @@ -0,0 +1,265 @@ +{ + "comparisons": [ + { + "actual_backend": "sqlite", + "allowed_diffs": [ + { + "allowed": true, + "case_id": "single_turn_text", + "field_path": "sessions.*.events.*.id", + "reason": "event ids are SDK-generated unless fixed by the replay case" + }, + { + "allowed": true, + "case_id": "single_turn_text", + "field_path": "sessions.*.events.*.timestamp", + "reason": "event timestamps are normalized away as non-business fields" + }, + { + "allowed": true, + "case_id": "single_turn_text", + "field_path": "sessions.*.summary.anchor_timestamp", + "reason": "summary timestamps are compared by validity and order, not cross-backend equality" + } + ], + "case_id": "single_turn_text", + "diffs": [], + "expected_backend": "inmemory" + }, + { + "actual_backend": "sqlite", + "allowed_diffs": [ + { + "allowed": true, + "case_id": "multi_turn_text", + "field_path": "sessions.*.events.*.id", + "reason": "event ids are SDK-generated unless fixed by the replay case" + }, + { + "allowed": true, + "case_id": "multi_turn_text", + "field_path": "sessions.*.events.*.timestamp", + "reason": "event timestamps are normalized away as non-business fields" + }, + { + "allowed": true, + "case_id": "multi_turn_text", + "field_path": "sessions.*.summary.anchor_timestamp", + "reason": "summary timestamps are compared by validity and order, not cross-backend equality" + } + ], + "case_id": "multi_turn_text", + "diffs": [], + "expected_backend": "inmemory" + }, + { + "actual_backend": "sqlite", + "allowed_diffs": [ + { + "allowed": true, + "case_id": "tool_call_response", + "field_path": "sessions.*.events.*.id", + "reason": "event ids are SDK-generated unless fixed by the replay case" + }, + { + "allowed": true, + "case_id": "tool_call_response", + "field_path": "sessions.*.events.*.timestamp", + "reason": "event timestamps are normalized away as non-business fields" + }, + { + "allowed": true, + "case_id": "tool_call_response", + "field_path": "sessions.*.summary.anchor_timestamp", + "reason": "summary timestamps are compared by validity and order, not cross-backend equality" + } + ], + "case_id": "tool_call_response", + "diffs": [], + "expected_backend": "inmemory" + }, + { + "actual_backend": "sqlite", + "allowed_diffs": [ + { + "allowed": true, + "case_id": "state_overwrite", + "field_path": "sessions.*.events.*.id", + "reason": "event ids are SDK-generated unless fixed by the replay case" + }, + { + "allowed": true, + "case_id": "state_overwrite", + "field_path": "sessions.*.events.*.timestamp", + "reason": "event timestamps are normalized away as non-business fields" + }, + { + "allowed": true, + "case_id": "state_overwrite", + "field_path": "sessions.*.summary.anchor_timestamp", + "reason": "summary timestamps are compared by validity and order, not cross-backend equality" + } + ], + "case_id": "state_overwrite", + "diffs": [], + "expected_backend": "inmemory" + }, + { + "actual_backend": "sqlite", + "allowed_diffs": [ + { + "allowed": true, + "case_id": "memory_write_read", + "field_path": "sessions.*.events.*.id", + "reason": "event ids are SDK-generated unless fixed by the replay case" + }, + { + "allowed": true, + "case_id": "memory_write_read", + "field_path": "sessions.*.events.*.timestamp", + "reason": "event timestamps are normalized away as non-business fields" + }, + { + "allowed": true, + "case_id": "memory_write_read", + "field_path": "sessions.*.summary.anchor_timestamp", + "reason": "summary timestamps are compared by validity and order, not cross-backend equality" + } + ], + "case_id": "memory_write_read", + "diffs": [], + "expected_backend": "inmemory" + }, + { + "actual_backend": "sqlite", + "allowed_diffs": [ + { + "allowed": true, + "case_id": "summary_generation", + "field_path": "sessions.*.events.*.id", + "reason": "event ids are SDK-generated unless fixed by the replay case" + }, + { + "allowed": true, + "case_id": "summary_generation", + "field_path": "sessions.*.events.*.timestamp", + "reason": "event timestamps are normalized away as non-business fields" + }, + { + "allowed": true, + "case_id": "summary_generation", + "field_path": "sessions.*.summary.anchor_timestamp", + "reason": "summary timestamps are compared by validity and order, not cross-backend equality" + } + ], + "case_id": "summary_generation", + "diffs": [], + "expected_backend": "inmemory" + }, + { + "actual_backend": "sqlite", + "allowed_diffs": [ + { + "allowed": true, + "case_id": "summary_update", + "field_path": "sessions.*.events.*.id", + "reason": "event ids are SDK-generated unless fixed by the replay case" + }, + { + "allowed": true, + "case_id": "summary_update", + "field_path": "sessions.*.events.*.timestamp", + "reason": "event timestamps are normalized away as non-business fields" + }, + { + "allowed": true, + "case_id": "summary_update", + "field_path": "sessions.*.summary.anchor_timestamp", + "reason": "summary timestamps are compared by validity and order, not cross-backend equality" + } + ], + "case_id": "summary_update", + "diffs": [], + "expected_backend": "inmemory" + }, + { + "actual_backend": "sqlite", + "allowed_diffs": [ + { + "allowed": true, + "case_id": "summary_truncation_context", + "field_path": "sessions.*.events.*.id", + "reason": "event ids are SDK-generated unless fixed by the replay case" + }, + { + "allowed": true, + "case_id": "summary_truncation_context", + "field_path": "sessions.*.events.*.timestamp", + "reason": "event timestamps are normalized away as non-business fields" + }, + { + "allowed": true, + "case_id": "summary_truncation_context", + "field_path": "sessions.*.summary.anchor_timestamp", + "reason": "summary timestamps are compared by validity and order, not cross-backend equality" + } + ], + "case_id": "summary_truncation_context", + "diffs": [], + "expected_backend": "inmemory" + }, + { + "actual_backend": "sqlite", + "allowed_diffs": [ + { + "allowed": true, + "case_id": "duplicate_write_recovery", + "field_path": "sessions.*.events.*.id", + "reason": "event ids are SDK-generated unless fixed by the replay case" + }, + { + "allowed": true, + "case_id": "duplicate_write_recovery", + "field_path": "sessions.*.events.*.timestamp", + "reason": "event timestamps are normalized away as non-business fields" + }, + { + "allowed": true, + "case_id": "duplicate_write_recovery", + "field_path": "sessions.*.summary.anchor_timestamp", + "reason": "summary timestamps are compared by validity and order, not cross-backend equality" + } + ], + "case_id": "duplicate_write_recovery", + "diffs": [], + "expected_backend": "inmemory" + }, + { + "actual_backend": "sqlite", + "allowed_diffs": [ + { + "allowed": true, + "case_id": "wrong_session_recovery", + "field_path": "sessions.*.events.*.id", + "reason": "event ids are SDK-generated unless fixed by the replay case" + }, + { + "allowed": true, + "case_id": "wrong_session_recovery", + "field_path": "sessions.*.events.*.timestamp", + "reason": "event timestamps are normalized away as non-business fields" + }, + { + "allowed": true, + "case_id": "wrong_session_recovery", + "field_path": "sessions.*.summary.anchor_timestamp", + "reason": "summary timestamps are compared by validity and order, not cross-backend equality" + } + ], + "case_id": "wrong_session_recovery", + "diffs": [], + "expected_backend": "inmemory" + } + ], + "generated_at": "deterministic-test-run" +} \ No newline at end of file diff --git a/tests/sessions/replay_cases/issue_89_replay_cases.jsonl b/tests/sessions/replay_cases/issue_89_replay_cases.jsonl new file mode 100644 index 00000000..a65c1625 --- /dev/null +++ b/tests/sessions/replay_cases/issue_89_replay_cases.jsonl @@ -0,0 +1,10 @@ +{"case_id":"single_turn_text","description":"Single user input and agent text output.","operations":[{"op":"event","author":"user","text":"hello agent"},{"op":"event","author":"assistant","text":"hello user"}]} +{"case_id":"multi_turn_text","description":"Multiple user and assistant turns appended to one session.","operations":[{"op":"event","author":"user","text":"turn one question"},{"op":"event","author":"assistant","text":"turn one answer"},{"op":"event","author":"user","text":"turn two question"},{"op":"event","author":"assistant","text":"turn two answer"}]} +{"case_id":"tool_call_response","description":"Assistant function_call followed by tool function_response and final text.","operations":[{"op":"event","author":"user","text":"weather in shenzhen"},{"op":"tool_call","author":"assistant","call_id":"call-weather","name":"lookup_weather","args":{"city":"shenzhen"}},{"op":"tool_response","author":"tool","call_id":"call-weather","name":"lookup_weather","response":{"forecast":"sunny","temperature_c":30}},{"op":"event","author":"assistant","text":"shenzhen is sunny at 30c"}]} +{"case_id":"state_overwrite","description":"Repeated session state writes and overwrite of the same key.","operations":[{"op":"state","key":"topic","value":"travel"},{"op":"state","key":"topic","value":"finance"},{"op":"state","key":"step","value":2},{"op":"event","author":"assistant","text":"state updated"}]} +{"case_id":"memory_write_read","description":"Store preference and fact memories, then search them.","operations":[{"op":"event","author":"user","text":"memory preference tea oolong"},{"op":"memory_store"},{"op":"memory_search","query":"oolong"},{"op":"event","author":"user","text":"memory fact project codename river"},{"op":"memory_store"},{"op":"memory_search","query":"river"}]} +{"case_id":"summary_generation","description":"Long conversation creates a v1 summary.","operations":[{"op":"event","author":"user","text":"summary seed one has enough context"},{"op":"event","author":"assistant","text":"summary response one has enough context"},{"op":"event","author":"user","text":"summary seed two has enough context"},{"op":"event","author":"assistant","text":"summary response two has enough context"},{"op":"event","author":"user","text":"summary seed three has enough context"},{"op":"summary","expected_revision":"v1"}]} +{"case_id":"summary_update","description":"A second summary generation replaces the previous observable revision.","operations":[{"op":"event","author":"user","text":"summary update one has enough context"},{"op":"event","author":"assistant","text":"summary update answer one has enough context"},{"op":"event","author":"user","text":"summary update two has enough context"},{"op":"event","author":"assistant","text":"summary update answer two has enough context"},{"op":"summary","expected_revision":"v1"},{"op":"event","author":"user","text":"summary update three has enough context"},{"op":"event","author":"assistant","text":"summary update answer three has enough context"},{"op":"event","author":"user","text":"summary update four has enough context"},{"op":"summary","expected_revision":"v2"}]} +{"case_id":"summary_truncation_context","description":"Summary anchor plus retained events and new events restore active context.","operations":[{"op":"event","author":"user","text":"context alpha decision keep"},{"op":"event","author":"assistant","text":"context beta answer keep"},{"op":"event","author":"user","text":"context gamma followup keep"},{"op":"event","author":"assistant","text":"context delta answer keep"},{"op":"event","author":"user","text":"context epsilon recent keep"},{"op":"summary","expected_revision":"v1"},{"op":"event","author":"assistant","text":"post summary continuation"}]} +{"case_id":"duplicate_write_recovery","description":"Repeated session and memory writes should not create replay-visible duplicates.","operations":[{"op":"event","event_id":"duplicate-event-1","author":"user","text":"duplicate recovery input"},{"op":"event","author":"assistant","text":"duplicate recovery output"},{"op":"duplicate_update"},{"op":"memory_store"},{"op":"memory_store"},{"op":"memory_search","query":"duplicate"}]} +{"case_id":"wrong_session_recovery","description":"Two sessions keep summary ownership isolated.","operations":[{"op":"event","author":"user","text":"primary session content one"},{"op":"event","author":"assistant","text":"primary session content two"},{"op":"event","author":"user","text":"primary session content three"},{"op":"summary","expected_revision":"v1"},{"op":"other_session_event","session_id":"other-session","author":"user","text":"other session content one"},{"op":"other_session_event","session_id":"other-session","author":"assistant","text":"other session content two"},{"op":"other_session_event","session_id":"other-session","author":"user","text":"other session content three"},{"op":"other_session_summary","session_id":"other-session","expected_revision":"v1"}]} diff --git a/tests/sessions/test_replay_consistency.py b/tests/sessions/test_replay_consistency.py new file mode 100644 index 00000000..14475b29 --- /dev/null +++ b/tests/sessions/test_replay_consistency.py @@ -0,0 +1,802 @@ +# Tencent is pleased to support the open source community by making tRPC-Agent-Python available. +# +# Copyright (C) 2026 Tencent. All rights reserved. +# +# tRPC-Agent-Python is licensed under Apache-2.0. +"""Replay consistency harness for Session, Memory, and Summary backends.""" + +from __future__ import annotations + +import copy +import json +import os +import re +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Callable + +import pytest + +from trpc_agent_sdk.abc import MemoryServiceConfig +from trpc_agent_sdk.events import Event +from trpc_agent_sdk.memory import InMemoryMemoryService, SqlMemoryService +from trpc_agent_sdk.sessions import InMemorySessionService, Session, SessionServiceConfig +from trpc_agent_sdk.sessions import SessionSummarizer, SqlSessionService, SummarizerSessionManager +from trpc_agent_sdk.types import Content, EventActions, FunctionCall, FunctionResponse, Part + +APP_NAME = "replay-app" +USER_ID = "replay-user" +REPORT_PATH = Path("session_memory_summary_diff_report.json") +CASES_PATH = Path(__file__).with_name("replay_cases") / "issue_89_replay_cases.jsonl" +SUMMARY_PREFIX = "Previous conversation summary: " + + +class _FakeSummaryModel: + name = "deterministic-summary-model" + + +class _DeterministicSummarizer(SessionSummarizer): + """A real SessionSummarizer subclass with deterministic text generation.""" + + def __init__(self) -> None: + super().__init__( + model=_FakeSummaryModel(), + check_summarizer_functions=[lambda session: bool(session.events)], + keep_recent_count=2, + ) + self._revision_by_session: dict[str, int] = {} + + async def _compress_session_to_summary( + self, + events: list[Event], + session_id: str, + ctx=None, + ) -> str: + revision = self._revision_by_session.get(session_id, 0) + 1 + self._revision_by_session[session_id] = revision + covered = "|".join(_event_text(event) for event in events if _event_text(event)) + return f"{session_id} summary revision v{revision}: {covered}" + + +@dataclass +class _Backend: + name: str + session_service: Any + memory_service: Any + db_url: str | None = None + + +@dataclass +class _ReplayContext: + backend: _Backend + sessions: dict[str, Session] + memory_searches: list[dict[str, Any]] = field(default_factory=list) + summary_generations: dict[str, int] = field(default_factory=dict) + + +def _load_replay_cases() -> list[dict[str, Any]]: + with CASES_PATH.open(encoding="utf-8") as handle: + return [json.loads(line) for line in handle if line.strip()] + + +def _memory_config() -> MemoryServiceConfig: + config = MemoryServiceConfig(enabled=True) + config.clean_ttl_config() + return config + + +def _session_config() -> SessionServiceConfig: + config = SessionServiceConfig(store_historical_events=True) + config.clean_ttl_config() + return config + + +def _summarizer_manager() -> SummarizerSessionManager: + return SummarizerSessionManager( + model=_FakeSummaryModel(), + summarizer=_DeterministicSummarizer(), + auto_summarize=True, + ) + + +async def _build_inmemory_backend() -> _Backend: + return _Backend( + name="inmemory", + session_service=InMemorySessionService( + session_config=_session_config(), + summarizer_manager=_summarizer_manager(), + ), + memory_service=InMemoryMemoryService(memory_service_config=_memory_config()), + ) + + +async def _build_sqlite_backend(tmp_path: Path) -> _Backend: + tmp_path.mkdir(parents=True, exist_ok=True) + db_url = f"sqlite:///{(tmp_path / 'replay-consistency.db').as_posix()}" + session_service = SqlSessionService( + db_url=db_url, + session_config=_session_config(), + summarizer_manager=_summarizer_manager(), + is_async=False, + ) + memory_service = SqlMemoryService( + db_url=db_url, + is_async=False, + memory_service_config=_memory_config(), + ) + await session_service._sql_storage.create_sql_engine() + await memory_service._sql_storage.create_sql_engine() + return _Backend( + name="sqlite", + session_service=session_service, + memory_service=memory_service, + db_url=db_url, + ) + + +async def _close_backend(backend: _Backend) -> None: + await backend.memory_service.close() + await backend.session_service.close() + + +async def _reopen_sqlite_session_backend(db_url: str) -> _Backend: + session_service = SqlSessionService( + db_url=db_url, + session_config=_session_config(), + summarizer_manager=_summarizer_manager(), + is_async=False, + ) + await session_service._sql_storage.create_sql_engine() + return _Backend(name="sqlite-cold", session_service=session_service, memory_service=None, db_url=db_url) + + +async def _ensure_session(ctx: _ReplayContext, session_id: str) -> Session: + if session_id not in ctx.sessions: + await ctx.backend.session_service.create_session( + app_name=APP_NAME, + user_id=USER_ID, + session_id=session_id, + ) + ctx.sessions[session_id] = await _refresh_session(ctx, session_id) + return ctx.sessions[session_id] + + +async def _refresh_session(ctx: _ReplayContext, session_id: str) -> Session: + session = await ctx.backend.session_service.get_session( + app_name=APP_NAME, + user_id=USER_ID, + session_id=session_id, + ) + assert session is not None + return session + + +async def _append_event(ctx: _ReplayContext, session_id: str, event: Event) -> None: + session = await _ensure_session(ctx, session_id) + await ctx.backend.session_service.append_event(session, event) + ctx.sessions[session_id] = await _refresh_session(ctx, session_id) + + +def _text_event( + *, + author: str, + text: str, + invocation_id: str, + event_id: str | None = None, + state_delta: dict[str, Any] | None = None, + timestamp: float | None = None, +) -> Event: + return Event( + id=event_id or "", + invocation_id=invocation_id, + author=author, + content=Content(parts=[Part.from_text(text=text)]), + actions=EventActions(state_delta=state_delta or {}), + timestamp=timestamp or _replay_timestamp(0), + ) + + +def _tool_call_event(operation: dict[str, Any], invocation_id: str, timestamp: float) -> Event: + return Event( + invocation_id=invocation_id, + author=operation["author"], + timestamp=timestamp, + content=Content( + parts=[ + Part( + function_call=FunctionCall( + id=operation["call_id"], + name=operation["name"], + args=operation["args"], + ) + ) + ] + ), + ) + + +def _tool_response_event(operation: dict[str, Any], invocation_id: str, timestamp: float) -> Event: + return Event( + invocation_id=invocation_id, + author=operation["author"], + timestamp=timestamp, + content=Content( + parts=[ + Part( + function_response=FunctionResponse( + id=operation["call_id"], + name=operation["name"], + response=operation["response"], + ) + ) + ] + ), + ) + + +def _replay_timestamp(index: int) -> float: + return 2_000_000_000.0 + index + + +async def _run_case(case: dict[str, Any], backend: _Backend) -> dict[str, Any]: + primary_session_id = f"{case['case_id']}-session" + ctx = _ReplayContext(backend=backend, sessions={}) + await _ensure_session(ctx, primary_session_id) + + for index, operation in enumerate(case["operations"]): + invocation_id = f"{case['case_id']}-inv-{index}" + op = operation["op"] + if op == "event": + event = _text_event( + event_id=operation.get("event_id"), + invocation_id=invocation_id, + author=operation["author"], + text=operation["text"], + timestamp=_replay_timestamp(index), + ) + await _append_event(ctx, primary_session_id, event) + elif op == "tool_call": + await _append_event( + ctx, + primary_session_id, + _tool_call_event(operation, invocation_id, _replay_timestamp(index)), + ) + elif op == "tool_response": + await _append_event( + ctx, + primary_session_id, + _tool_response_event(operation, invocation_id, _replay_timestamp(index)), + ) + elif op == "state": + event = _text_event( + invocation_id=invocation_id, + author="system", + text=f"state {operation['key']} updated", + state_delta={operation["key"]: operation["value"]}, + timestamp=_replay_timestamp(index), + ) + await _append_event(ctx, primary_session_id, event) + elif op == "memory_store": + await ctx.backend.memory_service.store_session(ctx.sessions[primary_session_id]) + elif op == "memory_search": + response = await ctx.backend.memory_service.search_memory( + f"{APP_NAME}/{USER_ID}", + operation["query"], + limit=10, + ) + results = [_normalize_memory_entry(memory.model_dump(mode="json")) for memory in response.memories] + ctx.memory_searches.append( + { + "query": operation["query"], + "results": sorted(results, key=lambda item: json.dumps(item, sort_keys=True)), + } + ) + elif op == "summary": + await ctx.backend.session_service.create_session_summary(ctx.sessions[primary_session_id]) + ctx.sessions[primary_session_id] = await _refresh_session(ctx, primary_session_id) + ctx.summary_generations[primary_session_id] = ctx.summary_generations.get(primary_session_id, 0) + 1 + elif op == "duplicate_update": + await ctx.backend.session_service.update_session(ctx.sessions[primary_session_id]) + ctx.sessions[primary_session_id] = await _refresh_session(ctx, primary_session_id) + elif op == "other_session_event": + session_id = operation["session_id"] + event = _text_event( + invocation_id=invocation_id, + author=operation["author"], + text=operation["text"], + timestamp=_replay_timestamp(index), + ) + await _append_event(ctx, session_id, event) + elif op == "other_session_summary": + session_id = operation["session_id"] + await ctx.backend.session_service.create_session_summary(ctx.sessions[session_id]) + ctx.sessions[session_id] = await _refresh_session(ctx, session_id) + ctx.summary_generations[session_id] = ctx.summary_generations.get(session_id, 0) + 1 + else: + raise AssertionError(f"unknown replay operation: {op}") + + return await _snapshot(case["case_id"], ctx) + + +async def _snapshot(case_id: str, ctx: _ReplayContext) -> dict[str, Any]: + sessions = {} + for session_id in sorted(ctx.sessions): + reloaded = await ctx.backend.session_service.get_session( + app_name=APP_NAME, + user_id=USER_ID, + session_id=session_id, + ) + assert reloaded is not None + runtime_summary = await ctx.backend.session_service.get_session_summary(reloaded) + sessions[session_id] = _normalize_session( + reloaded, + runtime_summary=runtime_summary, + observed_generation_ordinal=ctx.summary_generations.get(session_id, 0), + ) + return { + "case_id": case_id, + "sessions": sessions, + "memory": { + "searches": sorted(ctx.memory_searches, key=lambda item: item["query"]), + }, + } + + +async def _cold_sqlite_snapshot(case_id: str, warm_backend: _Backend, warm_snapshot: dict[str, Any]) -> dict[str, Any]: + assert warm_backend.db_url is not None + cold_backend = await _reopen_sqlite_session_backend(warm_backend.db_url) + try: + sessions = {} + for session_id, warm_session in warm_snapshot["sessions"].items(): + reloaded = await cold_backend.session_service.get_session( + app_name=APP_NAME, + user_id=USER_ID, + session_id=session_id, + ) + assert reloaded is not None + sessions[session_id] = _normalize_session( + reloaded, + runtime_summary=None, + observed_generation_ordinal=warm_session["summary"]["observed_generation_ordinal"], + ) + return { + "case_id": case_id, + "sessions": sessions, + "memory": warm_snapshot["memory"], + } + finally: + await cold_backend.session_service.close() + + +def _normalize_session( + session: Session, + *, + runtime_summary: str | None, + observed_generation_ordinal: int, +) -> dict[str, Any]: + active_events = [_normalize_event(event, index) for index, event in enumerate(session.events)] + historical_events = [_normalize_event(event, index) for index, event in enumerate(session.historical_events)] + active_summary_anchors = [event for event in session.events if event.is_summary_event()] + historical_summary_anchors = [event for event in session.historical_events if event.is_summary_event()] + all_summary_anchors = active_summary_anchors + historical_summary_anchors + current_anchor = active_summary_anchors[0] if active_summary_anchors else None + current_text = _summary_text(current_anchor) if current_anchor else None + return { + "session_id": session.id, + "state": _stable_json(session.state), + "events": active_events, + "historical_events": historical_events, + "summary": { + "runtime_text": runtime_summary, + "persisted_text": current_text, + "current_revision": _summary_revision(current_text), + "observed_generation_ordinal": observed_generation_ordinal, + "persisted_version": { + "status": "unsupported", + "reason": "SDK exposes observable summary revision state, not a dedicated persisted version field.", + }, + "anchor_id": current_anchor.id if current_anchor else None, + "anchor_timestamp": current_anchor.timestamp if current_anchor else None, + "anchor_count": len(all_summary_anchors), + "active_anchor_count": len(active_summary_anchors), + "historical_anchor_count": len(historical_summary_anchors), + "coverage": { + "historical_event_count": len(session.historical_events), + "active_event_count": len(session.events), + "has_summary_anchor": current_anchor is not None, + "has_recent_events_after_summary": bool(current_anchor and len(session.events) > 1), + }, + "session_owner": session.id if current_anchor else None, + }, + } + + +def _normalize_event(event: Event, index: int) -> dict[str, Any]: + return { + "event_index": index, + "author": event.author, + "invocation_id": event.invocation_id, + "parts": [_normalize_part(part) for part in (event.content.parts if event.content else [])], + "state_delta": _stable_json(event.actions.state_delta if event.actions else {}), + "is_summary": event.is_summary_event(), + "version": event.version, + } + + +def _normalize_part(part: Part) -> dict[str, Any]: + if part.text is not None: + return {"text": part.text} + if part.function_call is not None: + return { + "function_call": { + "id": part.function_call.id, + "name": part.function_call.name, + "args": _stable_json(part.function_call.args), + } + } + if part.function_response is not None: + return { + "function_response": { + "id": part.function_response.id, + "name": part.function_response.name, + "response": _stable_json(part.function_response.response), + } + } + return {} + + +def _normalize_memory_entry(entry: dict[str, Any]) -> dict[str, Any]: + parts = entry.get("content", {}).get("parts", []) + return { + "author": entry.get("author"), + "parts": [_stable_json(part) for part in parts], + } + + +def _stable_json(value: Any) -> Any: + return json.loads(json.dumps(value or {}, sort_keys=True, default=str)) + + +def _event_text(event: Event) -> str: + if not event.content or not event.content.parts: + return "" + return "".join(part.text or "" for part in event.content.parts) + + +def _summary_text(event: Event | None) -> str | None: + text = _event_text(event) if event else "" + if text.startswith(SUMMARY_PREFIX): + return text[len(SUMMARY_PREFIX):] + return text or None + + +def _summary_revision(summary_text: str | None) -> str | None: + if not summary_text: + return None + match = re.search(r"\bv\d+\b", summary_text) + return match.group(0) if match else None + + +def _compare_snapshots( + case_id: str, + expected: dict[str, Any], + actual: dict[str, Any], + *, + expected_backend: str, + actual_backend: str, +) -> list[dict[str, Any]]: + diffs: list[dict[str, Any]] = [] + _compare_value(case_id, expected, actual, "", diffs, expected_backend, actual_backend) + return diffs + + +def _compare_value( + case_id: str, + expected: Any, + actual: Any, + path: str, + diffs: list[dict[str, Any]], + expected_backend: str, + actual_backend: str, +) -> None: + if _is_allowed_normalized_difference(path): + return + if expected == actual: + return + if isinstance(expected, dict) and isinstance(actual, dict): + for key in sorted(set(expected) | set(actual)): + _compare_value( + case_id, + expected.get(key), + actual.get(key), + f"{path}.{key}" if path else key, + diffs, + expected_backend, + actual_backend, + ) + return + if isinstance(expected, list) and isinstance(actual, list): + for index in range(max(len(expected), len(actual))): + left = expected[index] if index < len(expected) else None + right = actual[index] if index < len(actual) else None + _compare_value( + case_id, + left, + right, + f"{path}[{index}]", + diffs, + expected_backend, + actual_backend, + ) + return + diffs.append( + { + "case_id": case_id, + "expected_backend": expected_backend, + "actual_backend": actual_backend, + "section": path.split(".", 1)[0], + "session_id": _extract_session_id(path), + "event_index": _extract_event_index(path), + "summary_id": _extract_summary_id(path, actual), + "field_path": path, + "expected": expected, + "actual": actual, + "allowed": False, + "reason": "normalized SDK-observable replay state differs", + } + ) + + +def _is_allowed_normalized_difference(path: str) -> bool: + return path.endswith("summary.anchor_id") or path.endswith("summary.anchor_timestamp") + + +def _extract_session_id(path: str) -> str | None: + match = re.search(r"sessions\.([^.[]+)", path) + return match.group(1) if match else None + + +def _extract_event_index(path: str) -> int | None: + match = re.search(r"events\[(\d+)\]", path) + return int(match.group(1)) if match else None + + +def _extract_summary_id(path: str, actual: Any) -> str | None: + return str(actual) if path.endswith("summary.anchor_id") and actual is not None else None + + +def _allowed_diffs(case_id: str) -> list[dict[str, Any]]: + return [ + { + "case_id": case_id, + "field_path": "sessions.*.events.*.id", + "allowed": True, + "reason": "event ids are SDK-generated unless fixed by the replay case", + }, + { + "case_id": case_id, + "field_path": "sessions.*.events.*.timestamp", + "allowed": True, + "reason": "event timestamps are normalized away as non-business fields", + }, + { + "case_id": case_id, + "field_path": "sessions.*.summary.anchor_timestamp", + "allowed": True, + "reason": "summary timestamps are compared by validity and order, not cross-backend equality", + }, + ] + + +def _write_report(comparisons: list[dict[str, Any]]) -> None: + report = { + "generated_at": "deterministic-test-run", + "comparisons": comparisons, + } + REPORT_PATH.write_text(json.dumps(report, indent=2, sort_keys=True), encoding="utf-8") + + +def _assert_summary_timestamps(snapshot: dict[str, Any]) -> None: + for session in snapshot["sessions"].values(): + summary = session["summary"] + if not summary["anchor_timestamp"]: + continue + assert isinstance(summary["anchor_timestamp"], float) + assert summary["anchor_timestamp"] > 0 + + +def _assert_sqlite_cold_summary(warm: dict[str, Any], cold: dict[str, Any]) -> None: + for session_id, warm_session in warm["sessions"].items(): + warm_summary = warm_session["summary"] + cold_summary = cold["sessions"][session_id]["summary"] + assert cold_summary["runtime_text"] is None + assert cold_summary["persisted_text"] == warm_summary["persisted_text"] + assert cold_summary["anchor_id"] == warm_summary["anchor_id"] + assert cold_summary["anchor_timestamp"] == warm_summary["anchor_timestamp"] + + +Mutation = tuple[str, Callable[[dict[str, Any]], None]] + + +def _mutations() -> list[Mutation]: + return [ + ("event_missing", lambda snapshot: snapshot["sessions"][next(iter(snapshot["sessions"]))]["events"].pop(0)), + ( + "event_duplicate", + lambda snapshot: snapshot["sessions"][next(iter(snapshot["sessions"]))]["events"].append( + copy.deepcopy(snapshot["sessions"][next(iter(snapshot["sessions"]))]["events"][0]) + ), + ), + ("event_order", _swap_first_two_events), + ("event_content", _mutate_first_event_text), + ("state_missing", lambda snapshot: snapshot["sessions"][next(iter(snapshot["sessions"]))]["state"].clear()), + ( + "state_dirty", + lambda snapshot: snapshot["sessions"][next(iter(snapshot["sessions"]))]["state"].update({"dirty": True}), + ), + ("memory_missing", lambda snapshot: snapshot["memory"]["searches"].clear()), + ("memory_value", _mutate_memory_result), + ("summary_lost", _mutate_summary_lost), + ("summary_overwrite_wrong", _mutate_summary_overwrite), + ("summary_wrong_session", _mutate_summary_wrong_session), + ] + + +def _first_session(snapshot: dict[str, Any]) -> dict[str, Any]: + return snapshot["sessions"][next(iter(snapshot["sessions"]))] + + +def _swap_first_two_events(snapshot: dict[str, Any]) -> None: + events = _first_session(snapshot)["events"] + events[0], events[1] = events[1], events[0] + + +def _mutate_first_event_text(snapshot: dict[str, Any]) -> None: + first_part = _first_session(snapshot)["events"][0]["parts"][0] + first_part["text"] = "mutated text" + + +def _mutate_memory_result(snapshot: dict[str, Any]) -> None: + searches = snapshot["memory"]["searches"] + if searches and searches[0]["results"]: + searches[0]["results"][0]["author"] = "wrong-author" + else: + searches.append({"query": "missing", "results": [{"author": "wrong-author", "parts": []}]}) + + +def _mutate_summary_lost(snapshot: dict[str, Any]) -> None: + summary = _first_session(snapshot)["summary"] + summary["persisted_text"] = None + summary["current_revision"] = None + summary["anchor_id"] = None + summary["anchor_count"] = 0 + + +def _mutate_summary_overwrite(snapshot: dict[str, Any]) -> None: + summary = _first_session(snapshot)["summary"] + summary["persisted_text"] = "stale summary revision v1" + summary["current_revision"] = "v1" + + +def _mutate_summary_wrong_session(snapshot: dict[str, Any]) -> None: + _first_session(snapshot)["summary"]["session_owner"] = "wrong-session" + + +@pytest.mark.asyncio +async def test_replay_cases_compare_inmemory_and_sqlite_without_false_positives(tmp_path: Path): + cases = _load_replay_cases() + assert len(cases) == 10 + + comparisons = [] + false_positive_count = 0 + for case in cases: + inmemory = await _build_inmemory_backend() + sqlite = await _build_sqlite_backend(tmp_path / case["case_id"]) + try: + inmemory_snapshot = await _run_case(case, inmemory) + sqlite_snapshot = await _run_case(case, sqlite) + cold_snapshot = await _cold_sqlite_snapshot(case["case_id"], sqlite, sqlite_snapshot) + diffs = _compare_snapshots( + case["case_id"], + inmemory_snapshot, + sqlite_snapshot, + expected_backend="inmemory", + actual_backend="sqlite", + ) + false_positive_count += 1 if diffs else 0 + comparisons.append( + { + "case_id": case["case_id"], + "expected_backend": "inmemory", + "actual_backend": "sqlite", + "diffs": diffs, + "allowed_diffs": _allowed_diffs(case["case_id"]), + } + ) + _assert_summary_timestamps(inmemory_snapshot) + _assert_summary_timestamps(sqlite_snapshot) + _assert_sqlite_cold_summary(sqlite_snapshot, cold_snapshot) + finally: + await _close_backend(inmemory) + await _close_backend(sqlite) + + _write_report(comparisons) + assert false_positive_count / len(cases) <= 0.05 + assert all(not comparison["diffs"] for comparison in comparisons) + + +@pytest.mark.asyncio +async def test_injected_inconsistencies_are_detected(tmp_path: Path): + cases = {item["case_id"]: item for item in _load_replay_cases()} + baselines = {} + for case_id in ["summary_update", "state_overwrite", "memory_write_read"]: + backend = await _build_sqlite_backend(tmp_path / case_id) + try: + baselines[case_id] = await _run_case(cases[case_id], backend) + finally: + await _close_backend(backend) + + missed = [] + for mutation_name, mutate in _mutations(): + if mutation_name.startswith("state"): + case_id = "state_overwrite" + elif mutation_name.startswith("memory"): + case_id = "memory_write_read" + else: + case_id = "summary_update" + baseline = baselines[case_id] + mutated = copy.deepcopy(baseline) + mutate(mutated) + diffs = _compare_snapshots( + case_id, + baseline, + mutated, + expected_backend="baseline", + actual_backend=mutation_name, + ) + if not diffs: + missed.append(mutation_name) + assert missed == [] + + +@pytest.mark.asyncio +async def test_summary_loss_overwrite_and_wrong_session_are_detected(tmp_path: Path): + case = next(item for item in _load_replay_cases() if item["case_id"] == "summary_update") + backend = await _build_sqlite_backend(tmp_path) + try: + baseline = await _run_case(case, backend) + finally: + await _close_backend(backend) + + for mutation_name, mutate in [ + ("summary_lost", _mutate_summary_lost), + ("summary_overwrite_wrong", _mutate_summary_overwrite), + ("summary_wrong_session", _mutate_summary_wrong_session), + ]: + mutated = copy.deepcopy(baseline) + mutate(mutated) + diffs = _compare_snapshots( + case["case_id"], + baseline, + mutated, + expected_backend="baseline", + actual_backend=mutation_name, + ) + assert diffs, mutation_name + + +def test_diff_report_shape(): + assert REPORT_PATH.exists() + report = json.loads(REPORT_PATH.read_text(encoding="utf-8")) + assert "comparisons" in report + assert len(report["comparisons"]) == 10 + for comparison in report["comparisons"]: + assert {"case_id", "expected_backend", "actual_backend", "diffs", "allowed_diffs"} <= comparison.keys() + for diff in comparison["diffs"]: + assert {"session_id", "event_index", "summary_id", "field_path", "expected", "actual"} <= diff.keys() + + +def test_redis_integration_is_env_gated(): + if not os.getenv("TRPC_AGENT_REPLAY_REDIS_URL"): + pytest.skip("Set TRPC_AGENT_REPLAY_REDIS_URL to enable Redis replay consistency integration.") + assert os.getenv("TRPC_AGENT_REPLAY_REDIS_URL")