feat: Session/Memory 多后端回放一致性测试框架 (issue #89)

.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

tests/sessions/replay/README.md

@@ -0,0 +1,87 @@
+# Session / Memory / Summary 多后端回放一致性测试
+
+用同一组标准化 Agent 轨迹驱动 InMemory / SQLite / Redis 三个后端,经四段管线
+`load → replay_case → 后端中立快照 → compare → report` 比较**事件 / state / memory / summary** 的一致性。
+既是测试工具,也是后端实现质量的基准。完整设计见同目录
+[`2026-07-13-session-memory-replay-consistency-design.md`](2026-07-13-session-memory-replay-consistency-design.md),
+实施步骤见 [`IMPLEMENTATION_PLAN.md`](IMPLEMENTATION_PLAN.md)。
+
+## 快速运行
+
+```bash
+# 轻量模式(默认,无需任何外部依赖,≤30s):InMemory vs SQLite
+PYTHONUTF8=1 pytest tests/sessions/test_replay_consistency.py \
+                    tests/sessions/test_replay_injections.py \
+                    tests/sessions/test_replay_unit.py -v
+
+# 启用 Redis 集成模式(需要本机/CI 有可达 Redis)
+TRPC_REPLAY_REDIS_URL=redis://localhost:6379/0 PYTHONUTF8=1 pytest tests/sessions/ -v
+```
+
+## 测试需要注意的地方
+
+### 1. 完全确定性,无需 API Key / 网络
+- **无 LLM、无真实网络**:summary 用 `DeterministicSummarizer`(覆写 `_compress_session_to_summary`,
+  返回确定性文本),时间戳 / 自动 id / invocation_id 经占位符归一化。**不需要任何模型 API Key**,CI 可离线跑。
+- 因此结果可复现:同一组 case 在同一后端组合下,差异报告逐字节稳定。
+
+### 2. 后端启用由环境变量门控(不可用自动 skip,不会 fail)
+| 环境变量 | 作用 | 默认 |
+|---|---|---|
+| `TRPC_REPLAY_SQL_URL` | 自定义 SQL 连接串 | `sqlite:///:memory:`(轻量) |
+| `TRPC_REPLAY_REDIS_URL` | 设置即启用 Redis 集成模式 | **未设置 → Redis 用例 `pytest.skip`** |
+
+- **轻量模式默认就跑**(InMemory + SQLite `:memory:`),**不要求**本地装 Redis/MySQL。
+- Redis 不可用时测试**自动 skip**,不会报错失败 —— 不要为了让 CI 变绿而注释掉 Redis 用例。
+
+### 3. 正向(正确场景)+ 负向(错误场景)双向验证
+同一组 10 条标准化 case 同时承担两个方向的验证,**不要只看绿灯**:
+- **正向**(`test_replay_consistency.py`):case **不注入** → 各后端应 100% `match`,
+  断言 `false_positive_rate == 0.0`(不误报)。
+- **负向**(`test_replay_injections.py`):case 经 `injectors.py` **程序化注入不一致**
+  (快照层 8 种 kind + 端到端改 SQL 行 / Redis key)→ 必须 100% 检出(不漏报)。
+
+### 4. 已知 drift 是「框架价值」,不是测试 bug
+- `summary_update` / `summary_truncation` 两个 case **会**检出 SQLite summary 持久化漂移
+  (`create_session_summary` 后 SQLite `get_session` 读回的 events 顺序 / historical_events / summary
+  与 InMemory 不一致,类 issue #163 的 summarizer 锚点问题)。
+- 这是框架**正确发现**的 SDK 真问题,以 `KNOWN_DRIFT` 标记、**不计入误报率分母**,
+  遵循设计 §8「只报告不改」—— 修 bug 另开 issue/PR,**不要在本测试里用 `allowed_diff` 把它掩盖掉**。
+
+### 5. `allowed_diff` 治理:严禁滥用
+- 每条 `allowed_diff` 必须带 `reason`,且有**条数上限(8/ case)与占比上限(10%)**。
+- 用 JSONPath **精确匹配**(`events[0].timestamp`),禁止 `*.id` 这类过宽规则(会误放业务 id)。
+- 任何「为了让用例过」而塞进 `allowed_diff` 的真不一致,都会被 governance 测试拒绝。
+
+### 6. Windows 运行注意
+- 必须 `PYTHONUTF8=1`(否则中文 case / 报告 JSON 序列化乱码)。
+- `python-magic` 在 Windows 上会致 SDK 导入崩溃,需改用 `python-magic-bin`(venv 内替换)。
+
+### 7. 报告产物位置
+- 运行 `test_replay_consistency.py` 会生成 / 覆盖 `tests/sessions/session_memory_summary_diff_report.json`
+  (schema_version=3,每条 diff 内联 `session_id` / `event_index` / `summary_id` / `field_path` + 双后端值,
+  不嵌全量 snapshot)。该文件作为测试基线产物**已纳入版本管理**,review/排障时可直接查看。
+
+## 目录结构
+
+```
+tests/sessions/replay/
+├── README.md                                          # 本文件
+├── 2026-07-13-session-memory-replay-consistency-design.md     # 设计文档
+├── IMPLEMENTATION_PLAN.md          # 实施计划(无日期前缀)
+├── __init__.py            # 包入口 + 设计说明
+├── harness.py             # 数据模型 + replay_case() 驱动
+├── normalizer.py          # 占位符归一化
+├── comparator.py          # 递归比较 + DiffEntry(内联定位)
+├── allowed_diff.py        # JSONPath 精确匹配 + 覆盖率治理
+├── summary_checks.py      # summary 三类专项(loss/overwrite/affiliation)
+├── injectors.py           # 快照层 + 端到端后端注入(错误场景)
+├── report.py              # schema_version=3 差异报告
+├── backends.py            # 三后端实例化 + env 门控 + 确定性 summarizer
+└── replay_cases/cases.jsonl   # 10 条标准化轨迹
+tests/sessions/
+├── test_replay_consistency.py    # 主 E2E(正向:一致性 + FPR)
+├── test_replay_injections.py     # 注入检出(负向:错误场景)
+├── test_replay_unit.py           # 模块单测
+└── session_memory_summary_diff_report.json   # 报告产物(运行时生成)
+```

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] = []

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}")

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(), is_async=True)
+    mem = RedisMemoryService(db_url=url, enabled=True, is_async=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