sessions: 添加多后端回放一致性测试框架

[Wsp030914]

新增 Session、Memory 和 Summary 多后端回放一致性测试框架，使用十条标准
轨迹对比 InMemory 与 SQLite，并支持通过环境变量启用 Redis。框架覆盖对话、
工具调用、state、memory、summary、事件截断、重复写入和异常恢复。设置
TRPC_REPLAY_BACKENDS=in_memory 可仅运行完整十条 InMemory 轻量轨迹。

比较器仅归一化自动 ID、相对时间和序列化顺序。业务字段及 Summary 归属、
版本和覆盖关系严格比较；后端差异必须匹配精确 allowed_diff。报告可定位到
session、event 或 summary 及具体字段路径。

十一类注入异常全部检出，正常场景误报率为 0%，三类 Summary 问题检出率
均为 100%。默认测试结果为 55 passed、12 optional skipped，分支覆盖率
为 92.35%，YAPF 和 flake8 检查通过。补充同一 runner 重载状态隔离测试，
并断言 unknown-outcome 重试不会触发第二次 append_event。

使用 deepseek-v4-flash 完成真实 Tool Call 验证。正常 InMemory/SQLite
比较为零差异；Chinese 到 ChinesX 的工具结果漂移及 email 到 emaiX 的
记忆漂移均被精确检出。证据见
tests/sessions/real_agent_replay_validation_report.json。

Fixes #89

RELEASE NOTES: NONE

[github-actions]

CLA Assistant Lite bot All contributors have signed the CLA ✍️ ✅

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
⚠️ Please upload report for BASE (main@6a2f7f9). Learn more about missing BASE report.

Additional details and impacted files

@@            Coverage Diff             @@
##             main        #240   +/-   ##
==========================================
  Coverage        ?   87.86678%           
==========================================
  Files           ?         482           
  Lines           ?       45157           
  Branches        ?           0           
==========================================
  Hits            ?       39678           
  Misses          ?        5479           
  Partials        ?           0           

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[helloopenworld]

AI Code Review

我已完成审查。该 PR 是一个综合性、自验证的测试工具，对生产环境 API 的使用准确。让我来整理一下审查结论。

发现的问题

🚨 Critical

未发现 Critical 级别问题。

⚠️ Warning

• tests/sessions/replay_harness.py:399-411：reload_snapshot 复用同一 ReplayRunner 实例，但未重置 self.memory_results/self.summary_checkpoints/self.failures 等运行态字段

  • reload_snapshot 用于重建服务后读取持久化数据，但它沿用了原 runner 在首次 run() 中累积的 memory_results、summary_checkpoints、failures 与 summary_generation。在 _run_backend_pair 中每次都新建 runner（ReplayRunner(reopened)），所以当前测试路径不触发；但该方法作为可复用 API 被公开导出，若调用者在同一实例上先 run 再 reload_snapshot，snapshot() 会把旧 summary_checkpoints/failures 一并写入重载快照，污染跨后端对比。建议在 reload_snapshot 开头重置这些字段，或在文档/实现上明确每次必须新建 runner。

• tests/sessions/replay_harness.py:305-315：_unknown_outcome_retry 的幂等判定基于 event.id，未覆盖并发或写入异常导致事件已部分提交但不可见的边界

  • 该处理器先 append 再抛 InjectedReplayFailure，随后重新读取判定 event 是否存在以决定是否重写。判定逻辑本身正确（生产 append_event 成功即提交），但若未来后端出现“提交成功但随后读取不可见”的时序（如 Redis 主从延迟），会触发重复写入。当前后端（in_memory/sqlite）无此问题，仅作为测试有效性提示：建议在 test_unknown_outcome_retry_writes_event_once 中补一条“重试路径在事件已存在时不再写入”的显式断言（现仅断言最终计数为 1，未区分“未重写”与“重写后又被去重”）。

💡 Suggestion

• tests/sessions/replay_harness.py:118-125：_cleanup_redis 直接访问 memory_service._redis_storage 私有属性
  • 该清理逻辑依赖生产实现细节（_redis_storage 属性名与 delete/create_db_session 私有方法），一旦生产重构会静默断裂且仅在 Redis 集成测试（默认跳过）中暴露。建议在 RedisMemoryService 上提供公开的 delete_session(session) 语义方法，或在此处加注释明确这一耦合约束，便于后续维护。

总结

整体风险低：该 PR 是一套自验证的回放一致性测试框架，对生产 API 的调用（含私有 _sql_storage/_redis_storage、key 格式、配置与构造签名）均已与生产源码核对一致，核心重试/幂等/快照对比逻辑正确。无必须修复的阻塞性问题；上述 Warning 均属可维护性与测试有效性层面的改进。

测试建议

• 在 test_unknown_outcome_retry_writes_event_once 基础上，补充“事件已存在时重试路径不触发第二次 append_event”的断言（可通过 mock 计数 append 调用次数实现），以显式覆盖幂等判定分支。
• 若 reload_snapshot 保留为公共 API，补充一条“同一 runner 实例先 run 再 reload 时运行态字段不串入重载快照”的测试。

[Wsp030914]

I have read the CLA Document and I hereby sign the CLA

[Wsp030914]

AI Code Review

我已完成审查。该 PR 是一个综合性、自验证的测试工具，对生产环境 API 的使用准确。让我来整理一下审查结论。

发现的问题

🚨 Critical

未发现 Critical 级别问题。

⚠️ Warning

• tests/sessions/replay_harness.py:399-411：reload_snapshot 复用同一 ReplayRunner 实例，但未重置 self.memory_results/self.summary_checkpoints/self.failures 等运行态字段

  • reload_snapshot 用于重建服务后读取持久化数据，但它沿用了原 runner 在首次 run() 中累积的 memory_results、summary_checkpoints、failures 与 summary_generation。在 _run_backend_pair 中每次都新建 runner（ReplayRunner(reopened)），所以当前测试路径不触发；但该方法作为可复用 API 被公开导出，若调用者在同一实例上先 run 再 reload_snapshot，snapshot() 会把旧 summary_checkpoints/failures 一并写入重载快照，污染跨后端对比。建议在 reload_snapshot 开头重置这些字段，或在文档/实现上明确每次必须新建 runner。

• tests/sessions/replay_harness.py:305-315：_unknown_outcome_retry 的幂等判定基于 event.id，未覆盖并发或写入异常导致事件已部分提交但不可见的边界

  • 该处理器先 append 再抛 InjectedReplayFailure，随后重新读取判定 event 是否存在以决定是否重写。判定逻辑本身正确（生产 append_event 成功即提交），但若未来后端出现“提交成功但随后读取不可见”的时序（如 Redis 主从延迟），会触发重复写入。当前后端（in_memory/sqlite）无此问题，仅作为测试有效性提示：建议在 test_unknown_outcome_retry_writes_event_once 中补一条“重试路径在事件已存在时不再写入”的显式断言（现仅断言最终计数为 1，未区分“未重写”与“重写后又被去重”）。

💡 Suggestion

• tests/sessions/replay_harness.py:118-125：_cleanup_redis 直接访问 memory_service._redis_storage 私有属性

  • 该清理逻辑依赖生产实现细节（_redis_storage 属性名与 delete/create_db_session 私有方法），一旦生产重构会静默断裂且仅在 Redis 集成测试（默认跳过）中暴露。建议在 RedisMemoryService 上提供公开的 delete_session(session) 语义方法，或在此处加注释明确这一耦合约束，便于后续维护。

总结

整体风险低：该 PR 是一套自验证的回放一致性测试框架，对生产 API 的调用（含私有 _sql_storage/_redis_storage、key 格式、配置与构造签名）均已与生产源码核对一致，核心重试/幂等/快照对比逻辑正确。无必须修复的阻塞性问题；上述 Warning 均属可维护性与测试有效性层面的改进。

测试建议

• 在 test_unknown_outcome_retry_writes_event_once 基础上，补充“事件已存在时重试路径不触发第二次 append_event”的断言（可通过 mock 计数 append 调用次数实现），以显式覆盖幂等判定分支。
• 若 reload_snapshot 保留为公共 API，补充一条“同一 runner 实例先 run 再 reload 时运行态字段不串入重载快照”的测试。

感谢 Review，已在 9aa9d1f 修复：

1. reload_snapshot 重载前会清空 memory、summary checkpoint、failure 和
   generation 等运行态；新增同一 runner 执行 run → reload 的回归测试。

2. unknown-outcome 测试新增 append_event 调用计数，确认事件已存在时不会
   发生第二次写入。

3. Redis 私有存储访问已补充耦合原因及使用范围说明。

[helloopenworld]

AI Code Review

好的 —— 只涉及 tests/sessions/。此 PR 是纯测试代码。CI 会运行 pytest ... tests/ 以及 --cov-fail-under=80，因此这些测试必须在收集阶段通过，且不能破坏覆盖率。

现在让我仔细重新审查 _unknown_outcome_retry 中的 self.session = stored —— 在条件性的重新追加（re-append）之后，stored 是重新读取之前的陈旧 session。测试 test_unknown_outcome_retry_writes_event_once 断言 append_event.await_count == 1 且 ids.count("retry-event") == 1。但重试逻辑会调用 _append，进而调用 append_event；如果事件已经提交，则不会再次调用。所以 await_count=1。好的。但请注意：重新追加（re-append）会调用 self._append(operation, operation_index)，它会使用 session = self._require_session() —— 也就是 self.session。此时 self.session 是 stored（刚读取的）。然后追加到该 session。但之后 self.session = stored（相同的对象）。没问题。

等等 —— 这里有个细微的 bug。在 _unknown_outcome_retry 中，如果 stored 中已经有了该事件（已提交），它会跳过重新追加并设置 self.session = stored。很好。但如果由于某种原因没有提交，它会重新追加，但 stored（刚读取的 session）不会被更新以包含新事件，且 self.session = stored 指向的是预追加（pre-append）的 session。然而，_append 使用的是 self._require_session()，它返回 self.session（即 stored 对象），并调用了 append_event(session, event)，这会就地修改 session 对象（针对内存服务）。所以 stored 会被更新。OK，问题不大。

让我再检查一个真正的问题：test_in_memory_only_lightweight_mode 中存在模块级 fixture 和环境变量过滤。以及 conftest.py 中的跳过过滤器通过 item.name 进行匹配 —— item.name 是完整的测试节点 ID，对于参数化测试，它包含了参数。IN_MEMORY_TEST_NAME = "test_in_memory_only_lightweight_mode" 匹配的是非参数化的测试名称。对于非参数化的 test_in_memory_only_lightweight_mode，item.name 确实就是 "test_in_memory_only_lightweight_mode"。好的。

现在让我看看一个真正的正确性问题：在 BackendBundle.close 中，Redis 的清理使用了 self.memory_service._redis_storage —— 我已经确认该属性存在。以及 delete(connection, key) 的签名 —— delete(self, conn, key, conditions=None)。调用方式是 delete(connection, key)。没问题。

让我检查 _cleanup_redis 潜在的 Key 清理问题 —— 它只删除 memory:{save_key}:{id}，但 store_session 可能通过 delete(redis_session, key) + rpush 创建了键。单个 delete 操作即可清除它。OK。但其他会话的内存键呢？save_key 是基于 user 的；使用相同 save_key 但不同 session_id 的会话会有不同的键。由于 Redis 后端使用了带有命名空间的身份标识，因此每次运行都是隔离的。没问题。

现在看看一个真正值得关注的问题 —— conftest.py 中的 pytest_collection_modifyitems 引用了 test_replay_real_agent.py，并且会跳过除内存专用测试之外的所有测试。但 test_replay_real_agent.py 中并没有 test_in_memory_only_lightweight_mode —— 它只存在于 test_replay_consistency.py 中。因此在内存模式下，test_replay_real_agent.py 中的所有测试都会被跳过。这没问题（它们无论如何都需要 API keys）。

让我验证最后一项 —— _with_final_memory 和 compare_persisted_snapshots。在 compare_persisted_snapshots 中，它调用了 _with_final_memory(left) 和 _with_final_memory(right)，两者都弹出 memory_final 并赋值给 memory。对于 sqlite_reloaded 快照，memory 为 {}（在 reload_snapshot 中已清除），而 memory_final 包含重新读取的结果。对于 in_memory（实时）快照，memory 包含即时结果，memory_final 包含最终读取结果。所以比较 in_memory.memory_final 和 sqlite_reloaded.memory_final。很好。

但这里有一个微妙的问题：在 compare_persisted_snapshots 中，操作是原地修改快照（_with_final_memory 执行了深拷贝，所以没问题）。之后在 test_injected_inconsistency_is_detected 中，它在 _with_final_memory 之后没有重用 pair["sqlite"] —— 实际上它直接使用了 pair["sqlite"] 和 pair["in_memory"]。由于 _with_final_memory 进行了深拷贝，原始数据未被触动。很好。

现在，关于 mutate_snapshot —— 它进行了深拷贝。没问题。

让我检查 test_reloaded_memory_uses_final_read_without_rewriting_trace_result 中潜在的真正问题 —— reloaded["memory"] == {}。在 reload_snapshot 中，self.memory_results.clear() 会使 memory 变为空。快照的 memory 字典是通过 self.memory_results 构建的。清空后 → 为空。很好。

现在让我检查最重要的真正 bug 候选：_normalize_value 和 set 处理。快照不应包含 set 类型（都是来自 model_dump(mode="json") 的 JSON）。没问题。

让我验证 test_normalization_detects_invalid_event_timestamp 和时间戳排名逻辑。在 _normalize_events 中，timestamp_ranks[item["timestamp"]] —— 如果时间戳是 -1.0（突变），它会被添加到 timestamps 集合中（因为 is not None）。所以排名映射包含 -1.0。item["timestamp"] 变为它的排名。时间轴约束检查 events_all_valid = all(_valid_timestamp(value) for value in event_values)，其中 event_values 是原始（非排名）时间戳。等等 —— _timeline_constraints 是在 _normalize_value 对结果进行递归处理之前，在 normalize_snapshot 中对原始快照计算的？让我重新检查：normalize_snapshot 执行 result["timeline_constraints"] = _timeline_constraints(result)，此时 result 仍然是深拷贝的原始快照（事件具有原始时间戳，尚未排名）。所以 event_values = 原始时间戳，包括 -1.0。_valid_timestamp(-1.0) = False。所以 events_all_valid = False。然后事件被归一化（时间戳→排名）。比较双方：左侧 (in_memory) events_all_valid=True，右侧 (mutated sqlite) = False → 在 /timeline_constraints/events_all_valid 处产生差异。很好，测试通过。

现在 —— 等等，关于 _timeline_constraints 的一个担忧。它在事件归一化之前计算，使用原始时间戳。但 _normalize_value 之后会规范化整个字典，包括 timeline_constraints。布尔值保持布尔值。所以最终的 timeline_constraints 包含从原始时间戳派生的布尔值。对于实时和重新加载的比较，两者都有有效的原始时间戳 → 一致。对于突变的快照，events_all_valid 为 False → 检测到差异。没问题。

我现在已经有足够的信息了。让我写下审查报告。让我重新考虑模块级异步 fixture 问题的确定性，并确认 pytest-asyncio 的行为。鉴于仓库中的 _reset_manager 异步 fixture（函数级作用域）工作正常，而 asyncio_mode=auto 确实支持异步 fixture，真正的问题是 模块级作用域 是否在没有 asyncio_default_fixture_loop_scope 的情况下工作。

在 pytest-asyncio>=0.23 中，引入了 asyncio_default_fixture_loop_scope 配置选项，默认为 function。当异步 fixture 具有不同的作用域（模块级）且没有显式的循环作用域时，较新的版本（>=0.24，然后 >=0.25 随着弃用警告移除了严格的默认值）会发出 PytestDeprecationWarning/错误。行为因版本而异。由于 pytest-asyncio 未指定版本，且新安装会拉取最新版本，这可能会导致测试阶段失败或发出弃用警告。我将此标记为 Warning（警告）（因为版本不确定，它可能仍然通过，但这是一个在版本升级下脆弱的真正兼容性风险）。

让我同时也检查一下 cov-fail-under=80 是否有风险 —— 添加测试文件会增加代码行数，但不会增加 trpc_agent_sdk/ 的覆盖范围（测试目录被 --cov=trpc_agent_sdk 排除）。因此，新的测试文件不会降低覆盖率。没问题。

再检查一个真正的正确性项目：test_in_memory_only_lightweight_mode 是一个模块级 fixture 的异步测试。但它是一个独立的异步测试（不使用 replay_matrix）。在内存模式下，conftest 会跳过除该测试之外的所有重放测试。该测试创建了自己的后端。没问题。

让我验证 Redis 清理调用 delete_session 是否正确清理了 —— 以及当 TRPC_REPLAY_REDIS_URL 未设置（CI 环境的默认情况）时，跳过列表引用 REDIS_ENVIRONMENT_VARIABLE 是否正确。没问题。

现在让我最终确认 _corrupt_tool_response —— 它遍历事件，检查 event["content"]["parts"][0].get("function_response")。但来自 model_dump 的事件字典：parts[0] 是一个字典，例如 {"function_response": {...}}。.get("function_response") 返回字典或 None。没问题。然后是 payload.get("preferred_language")。对于模拟用例（_tool_memory_case 基于 tool_round_trip + 记忆事实），工具响应是 {"name": "weather", "id": "call-weather", "response": {"temperature": 28}} —— 没有 preferred_language，但有 temperature（整数）。所以它命中了 payload["temperature"] += 1 分支 → 将 28 变为 29。差异被检测为事件差异。test_subtle_realistic_drift_is_detected 断言 any(diff.category == "events" and "function_response" in diff.field_path ...)。字段路径将是 /events/N/content/parts/0/function_response/response/temperature。包含 "function_response"。很好。

对于真实代理，工具响应包含 preferred_language。_corrupt_tool_response 修改了它。很好。

我很有把握了。让我写出审查报告。让我重新确认生成报告 JSON（提交的 session_memory_summary_diff_report.json）中 allowed_diff_audit 的 hit_count 值 —— 它显示 /summary/cache_present、/summary/original_event_count、/summary/compressed_event_count、/summary/generation 的 hit_count 为 4。有 4 个包含摘要的案例（duplicate_retry、summary_create、summary_truncation、summary_update）。每个案例在重载时有 1 次差异。所以 4 次命中。/summary_checkpoints/0 hit_count=4，/summary_checkpoints/1 hit_count=1（仅 summary_update），/failures/0 hit_count=1（仅 write_recovery）。总计。并且 _validate_allowed_diff_usage 要求所有规则都有 hit_count>0 —— 是的。并且测试断言 all(item["hit_count"] > 0 ...)。很好。

现在，最后一个真正的问题：提交的 JSON session_memory_summary_diff_report.json 是重新生成的。测试 test_report_contains_every_backend_and_case 断言 _stable_report(root_report) == _stable_report(persisted) —— 意味着提交的 JSON 必须与测试运行时生成的结果完全匹配（归一化持续时间后）。这很脆弱：如果任何后端行为发生改变，或者排序发生改变，这个断言就会失败。但更重要的是，由于持续时间被置零，并且报告是确定性排序的，只要后端行为一致，它就应该稳定。鉴于测试也断言了一致性，这是可以接受的。不是缺陷。

让我重新考虑是否存在 JSON 内容/结构漂移的风险，导致提交的报告与生成的报告不匹配。_stable_report 会将运行持续时间置零，但不会处理快照持续时间。生成的报告 runs 包含来自 pair["durations"] 的 duration_seconds。等等 —— run_record 包含 duration_seconds。_stable_report 将 run["duration_seconds"] = 0。所以双方所有的持续时间都被置零。其他所有内容（事件计数、状态键、比较、差异、审计）都是确定性的。应该匹配。好的。

我现在来写审查报告。考虑到这些全都是新文件，让我重新考虑 pr.diff 中是否还有我遗漏的严重（Critical）问题。让我重新检查 _validate_summary，看看当 summary is None and not summary_events 与预期的生成计数不匹配时，是否存在真正的正确性 bug。

在 validate_snapshot 中，require_runtime 默认为 True。对于重载的快照，require_cache=False, require_runtime=False。memory_field = "memory_final"。对于 summary_update 的 sqlite_reloaded，summary_generation 预期=2。_validate_summary：expected.summary_generation=2 != 0。summary —— 对于重载，get_session_summary 返回 None（缓存是进程局部的），但 summary_events（来自 session.events 的锚点）存在。所以 summary 字典是通过 _summary_to_dict(summary=None, summary_events=[anchor], generation=0, session_id) 构建的 → cache_present=False，generation=0。然后 if require_cache and summary["generation"] != expected.summary_generation —— require_cache=False，跳过。if expected.summary_fact not in summary["text"] —— 检查事实是否在文本中。锚点文本包含摘要。summary_fact="Initial requirement" 应该在摘要文本中。OK。if summary["session_id"] != snapshot["session_id"] or summary["anchor_count"] != 1 —— 两者都正常。if require_cache and (not cache_present ...) —— require_cache 为 False，跳过。所以没有错误。很好。

我确信没有严重（Critical）的缺陷了。让我定稿。让我重新考虑模块级 fixture 的问题是否应该提升为严重（Critical）。鉴于不确定性（取决于安装的 pytest-asyncio 版本），且无法验证，我将其保留为 Warning（警告），并带有一定程度的不确定性。实际上 —— 由于新版本 pytest-asyncio 中自动模式下的模块级异步 fixture 在没有 asyncio_default_fixture_loop_scope 的情况下会失败或发出弃用警告，且 CI 执行 pip install -r requirements-test.txt（未锁定版本 → 最新版本），这是 CI 中一个真实的失败风险。但因为我无法确认安装的版本，并且测试可能已经通过（提交的报告显示成功运行），我将把它标记为 Warning（警告）。让我把它写出来。

实际上，提交的报告显示运行成功（"status": "passed"），这表明测试确实运行并成功了。这暗示安装的 pytest-asyncio 版本容忍了模块级异步 fixture。因此，升级版本后它很脆弱但目前可以通过。Warning（警告）是正确的级别。

让我写出最终审查报告。## 发现的问题

🚨 Critical

（无）

⚠️ Warning

• tests/sessions/test_replay_consistency.py:90-91：模块级 async fixture 缺少 loop scope 配置，存在 CI 兼容风险

  • @pytest.fixture(scope="module") async def replay_matrix(...) 是本仓库唯一的模块级 async fixture（其余模块级 fixture 均为同步函数）。仓库仅配置了 asyncio_mode = "auto"，未配置 asyncio_default_fixture_loop_scope，且 pytest-asyncio 在依赖中未锁版本（CI 走 pip install -r requirements-test.txt 拉取最新版）。在新版 pytest-asyncio 中，模块级 async fixture 在缺少显式 loop scope 时会触发弃用告警甚至错误，可能导致 CI test 作业失败。建议在 pyproject.toml 的 [tool.pytest.ini_options] 补充 asyncio_default_fixture_loop_scope = "module"（或将该 fixture 改为 pytest_asyncio.fixture 并显式声明 loop scope）以稳定行为。

    @pytest.fixture(scope="module")
    async def replay_matrix(tmp_path_factory):
        ...

• tests/sessions/replay_harness.py:187-202：Redis 后端创建依赖私有的 decode_responses 透传，且清理逻辑耦合私有存储 API

  • create_redis_backend 通过 **kwargs 传 decode_responses=True 给 RedisSessionService/RedisMemoryService，最终落到 RedisStorage.__init__ 的 from_url；该路径依赖 RedisMemoryService 暴露 _redis_storage 私有属性（_cleanup_redis 直接访问 self.memory_service._redis_storage.create_db_session() 与 delete）。一旦生产侧重命名/下线该私有属性或 Redis 客户端不再接受 decode_responses，Redis 集成测试会在运行期报错而非收集期失败。该路径仅在 TRPC_REPLAY_REDIS_URL 设置时触发，CI 默认跳过，属可控风险；建议在清理处对缺失属性做显式校验或加注释说明耦合契约。

• tests/sessions/test_replay_consistency.py:1714-1715：提交的基线报告与生成报告做整体相等断言，脆弱

  • assert _stable_report(root_report) == _stable_report(persisted) 要求仓库内 session_memory_summary_diff_report.json 与本次运行生成报告在归一化后逐字段相等。该断言把测试输出与一份手维护的 JSON 强绑定：任何后端行为、排序或字段调整都会让此断言失败，且失败信息不指向具体回归点。建议改为按关键字段（metrics、unexpected_diff_count、allowed_diff_audit 命中）做结构化断言，或在失败时输出 diff 以降低维护成本。

💡 Suggestion

• tests/sessions/replay_harness.py:319-321：_unknown_memory_retry 连续两次 store_session 但未校验幂等
  • 该操作模拟内存写入重试，依赖 store_session 内部 delete+重写实现幂等；若后续 memory 服务改为增量写入，重试会产生重复记忆而本测试无法察觉。可在 ExpectedOutcome 中显式标注该 case 的幂等契约，便于后续维护者识别。

总结

整体风险可控：本 PR 为纯新增测试（tests/sessions/ 下 9 个文件），未改动生产代码，正常 CI 路径（无 Redis、无真实模型 key）下逻辑自洽且与生产 API 契约一致。最值得处理的是模块级 async fixture 缺少 asyncio_default_fixture_loop_scope，在 pytest-asyncio 未锁版本时构成 CI 兼容隐患；不存在必须修复的 Critical 问题。

测试建议

• 建议在 CI 中显式跑一次 TRPC_REPLAY_BACKENDS=in_memory 模式，验证 conftest.py 的跳过过滤与 test_in_memory_only_lightweight_mode 在仅内存模式下的一致性。
• 建议补一个针对模块级 async fixture 的冒烟用例或在 CI 升级 pytest-asyncio 后验证 replay_matrix 仍可正常初始化（覆盖 loop scope 行为）。

[helloopenworld]

AI Code Review

我已经完成了静态审查。以下是我的审查结论。

发现的问题

⚠️ Warning

• tests/sessions/replay_harness.py:126-127、tests/sessions/replay_harness.py:155-156、tests/sessions/replay_harness.py:627-628：测试直接访问 SDK 私有存储接口（memory_service._redis_storage.create_db_session()/delete()、session_service._sql_storage.create_sql_engine()、session_service._sql_storage.commit）
  • 这些 _ 前缀成员是 SDK 内部实现细节，没有任何稳定性契约。一旦 RedisStorage/SqlStorage 重构方法名或上下文管理器契约（如 create_db_session），CI 上的回放测试会整体崩溃，且崩溃点远离真实回归。commit 的 monkey patch（test_replay_consistency.py:1973、2017）尤其脆弱。建议通过 SDK 公共接口（如为 RedisMemoryService/SqlStorage 增加受控的测试清理/事务注入点）暴露这些能力，或在测试中显式声明该耦合并加版本护栏，避免私有 API 漂移静默阻断 CI。

💡 Suggestion

• tests/sessions/test_replay_consistency.py:91：@pytest_asyncio.fixture(scope="module", loop_scope="module") 依赖 pytest-asyncio>=0.23，而 requirements-test.txt 未锁定 pytest-asyncio 版本。若 CI 环境降级到旧版本会直接 TypeError。建议在 requirements-test.txt 显式下限 pytest-asyncio>=0.23，消除隐性版本依赖。

总结

该 PR 是一套结构清晰、带自校验（规模/函数复杂度/魔法数 lint）的回放一致性测试，核心回放/比较/归一化逻辑与 SDK 行为核对一致，未发现明确的逻辑或安全 Critical 问题。主要风险在于测试与 SDK 私有存储接口的深度耦合，以及 pytest-asyncio 版本未锁定，二者都会在 SDK/依赖演进时让 CI 产生远离根因的失败。

测试建议

• 暂无额外测试建议；现有用例已覆盖正常一致性、重放幂等、写失败回滚、summary 提交失败、mutation 检测等关键路径。若采纳上述私有接口耦合的修复方向，建议补一个"SDK 私有存储符号被重命名时测试给出可读报错"的护栏测试，避免静默全量失败。

[helloopenworld]

AI Code Review

确认 — SDK 会剥离临时状态。很好。测试有效。

我已经完成了全面审查。该 diff 质量很高，自洽性好，SDK API 调用也是正确的。我来整理最终结论，附上少量真正值得注意的问题。

发现的问题

🚨 Critical

未发现 Critical 级别问题。

⚠️ Warning

• tests/sessions/test_replay_consistency.py:1710-1711：生成的报告与已提交报告做精确比对，脆弱性强

  • test_report_contains_every_backend_and_case 在未配置 Redis 时断言 _stable_report(root_report) == _stable_report(persisted)，即新生成报告必须与仓库内 session_memory_summary_diff_report.json 完全一致（仅忽略 duration）。任何对快照结构、归一化或 allowed-diff 的合理演进都会触发 CI 失败，且需要手工重新生成该 JSON。建议改为断言关键字段（schema_version、metrics、unexpected_diff_count、allowed_diff_audit 命中情况）而非整体结构相等，以降低维护成本。

• tests/sessions/test_replay_consistency.py:1536-1538：模块级 replay_matrix fixture 聚合全部用例，单点失败会拖垮整模块诊断

  • 该 scope="module" fixture 一次性跑完 10 个 case × 3 后端，任何单个 case/后端异常都会使 fixture 抛错，导致本模块所有参数化测试以 fixture 错误而非具体 case 失败呈现，定位成本高。建议在 fixture 内对每个 case 单独捕获并汇总错误，或在失败时保留已完成结果以便定位。

💡 Suggestion

• tests/sessions/test_replay_consistency.py:1939-1940：_find_diff 使用无默认值的 next(...)，未命中时抛 StopIteration 而非清晰断言

  • 测试失败信息会变成 StopIteration，难以判断是“未检测到差异”还是其他问题。建议改为 next((d for d in diffs if d.field_path == path), None) 并对 None 显式 assert 出可读消息。

• tests/sessions/test_replay_real_agent.py:2348 与 tests/sessions/test_replay_consistency.py:1541：异步测试标注不一致

  • real_agent 文件用 @pytest.mark.asyncio，consistency 文件依赖 asyncio_mode=auto。两种方式都能跑通，但混用会让人误以为存在模式差异。建议统一为依赖 auto 模式（去掉多余标记）或统一显式标记。

总结

该 PR 新增的 Session/Memory/Summary 回放一致性测试框架整体逻辑正确、自洽，对 SDK 公共 API（SessionServiceConfig、MemoryServiceConfig、SummarizerSessionManager、save_key、is_summary_event 等）的调用均与仓库实现匹配，pytest-asyncio>=0.24.0 的版本约束也与 loop_scope 用法相容。未发现必须修复的阻塞性问题；两项 Warning 均为测试可维护性/诊断性方面的改进，不影响当前正确性。

测试建议

暂无额外测试建议。现有用例已覆盖正常一致性、持久化重开、注入异常检测、并发隔离、Redis 可选集成等关键路径；若采纳上述 Warning，建议补充“报告结构演进时仅关键字段比对”的单测以替代当前的全文比对。

[helloopenworld]

AI Code Review

现在让我检查一下在 _validate_allowed_diff_usage 中可能存在的一个实际问题：当 Redis 启用时，执行对包含 ("in_memory", "redis")。对于该对，没有允许的差异规则，因此没有规则适用 → 没问题。对于 ("in_memory", "sqlite_reloaded")，所有规则的 hit_count > 0。_append_redis_report_data 会添加 redis 运行和比较。测试断言 len(runs) == N*(3 + redis_enabled)。当启用 redis 时，每个用例会增加 1 个 redis 运行。但 redis 比较 ("in_memory","redis") —— 没有添加相应的 sqlite_reloaded redis 对。比较次数 = N*2 + N（redis 对）。断言 comparisons == N*(2 + redis_enabled)。很好。

但确实存在一个问题：当启用 redis 时，_validate_allowed_diff_usage 仍然有效，因为 redis 对没有规则。并且 test_report_contains_every_backend_and_case 仅在 not redis_enabled 时断言 _report_contract 根匹配（第 1715 行）。所以启用 redis 的情况并没有根据根目录进行检查。没问题。

现在让我验证一下是否存在实际的 Bug：_allowed_diff_audit 中的 hit_count 是通过 diff["field_path"] == rule.field_path AND diff["allowed"] 计算的。对于 ("in_memory","sqlite_reloaded") 对，针对 write_recovery 出现了 /failures/0 差异，allowed=True → hit_count=1。但对于其他案例，没有 /failures/0 差异。hit_count 在所有案例中进行聚合。所以 /failures/0 规则的 hit_count=1 > 0。很好。

等等，确实存在一个细微的问题：_validate_allowed_diff_usage 检查 tuple(item["backend_pair"]) in executed_pairs。对于 redis 测试 test_redis_integration_uses_isolated_namespace，它直接使用 compare_snapshots，而不是 build_report。所以那里没有验证。没问题。

我已经进行了彻底的审查。这个测试工具结构良好。让我总结一下真正值得注意的问题。老实说，我没有发现任何 Critical（严重）问题。让我重新考虑一下是否真的存在 Warning（警告）级别的问题。

有一个确实值得注意的问题：_corrupt_tool_response 修改了 parts[0]，但如果一个事件有多个部分，则只会损坏第一个。不过对于这些案例来说，每个事件只有一个部分。问题不大。

让我重新考虑一下真正的测试有效性问题：完整的重放一致性矩阵（test_public_replay_case_is_consistent，即实际的核心跨后端一致性检查）仅在 pytest tests/（覆盖率步骤）中运行，而不是在专门的重放步骤中运行。实际上它确实在覆盖率步骤中运行了。所以它是覆盖到的。没问题。

让我重新考虑一下 _model_config 调用模块顶层的 pytest.skip —— 实际上它是在 _capture_real_case 内部调用的。test_real_agent_tool_trace_replays_consistently 调用了 _capture_real_case → _model_config → 如果没有 env 则跳过。但是 test_subtle_realistic_drift_is_detected 不需要模型，也没有跳过 —— 它总是运行。因此在覆盖率步骤中，test_subtle_realistic_drift_is_detected 总是运行（很好），而 test_real_agent_tool_trace_replays_consistently 会跳过。没问题。

让我最后检查一下真正的边界情况问题 —— 在未设置 env 的情况下 os.environ[MODEL_NAME_ENV] 里的 _write_real_report。但 _write_real_report 仅在设置了 REPORT_PATH_ENV 时才被调用，而这与模型 env 是分开的。如果有人设置了 TRPC_REAL_REPLAY_REPORT_PATH 但没有设置 TRPC_AGENT_MODEL_NAME，那么 _capture_real_case 会先跳过（因为模型 env 缺失），所以 _write_real_report 永远不会被执行到。所以 os.environ[MODEL_NAME_ENV] 是安全的，因为它在 _model_config 通过后才被访问。等等 —— _write_real_report 使用了 os.environ[MODEL_NAME_ENV]。如果 _capture_real_case 通过了（所有模型 env 都已设置），那么 MODEL_NAME_ENV 就被设置了。所以是安全的。没问题。

我现在很有把握了。这个 PR 很稳健。让我写下总结。老实说，没有 Critical 问题，也许有 1-2 个小的 Suggestion（建议）或 Warning（警告）。让我重新考虑一下是否真的有任何实质性问题。

让我重新考虑一个真正的问题：_search_memory 中的 result_key 是用 _next_memory_result_key(self.memory_results, query) 计算的，但 reload_snapshot 是用 self.memory_query_by_key 计算的。这些字典是不同的。在重载时，memory_query_by_key 被清空然后重新填充。在第一次搜索查询时，它返回 query。但键随后被用作 memory_results（在快照期间也为空）和 memory_query_by_key 中的键。在 reload_snapshot 中，它仅填充 memory_query_by_key（而不是 memory_results）。然后 snapshot() 从空的 memory_results 构建 "memory" → {}。并且 _read_final_memory 遍历 memory_query_by_key。所以重载后的 memory_final 键与 memory_query_by_key 的键匹配。对于实时运行，memory_results 和 memory_query_by_key 是通过 _next_memory_result_key(self.memory_results, ...) 一起填充的。因此，键在实时运行的两个字典中是匹配的。对于重载，键是基于 memory_query_by_key（为空）计算的。因为单个查询的键都是 query，所以它们是匹配的。没问题。

好了，我认为这个 PR 没有发现阻塞性问题。让我进行总结。老实说，这个测试工具非常全面，我没有发现明确的缺陷。根据说明，如果没有明确的问题，我应该直接说明没有发现阻塞性问题，而不是为了凑数而提出低价值的建议。

让我做一个最终的检查 —— DeterministicSummaryModel.generate_async 解析 request.contents[0].parts[0].text —— 如果提示结构不同，这可能会引发 IndexError。但它是确定性使用的。问题不大。

让我再核实一下真正潜在的问题：_summary_text 使用了 event.get_text() 和 marker = "Previous conversation summary: "。摘要锚点内容是通过摘要管理器构建的。如果管理器的格式发生变化，这就会出问题。但这属于测试维护风险，而不是当前的问题。

我很有把握。让我写下最终的审查意见。我没有发现 Critical 或 Warning 级别的问题。根据说明，如果没有明确的问题，我应该直接说明，而不是提出低价值的建议。不过，让我重新考虑一下，是否确实存在值得提出的维护性建议。

实际上，确实有一个真正的小观察值得提出：test_in_memory_only_lightweight_mode 和 conftest 跳过机制意味着轻量级模式的 CI 步骤实际上只执行了一个测试；完整的矩阵验证依赖于覆盖率步骤。但这没问题。我最多把它作为建议提一下，但这与设计意图相符。

老实说，我会报告没有发现阻塞性问题，并带有一两个真正可维护性的建议。让我重新考虑一下什么才是真正有价值的。

一个真正合理的问题：BackendBundle.close() 在收集错误后会重新抛出 errors[0]，丢弃了其他错误，并丢弃了原始的 traceback 上下文（它抛出的是存储的异常对象，这会保留 __context__，但会丢失 as a new raise）。问题不大 —— 测试工具。

我就如实报告吧。没有 Critical 问题。我会包含一个关于已提交报告制品的真正可维护性建议。我已对 pr.diff（共 2403 行，新增 tests/sessions 下的 replay 测试框架与 CI/依赖改动）及其相关 SDK 上下文进行了完整审查。下面是结论。

发现的问题

未发现 Critical 或 Warning 级别的明确问题。新增的 replay harness（replay_cases.py / replay_harness.py / replay_report.py）、两套测试与 CI/依赖改动在正确性、安全性、稳定性、兼容性上均未发现可从 diff 与上下文验证的缺陷：handler 调度、重试幂等（_unknown_outcome_retry / _unknown_summary_retry）、内存点-in-time 与 reload 读模型、归一化/白名单 diff 审计、超时与资源释放（BackendBundle.close 即使首个 close 失败仍关闭第二个并聚合异常）等关键路径均与对应断言及提交的基线报告一致；所依赖的 SDK API（save_key、search_memory(key, query)、summarizer_manager、SessionServiceConfig.clean_ttl_config、SqlSessionService(is_async=...) 等）均存在且签名匹配。

💡 Suggestion

• tests/sessions/real_agent_replay_validation_report.json、tests/sessions/session_memory_summary_diff_report.json：提交了两份测试运行产物 JSON。
  • 这两份文件仅在 TRPC_REAL_REPLAY_REPORT_PATH 设置或 _report_contract 比对时被读取，本身不阻塞 CI；但作为仓库内固定产物，会随 SDK 行为/归一化规则演进而过期，后续维护时容易与实际运行结果漂移。建议在文件头或 DESIGN.md 中标注"由测试生成、可通过删除后重跑再生"，避免后续被误当作契约固化。
• tests/sessions/conftest.py:28-30：InMemory-only 模式下，整个 test_replay_real_agent.py 因按文件名跳过而完全不执行（包含不依赖真实模型的 test_subtle_realistic_drift_is_detected）。
  • 当前设计为轻量模式，非缺陷；但该轻量 CI 步骤实际仅运行 test_in_memory_only_lightweight_mode 一个用例，跨后端一致性矩阵的真实覆盖仍依赖 pytest --cov ... tests/ 这一全量步骤。可维护性上建议在该 step 名或注释中点明"轻量冒烟，非完整矩阵"，避免后续误读为完整 replay 校验。

总结

整体风险低，未发现必须修复或建议修复的明确缺陷；新增 replay 一致性测试框架逻辑自洽、与 SDK 上下文一致，CI 与依赖变更（pytest-asyncio>=0.24.0、新增 InMemory-only 步骤、conftest skip 策略）均为兼容性安全改动。

测试建议

暂无额外测试建议。现有用例已覆盖核心风险路径：跨后端一致性、写恢复/重试幂等（test_unknown_outcome_retry_writes_event_once）、reload 后读模型隔离（test_reloaded_memory_uses_final_read_without_rewriting_trace_result）、注入变异检测（11 个 mutation）、报告白名单审计与未用规则拒绝。仅建议在后续接入 Redis 时补充 compare_persisted_snapshots 对 Redis reload 路径的断言（当前 Redis 仅做 live compare_snapshots，无 reload 比较）。

[helloopenworld]

AI Code Review

发现的问题

审查了 pr.diff 全部变更（CI workflow、requirements-test.txt、tests/sessions/ 下新增的 replay harness、报告与测试），并对照 SDK 真实签名（SessionSummary/Session.save_key/Event.is_summary_event/SessionServiceConfig.clean_ttl_config/Redis 服务 **kwargs 透传 decode_responses/create_session_summary 等）以及 pyproject.toml 的 asyncio_mode = "auto"、loop_scope="module"（需 pytest-asyncio>=0.24，与 requirements 改动一致）逐一验证。

未发现明显阻塞问题。harness 的字段访问、归一化/比对逻辑、注入失败重放路径（_unknown_outcome_retry/_unknown_summary_retry 保证只写一次）、临时态 temp: 不持久化校验、allowed-diff 审计与 unused-rule 校验、Redis 命名空间隔离、conftest 的 InMemory-only 跳过策略（仅保留 test_in_memory_only_lightweight_mode 与 test_subtle_realistic_drift_is_detected，二者均不依赖外部模型/Redis）均与 SDK 行为及 CI 配置自洽。

💡 Suggestion

• .github/workflows/ci.yml:13：步骤命名为 "Run InMemory-only replay smoke test (not full matrix)"，但保留执行的 test_subtle_realistic_drift_is_detected 实际通过 _replay_pair 同时创建并运行了 InMemory 与 SQLite 两个后端（tests/sessions/test_replay_real_agent.py:283-294）。建议修正注释/命名以避免后续维护者误判该步骤只跑内存后端。
• tests/sessions/session_memory_summary_diff_report.json:1：作为测试生成产物被提交并由 test_report_contains_every_backend_and_case 通过 _report_contract 做稳定性比对（test_replay_consistency.py:720）。DESIGN.md 已说明可重跑再生，但只要新增/调整 case 或 allowed-diff，这份大单行 JSON 即过时并阻塞 CI；建议在 PR 说明中标注其"再生步骤"或考虑改为 .gitignore + 断言仅校验结构契约，降低维护耦合。

总结

整体风险低，不存在必须修复的安全或正确性问题；新增 replay 测试与 SDK 真实接口完全对齐，CI 轻量模式与全量矩阵职责清晰。

测试建议

暂无额外测试建议。现有用例已覆盖一致性、持久化重开、注入异常检测、并发隔离与 Redis 可选集成等关键路径；若后续扩展 REPLAY_CASES，记得同步重跑并更新两份 JSON 产物以匹配 _report_contract 断言。