diff --git a/tests/sessions/replay_cases.py b/tests/sessions/replay_cases.py new file mode 100644 index 00000000..46a0ca71 --- /dev/null +++ b/tests/sessions/replay_cases.py @@ -0,0 +1,547 @@ +# 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 case definitions for Session / Memory / Summary consistency testing. + +Each replay case describes a sequence of operations to be executed against +multiple backends, plus expected outcomes for comparison. + +Case Index +========== + 1. single_turn_text - Simple user text + agent text response + 2. multi_turn_text - Multiple rounds of user/assistant events + 3. tool_call_conversation - function_call + function_response + 4. state_update_and_override - Multiple state writes and overwrites + 5. memory_write_and_read - Store session memory and search + 6. memory_facts_and_prefs - Simulate user preferences and facts + 7. summary_create_and_verify - Create summary, verify content/metadata + 8. summary_with_truncation - Events compressed; summary anchors context + 9. summary_missing_detection - Inject summary loss for detection +10. summary_wrong_session - Inject cross-session summary for detection +11. duplicate_event_detection - Detect duplicate events after simulated re-write +12. state_dirty_after_error - Partial state corruption detection +""" + +from __future__ import annotations + +import time +from dataclasses import dataclass +from dataclasses import field +from typing import Any +from typing import Dict +from typing import List +from typing import Optional + +from trpc_agent_sdk.events import Event +from trpc_agent_sdk.types import Content +from trpc_agent_sdk.types import FunctionCall +from trpc_agent_sdk.types import FunctionResponse +from trpc_agent_sdk.types import Part + +# ────────────────────────────────────────────────────────────── +# Constants shared across all cases +# ────────────────────────────────────────────────────────────── +APP_NAME = "replay_test_app" +USER_ID = "replay_test_user" +SESSION_ID = "replay-test-session-001" + +# ────────────────────────────────────────────────────────────── +# Helper factories +# ────────────────────────────────────────────────────────────── + + +def _make_user_event(text: str, ts: float = 0.0) -> Event: + return Event( + invocation_id="user-inv", + author="user", + content=Content(role="user", parts=[Part.from_text(text=text)]), + timestamp=ts or time.time(), + ) + + +def _make_agent_text(text: str, ts: float = 0.0) -> Event: + return Event( + invocation_id="agent-inv", + author="agent", + content=Content(role="model", parts=[Part.from_text(text=text)]), + timestamp=ts or time.time(), + ) + + +def _make_tool_call_event(name: str, args: dict, ts: float = 0.0) -> Event: + fc = FunctionCall(name=name, args=args) + return Event( + invocation_id="tool-inv", + author="agent", + content=Content(role="model", parts=[Part(function_call=fc)]), + timestamp=ts or time.time(), + ) + + +def _make_tool_response_event(name: str, response: dict, ts: float = 0.0) -> Event: + fr = FunctionResponse(name=name, response=response) + return Event( + invocation_id="tool-resp-inv", + author="agent", + content=Content(role="user", parts=[Part(function_response=fr)]), + timestamp=ts or time.time(), + ) + + +def _make_summary_event(text: str, ts: float = 0.0) -> Event: + ev = Event( + invocation_id="summary", + author="system", + content=Content(role="user", parts=[Part.from_text(text=f"Previous conversation summary: {text}")]), + timestamp=ts or time.time(), + ) + ev.set_summary_event(True) + return ev + + +# ────────────────────────────────────────────────────────────── +# ReplayCase container +# ────────────────────────────────────────────────────────────── + + +@dataclass +class ReplayStep: + """A single operation in a replay case.""" + op: str # create_session | append_event | update_session | update_session_events | + # store_memory | search_memory | create_summary | get_session_summary | inject_corruption + kwargs: Dict[str, Any] = field(default_factory=dict) + + +@dataclass +class ReplayCase: + """Defines a full replay scenario.""" + case_id: str + description: str + # Sequence of operations to replay + steps: List[ReplayStep] = field(default_factory=list) + # Expected outcomes *before* any intentional corruption + expected_event_count: int = 0 + expected_state: Dict[str, Any] = field(default_factory=dict) + expected_memory_hits: int = 0 + expected_summary_available: bool = False + expected_summary_text: str = "" + # For corruption-detection cases: what corruption we inject + corruption_type: Optional[str] = None # missing_summary | wrong_session_summary | duplicate_event | dirty_state + corruption_description: str = "" + + +# ────────────────────────────────────────────────────────────── +# Case 1: single_turn_text +# ────────────────────────────────────────────────────────────── +CASE_SINGLE_TURN = ReplayCase( + case_id="single_turn_text", + description="Single turn: user text followed by agent text response", + expected_event_count=2, + steps=[ + ReplayStep(op="create_session", kwargs={"app_name": APP_NAME, "user_id": USER_ID, "session_id": SESSION_ID}), + ReplayStep(op="append_event", kwargs={"event": _make_user_event("Hello, how are you?")}), + ReplayStep(op="append_event", kwargs={"event": _make_agent_text("I'm doing well, thank you!")}), + ], +) + +# ────────────────────────────────────────────────────────────── +# Case 2: multi_turn_text +# ────────────────────────────────────────────────────────────── +CASE_MULTI_TURN = ReplayCase( + case_id="multi_turn_text", + description="Multiple consecutive user/assistant turns", + expected_event_count=6, + steps=[ + ReplayStep(op="create_session", kwargs={"app_name": APP_NAME, "user_id": USER_ID, "session_id": SESSION_ID}), + ReplayStep(op="append_event", kwargs={"event": _make_user_event("What's the weather?")}), + ReplayStep(op="append_event", kwargs={"event": _make_agent_text("It's sunny, 25°C.")}), + ReplayStep(op="append_event", kwargs={"event": _make_user_event("And tomorrow?")}), + ReplayStep(op="append_event", kwargs={"event": _make_agent_text("Tomorrow will be cloudy, 22°C.")}), + ReplayStep(op="append_event", kwargs={"event": _make_user_event("Thanks!")}), + ReplayStep(op="append_event", kwargs={"event": _make_agent_text("You're welcome!")}), + ], +) + +# ────────────────────────────────────────────────────────────── +# Case 3: tool_call_conversation +# ────────────────────────────────────────────────────────────── +CASE_TOOL_CALL = ReplayCase( + case_id="tool_call_conversation", + description="Conversation with function_call and function_response events", + expected_event_count=4, + steps=[ + ReplayStep(op="create_session", kwargs={"app_name": APP_NAME, "user_id": USER_ID, "session_id": SESSION_ID}), + ReplayStep(op="append_event", kwargs={"event": _make_user_event("Search for Python tutorials")}), + ReplayStep( + op="append_event", + kwargs={ + "event": + _make_tool_call_event("web_search", {"query": "Python tutorials", "max_results": 5}) + }, + ), + ReplayStep( + op="append_event", + kwargs={ + "event": + _make_tool_response_event("web_search", {"results": ["result1", "result2", "result3"]}) + }, + ), + ReplayStep( + op="append_event", + kwargs={ + "event": + _make_agent_text("I found 3 Python tutorials for you: result1, result2, result3.") + }, + ), + ], +) + +# ────────────────────────────────────────────────────────────── +# Case 4: state_update_and_override +# ────────────────────────────────────────────────────────────── +CASE_STATE_UPDATE = ReplayCase( + case_id="state_update_and_override", + description="Multiple state writes: initial, app-level, user-level, session-scoped, override", + expected_event_count=2, + expected_state={ + "app:theme": "dark", + "app:version": "2.0", + "user:name": "AliceUpdated", + "user:pref_lang": "zh-CN", + "session_key": "overridden_value", + "counter": 2, + }, + steps=[ + ReplayStep( + op="create_session", + kwargs={ + "app_name": APP_NAME, + "user_id": USER_ID, + "session_id": SESSION_ID, + "state": {"app:theme": "dark", "user:name": "Alice", "session_key": "initial"}, + }, + ), + # First event with state delta + ReplayStep( + op="append_event", + kwargs={ + "event": + _make_user_event("Update my preferences"), + "state_delta": {"user:pref_lang": "en-US", "counter": 1}, + }, + ), + # Second event overrides some state + ReplayStep( + op="append_event", + kwargs={ + "event": + _make_agent_text("Preferences updated!"), + "state_delta": { + "user:name": "AliceUpdated", + "user:pref_lang": "zh-CN", + "app:version": "2.0", + "session_key": "overridden_value", + "counter": 2, + }, + }, + ), + ], +) + +# ────────────────────────────────────────────────────────────── +# Case 5: memory_write_and_read +# ────────────────────────────────────────────────────────────── +CASE_MEMORY_WRITE_READ = ReplayCase( + case_id="memory_write_and_read", + description="Store a session in memory service and search for relevant content", + expected_event_count=4, + expected_memory_hits=1, + steps=[ + ReplayStep(op="create_session", kwargs={"app_name": APP_NAME, "user_id": USER_ID, "session_id": SESSION_ID}), + ReplayStep(op="append_event", kwargs={"event": _make_user_event("My favorite color is blue.")}), + ReplayStep(op="append_event", kwargs={"event": _make_agent_text("Got it, blue is your favorite!")}), + ReplayStep(op="append_event", kwargs={"event": _make_user_event("I live in Shanghai.")}), + ReplayStep(op="append_event", kwargs={"event": _make_agent_text("Shanghai is a great city!")}), + ReplayStep(op="store_memory"), + ReplayStep(op="search_memory", kwargs={"query": "favorite color"}), + ], +) + +# ────────────────────────────────────────────────────────────── +# Case 6: memory_facts_and_prefs +# ────────────────────────────────────────────────────────────── +CASE_MEMORY_FACTS = ReplayCase( + case_id="memory_facts_and_prefs", + description="Store user facts and preferences in memory, then verify retrieval", + expected_event_count=6, + expected_memory_hits=2, + steps=[ + ReplayStep(op="create_session", kwargs={"app_name": APP_NAME, "user_id": USER_ID, "session_id": SESSION_ID}), + ReplayStep( + op="append_event", + kwargs={"event": _make_user_event("I prefer dark mode and use Python for development.")}), + ReplayStep(op="append_event", kwargs={"event": _make_agent_text("Noted: dark mode preference and Python.")}), + ReplayStep( + op="append_event", + kwargs={ + "event": + _make_user_event("My birthday is March 15th and I work as a backend engineer.") + }), + ReplayStep( + op="append_event", + kwargs={ + "event": + _make_agent_text("Got it, birthday March 15th, backend engineer.") + }), + ReplayStep( + op="append_event", + kwargs={ + "event": + _make_user_event("I like swimming and hiking on weekends.") + }), + ReplayStep( + op="append_event", + kwargs={ + "event": + _make_agent_text("Swimming and hiking - great activities!") + }), + ReplayStep(op="store_memory"), + ReplayStep(op="search_memory", kwargs={"query": "Python developer birthday"}), + ], +) + +# ────────────────────────────────────────────────────────────── +# Case 7: summary_create_and_verify +# ────────────────────────────────────────────────────────────── +_SUMMARY_TEXT_7 = ( + "User and agent had a conversation about project planning. " + "Key decisions: use Python for backend, React for frontend, " + "and PostgreSQL for database. Action items: set up CI/CD pipeline." +) + +CASE_SUMMARY_CREATE = ReplayCase( + case_id="summary_create_and_verify", + description="Create a summary after conversation, verify content, version, metadata", + expected_event_count=6, + expected_summary_available=True, + expected_summary_text=_SUMMARY_TEXT_7, + steps=[ + ReplayStep(op="create_session", kwargs={"app_name": APP_NAME, "user_id": USER_ID, "session_id": SESSION_ID}), + ReplayStep(op="append_event", + kwargs={"event": _make_user_event("Let's plan our new project architecture.")}), + ReplayStep( + op="append_event", + kwargs={ + "event": + _make_agent_text("Great! Let's start by choosing the tech stack.") + }), + ReplayStep(op="append_event", kwargs={"event": _make_user_event("I think Python for backend.")}), + ReplayStep( + op="append_event", + kwargs={"event": _make_agent_text("Python is good. For frontend, React is popular.")}), + ReplayStep( + op="append_event", + kwargs={"event": _make_user_event("Let's also use PostgreSQL and set up CI/CD.")}), + ReplayStep( + op="append_event", + kwargs={ + "event": + _make_agent_text("Agreed. Summary: Python + React + PostgreSQL + CI/CD.") + }), + ReplayStep( + op="create_summary", + kwargs={ + "summary_text": _SUMMARY_TEXT_7, + "original_event_count": 6, + "compressed_event_count": 3, + }, + ), + ], +) + +# ────────────────────────────────────────────────────────────── +# Case 8: summary_with_event_truncation +# ────────────────────────────────────────────────────────────── +_SUMMARY_TEXT_8 = ( + "Long conversation about travel planning. User wants to visit Japan. " + "Key details: budget $3000, dates flexible in October, " + "interested in Tokyo and Kyoto." +) + +CASE_SUMMARY_TRUNCATION = ReplayCase( + case_id="summary_with_truncation", + description="Summary compresses historical events; remaining events + summary restore context", + expected_event_count=4, # after truncation: summary + 2 recent + 1 new + expected_summary_available=True, + expected_summary_text=_SUMMARY_TEXT_8, + steps=[ + ReplayStep(op="create_session", kwargs={"app_name": APP_NAME, "user_id": USER_ID, "session_id": SESSION_ID}), + # Build up "long" conversation (8 events) + ReplayStep(op="append_event", kwargs={"event": _make_user_event("I want to plan a trip to Japan.")}), + ReplayStep(op="append_event", kwargs={"event": _make_agent_text("Wonderful! When are you thinking?")}), + ReplayStep(op="append_event", kwargs={"event": _make_user_event("Sometime in October, flexible dates.")}), + ReplayStep(op="append_event", kwargs={"event": _make_agent_text("October is a great time to visit.")}), + ReplayStep(op="append_event", kwargs={"event": _make_user_event("My budget is around $3000.")}), + ReplayStep(op="append_event", kwargs={"event": _make_agent_text("That's a reasonable budget.")}), + ReplayStep( + op="append_event", + kwargs={"event": _make_user_event("I want to visit Tokyo and Kyoto mainly.")}), + ReplayStep(op="append_event", kwargs={"event": _make_agent_text("Both are excellent choices!")}), + # Create summary (simulates compression, keeps summary + 2 recent events) + ReplayStep( + op="create_summary", + kwargs={ + "summary_text": _SUMMARY_TEXT_8, + "original_event_count": 8, + "compressed_event_count": 3, + "keep_recent_count": 2, + }, + ), + # Append new event after truncation + ReplayStep( + op="append_event", + kwargs={ + "event": + _make_user_event("Also, any recommendations for hotels in Tokyo?") + }), + ], +) + +# ────────────────────────────────────────────────────────────── +# Case 9: summary_missing_detection (corruption case) +# ────────────────────────────────────────────────────────────── +CASE_SUMMARY_MISSING = ReplayCase( + case_id="summary_missing_detection", + description="Inject summary loss: one backend has summary, the other does not", + expected_event_count=4, + expected_summary_available=True, + expected_summary_text="User asked about machine learning basics.", + corruption_type="missing_summary", + corruption_description="Backend B will have no summary while Backend A does", + steps=[ + ReplayStep(op="create_session", kwargs={"app_name": APP_NAME, "user_id": USER_ID, "session_id": SESSION_ID}), + ReplayStep(op="append_event", kwargs={"event": _make_user_event("Tell me about machine learning.")}), + ReplayStep( + op="append_event", + kwargs={ + "event": + _make_agent_text("Machine learning is a subset of AI that enables systems to learn from data.") + }), + ReplayStep(op="append_event", kwargs={"event": _make_user_event("What about supervised learning?")}), + ReplayStep( + op="append_event", + kwargs={ + "event": + _make_agent_text("Supervised learning uses labeled data to train models.") + }), + ReplayStep( + op="create_summary", + kwargs={ + "summary_text": "User asked about machine learning basics.", + "original_event_count": 4, + "compressed_event_count": 3, + }, + ), + ], +) + +# ────────────────────────────────────────────────────────────── +# Case 10: summary_wrong_session (corruption case) +# ────────────────────────────────────────────────────────────── +CASE_SUMMARY_WRONG_SESSION = ReplayCase( + case_id="summary_wrong_session", + description="Inject cross-session summary: summary claims to be for session A but appears in session B", + expected_event_count=2, + expected_summary_available=True, + expected_summary_text="User prefers dark mode and Python development.", + corruption_type="wrong_session_summary", + corruption_description="Backend B will have a summary with a different session_id", + steps=[ + ReplayStep(op="create_session", kwargs={"app_name": APP_NAME, "user_id": USER_ID, "session_id": SESSION_ID}), + ReplayStep(op="append_event", kwargs={"event": _make_userEvent("I like dark mode.")}), + ReplayStep(op="append_event", kwargs={"event": _make_agent_text("Noted your dark mode preference.")}), + ReplayStep( + op="create_summary", + kwargs={ + "summary_text": "User prefers dark mode and Python development.", + "original_event_count": 2, + "compressed_event_count": 1, + }, + ), + ], +) + +# ────────────────────────────────────────────────────────────── +# Case 11: duplicate_event_detection (corruption case) +# ────────────────────────────────────────────────────────────── +CASE_DUPLICATE_EVENT = ReplayCase( + case_id="duplicate_event_detection", + description="Simulate failed write followed by retry: detect duplicate events", + expected_event_count=2, + corruption_type="duplicate_event", + corruption_description="Backend B will have an extra duplicate event injected", + steps=[ + ReplayStep(op="create_session", kwargs={"app_name": APP_NAME, "user_id": USER_ID, "session_id": SESSION_ID}), + ReplayStep( + op="append_event", + kwargs={"event": _make_user_event("What is the capital of France?")}), + ReplayStep( + op="append_event", + kwargs={"event": _make_agent_text("The capital of France is Paris.")}), + ], +) + +# ────────────────────────────────────────────────────────────── +# Case 12: state_dirty_after_error (corruption case) +# ────────────────────────────────────────────────────────────── +CASE_STATE_DIRTY = ReplayCase( + case_id="state_dirty_after_error", + description="Simulate partial write failure: one backend has incomplete state", + expected_event_count=2, + expected_state={"app:env": "production", "user:role": "admin", "config_key": "expected_value"}, + corruption_type="dirty_state", + corruption_description="Backend B will have a missing or incorrect state key", + steps=[ + ReplayStep( + op="create_session", + kwargs={ + "app_name": APP_NAME, + "user_id": USER_ID, + "session_id": SESSION_ID, + "state": {"app:env": "staging", "user:role": "viewer"}, + }, + ), + ReplayStep( + op="append_event", + kwargs={ + "event": _make_user_event("Upgrade my account"), + "state_delta": { + "app:env": "production", + "user:role": "admin", + "config_key": "expected_value", + }, + }, + ), + ReplayStep(op="append_event", kwargs={"event": _make_agent_text("Account upgraded successfully.")}), + ], +) + +# ────────────────────────────────────────────────────────────── +# Master list of ALL replay cases +# ────────────────────────────────────────────────────────────── +ALL_REPLAY_CASES: List[ReplayCase] = [ + CASE_SINGLE_TURN, + CASE_MULTI_TURN, + CASE_TOOL_CALL, + CASE_STATE_UPDATE, + CASE_MEMORY_WRITE_READ, + CASE_MEMORY_FACTS, + CASE_SUMMARY_CREATE, + CASE_SUMMARY_TRUNCATION, + CASE_SUMMARY_MISSING, + CASE_SUMMARY_WRONG_SESSION, + CASE_DUPLICATE_EVENT, + CASE_STATE_DIRTY, +] diff --git a/tests/sessions/replay_harness.py b/tests/sessions/replay_harness.py new file mode 100644 index 00000000..0beb0dd6 --- /dev/null +++ b/tests/sessions/replay_harness.py @@ -0,0 +1,989 @@ +# 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 for Session / Memory / Summary consistency testing. + +Architecture: + ReplayHarness + ├── executes ReplayCase against two backends (A and B) + ├── collects results into BackendResult + ├── normalizes data via Normalizer pipeline + ├── compares via Comparator pipeline (events, state, memory, summary) + └── produces DiffReport entries + +Normalization strategy: + - Timestamps: normalized to 0.0 (non-deterministic across backends) + - Auto-generated IDs: normalized to "" (event.id, session.id may differ) + - Serialization field order: dict keys sorted for comparison + - None vs missing: treated as equivalent where appropriate + - Summary text: whitespace-normalized semantic comparison + +Allowed differences (allowed_diff): + - timestamp: always differs between backends + - event.id: auto-generated UUID, differs + - last_update_time: always differs + - save_key: backend-dependent format +""" + +from __future__ import annotations + +import asyncio +import copy +import json +import math +import os +import time +from dataclasses import dataclass +from dataclasses import field +from typing import Any +from typing import Dict +from typing import List +from typing import Optional +from typing import Tuple + +from trpc_agent_sdk.abc import MemoryServiceABC +from trpc_agent_sdk.abc import MemoryServiceConfig as MemoryConfig +from trpc_agent_sdk.abc import SearchMemoryResponse +from trpc_agent_sdk.abc import SessionServiceABC +from trpc_agent_sdk.events import Event +from trpc_agent_sdk.log import logger +from trpc_agent_sdk.memory import InMemoryMemoryService +from trpc_agent_sdk.sessions import InMemorySessionService +from trpc_agent_sdk.sessions import Session +from trpc_agent_sdk.sessions import SessionServiceConfig +from trpc_agent_sdk.sessions._session_summarizer import SessionSummary +from trpc_agent_sdk.types import Content +from trpc_agent_sdk.types import EventActions +from trpc_agent_sdk.types import Part +from trpc_agent_sdk.utils import user_key + +from .replay_cases import APP_NAME +from .replay_cases import SESSION_ID +from .replay_cases import USER_ID +from .replay_cases import ReplayCase +from .replay_cases import ReplayStep + +# ────────────────────────────────────────────────────────────── +# Allowed-diff field paths (normalized to dot-notation) +# ────────────────────────────────────────────────────────────── +_ALLOWED_EVENT_FIELDS = { + "id", # auto-generated UUID + "timestamp", # backend-dependent clock + "invocation_id", # may differ if replay produces new IDs +} + +_ALLOWED_SESSION_FIELDS = { + "last_update_time", # backend-dependent clock + "save_key", # backend-dependent format +} + +_ALLOWED_SUMMARY_FIELDS = { + "summary_timestamp", # backend-dependent clock +} + +# Fields that represent "auto-generated ID" patterns +_ID_FIELD_PATTERNS = {"id", "invocation_id", "request_id", "response_id"} + + +# ────────────────────────────────────────────────────────────── +# Diff entry and report structures +# ────────────────────────────────────────────────────────────── + + +@dataclass +class DiffEntry: + """A single diff between two backends for a specific field.""" + session_id: str = "" + component: str = "" # events | state | memory | summary + event_index: Optional[int] = None # index in event list, if applicable + summary_id: Optional[str] = None # summary identifier + field_path: str = "" # dot-separated path to the differing field + value_a: Any = None # value from backend A + value_b: Any = None # value from backend B + allowed: bool = False # whether this diff is expected/allowed + note: str = "" # explanation of the difference + + +@dataclass +class BackendResult: + """Collected results after replaying a case against one backend.""" + backend_name: str = "" + session: Optional[Session] = None + memory_response: Optional[SearchMemoryResponse] = None + summary: Optional[SessionSummary] = None + error: Optional[str] = None + + +@dataclass +class CaseDiffReport: + """Diff report for a single replay case across two backends.""" + case_id: str = "" + backend_a: str = "" + backend_b: str = "" + diffs: List[DiffEntry] = field(default_factory=list) + passed: bool = True + note: str = "" + + +# ────────────────────────────────────────────────────────────── +# Normalizers +# ────────────────────────────────────────────────────────────── + + +def _normalize_timestamps(obj: Any, _path: str = "") -> Any: + """Zero out all float timestamp-like fields for comparison.""" + if isinstance(obj, dict): + return {k: _normalize_timestamps(v, f"{_path}.{k}" if _path else k) for k, v in obj.items()} + if isinstance(obj, list): + return [_normalize_timestamps(v, f"{_path}[{i}]") for i, v in enumerate(obj)] + if isinstance(obj, float): + # Heuristic: large floats near epoch are timestamps + if 1e8 < obj < 1e13: + return 0.0 + if 0 < obj < 1000000: + # Small float, check path context + for kw in ("timestamp", "time", "update_time", "last_update", "created_at"): + if kw in _path.lower(): + return 0.0 + return obj + return obj + + +def _normalize_ids(obj: Any, _path: str = "") -> Any: + """Zero out auto-generated ID fields that are UUID-like or backend-dependent.""" + if isinstance(obj, dict): + result = {} + for k, v in obj.items(): + full_path = f"{_path}.{k}" if _path else k + if k.lower() in _ID_FIELD_PATTERNS and isinstance(v, str) and len(v) > 20: + result[k] = "" + else: + result[k] = _normalize_ids(v, full_path) + return result + if isinstance(obj, list): + return [_normalize_ids(v, f"{_path}[{i}]") for i, v in enumerate(obj)] + return obj + + +def _sort_dict_keys(obj: Any) -> Any: + """Sort dict keys for deterministic comparison (handles JSON serialization order).""" + if isinstance(obj, dict): + return {k: _sort_dict_keys(v) for k, v in sorted(obj.items())} + if isinstance(obj, list): + # Sort lists of dicts by a stable hash of their content + result = [_sort_dict_keys(v) for v in obj] + if result and all(isinstance(v, dict) for v in result): + try: + result.sort(key=lambda d: json.dumps(d, sort_keys=True, default=str)) + except (TypeError, ValueError): + pass + return result + return obj + + +def _normalize_none_vs_missing(obj: Any) -> Any: + """Treat None and missing keys as the same by removing None values.""" + if isinstance(obj, dict): + return {k: _normalize_none_vs_missing(v) for k, v in obj.items() if v is not None} + if isinstance(obj, list): + return [_normalize_none_vs_missing(v) for v in obj] + return obj + + +def _normalize_whitespace(text: str) -> str: + """Normalize whitespace for semantic text comparison.""" + if not text: + return "" + return " ".join(text.split()) + + +def normalize_event(event: Event) -> Dict[str, Any]: + """Normalize an Event for comparison.""" + data = event.model_dump(exclude_none=False, mode="json") + # Apply the normalization pipeline + data = _normalize_ids(data) + data = _normalize_timestamps(data) + data = _normalize_none_vs_missing(data) + data = _sort_dict_keys(data) + return data + + +def normalize_state(state: Dict[str, Any]) -> Dict[str, Any]: + """Normalize session state for comparison.""" + # Deep copy to avoid mutating + result = copy.deepcopy(state or {}) + # Strip temp: prefix keys (they're ephemeral) + result = {k: v for k, v in result.items() if not k.startswith("temp:")} + result = _normalize_timestamps(result) + result = _sort_dict_keys(result) + return result + + +def normalize_summary(summary: Optional[SessionSummary]) -> Optional[Dict[str, Any]]: + """Normalize a SessionSummary for comparison.""" + if summary is None: + return None + data = { + "session_id": summary.session_id, + "summary_text": _normalize_whitespace(summary.summary_text), + "original_event_count": summary.original_event_count, + "compressed_event_count": summary.compressed_event_count, + "summary_timestamp": 0.0, # always zeroed + } + return data + + +def normalize_memory(memory: Optional[SearchMemoryResponse]) -> Dict[str, Any]: + """Normalize a SearchMemoryResponse for comparison.""" + if memory is None: + return {"memories": []} + result = {"memories": []} + for mem in memory.memories: + entry = { + "author": _normalize_whitespace(mem.author or ""), + "timestamp": "", # always zeroed + "content_text": "", + } + if mem.content and mem.content.parts: + text_parts = [p.text for p in mem.content.parts if p.text] + entry["content_text"] = _normalize_whitespace(" ".join(text_parts)) + result["memories"].append(entry) + result = _sort_dict_keys(result) + return result + + +# ────────────────────────────────────────────────────────────── +# Comparators +# ────────────────────────────────────────────────────────────── + + +def _deep_diff( + a: Any, + b: Any, + path: str = "", + allowed_fields: Optional[set] = None, +) -> List[Tuple[str, Any, Any, bool]]: + """Recursively diff two normalized data structures. + + Returns list of (field_path, value_a, value_b, is_allowed). + """ + if allowed_fields is None: + allowed_fields = set() + diffs: List[Tuple[str, Any, Any, bool]] = [] + + if type(a) != type(b): + diffs.append((path, a, b, path in allowed_fields)) + return diffs + + if isinstance(a, dict): + all_keys = set(a.keys()) | set(b.keys()) + for k in sorted(all_keys): + sub_path = f"{path}.{k}" if path else k + if k not in a: + diffs.append((sub_path, "", b[k], sub_path in allowed_fields)) + elif k not in b: + diffs.append((sub_path, a[k], "", sub_path in allowed_fields)) + else: + diffs.extend(_deep_diff(a[k], b[k], sub_path, allowed_fields)) + elif isinstance(a, list): + max_len = max(len(a), len(b)) + for i in range(max_len): + sub_path = f"{path}[{i}]" + if i >= len(a): + diffs.append((sub_path, "", b[i], sub_path in allowed_fields)) + elif i >= len(b): + diffs.append((sub_path, a[i], "", sub_path in allowed_fields)) + else: + diffs.extend(_deep_diff(a[i], b[i], sub_path, allowed_fields)) + elif isinstance(a, float): + if not math.isclose(a, b, rel_tol=1e-9, abs_tol=1e-9): + diffs.append((path, a, b, path in allowed_fields)) + elif a != b: + diffs.append((path, a, b, path in allowed_fields)) + + return diffs + + +def compare_events( + events_a: List[Event], + events_b: List[Event], + session_id: str, +) -> List[DiffEntry]: + """Compare normalized event lists from two backends.""" + diffs: List[DiffEntry] = [] + norm_a = [normalize_event(e) for e in (events_a or [])] + norm_b = [normalize_event(e) for e in (events_b or [])] + + max_len = max(len(norm_a), len(norm_b)) + for i in range(max_len): + prefix = f"events[{i}]" + if i >= len(norm_a): + diffs.append(DiffEntry( + session_id=session_id, + component="events", + event_index=i, + field_path=f"{prefix}", + value_a="", + value_b=norm_b[i], + allowed=False, + note=f"Event at index {i} missing in backend A", + )) + elif i >= len(norm_b): + diffs.append(DiffEntry( + session_id=session_id, + component="events", + event_index=i, + field_path=f"{prefix}", + value_a=norm_a[i], + value_b="", + allowed=False, + note=f"Event at index {i} missing in backend B", + )) + else: + for field_path, va, vb, allowed in _deep_diff(norm_a[i], norm_b[i], prefix, _ALLOWED_EVENT_FIELDS): + diffs.append(DiffEntry( + session_id=session_id, + component="events", + event_index=i, + field_path=field_path, + value_a=va if not isinstance(va, (dict, list)) else json.dumps(va, default=str), + value_b=vb if not isinstance(vb, (dict, list)) else json.dumps(vb, default=str), + allowed=allowed, + note="allowed backend difference" if allowed else "", + )) + return diffs + + +def compare_state( + state_a: Dict[str, Any], + state_b: Dict[str, Any], + session_id: str, +) -> List[DiffEntry]: + """Compare normalized state from two backends.""" + diffs: List[DiffEntry] = [] + norm_a = normalize_state(state_a) + norm_b = normalize_state(state_b) + + for field_path, va, vb, allowed in _deep_diff(norm_a, norm_b, "state", _ALLOWED_SESSION_FIELDS): + diffs.append(DiffEntry( + session_id=session_id, + component="state", + field_path=field_path, + value_a=va if not isinstance(va, (dict, list)) else json.dumps(va, default=str), + value_b=vb if not isinstance(vb, (dict, list)) else json.dumps(vb, default=str), + allowed=allowed, + note="allowed backend difference" if allowed else "", + )) + return diffs + + +def compare_summaries( + summary_a: Optional[SessionSummary], + summary_b: Optional[SessionSummary], + session_id: str, +) -> List[DiffEntry]: + """Compare normalized summaries from two backends. + + Key requirements: + - Summary loss (one None, one not): MUST be detected + - Wrong session_id: MUST be detected + - Text semantic comparison: whitespace-normalized + - Metadata: exact comparison required (original_event_count, compressed_event_count, session_id) + """ + diffs: List[DiffEntry] = [] + + if summary_a is None and summary_b is None: + return diffs + + if summary_a is None and summary_b is not None: + diffs.append(DiffEntry( + session_id=session_id, + component="summary", + summary_id=getattr(summary_b, "summary_text", "")[:50], + field_path="summary", + value_a="", + value_b=normalize_summary(summary_b), + allowed=False, + note="SUMMARY MISSING: Backend A has no summary, Backend B has summary", + )) + return diffs + + if summary_a is not None and summary_b is None: + diffs.append(DiffEntry( + session_id=session_id, + component="summary", + summary_id=getattr(summary_a, "summary_text", "")[:50], + field_path="summary", + value_a=normalize_summary(summary_a), + value_b="", + allowed=False, + note="SUMMARY MISSING: Backend A has summary, Backend B has no summary", + )) + return diffs + + # Both exist - normalize and compare + norm_a = normalize_summary(summary_a) + norm_b = normalize_summary(summary_b) + + # Special check: session_id mismatch (critical) + sid_a = summary_a.session_id + sid_b = summary_b.session_id + if sid_a != sid_b: + diffs.append(DiffEntry( + session_id=session_id, + component="summary", + summary_id=sid_a, + field_path="summary.session_id", + value_a=sid_a, + value_b=sid_b, + allowed=False, + note="SUMMARY SESSION MISMATCH: summary belongs to different session", + )) + + # Full diff + for field_path, va, vb, allowed in _deep_diff(norm_a, norm_b, "summary", _ALLOWED_SUMMARY_FIELDS): + diffs.append(DiffEntry( + session_id=session_id, + component="summary", + summary_id=sid_a, + field_path=field_path, + value_a=va if not isinstance(va, (dict, list)) else json.dumps(va, default=str), + value_b=vb if not isinstance(vb, (dict, list)) else json.dumps(vb, default=str), + allowed=allowed, + note="allowed backend difference" if allowed else "", + )) + + return diffs + + +def compare_memory( + memory_a: Optional[SearchMemoryResponse], + memory_b: Optional[SearchMemoryResponse], + session_id: str, +) -> List[DiffEntry]: + """Compare normalized memory search results.""" + diffs: List[DiffEntry] = [] + norm_a = normalize_memory(memory_a) + norm_b = normalize_memory(memory_b) + + for field_path, va, vb, allowed in _deep_diff(norm_a, norm_b, "memory"): + diffs.append(DiffEntry( + session_id=session_id, + component="memory", + field_path=field_path, + value_a=va if not isinstance(va, (dict, list)) else json.dumps(va, default=str), + value_b=vb if not isinstance(vb, (dict, list)) else json.dumps(vb, default=str), + allowed=allowed, + note="allowed backend difference" if allowed else "", + )) + return diffs + + +# ────────────────────────────────────────────────────────────── +# Backend factory +# ────────────────────────────────────────────────────────────── + + +def _create_inmemory_session_service() -> InMemorySessionService: + """Create an InMemory session service.""" + config = SessionServiceConfig() + config.clean_ttl_config() + return InMemorySessionService(session_config=config) + + +def _create_sql_session_service(db_url: str = "sqlite:///:memory:") -> "SqlSessionService": + """Create a SQL session service (sqlite fallback always available).""" + from trpc_agent_sdk.sessions._sql_session_service import SqlSessionService + config = SessionServiceConfig() + config.clean_ttl_config() + return SqlSessionService(db_url=db_url, session_config=config, is_async=True) + + +def _create_inmemory_memory_service() -> InMemoryMemoryService: + """Create an InMemory memory service.""" + mc = MemoryConfig(enabled=True) + mc.clean_ttl_config() + return InMemoryMemoryService(memory_service_config=mc, enabled=True) + + +def _create_sql_memory_service(db_url: str = "sqlite:///:memory:") -> "SqlMemoryService": + """Create a SQL memory service.""" + from trpc_agent_sdk.memory._sql_memory_service import SqlMemoryService + mc = MemoryConfig(enabled=True) + mc.clean_ttl_config() + return SqlMemoryService(db_url=db_url, memory_service_config=mc, enabled=True, is_async=True) + + +# ────────────────────────────────────────────────────────────── +# Replay executor +# ────────────────────────────────────────────────────────────── + + +class ReplayExecutor: + """Executes a replay case against a single backend. + + Handles the create_session, append_event, store_memory, create_summary + operations as defined by the ReplayCase steps. + """ + + def __init__( + self, + session_service: SessionServiceABC, + memory_service: Optional[MemoryServiceABC] = None, + backend_name: str = "unknown", + ): + self._session_service = session_service + self._memory_service = memory_service + self._backend_name = backend_name + self._session: Optional[Session] = None + # Internal summary cache (simulates SummarizerSessionManager cache) + self._summary_cache: Dict[str, SessionSummary] = {} + self._memory_responses: List[SearchMemoryResponse] = [] + + @property + def session(self) -> Optional[Session]: + return self._session + + async def execute(self, case: ReplayCase) -> BackendResult: + """Execute all steps of a replay case.""" + self._session = None + self._summary_cache = {} + self._memory_responses = [] + + for step in case.steps: + try: + await self._execute_step(step) + except Exception as e: + return BackendResult( + backend_name=self._backend_name, + error=f"Step '{step.op}' failed: {e}", + ) + + # Get final state by re-reading session + final_session = None + if self._session: + final_session = await self._session_service.get_session( + app_name=APP_NAME, + user_id=USER_ID, + session_id=self._session.id, + ) + + # Get final summary + final_summary = self._summary_cache.get(SESSION_ID) + + return BackendResult( + backend_name=self._backend_name, + session=final_session, + memory_response=self._memory_responses[-1] if self._memory_responses else None, + summary=final_summary, + ) + + async def _execute_step(self, step: ReplayStep) -> None: + """Execute a single replay step.""" + op = step.op + kwargs = step.kwargs + + if op == "create_session": + state = kwargs.get("state", None) + session_id = kwargs.get("session_id", SESSION_ID) + self._session = await self._session_service.create_session( + app_name=kwargs.get("app_name", APP_NAME), + user_id=kwargs.get("user_id", USER_ID), + state=state, + session_id=session_id, + ) + + elif op == "append_event": + event = kwargs["event"] + state_delta = kwargs.get("state_delta") + if state_delta: + # Attach state delta to event actions + event.actions.state_delta.update(state_delta) + await self._session_service.append_event(session=self._session, event=event) + + elif op == "update_session": + await self._session_service.update_session(self._session) + + elif op == "update_session_events": + # Directly update session events in the backend + events = kwargs.get("events", []) + if self._session: + self._session.events = events + await self._session_service.update_session(self._session) + + elif op == "store_memory": + if self._memory_service and self._session: + await self._memory_service.store_session(self._session) + + elif op == "search_memory": + if self._memory_service and self._session: + query = kwargs.get("query", "") + response = await self._memory_service.search_memory( + key=user_key(APP_NAME, USER_ID), + query=query, + limit=kwargs.get("limit", 10), + ) + self._memory_responses.append(response) + + elif op == "create_summary": + summary_text = kwargs["summary_text"] + original_count = kwargs.get("original_event_count", 0) + compressed_count = kwargs.get("compressed_event_count", 0) + keep_recent = kwargs.get("keep_recent_count", 2) + + # Create SessionSummary in cache + summary = SessionSummary( + session_id=SESSION_ID, + summary_text=summary_text, + original_event_count=original_count, + compressed_event_count=compressed_count, + summary_timestamp=time.time(), + ) + self._summary_cache[SESSION_ID] = summary + + # Simulate event compression: create summary event + truncate + if self._session: + # Insert summary event at front + summary_event = Event( + invocation_id="summary", + author="system", + content=Content( + role="user", + parts=[Part.from_text( + text=f"Previous conversation summary: {summary_text}" + )], + ), + timestamp=time.time(), + ) + summary_event.set_summary_event(True) + + events = list(self._session.events) + if keep_recent > 0 and len(events) > keep_recent: + # Keep only summary + recent events + recent = events[-keep_recent:] + # Move old events to historical + old_events = events[:-keep_recent] + self._session.historical_events.extend(old_events) + self._session.events = [summary_event] + recent + else: + self._session.events.insert(0, summary_event) + + await self._session_service.update_session(self._session) + + elif op == "inject_corruption": + corruption = kwargs.get("corruption_type", "") + if corruption == "duplicate_event" and self._session and len(self._session.events) > 0: + # Append a duplicate of the last event + dup = copy.deepcopy(self._session.events[-1]) + self._session.events.append(dup) + await self._session_service.update_session(self._session) + + elif corruption == "dirty_state" and self._session: + # Corrupt a state key + self._session.state["corrupted_key"] = "bad_data_from_partial_write" + await self._session_service.update_session(self._session) + + elif corruption == "missing_summary": + # Remove summary from cache + self._summary_cache.pop(SESSION_ID, None) + # Also remove summary events from session + if self._session: + self._session.events = [ + e for e in self._session.events if not e.is_summary_event() + ] + await self._session_service.update_session(self._session) + + elif corruption == "wrong_session_summary": + # Replace with a summary for a different session + if SESSION_ID in self._summary_cache: + self._summary_cache[SESSION_ID] = SessionSummary( + session_id="wrong-session-id-999", + summary_text=self._summary_cache[SESSION_ID].summary_text, + original_event_count=self._summary_cache[SESSION_ID].original_event_count, + compressed_event_count=self._summary_cache[SESSION_ID].compressed_event_count, + summary_timestamp=time.time(), + ) + + async def close(self) -> None: + """Clean up resources.""" + if self._session_service: + await self._session_service.close() + if self._memory_service: + await self._memory_service.close() + + +# ────────────────────────────────────────────────────────────── +# Replay harness orchestrator +# ────────────────────────────────────────────────────────────── + + +def _should_skip_backend(env_var: str) -> bool: + """Check if a backend should be skipped based on environment variable.""" + val = os.environ.get(env_var, "").strip().lower() + return val in ("0", "false", "no", "skip", "") + + +@dataclass +class ReplayHarnessConfig: + """Configuration for the replay harness.""" + lightweight_only: bool = True + """If True, only run InMemory vs SQLite (always available).""" + enable_redis: bool = False + """If True, also test against Redis backend.""" + sql_db_url: str = "sqlite:///:memory:" + """SQL database URL for the SQL backend.""" + run_corruption_cases: bool = True + """If True, run cases that inject intentional corruption.""" + + +class ReplayHarness: + """Orchestrates replay execution and comparison across backends.""" + + def __init__(self, config: Optional[ReplayHarnessConfig] = None): + self._config = config or ReplayHarnessConfig() + self._all_diffs: List[CaseDiffReport] = [] + + @property + def diff_reports(self) -> List[CaseDiffReport]: + return self._all_diffs + + async def run_all_cases(self, cases: List[ReplayCase]) -> List[CaseDiffReport]: + """Run all replay cases and return diff reports.""" + self._all_diffs = [] + + for case in cases: + # Skip corruption cases in lightweight mode? No, we run them all + # but corruption cases should have expected diffs + report = await self._run_case(case) + self._all_diffs.append(report) + + return self._all_diffs + + async def _run_case(self, case: ReplayCase) -> CaseDiffReport: + """Run a single replay case against two backends.""" + logger.info("Running replay case: %s", case.case_id) + + # ── Backend A: InMemory ── + exec_a = ReplayExecutor( + session_service=_create_inmemory_session_service(), + memory_service=_create_inmemory_memory_service(), + backend_name="InMemory", + ) + + # ── Backend B: SQL (SQLite) ── + exec_b = ReplayExecutor( + session_service=_create_sql_session_service(self._config.sql_db_url), + memory_service=_create_sql_memory_service(self._config.sql_db_url), + backend_name="SQL", + ) + + try: + # Execute case on both backends + result_a, result_b = await asyncio.gather( + exec_a.execute(case), + exec_b.execute(case), + ) + + # ── Optional: inject corruption into backend B ── + if case.corruption_type and self._config.run_corruption_cases: + await self._inject_corruption(exec_b, case, result_b) + + # Re-read backend B after corruption injection if needed + if case.corruption_type and self._config.run_corruption_cases: + if result_b.session: + result_b.session = await exec_b._session_service.get_session( + app_name=APP_NAME, user_id=USER_ID, session_id=SESSION_ID, + ) + + # ── Compare ── + report = self._compare_results(case, result_a, result_b) + + # Determine pass/fail + unallowed_diffs = [d for d in report.diffs if not d.allowed] + if case.corruption_type: + # For corruption cases, we expect diffs + report.passed = len(unallowed_diffs) > 0 + report.note = f"Corruption case '{case.corruption_type}': expected diff detected" if report.passed else \ + f"FAILED: corruption '{case.corruption_type}' was NOT detected" + else: + report.passed = len(unallowed_diffs) == 0 + report.note = "All diffs are allowed" if report.passed else \ + f"FAILED: {len(unallowed_diffs)} unexpected diffs found" + + return report + + finally: + await exec_a.close() + await exec_b.close() + + async def _inject_corruption( + self, + executor: ReplayExecutor, + case: ReplayCase, + result: BackendResult, + ) -> None: + """Inject intentional corruption into backend B for testing.""" + ct = case.corruption_type + + if ct == "duplicate_event" and executor.session and executor.session.events: + executor.session.events.append(copy.deepcopy(executor.session.events[-1])) + await executor._session_service.update_session(executor.session) + + elif ct == "dirty_state" and executor.session: + executor.session.state["config_key"] = "corrupted_value" + await executor._session_service.update_session(executor.session) + + elif ct == "missing_summary": + executor._summary_cache.pop(SESSION_ID, None) + if executor.session: + executor.session.events = [ + e for e in executor.session.events if not e.is_summary_event() + ] + await executor._session_service.update_session(executor.session) + + elif ct == "wrong_session_summary": + if SESSION_ID in executor._summary_cache: + existing = executor._summary_cache[SESSION_ID] + executor._summary_cache[SESSION_ID] = SessionSummary( + session_id="wrong-session-id-999", + summary_text=existing.summary_text, + original_event_count=existing.original_event_count, + compressed_event_count=existing.compressed_event_count, + summary_timestamp=time.time(), + ) + + def _compare_results( + self, + case: ReplayCase, + result_a: BackendResult, + result_b: BackendResult, + ) -> CaseDiffReport: + """Compare results from two backends.""" + report = CaseDiffReport( + case_id=case.case_id, + backend_a=result_a.backend_name, + backend_b=result_b.backend_name, + ) + + sid = SESSION_ID + + # Handle errors + if result_a.error: + report.diffs.append(DiffEntry( + session_id=sid, component="error", field_path="backend_a", + value_a=result_a.error, value_b="", allowed=False, + note=f"Backend A error: {result_a.error}", + )) + if result_b.error: + report.diffs.append(DiffEntry( + session_id=sid, component="error", field_path="backend_b", + value_a="", value_b=result_b.error, allowed=False, + note=f"Backend B error: {result_b.error}", + )) + if result_a.error or result_b.error: + report.passed = False + return report + + # Compare events + events_a = result_a.session.events if result_a.session else [] + events_b = result_b.session.events if result_b.session else [] + report.diffs.extend(compare_events(events_a, events_b, sid)) + + # Compare state + state_a = result_a.session.state if result_a.session else {} + state_b = result_b.session.state if result_b.session else {} + report.diffs.extend(compare_state(state_a, state_b, sid)) + + # Compare memory + report.diffs.extend(compare_memory(result_a.memory_response, result_b.memory_response, sid)) + + # Compare summary + summary_a = result_a.summary + summary_b = result_b.summary + report.diffs.extend(compare_summaries(summary_a, summary_b, sid)) + + return report + + def generate_report_json(self, output_path: str) -> None: + """Generate a JSON diff report file.""" + report_data = { + "generated_at": time.time(), + "backend_a": "InMemory", + "backend_b": "SQL", + "total_cases": len(self._all_diffs), + "cases": [], + "summary": { + "passed": 0, + "failed": 0, + "total_diffs": 0, + "unallowed_diffs": 0, + }, + } + + for case_report in self._all_diffs: + unallowed = [d for d in case_report.diffs if not d.allowed] + entry = { + "case_id": case_report.case_id, + "passed": case_report.passed, + "note": case_report.note, + "total_diffs": len(case_report.diffs), + "unallowed_diffs": len(unallowed), + "diffs": [ + { + "session_id": d.session_id, + "component": d.component, + "event_index": d.event_index, + "summary_id": d.summary_id, + "field_path": d.field_path, + "value_a": str(d.value_a)[:500] if d.value_a is not None else None, + "value_b": str(d.value_b)[:500] if d.value_b is not None else None, + "allowed": d.allowed, + "note": d.note, + } + for d in case_report.diffs + ], + } + report_data["cases"].append(entry) + if case_report.passed: + report_data["summary"]["passed"] += 1 + else: + report_data["summary"]["failed"] += 1 + report_data["summary"]["total_diffs"] += len(case_report.diffs) + report_data["summary"]["unallowed_diffs"] += len(unallowed) + + with open(output_path, "w", encoding="utf-8") as f: + json.dump(report_data, f, indent=2, ensure_ascii=False, default=str) + + logger.info("Diff report written to %s", output_path) + + +# ────────────────────────────────────────────────────────────── +# Convenience runner +# ────────────────────────────────────────────────────────────── + + +async def run_replay_harness( + cases: Optional[List[ReplayCase]] = None, + config: Optional[ReplayHarnessConfig] = None, + output_report: str = "session_memory_summary_diff_report.json", +) -> Tuple[List[CaseDiffReport], ReplayHarness]: + """Run the replay harness and generate a diff report. + + Args: + cases: List of replay cases to run. If None, runs all cases. + config: Harness configuration. + output_report: Path for the JSON diff report. + + Returns: + Tuple of (diff reports, harness instance). + """ + if cases is None: + from .replay_cases import ALL_REPLAY_CASES + cases = ALL_REPLAY_CASES + + harness = ReplayHarness(config=config) + reports = await harness.run_all_cases(cases) + harness.generate_report_json(output_report) + return reports, harness diff --git a/tests/sessions/test_replay_consistency.py b/tests/sessions/test_replay_consistency.py new file mode 100644 index 00000000..f08bcf84 --- /dev/null +++ b/tests/sessions/test_replay_consistency.py @@ -0,0 +1,383 @@ +# 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 replay consistency tests. + +Design Overview +=============== + +The replay harness validates cross-backend consistency for Session, Memory, and +Summary operations. It replays identical operation sequences against two +backends (InMemory and SQL/SQLite) and compares the normalized results. + +Normalization Strategy +----------------------- +- **Timestamps**: All float timestamps are zeroed to 0.0 (non-deterministic). +- **Auto-generated IDs**: UUID-like IDs (event.id, invocation_id, etc.) are + zeroed to "" for comparison. +- **Dict key order**: Keys are sorted for deterministic serialization. +- **None vs missing**: None values are stripped to treat them equivalent to + missing keys. +- **Whitespace**: Text content is whitespace-normalized for semantic comparison. + +Summary Comparison Strategy +---------------------------- +- **Content semantic**: Summary text is compared whitespace-normalized. +- **Metadata exact**: session_id, original_event_count, compressed_event_count + must match exactly - no fuzzy matching. +- **Summary loss**: If one backend has a summary and the other doesn't, this is + a critical error (unallowed diff). +- **Cross-session summary**: If summary.session_id differs from the expected + session, this is flagged as a critical error. +- **Summary timestamp**: Allowed diff (zeroed during normalization). + +Allowed Differences +-------------------- +- Event: id, timestamp, invocation_id +- Session: last_update_time, save_key +- Summary: summary_timestamp + +Backend Access Modes +--------------------- +- **Lightweight (default)**: InMemory vs SQLite (in-memory). No external dependencies. + Target: < 30 seconds for all cases. +- **Integration**: Set environment variables to enable Redis or external SQL: + - TRPC_REPLAY_REDIS_URL=redis://localhost:6379/0 + - TRPC_REPLAY_SQL_URL=mysql+pymysql://user:pass@localhost/db + (Redis/SQL integration is tested only if the backend imports succeed.) + +Acceptance Criteria +-------------------- +1. InMemory vs SQLite (lightweight) runs all 12 cases. +2. 10 corruption cases must 100% detect injected inconsistencies. +3. Normal cases false-positive rate <= 5%. +4. Summary loss, wrong-session summary, summary overwrite error: 100% detection. +5. Diff report locates: session_id, event_index/summary_id, field_path, values. +""" + +import asyncio +import json +import os +import time + +import pytest + +from trpc_agent_sdk.events import Event +from trpc_agent_sdk.sessions._session_summarizer import SessionSummary +from trpc_agent_sdk.types import Content +from trpc_agent_sdk.types import Part + +# ────────────────────────────────────────────────────────────── +# Fixtures +# ────────────────────────────────────────────────────────────── + + +@pytest.fixture(scope="module") +def event_loop(): + """Create a module-scoped event loop for async tests.""" + loop = asyncio.new_event_loop() + yield loop + loop.close() + + +@pytest.fixture +def lightweight_config(): + """Lightweight harness configuration (InMemory vs SQLite).""" + from tests.sessions.replay_harness import ReplayHarnessConfig + return ReplayHarnessConfig( + lightweight_only=True, + sql_db_url="sqlite:///:memory:", + run_corruption_cases=True, + ) + + +# ────────────────────────────────────────────────────────────── +# Test cases: Normal (non-corruption) cases +# ────────────────────────────────────────────────────────────── + +NORMAL_CASES = [ + "single_turn_text", + "multi_turn_text", + "tool_call_conversation", + "state_update_and_override", + "memory_write_and_read", + "memory_facts_and_prefs", + "summary_create_and_verify", + "summary_with_truncation", +] + +CORRUPTION_CASES = [ + "summary_missing_detection", + "summary_wrong_session", + "duplicate_event_detection", + "state_dirty_after_error", +] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("case_id", NORMAL_CASES) +async def test_normal_case_consistency(case_id, lightweight_config): + """Test that normal replay cases produce no unexpected diffs.""" + from tests.sessions.replay_cases import ALL_REPLAY_CASES + from tests.sessions.replay_harness import run_replay_harness + + case = next(c for c in ALL_REPLAY_CASES if c.case_id == case_id) + reports, _ = await run_replay_harness(cases=[case], config=lightweight_config, output_report="") + + report = reports[0] + unallowed_diffs = [d for d in report.diffs if not d.allowed] + + assert report.passed, ( + f"Case '{case_id}' FAILED with {len(unallowed_diffs)} unexpected diffs:\n" + + "\n".join( + f" [{d.component}] {d.field_path}: A={d.value_a}, B={d.value_b}" + for d in unallowed_diffs[:10] + ) + ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("case_id", CORRUPTION_CASES) +async def test_corruption_case_detection(case_id, lightweight_config): + """Test that corruption cases correctly detect the injected inconsistencies.""" + from tests.sessions.replay_cases import ALL_REPLAY_CASES + from tests.sessions.replay_harness import run_replay_harness + + case = next(c for c in ALL_REPLAY_CASES if c.case_id == case_id) + reports, _ = await run_replay_harness(cases=[case], config=lightweight_config, output_report="") + + report = reports[0] + unallowed_diffs = [d for d in report.diffs if not d.allowed] + + assert len(unallowed_diffs) > 0, ( + f"Corruption case '{case_id}' ({case.corruption_type}) did NOT detect any inconsistency. " + f"Expected diffs for corruption: {case.corruption_description}" + ) + + # Verify the diff is in the right component + if case.corruption_type == "missing_summary": + summary_diffs = [d for d in unallowed_diffs if d.component == "summary"] + assert len(summary_diffs) > 0, f"Expected summary diff for missing_summary, got none" + + if case.corruption_type == "wrong_session_summary": + summary_diffs = [d for d in unallowed_diffs if d.component == "summary"] + session_mismatch = [d for d in summary_diffs if "session_id" in d.field_path or "SESSION MISMATCH" in d.note] + assert len(session_mismatch) > 0, f"Expected session_id mismatch for wrong_session_summary" + + if case.corruption_type == "duplicate_event": + event_diffs = [d for d in unallowed_diffs if d.component == "events"] + assert len(event_diffs) > 0, f"Expected event diff for duplicate_event" + + if case.corruption_type == "dirty_state": + state_diffs = [d for d in unallowed_diffs if d.component == "state"] + assert len(state_diffs) > 0, f"Expected state diff for dirty_state" + + +@pytest.mark.asyncio +async def test_summary_loss_detection_specific(): + """Specifically verify summary loss detection: one side has summary, other doesn't.""" + from tests.sessions.replay_cases import CASE_SUMMARY_MISSING + from tests.sessions.replay_harness import run_replay_harness + from tests.sessions.replay_harness import ReplayHarnessConfig + + config = ReplayHarnessConfig(lightweight_only=True, sql_db_url="sqlite:///:memory:", run_corruption_cases=True) + reports, _ = await run_replay_harness(cases=[CASE_SUMMARY_MISSING], config=config, output_report="") + + report = reports[0] + summary_diffs = [d for d in report.diffs if d.component == "summary" and not d.allowed] + + # Must have at least one summary diff + assert len(summary_diffs) >= 1, "Summary loss was NOT detected" + + # Every summary diff should mention "" on one side + has_missing = any( + str(d.value_a) == "" or str(d.value_b) == "" + for d in summary_diffs + ) + assert has_missing, "Summary loss diff should show '' on one side" + + +@pytest.mark.asyncio +async def test_summary_wrong_session_detection_specific(): + """Specifically verify wrong-session summary detection.""" + from tests.sessions.replay_cases import CASE_SUMMARY_WRONG_SESSION + from tests.sessions.replay_harness import run_replay_harness + from tests.sessions.replay_harness import ReplayHarnessConfig + + config = ReplayHarnessConfig(lightweight_only=True, sql_db_url="sqlite:///:memory:", run_corruption_cases=True) + reports, _ = await run_replay_harness(cases=[CASE_SUMMARY_WRONG_SESSION], config=config, output_report="") + + report = reports[0] + summary_diffs = [d for d in report.diffs if d.component == "summary" and not d.allowed] + + assert len(summary_diffs) >= 1, "Wrong-session summary was NOT detected" + + # Check that a session_id-related diff exists + session_diffs = [ + d for d in summary_diffs + if "session_id" in d.field_path.lower() or "session" in d.note.lower() + ] + assert len(session_diffs) >= 1, f"Expected session_id mismatch, got diffs: {summary_diffs}" + + +@pytest.mark.asyncio +async def test_summary_overwrite_detection(): + """Verify summary overwrite error detection: two different summaries for same session.""" + from tests.sessions.replay_cases import APP_NAME, USER_ID, SESSION_ID + from tests.sessions.replay_harness import ( + _create_inmemory_session_service, + _create_sql_session_service, + _create_inmemory_memory_service, + _create_sql_memory_service, + ReplayExecutor, + ) + + exec_a = ReplayExecutor( + session_service=_create_inmemory_session_service(), + memory_service=_create_inmemory_memory_service(), + backend_name="InMemory", + ) + exec_b = ReplayExecutor( + session_service=_create_sql_session_service("sqlite:///:memory:"), + memory_service=_create_sql_memory_service("sqlite:///:memory:"), + backend_name="SQL", + ) + + try: + # Both create session and add events + for ex in (exec_a, exec_b): + session = await ex._session_service.create_session( + app_name=APP_NAME, user_id=USER_ID, session_id=SESSION_ID, + ) + ex._session = session + event = Event( + invocation_id="u1", author="user", + content=Content( + role="user", parts=[Part.from_text(text="Hello")] + ), + ) + await ex._session_service.append_event(session=session, event=event) + + # Backend A: summary version 1 + exec_a._summary_cache[SESSION_ID] = SessionSummary( + session_id=SESSION_ID, summary_text="Summary version 1", + original_event_count=1, compressed_event_count=1, + summary_timestamp=time.time(), + ) + # Backend B: summary version 2 (different text - overwrite) + exec_b._summary_cache[SESSION_ID] = SessionSummary( + session_id=SESSION_ID, summary_text="Summary version 2 - OVERWRITTEN", + original_event_count=1, compressed_event_count=1, + summary_timestamp=time.time(), + ) + + # Compare + from tests.sessions.replay_harness import compare_summaries + diffs = compare_summaries( + exec_a._summary_cache.get(SESSION_ID), + exec_b._summary_cache.get(SESSION_ID), + SESSION_ID, + ) + + unallowed = [d for d in diffs if not d.allowed] + assert len(unallowed) >= 1, f"Summary overwrite was NOT detected, diffs={diffs}" + + # The diff should be about summary_text + text_diffs = [d for d in unallowed if "text" in d.field_path.lower()] + assert len(text_diffs) >= 1, f"Expected summary_text diff, got: {unallowed}" + + finally: + await exec_a.close() + await exec_b.close() + + +# ────────────────────────────────────────────────────────────── +# Full integration test (runs all cases, generates report) +# ────────────────────────────────────────────────────────────── + + +@pytest.mark.asyncio +async def test_full_replay_suite_and_report(lightweight_config, tmp_path): + """Run all 12 replay cases, generate diff report, verify acceptance criteria.""" + from tests.sessions.replay_cases import ALL_REPLAY_CASES + from tests.sessions.replay_harness import run_replay_harness + + report_path = str(tmp_path / "session_memory_summary_diff_report.json") + reports, harness = await run_replay_harness( + cases=ALL_REPLAY_CASES, + config=lightweight_config, + output_report=report_path, + ) + + # Verify report file exists and is valid JSON + assert os.path.exists(report_path) + with open(report_path, "r", encoding="utf-8") as f: + report_data = json.load(f) + assert "cases" in report_data + assert len(report_data["cases"]) == 12 + + # Acceptance criteria checks + normal_cases = [r for r in reports if r.case_id in NORMAL_CASES] + corruption_cases = [r for r in reports if r.case_id in CORRUPTION_CASES] + + # 1. All normal cases must pass + for r in normal_cases: + assert r.passed, f"Normal case '{r.case_id}' FAILED: {r.note}" + + # 2. All corruption cases must detect the issue + for r in corruption_cases: + assert r.passed, f"Corruption case '{r.case_id}' did NOT detect issue: {r.note}" + + # 3. False positive rate check for normal cases + for r in normal_cases: + unallowed = [d for d in r.diffs if not d.allowed] + assert len(unallowed) == 0, ( + f"Normal case '{r.case_id}' has {len(unallowed)} unallowed diffs (false positives)" + ) + + # 4. Summary error detection must be 100% + summary_cases = [ + r for r in reports + if r.case_id in ("summary_missing_detection", "summary_wrong_session") + ] + for r in summary_cases: + summary_diffs = [d for d in r.diffs if d.component == "summary" and not d.allowed] + assert len(summary_diffs) >= 1, f"Summary error not detected for '{r.case_id}'" + + # 5. Diff report must contain field paths and values + all_case_diffs = [d for r in reports for d in r.diffs] + assert len(all_case_diffs) > 0, "No diffs at all - even allowed diffs should exist" + for d in all_case_diffs[:5]: # spot check first few + assert d.session_id, "Diff entry missing session_id" + assert d.field_path, "Diff entry missing field_path" + + print(f"\nAll 12 replay cases completed successfully.") + print(f" Normal cases passed: {len(normal_cases)}/{len(normal_cases)}") + print(f" Corruption cases detected: {len(corruption_cases)}/{len(corruption_cases)}") + print(f" Report: {report_path}") + + +@pytest.mark.asyncio +async def test_diff_report_field_accuracy(lightweight_config, tmp_path): + """Verify diff report contains session_id, event_index, field_path, and values.""" + from tests.sessions.replay_cases import CASE_STATE_UPDATE + from tests.sessions.replay_harness import run_replay_harness + + report_path = str(tmp_path / "field_accuracy_report.json") + reports, _ = await run_replay_harness( + cases=[CASE_STATE_UPDATE], + config=lightweight_config, + output_report=report_path, + ) + + with open(report_path, "r", encoding="utf-8") as f: + data = json.load(f) + + case_data = data["cases"][0] + for diff in case_data.get("diffs", []): + assert "session_id" in diff, f"Missing session_id in diff: {diff}" + assert "field_path" in diff, f"Missing field_path in diff: {diff}" + # At least one of event_index or summary_id should be set for relevant components + assert "component" in diff, f"Missing component in diff: {diff}"