test: add session memory summary replay framework

.gitignore

@@ -27,3 +27,6 @@ pyrightconfig.json
 
 # spec-workflow tool artifacts
 .spec-workflow
+
+# Replay harness runtime reports / 回放框架运行时报告
+/tests/sessions/artifacts/

tests/memory/test_sql_memory_service.py

@@ -278,6 +278,38 @@ async def test_store_skips_events_without_content(self):
         svc._sql_storage.add.assert_not_called()
         svc._sql_storage.commit.assert_not_called()
 
+    async def test_store_replaces_stale_session_memory(self, tmp_path):
+        """Verify a new SQL memory snapshot removes stale event rows.
+
+        验证再次保存 Session 完整快照时，SQL Memory 会删除已不在快照中的旧事件，
+        同时保留当前事件且不产生重复记录。
+        """
+        svc = SqlMemoryService(
+            db_url=f"sqlite:///{tmp_path / 'memory.sqlite3'}",
+            memory_service_config=_make_config_no_ttl(),
+        )
+        await svc._sql_storage.create_sql_engine()
+        current = _make_event("current marker", event_id="current")
+        original = _make_session(events=[
+            _make_event("obsolete marker", event_id="obsolete"),
+            current,
+        ])
+        replacement = _make_session(events=[current])
+
+        # Store the original snapshot, then replace it with one that omits obsolete.
+        # 先保存原始快照，再用不含 obsolete 事件的新快照替换。
+        await svc.store_session(original)
+        await svc.store_session(replacement)
+
+        # Search through the public API to verify stale deletion and retained uniqueness.
+        # 通过公开查询接口验证旧行已删除，保留事件仍且仅有一条。
+        obsolete = await svc.search_memory(original.save_key, "obsolete")
+        current_result = await svc.search_memory(original.save_key, "current")
+        assert obsolete.memories == []
+        assert len(current_result.memories) == 1
+        await svc.close()
+
+
 # ---------------------------------------------------------------------------
 # SqlMemoryService — search_memory
 # ---------------------------------------------------------------------------

tests/sessions/replay/IMPLEMENTATION_PLAN.md

@@ -10,7 +10,7 @@
 
 ## Global Constraints
 
-- **PR 干净**:全部落在 `tests/sessions/` 下 —— 测试代码 + 本计划/设计文档(置于 `tests/sessions/replay/`)+ 报告产物 `tests/sessions/session_memory_summary_diff_report.json`。**不改 `trpc_agent_sdk/` 生产代码**;发现的 SDK bug 只在报告/文档记录。
+- **PR 干净**:全部落在 `tests/sessions/` 下 —— 测试代码 + 本计划/设计文档(置于 `tests/sessions/replay/`)+ 报告产物 `tests/sessions/artifacts/session_memory_summary_diff_report.json`。若差异定位为 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。
@@ -236,7 +236,7 @@
 **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`,生成 `tests/sessions/session_memory_summary_diff_report.json`。
+- [ ] **Step 1:** 写 `test_replay_consistency_lightweight`:跑全部 10 case × `enabled_backends()`(轻量=in_memory+sqlite),断言正常 case 全 `match`、`false_positive_rate==0.0`,生成 `tests/sessions/artifacts/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、报告产物生成且可定位。

tests/sessions/replay/README.md

@@ -58,9 +58,9 @@ TRPC_REPLAY_REDIS_URL=redis://localhost:6379/0 PYTHONUTF8=1 pytest tests/session
 - `python-magic` 在 Windows 上会致 SDK 导入崩溃,需改用 `python-magic-bin`(venv 内替换)。
 
 ### 7. 报告产物位置
-- 运行 `test_replay_consistency.py` 会生成 / 覆盖 `tests/sessions/session_memory_summary_diff_report.json`
+- 运行 `test_replay_consistency.py` 会生成 / 覆盖 `tests/sessions/artifacts/session_memory_summary_diff_report.json`
   (schema_version=3,每条 diff 内联 `session_id` / `event_index` / `summary_id` / `field_path` + 双后端值,
-  不嵌全量 snapshot)。该文件作为测试基线产物**已纳入版本管理**,review/排障时可直接查看。
+  不嵌全量 snapshot)。该文件是可重复生成的运行时产物，目录已加入 `.gitignore`，需要时通过测试或 CLI 重新生成。
 
 ## 目录结构
 
@@ -83,5 +83,5 @@ tests/sessions/
 ├── test_replay_consistency.py    # 主 E2E(正向:一致性 + FPR)
 ├── test_replay_injections.py     # 注入检出(负向:错误场景)
 ├── test_replay_unit.py           # 模块单测
-└── session_memory_summary_diff_report.json   # 报告产物(运行时生成)
+└── artifacts/session_memory_summary_diff_report.json   # 报告产物(运行时生成，Git 忽略)
 ```

tests/sessions/replay/backends.py

@@ -28,6 +28,7 @@
 from trpc_agent_sdk.sessions._summarizer_manager import SummarizerSessionManager
 
 from .harness import ReplayBackend
+from .redis_support import redis_unavailable_reason
 from .report import BackendStatus
 
 
@@ -75,7 +76,10 @@ def sqlite_backend(db_url: str = "sqlite:///:memory:") -> ReplayBackend:
 
 
 def redis_backend(url: str) -> ReplayBackend:
-    svc = RedisSessionService(db_url=url, summarizer_manager=_manager(), session_config=_session_config(), is_async=True)
+    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)
 
@@ -98,11 +102,19 @@ def enabled_backends(tmp_path: Optional[str] = None, ) -> tuple[list[ReplayBacke
 
     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)))
+        # Redis clients connect lazily, so constructing the backend cannot
+        # prove the configured endpoint is reachable. Probe it explicitly
+        # before adding Redis to the replay candidates.
+        # Redis 客户端采用延迟连接；构造成功不代表服务可达，因此先执行 PING。
+        unavailable_reason = redis_unavailable_reason(redis_url)
+        if unavailable_reason:
+            statuses.append(BackendStatus(name="redis", status="skipped", reason=unavailable_reason))
+        else:
+            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"))
 

tests/sessions/replay/normalizer.py

@@ -14,6 +14,8 @@
 import json
 from typing import Any
 
+from trpc_agent_sdk.sessions import SESSION_SUMMARY_METADATA_KEY
+
 from .harness import NORMALIZED
 from .harness import ReplaySnapshot
 
@@ -27,6 +29,27 @@ def normalize_event(event: dict[str, Any]) -> dict[str, Any]:
     for key in VOLATILE_KEYS:
         if key in out:
             out[key] = NORMALIZED
+
+    # A structured Summary event repeats its generated ID and real update
+    # clock in metadata. Normalize only those volatile representations while
+    # keeping ownership, text, version and replacement presence strict.
+    # 结构化 Summary Event 会在元数据中重复自动 ID 和真实更新时间；这里只
+    # 归一化这些非业务表示，归属、正文、版本及覆盖关系是否存在仍严格比较。
+    custom_metadata = out.get("custom_metadata")
+    if isinstance(custom_metadata, dict):
+        custom_metadata = dict(custom_metadata)
+        summary_metadata = custom_metadata.get(SESSION_SUMMARY_METADATA_KEY)
+        if isinstance(summary_metadata, dict):
+            summary_metadata = dict(summary_metadata)
+            if isinstance(summary_metadata.get("summary_id"), str):
+                summary_metadata["summary_id"] = NORMALIZED
+            if isinstance(summary_metadata.get("summary_timestamp"), (int, float)):
+                summary_metadata["summary_timestamp"] = NORMALIZED
+            if isinstance(summary_metadata.get("replaces_summary_id"), str):
+                summary_metadata["replaces_summary_id"] = NORMALIZED
+            custom_metadata[SESSION_SUMMARY_METADATA_KEY] = summary_metadata
+            out["custom_metadata"] = custom_metadata
+
     # long_running_tool_ids: InMemory=None vs SQL=set() 的良性序列化差异,统一空值;
     # 一方有值一方空的真丢失仍会被检出。
     lr = out.get("long_running_tool_ids")

tests/sessions/replay/redis_support.py

@@ -0,0 +1,64 @@
+# 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.
+"""Redis availability helpers shared by replay integration tests.
+
+回放集成测试共用的 Redis 可用性检查工具。
+"""
+
+from __future__ import annotations
+
+import os
+from typing import Optional
+
+import pytest
+import redis
+
+
+def redis_unavailable_reason(redis_url: str) -> Optional[str]:
+    """Return an actionable reason when Redis cannot answer ``PING``.
+
+    当 Redis 无法响应 ``PING`` 时返回可操作的原因；可用时返回 ``None``。
+    The URL is deliberately omitted from the message because it may contain
+    credentials.
+
+    错误消息不回显 URL，避免泄露其中可能包含的认证信息。
+    """
+    client = None
+    try:
+        # Keep the probe short so an invalid opt-in URL does not stall the
+        # lightweight test suite.
+        # 使用短超时，避免错误的可选 URL 阻塞轻量测试流程。
+        client = redis.Redis.from_url(
+            redis_url,
+            socket_connect_timeout=0.5,
+            socket_timeout=0.5,
+        )
+        if client.ping():
+            return None
+        return "configured Redis did not return PONG"
+    except Exception as exc:  # noqa: BLE001 - availability probe must report every client failure
+        return f"configured Redis is unavailable: {type(exc).__name__}: {exc}"
+    finally:
+        if client is not None:
+            try:
+                client.close()
+            except Exception:  # noqa: BLE001 - cleanup must not hide the probe result
+                pass
+
+
+def require_replay_redis() -> str:
+    """Return the configured reachable Redis URL or skip the integration test.
+
+    返回已配置且可连接的 Redis URL；否则跳过可选集成测试。
+    """
+    redis_url = os.getenv("TRPC_REPLAY_REDIS_URL")
+    if not redis_url:
+        pytest.skip("TRPC_REPLAY_REDIS_URL is not configured")
+
+    reason = redis_unavailable_reason(redis_url)
+    if reason:
+        pytest.skip(f"{reason}; start Redis or unset TRPC_REPLAY_REDIS_URL")
+    return redis_url

tests/sessions/replay/report.py

@@ -13,6 +13,7 @@
 from __future__ import annotations
 
 import json
+from pathlib import Path
 from typing import Any
 from typing import Literal
 
@@ -140,6 +141,10 @@ def build_diff_report(
 
 
 def write_report(report: dict[str, Any], path: str) -> None:
-    """把报告写入 JSON 文件(默认 tests/sessions/session_memory_summary_diff_report.json,路径由调用方传入)。"""
-    with open(path, "w", encoding="utf-8") as f:
+    """把报告写入 JSON 文件(默认 tests/sessions/artifacts/session_memory_summary_diff_report.json,路径由调用方传入)。"""
+    # The report directory is intentionally generated and may not exist in a clean checkout.
+    # 报告目录属于运行时产物，干净检出中可能不存在，因此写入前主动创建。
+    output_path = Path(path)
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+    with output_path.open("w", encoding="utf-8") as f:
         json.dump(report, f, ensure_ascii=False, indent=2)

tests/sessions/replay/session-memory-replay-consistency-design.md

@@ -103,7 +103,7 @@ 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 故障
-tests/sessions/session_memory_summary_diff_report.json   # 报告产物(运行时生成)
+tests/sessions/artifacts/session_memory_summary_diff_report.json   # 报告产物(运行时生成)
 ```
 
 ---
@@ -408,7 +408,7 @@ Redis/MySQL 不可用时 `pytest.skip`(满足 issue"不要求本地装真 Redis/
 
 1. `tests/sessions/test_replay_consistency.py` + `tests/sessions/replay/` harness 包
 2. `tests/sessions/replay/replay_cases/*.jsonl`(10 条)
-3. `tests/sessions/session_memory_summary_diff_report.json`(运行时生成)
+3. `tests/sessions/artifacts/session_memory_summary_diff_report.json`(运行时生成)
 4. 150–300 字设计说明(本文档 §10 + 测试包 `__init__.py` docstring)
 5. 本设计文档 + 实施计划(均置于 `tests/sessions/replay/`,随测试代码同置)