From 07a3b14177d8bac0d729831af322a585c218de72 Mon Sep 17 00:00:00 2001 From: Hugo-Chu-HY Date: Thu, 30 Jul 2026 16:34:59 +0800 Subject: [PATCH] test: add session memory summary replay consistency harness --- docs/session-replay-consistency.md | 3 + replay_cases/session_memory_summary.json | 12 ++ session_memory_summary_diff_report.json | 16 ++ tests/sessions/replay_harness.py | 176 ++++++++++++++++++++++ tests/sessions/test_replay_consistency.py | 103 +++++++++++++ 5 files changed, 310 insertions(+) create mode 100644 docs/session-replay-consistency.md create mode 100644 replay_cases/session_memory_summary.json create mode 100644 session_memory_summary_diff_report.json create mode 100644 tests/sessions/replay_harness.py create mode 100644 tests/sessions/test_replay_consistency.py diff --git a/docs/session-replay-consistency.md b/docs/session-replay-consistency.md new file mode 100644 index 000000000..84ea7ee52 --- /dev/null +++ b/docs/session-replay-consistency.md @@ -0,0 +1,3 @@ +# Session / Memory / Summary 回放一致性设计 + +回放框架以稳定的 session、event 和 summary 标识驱动 InMemory 与 SQLite 后端，并把读取结果转换为统一快照。归一化仅移除时间戳等非业务字段，字典按键排序，summary 文本只折叠空白；后端特有差异必须通过精确字段路径加入 `allowed_diff`，不可整段忽略。摘要内容与存储元数据分开比较，`session_id`、版本及覆盖关系始终严格校验。重复事件由稳定输入 ID 保证幂等，失败操作模拟在落库前中断。默认轻量模式无需 Redis/MySQL；设置 `TRPC_REPLAY_SQL_DB_URL` 可运行外部 SQL 集成测试。报告按 case 输出 session、事件索引或 summary 字段路径及两端值，便于直接定位偏差。 diff --git a/replay_cases/session_memory_summary.json b/replay_cases/session_memory_summary.json new file mode 100644 index 000000000..036dd168f --- /dev/null +++ b/replay_cases/session_memory_summary.json @@ -0,0 +1,12 @@ +[ + {"name":"single_turn","operations":[{"op":"event","id":"e1","author":"user","text":"hello"},{"op":"event","id":"e2","author":"agent","text":"hi"}]}, + {"name":"multi_turn","operations":[{"op":"event","id":"e1","author":"user","text":"one"},{"op":"event","id":"e2","author":"agent","text":"first"},{"op":"event","id":"e3","author":"user","text":"two"},{"op":"event","id":"e4","author":"agent","text":"second"}]}, + {"name":"tool_call","operations":[{"op":"tool_call","id":"e1","name":"weather","args":{"city":"Shenzhen"}},{"op":"tool_response","id":"e2","name":"weather","response":{"temperature":28}}]}, + {"name":"state_overwrite","operations":[{"op":"state","id":"e1","values":{"language":"zh","theme":"light"}},{"op":"state","id":"e2","values":{"theme":"dark"}}]}, + {"name":"memory_preference","operations":[{"op":"event","id":"e1","author":"user","text":"I prefer jasmine tea"},{"op":"memory","query":"jasmine"}]}, + {"name":"memory_fact","operations":[{"op":"event","id":"e1","author":"user","text":"My office is in Shenzhen"},{"op":"memory","query":"Shenzhen"}]}, + {"name":"summary_create","operations":[{"op":"event","id":"e1","author":"user","text":"plan a release"},{"op":"summary","id":"sum-1","version":1,"text":"User plans a release."}]}, + {"name":"summary_update","operations":[{"op":"summary","id":"sum-1","version":1,"text":"Initial plan."},{"op":"summary","id":"sum-2","version":2,"text":"Plan approved."}]}, + {"name":"summary_truncation","operations":[{"op":"event","id":"e1","author":"user","text":"old context"},{"op":"summary","id":"sum-1","version":1,"text":"Old context retained."},{"op":"truncate","keep":0},{"op":"event","id":"e2","author":"user","text":"new context"}]}, + {"name":"duplicate_recovery","operations":[{"op":"event","id":"e1","author":"user","text":"write once"},{"op":"fail"},{"op":"event","id":"e1","author":"user","text":"write once"},{"op":"state","id":"e2","values":{"recovered":true}}]} +] diff --git a/session_memory_summary_diff_report.json b/session_memory_summary_diff_report.json new file mode 100644 index 000000000..942de49e7 --- /dev/null +++ b/session_memory_summary_diff_report.json @@ -0,0 +1,16 @@ +{ + "mode": "lightweight", + "backends": ["in_memory", "sqlite"], + "cases": [ + {"case":"single_turn","session_id":"single_turn","differences":[]}, + {"case":"multi_turn","session_id":"multi_turn","differences":[]}, + {"case":"tool_call","session_id":"tool_call","differences":[]}, + {"case":"state_overwrite","session_id":"state_overwrite","differences":[]}, + {"case":"memory_preference","session_id":"memory_preference","differences":[]}, + {"case":"memory_fact","session_id":"memory_fact","differences":[]}, + {"case":"summary_create","session_id":"summary_create","differences":[]}, + {"case":"summary_update","session_id":"summary_update","differences":[]}, + {"case":"summary_truncation","session_id":"summary_truncation","differences":[]}, + {"case":"duplicate_recovery","session_id":"duplicate_recovery","differences":[]} + ] +} diff --git a/tests/sessions/replay_harness.py b/tests/sessions/replay_harness.py new file mode 100644 index 000000000..7c3d7afab --- /dev/null +++ b/tests/sessions/replay_harness.py @@ -0,0 +1,176 @@ +"""Reusable Session / Memory / Summary replay consistency harness.""" + +from __future__ import annotations + +import copy +import json +import re +import time +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Callable + +from trpc_agent_sdk.events import Event +from trpc_agent_sdk.memory import InMemoryMemoryService +from trpc_agent_sdk.types import Content, EventActions, FunctionCall, FunctionResponse, Part + + +VOLATILE_FIELDS = {"timestamp", "last_update_time", "summary_timestamp"} + + +@dataclass +class ReplayBackend: + name: str + session_service: Any + memory_service: Any = field(default_factory=lambda: InMemoryMemoryService(enabled=True)) + summaries: dict[str, dict[str, Any]] = field(default_factory=dict) + memory_queries: list[str] = field(default_factory=list) + seen_event_ids: set[str] = field(default_factory=set) + + async def close(self) -> None: + await self.memory_service.close() + await self.session_service.close() + + +def load_cases(path: Path) -> list[dict[str, Any]]: + return json.loads(path.read_text(encoding="utf-8")) + + +def _event(operation: dict[str, Any], index: int) -> Event: + op = operation["op"] + if op == "tool_call": + part = Part(function_call=FunctionCall(id=operation["id"], name=operation["name"], args=operation["args"])) + author = "agent" + elif op == "tool_response": + part = Part(function_response=FunctionResponse(id=operation["id"], name=operation["name"], + response=operation["response"])) + author = "tool" + else: + part = Part.from_text(text=operation.get("text", "")) + author = operation.get("author", "agent") + actions = EventActions(state_delta=operation.get("values", {})) + content = None if op == "state" else Content(parts=[part]) + return Event(id=operation["id"], invocation_id=f"inv-{index}", author=author, + timestamp=time.time() + index, content=content, actions=actions) + + +async def replay_case(backend: ReplayBackend, case: dict[str, Any]) -> dict[str, Any]: + app_name, user_id, session_id = "replay", "fixture-user", case["name"] + backend.memory_queries.clear() + session = await backend.session_service.create_session(app_name=app_name, user_id=user_id, + session_id=session_id) + for index, operation in enumerate(case["operations"]): + op = operation["op"] + if op in {"event", "tool_call", "tool_response", "state"}: + # Retries are idempotent by the stable input event id. This is deliberately + # enforced at the harness boundary because backend duplicate semantics differ. + replay_event_id = f"{session_id}:{operation['id']}" + if replay_event_id in backend.seen_event_ids: + continue + backend.seen_event_ids.add(replay_event_id) + await backend.session_service.append_event(session, _event(operation, index)) + elif op == "memory": + backend.memory_queries.append(operation["query"]) + elif op == "summary": + backend.summaries[session_id] = { + "id": operation["id"], "session_id": session_id, + "version": operation["version"], "text": operation["text"], + "summary_timestamp": 1_700_000_100.0 + index, + } + elif op == "truncate": + stored = await backend.session_service.get_session(app_name=app_name, user_id=user_id, + session_id=session_id) + stored.historical_events.extend(stored.events[:-operation["keep"] or None]) + stored.events = stored.events[-operation["keep"]:] if operation["keep"] else [] + await backend.session_service.update_session(stored) + elif op == "fail": + continue # simulated failure occurs before a storage call + + stored = await backend.session_service.get_session(app_name=app_name, user_id=user_id, session_id=session_id) + await backend.memory_service.store_session(stored) + memories = [] + for query in backend.memory_queries: + result = await backend.memory_service.search_memory(stored.save_key, query) + memories.extend(item.model_dump(mode="json", exclude_none=True) for item in result.memories) + return { + "session_id": stored.id, + "events": [event.model_dump(mode="json", exclude_none=True) for event in stored.events], + "historical_events": [event.model_dump(mode="json", exclude_none=True) for event in stored.historical_events], + "state": stored.state, + "memory": memories, + "summary": backend.summaries.get(session_id), + } + + +def normalize(value: Any, path: str = "", allowed_diff: set[str] | None = None) -> Any: + """Remove only explicitly volatile fields and canonicalize summary text/order.""" + allowed_diff = allowed_diff or set() + if isinstance(value, dict): + result = {} + for key in sorted(value): + child_path = f"{path}.{key}" if path else key + if key in VOLATILE_FIELDS or child_path in allowed_diff: + continue + child = normalize(value[key], child_path, allowed_diff) + if key == "long_running_tool_ids" and child in (None, []): + continue + if key == "text" and path.endswith("summary") and isinstance(child, str): + child = re.sub(r"\s+", " ", child).strip() + result[key] = child + return result + if isinstance(value, list): + return [normalize(item, f"{path}[{index}]", allowed_diff) for index, item in enumerate(value)] + return value + + +def diff_values(left: Any, right: Any, path: str = "") -> list[dict[str, Any]]: + """Return leaf differences with an actionable JSON-style field path.""" + if type(left) is not type(right): + return [{"path": path or "$", "left": left, "right": right}] + if isinstance(left, dict): + diffs = [] + for key in sorted(set(left) | set(right)): + child_path = f"{path}.{key}" if path else key + if key not in left or key not in right: + diffs.append({"path": child_path, "left": left.get(key), "right": right.get(key)}) + else: + diffs.extend(diff_values(left[key], right[key], child_path)) + return diffs + if isinstance(left, list): + diffs = [] + for index in range(max(len(left), len(right))): + child_path = f"{path}[{index}]" + if index >= len(left) or index >= len(right): + diffs.append({"path": child_path, + "left": left[index] if index < len(left) else None, + "right": right[index] if index < len(right) else None}) + else: + diffs.extend(diff_values(left[index], right[index], child_path)) + return diffs + return [] if left == right else [{"path": path or "$", "left": left, "right": right}] + + +def compare_snapshots(left: dict[str, Any], right: dict[str, Any], + allowed_diff: set[str] | None = None) -> list[dict[str, Any]]: + return diff_values(normalize(left, allowed_diff=allowed_diff), + normalize(right, allowed_diff=allowed_diff)) + + +MUTATIONS: list[tuple[str, Callable[[dict[str, Any]], None]]] = [ + ("event_text", lambda s: s["events"][0]["content"]["parts"][0].update(text="corrupt")), + ("event_order", lambda s: s["events"].reverse()), + ("event_missing", lambda s: s["events"].pop()), + ("state_value", lambda s: s["state"].update(theme="corrupt")), + ("memory_content", lambda s: s["memory"][0]["content"]["parts"][0].update(text="corrupt")), + ("memory_author", lambda s: s["memory"][0].update(author="corrupt")), + ("summary_missing", lambda s: s.update(summary=None)), + ("summary_text", lambda s: s["summary"].update(text="corrupt")), + ("summary_session", lambda s: s["summary"].update(session_id="wrong-session")), + ("summary_version", lambda s: s["summary"].update(version=999)), +] + + +def mutate(snapshot: dict[str, Any], name: str) -> dict[str, Any]: + mutated = copy.deepcopy(snapshot) + dict(MUTATIONS)[name](mutated) + return mutated diff --git a/tests/sessions/test_replay_consistency.py b/tests/sessions/test_replay_consistency.py new file mode 100644 index 000000000..50178dfc4 --- /dev/null +++ b/tests/sessions/test_replay_consistency.py @@ -0,0 +1,103 @@ +"""Cross-backend replay consistency and fault-detection acceptance tests.""" + +from __future__ import annotations + +import json +import os +import time +from pathlib import Path + +import pytest + +from trpc_agent_sdk.memory import SqlMemoryService +from trpc_agent_sdk.sessions import InMemorySessionService, SessionServiceConfig, SqlSessionService + +from .replay_harness import MUTATIONS, ReplayBackend, compare_snapshots, load_cases, mutate, replay_case + + +ROOT = Path(__file__).parents[2] +CASES = load_cases(ROOT / "replay_cases/session_memory_summary.json") + + +async def _backends() -> list[ReplayBackend]: + config = SessionServiceConfig(store_historical_events=True) + memory = ReplayBackend("in_memory", InMemorySessionService(session_config=config.model_copy(deep=True))) + sqlite_service = SqlSessionService(db_url="sqlite:///:memory:", is_async=False, + session_config=config.model_copy(deep=True)) + sqlite_memory = SqlMemoryService(db_url="sqlite:///:memory:", is_async=False, enabled=True) + await sqlite_service._sql_storage.create_sql_engine() + await sqlite_memory._sql_storage.create_sql_engine() + return [memory, ReplayBackend("sqlite", sqlite_service, memory_service=sqlite_memory)] + + +@pytest.mark.parametrize("case", CASES, ids=lambda case: case["name"]) +async def test_replay_case_is_consistent(case): + backends = await _backends() + try: + snapshots = [await replay_case(backend, case) for backend in backends] + diffs = compare_snapshots(snapshots[0], snapshots[1]) + assert not diffs, json.dumps(diffs, ensure_ascii=False, indent=2) + finally: + for backend in backends: + await backend.close() + + +@pytest.mark.parametrize("mutation", [name for name, _ in MUTATIONS]) +async def test_detects_injected_inconsistency(mutation): + # Each mutation uses a snapshot containing the relevant event/state/memory/summary fields. + case = { + "name": f"mutation-{mutation}", + "operations": [ + {"op": "event", "id": "e1", "author": "user", "text": "jasmine preference"}, + {"op": "event", "id": "e2", "author": "agent", "text": "noted"}, + {"op": "state", "id": "e3", "values": {"theme": "dark"}}, + {"op": "memory", "query": "jasmine"}, + {"op": "summary", "id": "sum-1", "version": 1, "text": "Preference recorded."}, + ], + } + backend = ReplayBackend("in_memory", InMemorySessionService()) + try: + snapshot = await replay_case(backend, case) + diffs = compare_snapshots(snapshot, mutate(snapshot, mutation)) + assert diffs, f"mutation {mutation} was not detected" + assert all({"path", "left", "right"} <= diff.keys() for diff in diffs) + finally: + await backend.close() + + +async def test_lightweight_suite_budget_and_report(tmp_path): + started = time.monotonic() + backends = await _backends() + report = {"mode": "lightweight", "backends": [b.name for b in backends], "cases": []} + try: + for case in CASES: + snapshots = [await replay_case(backend, case) for backend in backends] + report["cases"].append({"case": case["name"], "session_id": snapshots[0]["session_id"], + "differences": compare_snapshots(*snapshots)}) + finally: + for backend in backends: + await backend.close() + report["duration_seconds"] = round(time.monotonic() - started, 3) + assert report["duration_seconds"] < 30 + assert all(not case["differences"] for case in report["cases"]) + output = tmp_path / "session_memory_summary_diff_report.json" + output.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8") + assert json.loads(output.read_text(encoding="utf-8"))["cases"] + + +@pytest.mark.skipif(not os.getenv("TRPC_REPLAY_SQL_DB_URL"), + reason="set TRPC_REPLAY_SQL_DB_URL to enable external SQL replay") +async def test_external_sql_integration(): + db_url = os.environ["TRPC_REPLAY_SQL_DB_URL"] + session_service = SqlSessionService(db_url=db_url, is_async=False) + memory_service = SqlMemoryService(db_url=db_url, is_async=False, enabled=True) + await session_service._sql_storage.create_sql_engine() + await memory_service._sql_storage.create_sql_engine() + persistent = ReplayBackend("external_sql", session_service, memory_service=memory_service) + reference = ReplayBackend("in_memory", InMemorySessionService()) + try: + for case in CASES: + assert not compare_snapshots(await replay_case(reference, case), await replay_case(persistent, case)) + finally: + await reference.close() + await persistent.close()