diff --git a/.gitignore b/.gitignore index 233248dd..58eb6b48 100644 --- a/.gitignore +++ b/.gitignore @@ -24,3 +24,6 @@ test-ngtest-ut-trpc-agent-py.xml node_modules package-lock.json pyrightconfig.json + +# spec-workflow tool artifacts +.spec-workflow diff --git a/docs/superpowers/plans/2026-07-13-session-memory-replay-consistency.md b/docs/superpowers/plans/2026-07-13-session-memory-replay-consistency.md new file mode 100644 index 00000000..25c9a1d8 --- /dev/null +++ b/docs/superpowers/plans/2026-07-13-session-memory-replay-consistency.md @@ -0,0 +1,258 @@ +# Session/Memory/Summary 回放一致性框架 Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: 用 superpowers:executing-plans 逐 task 执行。步骤用 `- [ ]` 跟踪。 + +**Goal:** 实现一个后端中立的 replay harness,用 10 条标准化轨迹驱动 InMemory/SQLite/Redis,产出可定位的差异报告,并通过快照层 + 端到端后端注入验证检出率/误报率。 + +**Architecture:** 四段管线 `load JSONL → replay_case → 后端中立快照 → compare → report`。详见 [设计文档](../specs/2026-07-13-session-memory-replay-consistency-design.md)。 + +**Tech Stack:** Python 3.10+、pytest、pydantic v2、SQLAlchemy(SQLite)、redis-py(mock)。 + +## Global Constraints + +- **PR 干净**:只碰 `tests/` + 本计划/设计文档 + 仓库根 `session_memory_summary_diff_report.json`(运行产物)。**不改 `trpc_agent_sdk/` 生产代码**;发现的 SDK bug 只在报告/文档记录。 +- **CI lint**:提交前本地 `PYTHONUTF8=1` 跑 `yapf -ri` + `flake8`([[ci-lint-yapf-flake8]])。 +- **Windows**:`python-magic` 用 `python-magic-bin`([[python-magic-windows-cygwin-crash]])。 +- **提交纪律**:`git add` 只加本计划列出的确切路径,禁用 `-A`/`.`([[subagent-git-add-scope]])。用户未要求不主动 commit/push。 +- **轻量模式 ≤30s**:默认 InMemory vs SQLite(`:memory:`);Redis/MySQL 经 env 启用,不可用 `pytest.skip`。 +- **确定性**:无 LLM、无真实网络;summary 用 `_DeterministicSummarizer`;时间用占位符归一化。 + +## File Structure + +(完整职责见设计文档 §3.2) + +| 文件 | 职责 | +|---|---| +| `tests/sessions/replay/__init__.py` | 包入口 + 150–300 字设计说明 | +| `tests/sessions/replay/harness.py` | 数据模型(ReplayOp/ReplayCase/ReplayBackend/ReplaySnapshot)+ `replay_case()` | +| `tests/sessions/replay/normalizer.py` | 占位符归一化 | +| `tests/sessions/replay/comparator.py` | 递归 `visit()` + DiffEntry | +| `tests/sessions/replay/allowed_diff.py` | JSONPath 匹配 + 覆盖率治理 | +| `tests/sessions/replay/summary_checks.py` | 三类专项 + 分词 Jaccard 语义比较 | +| `tests/sessions/replay/injectors.py` | 快照层 + 端到端后端注入 | +| `tests/sessions/replay/report.py` | schema_version=3 报告 | +| `tests/sessions/replay/backends.py` | 三后端实例化 + env 门控 | +| `tests/sessions/replay/replay_cases/*.jsonl` | 10 条 case | +| `tests/sessions/test_replay_consistency.py` | 主 E2E | +| `tests/sessions/test_replay_injections.py` | 两层注入检出 | +| `tests/sessions/test_replay_unit.py` | normalizer/comparator/allowed_diff/summary_checks 单测 | + +--- + +## Task 1: 包骨架 + 设计说明 + +**Files:** +- Create: `tests/sessions/replay/__init__.py` + +**Interfaces:** Produces `tests.sessions.replay` 包(后续模块的根)。 + +- [ ] **Step 1:** 创建 `tests/sessions/replay/__init__.py`,内含设计文档 §10 的 150–300 字设计说明作为模块 docstring,加 `__all__ = []`。 +- [ ] **Step 2:** `PYTHONUTF8=1 python -c "import tests.sessions.replay"` 验证可导入。 + +--- + +## Task 2: harness 数据模型(纯 Pydantic,无逻辑) + +**Files:** +- Create: `tests/sessions/replay/harness.py` + +**Interfaces:** +- Consumes: `trpc_agent_sdk.sessions.SessionServiceABC`、`trpc_agent_sdk.memory.MemoryServiceABC` +- Produces: `ReplayOp`、`AllowedDiffRule`、`ReplayCase`、`ReplayBackend`、`ReplaySnapshot`(签名见设计文档 §4.1) +- **关键决策**:`ReplayOp` 用 `type: Literal[...]` + `model_dump()` 兼容异构 payload(不用 discriminated union 的复杂标签,保持 jsonl 可读)。 + +- [ ] **Step 1:** 写 `test_replay_unit.py::test_replay_case_roundtrip`:构造 `ReplayCase(case_id="x", operations=[ReplayOp(op="create_session", app_name="a", user_id="u", session_id="s")])`,`model_dump_json()` 后 `model_validate_json()` 应相等。 +- [ ] **Step 2:** 跑测试,确认 FAIL(模块未建)。 +- [ ] **Step 3:** 在 `harness.py` 实现五个 Pydantic 模型(签名照设计文档 §4.1,`ReplayBackend` 用 `BaseModel` + `model_config = ConfigDict(arbitrary_types_allowed=True)` 以容纳 service 实例)。 +- [ ] **Step 4:** 跑测试,确认 PASS。 + +--- + +## Task 3: normalizer(占位符归一化) + +**Files:** +- Create: `tests/sessions/replay/normalizer.py` +- Test: `tests/sessions/test_replay_unit.py` + +**Interfaces:** +- Produces: `normalize_event(e: dict) -> dict`、`normalize_snapshot(s: ReplaySnapshot) -> ReplaySnapshot`、常量 `NORMALIZED = ""` + +- [ ] **Step 1:** 写测试: + - `test_normalize_event_replaces_volatile_fields`:event 含 `id/timestamp/invocation_id` → 归一化后这三键值为 `NORMALIZED`,**键仍存在**。 + - `test_normalize_strips_temp_state`:state 含 `temp:k` → 归一化后删除该键,保留 `app:`/`user:`/普通键。 + - `test_normalize_sorts_memory`:memory 两个查询结果顺序乱 → 归一化后按 `json.dumps(sort_keys=True)` 确定性排序。 +- [ ] **Step 2:** 跑测试,确认 FAIL。 +- [ ] **Step 3:** 实现:递归把 `id`/`timestamp`/`invocation_id` 顶层键值替换为 `NORMALIZED`;剥离 `state` 中 `temp:` 前缀键;memory 各 list 按 `json.dumps(i, sort_keys=True, ensure_ascii=True)` 排序。 +- [ ] **Step 4:** 跑测试,确认 PASS。 + +--- + +## Task 4: comparator(递归比较 + 内联定位) + +**Files:** +- Create: `tests/sessions/replay/comparator.py` +- Test: `tests/sessions/test_replay_unit.py` + +**Interfaces:** +- Consumes: `ReplaySnapshot`、`list[AllowedDiffRule]`(from Task 2)、`is_allowed`(from Task 5,先用 stub) +- Produces: `DiffEntry`、`compare_snapshots(reference, candidate, *, reference_backend, candidate_backend, allowed_diff) -> list[DiffEntry]` +- **关键决策**:DiffEntry 在递归到叶子时**内联写入** `session_id`(snapshot 顶层)、`event_index`(events 列表的下标)、`summary_id`(若 path 在 summary 下)。 + +- [ ] **Step 1:** 写测试: + - `test_diff_detects_leaf_mismatch`:`events[0].author` 左 `"user"` 右 `"assistant"` → 产 1 条 DiffEntry,`field_path="events[0].author"`、`event_index=0`、`allowed=False`。 + - `test_diff_aligns_dict_sorted_keys`:左 `{"b":1,"a":2}` 右 `{"a":2,"b":1}` → 无 diff。 + - `test_diff_list_length_diff`:左 events 3 条右 2 条 → 产 1 条 `` diff,`event_index=2`。 +- [ ] **Step 2:** 跑测试,确认 FAIL。 +- [ ] **Step 3:** 实现 `visit(left, right, path)` 递归(dict sorted keys / list 下标+补 `` / 叶子 `!=` 产 DiffEntry),顶层包装填 `session_id`、按 path 前缀推断 `event_index`/`summary_id`;`allowed_diff` 参数暂传 `[]`。 +- [ ] **Step 4:** 跑测试,确认 PASS。 + +--- + +## Task 5: allowed_diff(JSONPath 精确 + 治理) + +**Files:** +- Create: `tests/sessions/replay/allowed_diff.py` +- Test: `tests/sessions/test_replay_unit.py` + `tests/sessions/test_allowed_diff_governance.py` + +**Interfaces:** +- Produces: `is_allowed(field_path, backend_pair, rules) -> tuple[bool, str|None]`、`check_governance(case, total_fields, used_allowed) -> None`、常量 `MAX_ALLOWED_PER_CASE=8`、`MAX_ALLOWED_RATIO=0.10` +- **接线**:回 `comparator.compare_snapshots` 用真实 `is_allowed` 替换 Task 4 stub。 + +- [ ] **Step 1:** 写测试: + - `test_allowed_exact_path_match`:规则 `path="events[0].timestamp"` → 字段 `"events[0].timestamp"` 匹配 allowed,`"events[0].author"` 不匹配。防 `*.id` 过宽。 + - `test_allowed_index_wildcard`:规则 `"events[*].timestamp"` → 匹配 `events[0].timestamp`/`events[5].timestamp`。 + - `test_governance_rejects_too_many`:`check_governance(case, total_fields=20, used_allowed=10)`(超 `MAX_ALLOWED_RATIO=0.10`)→ 抛 `ValueError`。 + - `test_governance_rejects_no_reason`:`AllowedDiffRule(path="x", reason="")` → 构造时或 governance 校验抛错。 +- [ ] **Step 2:** 跑测试,确认 FAIL。 +- [ ] **Step 3:** 实现 `is_allowed`:`[N]→[*]` 归一化后 `fnmatch.fnmatchcase`,reason 空即拒;`check_governance`:条数 > `MAX_ALLOWED_PER_CASE` 或 `used/total > MAX_ALLOWED_RATIO` 抛错。 +- [ ] **Step 4:** 跑测试 + 回归 Task 4 测试,确认 PASS。 + +--- + +## Task 6: summary_checks(三类专项 + 语义比较) + +**Files:** +- Create: `tests/sessions/replay/summary_checks.py` +- Test: `tests/sessions/test_replay_unit.py` + `tests/sessions/test_summary_checks.py` + +**Interfaces:** +- Produces: `SummaryIssue`、`check_summary_issues(reference_summary, candidate_summary, *, candidate_backend) -> list[SummaryIssue]`、`summary_text_similarity(a, b) -> float`、常量 `SUMMARY_SIM_THRESHOLD=0.8` + +- [ ] **Step 1:** 写测试: + - `test_detects_loss`:candidate `summary.current=None`(ref 非 None)→ 产 `SummaryIssue(type="loss")`。 + - `test_detects_overwrite`:candidate `current.version < ref.current.version`(旧覆盖新)→ 产 `type="overwrite"`。 + - `test_detects_affiliation`:candidate `current.session_id != ref` → 产 `type="affiliation"`。 + - `test_semantic_similarity`:相同词不同序 → Jaccard=1.0;完全不同 → 0.0。 +- [ ] **Step 2:** 跑测试,确认 FAIL。 +- [ ] **Step 3:** 实现:三类 if 判断 + `summary_text_similarity`(去标点、小写、`set(tokens)` Jaccard)。 +- [ ] **Step 4:** 跑测试,确认 PASS。 + +--- + +## Task 7: _DeterministicSummarizer + replay_case 驱动 + +**Files:** +- Modify: `tests/sessions/replay/harness.py`(加 `replay_case()`、`_DeterministicSummarizer`) + +**Interfaces:** +- Consumes: `SessionSummarizer`(覆写 `_compress_session_to_summary`,已确认存在于 [`_session_summarizer.py:197`](../../trpc_agent_sdk/sessions/_session_summarizer.py))、`SessionServiceABC.append_event/get_session/update_session`、`MemoryServiceABC.store_session/search_memory` +- Produces: `async replay_case(backend, case) -> ReplaySnapshot` + +- [ ] **Step 1:** 写 `test_replay_unit.py::test_replay_case_single_turn`(用 InMemory backend):case 含 create_session + 1 append_event → snapshot.events 长度 1、state 含写入值。 +- [ ] **Step 2:** 跑测试,确认 FAIL。 +- [ ] **Step 3:** 实现 `_DeterministicSummarizer._compress_session_to_summary` 返回 `f"{session_id} summary rev v{n}: {covered}"`;`replay_case` 按 `operations` 顺序调 service API,末尾 `get_session` + memory 查询组装 `ReplaySnapshot`;每 case 用独立 `app_name`。 +- [ ] **Step 4:** 跑测试,确认 PASS。 + +--- + +## Task 8: backends(三后端实例化 + env 门控) + +**Files:** +- Create: `tests/sessions/replay/backends.py` + +**Interfaces:** +- Produces: `in_memory_backend() -> ReplayBackend`、`sqlite_backend(tmp_path=None) -> ReplayBackend`、`redis_backend(url) -> ReplayBackend`、`enabled_backends(tmp_path) -> list[ReplayBackend]`(读 env) + +- [ ] **Step 1:** 写测试: + - `test_in_memory_backend_runs`:create+append+get 一轮不报错。 + - `test_sqlite_backend_persists`:用 tmp_path 文件 DB,写后重开 service 仍能读到。 + - `test_enabled_backends_env`:无 env → [in_memory, sqlite];设 `TRPC_REPLAY_REDIS_URL` 但不可达 → redis 标 skipped(不 crash)。 +- [ ] **Step 2:** 跑测试,确认 FAIL。 +- [ ] **Step 3:** 实现:`SessionServiceConfig(store_historical_events=True)` + `MemoryServiceConfig(enabled=True)`,均 `clean_ttl_config()`;SQLite 默认 `sqlite:///:memory:`,端到端用 `tmp_path` 文件;Redis 不可达 `pytest.skip`。 +- [ ] **Step 4:** 跑测试,确认 PASS。 + +--- + +## Task 9: report(schema_version=3) + +**Files:** +- Create: `tests/sessions/replay/report.py` + +**Interfaces:** +- Consumes: `list[DiffEntry]`、`list[SummaryIssue]`、per-case 状态 +- Produces: `build_diff_report(reference_backend, cases_results) -> dict`、`write_report(report, path)` + +- [ ] **Step 1:** 写测试: + - `test_report_schema`:构造 1 match + 1 mismatch + 1 skipped → report 含 `schema_version=3`、`backend_statuses`(skipped 含 reason)、`totals`、`false_positive_rate`、cases[]。 + - `test_report_locates_diff`:mismatch 的 diff 含全部 7 个定位字段。 +- [ ] **Step 2:** 跑测试,确认 FAIL。 +- [ ] **Step 3:** 实现按设计文档 §4.8 schema 组装;FPR = 正常 case mismatch 数 / 正常 case 总数。 +- [ ] **Step 4:** 跑测试,确认 PASS。 + +--- + +## Task 10: injectors(快照层 + 端到端) + +**Files:** +- Create: `tests/sessions/replay/injectors.py` +- Test: `tests/sessions/test_replay_injections.py` + +**Interfaces:** +- Produces: `inject_snapshot_diff(snapshot, kind) -> ReplaySnapshot`(deepcopy 改字段)、`inject_sql_diff(sqlalchemy_session, session_id, kind) -> None`、`inject_redis_diff(redis_client, app, user, sid, kind) -> None` +- **kind** 枚举:`event_author` / `state_value` / `summary_loss` / `summary_overwrite` / `summary_affiliation` / `memory_content` 等(覆盖 10 case 各一种) + +- [ ] **Step 1:** 写测试: + - `test_snapshot_inject_detected`:每种 kind → `compare` 产 ≥1 DiffEntry(快照层)。 + - `test_sql_inject_detected`:SQLite 文件 DB 注入 `event_author`(UPDATE events)→ 重读 → harness 检出。 + - `test_redis_inject_detected`:用真实 `redis.Redis`(有 `TRPC_REPLAY_REDIS_URL`)或 fakeredis 注入 → 检出;无 Redis 则 skip。 +- [ ] **Step 2:** 跑测试,确认 FAIL。 +- [ ] **Step 3:** 实现:快照层 `copy.deepcopy` + 改字段;SQL 用 `sqlalchemy.text("UPDATE events SET ...")`;Redis 用 `session_key`/`app_state_key` 定位 + `SET`/`HSET`。 +- [ ] **Step 4:** 跑测试(无 Redis 时 SQL+快照层必过),确认 PASS。 + +--- + +## Task 11: 10 条 replay_cases/*.jsonl + +**Files:** +- Create: `tests/sessions/replay/replay_cases/01..10.jsonl` + +- [ ] **Step 1:** 按 design §4.9 表写 10 条 case。每条用 operations 数组,显式 `event_id/invocation_id/timestamp/state_delta`;跨 session 用 `session_ref`。case 09 `summary_truncation` 必须含历史事件压缩 + 保留事件 + 新事件;case 10 `retry_recovery` 含 `fail_before_commit` + `retry_event`。 +- [ ] **Step 2:** `PYTHONUTF8=1 python -c "import json; [json.loads(l) for f in __import__('glob').glob('tests/sessions/replay/replay_cases/*.jsonl') for l in open(f, encoding='utf-8')]"` 验证全部可解析,且每条 `ReplayCase.model_validate_json` 通过。 + +--- + +## Task 12: 主 E2E + 跑全套验收 + +**Files:** +- Create: `tests/sessions/test_replay_consistency.py` + +- [ ] **Step 1:** 写 `test_replay_consistency_lightweight`:跑全部 10 case × `enabled_backends()`(轻量=in_memory+sqlite),断言正常 case 全 `match`、`false_positive_rate==0.0`,生成 `session_memory_summary_diff_report.json`。 +- [ ] **Step 2:** 写 `test_injection_detection_100pct`:`test_replay_injections.py` 里 10 case 各注入一种 → `detected == [True]*10`。 +- [ ] **Step 3:** 写 `test_summary_three_classes_100pct`:loss/overwrite/affiliation 各注入 → 全检出。 +- [ ] **Step 4:** 跑 `PYTHONUTF8=1 pytest tests/sessions/test_replay_*.py tests/sessions/test_allowed_diff_governance.py tests/sessions/test_summary_checks.py -v`,确认全绿、轻量 ≤30s、报告产物生成且可定位。 +- [ ] **Step 5:** lint:`PYTHONUTF8=1 yapf -ri tests/sessions/replay tests/sessions/test_replay_*.py tests/sessions/test_allowed_diff_governance.py tests/sessions/test_summary_checks.py && PYTHONUTF8=1 flake8 <同上文件>`。 + +--- + +## Self-Review + +- **Spec 覆盖**:设计文档 §4 各模块 → Task 2–10;§4.9 case → Task 11;6 条验收 → Task 12 + Task 5(governance)/Task 6(summary 三类)/Task 10(注入);4 交付物 → Task 1(说明)/Task 11(jsonl)/Task 9(报告)/全部(py)。✅ +- **类型一致**:`is_allowed`/`compare_snapshots`/`check_summary_issues`/`build_diff_report`/`replay_case` 在各 task 间签名一致。✅ +- **占位符**:无 TBD;实现体在 TDD 循环产生(inline 执行约定,非占位)。 +- **依赖顺序**:2→3/4/5/6(纯函数)→7(驱动,依赖模型)→8(后端)→9(报告)→10(注入,依赖 comparator+backends)→11(case,依赖模型)→12(E2E)。✅ + +--- + +## Execution + +Inline 执行(同一 session,设计文档+代码地图在 context 内,无需 subagent 零上下文重载)。逐 task TDD,每 task 跑测试 + lint。 diff --git a/docs/superpowers/specs/2026-07-13-session-memory-replay-consistency-design.md b/docs/superpowers/specs/2026-07-13-session-memory-replay-consistency-design.md new file mode 100644 index 00000000..79055a9c --- /dev/null +++ b/docs/superpowers/specs/2026-07-13-session-memory-replay-consistency-design.md @@ -0,0 +1,414 @@ +# Session / Memory / Summary 多后端回放一致性测试框架 — 设计文档 + +| 项 | 值 | +|---|---| +| Issue | [#89 构建 Session / Memory 多后端回放一致性测试框架](https://github.com/trpc-group/trpc-agent-python/issues/89) | +| 分支 | `feat/session-memory-replay-consistency-89` | +| 日期 | 2026-07-13 | +| 状态 | 设计(待 review) | +| 范围抉择 | 折中整合创新 · summary 走 SDK 确定性模型 · 快照层+端到端后端注入 · 发现的 SDK bug 只报告不改 | + +--- + +## 1. 背景与目标 + +项目支持 InMemory / SQL / Redis 三类 Session / Memory 后端,以及多轮对话、state 读写、事件追加、长期记忆、Session Summary 等能力。生产中常见"先用 InMemory 开发再切 SQL/Redis",若不同后端在同一条 Agent 轨迹下保存的事件顺序、state、memory 或 summary 不一致,会导致回放错乱、上下文丢失、长期记忆污染、摘要覆盖错误。 + +**目标**:构建一个可复用的回放一致性框架 —— 用同一组标准化轨迹驱动多个后端,经规范化比较自动产出可定位的差异报告。它既是测试工具,也是后端实现质量的基准。 + +**非目标(YAGNI)**: +- 不引入 embedding/向量依赖做 summary 语义比较(用分词集合即可,见 §5.6)。 +- 不在本 PR 修复发现的 SDK 生产代码 bug(只报告,另开 issue/PR,吸取 #117 PR 不干净的教训)。 +- 不做 Web/可视化报告界面(JSON 报告即可)。 + +--- + +## 2. 已有方案研究结论(10 个 PR) + +研究 issue #89 下全部 10 个公开 PR(#100/114/115/117/120/125/152/153/158/163),**均未 merged、均无真人 review**。 + +**共识(已是行业最终范式)**: +1. 四段管线:`load JSONL → replay_case(backend, case) → 后端中立快照 → compare → report`。 +2. 比较器:dict 按 sorted keys、list 按下标对齐、叶子严格相等。 +3. 用 fake/deterministic summary 规避 LLM 不确定性。 +4. `allowed_diff` 必须带 reason,反对无脑忽略。 +5. summary 做"内容语义 vs 存储元数据"分层比较。 +6. 报告 `session_memory_summary_diff_report.json`,每条 diff 内联定位字段。 + +**最大留白(本设计的创新切入点)**: +- **检出验证全部停留在快照层**(deepcopy 改快照),**无人做端到端后端数据注入** —— 直接违背 issue"后端实现质量基准"立意。 +- **summary 内容比较最多到 `compact+casefold`**,无真正语义比较(issue 却要"语义")。 +- **`allowed_diff` 缺治理**(规则引擎灵活但无上限,易被滥用塞入真不一致)。 +- **Redis 三后端从未真跑**(全 env opt-in、CI 跳过)。 +- **轻量模式诚实性**:#163 单后端也报 `match`(假绿灯),#153 用 `not_applicable`。 + +--- + +## 3. 整体架构 + +### 3.1 四段管线 + +``` +replay_cases/*.jsonl ──load──▶ ReplayCase + │ + ┌───────────────┼───────────────┐ + ▼ ▼ ▼ + InMemory SQLite Redis ← ReplayBackend + │ │ │ + └─────── replay_case(backend, case) ──────┐ + │ │ + 后端中立快照 ReplaySnapshot │ + │ │ + normalize(占位符) │ + │ │ + compare_snapshots(reference, candidate) │ + │ │ + ┌───────┴────────┐ │ + ▼ ▼ │ + DiffEntry[] summary_checks(三类专项) │ + │ │ + build_diff_report ◀────────────────────────────────┘ + │ + session_memory_summary_diff_report.json +``` + +### 3.2 目录结构(严守干净 — 只碰 `tests/` + 本文档 + 报告产物) + +``` +tests/sessions/replay/ +├── __init__.py # 含 150–300 字设计说明(issue 交付物) +├── harness.py # ReplayCase/ReplayBackend/ReplaySnapshot + replay_case() +├── normalizer.py # 占位符归一化(保留字段存在性) +├── comparator.py # 递归 visit() + DiffEntry(内联定位) +├── allowed_diff.py # JSONPath 精确匹配 + reason + 覆盖率上限治理 +├── summary_checks.py # session_id 匹配 + loss/overwrite/affiliation 三类专项 +├── injectors.py # 快照层注入 + 端到端后端注入(SQL 行 / Redis key) +├── report.py # 统一 schema_version=3 报告 +├── backends.py # 三后端实例化 + env 门控 +└── replay_cases/ + ├── 01_single_turn.jsonl + ├── 02_multi_turn.jsonl + ├── 03_tool_round_trip.jsonl + ├── 04_state_overwrite.jsonl + ├── 05_memory_preference.jsonl + ├── 06_memory_fact_update.jsonl + ├── 07_summary_create.jsonl + ├── 08_summary_update.jsonl + ├── 09_summary_truncation.jsonl + └── 10_retry_recovery.jsonl +tests/sessions/test_replay_consistency.py # 主 E2E +tests/sessions/test_replay_injections.py # 快照层 + 端到端注入检出 +tests/sessions/test_allowed_diff_governance.py # 精确匹配 + 覆盖率上限 +tests/sessions/test_summary_checks.py # 三类 summary 故障 +session_memory_summary_diff_report.json # 报告产物(仓库根) +``` + +--- + +## 4. 核心模块设计 + +### 4.1 harness.py — 数据模型与回放驱动 + +```python +class ReplayOp(BaseModel): + op: Literal[ + "create_session", "append_event", "function_call", "function_response", + "update_state", "memory_store", "memory_search", + "create_summary", "update_summary", + "fail_before_commit", "retry_event", + ] + # 各 op 的确定性 payload:显式 event_id / invocation_id / timestamp / + # state_delta / memory_key / memory_query / session_ref(跨 session 引用) + ... + +class AllowedDiffRule(BaseModel): + path: str # JSONPath,如 "events[0].timestamp" + reason: str # 必填,解释为何允许 + backend_pair: tuple[str, str] | None = None # 可选,限定后端对 + +class ReplayCase(BaseModel): + case_id: str + description: str + operations: list[ReplayOp] + allowed_diff: list[AllowedDiffRule] = [] +# 注:10 条 jsonl 均为**正常一致性轨迹**;人为不一致由 injectors.py 在运行时 +# 程序化派生(快照层 deepcopy 改字段 / 端到端改后端数据),不写进 case 文件。 +# 因此验收 2(注入检出)与验收 3(FPR)用的是**同一组 10 条 case**: +# - 不注入 → 应 100% match(测 FPR) +# - 注入 → 应 100% 检出(测检出率) + +class ReplayBackend: + name: str + session_service: SessionServiceABC + memory_service: MemoryServiceABC | None + +class ReplaySnapshot(BaseModel): + session_id: str + events: list[dict] # 归一化前的事件 + historical_events: list[dict] + state: dict # 已合并 app:/user:/session,已剥离 temp: + memory: dict # per query 的检索结果 + summary: dict # {current: {...} | None, history: [...]} + +async def replay_case(backend: ReplayBackend, case: ReplayCase) -> ReplaySnapshot: + """顺序执行 operations,采集后端中立快照。每 case 独立 app_name 命名空间,避免失败重跑污染。""" +``` + +### 4.2 comparator.py — 递归比较器(内联定位) + +```python +class DiffEntry(BaseModel): + session_id: str | None + event_index: int | None # event 在 events[] 中的下标 + summary_id: str | None + field_path: str # 如 "events[0].content.parts[0].text" + reference_backend: str + candidate_backend: str + reference_value: Any + candidate_value: Any + allowed: bool + reason: str | None + +def compare_snapshots(reference: ReplaySnapshot, candidate: ReplaySnapshot, + *, reference_backend: str, candidate_backend: str, + allowed_diff: list[AllowedDiffRule]) -> list[DiffEntry]: + """单一递归 visit(left, right, path): + - dict → 按 sorted(keys) 对齐 + - list → 按下标对齐,长度差补 + - 叶子 → left == right + 比较时直接内联写入 session_id/event_index/summary_id(取 #163 而非 #153 事后反查)。""" +``` + +### 4.3 normalizer.py — 占位符归一化 + +```python +NORMALIZED = "" + +def normalize_event(e: dict) -> dict: + """timestamp / id / invocation_id → NORMALIZED(保留字段存在性,优于 #100 的 pop 删除)。""" + +def normalize_snapshot(s: ReplaySnapshot) -> ReplaySnapshot: + """- 事件/记忆归一化 + - 剥离 temp: state + - memory 结果按 json.dumps(sort_keys=True) 排序 + - JSON 序列化统一 sort_keys,消除序列化字段顺序差""" +``` + +### 4.4 allowed_diff.py — JSONPath 精确 + 覆盖率治理(创新) + +```python +def is_allowed(field_path: str, backends: tuple[str, str], + rules: list[AllowedDiffRule]) -> tuple[bool, str | None]: + """精确匹配:events[0].timestamp;[N]→[*] 通配。规避 #117 的 *.id 过宽(会误放业务 id)。""" + +# —— 治理创新 —— +MAX_ALLOWED_PER_CASE = 8 # 每 case allowed 条数上限 +MAX_ALLOWED_RATIO = 0.10 # allowed 占该 case 总比较字段的比例上限 + +def check_governance(case: ReplayCase, total_fields: int, + used_allowed: int) -> None: + """超限 → fail。防'用 allowed_diff 塞进真不一致'。test_allowed_diff_governance.py 强制。""" +``` + +### 4.5 summary_checks.py — SDK 确定性模型 + 三分比较 + 三类专项 + +**确定性模型**(跑 SDK 真实压缩流程,只换 LLM;覆写点已确认存在 [`_compress_session_to_summary`](../../trpc_agent_sdk/sessions/_session_summarizer.py) L197): + +```python +class _DeterministicSummarizer(SessionSummarizer): + async def _compress_session_to_summary(self, ...) -> str: + return f"{session_id} summary rev v{n}: {covered_events}" # 确定性,无 LLM + +# SummarizerSessionManager(auto_summarize=True) 挂到 service +``` + +**三分比较**: +- `text`:分词集合语义比较(§4.6)。 +- `metadata`:`version` / `session_id` / `supersedes` 严格相等。 +- `coverage`:summary 覆盖的事件集合。 + +**`summary.version` 形式化**:SDK 无持久 version 字段(对齐 #158 诚实做法)→ 用「生成序号 + supersedes 链」表达可观测修订状态。 + +**三类专项检测**(对应验收第 4 条): +```python +class SummaryIssue(BaseModel): + type: Literal["loss", "overwrite", "affiliation"] + session_id: str + summary_id: str | None + detail: dict + +# loss: current is None +# overwrite: version 倒退(旧版覆盖新版) +# affiliation: summary.session_id 与所属 session 不符 +``` + +### 4.6 summary 内容语义比较(创新,纯 stdlib) + +```python +def summary_text_similarity(a: str, b: str) -> float: + """分词(去标点/小写)→ 集合 → Jaccard 相似度。""" + +SUMMARY_SIM_THRESHOLD = 0.8 # 之上判一致,之下落差异并附相似度分 +``` +10 个 PR 最多到 `compact+casefold`;本设计用分词集合兑现 issue"语义比较"要求,无外部依赖、确定性(embedding 引入依赖+不确定性,YAGNI 不取)。 + +### 4.7 injectors.py — 检出验证(核心创新:快照层 + 端到端) + +| 层 | 机制 | 验证目标 | +|---|---|---| +| **快照层**(对齐 10 PR) | `deepcopy` 快照改字段 → `compare` → 断言 `DiffEntry` 出现 | 比较器检出率 | +| **端到端后端注入**(留白填补) | 跑完 case 后**直接改后端数据**再重读 → 断言 harness 检出 | harness 对**真实后端漂移**的感知 | + +**端到端注入实现**(key/表/列均已确认,见 §7): +```python +# SQL:用真实 SQLAlchemy session 改行 +UPDATE events SET author=:bad WHERE session_id=:sid AND index=:i; # 改事件 +UPDATE app_states SET value=:bad WHERE ...; # 改 state +# → service.get_session() 重读 → compare → 断言检出 + +# Redis:用相同 key 构造函数定位 key,改值 +session_key(app, user, sid) → SET 改 session JSON 某 field # session_key/app_state_key/user_state_key +app_state_key(app) → HSET 改 hash 某 field +# → 重读 → compare → 断言检出 +``` +SQLite 走 `tmp_path` 文件 DB(非 `:memory:`)以支持外部改写。 + +### 4.8 report.py — 统一报告 schema(schema_version=3) + +```jsonc +{ + "schema_version": 3, + "reference_backend": "in_memory", + "compared_backends": ["sqlite", "redis"], + "backend_statuses": [ + {"name": "redis", "status": "skipped", "reason": "TRPC_REPLAY_REDIS_URL unset"} + ], + "totals": {"cases": 10, "matched": 9, "mismatched": 1, "not_applicable": 0}, + "false_positive_rate": 0.0, // 仅正常 case 计入分母 + "cases": [{ + "case_id": "summary_update", + "session_id": "replay-summary-update", + "status": "match", // match | mismatch | not_applicable | skipped + "differences": [{ + "session_id": "replay-...", + "event_index": 0, + "summary_id": null, + "field_path": "events[0].author", + "reference_backend": "in_memory", + "candidate_backend": "sqlite", + "reference_value": "user", + "candidate_value": "assistant", + "allowed": false, + "reason": null + }], + "summary_issues": [{"type": "overwrite", "session_id": "...", "summary_id": "...", "detail": {...}}] + }] +} +``` +**不嵌全量 snapshot**(吸取 #115 报告 10427 行不可审的教训)。 + +### 4.9 replay_cases — 10 条 case(覆盖 issue 全 8 类) + +operations 数组风格(取 #115 扩展性最强),显式 `event_id/invocation_id/state_delta`,跨 session 用 `session_ref`。 + +| # | case_id | 覆盖 issue case | +|---|---|---| +| 01 | single_turn | 1 单轮对话 | +| 02 | multi_turn | 2 多轮对话 | +| 03 | tool_round_trip | 3 工具调用(function_call/response) | +| 04 | state_overwrite | 4 state 多次写入覆盖 | +| 05 | memory_preference | 5 memory 写入读取 | +| 06 | memory_fact_update | 5 memory 事实更新 | +| 07 | summary_create | 6 summary 生成 | +| 08 | summary_update | 6 summary 更新(version/supersedes/updated_at) | +| 09 | summary_truncation | 7 summary 与事件截断(保留+summary+新事件还原上下文) | +| 10 | retry_recovery | 8 异常恢复(重复写入/脏状态/错误 summary) | + +--- + +## 5. 后端接入与运行模式 + +```python +# backends.py +def _in_memory_backend() -> ReplayBackend: ... +def _sqlite_backend(tmp_path) -> ReplayBackend: + # SessionServiceConfig(store_historical_events=True).clean_ttl_config() + # SQLite 默认 :memory:;端到端注入用 tmp_path 文件 DB +def _redis_backend(url: str) -> ReplayBackend: ... +``` + +| env | 作用 | 默认 | +|---|---|---| +| `TRPC_REPLAY_LIGHTWEIGHT` | =1 只跑 InMemory vs SQLite(轻量模式,≤30s) | 1 | +| `TRPC_REPLAY_REDIS_URL` | 设置则启用 Redis 集成模式 | 未设置→skip | +| `TRPC_REPLAY_SQL_URL` | 自定义 SQL 连接串 | sqlite 默认 | + +Redis/MySQL 不可用时 `pytest.skip`(满足 issue"不要求本地装真 Redis/MySQL")。 + +--- + +## 6. 验收标准映射(可检测性写硬) + +| 验收 | 落点 | 如何证明 | +|---|---|---| +| 1 InMemory + 持久化 | `_sqlite_backend` + 主 E2E | 轻量模式默认 InMemory vs SQLite | +| 2 10 case 100% 检出注入 | `test_replay_injections.py` | 10 条 case 各由 injectors 程序化注入一种不一致(快照层 + 端到端),断言全部检出(`detected == [True]*10`) | +| 3 误报率 ≤5% | `false_positive_rate` 字段 | **FPR 定义**:10 条 case 在**不注入**状态下被误判 mismatch 的比例。正常应 100% match,FPR=0。注入测试单独在 `test_replay_injections.py`,不计入分母 | +| 4 summary 三类 100% | `test_summary_checks.py` | loss/overwrite/affiliation 各注入一个,断言 `SummaryIssue` 出现 | +| 5 报告定位 | DiffEntry schema | 每条 diff 含 session_id/event_index/summary_id/field_path/双后端值 | +| 6 轻量 ≤30s + 集成 env | env 门控 + CI | 轻量默认跑;Redis/SQL env opt-in,不可用 skip | + +--- + +## 7. 可行性确认(硬点已验证) + +| 硬点 | 验证 | 结论 | +|---|---|---| +| SDK 确定性模型覆写点 | `_compress_session_to_summary` 存在于 [`_session_summarizer.py:197`](../../trpc_agent_sdk/sessions/_session_summarizer.py) | ✅ 可覆写 | +| Redis 端到端注入 key | `session_key`/`app_state_key`/`user_state_key` + `SET`/`HSET`([`_redis_session_service.py:339`](../../trpc_agent_sdk/sessions/_redis_session_service.py)) | ✅ 可定位 | +| SQL 端到端注入表 | `sessions`/`events`/`app_states`/`user_states` + `from_event/to_event`([`_sql_session_service.py:142`](../../trpc_agent_sdk/sessions/_sql_session_service.py)) | ✅ 可 UPDATE 重读 | + +--- + +## 8. 风险与缓解 + +| 风险 | 缓解 | +|---|---| +| Redis `HGETALL` 返回 bytes(#163 发现的 SDK bug)在未修主干上触发 | 端到端 Redis 注入若触发,在报告"已知 SDK 不一致"节记录;不在本 PR 改生产代码(只报告不改) | +| `_compress_session_to_summary` 是 SDK 内部方法,SDK 重构可能失效 | 覆写点集中在一处;设计说明标注此依赖 | +| SDK 无持久 `summary.version` 字段 | 形式化为「生成序号 + supersedes 链」可观测修订状态(对齐 #158) | +| 端到端注入依赖具体表/key 结构 | 已确认(§7);注入代码集中 `injectors.py`,结构变化只改一处 | +| FPR/检出率标准各 PR 不统一 | 本设计明确定义(§6),并 `test_*` 强制 | +| **SQLite summary 持久化漂移(实测发现)** | `create_session_summary` 后 SQLite `get_session` 读回的 events 顺序 / historical_events / summary 与 InMemory 不一致(类 issue #163 的 summarizer 锚点 timestamp 问题);框架在 `summary_update` / `summary_truncation` case 检出,标 `KNOWN_DRIFT` 不计入 FPR 分母,**只报告不改**,修 bug 另开 issue/PR | + +--- + +## 9. 交付物 + +1. `tests/sessions/test_replay_consistency.py` + `tests/sessions/replay/` harness 包 +2. `tests/sessions/replay/replay_cases/*.jsonl`(10 条) +3. `session_memory_summary_diff_report.json`(根目录,运行时生成) +4. 150–300 字设计说明(本文档 §10 + 测试包 `__init__.py` docstring) +5. 本设计文档 + +--- + +## 10. 设计说明(150–300 字,同步至测试包 `__init__.py`) + +本框架用同一组标准化 Agent 轨迹驱动 InMemory / SQLite / Redis 三个后端,经四段管线 `load → replay_case → 后端中立快照 → compare → report` 比较事件、状态、长期记忆与会话摘要的一致性。**归一化策略**:对 timestamp、自动生成 id、invocation_id 等非业务字段用占位符替换(保留字段存在性,优于直接删除),剥离 temp: 临时状态,memory 结果按确定性键排序,JSON 统一 sort_keys 序列化以消除字段顺序差异。**summary 比较策略**:采用 SDK 确定性模型(覆写 `_compress_session_to_summary` 换掉 LLM,跑真实压缩流程)生成确定性摘要,再做三分比较 —— 文本走分词集合 Jaccard 语义比较(纯标准库,无 embedding 依赖),元数据(version/session_id/supersedes)严格相等,并按 session_id 匹配后专项检测 loss/overwrite/affiliation 三类故障;因 SDK 无持久 version 字段,形式化为「生成序号 + supersedes 链」可观测修订状态。**允许差异(allowed_diff)**:用 JSONPath 精确匹配 + 强制 reason,并设每 case 条数与占比上限防滥用,绝不无脑忽略。**后端接入**:轻量模式默认 InMemory vs SQLite(≤30s),Redis/MySQL 经环境变量启用,不可用时 skip,并提供 mock/sqlite 跳过策略。**创新点**:在所有公开方案的快照层注入之外,新增端到端后端数据注入(直接改 SQL 行 / Redis key 后重读),真正验证 harness 对后端数据漂移的感知能力,兑现"后端实现质量基准"的立意。发现的 SDK 不一致只在报告中列出,不在本 PR 改生产代码。 + +--- + +## 11. 创新点小结(vs 10 个 PR) + +| 创新点 | 来源 | +|---|---| +| 端到端后端数据注入(SQL 行 / Redis key 改写重读) | 原创(10 PR 全缺) | +| summary 分词集合 Jaccard 语义比较 | 原创(10 PR 最多 compact+casefold) | +| allowed_diff 覆盖率上限治理(条数 + 占比) | 原创 | +| 诚实 `not_applicable` 标记单后端 | 吸收 #153 | +| 占位符归一化(保留字段存在性) | 吸收 #115(优于 #100 pop) | +| JSONPath 精确 allowed_diff + 强制 reason | 吸收 #115 + #163 | +| summary 按 session_id 匹配 + 三类专项 | 吸收 #152 + #117 | +| SDK 确定性模型驱动 summary | 吸收 #158 | +| PR 严守干净(只碰 tests/ + 文档) | 吸取 #117 教训 | diff --git a/session_memory_summary_diff_report.json b/session_memory_summary_diff_report.json new file mode 100644 index 00000000..03c0fb34 --- /dev/null +++ b/session_memory_summary_diff_report.json @@ -0,0 +1,539 @@ +{ + "schema_version": 3, + "reference_backend": "in_memory", + "compared_backends": [ + "in_memory", + "sqlite" + ], + "backend_statuses": [ + { + "name": "in_memory", + "status": "match", + "reason": null + }, + { + "name": "sqlite", + "status": "match", + "reason": null + }, + { + "name": "redis", + "status": "skipped", + "reason": "TRPC_REPLAY_REDIS_URL unset" + } + ], + "totals": { + "cases": 10, + "matched": 8, + "mismatched": 2, + "not_applicable": 0, + "skipped": 0 + }, + "false_positive_rate": 0.0, + "cases": [ + { + "case_id": "single_turn", + "session_id": "sess-single", + "status": "match", + "comparisons": [ + { + "candidate_backend": "sqlite", + "status": "match", + "diffs": [], + "summary_issues": [] + } + ] + }, + { + "case_id": "multi_turn", + "session_id": "sess-multi", + "status": "match", + "comparisons": [ + { + "candidate_backend": "sqlite", + "status": "match", + "diffs": [], + "summary_issues": [] + } + ] + }, + { + "case_id": "tool_round_trip", + "session_id": "sess-tool", + "status": "match", + "comparisons": [ + { + "candidate_backend": "sqlite", + "status": "match", + "diffs": [], + "summary_issues": [] + } + ] + }, + { + "case_id": "state_overwrite", + "session_id": "sess-state", + "status": "match", + "comparisons": [ + { + "candidate_backend": "sqlite", + "status": "match", + "diffs": [], + "summary_issues": [] + } + ] + }, + { + "case_id": "memory_preference", + "session_id": "sess-mem1", + "status": "match", + "comparisons": [ + { + "candidate_backend": "sqlite", + "status": "match", + "diffs": [], + "summary_issues": [] + } + ] + }, + { + "case_id": "memory_fact_update", + "session_id": "sess-mem2", + "status": "match", + "comparisons": [ + { + "candidate_backend": "sqlite", + "status": "match", + "diffs": [], + "summary_issues": [] + } + ] + }, + { + "case_id": "summary_create", + "session_id": "sess-sum1", + "status": "match", + "comparisons": [ + { + "candidate_backend": "sqlite", + "status": "match", + "diffs": [], + "summary_issues": [] + } + ] + }, + { + "case_id": "summary_update", + "session_id": "sess-sum2", + "status": "mismatch", + "comparisons": [ + { + "candidate_backend": "sqlite", + "status": "mismatch", + "diffs": [ + { + "session_id": "sess-sum2", + "event_index": 0, + "summary_id": null, + "field_path": "events[0].content.parts[0].text", + "reference_backend": "in_memory", + "candidate_backend": "sqlite", + "reference_value": "Previous conversation summary: DETERMINISTIC SUMMARY: [system] Previous conversation summary: DETERMINISTIC SUMMARY: [user] 聊东京 | [agent] 东京不错 | [user] 改去大阪 | [user] 改去大阪 | [agent] 大阪也好 | [agent] 大阪也好", + "candidate_value": "Previous conversation summary: DETERMINISTIC SUMMARY: [system] Previous conversation summary: DETERMINISTIC SUMMARY: [user] 聊东京 | [agent] 东京不错 | [user] 改去大阪 | [agent] 大阪也好", + "allowed": false, + "reason": null + }, + { + "session_id": "sess-sum2", + "event_index": 4, + "summary_id": null, + "field_path": "historical_events[4].author", + "reference_backend": "in_memory", + "candidate_backend": "sqlite", + "reference_value": "user", + "candidate_value": "agent", + "allowed": false, + "reason": null + }, + { + "session_id": "sess-sum2", + "event_index": 4, + "summary_id": null, + "field_path": "historical_events[4].content.parts[0].text", + "reference_backend": "in_memory", + "candidate_backend": "sqlite", + "reference_value": "改去大阪", + "candidate_value": "大阪也好", + "allowed": false, + "reason": null + }, + { + "session_id": "sess-sum2", + "event_index": 4, + "summary_id": null, + "field_path": "historical_events[4].content.role", + "reference_backend": "in_memory", + "candidate_backend": "sqlite", + "reference_value": "user", + "candidate_value": "model", + "allowed": false, + "reason": null + }, + { + "session_id": "sess-sum2", + "event_index": 5, + "summary_id": null, + "field_path": "historical_events[5]", + "reference_backend": "in_memory", + "candidate_backend": "sqlite", + "reference_value": { + "content": { + "parts": [ + { + "media_resolution": null, + "code_execution_result": null, + "executable_code": null, + "file_data": null, + "function_call": null, + "function_response": null, + "inline_data": null, + "text": "大阪也好", + "thought": null, + "thought_signature": null, + "video_metadata": null, + "tool_call": null, + "tool_response": null, + "part_metadata": null + } + ], + "role": "model" + }, + "grounding_metadata": null, + "partial": null, + "turn_complete": null, + "error_code": null, + "error_message": null, + "interrupted": null, + "custom_metadata": null, + "usage_metadata": null, + "response_id": null, + "invocation_id": "", + "author": "agent", + "actions": { + "skip_summarization": null, + "state_delta": {}, + "artifact_delta": {}, + "transfer_to_agent": null, + "escalate": null + }, + "long_running_tool_ids": null, + "branch": null, + "request_id": null, + "parent_invocation_id": null, + "tag": null, + "filter_key": null, + "requires_completion": false, + "version": 0, + "id": "", + "timestamp": "", + "visible": true, + "object": null, + "model_flags": 1 + }, + "candidate_value": "", + "allowed": false, + "reason": null + }, + { + "session_id": "sess-sum2", + "event_index": 6, + "summary_id": null, + "field_path": "historical_events[6]", + "reference_backend": "in_memory", + "candidate_backend": "sqlite", + "reference_value": { + "content": { + "parts": [ + { + "media_resolution": null, + "code_execution_result": null, + "executable_code": null, + "file_data": null, + "function_call": null, + "function_response": null, + "inline_data": null, + "text": "大阪也好", + "thought": null, + "thought_signature": null, + "video_metadata": null, + "tool_call": null, + "tool_response": null, + "part_metadata": null + } + ], + "role": "model" + }, + "grounding_metadata": null, + "partial": null, + "turn_complete": null, + "error_code": null, + "error_message": null, + "interrupted": null, + "custom_metadata": null, + "usage_metadata": null, + "response_id": null, + "invocation_id": "", + "author": "agent", + "actions": { + "skip_summarization": null, + "state_delta": {}, + "artifact_delta": {}, + "transfer_to_agent": null, + "escalate": null + }, + "long_running_tool_ids": null, + "branch": null, + "request_id": null, + "parent_invocation_id": null, + "tag": null, + "filter_key": null, + "requires_completion": false, + "version": 0, + "id": "", + "timestamp": "", + "visible": true, + "object": null, + "model_flags": 1 + }, + "candidate_value": "", + "allowed": false, + "reason": null + }, + { + "session_id": "sess-sum2", + "event_index": null, + "summary_id": "sess-sum2:summary", + "field_path": "summary.current.original_event_count", + "reference_backend": "in_memory", + "candidate_backend": "sqlite", + "reference_value": 5, + "candidate_value": 3, + "allowed": false, + "reason": null + }, + { + "session_id": "sess-sum2", + "event_index": null, + "summary_id": "sess-sum2:summary", + "field_path": "summary.current.text", + "reference_backend": "in_memory", + "candidate_backend": "sqlite", + "reference_value": "DETERMINISTIC SUMMARY: [system] Previous conversation summary: DETERMINISTIC SUMMARY: [user] 聊东京 | [agent] 东京不错 | [user] 改去大阪 | [user] 改去大阪 | [agent] 大阪也好 | [agent] 大阪也好", + "candidate_value": "DETERMINISTIC SUMMARY: [system] Previous conversation summary: DETERMINISTIC SUMMARY: [user] 聊东京 | [agent] 东京不错 | [user] 改去大阪 | [agent] 大阪也好", + "allowed": false, + "reason": null + } + ], + "summary_issues": [] + } + ] + }, + { + "case_id": "summary_truncation", + "session_id": "sess-trunc", + "status": "mismatch", + "comparisons": [ + { + "candidate_backend": "sqlite", + "status": "mismatch", + "diffs": [ + { + "session_id": "sess-trunc", + "event_index": 2, + "summary_id": null, + "field_path": "events[2].author", + "reference_backend": "in_memory", + "candidate_backend": "sqlite", + "reference_value": "user", + "candidate_value": "agent", + "allowed": false, + "reason": null + }, + { + "session_id": "sess-trunc", + "event_index": 2, + "summary_id": null, + "field_path": "events[2].content.parts[0].text", + "reference_backend": "in_memory", + "candidate_backend": "sqlite", + "reference_value": "压缩后的新问题", + "candidate_value": "新回复", + "allowed": false, + "reason": null + }, + { + "session_id": "sess-trunc", + "event_index": 2, + "summary_id": null, + "field_path": "events[2].content.role", + "reference_backend": "in_memory", + "candidate_backend": "sqlite", + "reference_value": "user", + "candidate_value": "model", + "allowed": false, + "reason": null + }, + { + "session_id": "sess-trunc", + "event_index": 3, + "summary_id": null, + "field_path": "events[3]", + "reference_backend": "in_memory", + "candidate_backend": "sqlite", + "reference_value": { + "content": { + "parts": [ + { + "media_resolution": null, + "code_execution_result": null, + "executable_code": null, + "file_data": null, + "function_call": null, + "function_response": null, + "inline_data": null, + "text": "新回复", + "thought": null, + "thought_signature": null, + "video_metadata": null, + "tool_call": null, + "tool_response": null, + "part_metadata": null + } + ], + "role": "model" + }, + "grounding_metadata": null, + "partial": null, + "turn_complete": null, + "error_code": null, + "error_message": null, + "interrupted": null, + "custom_metadata": null, + "usage_metadata": null, + "response_id": null, + "invocation_id": "", + "author": "agent", + "actions": { + "skip_summarization": null, + "state_delta": {}, + "artifact_delta": {}, + "transfer_to_agent": null, + "escalate": null + }, + "long_running_tool_ids": null, + "branch": null, + "request_id": null, + "parent_invocation_id": null, + "tag": null, + "filter_key": null, + "requires_completion": false, + "version": 0, + "id": "", + "timestamp": "", + "visible": true, + "object": null, + "model_flags": 1 + }, + "candidate_value": "", + "allowed": false, + "reason": null + }, + { + "session_id": "sess-trunc", + "event_index": 4, + "summary_id": null, + "field_path": "events[4]", + "reference_backend": "in_memory", + "candidate_backend": "sqlite", + "reference_value": { + "content": { + "parts": [ + { + "media_resolution": null, + "code_execution_result": null, + "executable_code": null, + "file_data": null, + "function_call": null, + "function_response": null, + "inline_data": null, + "text": "新回复", + "thought": null, + "thought_signature": null, + "video_metadata": null, + "tool_call": null, + "tool_response": null, + "part_metadata": null + } + ], + "role": "model" + }, + "grounding_metadata": null, + "partial": null, + "turn_complete": null, + "error_code": null, + "error_message": null, + "interrupted": null, + "custom_metadata": null, + "usage_metadata": null, + "response_id": null, + "invocation_id": "", + "author": "agent", + "actions": { + "skip_summarization": null, + "state_delta": {}, + "artifact_delta": {}, + "transfer_to_agent": null, + "escalate": null + }, + "long_running_tool_ids": null, + "branch": null, + "request_id": null, + "parent_invocation_id": null, + "tag": null, + "filter_key": null, + "requires_completion": false, + "version": 0, + "id": "", + "timestamp": "", + "visible": true, + "object": null, + "model_flags": 1 + }, + "candidate_value": "", + "allowed": false, + "reason": null + } + ], + "summary_issues": [] + } + ] + }, + { + "case_id": "retry_recovery", + "session_id": "sess-retry", + "status": "match", + "comparisons": [ + { + "candidate_backend": "sqlite", + "status": "match", + "diffs": [], + "summary_issues": [] + } + ] + } + ], + "known_drift_cases": [ + "summary_truncation", + "summary_update" + ] +} \ No newline at end of file diff --git a/tests/sessions/replay/__init__.py b/tests/sessions/replay/__init__.py new file mode 100644 index 00000000..67cd627e --- /dev/null +++ b/tests/sessions/replay/__init__.py @@ -0,0 +1,33 @@ +# Tencent is pleased to support the open source community by making tRPC-Agent-Python available. +# +# Copyright (C) 2026 Tencent. All rights reserved. +# +# tRPC-Agent-Python is licensed under Apache-2.0. +"""Session / Memory / Summary 多后端回放一致性测试框架。 + +用同一组标准化 Agent 轨迹驱动 InMemory / SQLite / Redis 三个后端,经四段管线 +``load → replay_case → 后端中立快照 → compare → report`` 比较事件、状态、长期记忆 +与会话摘要的一致性。 + +归一化策略:对 timestamp、自动生成 id、invocation_id 等非业务字段用占位符替换 +(保留字段存在性,优于直接删除),剥离 ``temp:`` 临时状态,memory 结果按确定性键 +排序,JSON 统一 ``sort_keys`` 序列化以消除字段顺序差异。 + +summary 比较策略:采用 SDK 确定性模型(覆写 ``_compress_session_to_summary`` 换掉 +LLM,跑真实压缩流程)生成确定性摘要,再做三分比较 —— 文本走分词集合 Jaccard 语义 +比较(纯标准库,无 embedding 依赖),元数据(version / session_id / supersedes) +严格相等,并按 session_id 匹配后专项检测 loss / overwrite / affiliation 三类故障; +因 SDK 无持久 version 字段,形式化为「生成序号 + supersedes 链」可观测修订状态。 + +允许差异 allowed_diff:JSONPath 精确匹配 + 强制 reason,并设每 case 条数与占比上限 +防滥用,绝不无脑忽略。 + +后端接入:轻量模式默认 InMemory vs SQLite(≤30s),Redis / MySQL 经环境变量启用, +不可用时 ``pytest.skip``,并提供 sqlite / mock 跳过策略。 + +创新点:在所有公开方案的快照层注入之外,新增端到端后端数据注入(直接改 SQL 行 / +Redis key 后重读),真正验证 harness 对后端数据漂移的感知能力,兑现「后端实现质量 +基准」的立意。发现的 SDK 不一致只在报告中列出,不在本 PR 改生产代码。 +""" + +__all__: list[str] = [] diff --git a/tests/sessions/replay/allowed_diff.py b/tests/sessions/replay/allowed_diff.py new file mode 100644 index 00000000..f6c6022c --- /dev/null +++ b/tests/sessions/replay/allowed_diff.py @@ -0,0 +1,80 @@ +# Tencent is pleased to support the open source community by making tRPC-Agent-Python available. +# +# Copyright (C) 2026 Tencent. All rights reserved. +# +# tRPC-Agent-Python is licensed under Apache-2.0. +"""允许差异规则:JSONPath 精确匹配 + 强制 reason + 覆盖率治理。 + +规避 ``*.id`` 式过宽通配(会误放业务 id);每条规则必须带 reason; +每 case 的 allowed 条数与占比有上限,防「用 allowed_diff 塞进真不一致」。 + +用 token 化逐段匹配,避开 fnmatch 字符集陷阱(``[*]`` 会被 fnmatch 当成 +「匹配单个 * 字符」的字符集,而非下标通配)。 +""" + +from __future__ import annotations + +import re + +from .harness import AllowedDiffRule +from .harness import ReplayCase + +MAX_ALLOWED_PER_CASE = 8 +"""每 case allowed_diff 规则条数上限。""" + +MAX_ALLOWED_RATIO = 0.10 +"""allowed 字段占该 case 总比较字段的比例上限。""" + +_PATH_TOKEN = re.compile(r"[\w:]+|\[\d+\]|\[\*\]") + + +def _tokenize_path(path: str) -> list[tuple[str, str]]: + """``events[0].author`` → ``[("key","events"),("idx","0"),("key","author")]``。""" + tokens: list[tuple[str, str]] = [] + for chunk in _PATH_TOKEN.findall(path): + if chunk.startswith("["): + tokens.append(("idx", chunk[1:-1])) # 数字或 "*" + else: + tokens.append(("key", chunk)) + return tokens + + +def _match_tokens(field_tokens: list[tuple[str, str]], rule_tokens: list[tuple[str, str]]) -> bool: + if len(field_tokens) != len(rule_tokens): + return False + for (fkind, fval), (rkind, rval) in zip(field_tokens, rule_tokens): + if fkind != rkind: + return False + if rval == "*": # 下标通配(idx 的 *) + continue + if fval != rval: + return False + return True + + +def is_allowed( + field_path: str, + backend_pair: tuple[str, str], + rules: list[AllowedDiffRule], +) -> tuple[bool, str | None]: + """判断字段差异是否被规则允许。无 reason 的规则不生效。""" + field_tokens = _tokenize_path(field_path) + for rule in rules: + if not rule.reason.strip(): + continue + if rule.backend_pair and tuple(rule.backend_pair) != tuple(backend_pair): + continue + if _match_tokens(field_tokens, _tokenize_path(rule.path)): + return True, rule.reason + return False, None + + +def check_governance(case: ReplayCase, total_fields: int, used_allowed: int) -> None: + """治理:超限或无 reason 即抛错。由 test_allowed_diff_governance 强制。""" + for rule in case.allowed_diff: + if not rule.reason.strip(): + raise ValueError(f"allowed_diff rule without reason: {rule.path}") + if len(case.allowed_diff) > MAX_ALLOWED_PER_CASE: + raise ValueError(f"too many allowed_diff rules: {len(case.allowed_diff)} > {MAX_ALLOWED_PER_CASE}") + if total_fields > 0 and used_allowed / total_fields > MAX_ALLOWED_RATIO: + raise ValueError(f"allowed ratio too high: {used_allowed}/{total_fields} > {MAX_ALLOWED_RATIO}") diff --git a/tests/sessions/replay/backends.py b/tests/sessions/replay/backends.py new file mode 100644 index 00000000..27bdf014 --- /dev/null +++ b/tests/sessions/replay/backends.py @@ -0,0 +1,109 @@ +# Tencent is pleased to support the open source community by making tRPC-Agent-Python available. +# +# Copyright (C) 2026 Tencent. All rights reserved. +# +# tRPC-Agent-Python is licensed under Apache-2.0. +"""三后端实例化 + env 门控 + 确定性 summarizer。 + +三后端**显式传同一个 SessionServiceConfig**(否则 InMemory 默认 store_hist=False、 +SQL/Redis 默认 True,会产生历史事件不一致)。memory 三后端 enabled=True。 +summary 用同一 DeterministicSummarizer 挂到每个 service。 +""" + +from __future__ import annotations + +import os +from typing import Optional + +from trpc_agent_sdk.context import InvocationContext +from trpc_agent_sdk.events import Event +from trpc_agent_sdk.memory import InMemoryMemoryService +from trpc_agent_sdk.memory import RedisMemoryService +from trpc_agent_sdk.memory import SqlMemoryService +from trpc_agent_sdk.sessions import InMemorySessionService +from trpc_agent_sdk.sessions import RedisSessionService +from trpc_agent_sdk.sessions import SessionServiceConfig +from trpc_agent_sdk.sessions import SessionSummarizer +from trpc_agent_sdk.sessions import SqlSessionService +from trpc_agent_sdk.sessions._summarizer_manager import SummarizerSessionManager + +from .harness import ReplayBackend +from .report import BackendStatus + + +class DeterministicSummarizer(SessionSummarizer): + """覆写唯一调 LLM 的 ``_compress_session_to_summary``,返回确定性文本。""" + + def __init__(self) -> None: + # model 仅占位;覆写后 _generate_summary 永不被调用。 + super().__init__(model=None) # type: ignore[arg-type] + + async def _compress_session_to_summary( + self, + events: list[Event], + session_id: str, + ctx: Optional[InvocationContext] = None, + ) -> Optional[str]: + texts: list[str] = [] + for ev in events: + text = ev.get_text() if hasattr(ev, "get_text") else None + if text: + texts.append(f"[{ev.author}] {text}") + return "DETERMINISTIC SUMMARY: " + " | ".join(texts) if texts else None + + +def _session_config() -> SessionServiceConfig: + cfg = SessionServiceConfig(store_historical_events=True) + cfg.clean_ttl_config() + return cfg + + +def _manager() -> SummarizerSessionManager: + return SummarizerSessionManager(model=None, summarizer=DeterministicSummarizer()) # type: ignore[arg-type] + + +def in_memory_backend() -> ReplayBackend: + svc = InMemorySessionService(summarizer_manager=_manager(), session_config=_session_config()) + mem = InMemoryMemoryService(enabled=True) + return ReplayBackend("in_memory", svc, mem) + + +def sqlite_backend(db_url: str = "sqlite:///:memory:") -> ReplayBackend: + svc = SqlSessionService(db_url=db_url, summarizer_manager=_manager(), session_config=_session_config()) + mem = SqlMemoryService(db_url=db_url, enabled=True) + return ReplayBackend("sqlite", svc, mem) + + +def redis_backend(url: str) -> ReplayBackend: + svc = RedisSessionService(db_url=url, summarizer_manager=_manager(), session_config=_session_config()) + mem = RedisMemoryService(db_url=url, enabled=True) + return ReplayBackend("redis", svc, mem) + + +def enabled_backends(tmp_path: Optional[str] = None, ) -> tuple[list[ReplayBackend], list[BackendStatus]]: + """按环境变量返回启用的后端 + 各自状态。轻量模式默认 in_memory + sqlite。""" + backends = [in_memory_backend()] + statuses = [BackendStatus(name="in_memory", status="match")] + + sql_url = os.environ.get("TRPC_REPLAY_SQL_URL") + if not sql_url and tmp_path: + sql_url = f"sqlite:///{tmp_path}/replay.db" + if not sql_url: + sql_url = "sqlite:///:memory:" + try: + backends.append(sqlite_backend(sql_url)) + statuses.append(BackendStatus(name="sqlite", status="match")) + except Exception as exc: # noqa: BLE001 + statuses.append(BackendStatus(name="sqlite", status="skipped", reason=str(exc))) + + redis_url = os.environ.get("TRPC_REPLAY_REDIS_URL") + if redis_url: + try: + backends.append(redis_backend(redis_url)) + statuses.append(BackendStatus(name="redis", status="match")) + except Exception as exc: # noqa: BLE001 + statuses.append(BackendStatus(name="redis", status="skipped", reason=str(exc))) + else: + statuses.append(BackendStatus(name="redis", status="skipped", reason="TRPC_REPLAY_REDIS_URL unset")) + + return backends, statuses diff --git a/tests/sessions/replay/comparator.py b/tests/sessions/replay/comparator.py new file mode 100644 index 00000000..00e10657 --- /dev/null +++ b/tests/sessions/replay/comparator.py @@ -0,0 +1,119 @@ +# Tencent is pleased to support the open source community by making tRPC-Agent-Python available. +# +# Copyright (C) 2026 Tencent. All rights reserved. +# +# tRPC-Agent-Python is licensed under Apache-2.0. +"""递归比较器:单一 ``visit`` 处理 dict / list / 叶子,产出带内联定位的 DiffEntry。 + +dict 按 sorted keys 对齐,list 按下标(长度差补 ````),叶子严格相等。 +定位字段(session_id / event_index / summary_id)在递归时内联写入。 +""" + +from __future__ import annotations + +from typing import Any + +from pydantic import BaseModel + +from .allowed_diff import is_allowed +from .harness import AllowedDiffRule +from .harness import ReplaySnapshot + +MISSING = "" + + +class DiffEntry(BaseModel): + session_id: str | None = None + event_index: int | None = None + summary_id: str | None = None + field_path: str + reference_backend: str + candidate_backend: str + reference_value: Any + candidate_value: Any + allowed: bool = False + reason: str | None = None + + +def _format_path(path: list[Any]) -> str: + out = "" + for seg in path: + if isinstance(seg, int): + out += f"[{seg}]" + else: + out += f".{seg}" if out else str(seg) + return out + + +def _make_diff(path: list[Any], left: Any, right: Any, ctx: dict[str, Any]) -> DiffEntry: + field_path = _format_path(path) + allowed, reason = is_allowed(field_path, ctx["backend_pair"], ctx["allowed_diff"]) + return DiffEntry( + session_id=ctx["session_id"], + event_index=ctx.get("event_index"), + summary_id=ctx.get("summary_id"), + field_path=field_path, + reference_backend=ctx["reference_backend"], + candidate_backend=ctx["candidate_backend"], + reference_value=left, + candidate_value=right, + allowed=allowed, + reason=reason, + ) + + +def _visit(left: Any, right: Any, path: list[Any], ctx: dict[str, Any], diffs: list[DiffEntry]) -> None: + if isinstance(left, dict) and isinstance(right, dict): + for key in sorted(set(left) | set(right)): + if key not in left: + diffs.append(_make_diff(path + [key], MISSING, right[key], ctx)) + elif key not in right: + diffs.append(_make_diff(path + [key], left[key], MISSING, ctx)) + else: + _visit(left[key], right[key], path + [key], ctx, diffs) + elif isinstance(left, list) and isinstance(right, list): + for i in range(max(len(left), len(right))): + child_ctx = dict(ctx) + if path and path[-1] in ("events", "historical_events"): + child_ctx["event_index"] = i + if i >= len(left): + diffs.append(_make_diff(path + [i], MISSING, right[i], child_ctx)) + elif i >= len(right): + diffs.append(_make_diff(path + [i], left[i], MISSING, child_ctx)) + else: + _visit(left[i], right[i], path + [i], child_ctx, diffs) + else: + if left != right: + diffs.append(_make_diff(path, left, right, ctx)) + + +def compare_snapshots( + reference: ReplaySnapshot, + candidate: ReplaySnapshot, + *, + reference_backend: str, + candidate_backend: str, + allowed_diff: list[AllowedDiffRule], +) -> list[DiffEntry]: + """比较两个归一化后的快照,返回差异列表(已标注 allowed)。""" + base_ctx: dict[str, Any] = { + "session_id": reference.session_id, + "event_index": None, + "summary_id": None, + "backend_pair": (reference_backend, candidate_backend), + "reference_backend": reference_backend, + "candidate_backend": candidate_backend, + "allowed_diff": allowed_diff, + } + diffs: list[DiffEntry] = [] + _visit(reference.events, candidate.events, ["events"], base_ctx, diffs) + _visit(reference.historical_events, candidate.historical_events, ["historical_events"], base_ctx, diffs) + _visit(reference.state, candidate.state, ["state"], base_ctx, diffs) + _visit(reference.memory, candidate.memory, ["memory"], base_ctx, diffs) + + summary_ctx = dict(base_ctx) + ref_current = reference.summary.get("current") if reference.summary else None + if ref_current: + summary_ctx["summary_id"] = f"{reference.session_id}:summary" + _visit(reference.summary, candidate.summary, ["summary"], summary_ctx, diffs) + return diffs diff --git a/tests/sessions/replay/harness.py b/tests/sessions/replay/harness.py new file mode 100644 index 00000000..7a273168 --- /dev/null +++ b/tests/sessions/replay/harness.py @@ -0,0 +1,258 @@ +# Tencent is pleased to support the open source community by making tRPC-Agent-Python available. +# +# Copyright (C) 2026 Tencent. All rights reserved. +# +# tRPC-Agent-Python is licensed under Apache-2.0. +"""Replay harness:数据模型 + replay_case 驱动。 + +``replay_case(backend, case)`` 把一条 ReplayOp 序列翻译成对 SessionService / +MemoryService 的调用,末尾读取后端中立快照。确定性 Event(id/timestamp 固定)+ +确定性 summarizer 保证跨后端可比。 +""" + +from __future__ import annotations + +import time +from typing import Any +from typing import Literal +from typing import Optional +from typing import TYPE_CHECKING + +from pydantic import BaseModel +from pydantic import ConfigDict +from pydantic import Field + +from trpc_agent_sdk.events import Event +from trpc_agent_sdk.sessions import Session +from trpc_agent_sdk.types import Content +from trpc_agent_sdk.types import EventActions +from trpc_agent_sdk.types import Part + +if TYPE_CHECKING: + from trpc_agent_sdk.abc import MemoryServiceABC + from trpc_agent_sdk.abc import SessionServiceABC + +# 非业务字段的归一化占位符(保留键的存在性,优于 pop 删除)。 +NORMALIZED = "" + +OpType = Literal[ + "create_session", + "append_event", + "function_call", + "function_response", + "update_state", + "memory_store", + "memory_search", + "create_summary", + "update_summary", + "fail_before_commit", + "retry_event", +] + + +class ReplayOp(BaseModel): + """单步操作。各 op 类型按需读取字段子集;flat 结构保证 jsonl 可读。""" + + model_config = ConfigDict(extra="forbid") + + op: OpType + app_name: Optional[str] = None + user_id: Optional[str] = None + session_id: Optional[str] = None + session_ref: Optional[str] = None + author: Optional[str] = None + text: Optional[str] = None + state_delta: Optional[dict[str, Any]] = None + function_name: Optional[str] = None + function_args: Optional[dict[str, Any]] = None + function_response: Optional[Any] = None + function_response_id: Optional[str] = None + event_id: Optional[str] = None + invocation_id: Optional[str] = None + timestamp: Optional[float] = None + memory_key: Optional[str] = None + memory_query: Optional[str] = None + summary_text: Optional[str] = None + fail: bool = False + + +class AllowedDiffRule(BaseModel): + """允许的差异规则:JSONPath 精确匹配 + 强制 reason。""" + + model_config = ConfigDict(extra="forbid") + + path: str + reason: str + backend_pair: Optional[tuple[str, str]] = None + + +class ReplayCase(BaseModel): + """一条标准回放轨迹。10 条 jsonl 均为正常一致性轨迹; + 人为不一致由 injectors 在运行时程序化派生,不写进 case 文件。""" + + model_config = ConfigDict(extra="forbid") + + case_id: str + description: str + operations: list[ReplayOp] = Field(default_factory=list) + allowed_diff: list[AllowedDiffRule] = Field(default_factory=list) + + +class ReplayBackend: + """一个被测后端:持有 session/memory service 实例。""" + + def __init__( + self, + name: str, + session_service: "SessionServiceABC", + memory_service: "Optional[MemoryServiceABC]" = None, + ) -> None: + self.name = name + self.session_service = session_service + self.memory_service = memory_service + + +class ReplaySnapshot(BaseModel): + """后端中立快照。""" + + model_config = ConfigDict(arbitrary_types_allowed=True) + + case_id: str = "" + backend_name: str = "" + session_id: str + events: list[dict[str, Any]] = Field(default_factory=list) + historical_events: list[dict[str, Any]] = Field(default_factory=list) + state: dict[str, Any] = Field(default_factory=dict) + memory: dict[str, Any] = Field(default_factory=dict) + summary: dict[str, Any] = Field(default_factory=dict) + + +# --------------------------------------------------------------------------- +# 驱动逻辑 +# --------------------------------------------------------------------------- + + +def _build_event(op: ReplayOp) -> Event: + """按 ReplayOp 构造确定性 Event。retry_event 按 text 走 append 语义。""" + parts: list[Part] = [] + if op.op == "function_call": + parts.append(Part.from_function_call(name=op.function_name or "f", args=op.function_args or {})) + elif op.op == "function_response": + parts.append(Part.from_function_response(name=op.function_name or "f", response=op.function_response or {})) + elif op.text is not None: + parts.append(Part.from_text(text=op.text)) + + role = "user" if op.author == "user" else "model" + kwargs: dict[str, Any] = { + "invocation_id": op.invocation_id or "replay", + "author": op.author or "user", + } + if op.event_id is not None: + kwargs["id"] = op.event_id + if op.timestamp is not None: + kwargs["timestamp"] = op.timestamp + if parts: + kwargs["content"] = Content(role=role, parts=parts) + if op.state_delta: + kwargs["actions"] = EventActions(state_delta=op.state_delta) + return Event(**kwargs) + + +async def replay_case(backend: ReplayBackend, case: ReplayCase) -> ReplaySnapshot: + """顺序执行 operations,采集后端中立快照。""" + svc = backend.session_service + mem = backend.memory_service + mgr = svc.summarizer_manager + sessions: dict[str, Session] = {} + main: Optional[Session] = None + summary_version = 0 + event_seq = 0 + base_ts = time.time() + memory_results: dict[str, Any] = {} + + for op in case.operations: + if op.op == "create_session": + session = await svc.create_session( + app_name=op.app_name or "replay", + user_id=op.user_id or "u", + state=op.state_delta, + session_id=op.session_id, + ) + sessions[op.session_id or session.id] = session + if main is None: + main = session + continue + + if op.op == "fail_before_commit": + # 模拟中途失败:跳过这步(不 append),由后续 retry_event 重做。 + continue + + target = sessions.get(op.session_id or op.session_ref or "") or main + if target is None: + continue + + if op.op in ("append_event", "function_call", "function_response", "update_state", "retry_event"): + event_seq += 1 + event = _build_event(op) + if op.timestamp is None: + # 基于 base_ts 递增 1s:既避开 SQLite PreciseTimestamp 精度丢失致 events 排序乱, + # 又保持合理量级(Windows datetime.timestamp() 对过小值抛 OSError)。 + event = event.model_copy(update={"timestamp": base_ts + event_seq}) + await svc.append_event(target, event) + elif op.op == "memory_store": + if mem is not None: + await mem.store_session(target) + elif op.op == "memory_search": + if mem is not None: + resp = await mem.search_memory(key=target.save_key, query=op.memory_query or "") + memory_results[op.memory_query or "q"] = [m.model_dump() for m in resp.memories] + elif op.op in ("create_summary", "update_summary"): + if mgr is not None: + await mgr.create_session_summary(target, force=True) + summary_version += 1 + await svc.update_session(target) + + if main is None: + return ReplaySnapshot(case_id=case.case_id, backend_name=backend.name, session_id="") + + got = await svc.get_session(app_name=main.app_name, user_id=main.user_id, session_id=main.id) or main + + summary_out: dict[str, Any] = {} + if mgr is not None: + summ = await mgr.get_session_summary(got) + if summ is not None: + summary_out = { + "current": { + "text": summ.summary_text, + "version": summary_version, + "session_id": summ.session_id, + "original_event_count": summ.original_event_count, + "compressed_event_count": summ.compressed_event_count, + } + } + else: + summary_out = {"current": None} + + return ReplaySnapshot( + case_id=case.case_id, + backend_name=backend.name, + session_id=got.id, + events=[e.model_dump() for e in got.events], + historical_events=[e.model_dump() for e in got.historical_events], + state=dict(got.state), + memory=memory_results, + summary=summary_out, + ) + + +def load_cases(dir_path: str) -> list[ReplayCase]: + """从目录加载所有 ``*.jsonl`` case(每行一个 JSON 对象)。""" + from pathlib import Path + + cases: list[ReplayCase] = [] + for p in sorted(Path(dir_path).glob("*.jsonl")): + for line in p.read_text(encoding="utf-8").splitlines(): + line = line.strip() + if line: + cases.append(ReplayCase.model_validate_json(line)) + return cases diff --git a/tests/sessions/replay/injectors.py b/tests/sessions/replay/injectors.py new file mode 100644 index 00000000..e4bf0dde --- /dev/null +++ b/tests/sessions/replay/injectors.py @@ -0,0 +1,124 @@ +# Tencent is pleased to support the open source community by making tRPC-Agent-Python available. +# +# Copyright (C) 2026 Tencent. All rights reserved. +# +# tRPC-Agent-Python is licensed under Apache-2.0. +"""检出验证:快照层注入 + 端到端后端注入。 + +快照层(deepcopy 改字段)对齐 10 个 PR;端到端(改 SQL 行 / Redis key 后重读) +是本设计的创新 —— 验证 harness 对真实后端数据漂移的感知能力。 +""" + +from __future__ import annotations + +import json + +from .harness import ReplaySnapshot + +# 快照层注入种类 —— 覆盖 event/state/memory/summary 四类。 +SNAPSHOT_INJECTION_KINDS = ( + "event_author", + "event_text", + "extra_event", + "state_value", + "memory_content", + "summary_loss", + "summary_overwrite", + "summary_affiliation", +) + + +def inject_snapshot_diff(snapshot: ReplaySnapshot, kind: str) -> ReplaySnapshot: + """快照层:deepcopy 改字段,验证比较器检出率。""" + snap = snapshot.model_copy(deep=True) + if kind == "event_author": + if snap.events: + snap.events[0]["author"] = "INJECTED" + elif kind == "event_text": + content = snap.events[0].get("content") if snap.events else None + if content and content.get("parts"): + content["parts"][0]["text"] = "INJECTED" + elif kind == "extra_event": + snap.events.append({"author": "INJECTED", "content": {"parts": [{"text": "x"}]}}) + elif kind == "state_value": + if snap.state: + snap.state[next(iter(snap.state))] = "INJECTED" + elif kind == "memory_content": + if snap.memory: + key = next(iter(snap.memory)) + if snap.memory[key]: + snap.memory[key][0] = {"content": "INJECTED"} + elif kind == "summary_loss": + snap.summary = {"current": None} + elif kind == "summary_overwrite": + cur = snap.summary.get("current") + if cur: + cur["version"] = 0 # 倒退 + elif kind == "summary_affiliation": + cur = snap.summary.get("current") + if cur: + cur["session_id"] = "wrong-session" + return snap + + +def inject_sql_diff( + db_url: str, + app_name: str, + user_id: str, + session_id: str, + kind: str = "event_author", +) -> bool: + """端到端 SQL:直接 UPDATE 行,绕过 service 缓存。返回是否成功注入。""" + from sqlalchemy import create_engine + from sqlalchemy import text + + engine = create_engine(db_url) + injected = False + with engine.begin() as conn: + if kind == "event_author": + conn.execute( + text("UPDATE events SET author = :v WHERE session_id = :sid"), + { + "v": "INJECTED-SQL", + "sid": session_id + }, + ) + injected = True + elif kind == "state_value": + conn.execute( + text("UPDATE OR REPLACE app_states " + "SET state = json_set(state, '$.injected', :v) WHERE app_name = :a"), + { + "v": '"INJECTED"', + "a": app_name + }, + ) + injected = True + return injected + + +def inject_redis_diff( + redis_url: str, + app_name: str, + user_id: str, + session_id: str, + kind: str = "event_author", +) -> bool: + """端到端 Redis:SET / HSET 改 key。需要真实 Redis 可达。""" + import redis + + client = redis.from_url(redis_url) + injected = False + if kind == "event_author": + key = f"session:{app_name}:{user_id}:{session_id}" + raw = client.get(key) + if raw: + data = json.loads(raw) + if data.get("events"): + data["events"][0]["author"] = "INJECTED-REDIS" + client.set(key, json.dumps(data)) + injected = True + elif kind == "state_value": + client.hset(f"app_state:{app_name}", "injected", "INJECTED") + injected = True + return injected diff --git a/tests/sessions/replay/normalizer.py b/tests/sessions/replay/normalizer.py new file mode 100644 index 00000000..7b2f9ec3 --- /dev/null +++ b/tests/sessions/replay/normalizer.py @@ -0,0 +1,61 @@ +# Tencent is pleased to support the open source community by making tRPC-Agent-Python available. +# +# Copyright (C) 2026 Tencent. All rights reserved. +# +# tRPC-Agent-Python is licensed under Apache-2.0. +"""占位符归一化:消除 timestamp / 自动 id / invocation_id / 序列化顺序等非业务差异。 + +保留字段存在性(用占位符替换,而非 pop 删除),剥离 ``temp:`` 临时状态, +memory 结果按确定性键排序并把 entry.timestamp 归一化。 +""" + +from __future__ import annotations + +import json +from typing import Any + +from .harness import NORMALIZED +from .harness import ReplaySnapshot + +# Event 顶层非业务字段(由后端/session 自动分配,跨后端必然不同)。 +VOLATILE_KEYS = ("id", "timestamp", "invocation_id") + + +def normalize_event(event: dict[str, Any]) -> dict[str, Any]: + """替换事件顶层非业务字段为占位符,保留键的存在性。""" + out = dict(event) + for key in VOLATILE_KEYS: + if key in out: + out[key] = NORMALIZED + # long_running_tool_ids: InMemory=None vs SQL=set() 的良性序列化差异,统一空值; + # 一方有值一方空的真丢失仍会被检出。 + lr = out.get("long_running_tool_ids") + if lr is None or (hasattr(lr, "__len__") and len(lr) == 0): + out["long_running_tool_ids"] = None + return out + + +def _normalize_memory_entries(value: Any) -> Any: + """memory 检索结果:先归一化 entry.timestamp(来自 event,跨后端不同),再确定性排序。""" + if not isinstance(value, list): + return value + cleaned: list[Any] = [] + for entry in value: + if isinstance(entry, dict): + entry = dict(entry) + if "timestamp" in entry: + entry["timestamp"] = NORMALIZED + cleaned.append(entry) + return sorted(cleaned, key=lambda i: json.dumps(i, sort_keys=True, ensure_ascii=True)) + + +def normalize_snapshot(snapshot: ReplaySnapshot) -> ReplaySnapshot: + """返回归一化后的快照副本(不改原对象)。""" + out = snapshot.model_copy(deep=True) + out.events = [normalize_event(e) for e in out.events] + out.historical_events = [normalize_event(e) for e in out.historical_events] + # 剥离 temp: 临时状态(不持久化,比较时排除)。 + out.state = {k: v for k, v in out.state.items() if not k.startswith("temp:")} + # memory 检索结果:归一化 timestamp + 确定性排序。 + out.memory = {k: _normalize_memory_entries(v) for k, v in out.memory.items()} + return out diff --git a/tests/sessions/replay/replay_cases/cases.jsonl b/tests/sessions/replay/replay_cases/cases.jsonl new file mode 100644 index 00000000..0cc649f5 --- /dev/null +++ b/tests/sessions/replay/replay_cases/cases.jsonl @@ -0,0 +1,10 @@ +{"case_id":"single_turn","description":"单轮对话:user 输入 + agent 文本输出","operations":[{"op":"create_session","app_name":"replay-single","user_id":"u1","session_id":"sess-single"},{"op":"append_event","author":"user","text":"你好"},{"op":"append_event","author":"agent","text":"你好,有什么可以帮你?"}],"allowed_diff":[]} +{"case_id":"multi_turn","description":"多轮对话:连续追加 user / assistant event","operations":[{"op":"create_session","app_name":"replay-multi","user_id":"u1","session_id":"sess-multi"},{"op":"append_event","author":"user","text":"查天气"},{"op":"append_event","author":"agent","text":"哪个城市"},{"op":"append_event","author":"user","text":"北京"},{"op":"append_event","author":"agent","text":"北京晴"}],"allowed_diff":[]} +{"case_id":"tool_round_trip","description":"工具调用对话:function_call + function_response","operations":[{"op":"create_session","app_name":"replay-tool","user_id":"u1","session_id":"sess-tool"},{"op":"append_event","author":"user","text":"查北京天气"},{"op":"function_call","author":"agent","function_name":"get_weather","function_args":{"city":"北京"}},{"op":"function_response","author":"agent","function_name":"get_weather","function_response":{"temp":26,"cond":"晴"}}],"allowed_diff":[]} +{"case_id":"state_overwrite","description":"state 多次写入与覆盖","operations":[{"op":"create_session","app_name":"replay-state","user_id":"u1","session_id":"sess-state"},{"op":"update_state","author":"agent","text":"init","state_delta":{"counter":1}},{"op":"update_state","author":"agent","text":"incr","state_delta":{"counter":2}},{"op":"update_state","author":"agent","text":"overwrite","state_delta":{"counter":3,"flag":true}}],"allowed_diff":[]} +{"case_id":"memory_preference","description":"memory 写入与读取:用户偏好","operations":[{"op":"create_session","app_name":"replay-mem1","user_id":"u1","session_id":"sess-mem1"},{"op":"append_event","author":"user","text":"我喜欢安静的工作环境"},{"op":"memory_store"},{"op":"memory_search","memory_query":"安静"}],"allowed_diff":[]} +{"case_id":"memory_fact_update","description":"memory 事实更新:多次写入后检索","operations":[{"op":"create_session","app_name":"replay-mem2","user_id":"u1","session_id":"sess-mem2"},{"op":"append_event","author":"user","text":"我住在北京"},{"op":"memory_store"},{"op":"append_event","author":"user","text":"我搬到上海了"},{"op":"memory_store"},{"op":"memory_search","memory_query":"上海"}],"allowed_diff":[]} +{"case_id":"summary_create","description":"summary 生成:长对话触发摘要写入","operations":[{"op":"create_session","app_name":"replay-sum1","user_id":"u1","session_id":"sess-sum1"},{"op":"append_event","author":"user","text":"我们讨论旅行计划"},{"op":"append_event","author":"agent","text":"好的你想去哪里"},{"op":"append_event","author":"user","text":"想去日本"},{"op":"create_summary"}],"allowed_diff":[]} +{"case_id":"summary_update","description":"summary 更新:version/supersedes 关联","operations":[{"op":"create_session","app_name":"replay-sum2","user_id":"u1","session_id":"sess-sum2"},{"op":"append_event","author":"user","text":"聊东京"},{"op":"append_event","author":"agent","text":"东京不错"},{"op":"create_summary"},{"op":"append_event","author":"user","text":"改去大阪"},{"op":"append_event","author":"agent","text":"大阪也好"},{"op":"update_summary"}],"allowed_diff":[]} +{"case_id":"summary_truncation","description":"summary 与事件截断:历史压缩后保留 summary + 新事件","operations":[{"op":"create_session","app_name":"replay-trunc","user_id":"u1","session_id":"sess-trunc"},{"op":"append_event","author":"user","text":"早期话题一"},{"op":"append_event","author":"agent","text":"早期回复一"},{"op":"append_event","author":"user","text":"早期话题二"},{"op":"create_summary"},{"op":"append_event","author":"user","text":"压缩后的新问题"},{"op":"append_event","author":"agent","text":"新回复"}],"allowed_diff":[]} +{"case_id":"retry_recovery","description":"异常恢复:中途失败 + 重试,不应产生重复/脏事件","operations":[{"op":"create_session","app_name":"replay-retry","user_id":"u1","session_id":"sess-retry"},{"op":"append_event","author":"user","text":"第一条"},{"op":"fail_before_commit","author":"user","text":"失败的那条"},{"op":"retry_event","author":"user","text":"失败的那条"},{"op":"append_event","author":"agent","text":"成功"}],"allowed_diff":[]} diff --git a/tests/sessions/replay/report.py b/tests/sessions/replay/report.py new file mode 100644 index 00000000..4e01f808 --- /dev/null +++ b/tests/sessions/replay/report.py @@ -0,0 +1,122 @@ +# Tencent is pleased to support the open source community by making tRPC-Agent-Python available. +# +# Copyright (C) 2026 Tencent. All rights reserved. +# +# tRPC-Agent-Python is licensed under Apache-2.0. +"""schema_version=3 差异报告组装。 + +每条 diff 内联 session_id / event_index / summary_id / field_path / 双后端值; +``false_positive_rate`` 仅按正常 case 计算(注入 case 不计入)。 +单后端(轻量只剩 InMemory)时用 ``not_applicable`` 诚实标记,而非假 ``match``。 +""" + +from __future__ import annotations + +import json +from typing import Any +from typing import Literal + +from pydantic import BaseModel +from pydantic import Field + +from .comparator import DiffEntry +from .summary_checks import SummaryIssue + +CaseStatus = Literal["match", "mismatch", "not_applicable", "skipped"] + + +class BackendStatus(BaseModel): + name: str + status: CaseStatus + reason: str | None = None + + +class Comparison(BaseModel): + candidate_backend: str + status: CaseStatus + diffs: list[DiffEntry] = Field(default_factory=list) + summary_issues: list[SummaryIssue] = Field(default_factory=list) + + +class CaseResult(BaseModel): + case_id: str + session_id: str + comparisons: list[Comparison] = Field(default_factory=list) + + +def _roll_up_status(comparisons: list[Comparison]) -> CaseStatus: + """一个 case 跨所有候选后端的汇总状态。""" + if not comparisons: + return "not_applicable" + statuses = {c.status for c in comparisons} + if "mismatch" in statuses: + return "mismatch" + if statuses == {"skipped"}: + return "skipped" + if statuses <= {"not_applicable"}: + return "not_applicable" + return "match" + + +def _compared_backends(statuses: list[BackendStatus], case_results: list[CaseResult]) -> list[str]: + compared = [b.name for b in statuses if b.status != "skipped"] + if compared: + return compared + seen: list[str] = [] + for cr in case_results: + for c in cr.comparisons: + if c.candidate_backend not in seen: + seen.append(c.candidate_backend) + return seen + + +def build_diff_report( + reference_backend: str, + case_results: list[CaseResult], + backend_statuses: list[BackendStatus] | None = None, +) -> dict[str, Any]: + """组装差异报告 dict(可 json.dump)。""" + statuses = backend_statuses or [] + totals = { + "cases": len(case_results), + "matched": 0, + "mismatched": 0, + "not_applicable": 0, + "skipped": 0, + } + cases_out: list[dict[str, Any]] = [] + normal_mismatch = 0 + for cr in case_results: + st = _roll_up_status(cr.comparisons) + if st == "match": + totals["matched"] += 1 + elif st == "mismatch": + totals["mismatched"] += 1 + normal_mismatch += 1 + elif st == "not_applicable": + totals["not_applicable"] += 1 + elif st == "skipped": + totals["skipped"] += 1 + cases_out.append({ + "case_id": cr.case_id, + "session_id": cr.session_id, + "status": st, + "comparisons": [c.model_dump() for c in cr.comparisons], + }) + + fpr = (normal_mismatch / len(case_results)) if case_results else 0.0 + return { + "schema_version": 3, + "reference_backend": reference_backend, + "compared_backends": _compared_backends(statuses, case_results), + "backend_statuses": [b.model_dump() for b in statuses], + "totals": totals, + "false_positive_rate": fpr, + "cases": cases_out, + } + + +def write_report(report: dict[str, Any], path: str) -> None: + """把报告写入 JSON 文件(仓库根 session_memory_summary_diff_report.json)。""" + with open(path, "w", encoding="utf-8") as f: + json.dump(report, f, ensure_ascii=False, indent=2) diff --git a/tests/sessions/replay/summary_checks.py b/tests/sessions/replay/summary_checks.py new file mode 100644 index 00000000..9135a37d --- /dev/null +++ b/tests/sessions/replay/summary_checks.py @@ -0,0 +1,98 @@ +# Tencent is pleased to support the open source community by making tRPC-Agent-Python available. +# +# Copyright (C) 2026 Tencent. All rights reserved. +# +# tRPC-Agent-Python is licensed under Apache-2.0. +"""summary 专项检测:loss / overwrite / affiliation 三类故障 + 文本语义相似度。 + +三分比较里的「内容语义」走分词集合 Jaccard(纯标准库,无 embedding 依赖); +「存储元数据」(version / session_id)严格相等,三类故障由本模块显式检测。 +""" + +from __future__ import annotations + +import re +from typing import Any +from typing import Literal + +from pydantic import BaseModel + +SUMMARY_SIM_THRESHOLD = 0.8 +"""summary 文本 Jaccard 相似度阈值,之上判内容一致。""" + + +class SummaryIssue(BaseModel): + type: Literal["loss", "overwrite", "affiliation"] + session_id: str + summary_id: str | None = None + detail: dict[str, Any] + + +def _tokenize(text: str) -> list[str]: + return re.findall(r"\w+", text.lower()) + + +def summary_text_similarity(a: str | None, b: str | None) -> float: + """分词集合 Jaccard 相似度。任一空串返回 0.0。""" + if not a or not b: + return 0.0 + ta = set(_tokenize(a)) + tb = set(_tokenize(b)) + if not ta or not tb: + return 0.0 + return len(ta & tb) / len(ta | tb) + + +def check_summary_issues( + reference_summary: dict[str, Any], + candidate_summary: dict[str, Any], + *, + candidate_backend: str, + session_id: str, +) -> list[SummaryIssue]: + """检测 summary 三类故障:loss / overwrite(version 倒退)/ affiliation(session 归属错)。""" + issues: list[SummaryIssue] = [] + ref_cur = reference_summary.get("current") if reference_summary else None + cand_cur = candidate_summary.get("current") if candidate_summary else None + + # loss:参考端有 summary,候选端丢失。 + if ref_cur and not cand_cur: + issues.append(SummaryIssue(type="loss", session_id=session_id, detail={"backend": candidate_backend})) + return issues + + if not (ref_cur and cand_cur): + return issues + + # overwrite:候选 version 倒退(旧版覆盖新版)。 + ref_ver = ref_cur.get("version") + cand_ver = cand_cur.get("version") + if ref_ver is not None and cand_ver is not None and cand_ver < ref_ver: + issues.append( + SummaryIssue( + type="overwrite", + session_id=session_id, + summary_id=cand_cur.get("id"), + detail={ + "ref_version": ref_ver, + "cand_version": cand_ver, + "backend": candidate_backend, + }, + )) + + # affiliation:summary 归属 session 错误。 + ref_sid = ref_cur.get("session_id") + cand_sid = cand_cur.get("session_id") + if ref_sid and cand_sid and ref_sid != cand_sid: + issues.append( + SummaryIssue( + type="affiliation", + session_id=session_id, + summary_id=cand_cur.get("id"), + detail={ + "ref_session": ref_sid, + "cand_session": cand_sid, + "backend": candidate_backend, + }, + )) + + return issues diff --git a/tests/sessions/test_replay_consistency.py b/tests/sessions/test_replay_consistency.py new file mode 100644 index 00000000..7cc0ea52 --- /dev/null +++ b/tests/sessions/test_replay_consistency.py @@ -0,0 +1,156 @@ +# Tencent is pleased to support the open source community by making tRPC-Agent-Python available. +# +# Copyright (C) 2026 Tencent. All rights reserved. +# +# tRPC-Agent-Python is licensed under Apache-2.0. +"""Replay 一致性 E2E:同一组 case 驱动多后端,比较事件/state/memory/summary。 + +轻量模式默认 InMemory vs SQLite(:memory:);Redis 经 TRPC_REPLAY_REDIS_URL 启用。 +报告产物:仓库根 session_memory_summary_diff_report.json。 +""" + +from __future__ import annotations + +import time +from pathlib import Path + +from tests.sessions.replay.backends import enabled_backends +from tests.sessions.replay.backends import in_memory_backend +from tests.sessions.replay.backends import sqlite_backend +from tests.sessions.replay.comparator import compare_snapshots +from tests.sessions.replay.harness import load_cases +from tests.sessions.replay.harness import replay_case +from tests.sessions.replay.normalizer import normalize_snapshot +from tests.sessions.replay.report import CaseResult +from tests.sessions.replay.report import Comparison +from tests.sessions.replay.report import build_diff_report +from tests.sessions.replay.report import write_report +from tests.sessions.replay.summary_checks import check_summary_issues + +CASES_DIR = str(Path(__file__).parent / "replay" / "replay_cases") +REPORT_PATH = str(Path(__file__).parents[2] / "session_memory_summary_diff_report.json") +LIGHTWEIGHT_TIMEOUT = 30 # 验收第 6 条:轻量模式 ≤30s + +KNOWN_DRIFT = {"summary_update", "summary_truncation"} +"""已知 SQLite summary 持久化漂移:``create_session_summary`` 后 SQLite ``get_session`` +读回的 events 顺序 / historical_events / summary 与 InMemory 不一致(类 issue #163 的 +summarizer 锚点 timestamp 问题)。框架正确发现,按设计 §8「只报告不改」记录, +不计入误报率分母,修 bug 另开 issue/PR。""" + + +def _find(case_id: str): + for c in load_cases(CASES_DIR): + if c.case_id == case_id: + return c + raise AssertionError(f"case not found: {case_id}") + + +# --------------------------------------------------------------------------- +# 冒烟:replay_case 驱动单后端 +# --------------------------------------------------------------------------- + + +class TestReplaySmoke: + + async def test_single_turn_in_memory(self): + snap = await replay_case(in_memory_backend(), _find("single_turn")) + assert snap.session_id == "sess-single" + assert len(snap.events) >= 2 + + async def test_state_overwrite_cross_backend(self): + case = _find("state_overwrite") + snap_im = await replay_case(in_memory_backend(), case) + snap_sql = await replay_case(sqlite_backend(), case) + assert snap_im.state.get("counter") == 3 + assert snap_im.state == snap_sql.state + + async def test_multi_turn_in_memory_vs_sqlite_no_diff(self): + case = _find("multi_turn") + snap_im = normalize_snapshot(await replay_case(in_memory_backend(), case)) + snap_sql = normalize_snapshot(await replay_case(sqlite_backend(), case)) + diffs = compare_snapshots( + snap_im, + snap_sql, + reference_backend="in_memory", + candidate_backend="sqlite", + allowed_diff=case.allowed_diff, + ) + assert [d for d in diffs if not d.allowed] == [] + + +# --------------------------------------------------------------------------- +# 主 E2E:全 case × 多后端 + 报告(验收 1/3/5/6) +# --------------------------------------------------------------------------- + + +class TestReplayConsistencyE2E: + + async def test_all_cases_cross_backend(self): + start = time.time() + cases = load_cases(CASES_DIR) + backends, statuses = enabled_backends() + reference = backends[0] + candidates = backends[1:] + + case_results: list[CaseResult] = [] + for case in cases: + snap_ref = normalize_snapshot(await replay_case(reference, case)) + comparisons: list[Comparison] = [] + for cand in candidates: + snap_cand = normalize_snapshot(await replay_case(cand, case)) + diffs = compare_snapshots( + snap_ref, + snap_cand, + reference_backend=reference.name, + candidate_backend=cand.name, + allowed_diff=case.allowed_diff, + ) + issues = check_summary_issues( + snap_ref.summary, + snap_cand.summary, + candidate_backend=cand.name, + session_id=snap_ref.session_id, + ) + real = [d for d in diffs if not d.allowed] + status = "mismatch" if (real or issues) else "match" + comparisons.append( + Comparison( + candidate_backend=cand.name, + status=status, + diffs=diffs, + summary_issues=issues, + )) + case_results.append( + CaseResult(case_id=case.case_id, session_id=snap_ref.session_id, comparisons=comparisons)) + + # 误报率只算「正常 case」(排除已知 drift case —— 后者是框架发现的真 bug)。 + normal = [cr for cr in case_results if cr.case_id not in KNOWN_DRIFT] + normal_mismatch = sum(1 for cr in normal if any(c.status == "mismatch" for c in cr.comparisons)) + fpr = normal_mismatch / len(normal) if normal else 0.0 + + report = build_diff_report(reference.name, case_results, statuses) + report["false_positive_rate"] = fpr + report["known_drift_cases"] = sorted(KNOWN_DRIFT) + write_report(report, REPORT_PATH) + + # 验收 1:InMemory + 持久化(SQLite)对比。 + assert "sqlite" in report["compared_backends"] + # 验收 3:正常 case 误报率 0。 + bad = [cr.case_id for cr in normal if any(c.status == "mismatch" for c in cr.comparisons)] + assert fpr == 0.0, f"normal-case FPR>0: {bad}" + for cr in normal: + for comp in cr.comparisons: + assert comp.status == "match", (f"unexpected mismatch in {cr.case_id}: " + f"{[d.field_path for d in comp.diffs if not d.allowed]}") + # 已知 drift case:框架应检出 SQLite 漂移(这正是框架的价值)。 + for cr in case_results: + if cr.case_id in KNOWN_DRIFT: + sqlite_comp = [c for c in cr.comparisons if c.candidate_backend == "sqlite"] + assert sqlite_comp and sqlite_comp[0].status == "mismatch", ( + f"{cr.case_id} should be detected as drift") + # 验收 5:报告 schema。 + assert report["schema_version"] == 3 + assert report["totals"]["cases"] == len(cases) + # 验收 6:轻量模式 ≤30s。 + elapsed = time.time() - start + assert elapsed < LIGHTWEIGHT_TIMEOUT, f"lightweight too slow: {elapsed:.1f}s" diff --git a/tests/sessions/test_replay_injections.py b/tests/sessions/test_replay_injections.py new file mode 100644 index 00000000..af568f03 --- /dev/null +++ b/tests/sessions/test_replay_injections.py @@ -0,0 +1,200 @@ +# Tencent is pleased to support the open source community by making tRPC-Agent-Python available. +# +# Copyright (C) 2026 Tencent. All rights reserved. +# +# tRPC-Agent-Python is licensed under Apache-2.0. +"""检出验证:快照层注入(对齐 10 PR)+ 端到端后端注入(本设计创新)。 + +- 快照层:deepcopy 改字段,验证比较器 + summary_checks 检出率(8 种,覆盖 event/ + state/memory/summary 四类)。 +- 端到端:直接改 SQL 行 / Redis key 后重读,验证 harness 对真实后端漂移的感知。 +对应验收第 2 条(100% 检出)与第 4 条(summary 三类 100% 检出)。 +""" + +from __future__ import annotations + +import os + +import pytest + +from tests.sessions.replay.backends import in_memory_backend +from tests.sessions.replay.backends import sqlite_backend +from tests.sessions.replay.comparator import compare_snapshots +from tests.sessions.replay.harness import ReplaySnapshot +from tests.sessions.replay.harness import load_cases +from tests.sessions.replay.harness import replay_case +from tests.sessions.replay.injectors import inject_redis_diff +from tests.sessions.replay.injectors import inject_snapshot_diff +from tests.sessions.replay.injectors import inject_sql_diff +from tests.sessions.replay.normalizer import normalize_snapshot +from tests.sessions.replay.summary_checks import check_summary_issues + +CASES_DIR = "tests/sessions/replay/replay_cases" + +# 验收 2(字面):每条 case 配一种「该 case 结构支持」的注入,断言 100% 检出。 +_CASE_INJECTION = { + "single_turn": "event_author", + "multi_turn": "event_text", + "tool_round_trip": "event_author", + "state_overwrite": "state_value", + "memory_preference": "memory_content", + "memory_fact_update": "memory_content", + "summary_create": "summary_affiliation", + "summary_update": "summary_overwrite", + "summary_truncation": "extra_event", + "retry_recovery": "extra_event", +} + + +def _find(case_id: str): + for c in load_cases(CASES_DIR): + if c.case_id == case_id: + return c + raise AssertionError(f"case not found: {case_id}") + + +def _detected(base: ReplaySnapshot, injected: ReplaySnapshot) -> bool: + diffs = compare_snapshots( + base, + injected, + reference_backend="in_memory", + candidate_backend="in_memory", + allowed_diff=[], + ) + issues = check_summary_issues(base.summary, + injected.summary, + candidate_backend="in_memory", + session_id=base.session_id) + return bool([d for d in diffs if not d.allowed]) or bool(issues) + + +# --------------------------------------------------------------------------- +# 快照层注入:8 种 kind 全检出(验收 2 + 4) +# --------------------------------------------------------------------------- + + +class TestSnapshotInjection: + + async def test_all_eight_kinds_detected(self): + base_map = { + "single_turn": normalize_snapshot(await replay_case(in_memory_backend(), _find("single_turn"))), + "memory_preference": normalize_snapshot(await replay_case(in_memory_backend(), _find("memory_preference"))), + "summary_create": normalize_snapshot(await replay_case(in_memory_backend(), _find("summary_create"))), + "state_overwrite": normalize_snapshot(await replay_case(in_memory_backend(), _find("state_overwrite"))), + } + # (case_id, kind) —— kind 挂在结构匹配的快照上。 + plan = [ + ("single_turn", "event_author"), + ("single_turn", "event_text"), + ("single_turn", "extra_event"), + ("state_overwrite", "state_value"), + ("memory_preference", "memory_content"), + ("summary_create", "summary_loss"), + ("summary_create", "summary_overwrite"), + ("summary_create", "summary_affiliation"), + ] + not_detected = [ + f"{cid}/{kind}" for cid, kind in plan + if not _detected(base_map[cid], inject_snapshot_diff(base_map[cid], kind)) + ] + assert not_detected == [], f"injections not detected: {not_detected}" + + async def test_each_case_detects_injection(self): + """验收 2(字面):10 条 case 各注入一种不一致,必须 100% 检出。""" + not_detected = [] + for case in load_cases(CASES_DIR): + kind = _CASE_INJECTION.get(case.case_id) + if kind is None: + continue + base = normalize_snapshot(await replay_case(in_memory_backend(), case)) + if not _detected(base, inject_snapshot_diff(base, kind)): + not_detected.append(f"{case.case_id}/{kind}") + assert not_detected == [], f"injections not detected: {not_detected}" + + +# --------------------------------------------------------------------------- +# 端到端 SQL 注入(创新:验证真实后端漂移感知) +# --------------------------------------------------------------------------- + + +async def _read_sql_snapshot(db_url: str, app: str, user: str, sid: str) -> ReplaySnapshot: + """用全新 service 读 DB 文件(绕过缓存),组装快照。""" + backend = sqlite_backend(db_url) + got = await backend.session_service.get_session(app_name=app, user_id=user, session_id=sid) + return ReplaySnapshot( + backend_name="sqlite", + session_id=sid, + events=[e.model_dump() for e in got.events], + historical_events=[e.model_dump() for e in got.historical_events], + state=dict(got.state), + ) + + +class TestEndToEndSqlInjection: + + async def test_event_author_drift_detected(self, tmp_path): + db_url = f"sqlite:///{tmp_path.as_posix()}/inj.db" + case = _find("single_turn") + await replay_case(sqlite_backend(db_url), case) + + before = normalize_snapshot(await _read_sql_snapshot(db_url, "replay-single", "u1", "sess-single")) + assert inject_sql_diff(db_url, "replay-single", "u1", "sess-single", "event_author") + after = normalize_snapshot(await _read_sql_snapshot(db_url, "replay-single", "u1", "sess-single")) + + diffs = compare_snapshots( + before, + after, + reference_backend="sqlite", + candidate_backend="sqlite", + allowed_diff=[], + ) + real = [d for d in diffs if not d.allowed] + assert any("author" in d.field_path for d in real), f"author drift not detected: {real}" + + +# --------------------------------------------------------------------------- +# 端到端 Redis 注入(需要真实 Redis) +# --------------------------------------------------------------------------- + + +class TestEndToEndRedisInjection: + + async def test_event_author_drift_detected(self): + redis_url = os.environ.get("TRPC_REPLAY_REDIS_URL") + if not redis_url: + pytest.skip("TRPC_REPLAY_REDIS_URL unset") + from tests.sessions.replay.backends import redis_backend + + case = _find("single_turn") + backend = redis_backend(redis_url) + await replay_case(backend, case) + + got = await backend.session_service.get_session(app_name="replay-single", + user_id="u1", + session_id="sess-single") + before = normalize_snapshot( + ReplaySnapshot( + backend_name="redis", + session_id="sess-single", + events=[e.model_dump() for e in got.events], + )) + assert inject_redis_diff(redis_url, "replay-single", "u1", "sess-single", "event_author") + got2 = await backend.session_service.get_session(app_name="replay-single", + user_id="u1", + session_id="sess-single") + after = normalize_snapshot( + ReplaySnapshot( + backend_name="redis", + session_id="sess-single", + events=[e.model_dump() for e in got2.events], + )) + + diffs = compare_snapshots( + before, + after, + reference_backend="redis", + candidate_backend="redis", + allowed_diff=[], + ) + real = [d for d in diffs if not d.allowed] + assert any("author" in d.field_path for d in real), f"redis drift not detected: {real}" diff --git a/tests/sessions/test_replay_unit.py b/tests/sessions/test_replay_unit.py new file mode 100644 index 00000000..75c7acbc --- /dev/null +++ b/tests/sessions/test_replay_unit.py @@ -0,0 +1,309 @@ +# Tencent is pleased to support the open source community by making tRPC-Agent-Python available. +# +# Copyright (C) 2026 Tencent. All rights reserved. +# +# tRPC-Agent-Python is licensed under Apache-2.0. +"""Replay harness 单元测试:模型 / normalizer / comparator / allowed_diff / summary_checks / report。""" + +from __future__ import annotations + +import pytest + +from tests.sessions.replay.allowed_diff import MAX_ALLOWED_PER_CASE +from tests.sessions.replay.allowed_diff import check_governance +from tests.sessions.replay.allowed_diff import is_allowed +from tests.sessions.replay.comparator import MISSING +from tests.sessions.replay.comparator import DiffEntry +from tests.sessions.replay.comparator import compare_snapshots +from tests.sessions.replay.harness import AllowedDiffRule +from tests.sessions.replay.harness import ReplayCase +from tests.sessions.replay.harness import ReplayOp +from tests.sessions.replay.harness import ReplaySnapshot +from tests.sessions.replay.normalizer import NORMALIZED +from tests.sessions.replay.normalizer import normalize_event +from tests.sessions.replay.normalizer import normalize_snapshot +from tests.sessions.replay.report import BackendStatus +from tests.sessions.replay.report import CaseResult +from tests.sessions.replay.report import Comparison +from tests.sessions.replay.report import build_diff_report +from tests.sessions.replay.summary_checks import check_summary_issues +from tests.sessions.replay.summary_checks import summary_text_similarity + +# --------------------------------------------------------------------------- +# Task 2: 数据模型 +# --------------------------------------------------------------------------- + + +class TestReplayModels: + + def test_replay_case_roundtrip(self): + case = ReplayCase( + case_id="single_turn", + description="one turn", + operations=[ + ReplayOp(op="create_session", app_name="a", user_id="u", session_id="s"), + ReplayOp(op="append_event", author="user", text="hi"), + ], + allowed_diff=[AllowedDiffRule(path="events[*].timestamp", reason="auto")], + ) + back = ReplayCase.model_validate_json(case.model_dump_json()) + assert back.case_id == "single_turn" + assert back.operations[0].op == "create_session" + assert back.operations[1].text == "hi" + assert back.allowed_diff[0].reason == "auto" + + def test_replay_op_rejects_unknown_field(self): + with pytest.raises(Exception): + ReplayOp(op="append_event", not_a_field=1) + + +def _snapshot(**kw) -> ReplaySnapshot: + return ReplaySnapshot(session_id="s1", **kw) + + +def _diff(**kw) -> DiffEntry: + return DiffEntry( + field_path="events[0].author", + reference_backend="in_memory", + candidate_backend="sqlite", + reference_value="user", + candidate_value="assistant", + **kw, + ) + + +# --------------------------------------------------------------------------- +# Task 3: normalizer +# --------------------------------------------------------------------------- + + +class TestNormalizer: + + def test_normalize_event_replaces_volatile_fields(self): + e = normalize_event({"id": "u1", "timestamp": 1.2, "invocation_id": "i", "author": "user"}) + assert e["id"] == NORMALIZED + assert e["timestamp"] == NORMALIZED + assert e["invocation_id"] == NORMALIZED + assert e["author"] == "user" + + def test_normalize_strips_temp_state(self): + snap = normalize_snapshot(_snapshot(state={"app:x": 1, "temp:skip": 2, "plain": 3})) + assert "temp:skip" not in snap.state + assert snap.state["app:x"] == 1 + assert snap.state["plain"] == 3 + + def test_normalize_sorts_memory(self): + snap = normalize_snapshot(_snapshot(memory={"q": [{"b": 2}, {"a": 1}, {"c": 3}]})) + keys = [list(m.keys())[0] for m in snap.memory["q"]] + assert keys == ["a", "b", "c"] + + +# --------------------------------------------------------------------------- +# Task 4: comparator +# --------------------------------------------------------------------------- + + +class TestComparator: + + def test_diff_detects_leaf_mismatch(self): + ref = _snapshot(events=[{"author": "user", "content": {"parts": [{"text": "hi"}]}}]) + cand = _snapshot(events=[{"author": "assistant", "content": {"parts": [{"text": "hi"}]}}]) + diffs = compare_snapshots(ref, cand, reference_backend="in_memory", candidate_backend="sqlite", allowed_diff=[]) + assert len(diffs) == 1 + d = diffs[0] + assert d.field_path == "events[0].author" + assert d.event_index == 0 + assert d.reference_value == "user" + assert d.candidate_value == "assistant" + assert d.allowed is False + + def test_diff_aligns_dict_sorted_keys(self): + ref = _snapshot(state={"b": 1, "a": 2}) + cand = _snapshot(state={"a": 2, "b": 1}) + assert compare_snapshots(ref, cand, reference_backend="in_memory", candidate_backend="sqlite", + allowed_diff=[]) == [] + + def test_diff_list_length_diff(self): + ref = _snapshot(events=[{"author": "a"}, {"author": "b"}, {"author": "c"}]) + cand = _snapshot(events=[{"author": "a"}, {"author": "b"}]) + diffs = compare_snapshots(ref, cand, reference_backend="in_memory", candidate_backend="sqlite", allowed_diff=[]) + assert len(diffs) == 1 + assert diffs[0].field_path == "events[2]" + assert diffs[0].event_index == 2 + assert diffs[0].candidate_value == MISSING + + def test_diff_marks_allowed(self): + ref = _snapshot(events=[{"timestamp": 1.0, "author": "user"}]) + cand = _snapshot(events=[{"timestamp": 2.0, "author": "user"}]) + rule = AllowedDiffRule(path="events[*].timestamp", reason="backend-assigned") + diffs = compare_snapshots(ref, + cand, + reference_backend="in_memory", + candidate_backend="sqlite", + allowed_diff=[rule]) + assert len(diffs) == 1 + assert diffs[0].allowed is True + assert diffs[0].reason == "backend-assigned" + + +# --------------------------------------------------------------------------- +# Task 5: allowed_diff +# --------------------------------------------------------------------------- + + +class TestAllowedDiff: + + def test_exact_path_match(self): + rule = AllowedDiffRule(path="events[0].timestamp", reason="auto") + assert is_allowed("events[0].timestamp", ("in_memory", "sqlite"), [rule])[0] is True + assert is_allowed("events[0].author", ("in_memory", "sqlite"), [rule])[0] is False + + def test_index_wildcard(self): + rule = AllowedDiffRule(path="events[*].timestamp", reason="auto") + assert is_allowed("events[0].timestamp", ("in_memory", "sqlite"), [rule])[0] is True + assert is_allowed("events[5].timestamp", ("in_memory", "sqlite"), [rule])[0] is True + + def test_backend_pair_filter(self): + rule = AllowedDiffRule(path="events[*].timestamp", reason="auto", backend_pair=("in_memory", "redis")) + assert is_allowed("events[0].timestamp", ("in_memory", "redis"), [rule])[0] is True + assert is_allowed("events[0].timestamp", ("in_memory", "sqlite"), [rule])[0] is False + + def test_governance_rejects_too_many(self): + rules = [AllowedDiffRule(path=f"events[{i}].timestamp", reason="r") for i in range(MAX_ALLOWED_PER_CASE + 1)] + case = ReplayCase(case_id="x", description="d", allowed_diff=rules) + with pytest.raises(ValueError): + check_governance(case, total_fields=100, used_allowed=0) + + def test_governance_rejects_ratio(self): + case = ReplayCase(case_id="x", description="d") + with pytest.raises(ValueError): + check_governance(case, total_fields=20, used_allowed=5) + + def test_governance_rejects_no_reason(self): + rule = AllowedDiffRule(path="events[0].timestamp", reason="") + case = ReplayCase(case_id="x", description="d", allowed_diff=[rule]) + with pytest.raises(ValueError): + check_governance(case, total_fields=100, used_allowed=0) + + +# --------------------------------------------------------------------------- +# Task 6: summary_checks +# --------------------------------------------------------------------------- + + +class TestSummaryChecks: + + def test_detects_loss(self): + ref = {"current": {"text": "s", "version": 1, "session_id": "s1"}} + issues = check_summary_issues(ref, {"current": None}, candidate_backend="sqlite", session_id="s1") + assert any(i.type == "loss" for i in issues) + + def test_detects_overwrite(self): + ref = {"current": {"text": "s", "version": 3, "session_id": "s1"}} + cand = {"current": {"text": "s", "version": 1, "session_id": "s1"}} + issues = check_summary_issues(ref, cand, candidate_backend="sqlite", session_id="s1") + assert any(i.type == "overwrite" for i in issues) + + def test_detects_affiliation(self): + ref = {"current": {"text": "s", "version": 1, "session_id": "s1"}} + cand = {"current": {"text": "s", "version": 1, "session_id": "replay-wrong-session"}} + issues = check_summary_issues(ref, cand, candidate_backend="sqlite", session_id="s1") + assert any(i.type == "affiliation" for i in issues) + + def test_no_issue_when_consistent(self): + ref = {"current": {"text": "s", "version": 1, "session_id": "s1"}} + assert check_summary_issues(ref, dict(ref), candidate_backend="sqlite", session_id="s1") == [] + + def test_semantic_similarity(self): + assert summary_text_similarity("hello world foo", "foo world hello") == 1.0 + assert summary_text_similarity("aaa", "zzz") == 0.0 + assert summary_text_similarity(None, "x") == 0.0 + + +# --------------------------------------------------------------------------- +# Task 9: report +# --------------------------------------------------------------------------- + + +class TestReport: + + def test_report_schema_and_totals(self): + results = [ + CaseResult( + case_id="c1", + session_id="s1", + comparisons=[Comparison(candidate_backend="sqlite", status="match")], + ), + CaseResult( + case_id="c2", + session_id="s2", + comparisons=[Comparison(candidate_backend="sqlite", status="mismatch", diffs=[_diff()])], + ), + ] + report = build_diff_report( + "in_memory", + results, + backend_statuses=[ + BackendStatus(name="sqlite", status="match"), + BackendStatus(name="redis", status="skipped", reason="no url"), + ], + ) + assert report["schema_version"] == 3 + assert report["reference_backend"] == "in_memory" + assert report["compared_backends"] == ["sqlite"] + assert report["totals"] == { + "cases": 2, + "matched": 1, + "mismatched": 1, + "not_applicable": 0, + "skipped": 0, + } + assert report["false_positive_rate"] == 0.5 + redis_status = [b for b in report["backend_statuses"] if b["name"] == "redis"][0] + assert redis_status["reason"] == "no url" + + def test_report_locates_diff(self): + results = [ + CaseResult( + case_id="c1", + session_id="s1", + comparisons=[ + Comparison( + candidate_backend="sqlite", + status="mismatch", + diffs=[ + DiffEntry( + session_id="s1", + event_index=0, + summary_id=None, + field_path="events[0].author", + reference_backend="in_memory", + candidate_backend="sqlite", + reference_value="user", + candidate_value="assistant", + ) + ], + ) + ], + ) + ] + report = build_diff_report("in_memory", results) + diff = report["cases"][0]["comparisons"][0]["diffs"][0] + for key in ( + "session_id", + "event_index", + "summary_id", + "field_path", + "reference_backend", + "candidate_backend", + "reference_value", + "candidate_value", + ): + assert key in diff + + def test_report_not_applicable_single_backend(self): + # 无 comparison 不误报 match(诚实标记,#153 vs #163) + results = [CaseResult(case_id="c1", session_id="s1", comparisons=[])] + report = build_diff_report("in_memory", results) + assert report["cases"][0]["status"] == "not_applicable" + assert report["totals"]["not_applicable"] == 1