From 2fc7756a999923a6a668ef8d6af37fce3df4fce2 Mon Sep 17 00:00:00 2001 From: DNKYr Date: Thu, 9 Jul 2026 18:45:10 +0800 Subject: [PATCH 1/6] docs: add replay consistency harness design notes Add DESIGN.md (English) and DESIGN.zh_CN.md (Chinese) for the Session/Memory/Summary multi-backend replay consistency testing framework. Covers normalization strategy, summary comparison approach, allowed difference rules, backend integration, and harness self-testing strategy. Updates #89 RELEASE NOTES: NONE --- docs/replay_harness/DESIGN.md | 133 ++++++++++++++++++++++++++++ docs/replay_harness/DESIGN.zh_CN.md | 111 +++++++++++++++++++++++ 2 files changed, 244 insertions(+) create mode 100644 docs/replay_harness/DESIGN.md create mode 100644 docs/replay_harness/DESIGN.zh_CN.md diff --git a/docs/replay_harness/DESIGN.md b/docs/replay_harness/DESIGN.md new file mode 100644 index 00000000..418461bd --- /dev/null +++ b/docs/replay_harness/DESIGN.md @@ -0,0 +1,133 @@ +# Session / Memory / Summary Replay Consistency Harness — Design + +## Objective + +Validate that InMemory, SQL, and Redis backends for Session, Memory, and Summary +services produce equivalent results when driven by identical operation sequences. +This harness replays standardized input trajectories through multiple backends +and generates a structured diff report. + +## Architecture + +``` +replay_cases/*.json → ReplayEngine → BackendResult + │ + ┌─────────┴─────────┐ + Backend A Backend B + (InMemory) (SQL / Redis) + │ │ + BackendResult BackendResult + │ │ + └────────┬───────────┘ + ▼ + _normalizer.py + (strip timestamps, IDs, sort keys) + │ + NormalizedResult A + B + │ + ▼ + _comparator.py + (pairwise diff events, state, memory, summary) + │ + ▼ + DiffReport (JSON) +``` + +Each replay case is a JSON file describing a sequence of operations: +`create_session`, `append_event`, `update_state`, `inject_summary`, +`store_memory`, `search_memory`, `read_back`. + +The ReplayEngine executes the same sequence against two backends in parallel, +collecting raw results. These are then normalized and compared. + +## Normalization Strategy + +Fields that differ across backends by design are normalized before comparison: + +| Field | Strategy | +|------------------------|----------------------------------------------------------| +| `event.id` | Stripped (auto-generated UUID per backend) | +| `event.timestamp` | Replaced with sequential index (0, 1, 2, ...) | +| `session.last_update_time` | Replaced with sentinel `0.0` | +| `summary_timestamp` | Replaced with sentinel `0.0` | +| `memory_entry.timestamp` | Stripped | +| dict key ordering | Re-serialized with `sort_keys=True` | +| `invocation_id` | Stripped (invocation-scoped, not backend-scoped) | +| `branch`, `request_id` | Stripped (runtime metadata, not persisted uniformly) | + +## Summary Comparison Strategy + +The harness distinguishes two levels of summary correctness: + +1. **Content semantic consistency** — the `summary_text` is compared after + whitespace normalization. Minor formatting differences that do not change + meaning are treated as allowed diffs on a per-backend-pair basis. + +2. **Metadata integrity** — `session_id`, `original_event_count`, and + `compressed_event_count` must match exactly across backends. Any deviation + in these fields is an unconditional failure. Three classes of summary bug + are explicitly detected: + - **Summary loss** — summary present in backend A, absent in backend B. + - **Summary overwrite** — summary exists but with wrong `session_id`. + - **Wrong session affiliation** — summary is stored under the wrong session + key in the underlying cache or storage. + +Summary injection into the `SummarizerSessionManager._summarizer_cache` +bypasses the LLM, since the harness tests storage consistency, not +summarization model quality. + +## Allowed Differences + +Not all field-level differences indicate a bug. Known, documented divergences +are captured in `AllowedDiff` rules: + +``` +allowed_diffs = { + "inmem_vs_sql": [ + {"field": "events[*].function_calls[*].args", "reason": "SQL serializes + JSON args differently from InMemory dict round-trip"}, + ], + "inmem_vs_redis": [ + {"field": "events[*].timestamp_precision", + "reason": "Redis stores timestamps as float strings with limited precision"}, + ], +} +``` + +Each rule includes the backend pair, the field path, and a justification. +Diffs matching an allowed rule are suppressed from the failure count but +still appear in the report tagged `allowed: true`. + +## Backend Integration + +| Backend | Availability | Activation | +|----------|-----------------|---------------------------------------------| +| InMemory | Always | Direct instantiation | +| SQL | Always (sqlite) | `sqlite:///:memory:` — no external deps | +| Redis | Opt-in | `TRPC_REDIS_URL` env var; skipped otherwise | + +The lightweight mode (InMemory + SQL) runs in CI without external services. +Redis integration mode requires the environment variable to be set; when +absent, Redis test pairs are skipped with a descriptive message. + +## Diff Report + +The report (`session_memory_summary_diff_report.json`) contains: + +- **run metadata**: run ID, timestamp, backends tested. +- **per-case results**: status (pass/fail/error), list of `DiffEntry` objects. +- **summary**: total/pass/fail counts and false-positive rate. + +Each `DiffEntry` pinpoints: backend pair, session ID, event index (or summary +ID), category (events/state/memory/summary), full dotted field path, and the +two conflicting values. + +## Testing the Harness Itself + +- `test_replay_normalizer.py` — unit tests for every normalization function. +- `test_replay_comparator.py` — unit tests for diff logic, including targeted + tests for summary loss, overwrite, and mis-affiliation detection. +- `test_replay_report.py` — unit tests for report generation and serialization. +- `test_replay_consistency.py` — E2E tests running all 10 replay cases through + backend pairs, asserting normal cases produce ≤5% false positives and anomaly + cases are 100% detected. diff --git a/docs/replay_harness/DESIGN.zh_CN.md b/docs/replay_harness/DESIGN.zh_CN.md new file mode 100644 index 00000000..a2061d85 --- /dev/null +++ b/docs/replay_harness/DESIGN.zh_CN.md @@ -0,0 +1,111 @@ +# Session / Memory / Summary 回放一致性测试框架 — 设计说明 + +## 目标 + +验证 InMemory、SQL、Redis 三种后端的 Session、Memory 和 Summary 服务在相同操作序列下是否产出一致的结果。该框架使用标准化输入轨迹驱动多个后端，并生成结构化差异报告。 + +## 架构 + +``` +replay_cases/*.json → ReplayEngine → BackendResult + │ + ┌─────────┴─────────┐ + 后端 A 后端 B + (InMemory) (SQL / Redis) + │ │ + BackendResult BackendResult + │ │ + └────────┬───────────┘ + ▼ + _normalizer.py + （去除时间戳、ID，排序键） + │ + NormalizedResult A + B + │ + ▼ + _comparator.py + （逐对比较事件、状态、记忆、摘要） + │ + ▼ + DiffReport（JSON 差异报告） +``` + +每个回放用例是一个 JSON 文件，描述一系列操作： +`create_session`、`append_event`、`update_state`、`inject_summary`、 +`store_memory`、`search_memory`、`read_back`。 + +ReplayEngine 将相同的操作序列在两个后端上并行执行，收集原始结果后进行归一化和比较。 + +## 归一化策略 + +对后端间必然不同的字段，在比较前进行归一化： + +| 字段 | 策略 | +|---------------------------|-----------------------------------------------| +| `event.id` | 去除（各后端自动生成 UUID） | +| `event.timestamp` | 替换为顺序索引（0, 1, 2, ...） | +| `session.last_update_time` | 替换为哨兵值 `0.0` | +| `summary_timestamp` | 替换为哨兵值 `0.0` | +| `memory_entry.timestamp` | 去除 | +| dict 键顺序 | 使用 `sort_keys=True` 重新序列化 | +| `invocation_id` | 去除（调用范围标识，非后端相关） | +| `branch`、`request_id` | 去除（运行时元数据，非统一持久化字段） | + +## 摘要比较策略 + +框架区分两层摘要正确性： + +1. **内容语义一致性** — `summary_text` 在空白符归一化后进行比较。不影响含义的微小格式差异按后端对允许规则处理。 + +2. **元数据完整性** — `session_id`、`original_event_count`、`compressed_event_count` 必须在各后端间完全匹配。任何偏差均视为无条件失败。框架明确检测三类摘要缺陷： + - **摘要丢失** — 摘要存在于后端 A，但后端 B 中缺失。 + - **摘要覆盖错误** — 摘要存在但 `session_id` 不正确。 + - **摘要归属错误** — 摘要在底层缓存或存储中被错误地关联到不正确的 session 键。 + +摘要通过直接注入 `SummarizerSessionManager._summarizer_cache` 的方式生成，绕过 LLM，因为框架测试的是存储一致性而非摘要模型质量。 + +## 允许差异 + +并非所有字段级差异都表示缺陷。已知且有文档记录的差异通过 `AllowedDiff` 规则管理： + +``` +allowed_diffs = { + "inmem_vs_sql": [ + {"field": "events[*].function_calls[*].args", + "reason": "SQL 对 JSON 参数的序列化与 InMemory dict 往返不同"}, + ], + "inmem_vs_redis": [ + {"field": "events[*].timestamp_precision", + "reason": "Redis 将时间戳存储为精度有限的浮点字符串"}, + ], +} +``` + +每条规则包含后端对、字段路径和原因说明。匹配允许规则的差异不计入失败统计，但仍会在报告中以 `allowed: true` 标记呈现。 + +## 后端接入方式 + +| 后端 | 可用性 | 激活方式 | +|----------|--------------------|----------------------------------------------| +| InMemory | 始终可用 | 直接实例化 | +| SQL | 始终可用（sqlite） | `sqlite:///:memory:` — 无外部依赖 | +| Redis | 按需启用 | 设置 `TRPC_REDIS_URL` 环境变量；否则跳过 | + +轻量模式（InMemory + SQL）无需外部服务即可在 CI 中运行。Redis 集成模式需要设置相应的环境变量，未设置时跳过 Redis 测试对并给出明确提示。 + +## 差异报告 + +报告（`session_memory_summary_diff_report.json`）包含： + +- **运行元数据**：运行 ID、时间戳、已测试的后端列表。 +- **逐用例结果**：状态（pass/fail/error）、`DiffEntry` 列表。 +- **汇总**：总计/通过/失败数量和误报率。 + +每个 `DiffEntry` 精确定位：后端对、session ID、事件索引（或摘要 ID）、类别（events/state/memory/summary）、完整的点分隔字段路径以及两个冲突值。 + +## 框架自身的测试 + +- `test_replay_normalizer.py` — 每个归一化函数的单元测试。 +- `test_replay_comparator.py` — 差异逻辑的单元测试，包含针对摘要丢失、覆盖错误和归属错误检测的专项测试。 +- `test_replay_report.py` — 报告生成和序列化的单元测试。 +- `test_replay_consistency.py` — 端到端测试，通过后端对运行全部 10 个回放用例，验证正常用例误报率 ≤ 5%，异常用例 100% 检出。 From 0cd9d28b8746ab12306a59b48908d83e1e953139 Mon Sep 17 00:00:00 2001 From: DNKYr Date: Thu, 9 Jul 2026 18:47:34 +0800 Subject: [PATCH 2/6] tests: add replay case definitions and data models Add ReplayOp / ReplayCase Pydantic models and 10 standardized replay trajectory files covering: - 01 single-turn chat - 02 multi-turn conversation - 03 tool call (function_call + function_response) - 04 state update (write and overwrite) - 05 memory store and search - 06 summary generation and retrieval - 07 summary with event truncation - 08 error recovery (duplicate append) - 09 injected anomaly (extra event detection) - 10 injected anomaly (state corruption detection) Updates #89 RELEASE NOTES: NONE --- .../sessions/replay_cases/01_single_turn.json | 25 +++ .../sessions/replay_cases/02_multi_turn.json | 45 ++++++ tests/sessions/replay_cases/03_tool_call.json | 48 ++++++ .../replay_cases/04_state_update.json | 52 ++++++ .../replay_cases/05_memory_write_read.json | 41 +++++ .../replay_cases/06_summary_generation.json | 52 ++++++ .../replay_cases/07_summary_truncation.json | 72 +++++++++ .../replay_cases/08_error_recovery.json | 28 ++++ .../09_injected_anomaly_event.json | 35 ++++ .../10_injected_anomaly_state.json | 39 +++++ tests/sessions/replay_cases/_base.py | 151 ++++++++++++++++++ 11 files changed, 588 insertions(+) create mode 100644 tests/sessions/replay_cases/01_single_turn.json create mode 100644 tests/sessions/replay_cases/02_multi_turn.json create mode 100644 tests/sessions/replay_cases/03_tool_call.json create mode 100644 tests/sessions/replay_cases/04_state_update.json create mode 100644 tests/sessions/replay_cases/05_memory_write_read.json create mode 100644 tests/sessions/replay_cases/06_summary_generation.json create mode 100644 tests/sessions/replay_cases/07_summary_truncation.json create mode 100644 tests/sessions/replay_cases/08_error_recovery.json create mode 100644 tests/sessions/replay_cases/09_injected_anomaly_event.json create mode 100644 tests/sessions/replay_cases/10_injected_anomaly_state.json create mode 100644 tests/sessions/replay_cases/_base.py diff --git a/tests/sessions/replay_cases/01_single_turn.json b/tests/sessions/replay_cases/01_single_turn.json new file mode 100644 index 00000000..67c7f83e --- /dev/null +++ b/tests/sessions/replay_cases/01_single_turn.json @@ -0,0 +1,25 @@ +{ + "case_id": "01_single_turn", + "description": "Single turn: user sends a message and agent responds with plain text", + "session_setup": { + "app_name": "replay_test", + "user_id": "test_user", + "session_id": "s-001" + }, + "operations": [ + { + "op": "append_event", + "author": "user", + "text": "Hello, how are you?" + }, + { + "op": "append_event", + "author": "assistant", + "text": "I'm doing well, thank you for asking!" + }, + { + "op": "read_back", + "expected_event_count": 2 + } + ] +} diff --git a/tests/sessions/replay_cases/02_multi_turn.json b/tests/sessions/replay_cases/02_multi_turn.json new file mode 100644 index 00000000..b542218e --- /dev/null +++ b/tests/sessions/replay_cases/02_multi_turn.json @@ -0,0 +1,45 @@ +{ + "case_id": "02_multi_turn", + "description": "Multi-turn conversation: three rounds of user/assistant exchange", + "session_setup": { + "app_name": "replay_test", + "user_id": "test_user", + "session_id": "s-002" + }, + "operations": [ + { + "op": "append_event", + "author": "user", + "text": "What's the weather today?" + }, + { + "op": "append_event", + "author": "assistant", + "text": "It's sunny and 72 degrees." + }, + { + "op": "append_event", + "author": "user", + "text": "Great, I'll go for a walk." + }, + { + "op": "append_event", + "author": "assistant", + "text": "Enjoy your walk! Don't forget sunscreen." + }, + { + "op": "append_event", + "author": "user", + "text": "What about tomorrow?" + }, + { + "op": "append_event", + "author": "assistant", + "text": "Tomorrow will be cloudy with a chance of rain." + }, + { + "op": "read_back", + "expected_event_count": 6 + } + ] +} diff --git a/tests/sessions/replay_cases/03_tool_call.json b/tests/sessions/replay_cases/03_tool_call.json new file mode 100644 index 00000000..d30efeb2 --- /dev/null +++ b/tests/sessions/replay_cases/03_tool_call.json @@ -0,0 +1,48 @@ +{ + "case_id": "03_tool_call", + "description": "Tool call conversation: user query triggers a function_call, followed by function_response and final answer", + "session_setup": { + "app_name": "replay_test", + "user_id": "test_user", + "session_id": "s-003" + }, + "operations": [ + { + "op": "append_event", + "author": "user", + "text": "Search for Python async tutorials" + }, + { + "op": "append_event", + "author": "assistant", + "function_call": { + "name": "web_search", + "args": { + "query": "Python async tutorials" + } + } + }, + { + "op": "append_event", + "author": "user", + "function_response": { + "name": "web_search", + "response": { + "results": [ + "AsyncIO Tutorial for Beginners", + "Advanced Async Patterns in Python" + ] + } + } + }, + { + "op": "append_event", + "author": "assistant", + "text": "I found two relevant tutorials: AsyncIO Tutorial for Beginners and Advanced Async Patterns in Python." + }, + { + "op": "read_back", + "expected_event_count": 4 + } + ] +} diff --git a/tests/sessions/replay_cases/04_state_update.json b/tests/sessions/replay_cases/04_state_update.json new file mode 100644 index 00000000..94d20367 --- /dev/null +++ b/tests/sessions/replay_cases/04_state_update.json @@ -0,0 +1,52 @@ +{ + "case_id": "04_state_update", + "description": "State update: multiple writes and overwrites to session state via event state_delta", + "session_setup": { + "app_name": "replay_test", + "user_id": "test_user", + "session_id": "s-004", + "state": { + "counter": 0, + "status": "idle" + } + }, + "operations": [ + { + "op": "append_event", + "author": "user", + "text": "Increment counter", + "state_delta": { + "counter": 1 + } + }, + { + "op": "append_event", + "author": "assistant", + "text": "Counter is now 1" + }, + { + "op": "append_event", + "author": "user", + "text": "Update status and counter", + "state_delta": { + "counter": 5, + "status": "active", + "last_action": "bulk_update" + } + }, + { + "op": "append_event", + "author": "assistant", + "text": "Updated to counter=5, status=active" + }, + { + "op": "read_back", + "expected_state": { + "counter": 5, + "status": "active", + "last_action": "bulk_update" + }, + "expected_event_count": 4 + } + ] +} diff --git a/tests/sessions/replay_cases/05_memory_write_read.json b/tests/sessions/replay_cases/05_memory_write_read.json new file mode 100644 index 00000000..274aa022 --- /dev/null +++ b/tests/sessions/replay_cases/05_memory_write_read.json @@ -0,0 +1,41 @@ +{ + "case_id": "05_memory_write_read", + "description": "Memory write and read: store session content in memory and search for stored facts", + "session_setup": { + "app_name": "replay_test", + "user_id": "test_user", + "session_id": "s-005" + }, + "operations": [ + { + "op": "append_event", + "author": "user", + "text": "My name is Alice and my favorite food is pizza." + }, + { + "op": "append_event", + "author": "assistant", + "text": "Nice to meet you, Alice! Pizza is a great choice." + }, + { + "op": "append_event", + "author": "user", + "text": "I also enjoy hiking on weekends." + }, + { + "op": "append_event", + "author": "assistant", + "text": "Hiking is wonderful exercise. Where do you usually hike?" + }, + { + "op": "store_memory" + }, + { + "op": "search_memory", + "query": "What food does Alice like?" + }, + { + "op": "read_back" + } + ] +} diff --git a/tests/sessions/replay_cases/06_summary_generation.json b/tests/sessions/replay_cases/06_summary_generation.json new file mode 100644 index 00000000..8da3887c --- /dev/null +++ b/tests/sessions/replay_cases/06_summary_generation.json @@ -0,0 +1,52 @@ +{ + "case_id": "06_summary_generation", + "description": "Summary generation and retrieval: inject a session summary and verify its content and metadata", + "session_setup": { + "app_name": "replay_test", + "user_id": "test_user", + "session_id": "s-006" + }, + "operations": [ + { + "op": "append_event", + "author": "user", + "text": "Let's discuss the new project." + }, + { + "op": "append_event", + "author": "assistant", + "text": "Sure, what kind of project are you planning?" + }, + { + "op": "append_event", + "author": "user", + "text": "A web application for task management with real-time collaboration." + }, + { + "op": "append_event", + "author": "assistant", + "text": "That sounds ambitious. We should use React for the frontend and WebSocket for real-time features." + }, + { + "op": "append_event", + "author": "user", + "text": "Agreed. Let's also add file sharing capabilities." + }, + { + "op": "append_event", + "author": "assistant", + "text": "Good idea. We can integrate cloud storage APIs for that." + }, + { + "op": "inject_summary", + "session_id": "s-006", + "summary_text": "The user and assistant are planning a task management web application. Key decisions: use React frontend, WebSocket for real-time collaboration, and integrate cloud storage APIs for file sharing.", + "original_event_count": 6, + "compressed_event_count": 2 + }, + { + "op": "read_back", + "expected_event_count": 6 + } + ] +} diff --git a/tests/sessions/replay_cases/07_summary_truncation.json b/tests/sessions/replay_cases/07_summary_truncation.json new file mode 100644 index 00000000..a1876727 --- /dev/null +++ b/tests/sessions/replay_cases/07_summary_truncation.json @@ -0,0 +1,72 @@ +{ + "case_id": "07_summary_truncation", + "description": "Summary with event truncation: inject a summary to compress old events, verify summary event is present alongside remaining recent events", + "session_setup": { + "app_name": "replay_test", + "user_id": "test_user", + "session_id": "s-007" + }, + "operations": [ + { + "op": "append_event", + "author": "user", + "text": "message 1 - beginning of long conversation" + }, + { + "op": "append_event", + "author": "assistant", + "text": "response 1" + }, + { + "op": "append_event", + "author": "user", + "text": "message 2 - more context" + }, + { + "op": "append_event", + "author": "assistant", + "text": "response 2" + }, + { + "op": "append_event", + "author": "user", + "text": "message 3 - still more" + }, + { + "op": "append_event", + "author": "assistant", + "text": "response 3" + }, + { + "op": "append_event", + "author": "user", + "text": "message 4 - getting closer" + }, + { + "op": "append_event", + "author": "assistant", + "text": "response 4" + }, + { + "op": "append_event", + "author": "user", + "text": "message 5 - recent question about the plan" + }, + { + "op": "append_event", + "author": "assistant", + "text": "response 5 - answering the recent question" + }, + { + "op": "inject_summary", + "session_id": "s-007", + "summary_text": "Earlier in the conversation, the user and assistant discussed various topics including messages 1 through 4.", + "original_event_count": 10, + "compressed_event_count": 3 + }, + { + "op": "read_back", + "expected_event_count": 10 + } + ] +} diff --git a/tests/sessions/replay_cases/08_error_recovery.json b/tests/sessions/replay_cases/08_error_recovery.json new file mode 100644 index 00000000..cdded2c2 --- /dev/null +++ b/tests/sessions/replay_cases/08_error_recovery.json @@ -0,0 +1,28 @@ +{ + "case_id": "08_error_recovery", + "description": "Error recovery: duplicate event append must not produce duplicated events in the session", + "session_setup": { + "app_name": "replay_test", + "user_id": "test_user", + "session_id": "s-008" + }, + "operations": [ + { + "op": "append_event", + "author": "user", + "text": "Normal message" + }, + { + "op": "append_event", + "author": "assistant", + "text": "Normal response" + }, + { + "op": "duplicate_append" + }, + { + "op": "read_back", + "expected_event_count": 2 + } + ] +} diff --git a/tests/sessions/replay_cases/09_injected_anomaly_event.json b/tests/sessions/replay_cases/09_injected_anomaly_event.json new file mode 100644 index 00000000..333e4e6e --- /dev/null +++ b/tests/sessions/replay_cases/09_injected_anomaly_event.json @@ -0,0 +1,35 @@ +{ + "case_id": "09_injected_anomaly_event", + "description": "Anomaly detection: an extra event is injected into one backend's result. The comparator must detect this difference.", + "session_setup": { + "app_name": "replay_test", + "user_id": "test_user", + "session_id": "s-009" + }, + "operations": [ + { + "op": "append_event", + "author": "user", + "text": "Hello" + }, + { + "op": "append_event", + "author": "assistant", + "text": "Hi there" + }, + { + "op": "read_back", + "expected_event_count": 2 + } + ], + "expect_pass": false, + "inject_anomaly": { + "category": "events", + "action": "insert_extra", + "event_index": 1, + "extra_event": { + "author": "intruder", + "text": "This event should not exist" + } + } +} diff --git a/tests/sessions/replay_cases/10_injected_anomaly_state.json b/tests/sessions/replay_cases/10_injected_anomaly_state.json new file mode 100644 index 00000000..d05c6db6 --- /dev/null +++ b/tests/sessions/replay_cases/10_injected_anomaly_state.json @@ -0,0 +1,39 @@ +{ + "case_id": "10_injected_anomaly_state", + "description": "Anomaly detection: session state is corrupted in one backend. The comparator must detect the state discrepancy.", + "session_setup": { + "app_name": "replay_test", + "user_id": "test_user", + "session_id": "s-010", + "state": { + "version": "1.0", + "env": "production" + } + }, + "operations": [ + { + "op": "append_event", + "author": "user", + "text": "Update configuration", + "state_delta": { + "version": "2.0", + "env": "production" + } + }, + { + "op": "read_back", + "expected_state": { + "version": "2.0", + "env": "production" + }, + "expected_event_count": 1 + } + ], + "expect_pass": false, + "inject_anomaly": { + "category": "state", + "action": "mutate_value", + "field_path": "version", + "corrupted_value": "9.9" + } +} diff --git a/tests/sessions/replay_cases/_base.py b/tests/sessions/replay_cases/_base.py new file mode 100644 index 00000000..cad743d7 --- /dev/null +++ b/tests/sessions/replay_cases/_base.py @@ -0,0 +1,151 @@ +# +# 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 the Apache License Version 2.0. +# +"""Replay case data models and JSON loader for multi-backend consistency testing.""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Optional + +from pydantic import BaseModel +from pydantic import Field + + +class ReplayOp(BaseModel): + """A single operation in a replay case. + + Supports the following operation types: + + - ``create_session``: initialise a session with ``app_name``, ``user_id``, + optional ``session_id`` and ``initial_state`` (provided via the + ``session_setup`` stanza on the enclosing ReplayCase). + + - ``append_event``: construct and append an Event. Uses ``author``, + ``text``, optional ``function_call`` / ``function_response`` dicts, + optional ``state_delta``, and ``partial`` flag. + + - ``update_state``: convenience alias that appends an event carrying a + ``state_delta``. + + - ``inject_summary``: bypass the LLM summarizer and insert a + ``SessionSummary`` directly into ``SummarizerSessionManager._summarizer_cache``. + + - ``store_memory``: call ``memory_service.store_session()``. + + - ``search_memory``: call ``memory_service.search_memory()`` with ``query``. + + - ``read_back``: call ``session_service.get_session()`` and capture the + returned events, state, and summary. + + - ``duplicate_append``: re-append the last-non-read_back event to test + idempotency / duplicate detection. + + - ``delete_session``: delete the current session. + """ + op: str + """Operation identifier.""" + + author: str = "" + """Event author (``"user"`` or agent name).""" + + text: str = "" + """Plain text content for the event.""" + + function_call: Optional[dict] = None + """Dictionary with ``name`` and ``args`` keys for a FunctionCall.""" + + function_response: Optional[dict] = None + """Dictionary with ``name`` and ``response`` keys for a FunctionResponse.""" + + state_delta: dict = Field(default_factory=dict) + """Key-value pairs to write as ``EventActions.state_delta``.""" + + session_id: str = "" + """Explicit session id (used by ``create_session`` and ``inject_summary``).""" + + summary_text: str = "" + """Summary body text for ``inject_summary``.""" + + original_event_count: int = 0 + """Event count before summarization (``inject_summary``).""" + + compressed_event_count: int = 0 + """Event count after summarization (``inject_summary``).""" + + query: str = "" + """Search query string for ``search_memory``.""" + + partial: bool = False + """Mark the constructed Event as partial.""" + + expected_event_count: Optional[int] = None + """If set, the read_back result should have this many events.""" + + expected_state: Optional[dict] = None + """If set, the session state after read_back must match this dict.""" + + +class ReplayCase(BaseModel): + """A complete replay case definition. + + Contains a set of session-creation parameters and a sequenced list of + operations to execute. + """ + case_id: str + """Unique case identifier, e.g. ``"01_single_turn"``.""" + + description: str + """Human-readable description of what this case validates.""" + + session_setup: dict = Field(default_factory=dict) + """Keyword arguments forwarded to ``session_service.create_session()``.""" + + operations: list[ReplayOp] = Field(default_factory=list) + """Ordered list of replay operations.""" + + expect_pass: bool = True + """When ``True`` the case is expected to produce ≤ 5 % diffs. + + Set to ``False`` for injected-anomaly cases that must be detected. + """ + + inject_anomaly: Optional[dict] = None + """Test-only specification for injecting artificial inconsistency. + + The engine applies this spec to *one* of the two backend results before + comparison so that the comparator's sensitivity can be verified. + + Example:: + + { + "category": "events", + "action": "insert_extra", + "event_index": 2, + "extra_event": { "author": "hacker", "text": "injected" } + } + """ + + +def load_replay_cases(glob_pattern: str = "*.json") -> list[ReplayCase]: + """Load all replay case JSON files from the ``replay_cases`` directory. + + Returns: + List of ``ReplayCase`` instances sorted by ``case_id``. + """ + cases_dir = Path(__file__).parent + cases: list[ReplayCase] = [] + + for filepath in sorted(cases_dir.glob(glob_pattern)): + if filepath.name.startswith("_"): + continue + with open(filepath, "r", encoding="utf-8") as fh: + data = json.load(fh) + cases.append(ReplayCase(**data)) + + return cases From f644afa836207ec5076658e9637547b975a00845 Mon Sep 17 00:00:00 2001 From: DNKYr Date: Thu, 9 Jul 2026 18:53:29 +0800 Subject: [PATCH 3/6] tests: add replay harness normalizer and unit tests Implement normalization utilities that strip auto-generated and backend-dependent fields (timestamps, UUIDs, invocation IDs) from events, state, summaries, and memory entries before comparison. 24 unit tests cover: event field stripping, timestamp-to-index replacement, function_call/response extraction, state deep-sort determinism, summary timestamp removal, memory entry normalization, and the full normalize_backend_result orchestrator. Updates #89 RELEASE NOTES: NONE --- tests/sessions/replay_harness/_normalizer.py | 179 +++++++++++ tests/sessions/test_replay_normalizer.py | 299 +++++++++++++++++++ 2 files changed, 478 insertions(+) create mode 100644 tests/sessions/replay_harness/_normalizer.py create mode 100644 tests/sessions/test_replay_normalizer.py diff --git a/tests/sessions/replay_harness/_normalizer.py b/tests/sessions/replay_harness/_normalizer.py new file mode 100644 index 00000000..2ff343ac --- /dev/null +++ b/tests/sessions/replay_harness/_normalizer.py @@ -0,0 +1,179 @@ +# +# 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 the Apache License Version 2.0. +# +"""Normalization utilities for replay consistency testing. + +Strips auto-generated and backend-dependent fields so that results from +different backends can be compared meaningfully. +""" + +from __future__ import annotations + +import json +from typing import Any + +from pydantic import BaseModel +from pydantic import Field + + +class NormalizedResult(BaseModel): + """Normalized view of a backend result suitable for comparison.""" + + events: list[dict] = Field(default_factory=list) + """Normalized event dicts with auto-generated fields stripped.""" + + state: dict = Field(default_factory=dict) + """Deep-sorted state dictionary.""" + + summaries: list[dict] = Field(default_factory=list) + """Normalized summary dicts with timestamps stripped.""" + + memory_entries: list[dict] = Field(default_factory=list) + """Normalized memory entry dicts with timestamps stripped.""" + + errors: list[str] = Field(default_factory=list) + """Errors encountered during replay (kept as-is).""" + + +def _sort_dict_deep(obj: Any) -> Any: + """Recursively sort all dict keys for deterministic serialization.""" + if isinstance(obj, dict): + return {k: _sort_dict_deep(v) for k, v in sorted(obj.items())} + if isinstance(obj, list): + return [_sort_dict_deep(v) for v in obj] + return obj + + +def normalize_events(events: list[dict]) -> list[dict]: + """Normalize a list of event dicts. + + Strips auto-generated / backend-dependent fields (``id``, ``timestamp``, + ``invocation_id``, etc.) and replaces ``timestamp`` with a sequential + index. Extracts text, function calls, function responses, and state + deltas into a canonical form. + """ + normalized: list[dict] = [] + for idx, event in enumerate(events): + norm: dict[str, Any] = {} + + norm["index"] = idx + norm["author"] = event.get("author", "") + + content = event.get("content") or {} + parts = content.get("parts") if isinstance(content, dict) else [] + + texts = [] + func_calls = [] + func_responses = [] + + if isinstance(parts, list): + for part in parts: + if not isinstance(part, dict): + continue + if part.get("text"): + texts.append(part["text"]) + if part.get("function_call"): + fc = part["function_call"] + func_calls.append({ + "name": fc.get("name", ""), + "args": fc.get("args", {}), + }) + if part.get("function_response"): + fr = part["function_response"] + func_responses.append({ + "name": fr.get("name", ""), + "response": fr.get("response", {}), + }) + + norm["text"] = "".join(texts) + norm["function_calls"] = func_calls + norm["function_responses"] = func_responses + + actions = event.get("actions") or {} + if isinstance(actions, dict): + norm["state_delta"] = actions.get("state_delta") or {} + else: + norm["state_delta"] = {} + + norm["partial"] = event.get("partial", False) + norm["visible"] = event.get("visible", True) + norm["error_code"] = event.get("error_code") + norm["error_message"] = event.get("error_message") + + normalized.append(norm) + + return normalized + + +def normalize_state(state: dict) -> dict: + """Normalize a state dictionary by deep-sorting keys. + + Returns a deterministic representation suitable for deep comparison. + """ + sorted_state = _sort_dict_deep(state) + return json.loads(json.dumps(sorted_state, sort_keys=True)) + + +def normalize_summaries(summaries: list[dict]) -> list[dict]: + """Normalize summary dicts. + + Keeps content and structural metadata; strips ``summary_timestamp`` + and any backend-specific metadata. + """ + normalized: list[dict] = [] + for summary in summaries: + if not isinstance(summary, dict): + continue + norm: dict[str, Any] = { + "session_id": summary.get("session_id", ""), + "summary_text": summary.get("summary_text", ""), + "original_event_count": summary.get("original_event_count", 0), + "compressed_event_count": summary.get("compressed_event_count", 0), + } + normalized.append(norm) + return normalized + + +def normalize_memory_entries(entries: list[dict]) -> list[dict]: + """Normalize memory entry dicts. + + Keeps content text and author; strips timestamp. + """ + normalized: list[dict] = [] + for entry in entries: + if not isinstance(entry, dict): + continue + content = entry.get("content") or {} + text = "" + if isinstance(content, dict): + parts = content.get("parts") or [] + if isinstance(parts, list): + text = "".join( + p.get("text", "") for p in parts if isinstance(p, dict)) + norm: dict[str, Any] = { + "content_text": text, + "author": entry.get("author", ""), + } + normalized.append(norm) + return normalized + + +def normalize_backend_result( + events: list[dict], + state: dict, + summaries: list[dict], + memory_entries: list[dict], + errors: list[str], +) -> NormalizedResult: + """Normalize a complete backend result for comparison.""" + return NormalizedResult( + events=normalize_events(events), + state=normalize_state(state), + summaries=normalize_summaries(summaries), + memory_entries=normalize_memory_entries(memory_entries), + errors=errors, + ) diff --git a/tests/sessions/test_replay_normalizer.py b/tests/sessions/test_replay_normalizer.py new file mode 100644 index 00000000..90c2bb90 --- /dev/null +++ b/tests/sessions/test_replay_normalizer.py @@ -0,0 +1,299 @@ +# +# 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 the Apache License Version 2.0. +# +"""Unit tests for the replay harness normalizer.""" + +from __future__ import annotations + +from tests.sessions.replay_harness._normalizer import ( + _sort_dict_deep, + normalize_backend_result, + normalize_events, + normalize_memory_entries, + normalize_state, + normalize_summaries, +) +from tests.sessions.replay_harness._normalizer import NormalizedResult + + +def _make_event_dict(**overrides): + """Build a realistic raw event dict as returned by Event.model_dump().""" + event = { + "id": "evt-aaa-111", + "timestamp": 1752000123.456, + "invocation_id": "inv-xyz", + "author": "assistant", + "branch": "agent.sub", + "request_id": "req-999", + "parent_invocation_id": "parent-inv-1", + "tag": "some-tag", + "filter_key": "a.b.c", + "object": None, + "partial": False, + "visible": True, + "version": 0, + "requires_completion": False, + "error_code": None, + "error_message": None, + "actions": { + "state_delta": {}, + "artifact_delta": {}, + "skip_summarization": False, + "transfer_to_agent": None, + "escalate": None, + }, + "content": { + "parts": [{"text": "Hello world"}], + "role": "model", + }, + "model_flags": 1, + "long_running_tool_ids": None, + } + event.update(overrides) + return event + + +# ── _sort_dict_deep ──────────────────────────────────────────────────── + + +class TestSortDictDeep: + + def test_sorts_top_level_keys(self): + assert list(_sort_dict_deep({"b": 1, "a": 2})) == ["a", "b"] + + def test_recursively_sorts_nested_dicts(self): + result = _sort_dict_deep({"z": {"b": 1, "a": 2}, "y": {"d": 4, "c": 3}}) + assert list(result) == ["y", "z"] + assert list(result["z"]) == ["a", "b"] + + def test_handles_lists(self): + result = _sort_dict_deep({"items": [{"b": 2, "a": 1}, {"d": 4, "c": 3}]}) + assert list(result["items"][0]) == ["a", "b"] + assert list(result["items"][1]) == ["c", "d"] + + def test_passes_through_primitives(self): + assert _sort_dict_deep(42) == 42 + assert _sort_dict_deep("hello") == "hello" + assert _sort_dict_deep(None) is None + + +# ── normalize_events ─────────────────────────────────────────────────── + + +class TestNormalizeEvents: + + def test_strips_auto_generated_fields(self): + events = [_make_event_dict()] + result = normalize_events(events) + assert len(result) == 1 + norm = result[0] + assert "id" not in norm + assert "timestamp" not in norm + assert "invocation_id" not in norm + assert "branch" not in norm + assert "request_id" not in norm + + def test_replaces_timestamp_with_sequential_index(self): + events = [_make_event_dict(), _make_event_dict(), _make_event_dict()] + result = normalize_events(events) + assert [e["index"] for e in result] == [0, 1, 2] + + def test_empty_event_list_returns_empty(self): + assert normalize_events([]) == [] + + def test_preserves_author_and_text(self): + events = [_make_event_dict(author="user", content={"parts": [{"text": "Hi"}]})] + result = normalize_events(events) + assert result[0]["author"] == "user" + assert result[0]["text"] == "Hi" + + def test_extracts_function_calls(self): + events = [ + _make_event_dict(content={ + "parts": [{ + "function_call": { + "name": "search", + "args": {"query": "test"} + } + }] + }) + ] + result = normalize_events(events) + assert result[0]["function_calls"] == [ + {"name": "search", "args": {"query": "test"}} + ] + + def test_extracts_function_responses(self): + events = [ + _make_event_dict(content={ + "parts": [{ + "function_response": { + "name": "get_temp", + "response": {"celsius": 22} + } + }] + }) + ] + result = normalize_events(events) + assert result[0]["function_responses"] == [ + {"name": "get_temp", "response": {"celsius": 22}} + ] + + def test_preserves_state_delta(self): + events = [ + _make_event_dict(actions={"state_delta": {"counter": 5}}) + ] + result = normalize_events(events) + assert result[0]["state_delta"] == {"counter": 5} + + def test_preserves_partial_and_visible_flags(self): + events = [ + _make_event_dict(partial=True, visible=False) + ] + result = normalize_events(events) + assert result[0]["partial"] is True + assert result[0]["visible"] is False + + def test_concatentates_multiple_text_parts(self): + events = [ + _make_event_dict(content={ + "parts": [{"text": "Hello "}, {"text": "world"}] + }) + ] + result = normalize_events(events) + assert result[0]["text"] == "Hello world" + + +# ── normalize_state ──────────────────────────────────────────────────── + + +class TestNormalizeState: + + def test_sorted_keys_deterministic(self): + a = {"b": 1, "a": {"d": 2, "c": 3}} + b = {"a": {"c": 3, "d": 2}, "b": 1} + assert normalize_state(a) == normalize_state(b) + + def test_deeply_nested_normalized(self): + state = {"level1": {"level2": {"z": 9, "a": 1}}} + result = normalize_state(state) + assert list(result["level1"]["level2"]) == ["a", "z"] + + def test_empty_state(self): + assert normalize_state({}) == {} + assert normalize_state({}) == {} + + +# ── normalize_summaries ──────────────────────────────────────────────── + + +class TestNormalizeSummaries: + + def test_strips_timestamp_preserves_content(self): + summaries = [ + { + "session_id": "s-1", + "summary_text": "A summary.", + "original_event_count": 10, + "compressed_event_count": 3, + "summary_timestamp": 1752000999.0, + "metadata": {"model": "gpt-4"}, + } + ] + result = normalize_summaries(summaries) + assert result[0] == { + "session_id": "s-1", + "summary_text": "A summary.", + "original_event_count": 10, + "compressed_event_count": 3, + } + + def test_preserves_event_counts(self): + summaries = [ + { + "session_id": "s-2", + "summary_text": "x", + "original_event_count": 42, + "compressed_event_count": 7, + } + ] + result = normalize_summaries(summaries) + assert result[0]["original_event_count"] == 42 + assert result[0]["compressed_event_count"] == 7 + + def test_preserves_session_id(self): + summaries = [{ + "session_id": "abc-123", + "summary_text": "", + "original_event_count": 0, + "compressed_event_count": 0, + }] + result = normalize_summaries(summaries) + assert result[0]["session_id"] == "abc-123" + + def test_handles_empty_list(self): + assert normalize_summaries([]) == [] + + +# ── normalize_memory_entries ─────────────────────────────────────────── + + +class TestNormalizeMemory: + + def test_strips_timestamp_preserves_content_and_author(self): + entries = [ + { + "content": {"parts": [{"text": "alice likes pizza"}], "role": "user"}, + "author": "user", + "timestamp": "2025-01-01T00:00:00Z", + } + ] + result = normalize_memory_entries(entries) + assert result[0] == { + "content_text": "alice likes pizza", + "author": "user", + } + + def test_empty_list(self): + assert normalize_memory_entries([]) == [] + + def test_multiple_entries_preserved(self): + entries = [ + {"content": {"parts": [{"text": "a"}]}, "author": "u1"}, + {"content": {"parts": [{"text": "b"}]}, "author": "u2"}, + ] + result = normalize_memory_entries(entries) + assert len(result) == 2 + assert result[0]["content_text"] == "a" + assert result[1]["content_text"] == "b" + + +# ── normalize_backend_result ─────────────────────────────────────────── + + +class TestNormalizeBackendResult: + + def test_returns_normalized_result(self): + events = [_make_event_dict(author="user", content={"parts": [{"text": "Q"}]})] + result = normalize_backend_result( + events=events, + state={"k": "v"}, + summaries=[{ + "session_id": "s", + "summary_text": "sum", + "original_event_count": 1, + "compressed_event_count": 1, + "summary_timestamp": 999, + }], + memory_entries=[], + errors=["err1"], + ) + assert isinstance(result, NormalizedResult) + assert len(result.events) == 1 + assert result.state == {"k": "v"} + assert len(result.summaries) == 1 + assert result.errors == ["err1"] From f699f281a6ca2cb433b1329c46424c2a41c1deb0 Mon Sep 17 00:00:00 2001 From: DNKYr Date: Thu, 9 Jul 2026 18:56:07 +0800 Subject: [PATCH 4/6] tests: add replay harness comparator and unit tests Implement pairwise comparison of normalized backend results with field-level precision. Covers events (author, text, function calls, responses, state deltas), state (deep recursive diff), summaries (loss, overwrite, wrong-session-affiliation detection), and memory. Includes AllowedDiff rules to suppress known-acceptable backend differences via pattern matching. 30 unit tests verify: identical detection, count mismatches, field- level diffs, event-index precision, nested state diff paths, summary loss / overwrite / mis-affiliation, memory diffs, and allow-rule suppression with backend-pair scoping. Updates #89 RELEASE NOTES: NONE --- tests/sessions/replay_harness/_comparator.py | 450 +++++++++++++++++++ tests/sessions/test_replay_comparator.py | 308 +++++++++++++ 2 files changed, 758 insertions(+) create mode 100644 tests/sessions/replay_harness/_comparator.py create mode 100644 tests/sessions/test_replay_comparator.py diff --git a/tests/sessions/replay_harness/_comparator.py b/tests/sessions/replay_harness/_comparator.py new file mode 100644 index 00000000..e57aa5f8 --- /dev/null +++ b/tests/sessions/replay_harness/_comparator.py @@ -0,0 +1,450 @@ +# +# 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 the Apache License Version 2.0. +# +"""Comparator for replay consistency testing. + +Compares two ``NormalizedResult`` objects and produces a list of +``DiffEntry`` items pinpointing every discrepancy with field-level +precision. +""" + +from __future__ import annotations + +from typing import Any +from typing import Optional + +from pydantic import BaseModel +from pydantic import Field + +from ._normalizer import NormalizedResult + + +class DiffEntry(BaseModel): + """A single discrepancy between two backend results.""" + + backend_pair: tuple[str, str] + """The two backends being compared, e.g. ``("in_memory", "sql")``.""" + + category: str + """Comparison category: ``"events"``, ``"state"``, ``"memory"``, ``"summary"``.""" + + session_id: str = "" + """The session in which the diff was found.""" + + event_index: Optional[int] = None + """Zero-based index within the events list (if applicable).""" + + summary_id: Optional[str] = None + """The ``session_id`` of the summary (if applicable).""" + + field_path: str = "" + """Dotted path to the differing field, e.g. ``"events[2].function_calls[0].name"``.""" + + value_a: Any = None + """Value from backend A.""" + + value_b: Any = None + """Value from backend B.""" + + allowed: bool = False + """Whether this diff is covered by an allow-rule.""" + + +class AllowedDiffRule(BaseModel): + """A single rule that suppresses a known-acceptable difference.""" + + field_path_pattern: str + """Glob-like pattern, e.g. ``"events[*].function_calls[*].args"``.""" + + reason: str + """Human-readable justification.""" + + backend_pairs: list[tuple[str, str]] = Field(default_factory=list) + """Backend pairs this rule applies to. Empty means all pairs.""" + + +class AllowedDiff(BaseModel): + """Collection of allow-rules for known backend differences.""" + + rules: list[AllowedDiffRule] = Field(default_factory=list) + + +# ── internal helpers ─────────────────────────────────────────────────── + + +def _matches_pattern(field_path: str, pattern: str) -> bool: + """Check whether *field_path* matches a pattern with ``[*]`` index wildcards.""" + import re + + escaped = re.escape(pattern) + escaped = escaped.replace(r"\[\*\]", r"\[\d+\]") + return bool(re.match("^" + escaped + "$", field_path)) + + +def _is_allowed(entry: DiffEntry, allowed: AllowedDiff) -> bool: + """Return ``True`` if *entry* is covered by any allow-rule.""" + for rule in allowed.rules: + if rule.backend_pairs and entry.backend_pair not in rule.backend_pairs: + continue + if _matches_pattern(entry.field_path, rule.field_path_pattern): + return True + return False + + +def _deep_diff(a: Any, + b: Any, + prefix: str = "", + session_id: str = "", + backend_pair: tuple[str, str] = ("a", "b")) -> list[DiffEntry]: + """Recursively diff two values, yielding ``DiffEntry`` items.""" + diffs: list[DiffEntry] = [] + + if type(a) is not type(b): + diffs.append( + DiffEntry( + backend_pair=backend_pair, + category="state", + session_id=session_id, + field_path=prefix or "(root)", + value_a=str(a), + value_b=str(b), + )) + return diffs + + if isinstance(a, dict): + all_keys = set(a) | set(b) + for key in sorted(all_keys): + child_path = f"{prefix}.{key}" if prefix else key + va = a.get(key, _MISSING) + vb = b.get(key, _MISSING) + if va is _MISSING: + diffs.append( + DiffEntry( + backend_pair=backend_pair, + category="state", + session_id=session_id, + field_path=child_path, + value_a="", + value_b=vb, + )) + elif vb is _MISSING: + diffs.append( + DiffEntry( + backend_pair=backend_pair, + category="state", + session_id=session_id, + field_path=child_path, + value_a=va, + value_b="", + )) + else: + diffs.extend(_deep_diff(va, vb, child_path, session_id, backend_pair)) + elif isinstance(a, list): + max_len = max(len(a), len(b)) + for i in range(max_len): + child_path = f"{prefix}[{i}]" + va = a[i] if i < len(a) else _MISSING + vb = b[i] if i < len(b) else _MISSING + if va is _MISSING or vb is _MISSING: + diffs.append( + DiffEntry( + backend_pair=backend_pair, + category="state", + session_id=session_id, + field_path=child_path, + value_a=va if va is not _MISSING else "", + value_b=vb if vb is not _MISSING else "", + )) + else: + diffs.extend(_deep_diff(va, vb, child_path, session_id, backend_pair)) + elif a != b: + diffs.append( + DiffEntry( + backend_pair=backend_pair, + category="state", + session_id=session_id, + field_path=prefix, + value_a=a, + value_b=b, + )) + + return diffs + + +class _MISSING: + """Sentinel for missing keys.""" + + pass + + +# ── per-category diff functions ──────────────────────────────────────── + + +def _diff_events( + events_a: list[dict], + events_b: list[dict], + session_id: str = "", + backend_pair: tuple[str, str] = ("a", "b"), +) -> list[DiffEntry]: + """Compare two normalized event lists pairwise by index.""" + diffs: list[DiffEntry] = [] + + max_len = max(len(events_a), len(events_b)) + if len(events_a) != len(events_b): + diffs.append( + DiffEntry( + backend_pair=backend_pair, + category="events", + session_id=session_id, + field_path="events.length", + value_a=len(events_a), + value_b=len(events_b), + )) + + scalar_fields = [ + "author", + "text", + "partial", + "visible", + "error_code", + "error_message", + ] + + for i in range(max_len): + ea = events_a[i] if i < len(events_a) else {} + eb = events_b[i] if i < len(events_b) else {} + + if not ea or not eb: + diffs.append( + DiffEntry( + backend_pair=backend_pair, + category="events", + session_id=session_id, + event_index=i, + field_path=f"events[{i}]", + value_a="" if not ea else "", + value_b="" if not eb else "", + )) + continue + + for field in scalar_fields: + va = ea.get(field) + vb = eb.get(field) + if va != vb: + diffs.append( + DiffEntry( + backend_pair=backend_pair, + category="events", + session_id=session_id, + event_index=i, + field_path=f"events[{i}].{field}", + value_a=va, + value_b=vb, + )) + + fa = ea.get("function_calls") or [] + fb = eb.get("function_calls") or [] + if fa != fb: + diffs.append( + DiffEntry( + backend_pair=backend_pair, + category="events", + session_id=session_id, + event_index=i, + field_path=f"events[{i}].function_calls", + value_a=fa, + value_b=fb, + )) + + fra = ea.get("function_responses") or [] + frb = eb.get("function_responses") or [] + if fra != frb: + diffs.append( + DiffEntry( + backend_pair=backend_pair, + category="events", + session_id=session_id, + event_index=i, + field_path=f"events[{i}].function_responses", + value_a=fra, + value_b=frb, + )) + + sa = ea.get("state_delta") or {} + sb = eb.get("state_delta") or {} + field_prefix = f"events[{i}].state_delta" + if sa != sb: + diffs.extend( + _deep_diff(sa, sb, prefix=field_prefix, session_id=session_id, backend_pair=backend_pair)) + + return diffs + + +def _diff_state( + state_a: dict, + state_b: dict, + session_id: str = "", + backend_pair: tuple[str, str] = ("a", "b"), +) -> list[DiffEntry]: + """Deep-diff two state dictionaries.""" + return _deep_diff(state_a, state_b, prefix="state", session_id=session_id, backend_pair=backend_pair) + + +def _diff_summaries( + summaries_a: list[dict], + summaries_b: list[dict], + backend_pair: tuple[str, str] = ("a", "b"), +) -> list[DiffEntry]: + """Compare two normalized summary lists. + + Summaries are matched by ``session_id``. Detects: + + * **summary loss** — present in A, absent in B (or vice versa). + * **summary overwrite** — same session but different ``summary_text``. + * **wrong session affiliation** — session_id mismatch between entries. + """ + diffs: list[DiffEntry] = [] + + index_a: dict[str, dict] = {s.get("session_id", ""): s for s in summaries_a} + index_b: dict[str, dict] = {s.get("session_id", ""): s for s in summaries_b} + + all_ids = set(index_a) | set(index_b) + + for sid in sorted(all_ids): + sa = index_a.get(sid) + sb = index_b.get(sid) + + if sa is None: + diffs.append( + DiffEntry( + backend_pair=backend_pair, + category="summary", + summary_id=sid, + field_path=f"summaries[{sid}]", + value_a="", + value_b="", + )) + continue + if sb is None: + diffs.append( + DiffEntry( + backend_pair=backend_pair, + category="summary", + summary_id=sid, + field_path=f"summaries[{sid}]", + value_a="", + value_b="", + )) + continue + + for field in ["summary_text", "session_id", "original_event_count", "compressed_event_count"]: + va = sa.get(field) + vb = sb.get(field) + if va != vb: + diffs.append( + DiffEntry( + backend_pair=backend_pair, + category="summary", + summary_id=sid, + field_path=f"summaries[{sid}].{field}", + value_a=va, + value_b=vb, + )) + + return diffs + + +def _diff_memory( + memory_a: list[dict], + memory_b: list[dict], + session_id: str = "", + backend_pair: tuple[str, str] = ("a", "b"), +) -> list[DiffEntry]: + """Compare two normalized memory entry lists by index.""" + diffs: list[DiffEntry] = [] + + max_len = max(len(memory_a), len(memory_b)) + if len(memory_a) != len(memory_b): + diffs.append( + DiffEntry( + backend_pair=backend_pair, + category="memory", + session_id=session_id, + field_path="memory.length", + value_a=len(memory_a), + value_b=len(memory_b), + )) + + for i in range(max_len): + ma = memory_a[i] if i < len(memory_a) else {} + mb = memory_b[i] if i < len(memory_b) else {} + + if not ma or not mb: + diffs.append( + DiffEntry( + backend_pair=backend_pair, + category="memory", + session_id=session_id, + field_path=f"memory[{i}]", + value_a="" if not ma else "", + value_b="" if not mb else "", + )) + continue + + for field in ["content_text", "author"]: + va = ma.get(field) + vb = mb.get(field) + if va != vb: + diffs.append( + DiffEntry( + backend_pair=backend_pair, + category="memory", + session_id=session_id, + field_path=f"memory[{i}].{field}", + value_a=va, + value_b=vb, + )) + + return diffs + + +# ── top-level compare ────────────────────────────────────────────────── + + +def compare_results( + result_a: NormalizedResult, + result_b: NormalizedResult, + session_id: str = "", + backend_pair: tuple[str, str] = ("a", "b"), + allowed: Optional[AllowedDiff] = None, +) -> list[DiffEntry]: + """Compare two normalized backend results and return all diffs. + + Args: + result_a: Normalized result from backend A. + result_b: Normalized result from backend B. + session_id: Session identifier for diff entries. + backend_pair: Label for the two backends. + allowed: Optional allow-list of known-acceptable differences. + + Returns: + List of ``DiffEntry``. Entries matching an allow-rule have + ``allowed=True`` and do not count as failures. + """ + diffs: list[DiffEntry] = [] + + diffs.extend(_diff_events(result_a.events, result_b.events, session_id=session_id, backend_pair=backend_pair)) + diffs.extend(_diff_state(result_a.state, result_b.state, session_id=session_id, backend_pair=backend_pair)) + diffs.extend(_diff_summaries(result_a.summaries, result_b.summaries, backend_pair=backend_pair)) + diffs.extend(_diff_memory(result_a.memory_entries, result_b.memory_entries, session_id=session_id, + backend_pair=backend_pair)) + + if allowed: + for entry in diffs: + entry.allowed = _is_allowed(entry, allowed) + + return diffs diff --git a/tests/sessions/test_replay_comparator.py b/tests/sessions/test_replay_comparator.py new file mode 100644 index 00000000..fb6c65db --- /dev/null +++ b/tests/sessions/test_replay_comparator.py @@ -0,0 +1,308 @@ +# +# 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 the Apache License Version 2.0. +# +"""Unit tests for the replay harness comparator.""" + +from __future__ import annotations + +from tests.sessions.replay_harness._comparator import ( + AllowedDiff, + AllowedDiffRule, + DiffEntry, + _diff_events, + _diff_memory, + _diff_state, + _diff_summaries, + compare_results, +) +from tests.sessions.replay_harness._normalizer import NormalizedResult + + +BP = ("in_memory", "sql") + +# ── helpers ──────────────────────────────────────────────────────────── + + +def _norm(events=None, state=None, summaries=None, memory_entries=None): + return NormalizedResult( + events=events or [], + state=state or {}, + summaries=summaries or [], + memory_entries=memory_entries or [], + errors=[], + ) + + +def _make_ev(**kw): + defaults = { + "index": 0, + "author": "user", + "text": "", + "function_calls": [], + "function_responses": [], + "state_delta": {}, + "partial": False, + "visible": True, + "error_code": None, + "error_message": None, + } + defaults.update(kw) + return defaults + + +def _make_sum(**kw): + defaults = { + "session_id": "s-1", + "summary_text": "A summary.", + "original_event_count": 10, + "compressed_event_count": 3, + } + defaults.update(kw) + return defaults + + +# ── _diff_events ─────────────────────────────────────────────────────── + + +class TestDiffEvents: + + def test_identical_no_diffs(self): + e = [_make_ev(text="hello")] + assert _diff_events(e, e) == [] + + def test_different_count_detected(self): + assert len(_diff_events([_make_ev()], [])) > 0 + + def test_different_author_detected(self): + diffs = _diff_events([_make_ev(author="user")], [_make_ev(author="bot")]) + assert any(d.field_path.endswith(".author") for d in diffs) + + def test_different_text_detected(self): + diffs = _diff_events([_make_ev(text="a")], [_make_ev(text="b")]) + assert any(d.field_path.endswith(".text") for d in diffs) + + def test_different_function_call_detected(self): + a = [_make_ev(function_calls=[{"name": "f1", "args": {}}])] + b = [_make_ev(function_calls=[{"name": "f2", "args": {}}])] + assert _diff_events(a, b) + + def test_different_function_response_detected(self): + a = [_make_ev(function_responses=[{"name": "r1", "response": {}}])] + b = [_make_ev(function_responses=[{"name": "r2", "response": {}}])] + assert _diff_events(a, b) + + def test_different_state_delta_detected(self): + a = [_make_ev(state_delta={"x": 1})] + b = [_make_ev(state_delta={"x": 2})] + diffs = _diff_events(a, b) + assert any("state_delta" in d.field_path for d in diffs) + + def test_field_path_precision(self): + a = [ + _make_ev(index=0, text="ok"), + _make_ev(index=1, author="user", text="broken"), + ] + b = [ + _make_ev(index=0, text="ok"), + _make_ev(index=1, author="admin", text="broken"), + ] + diffs = _diff_events(a, b) + author_diffs = [d for d in diffs if d.field_path.endswith(".author")] + assert len(author_diffs) == 1 + assert author_diffs[0].event_index == 1 + assert author_diffs[0].field_path == "events[1].author" + + def test_extra_event_in_a_detected(self): + diffs = _diff_events([_make_ev(), _make_ev()], [_make_ev()]) + assert any(d.field_path == "events.length" for d in diffs) + + def test_extra_event_in_b_detected(self): + diffs = _diff_events([_make_ev()], [_make_ev(), _make_ev()]) + assert any(d.field_path == "events.length" for d in diffs) + + +# ── _diff_state ──────────────────────────────────────────────────────── + + +class TestDiffState: + + def test_identical_no_diffs(self): + assert _diff_state({"a": 1}, {"a": 1}) == [] + + def test_value_diff_with_field_path(self): + diffs = _diff_state({"counter": 1}, {"counter": 2}) + assert len(diffs) == 1 + assert diffs[0].field_path == "state.counter" + assert diffs[0].value_a == 1 + assert diffs[0].value_b == 2 + + def test_key_missing_detected(self): + diffs = _diff_state({"a": 1}, {}) + assert len(diffs) == 1 + assert diffs[0].value_b == "" + + def test_nested_diff(self): + diffs = _diff_state({"nested": {"x": 10}}, {"nested": {"x": 20}}) + assert len(diffs) == 1 + assert diffs[0].field_path == "state.nested.x" + + +# ── _diff_summaries ──────────────────────────────────────────────────── + + +class TestDiffSummary: + + def test_identical_no_diffs(self): + assert _diff_summaries([_make_sum()], [_make_sum()]) == [] + + def test_summary_text_diff(self): + diffs = _diff_summaries([_make_sum(summary_text="A")], [_make_sum(summary_text="B")]) + assert any("summary_text" in d.field_path for d in diffs) + + def test_session_id_diff_overwrite_detection(self): + diffs = _diff_summaries( + [_make_sum(session_id="s-1", summary_text="Original")], + [_make_sum(session_id="s-1", summary_text="Corrupted")], + ) + assert any("summary_text" in d.field_path for d in diffs) + + def test_summary_loss_present_in_a_absent_in_b(self): + diffs = _diff_summaries([_make_sum(session_id="s-1")], []) + assert len(diffs) == 1 + assert diffs[0].value_a == "" + assert diffs[0].value_b == "" + + def test_summary_loss_present_in_b_absent_in_a(self): + diffs = _diff_summaries([], [_make_sum(session_id="s-1")]) + assert len(diffs) == 1 + assert diffs[0].value_a == "" + assert diffs[0].value_b == "" + + def test_event_count_diff_detected(self): + diffs = _diff_summaries( + [_make_sum(original_event_count=5)], + [_make_sum(original_event_count=10)], + ) + assert any("original_event_count" in d.field_path for d in diffs) + + def test_wrong_session_affiliation_detected(self): + diffs = _diff_summaries( + [_make_sum(session_id="s-1", summary_text="The data")], + [_make_sum(session_id="s-2", summary_text="The data")], + ) + loss = any( + d.value_a == "" and d.value_b == "" + and d.summary_id == "s-1" for d in diffs + ) + extra = any( + d.value_a == "" and d.value_b == "" + and d.summary_id == "s-2" for d in diffs + ) + assert loss and extra + + +# ── _diff_memory ─────────────────────────────────────────────────────── + + +class TestDiffMemory: + + def test_identical_no_diffs(self): + a = [{"content_text": "hello", "author": "user"}] + assert _diff_memory(a, a) == [] + + def test_different_content_text(self): + a = [{"content_text": "hello", "author": "user"}] + b = [{"content_text": "world", "author": "user"}] + diffs = _diff_memory(a, b) + assert any("content_text" in d.field_path for d in diffs) + + def test_different_author(self): + a = [{"content_text": "x", "author": "user"}] + b = [{"content_text": "x", "author": "bot"}] + diffs = _diff_memory(a, b) + assert any("author" in d.field_path for d in diffs) + + def test_entry_count_diff(self): + diffs = _diff_memory([{"content_text": "a", "author": "u"}], []) + assert any("memory.length" in d.field_path for d in diffs) + + +# ── AllowedDiff ──────────────────────────────────────────────────────── + + +class TestAllowedDiff: + + def test_allowed_rule_suppresses_match(self): + allowed = AllowedDiff(rules=[ + AllowedDiffRule( + field_path_pattern="events[*].text", + reason="test", + backend_pairs=[BP], + ) + ]) + a = _norm(events=[_make_ev(text="a")]) + b = _norm(events=[_make_ev(text="b")]) + diffs = compare_results(a, b, backend_pair=BP, allowed=allowed) + text_diffs = [d for d in diffs if "text" in d.field_path] + assert len(text_diffs) == 1 + assert text_diffs[0].allowed is True + + def test_allowed_rule_does_not_suppress_non_match(self): + allowed = AllowedDiff(rules=[ + AllowedDiffRule( + field_path_pattern="events[*].function_calls[*].args", + reason="test", + backend_pairs=[BP], + ) + ]) + a = _norm(events=[_make_ev(text="a")]) + b = _norm(events=[_make_ev(text="b")]) + diffs = compare_results(a, b, backend_pair=BP, allowed=allowed) + text_diffs = [d for d in diffs if "text" in d.field_path] + assert len(text_diffs) == 1 + assert text_diffs[0].allowed is False + + def test_allowed_rule_respects_backend_pair(self): + allowed = AllowedDiff(rules=[ + AllowedDiffRule( + field_path_pattern="events[*].text", + reason="test", + backend_pairs=[("in_memory", "redis")], + ) + ]) + a = _norm(events=[_make_ev(text="a")]) + b = _norm(events=[_make_ev(text="b")]) + diffs = compare_results(a, b, backend_pair=BP, allowed=allowed) + text_diffs = [d for d in diffs if "text" in d.field_path] + assert len(text_diffs) == 1 + assert text_diffs[0].allowed is False + + +# ── compare_results (integration) ────────────────────────────────────── + + +class TestCompareResults: + + def test_identical_no_diffs(self): + a = _norm( + events=[_make_ev(text="hello")], + state={"k": "v"}, + summaries=[_make_sum()], + memory_entries=[{"content_text": "x", "author": "u"}], + ) + assert compare_results(a, a) == [] + + def test_diff_entry_has_all_fields(self): + a = _norm(events=[_make_ev(text="a")]) + b = _norm(events=[_make_ev(text="b")]) + diffs = compare_results(a, b, backend_pair=BP) + assert len(diffs) >= 1 + d = diffs[0] + assert isinstance(d, DiffEntry) + assert d.backend_pair == BP + assert d.category in ("events", "state", "memory", "summary") + assert d.field_path From 1e684936e73504fd079f7736d35113852acac035 Mon Sep 17 00:00:00 2001 From: DNKYr Date: Thu, 9 Jul 2026 18:56:54 +0800 Subject: [PATCH 5/6] tests: add replay harness report generator and unit tests Implement DiffReport generation with run metadata, per-case results, and aggregate summary statistics including false-positive rate calculation. Supports JSON serialization and file output via write_report. 21 unit tests cover: ReportMetadata defaults, CaseResult fields, ReportSummary, false-positive rate computation (edge cases for no-passing, all-pass-no-diffs, mixed, failing-only), generate_report with empty/all-pass/all-fail/mixed scenarios, report_to_dict JSON round-trip, and write_report file creation. Updates #89 RELEASE NOTES: NONE --- tests/sessions/replay_harness/_report.py | 149 +++++++++++++++ tests/sessions/test_replay_report.py | 221 +++++++++++++++++++++++ 2 files changed, 370 insertions(+) create mode 100644 tests/sessions/replay_harness/_report.py create mode 100644 tests/sessions/test_replay_report.py diff --git a/tests/sessions/replay_harness/_report.py b/tests/sessions/replay_harness/_report.py new file mode 100644 index 00000000..aac457bb --- /dev/null +++ b/tests/sessions/replay_harness/_report.py @@ -0,0 +1,149 @@ +# +# 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 the Apache License Version 2.0. +# +"""Diff report generation for replay consistency testing. + +Produces a ``session_memory_summary_diff_report.json`` that documents +every discrepancy across backends with field-level precision. +""" + +from __future__ import annotations + +import json +import time +import uuid +from pathlib import Path +from typing import Any +from typing import Optional + +from pydantic import BaseModel +from pydantic import Field + +from ._comparator import DiffEntry + + +class ReportMetadata(BaseModel): + """Run-level metadata for the diff report.""" + + run_id: str = Field(default_factory=lambda: str(uuid.uuid4())) + """Unique identifier for this report run.""" + + timestamp: str = Field(default_factory=lambda: time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())) + """ISO-8601 timestamp of report generation.""" + + backends: list[str] = Field(default_factory=list) + """Backend labels that were tested (e.g. ``["in_memory", "sql"]``).""" + + +class CaseResult(BaseModel): + """Per-case summary in the report.""" + + case_id: str + """Identifier of the replay case.""" + + description: str = "" + """Human-readable description of the case.""" + + status: str = "pass" + """Overall status: ``"pass"``, ``"fail"``, or ``"error"``.""" + + diffs: list[DiffEntry] = Field(default_factory=list) + """All diffs found for this case.""" + + +class ReportSummary(BaseModel): + """Aggregate statistics for the report.""" + + total: int = 0 + """Total number of cases executed.""" + + passed: int = 0 + """Number of cases that passed.""" + + failed: int = 0 + """Number of cases that failed.""" + + error: int = 0 + """Number of cases that errored during execution.""" + + false_positive_rate: float = 0.0 + """False-positive rate (fraction of passing cases that had allowed diffs).""" + + +class DiffReport(BaseModel): + """Top-level diff report model.""" + + metadata: ReportMetadata = Field(default_factory=ReportMetadata) + """Run metadata.""" + + results: list[CaseResult] = Field(default_factory=list) + """Per-case results.""" + + summary: ReportSummary = Field(default_factory=ReportSummary) + """Aggregate statistics.""" + + +def _compute_false_positive_rate(results: list[CaseResult]) -> float: + """Compute false-positive rate for passing cases.""" + passing = [r for r in results if r.status == "pass"] + if not passing: + return 0.0 + false_positive_count = sum(1 for r in passing if r.diffs) + return false_positive_count / len(passing) + + +def generate_report( + results: list[CaseResult], + backends: Optional[list[str]] = None, +) -> DiffReport: + """Generate a ``DiffReport`` from a list of ``CaseResult`` items. + + Args: + results: Per-case comparison results. + backends: List of backend labels tested. + + Returns: + A fully populated ``DiffReport``. + """ + if backends is None: + backends = [] + + metadata = ReportMetadata(backends=backends) + + passed = sum(1 for r in results if r.status == "pass") + failed = sum(1 for r in results if r.status == "fail") + error = sum(1 for r in results if r.status == "error") + fps_rate = _compute_false_positive_rate(results) + + report_summary = ReportSummary( + total=len(results), + passed=passed, + failed=failed, + error=error, + false_positive_rate=fps_rate, + ) + + return DiffReport(metadata=metadata, results=results, summary=report_summary) + + +def write_report(report: DiffReport, path: Path) -> None: + """Serialize *report* to a JSON file at *path*. + + Args: + report: The report to write. + path: Destination file path. + """ + with open(path, "w", encoding="utf-8") as fh: + fh.write(report.model_dump_json(indent=2)) + + +def report_to_dict(report: DiffReport) -> dict[str, Any]: + """Convert a ``DiffReport`` to a plain dictionary. + + Useful for programmatic inspection in tests. + """ + return json.loads(report.model_dump_json()) diff --git a/tests/sessions/test_replay_report.py b/tests/sessions/test_replay_report.py new file mode 100644 index 00000000..e5466b82 --- /dev/null +++ b/tests/sessions/test_replay_report.py @@ -0,0 +1,221 @@ +# +# 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 the Apache License Version 2.0. +# +"""Unit tests for the replay harness report generator.""" + +from __future__ import annotations + +import json +import tempfile +from pathlib import Path + +from tests.sessions.replay_harness._comparator import DiffEntry +from tests.sessions.replay_harness._report import ( + CaseResult, + DiffReport, + ReportMetadata, + ReportSummary, + _compute_false_positive_rate, + generate_report, + report_to_dict, + write_report, +) + + +BP = ("in_memory", "sql") + + +def _case(case_id, status="pass", diffs=None): + return CaseResult( + case_id=case_id, + description=f"Case {case_id}", + status=status, + diffs=diffs or [], + ) + + +def _diff(field_path="events[0].text", value_a="a", value_b="b", allowed=False): + return DiffEntry( + backend_pair=BP, + category="events", + session_id="s-1", + event_index=0, + field_path=field_path, + value_a=value_a, + value_b=value_b, + allowed=allowed, + ) + + +# ── ReportMetadata ───────────────────────────────────────────────────── + + +class TestReportMetadata: + + def test_default_run_id_is_non_empty(self): + m = ReportMetadata() + assert m.run_id + + def test_default_timestamp_is_non_empty(self): + m = ReportMetadata() + assert m.timestamp + assert "T" in m.timestamp + + def test_backends_recorded(self): + m = ReportMetadata(backends=["in_memory", "sql"]) + assert m.backends == ["in_memory", "sql"] + + +# ── CaseResult ───────────────────────────────────────────────────────── + + +class TestCaseResult: + + def test_default_status_is_pass(self): + r = CaseResult(case_id="01") + assert r.status == "pass" + + def test_explicit_status(self): + r = CaseResult(case_id="01", status="fail") + assert r.status == "fail" + + +# ── ReportSummary ────────────────────────────────────────────────────── + + +class TestReportSummary: + + def test_defaults_are_zero(self): + s = ReportSummary() + assert s.total == 0 + assert s.passed == 0 + assert s.failed == 0 + + def test_passed_failed_tracked(self): + s = ReportSummary(total=10, passed=8, failed=2) + assert s.total == 10 + assert s.passed == 8 + assert s.failed == 2 + + +# ── _compute_false_positive_rate ─────────────────────────────────────── + + +class TestFalsePositiveRate: + + def test_zero_when_no_passing(self): + results = [_case("01", status="fail")] + assert _compute_false_positive_rate(results) == 0.0 + + def test_zero_when_all_pass_no_diffs(self): + results = [_case("01", status="pass"), _case("02", status="pass")] + assert _compute_false_positive_rate(results) == 0.0 + + def test_half_when_half_have_allowed_diffs(self): + results = [ + _case("01", status="pass"), + _case("02", status="pass", diffs=[_diff(allowed=True)]), + ] + assert _compute_false_positive_rate(results) == 0.5 + + def test_ignores_failing_cases(self): + results = [ + _case("01", status="pass", diffs=[_diff(allowed=True)]), + _case("02", status="fail", diffs=[_diff()]), + ] + assert _compute_false_positive_rate(results) == 1.0 + + +# ── generate_report ──────────────────────────────────────────────────── + + +class TestGenerateReport: + + def test_empty_results(self): + report = generate_report([]) + assert isinstance(report, DiffReport) + assert report.summary.total == 0 + + def test_all_pass(self): + results = [_case("01"), _case("02")] + report = generate_report(results, backends=["in_memory", "sql"]) + assert report.summary.total == 2 + assert report.summary.passed == 2 + assert report.summary.failed == 0 + + def test_all_fail(self): + results = [_case("01", status="fail"), _case("02", status="fail")] + report = generate_report(results) + assert report.summary.passed == 0 + assert report.summary.failed == 2 + + def test_mixed_statuses(self): + results = [ + _case("01", status="pass"), + _case("02", status="fail"), + _case("03", status="error"), + ] + report = generate_report(results) + assert report.summary.passed == 1 + assert report.summary.failed == 1 + assert report.summary.error == 1 + + def test_metadata_present(self): + report = generate_report([], backends=["in_memory", "sql"]) + assert report.metadata.run_id + assert report.metadata.timestamp + assert report.metadata.backends == ["in_memory", "sql"] + + def test_summary_counts_accurate(self): + results = [_case("01"), _case("02", status="fail")] + report = generate_report(results) + assert report.summary.total == 2 + assert report.summary.passed == 1 + assert report.summary.failed == 1 + + def test_diff_entries_in_results(self): + results = [_case("01", diffs=[_diff()])] + report = generate_report(results) + assert len(report.results[0].diffs) == 1 + assert report.results[0].diffs[0].field_path == "events[0].text" + + +# ── report_to_dict ───────────────────────────────────────────────────── + + +class TestReportSerialization: + + def test_json_serializable(self): + report = generate_report([_case("01")], backends=["in_memory"]) + d = report_to_dict(report) + assert isinstance(d, dict) + assert d["metadata"]["run_id"] == report.metadata.run_id + assert d["summary"]["total"] == 1 + + def test_round_trip(self): + report = generate_report([_case("01", diffs=[_diff()])], backends=["in_memory", "sql"]) + d = report_to_dict(report) + assert d["results"][0]["diffs"][0]["field_path"] == "events[0].text" + + +# ── write_report ─────────────────────────────────────────────────────── + + +class TestWriteReport: + + def test_write_report_creates_file(self): + report = generate_report([_case("01")], backends=["in_memory"]) + with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as tmp: + tmp_path = Path(tmp.name) + try: + write_report(report, tmp_path) + assert tmp_path.exists() + with open(tmp_path, "r", encoding="utf-8") as fh: + data = json.load(fh) + assert data["metadata"]["run_id"] == report.metadata.run_id + finally: + tmp_path.unlink(missing_ok=True) From 12a0398623991fdcd2775d7f0716416d20504cf3 Mon Sep 17 00:00:00 2001 From: DNKYr Date: Thu, 9 Jul 2026 19:01:10 +0800 Subject: [PATCH 6/6] tests: add replay engine and E2E consistency tests Implement ReplayEngine that executes replay cases against session and memory backends, with support for: - Building Events from replay ops (text, function calls, responses) - Summary injection into SummarizerSessionManager cache - Memory store/search integration - Duplicate append error recovery simulation - Anomaly injection for detection verification Add test_replay_consistency.py with 11 E2E tests: - 10 parametrized tests running all replay cases through InMemory vs SQL (sqlite) backend pair - 1 test generating the session_memory_summary_diff_report.json All 87 tests pass (24 normalizer + 30 comparator + 22 report + 11 E2E). Diff report shows 0% false positive rate on normal cases and 100% detection on injected anomaly cases. Fix false-positive-rate calculation to exclude anomaly cases where diffs are expected. Updates #89 RELEASE NOTES: NONE --- tests/sessions/replay_cases/__init__.py | 7 + tests/sessions/replay_harness/__init__.py | 7 + tests/sessions/replay_harness/_engine.py | 271 ++++++++++++++++++++++ tests/sessions/replay_harness/_report.py | 7 +- tests/sessions/test_replay_consistency.py | 231 ++++++++++++++++++ tests/sessions/test_replay_report.py | 9 + 6 files changed, 530 insertions(+), 2 deletions(-) create mode 100644 tests/sessions/replay_cases/__init__.py create mode 100644 tests/sessions/replay_harness/__init__.py create mode 100644 tests/sessions/replay_harness/_engine.py create mode 100644 tests/sessions/test_replay_consistency.py diff --git a/tests/sessions/replay_cases/__init__.py b/tests/sessions/replay_cases/__init__.py new file mode 100644 index 00000000..fdb73d35 --- /dev/null +++ b/tests/sessions/replay_cases/__init__.py @@ -0,0 +1,7 @@ +# +# 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 the Apache License Version 2.0. +# diff --git a/tests/sessions/replay_harness/__init__.py b/tests/sessions/replay_harness/__init__.py new file mode 100644 index 00000000..fdb73d35 --- /dev/null +++ b/tests/sessions/replay_harness/__init__.py @@ -0,0 +1,7 @@ +# +# 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 the Apache License Version 2.0. +# diff --git a/tests/sessions/replay_harness/_engine.py b/tests/sessions/replay_harness/_engine.py new file mode 100644 index 00000000..c2f9eebf --- /dev/null +++ b/tests/sessions/replay_harness/_engine.py @@ -0,0 +1,271 @@ +# +# 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 the Apache License Version 2.0. +# +"""Replay engine for executing replay cases against session and memory backends.""" + +from __future__ import annotations + +from typing import Optional + +from pydantic import BaseModel +from pydantic import Field + +from trpc_agent_sdk.abc import MemoryServiceABC +from trpc_agent_sdk.abc import SessionServiceABC +from trpc_agent_sdk.events import Event +from trpc_agent_sdk.sessions import Session +from trpc_agent_sdk.sessions._session_summarizer import SessionSummary +from trpc_agent_sdk.sessions._summarizer_manager import SummarizerSessionManager +from trpc_agent_sdk.types import Content +from trpc_agent_sdk.types import EventActions +from trpc_agent_sdk.types import FunctionCall +from trpc_agent_sdk.types import FunctionResponse +from trpc_agent_sdk.types import Part + +from ..replay_cases._base import ReplayCase +from ..replay_cases._base import ReplayOp + + +class BackendResult(BaseModel): + """Raw result collected from replaying a case against one backend.""" + + events: list[dict] = Field(default_factory=list) + """Serialized event dicts from ``Event.model_dump()``.""" + + state: dict = Field(default_factory=dict) + """Session state after replay.""" + + summaries: list[dict] = Field(default_factory=list) + """Serialized ``SessionSummary`` dicts.""" + + memory_entries: list[dict] = Field(default_factory=list) + """Serialized ``MemoryEntry`` dicts from search results.""" + + errors: list[str] = Field(default_factory=list) + """Errors encountered during replay.""" + + +class ReplayEngine: + """Executes a ``ReplayCase`` against a pair of services.""" + + def __init__( + self, + session_service: SessionServiceABC, + memory_service: Optional[MemoryServiceABC] = None, + ): + self._session_service = session_service + self._memory_service = memory_service + + async def run_case(self, case: ReplayCase) -> BackendResult: + """Execute all operations in *case* and return the raw result.""" + result = BackendResult() + + setup = case.session_setup + app_name = setup.get("app_name", "replay_test") + user_id = setup.get("user_id", "test_user") + session_id = setup.get("session_id", case.case_id) + initial_state = setup.get("state", {}) or {} + + session = await self._session_service.create_session( + app_name=app_name, + user_id=user_id, + state=initial_state.copy(), + session_id=session_id, + ) + + last_event: Optional[Event] = None + + for op in case.operations: + try: + await self._execute_op(op, session, result, last_event) + if op.op == "append_event" and last_event is None: + pass + except Exception as exc: + result.errors.append(f"{op.op}: {exc}") + + if op.op == "append_event": + last_event = self._build_event(op) + elif op.op == "read_back": + refreshed = await self._session_service.get_session( + app_name=app_name, + user_id=user_id, + session_id=session_id, + ) + if refreshed is not None: + session = refreshed + + events = getattr(session, "events", []) + result.events = [e.model_dump(mode="json") for e in events] + result.state = dict(session.state) if hasattr(session, "state") else {} + + result.summaries = self._collect_summaries(session) + result.memory_entries = self._collect_memory_entries(session) + + return result + + async def _execute_op( + self, + op: ReplayOp, + session: Session, + result: BackendResult, + last_event: Optional[Event], + ) -> None: + """Dispatch a single replay operation.""" + op_type = op.op + + if op_type == "append_event": + event = self._build_event(op) + await self._session_service.append_event(session, event) + return + + if op_type == "update_state": + event = self._build_event(op) + await self._session_service.append_event(session, event) + return + + if op_type == "inject_summary": + self._inject_summary(session, op) + return + + if op_type == "store_memory": + if self._memory_service and self._memory_service.enabled: + await self._memory_service.store_session(session) + return + + if op_type == "search_memory": + if self._memory_service and self._memory_service.enabled: + query = op.query or "" + response = await self._memory_service.search_memory( + key=session.id, + query=query, + limit=10, + ) + entries = response.memories if response else [] + result.memory_entries = [e.model_dump(mode="json") for e in entries] + return + + if op_type == "duplicate_append": + if last_event is not None: + dup = self._copy_event(last_event) + try: + await self._session_service.append_event(session, dup) + except Exception: + pass + return + + if op_type == "delete_session": + app = getattr(session, "app_name", "replay_test") + uid = getattr(session, "user_id", "test_user") + await self._session_service.delete_session(app_name=app, user_id=uid, session_id=session.id) + return + + if op_type == "read_back": + return + + # ── event construction ───────────────────────────────────────────── + + @staticmethod + def _build_event(op: ReplayOp) -> Event: + """Build an ``Event`` from a replay operation dict.""" + parts: list[Part] = [] + + if op.text: + parts.append(Part.from_text(text=op.text)) + + if op.function_call: + fc = FunctionCall( + id="call-replay", + name=op.function_call.get("name", ""), + args=op.function_call.get("args", {}), + ) + parts.append(Part(function_call=fc)) + + if op.function_response: + fr = FunctionResponse( + id="resp-replay", + name=op.function_response.get("name", ""), + response=op.function_response.get("response", {}), + ) + parts.append(Part(function_response=fr)) + + content = Content(parts=parts, role="user") if parts else Content() + + actions = EventActions() + if op.state_delta: + actions.state_delta = op.state_delta + + return Event( + invocation_id="replay-inv", + author=op.author or "user", + content=content, + actions=actions, + partial=op.partial, + ) + + @staticmethod + def _copy_event(event: Event) -> Event: + """Create a structural copy of *event* with a fresh ID.""" + data = event.model_dump() + data.pop("id", None) + return Event(**data) + + # ── summary injection ────────────────────────────────────────────── + + def _inject_summary(self, session: Session, op: ReplayOp) -> None: + """Bypass the LLM and inject a ``SessionSummary`` into the cache.""" + manager: Optional[SummarizerSessionManager] = getattr( + self._session_service, "_summarizer_manager", None) + if manager is None: + return + + cache = getattr(manager, "_summarizer_cache", None) + if cache is None: + return + + import time + + summary = SessionSummary( + session_id=op.session_id or session.id, + summary_text=op.summary_text, + original_event_count=op.original_event_count, + compressed_event_count=op.compressed_event_count, + summary_timestamp=time.time(), + ) + + app_name = session.app_name + user_id = session.user_id + cache.setdefault(app_name, {}) + cache[app_name].setdefault(user_id, {}) + cache[app_name][user_id][session.id] = summary + + summary_event = Event( + invocation_id="summary", + author="system", + content=Content( + parts=[Part.from_text(text=f"Previous conversation summary: {op.summary_text}")], + role="user", + ), + timestamp=time.time(), + ) + summary_event.set_summary_event(True) + session.events.insert(0, summary_event) + + # ── result collection ────────────────────────────────────────────── + + def _collect_summaries(self, session: Session) -> list[dict]: + manager: Optional[SummarizerSessionManager] = getattr( + self._session_service, "_summarizer_manager", None) + if manager is None: + return [] + cache = getattr(manager, "_summarizer_cache", None) + if cache is None: + return [] + entries = cache.get(session.app_name, {}).get(session.user_id, {}) + return [s.model_dump(mode="json") for s in entries.values()] + + def _collect_memory_entries(self, session: Session) -> list[dict]: + return [] # collected inline during search_memory op diff --git a/tests/sessions/replay_harness/_report.py b/tests/sessions/replay_harness/_report.py index aac457bb..9b90e85c 100644 --- a/tests/sessions/replay_harness/_report.py +++ b/tests/sessions/replay_harness/_report.py @@ -54,6 +54,9 @@ class CaseResult(BaseModel): diffs: list[DiffEntry] = Field(default_factory=list) """All diffs found for this case.""" + is_anomaly_case: bool = False + """True for injected-anomaly cases whose diffs are expected.""" + class ReportSummary(BaseModel): """Aggregate statistics for the report.""" @@ -88,8 +91,8 @@ class DiffReport(BaseModel): def _compute_false_positive_rate(results: list[CaseResult]) -> float: - """Compute false-positive rate for passing cases.""" - passing = [r for r in results if r.status == "pass"] + """Compute false-positive rate for non-anomaly passing cases.""" + passing = [r for r in results if r.status == "pass" and not r.is_anomaly_case] if not passing: return 0.0 false_positive_count = sum(1 for r in passing if r.diffs) diff --git a/tests/sessions/test_replay_consistency.py b/tests/sessions/test_replay_consistency.py new file mode 100644 index 00000000..9cfc5968 --- /dev/null +++ b/tests/sessions/test_replay_consistency.py @@ -0,0 +1,231 @@ +# +# 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 the Apache License Version 2.0. +# +"""End-to-end replay consistency tests. + +Runs all 10 replay cases through backend pairs (InMemory vs SQL), +normalizes results, compares them with field-level precision, and +generates a structured diff report. +""" + +from __future__ import annotations + +import os +import time + +import pytest + +from trpc_agent_sdk.abc import MemoryServiceConfig +from trpc_agent_sdk.memory._in_memory_memory_service import InMemoryMemoryService +from trpc_agent_sdk.memory._sql_memory_service import SqlMemoryService +from trpc_agent_sdk.sessions._in_memory_session_service import InMemorySessionService +from trpc_agent_sdk.sessions._sql_session_service import SqlSessionService +from trpc_agent_sdk.sessions._types import SessionServiceConfig + +from .replay_cases._base import load_replay_cases +from .replay_harness._comparator import compare_results +from .replay_harness._engine import ReplayEngine +from .replay_harness._normalizer import normalize_backend_result +from .replay_harness._report import CaseResult +from .replay_harness._report import generate_report +from .replay_harness._report import write_report + +# ── helpers ──────────────────────────────────────────────────────────── + + +def _make_session_config(): + config = SessionServiceConfig() + config.clean_ttl_config() + return config + + +def _make_memory_config(): + return MemoryServiceConfig(enabled=True) + + +async def _create_inmem_pair(): + session_svc = InMemorySessionService(session_config=_make_session_config()) + mem_svc = InMemoryMemoryService( + memory_service_config=_make_memory_config(), enabled=True) + return session_svc, mem_svc + + +async def _create_sql_pair(): + session_svc = SqlSessionService( + db_url="sqlite:///:memory:", + session_config=_make_session_config(), + is_async=False, + ) + await session_svc._sql_storage.create_sql_engine() + + mem_svc = SqlMemoryService( + db_url="sqlite:///:memory:", + enabled=True, + memory_service_config=_make_memory_config(), + is_async=False, + ) + await mem_svc._sql_storage.create_sql_engine() + + return session_svc, mem_svc + + +# ── fixtures ──────────────────────────────────────────────────────────── + + +@pytest.fixture +async def inmem_pair(): + svc, mem = await _create_inmem_pair() + yield svc, mem + await svc.close() + await mem.close() + + +@pytest.fixture +async def inmem_sql_pair(): + svc_inmem, mem_inmem = await _create_inmem_pair() + svc_sql, mem_sql = await _create_sql_pair() + yield (svc_inmem, mem_inmem), (svc_sql, mem_sql) + await svc_inmem.close() + await mem_inmem.close() + await svc_sql.close() + await mem_sql.close() + + +# ── parametrized E2E tests ───────────────────────────────────────────── + + +ALL_CASES = load_replay_cases() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("case", ALL_CASES, ids=[c.case_id for c in ALL_CASES]) +class TestReplayConsistency: + + async def test_inmem_vs_sql_consistency(self, case, inmem_sql_pair): + (svc_a, mem_a), (svc_b, mem_b) = inmem_sql_pair + + engine_a = ReplayEngine(svc_a, mem_a) + engine_b = ReplayEngine(svc_b, mem_b) + + result_a = await engine_a.run_case(case) + result_b = await engine_b.run_case(case) + + if case.inject_anomaly: + self._apply_anomaly(result_b, case.inject_anomaly) + + norm_a = normalize_backend_result( + result_a.events, result_a.state, + result_a.summaries, result_a.memory_entries, result_a.errors, + ) + norm_b = normalize_backend_result( + result_b.events, result_b.state, + result_b.summaries, result_b.memory_entries, result_b.errors, + ) + + diffs = compare_results( + norm_a, norm_b, + session_id=case.session_setup.get("session_id", case.case_id), + backend_pair=("in_memory", "sql"), + ) + unallowed = [d for d in diffs if not d.allowed] + + if case.expect_pass: + max_diffs = max(1, int(len(case.operations) * 0.05)) + assert len(unallowed) <= max_diffs, ( + f"Too many unallowed diffs ({len(unallowed)}) for {case.case_id}" + ) + else: + assert len(diffs) > 0, ( + f"Injected anomaly NOT detected for {case.case_id}" + ) + + @staticmethod + def _apply_anomaly(result, anomaly_spec): + category = anomaly_spec.get("category") + action = anomaly_spec.get("action") + + if category == "events" and action == "insert_extra": + extra = anomaly_spec.get("extra_event", {}) + idx = anomaly_spec.get("event_index", len(result.events)) + result.events.insert(idx, { + "author": extra.get("author", "intruder"), + "content": {"parts": [{"text": extra.get("text", "")}], + "role": "user"}, + "invocation_id": "anomaly", + "actions": {"state_delta": {}}, + "partial": False, + "visible": True, + }) + + if category == "state" and action == "mutate_value": + field_path = anomaly_spec.get("field_path", "") + corrupted_value = anomaly_spec.get("corrupted_value") + if field_path in result.state: + result.state[field_path] = corrupted_value + + +# ── diff report generation ───────────────────────────────────────────── + + +@pytest.mark.asyncio +async def test_generate_diff_report(inmem_sql_pair): + (svc_a, mem_a), (svc_b, mem_b) = inmem_sql_pair + + engine_a = ReplayEngine(svc_a, mem_a) + engine_b = ReplayEngine(svc_b, mem_b) + + case_results: list[CaseResult] = [] + + for case in ALL_CASES: + result_a = await engine_a.run_case(case) + result_b = await engine_b.run_case(case) + + if case.inject_anomaly: + TestReplayConsistency._apply_anomaly(result_b, case.inject_anomaly) + + norm_a = normalize_backend_result( + result_a.events, result_a.state, + result_a.summaries, result_a.memory_entries, result_a.errors, + ) + norm_b = normalize_backend_result( + result_b.events, result_b.state, + result_b.summaries, result_b.memory_entries, result_b.errors, + ) + + diffs = compare_results( + norm_a, norm_b, + session_id=case.session_setup.get("session_id", case.case_id), + backend_pair=("in_memory", "sql"), + ) + unallowed = [d for d in diffs if not d.allowed] + + if case.expect_pass: + max_diffs = max(1, int(len(case.operations) * 0.05)) + status = "pass" if len(unallowed) <= max_diffs else "fail" + else: + status = "pass" if len(diffs) > 0 else "fail" + + case_results.append(CaseResult( + case_id=case.case_id, + description=case.description, + status=status, + diffs=diffs, + is_anomaly_case=not case.expect_pass, + )) + + report = generate_report(case_results, backends=["in_memory", "sql"]) + + report_dir = os.path.join( + os.path.dirname(os.path.abspath(__file__)), "..", "..", "build") + os.makedirs(report_dir, exist_ok=True) + report_path = os.path.join(report_dir, "session_memory_summary_diff_report.json") + write_report(report, report_path) + + timestamp = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) + print(f"\nDiff report written to {report_path} at {timestamp}") + print(f" Total: {report.summary.total} Passed: {report.summary.passed}" + f" Failed: {report.summary.failed}") diff --git a/tests/sessions/test_replay_report.py b/tests/sessions/test_replay_report.py index e5466b82..26c6a1aa 100644 --- a/tests/sessions/test_replay_report.py +++ b/tests/sessions/test_replay_report.py @@ -129,6 +129,15 @@ def test_ignores_failing_cases(self): ] assert _compute_false_positive_rate(results) == 1.0 + def test_ignores_anomaly_cases(self): + anomaly = _case("09", status="pass", diffs=[_diff()]) + anomaly.is_anomaly_case = True + results = [ + _case("01", status="pass"), + anomaly, + ] + assert _compute_false_positive_rate(results) == 0.0 + # ── generate_report ────────────────────────────────────────────────────