From c796fd003ee0ad456e8c1f8827a8769a8b0578d3 Mon Sep 17 00:00:00 2001 From: guocfu Date: Fri, 10 Jul 2026 12:31:53 +0800 Subject: [PATCH] =?UTF-8?q?test:=20=E6=96=B0=E5=A2=9E=E5=A4=9A=E5=90=8E?= =?UTF-8?q?=E7=AB=AF=E5=9B=9E=E6=94=BE=E4=B8=80=E8=87=B4=E6=80=A7=E6=B5=8B?= =?UTF-8?q?=E8=AF=95=E6=A1=86=E6=9E=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests/sessions/replay_consistency/__init__.py | 199 +++++++ .../sessions/replay_consistency/assertions.py | 228 ++++++++ tests/sessions/replay_consistency/backends.py | 548 ++++++++++++++++++ .../replay_consistency/comparators.py | 171 ++++++ .../sessions/replay_consistency/constants.py | 126 ++++ tests/sessions/replay_consistency/fixtures.py | 27 + tests/sessions/replay_consistency/loaders.py | 61 ++ tests/sessions/replay_consistency/models.py | 26 + .../replay_consistency/normalizers.py | 223 +++++++ .../replay_cases/01_single_turn.jsonl | 3 + .../replay_cases/02_multi_turn.jsonl | 5 + .../replay_cases/03_tool_call.jsonl | 5 + .../replay_cases/04_state_update.jsonl | 5 + .../replay_cases/05_memory_store_search.jsonl | 12 + .../06_summary_event_update.jsonl | 7 + .../replay_cases/07_summary_truncation.jsonl | 6 + .../08_recovery_partial_event.jsonl | 4 + .../09_duplicate_event_replay.jsonl | 6 + .../10_branch_metadata_replay.jsonl | 5 + .../sessions/replay_consistency/reporters.py | 418 +++++++++++++ .../session_memory_summary_diff_report.json | 546 +++++++++++++++++ tests/sessions/test_replay_consistency.py | 455 +++++++++++++++ tests/sessions/test_replay_mutations.py | 292 ++++++++++ 23 files changed, 3378 insertions(+) create mode 100644 tests/sessions/replay_consistency/__init__.py create mode 100644 tests/sessions/replay_consistency/assertions.py create mode 100644 tests/sessions/replay_consistency/backends.py create mode 100644 tests/sessions/replay_consistency/comparators.py create mode 100644 tests/sessions/replay_consistency/constants.py create mode 100644 tests/sessions/replay_consistency/fixtures.py create mode 100644 tests/sessions/replay_consistency/loaders.py create mode 100644 tests/sessions/replay_consistency/models.py create mode 100644 tests/sessions/replay_consistency/normalizers.py create mode 100644 tests/sessions/replay_consistency/replay_cases/01_single_turn.jsonl create mode 100644 tests/sessions/replay_consistency/replay_cases/02_multi_turn.jsonl create mode 100644 tests/sessions/replay_consistency/replay_cases/03_tool_call.jsonl create mode 100644 tests/sessions/replay_consistency/replay_cases/04_state_update.jsonl create mode 100644 tests/sessions/replay_consistency/replay_cases/05_memory_store_search.jsonl create mode 100644 tests/sessions/replay_consistency/replay_cases/06_summary_event_update.jsonl create mode 100644 tests/sessions/replay_consistency/replay_cases/07_summary_truncation.jsonl create mode 100644 tests/sessions/replay_consistency/replay_cases/08_recovery_partial_event.jsonl create mode 100644 tests/sessions/replay_consistency/replay_cases/09_duplicate_event_replay.jsonl create mode 100644 tests/sessions/replay_consistency/replay_cases/10_branch_metadata_replay.jsonl create mode 100644 tests/sessions/replay_consistency/reporters.py create mode 100644 tests/sessions/replay_consistency/session_memory_summary_diff_report.json create mode 100644 tests/sessions/test_replay_consistency.py create mode 100644 tests/sessions/test_replay_mutations.py diff --git a/tests/sessions/replay_consistency/__init__.py b/tests/sessions/replay_consistency/__init__.py new file mode 100644 index 00000000..37ac118e --- /dev/null +++ b/tests/sessions/replay_consistency/__init__.py @@ -0,0 +1,199 @@ +# 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 consistency test framework for session and memory backends.""" + +from .constants import ALLOWED_DIFFS +from .constants import EXPECTED_REPLAY_CASE_FILES +from .constants import EXPECTED_REPLAY_CASE_NAMES +from .constants import REPLAY_CASES_DIR +from .constants import SUMMARY_TRUNCATION_ALLOWED_DIFF_PATHS +from .constants import SUMMARY_TRUNCATION_ALLOWED_DIFF_REASON +from .models import ReplayCase +from .loaders import MEMORY_REPLAY_CASES +from .loaders import REPLAY_CASES +from .loaders import load_jsonl_records +from .loaders import load_replay_cases +from .backends import BASELINE_BACKEND_NAME +from .backends import DEFAULT_BACKEND_CONFIG +from .backends import ENV_REDIS_BACKEND_NAME +from .backends import ENV_SQL_BACKEND_NAME +from .backends import MOCK_REDIS_BACKEND_NAME +from .backends import REDIS_URL_ENV +from .backends import SQLITE_BACKEND_NAME +from .backends import SQL_URL_ENV +from .backends import ReplayBackendConfig +from .backends import ReplayBackendUnavailable +from .backends import comparison_backend_names +from .backends import configured_memory_backend_names +from .backends import configured_session_backend_names +from .backends import create_sql_memory_service +from .backends import create_sql_session_service +from .backends import default_backend_matrix_enabled +from .backends import get_required_session +from .backends import make_memory_session +from .backends import resolve_backend_config +from .backends import run_memory_replay_case +from .backends import run_session_replay_case +from .fixtures import make_memory_config +from .fixtures import make_session_config +from .normalizers import all_normalized_events +from .normalizers import canonicalize +from .normalizers import compact_text +from .normalizers import event_texts +from .normalizers import make_event +from .normalizers import make_part +from .normalizers import normalize_content_parts +from .normalizers import normalize_event +from .normalizers import normalize_memory_entry +from .normalizers import normalize_memory_response +from .normalizers import normalize_part +from .normalizers import normalize_session +from .normalizers import normalize_summary_text +from .normalizers import summary_events +from .normalizers import summary_metadata +from .normalizers import summary_records_by_id +from .normalizers import summary_text +from .comparators import allowed_diff_reason +from .comparators import diff_dicts +from .comparators import diff_lists +from .comparators import diff_snapshots +from .comparators import event_at +from .comparators import extract_event_location +from .comparators import extract_summary_id +from .comparators import find_summary_id +from .comparators import is_event_diff +from .comparators import is_session_metadata_diff +from .comparators import is_state_diff +from .comparators import is_summary_diff +from .comparators import join_path +from .comparators import make_diff +from .reporters import build_case_diff_report +from .reporters import build_case_memory_report +from .reporters import build_case_session_report +from .reporters import build_diff_report +from .reporters import build_replay_diff_report +from .reporters import build_report_totals +from .reporters import build_summary_content_checks +from .reporters import build_summary_metadata_checks +from .reporters import combined_status +from .reporters import count_case_diffs +from .reporters import count_summary_check_mismatches +from .reporters import diff_status +from .reporters import iter_case_diffs +from .reporters import serialize_allowed_diffs +from .reporters import session_report_status +from .reporters import summary_comparison_records +from .reporters import summary_cache_records_by_id +from .reporters import summary_metadata_field_check +from .assertions import assert_all_diffs_allowed +from .assertions import assert_allowed_session_snapshot_variant +from .assertions import assert_memory_replay_case_snapshot +from .assertions import assert_replay_case_fixtures_load +from .assertions import assert_session_replay_case_snapshot +from .assertions import assert_summary_truncation_in_memory_snapshot +from .assertions import uses_allowed_snapshot_variant + +__all__ = [ + # Constants + "REPLAY_CASES_DIR", + "EXPECTED_REPLAY_CASE_NAMES", + "EXPECTED_REPLAY_CASE_FILES", + "SUMMARY_TRUNCATION_ALLOWED_DIFF_REASON", + "SUMMARY_TRUNCATION_ALLOWED_DIFF_PATHS", + "ALLOWED_DIFFS", + # Models + "ReplayCase", + # Loaders + "load_jsonl_records", + "load_replay_cases", + "REPLAY_CASES", + "MEMORY_REPLAY_CASES", + # Fixtures + "make_session_config", + "make_memory_config", + # Backends + "BASELINE_BACKEND_NAME", + "SQLITE_BACKEND_NAME", + "ENV_SQL_BACKEND_NAME", + "ENV_REDIS_BACKEND_NAME", + "MOCK_REDIS_BACKEND_NAME", + "DEFAULT_BACKEND_CONFIG", + "SQL_URL_ENV", + "REDIS_URL_ENV", + "ReplayBackendConfig", + "ReplayBackendUnavailable", + "comparison_backend_names", + "configured_session_backend_names", + "configured_memory_backend_names", + "default_backend_matrix_enabled", + "resolve_backend_config", + "create_sql_session_service", + "create_sql_memory_service", + "get_required_session", + "make_memory_session", + "run_session_replay_case", + "run_memory_replay_case", + # Normalizers + "make_event", + "make_part", + "normalize_session", + "normalize_event", + "normalize_content_parts", + "normalize_part", + "normalize_memory_response", + "normalize_memory_entry", + "summary_events", + "event_texts", + "all_normalized_events", + "compact_text", + "normalize_summary_text", + "canonicalize", + "summary_metadata", + "summary_records_by_id", + "summary_text", + # Comparators + "diff_snapshots", + "diff_dicts", + "diff_lists", + "join_path", + "make_diff", + "allowed_diff_reason", + "extract_event_location", + "extract_summary_id", + "event_at", + "find_summary_id", + "is_event_diff", + "is_state_diff", + "is_summary_diff", + "is_session_metadata_diff", + # Reporters + "build_replay_diff_report", + "build_case_diff_report", + "build_case_session_report", + "build_case_memory_report", + "build_report_totals", + "build_diff_report", + "build_summary_content_checks", + "build_summary_metadata_checks", + "summary_metadata_field_check", + "summary_comparison_records", + "summary_cache_records_by_id", + "session_report_status", + "combined_status", + "diff_status", + "count_summary_check_mismatches", + "count_case_diffs", + "iter_case_diffs", + "serialize_allowed_diffs", + # Assertions + "assert_all_diffs_allowed", + "assert_replay_case_fixtures_load", + "assert_session_replay_case_snapshot", + "assert_allowed_session_snapshot_variant", + "assert_summary_truncation_in_memory_snapshot", + "assert_memory_replay_case_snapshot", + "uses_allowed_snapshot_variant", +] diff --git a/tests/sessions/replay_consistency/assertions.py b/tests/sessions/replay_consistency/assertions.py new file mode 100644 index 00000000..2d5b7a1e --- /dev/null +++ b/tests/sessions/replay_consistency/assertions.py @@ -0,0 +1,228 @@ +# 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. +"""Assertion functions for replay consistency tests.""" + +from __future__ import annotations + +from typing import Any + +from .constants import ALLOWED_DIFFS +from .constants import EXPECTED_REPLAY_CASE_FILES +from .constants import EXPECTED_REPLAY_CASE_NAMES +from .constants import REPLAY_CASES_DIR +from .loaders import REPLAY_CASES +from .models import ReplayCase +from .normalizers import all_normalized_events +from .normalizers import event_texts +from .normalizers import summary_events + + +def assert_replay_case_fixtures_load() -> None: + """Assert that all replay case fixtures load correctly.""" + assert {replay_case.name for replay_case in REPLAY_CASES} == EXPECTED_REPLAY_CASE_NAMES + assert [path.name for path in sorted(REPLAY_CASES_DIR.glob("*.jsonl"))] == EXPECTED_REPLAY_CASE_FILES + + +def assert_session_replay_case_snapshot(case_name: str, snapshot: dict[str, Any]) -> None: + """Assert that a session snapshot matches expected values for a case.""" + if case_name == "single_turn": + assert event_texts(snapshot) == ["hello", "hi"] + assert snapshot["historical_events"] == [] + elif case_name == "multi_turn": + assert event_texts(snapshot) == [ + "My name is Alice.", + "Nice to meet you, Alice.", + "What is the weather in Paris?", + "Paris is sunny today.", + ] + assert snapshot["historical_events"] == [] + elif case_name == "tool_call": + events = all_normalized_events(snapshot) + assert any("function_call" in part for event in events for part in event["parts"]) + assert any("function_response" in part for event in events for part in event["parts"]) + elif case_name == "state_update": + assert snapshot["state"]["profile.name"] == "Alice" + assert snapshot["state"]["preference.color"] == "green" + elif case_name == "memory_store_search": + assert event_texts(snapshot) == [ + "Remember that Alice likes green tea.", + "Noted.", + "Alice prefers window seats when traveling.", + "Saved.", + "Alice's passport country is Canada.", + "Saved.", + "Summary: Alice is preparing a Shanghai travel profile from earlier preferences and facts.", + ] + elif case_name == "summary_generation_update": + assert_summary_generation_update_snapshot(snapshot) + elif case_name == "summary_truncation": + assert event_texts({"historical_events": snapshot["historical_events"], "events": []}) == [ + "My name is Alice.", + "Nice to meet you, Alice.", + ] + assert event_texts({"historical_events": [], "events": snapshot["events"]}) == [ + "Summary: Alice introduced herself.", + "What is my name?", + "Your name is Alice.", + ] + summary_evts = summary_events(snapshot) + assert len(summary_evts) == 1 + assert summary_evts[0]["version"] == 1 + assert summary_evts[0]["custom_metadata"]["session_id"] == snapshot["session_id"] + assert summary_evts[0]["custom_metadata"]["compressed_event_ids"] == ["event-1", "event-2"] + assert event_texts(snapshot) == [ + "My name is Alice.", + "Nice to meet you, Alice.", + "Summary: Alice introduced herself.", + "What is my name?", + "Your name is Alice.", + ] + elif case_name == "recovery_partial_event": + texts = event_texts(snapshot) + assert "stream chunk should not persist" not in texts + assert texts == ["Start a streaming answer.", "This is the completed answer."] + elif case_name == "duplicate_event_replay": + events = snapshot["events"] + assert event_texts(snapshot) == [ + "Retry this request.", + "Working on the retry.", + "Retry this request.", + "Working on the retry.", + "Final answer after retry.", + ] + assert [event["invocation_id"] for event in events] == [ + "inv-1", + "inv-2", + "inv-1", + "inv-2", + "inv-3", + ] + elif case_name == "branch_metadata_replay": + events = snapshot["events"] + assert event_texts(snapshot) == [ + "Route this task to two branches.", + "Design branch proposes the UI flow.", + "Ops branch checks deployment constraints.", + "Merged branch findings into one plan.", + ] + assert [event["branch"] for event in events] == [ + "main", + "main.design", + "main.ops", + "main", + ] + assert events[0]["custom_metadata"] == { + "attempt": 1, + "labels": ["root", "routing"], + "trace_id": "trace-branch-001", + } + assert events[1]["custom_metadata"]["score"] == {"quality": 0.91, "rank": 1} + assert events[2]["custom_metadata"]["branch_role"] == "ops" + assert events[3]["custom_metadata"]["decision"] == { + "accepted": True, + "owner": "agent", + } + assert events[3]["custom_metadata"]["merged_from"] == ["main.design", "main.ops"] + else: + raise AssertionError(f"Unexpected replay case: {case_name}") + + +def assert_all_diffs_allowed(report: list[dict[str, Any]]) -> None: + """Assert that all diffs in a report are allowed.""" + unexpected_diffs = [entry for entry in report if not entry["allowed"]] + assert unexpected_diffs == [] + + +def uses_allowed_snapshot_variant(case_name: str, backend_name: str) -> bool: + """Check if a case uses an allowed snapshot variant.""" + if case_name != "summary_truncation": + return False + return any( + rule["case_name"] == case_name and rule["backend_expected"] == backend_name + for rule in ALLOWED_DIFFS + ) + + +def assert_allowed_session_snapshot_variant(case_name: str, snapshot: dict[str, Any]) -> None: + """Assert an allowed snapshot variant.""" + if case_name == "summary_truncation": + assert_summary_truncation_in_memory_snapshot(snapshot) + return + raise AssertionError(f"Unexpected allowed divergence case: {case_name}") + + +def assert_summary_generation_update_snapshot(snapshot: dict[str, Any]) -> None: + """Assert a real generated summary snapshot.""" + all_summary_evts = summary_events(snapshot) + active_summary_evts = [event for event in snapshot["events"] if event["is_summary"]] + assert len(all_summary_evts) == 2 + assert len(active_summary_evts) == 1 + assert snapshot["summary"] is not None + summary_text = snapshot["summary"]["text"] + summary_metadata = snapshot["summary"]["metadata"] + + assert summary_text.startswith("summary(session-summary-generation-update):") + assert "Previous conversation summary:" in summary_text + assert "Audience is SDK developers and operators." in summary_text + assert "facts=3-events" in summary_text + + latest_summary_event = active_summary_evts[0] + assert latest_summary_event["author"] == "system" + assert latest_summary_event["invocation_id"] == "summary" + assert latest_summary_event["parts"][0]["text"] == ( + "Previous conversation summary: " + summary_text + ) + assert latest_summary_event["custom_metadata"] is None + assert latest_summary_event["version"] == 0 + + assert summary_metadata["session_id"] == snapshot["session_id"] + assert summary_metadata["manager_session_id"] == snapshot["session_id"] + assert summary_metadata["has_summary"] is True + assert summary_metadata["has_summary_timestamp"] is True + assert summary_metadata["summary_event_count"] == 1 + assert summary_metadata["summary_event_text"] == latest_summary_event["parts"][0]["text"] + assert summary_metadata["compressed_event_count"] == 3 + assert summary_metadata["historical_event_count"] == 5 + assert summary_metadata["original_event_count"] == 5 + assert summary_metadata["manager_compressed_event_count"] == 3 + + +def assert_summary_truncation_in_memory_snapshot(snapshot: dict[str, Any]) -> None: + """Assert the summary truncation in-memory snapshot variant.""" + assert snapshot["historical_events"] == [] + assert event_texts(snapshot) == [ + "My name is Alice.", + "Nice to meet you, Alice.", + "Summary: Alice introduced herself.", + "What is my name?", + "Your name is Alice.", + ] + summary_evts = summary_events(snapshot) + assert len(summary_evts) == 1 + assert summary_evts[0]["version"] == 1 + assert summary_evts[0]["custom_metadata"]["session_id"] == snapshot["session_id"] + assert summary_evts[0]["custom_metadata"]["compressed_event_ids"] == ["event-1", "event-2"] + + +def assert_memory_replay_case_snapshot(replay_case: ReplayCase, snapshot: dict[str, Any]) -> None: + """Assert that a memory snapshot matches expected values.""" + expected_searches = [] + for search_record in replay_case.memory_search_records: + expected_memories = search_record.get("expected_memories") + if expected_memories is None: + expected_memories = [ + {"author": "user", "text": text} + for text in search_record["expected_texts"] + ] + expected_searches.append({ + "query": search_record["query"], + "limit": search_record.get("limit", 10), + "memories": [ + {"author": memory["author"], "parts": [{"text": memory["text"]}]} + for memory in expected_memories + ], + }) + assert snapshot["searches"] == expected_searches diff --git a/tests/sessions/replay_consistency/backends.py b/tests/sessions/replay_consistency/backends.py new file mode 100644 index 00000000..11810c40 --- /dev/null +++ b/tests/sessions/replay_consistency/backends.py @@ -0,0 +1,548 @@ +# 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. +"""Backend factories and execution functions for replay consistency tests.""" + +from __future__ import annotations + +from contextlib import asynccontextmanager +from dataclasses import dataclass +from dataclasses import replace +import fnmatch +import os +import re +import time +from typing import AsyncGenerator +from typing import Any + +from trpc_agent_sdk.events import Event +from trpc_agent_sdk.memory._in_memory_memory_service import InMemoryMemoryService +from trpc_agent_sdk.memory._sql_memory_service import SqlMemoryService +from trpc_agent_sdk.models import LLMModel +from trpc_agent_sdk.models import LlmRequest +from trpc_agent_sdk.models import LlmResponse +from trpc_agent_sdk.sessions._in_memory_session_service import InMemorySessionService +from trpc_agent_sdk.sessions._session import Session +from trpc_agent_sdk.sessions._session_summarizer import SessionSummarizer +from trpc_agent_sdk.sessions._sql_session_service import SqlSessionService +from trpc_agent_sdk.sessions._summarizer_manager import SummarizerSessionManager +from trpc_agent_sdk.utils import user_key + +from .fixtures import make_memory_config +from .fixtures import make_session_config +from .models import ReplayCase +from .normalizers import make_event +from .normalizers import normalize_memory_response +from .normalizers import normalize_session + +BASELINE_BACKEND_NAME = "in_memory" +SQLITE_BACKEND_NAME = "sqlite_sql" +ENV_SQL_BACKEND_NAME = "env_sql" +ENV_REDIS_BACKEND_NAME = "env_redis" +MOCK_REDIS_BACKEND_NAME = "mock_redis" + +SQL_URL_ENV = "TRPC_AGENT_REPLAY_SQL_URL" +REDIS_URL_ENV = "TRPC_AGENT_REPLAY_REDIS_URL" + + +class ReplayBackendUnavailable(RuntimeError): + """Raised when an optional replay backend is requested but unavailable.""" + + +@dataclass(frozen=True) +class ReplayBackendConfig: + """Backend selection for replay consistency tests.""" + + sql_url: str | None = None + redis_url: str | None = None + + +DEFAULT_BACKEND_CONFIG = ReplayBackendConfig() + + +def resolve_backend_config( + backend_config: ReplayBackendConfig = DEFAULT_BACKEND_CONFIG, +) -> ReplayBackendConfig: + """Resolve optional external backend URLs from environment variables.""" + return replace( + backend_config, + sql_url=backend_config.sql_url or os.getenv(SQL_URL_ENV) or None, + redis_url=backend_config.redis_url or os.getenv(REDIS_URL_ENV) or None, + ) + + +def configured_session_backend_names( + backend_config: ReplayBackendConfig = DEFAULT_BACKEND_CONFIG, +) -> list[str]: + """Return configured session backend names.""" + backend_config = resolve_backend_config(backend_config) + names = [BASELINE_BACKEND_NAME] + if backend_config.sql_url: + names.append(ENV_SQL_BACKEND_NAME) + if backend_config.redis_url: + names.append(ENV_REDIS_BACKEND_NAME) + return names + + +def configured_memory_backend_names( + backend_config: ReplayBackendConfig = DEFAULT_BACKEND_CONFIG, +) -> list[str]: + """Return configured memory backend names.""" + backend_config = resolve_backend_config(backend_config) + names = [BASELINE_BACKEND_NAME] + if backend_config.sql_url: + names.append(ENV_SQL_BACKEND_NAME) + if backend_config.redis_url: + names.append(ENV_REDIS_BACKEND_NAME) + return names + + +def default_backend_matrix_enabled( + backend_config: ReplayBackendConfig = DEFAULT_BACKEND_CONFIG, +) -> bool: + """Return whether the deterministic default backend matrix is active.""" + return ( + backend_config == DEFAULT_BACKEND_CONFIG + and not os.getenv(SQL_URL_ENV) + and not os.getenv(REDIS_URL_ENV) + ) + + +def comparison_backend_names(snapshots: dict[str, dict[str, Any]]) -> list[str]: + """Return backend names to compare against the InMemory baseline.""" + return [name for name in snapshots if name != BASELINE_BACKEND_NAME] + + +class ReplayMockRedisStorage: + """In-memory RedisStorage replacement for replay tests.""" + + def __init__(self) -> None: + self._strings: dict[str, Any] = {} + self._hashes: dict[str, dict[str, Any]] = {} + self._lists: dict[str, list[Any]] = {} + + @asynccontextmanager + async def create_db_session(self): + yield object() + + async def execute_command(self, session: Any, command: Any) -> Any: + """Execute the Redis command subset used by replay services.""" + method = command.method.lower() + args = command.args + + if method == "set": + self._strings[args[0]] = args[1] + return True + if method == "get": + return self._strings.get(args[0]) + if method == "keys": + return sorted(self._matching_keys(args[0])) + if method == "hset": + key = args[0] + self._hashes.setdefault(key, {}) + for index in range(1, len(args), 2): + self._hashes[key][args[index]] = args[index + 1] + return True + if method == "hgetall": + return dict(self._hashes.get(args[0], {})) + if method == "rpush": + key = args[0] + self._lists.setdefault(key, []).extend(args[1:]) + return len(self._lists[key]) + if method == "lrange": + values = self._lists.get(args[0], []) + start = args[1] + stop = args[2] + end = None if stop == -1 else stop + 1 + return values[start:end] + if method == "type": + key = args[0] + if key in self._strings: + return "string" + if key in self._hashes: + return "hash" + if key in self._lists: + return "list" + return "none" + return None + + async def query(self, session: Any, pattern: str, conditions: Any) -> list[tuple[str, Any]]: + """Query matching mock Redis keys.""" + keys = sorted(self._matching_keys(pattern)) + if conditions.limit > 0: + keys = keys[:conditions.limit] + + results = [] + for key in keys: + if key in self._strings: + results.append((key, self._strings[key])) + elif key in self._hashes: + results.append((key, dict(self._hashes[key]))) + elif key in self._lists: + results.append((key, list(self._lists[key]))) + return results + + async def delete(self, session: Any, key: str, conditions: Any = None) -> None: + """Delete a mock Redis key.""" + self._strings.pop(key, None) + self._hashes.pop(key, None) + self._lists.pop(key, None) + + async def expire(self, session: Any, expire_obj: Any) -> None: + """Ignore TTL in the replay mock.""" + + async def close(self) -> None: + """Close mock storage.""" + + def _matching_keys(self, pattern: str) -> list[str]: + keys = set(self._strings) | set(self._hashes) | set(self._lists) + return [key for key in keys if fnmatch.fnmatch(key, pattern)] + + +class FakeReplayModel(LLMModel): + """Concrete test model used only to satisfy SessionSummarizer's model contract.""" + + @classmethod + def supported_models(cls) -> list[str]: + return [r"deterministic-replay-model"] + + async def _generate_async_impl( + self, + request: LlmRequest, + stream: bool = False, + ctx: Any | None = None, + ) -> AsyncGenerator[LlmResponse, None]: + if False: + yield LlmResponse() + + +class DeterministicSessionSummarizer(SessionSummarizer): + """No-LLM summarizer that still uses the real session summary flow.""" + + async def _compress_session_to_summary( + self, + events: list[Event], + session_id: str, + ctx: Any | None = None, + ) -> str | None: + fragments = [] + ordered_events = sorted( + events, + key=lambda event: ( + not event.is_summary_event(), + event.timestamp or 0, + event.invocation_id or "", + event.author or "", + event.get_text() or "", + ), + ) + for event in ordered_events: + text = (event.get_text() or "").strip() + if not text: + continue + normalized_text = re.sub(r"\s+", " ", text).strip() + fragments.append(f"{event.author or 'unknown'}={normalized_text}") + if not fragments: + return None + return f"summary({session_id}): {' | '.join(fragments)} | facts={len(events)}-events" + + +def make_summarizer_manager(keep_recent_count: int = 2) -> SummarizerSessionManager: + """Create a deterministic summarizer manager for replay summary cases.""" + model = FakeReplayModel(model_name="deterministic-replay-model") + summarizer = DeterministicSessionSummarizer( + model=model, + check_summarizer_functions=[lambda session: bool(session.events)], + keep_recent_count=keep_recent_count, + ) + return SummarizerSessionManager(model=model, summarizer=summarizer, auto_summarize=True) + + +async def create_sql_session_service( + session_config: Any, + db_url: str = "sqlite:///:memory:", +) -> SqlSessionService: + """Create a SQL session service.""" + service = SqlSessionService( + db_url=db_url, + session_config=session_config, + is_async=False, + ) + await service._sql_storage.create_sql_engine() + return service + + +async def create_sql_memory_service(memory_config: Any) -> SqlMemoryService: + """Create a SQLite in-memory SQL memory service.""" + return await create_sql_memory_service_for_url(memory_config, "sqlite:///:memory:") + + +async def create_sql_memory_service_for_url(memory_config: Any, db_url: str) -> SqlMemoryService: + """Create a SQL memory service.""" + service = SqlMemoryService(db_url=db_url, memory_service_config=memory_config, is_async=False) + await service._sql_storage.create_sql_engine() + return service + + +def create_redis_session_service(session_config: Any, db_url: str) -> Any: + """Create a Redis session service.""" + try: + from trpc_agent_sdk.sessions._redis_session_service import RedisSessionService + except ImportError as ex: + raise ReplayBackendUnavailable("Redis session dependencies are not installed") from ex + return RedisSessionService(db_url=db_url, session_config=session_config, is_async=False) + + +def create_mock_redis_session_service(session_config: Any) -> Any: + """Create a Redis session service backed by in-memory mock storage.""" + service = create_redis_session_service(session_config, "redis://mock") + service._redis_storage = ReplayMockRedisStorage() + return service + + +def create_redis_memory_service(memory_config: Any, db_url: str) -> Any: + """Create a Redis memory service.""" + try: + from trpc_agent_sdk.memory._redis_memory_service import RedisMemoryService + except ImportError as ex: + raise ReplayBackendUnavailable("Redis memory dependencies are not installed") from ex + return RedisMemoryService(db_url=db_url, memory_service_config=memory_config, is_async=False) + + +def create_mock_redis_memory_service(memory_config: Any) -> Any: + """Create a Redis memory service backed by in-memory mock storage.""" + service = create_redis_memory_service(memory_config, "redis://mock") + service._redis_storage = ReplayMockRedisStorage() + return service + + +async def create_session_services( + replay_case: ReplayCase, + backend_config: ReplayBackendConfig = DEFAULT_BACKEND_CONFIG, +) -> dict[str, Any]: + """Create configured session services for a replay case.""" + backend_config = resolve_backend_config(backend_config) + session_config = make_session_config(replay_case.session_config) + services = { + BASELINE_BACKEND_NAME: InMemorySessionService(session_config=session_config), + } + if backend_config.sql_url: + try: + services[ENV_SQL_BACKEND_NAME] = await create_sql_session_service( + make_session_config(replay_case.session_config), + db_url=backend_config.sql_url, + ) + except Exception: + services[SQLITE_BACKEND_NAME] = await create_sql_session_service( + make_session_config(replay_case.session_config), + ) + if backend_config.redis_url: + try: + services[ENV_REDIS_BACKEND_NAME] = create_redis_session_service( + make_session_config(replay_case.session_config), + backend_config.redis_url, + ) + except Exception: + services[MOCK_REDIS_BACKEND_NAME] = create_mock_redis_session_service( + make_session_config(replay_case.session_config), + ) + return services + + +async def create_memory_services( + replay_case: ReplayCase, + backend_config: ReplayBackendConfig = DEFAULT_BACKEND_CONFIG, +) -> dict[str, Any]: + """Create configured memory services for a replay case.""" + backend_config = resolve_backend_config(backend_config) + memory_config = make_memory_config(replay_case.memory_config) + services = { + BASELINE_BACKEND_NAME: InMemoryMemoryService(memory_service_config=memory_config), + } + if backend_config.sql_url: + try: + services[ENV_SQL_BACKEND_NAME] = await create_sql_memory_service_for_url( + make_memory_config(replay_case.memory_config), + backend_config.sql_url, + ) + except Exception: + services[SQLITE_BACKEND_NAME] = await create_sql_memory_service( + make_memory_config(replay_case.memory_config), + ) + if backend_config.redis_url: + try: + services[ENV_REDIS_BACKEND_NAME] = create_redis_memory_service( + make_memory_config(replay_case.memory_config), + backend_config.redis_url, + ) + except Exception: + services[MOCK_REDIS_BACKEND_NAME] = create_mock_redis_memory_service( + make_memory_config(replay_case.memory_config), + ) + return services + + +async def get_required_session(service: Any, replay_case: ReplayCase) -> Session: + """Get a session from the service, asserting it exists.""" + stored = await service.get_session( + app_name=replay_case.app_name, + user_id=replay_case.user_id, + session_id=replay_case.session_id, + ) + assert stored is not None + return stored + + +def make_memory_session(replay_case: ReplayCase) -> Session: + """Create a session for memory testing.""" + base_timestamp = 1700000000.0 + return Session( + id=replay_case.session_id, + app_name=replay_case.app_name, + user_id=replay_case.user_id, + save_key=user_key(replay_case.app_name, replay_case.user_id), + events=[make_event(event_record, base_timestamp) for event_record in replay_case.event_records], + ) + + +async def run_session_service_replay(service: Any, replay_case: ReplayCase) -> dict[str, Any]: + """Run one session replay case against a single service.""" + base_timestamp = time.time() + session = await service.create_session( + app_name=replay_case.app_name, + user_id=replay_case.user_id, + session_id=replay_case.session_id, + ) + summary_texts = [] + for event_index, event_record in enumerate(replay_case.event_records): + await service.append_event(session, make_event(event_record, base_timestamp)) + if event_index in replay_case.summary_points: + await service.create_session_summary(session) + summary_text = await service.get_session_summary(session) + assert summary_text + summary_texts.append(summary_text) + session = await get_required_session(service, replay_case) + + if len(summary_texts) > 1: + assert summary_texts[0] != summary_texts[-1] + + stored = await get_required_session(service, replay_case) + return await normalize_session(stored, service) + + +async def create_session_fallback_service(backend_name: str, replay_case: ReplayCase) -> tuple[str, Any] | None: + """Create a local fallback service for an unavailable external session backend.""" + if backend_name == ENV_SQL_BACKEND_NAME: + return ( + SQLITE_BACKEND_NAME, + await create_sql_session_service(make_session_config(replay_case.session_config)), + ) + if backend_name == ENV_REDIS_BACKEND_NAME: + return ( + MOCK_REDIS_BACKEND_NAME, + create_mock_redis_session_service(make_session_config(replay_case.session_config)), + ) + return None + + +async def run_session_replay_case( + replay_case: ReplayCase, + backend_config: ReplayBackendConfig = DEFAULT_BACKEND_CONFIG, +) -> dict[str, dict[str, Any]]: + """Run the same deterministic session trajectory against configured session backends.""" + services = await create_session_services(replay_case, backend_config) + if replay_case.summary_points: + for service in services.values(): + service.set_summarizer_manager(make_summarizer_manager(), force=True) + fallback_services = [] + try: + snapshots = {} + for backend_name, service in list(services.items()): + try: + snapshots[backend_name] = await run_session_service_replay(service, replay_case) + except Exception as ex: + fallback = await create_session_fallback_service(backend_name, replay_case) + if fallback is None: + raise + fallback_name, fallback_service = fallback + if replay_case.summary_points: + fallback_service.set_summarizer_manager(make_summarizer_manager(), force=True) + fallback_services.append(fallback_service) + try: + snapshots[fallback_name] = await run_session_service_replay(fallback_service, replay_case) + except Exception as fallback_ex: + raise ReplayBackendUnavailable( + f"Optional session backend {backend_name!r} is unavailable and " + f"fallback {fallback_name!r} failed: {fallback_ex}" + ) from ex + return snapshots + finally: + for service in list(services.values()) + fallback_services: + await service.close() + + +async def run_memory_service_replay(service: Any, replay_case: ReplayCase) -> dict[str, Any]: + """Run one memory replay case against a single service.""" + session = make_memory_session(replay_case) + await service.store_session(session) + searches = [] + for search_record in replay_case.memory_search_records: + response = await service.search_memory( + session.save_key, + search_record["query"], + limit=search_record.get("limit", 10), + ) + searches.append({ + "query": search_record["query"], + "limit": search_record.get("limit", 10), + "memories": normalize_memory_response(response), + }) + return {"searches": searches} + + +async def create_memory_fallback_service(backend_name: str, replay_case: ReplayCase) -> tuple[str, Any] | None: + """Create a local fallback service for an unavailable external memory backend.""" + if backend_name == ENV_SQL_BACKEND_NAME: + return ( + SQLITE_BACKEND_NAME, + await create_sql_memory_service(make_memory_config(replay_case.memory_config)), + ) + if backend_name == ENV_REDIS_BACKEND_NAME: + return ( + MOCK_REDIS_BACKEND_NAME, + create_mock_redis_memory_service(make_memory_config(replay_case.memory_config)), + ) + return None + + +async def run_memory_replay_case( + replay_case: ReplayCase, + backend_config: ReplayBackendConfig = DEFAULT_BACKEND_CONFIG, +) -> dict[str, dict[str, Any]]: + """Store the same deterministic session into configured memory backends and search it.""" + services = await create_memory_services(replay_case, backend_config) + fallback_services = [] + try: + snapshots = {} + for backend_name, service in list(services.items()): + try: + snapshots[backend_name] = await run_memory_service_replay(service, replay_case) + except Exception as ex: + fallback = await create_memory_fallback_service(backend_name, replay_case) + if fallback is None: + raise + fallback_name, fallback_service = fallback + fallback_services.append(fallback_service) + try: + snapshots[fallback_name] = await run_memory_service_replay(fallback_service, replay_case) + except Exception as fallback_ex: + raise ReplayBackendUnavailable( + f"Optional memory backend {backend_name!r} is unavailable and " + f"fallback {fallback_name!r} failed: {fallback_ex}" + ) from ex + return snapshots + finally: + for service in list(services.values()) + fallback_services: + await service.close() diff --git a/tests/sessions/replay_consistency/comparators.py b/tests/sessions/replay_consistency/comparators.py new file mode 100644 index 00000000..90cd1d67 --- /dev/null +++ b/tests/sessions/replay_consistency/comparators.py @@ -0,0 +1,171 @@ +# 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. +"""Comparison functions for replay consistency tests.""" + +from __future__ import annotations + +from typing import Any + +from .constants import ALLOWED_DIFFS + + +def diff_snapshots(expected: Any, actual: Any, field_path: str = "") -> list[dict[str, Any]]: + """Compare two snapshots and return differences.""" + if type(expected) is not type(actual): + return [make_diff(field_path, expected, actual)] + + if isinstance(expected, dict): + return diff_dicts(expected, actual, field_path) + + if isinstance(expected, list): + return diff_lists(expected, actual, field_path) + + if expected != actual: + return [make_diff(field_path, expected, actual)] + return [] + + +def diff_dicts(expected: dict[str, Any], actual: dict[str, Any], field_path: str) -> list[dict[str, Any]]: + """Compare two dicts and return differences.""" + diffs = [] + keys = sorted(set(expected) | set(actual)) + for key in keys: + child_path = join_path(field_path, key) + if key not in expected: + diffs.append(make_diff(child_path, "__missing__", actual[key])) + elif key not in actual: + diffs.append(make_diff(child_path, expected[key], "__missing__")) + else: + diffs.extend(diff_snapshots(expected[key], actual[key], child_path)) + return diffs + + +def diff_lists(expected: list[Any], actual: list[Any], field_path: str) -> list[dict[str, Any]]: + """Compare two lists and return differences.""" + diffs = [] + max_len = max(len(expected), len(actual)) + for index in range(max_len): + child_path = f"{field_path}[{index}]" if field_path else f"[{index}]" + if index >= len(expected): + diffs.append(make_diff(child_path, "__missing__", actual[index])) + elif index >= len(actual): + diffs.append(make_diff(child_path, expected[index], "__missing__")) + else: + diffs.extend(diff_snapshots(expected[index], actual[index], child_path)) + return diffs + + +def join_path(parent: str, child: str) -> str: + """Join two path components.""" + return f"{parent}.{child}" if parent else child + + +def make_diff(field_path: str, expected: Any, actual: Any) -> dict[str, Any]: + """Create a diff entry.""" + return { + "field_path": field_path, + "expected": expected, + "actual": actual, + } + + +def allowed_diff_reason(case_name: str, backend_expected: str, backend_actual: str, field_path: str) -> str | None: + """Check if a diff is allowed and return the reason.""" + for rule in ALLOWED_DIFFS: + if ( + rule["case_name"] == case_name + and rule["backend_expected"] == backend_expected + and rule["backend_actual"] == backend_actual + and field_path in rule["field_paths"] + ): + return rule["reason"] + return None + + +def extract_event_location(field_path: str) -> tuple[str | None, int | None]: + """Extract event collection and index from a field path.""" + for collection in ("events", "historical_events"): + prefix = f"{collection}[" + if field_path.startswith(prefix): + suffix = field_path[len(prefix):] + index_text = suffix.split("]", maxsplit=1)[0] + if index_text.isdigit(): + return collection, int(index_text) + return None, None + + +def extract_summary_id( + diff: dict[str, Any], + expected_snapshot: dict[str, Any], + actual_snapshot: dict[str, Any], +) -> str | None: + """Extract summary ID from a diff.""" + for value in (diff["expected"], diff["actual"]): + summary_id = find_summary_id(value) + if summary_id: + return summary_id + + event_collection, event_index = extract_event_location(diff["field_path"]) + if event_collection is None or event_index is None: + return None + + for snapshot in (expected_snapshot, actual_snapshot): + summary_id = find_summary_id(event_at(snapshot, event_collection, event_index)) + if summary_id: + return summary_id + return None + + +def event_at(snapshot: dict[str, Any], event_collection: str, event_index: int) -> dict[str, Any] | None: + """Get an event at a specific index from a snapshot.""" + events = snapshot.get(event_collection, []) + if not isinstance(events, list) or event_index >= len(events): + return None + event = events[event_index] + return event if isinstance(event, dict) else None + + +def find_summary_id(value: Any) -> str | None: + """Recursively find a summary ID in a value.""" + if isinstance(value, dict): + summary_id = value.get("summary_id") + if isinstance(summary_id, str): + return summary_id + custom_metadata = value.get("custom_metadata") + if isinstance(custom_metadata, dict): + summary_id = custom_metadata.get("summary_id") + if isinstance(summary_id, str): + return summary_id + for child in value.values(): + summary_id = find_summary_id(child) + if summary_id: + return summary_id + if isinstance(value, list): + for item in value: + summary_id = find_summary_id(item) + if summary_id: + return summary_id + return None + + +def is_event_diff(diff: dict[str, Any]) -> bool: + """Check if a diff is an event diff.""" + return diff["event_collection"] is not None + + +def is_state_diff(diff: dict[str, Any]) -> bool: + """Check if a diff is a state diff.""" + return diff["field_path"] == "state" or diff["field_path"].startswith("state.") + + +def is_summary_diff(diff: dict[str, Any]) -> bool: + """Check if a diff is a summary diff.""" + return diff["summary_id"] is not None + + +def is_session_metadata_diff(diff: dict[str, Any]) -> bool: + """Check if a diff is a session metadata diff.""" + return not is_event_diff(diff) and not is_state_diff(diff) and not is_summary_diff(diff) diff --git a/tests/sessions/replay_consistency/constants.py b/tests/sessions/replay_consistency/constants.py new file mode 100644 index 00000000..e2c29696 --- /dev/null +++ b/tests/sessions/replay_consistency/constants.py @@ -0,0 +1,126 @@ +# 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. +"""Constants for replay consistency tests.""" + +from pathlib import Path + +REPLAY_CASES_DIR = Path(__file__).parent / "replay_cases" + +EXPECTED_REPLAY_CASE_NAMES = { + "single_turn", + "multi_turn", + "tool_call", + "state_update", + "memory_store_search", + "summary_generation_update", + "summary_truncation", + "recovery_partial_event", + "duplicate_event_replay", + "branch_metadata_replay", +} + +EXPECTED_REPLAY_CASE_FILES = [ + "01_single_turn.jsonl", + "02_multi_turn.jsonl", + "03_tool_call.jsonl", + "04_state_update.jsonl", + "05_memory_store_search.jsonl", + "06_summary_event_update.jsonl", + "07_summary_truncation.jsonl", + "08_recovery_partial_event.jsonl", + "09_duplicate_event_replay.jsonl", + "10_branch_metadata_replay.jsonl", +] + +SUMMARY_TRUNCATION_ALLOWED_DIFF_REASON = ( + "InMemorySessionService keeps max_events-trimmed events in active storage instead of " + "moving them to historical_events; SQLite stores those filtered events in historical_events." +) + +SUMMARY_TRUNCATION_ALLOWED_DIFF_PATHS = { + "events[0].author", + "events[0].custom_metadata", + "events[0].invocation_id", + "events[0].is_summary", + "events[0].model_flags", + "events[0].parts[0].text", + "events[0].version", + "events[1].author", + "events[1].invocation_id", + "events[1].parts[0].text", + "events[2].author", + "events[2].custom_metadata", + "events[2].invocation_id", + "events[2].is_summary", + "events[2].model_flags", + "events[2].parts[0].text", + "events[2].version", + "events[3]", + "events[4]", + "historical_events[0]", + "historical_events[1]", +} + +SUMMARY_GENERATION_UPDATE_ALLOWED_DIFF_REASON = ( + "After real summary compression, InMemory keeps the generated summary event before " + "recent active events while SQLite reloads active and historical events in storage " + "order. Summary cache content and strict summary metadata must still match." +) + +SUMMARY_GENERATION_UPDATE_ALLOWED_DIFF_PATHS = { + "events[0].author", + "events[0].invocation_id", + "events[0].is_summary", + "events[0].model_flags", + "events[0].partial", + "events[0].parts[0].text", + "events[1].author", + "events[1].invocation_id", + "events[1].parts[0].text", + "events[2].author", + "events[2].invocation_id", + "events[2].is_summary", + "events[2].model_flags", + "events[2].partial", + "events[2].parts[0].text", + "historical_events[2].author", + "historical_events[2].invocation_id", + "historical_events[2].is_summary", + "historical_events[2].model_flags", + "historical_events[2].partial", + "historical_events[2].parts[0].text", + "historical_events[3].author", + "historical_events[3].invocation_id", + "historical_events[3].parts[0].text", + "historical_events[4].author", + "historical_events[4].invocation_id", + "historical_events[4].is_summary", + "historical_events[4].model_flags", + "historical_events[4].partial", + "historical_events[4].parts[0].text", +} + +PERSISTENT_BACKEND_NAMES = ("sqlite_sql", "mock_redis", "env_sql", "env_redis") + +ALLOWED_DIFFS = [ + { + "case_name": "summary_generation_update", + "backend_expected": "in_memory", + "backend_actual": backend_name, + "field_paths": SUMMARY_GENERATION_UPDATE_ALLOWED_DIFF_PATHS, + "reason": SUMMARY_GENERATION_UPDATE_ALLOWED_DIFF_REASON, + } + for backend_name in PERSISTENT_BACKEND_NAMES +] + [ + { + "case_name": "summary_truncation", + "backend_expected": "in_memory", + "backend_actual": backend_name, + "field_paths": SUMMARY_TRUNCATION_ALLOWED_DIFF_PATHS, + "reason": SUMMARY_TRUNCATION_ALLOWED_DIFF_REASON, + } + for backend_name in PERSISTENT_BACKEND_NAMES +] diff --git a/tests/sessions/replay_consistency/fixtures.py b/tests/sessions/replay_consistency/fixtures.py new file mode 100644 index 00000000..b7d41ce1 --- /dev/null +++ b/tests/sessions/replay_consistency/fixtures.py @@ -0,0 +1,27 @@ +# 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. +"""Configuration fixtures for replay consistency tests.""" + +from __future__ import annotations + +from typing import Any + +from trpc_agent_sdk.abc import MemoryServiceConfig +from trpc_agent_sdk.sessions._types import SessionServiceConfig + + +def make_session_config(config_data: dict[str, Any] | None = None) -> SessionServiceConfig: + """Create a SessionServiceConfig from dict.""" + config = SessionServiceConfig(**(config_data or {})) + config.clean_ttl_config() + return config + + +def make_memory_config(config_data: dict[str, Any] | None = None) -> MemoryServiceConfig: + """Create a MemoryServiceConfig from dict.""" + config = MemoryServiceConfig(**(config_data or {})) + config.clean_ttl_config() + return config diff --git a/tests/sessions/replay_consistency/loaders.py b/tests/sessions/replay_consistency/loaders.py new file mode 100644 index 00000000..af47b614 --- /dev/null +++ b/tests/sessions/replay_consistency/loaders.py @@ -0,0 +1,61 @@ +# Tencent is pleased to support the open source community by making tRPC-Agent-Python available. +# +# Copyright (C) 2026 Tencent. All rights reserved. +# +# tRPC-Agent-Python is licensed under Apache-2.0. +"""Loaders for replay consistency test cases.""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any + +from .constants import REPLAY_CASES_DIR +from .models import ReplayCase + + +def load_jsonl_records(path: Path) -> list[dict[str, Any]]: + """Load records from a JSONL file.""" + return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line.strip()] + + +def load_replay_cases() -> list[ReplayCase]: + """Load all replay cases from JSONL files.""" + cases = [] + for path in sorted(REPLAY_CASES_DIR.glob("*.jsonl")): + records = load_jsonl_records(path) + if not records: + raise ValueError(f"Replay case file is empty: {path}") + + case_record = records[0] + if case_record.get("record_type") != "case": + raise ValueError(f"First record must be a case record: {path}") + + event_records = [] + memory_search_records = [] + for index, record in enumerate(records[1:], start=2): + record_type = record.get("record_type") + if record_type == "event": + event_records.append(record) + elif record_type == "memory_search": + memory_search_records.append(record) + else: + raise ValueError(f"Unsupported record type at line {index}: {path}") + + cases.append(ReplayCase( + name=case_record["name"], + app_name=case_record["app_name"], + user_id=case_record["user_id"], + session_id=case_record["session_id"], + session_config=case_record.get("session_config", {}), + memory_config=case_record.get("memory_config", {}), + summary_points=case_record.get("summary_points", []), + event_records=event_records, + memory_search_records=memory_search_records, + )) + return cases + + +REPLAY_CASES = load_replay_cases() +MEMORY_REPLAY_CASES = [replay_case for replay_case in REPLAY_CASES if replay_case.memory_search_records] diff --git a/tests/sessions/replay_consistency/models.py b/tests/sessions/replay_consistency/models.py new file mode 100644 index 00000000..a6c3dd9c --- /dev/null +++ b/tests/sessions/replay_consistency/models.py @@ -0,0 +1,26 @@ +# 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. +"""Data models for replay consistency tests.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + + +@dataclass(frozen=True) +class ReplayCase: + """JSONL-driven replay case for deterministic backend comparison.""" + + name: str + app_name: str + user_id: str + session_id: str + session_config: dict[str, Any] + memory_config: dict[str, Any] + summary_points: list[int] + event_records: list[dict[str, Any]] + memory_search_records: list[dict[str, Any]] diff --git a/tests/sessions/replay_consistency/normalizers.py b/tests/sessions/replay_consistency/normalizers.py new file mode 100644 index 00000000..be4195ab --- /dev/null +++ b/tests/sessions/replay_consistency/normalizers.py @@ -0,0 +1,223 @@ +# 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. +"""Normalization functions for replay consistency tests.""" + +from __future__ import annotations + +import json +from typing import Any + +from trpc_agent_sdk.events import Event +from trpc_agent_sdk.events._event import _EVENT_FLAG_SUMMARY +from trpc_agent_sdk.sessions._session import Session +from trpc_agent_sdk.types import Content +from trpc_agent_sdk.types import EventActions +from trpc_agent_sdk.types import FunctionCall +from trpc_agent_sdk.types import FunctionResponse +from trpc_agent_sdk.types import MemoryEntry +from trpc_agent_sdk.types import Part +from trpc_agent_sdk.types import SearchMemoryResponse + + +def make_event(record: dict[str, Any], base_timestamp: float) -> Event: + """Create an Event from a replay record.""" + model_flags = 0 + if record.get("is_summary"): + model_flags |= _EVENT_FLAG_SUMMARY + return Event( + id=record["id"], + invocation_id=record["invocation_id"], + author=record["author"], + content=Content(parts=[make_part(part) for part in record.get("parts", [])]), + actions=EventActions(**record.get("actions", {})), + branch=record.get("branch"), + timestamp=base_timestamp + (record.get("timestamp_offset_ms", 0) / 1000.0), + partial=record.get("partial", False), + version=record.get("version", 0), + custom_metadata=record.get("custom_metadata"), + model_flags=model_flags, + ) + + +def make_part(part_record: dict[str, Any]) -> Part: + """Create a Part from a replay part record.""" + if "text" in part_record: + return Part.from_text(text=part_record["text"]) + if "function_call" in part_record: + return Part(function_call=FunctionCall.model_validate(part_record["function_call"])) + if "function_response" in part_record: + return Part(function_response=FunctionResponse.model_validate(part_record["function_response"])) + raise ValueError(f"Unsupported replay part: {part_record}") + + +async def normalize_session(session: Session, service: Any) -> dict[str, Any]: + """Normalize a session for comparison.""" + return { + "session_id": session.id, + "app_name": session.app_name, + "user_id": session.user_id, + "state": session.state, + "conversation_count": session.conversation_count, + "events": [normalize_event(event) for event in session.events], + "historical_events": [normalize_event(event) for event in session.historical_events], + "summary": await normalize_session_summary(session, service), + } + + +async def normalize_session_summary(session: Session, service: Any) -> dict[str, Any] | None: + """Normalize a session summary for comparison.""" + summary_text = await service.get_session_summary(session) + if summary_text is None: + return None + + summary_events_list = [event for event in session.events if event.is_summary_event()] + summary_event_text = summary_events_list[-1].get_text() if summary_events_list else None + metadata = { + "session_id": session.id, + "has_summary": True, + "summary_event_count": len(summary_events_list), + "summary_event_text": compact_text(summary_event_text), + "compressed_event_count": len(session.events), + "historical_event_count": len(session.historical_events), + } + + manager = service.summarizer_manager + if manager is not None: + manager_summary = await manager.get_session_summary(session) + if manager_summary is not None: + metadata.update({ + "manager_session_id": manager_summary.session_id, + "original_event_count": manager_summary.original_event_count, + "manager_compressed_event_count": manager_summary.compressed_event_count, + "has_summary_timestamp": bool(manager_summary.summary_timestamp), + }) + + return { + "text": compact_text(summary_text), + "metadata": canonicalize(metadata), + } + + +def normalize_event(event: Event) -> dict[str, Any]: + """Normalize an event for comparison.""" + return { + "invocation_id": event.invocation_id, + "author": event.author, + "branch": event.branch, + "partial": event.partial, + "model_flags": event.model_flags, + "version": event.version, + "is_summary": event.is_summary_event(), + "actions": event.actions.model_dump(exclude_none=True, mode="json"), + "custom_metadata": event.custom_metadata, + "parts": normalize_content_parts(event), + } + + +def normalize_content_parts(event: Event) -> list[dict[str, Any]]: + """Normalize content parts from an event.""" + if not event.content or not event.content.parts: + return [] + return [normalize_part(part) for part in event.content.parts] + + +def normalize_part(part: Part) -> dict[str, Any]: + """Normalize a part for comparison.""" + normalized_part = {} + if part.text is not None: + normalized_part["text"] = part.text + if part.function_call is not None: + normalized_part["function_call"] = part.function_call.model_dump(exclude_none=True, mode="json") + if part.function_response is not None: + normalized_part["function_response"] = part.function_response.model_dump(exclude_none=True, mode="json") + return normalized_part + + +def normalize_memory_response(response: SearchMemoryResponse) -> list[dict[str, Any]]: + """Normalize a memory search response for comparison.""" + memories = [normalize_memory_entry(memory) for memory in response.memories] + return sorted(memories, key=lambda memory: (memory["author"] or "", json.dumps(memory["parts"], sort_keys=True))) + + +def normalize_memory_entry(memory: MemoryEntry) -> dict[str, Any]: + """Normalize a memory entry for comparison.""" + return { + "author": memory.author, + "parts": [normalize_part(part) for part in memory.content.parts or []], + } + + +def summary_events(snapshot: dict[str, Any]) -> list[dict[str, Any]]: + """Extract summary events from a snapshot.""" + return [event for event in all_normalized_events(snapshot) if event["is_summary"]] + + +def event_texts(snapshot: dict[str, Any]) -> list[str]: + """Extract event texts from a snapshot.""" + return [part["text"] for event in all_normalized_events(snapshot) for part in event["parts"] if "text" in part] + + +def all_normalized_events(snapshot: dict[str, Any]) -> list[dict[str, Any]]: + """Get all normalized events from a snapshot.""" + return snapshot["historical_events"] + snapshot["events"] + + +def compact_text(text: str | None) -> str | None: + """Compact whitespace in text.""" + if text is None: + return None + return " ".join(text.split()) + + +def normalize_summary_text(text: str | None) -> str | None: + """Normalize summary text for comparison.""" + if text is None: + return None + return compact_text(text).casefold() + + +def canonicalize(value: Any) -> Any: + """Canonicalize a value for comparison (sort dicts, etc.).""" + if isinstance(value, dict): + return {key: canonicalize(value[key]) for key in sorted(value)} + if isinstance(value, list): + return [canonicalize(item) for item in value] + return value + + +def summary_metadata(event: dict[str, Any]) -> dict[str, Any]: + """Extract summary metadata from an event.""" + custom_metadata = event.get("custom_metadata") or {} + return { + "summary_id": custom_metadata.get("summary_id"), + "session_id": custom_metadata.get("session_id"), + "event_version": event.get("version"), + "summary_version": custom_metadata.get("summary_version"), + "supersedes": custom_metadata.get("supersedes"), + "source_event_ids": custom_metadata.get("source_event_ids"), + "compressed_event_ids": custom_metadata.get("compressed_event_ids"), + "updated_at_offset_ms": custom_metadata.get("updated_at_offset_ms"), + } + + +def summary_records_by_id(snapshot: dict[str, Any]) -> dict[str, dict[str, Any]]: + """Get summary records from events by ID.""" + records = {} + for index, event in enumerate(all_normalized_events(snapshot)): + if not event.get("is_summary"): + continue + metadata = summary_metadata(event) + summary_id = metadata.get("summary_id") or f"summary_index:{index}" + records[summary_id] = { + "text": summary_text(event), + "metadata": metadata, + } + return records + + +def summary_text(event: dict[str, Any]) -> str: + """Extract text from a summary event.""" + return "".join(part["text"] for part in event.get("parts", []) if "text" in part) diff --git a/tests/sessions/replay_consistency/replay_cases/01_single_turn.jsonl b/tests/sessions/replay_consistency/replay_cases/01_single_turn.jsonl new file mode 100644 index 00000000..b9a34719 --- /dev/null +++ b/tests/sessions/replay_consistency/replay_cases/01_single_turn.jsonl @@ -0,0 +1,3 @@ +{"record_type":"case","name":"single_turn","app_name":"replay_app","user_id":"user-1","session_id":"session-1","session_config":{"store_historical_events":true}} +{"record_type":"event","id":"event-1","invocation_id":"inv-1","author":"user","timestamp_offset_ms":0,"parts":[{"text":"hello"}]} +{"record_type":"event","id":"event-2","invocation_id":"inv-2","author":"agent","timestamp_offset_ms":1,"parts":[{"text":"hi"}]} diff --git a/tests/sessions/replay_consistency/replay_cases/02_multi_turn.jsonl b/tests/sessions/replay_consistency/replay_cases/02_multi_turn.jsonl new file mode 100644 index 00000000..efaa1cc1 --- /dev/null +++ b/tests/sessions/replay_consistency/replay_cases/02_multi_turn.jsonl @@ -0,0 +1,5 @@ +{"record_type":"case","name":"multi_turn","app_name":"replay_app","user_id":"user-1","session_id":"session-multi-turn","session_config":{"store_historical_events":true}} +{"record_type":"event","id":"event-1","invocation_id":"inv-1","author":"user","timestamp_offset_ms":0,"parts":[{"text":"My name is Alice."}]} +{"record_type":"event","id":"event-2","invocation_id":"inv-2","author":"agent","timestamp_offset_ms":1,"parts":[{"text":"Nice to meet you, Alice."}]} +{"record_type":"event","id":"event-3","invocation_id":"inv-3","author":"user","timestamp_offset_ms":2,"parts":[{"text":"What is the weather in Paris?"}]} +{"record_type":"event","id":"event-4","invocation_id":"inv-4","author":"agent","timestamp_offset_ms":3,"parts":[{"text":"Paris is sunny today."}]} diff --git a/tests/sessions/replay_consistency/replay_cases/03_tool_call.jsonl b/tests/sessions/replay_consistency/replay_cases/03_tool_call.jsonl new file mode 100644 index 00000000..0981a543 --- /dev/null +++ b/tests/sessions/replay_consistency/replay_cases/03_tool_call.jsonl @@ -0,0 +1,5 @@ +{"record_type":"case","name":"tool_call","app_name":"replay_app","user_id":"user-1","session_id":"session-tool-call","session_config":{"store_historical_events":true}} +{"record_type":"event","id":"event-1","invocation_id":"inv-1","author":"user","timestamp_offset_ms":0,"parts":[{"text":"What is the weather in Paris?"}]} +{"record_type":"event","id":"event-2","invocation_id":"inv-2","author":"agent","timestamp_offset_ms":1,"parts":[{"function_call":{"name":"get_weather_report","args":{"city":"Paris"}}}]} +{"record_type":"event","id":"event-3","invocation_id":"inv-2","author":"tool","timestamp_offset_ms":2,"parts":[{"function_response":{"name":"get_weather_report","response":{"status":"success","weather":"sunny"}}}]} +{"record_type":"event","id":"event-4","invocation_id":"inv-3","author":"agent","timestamp_offset_ms":3,"parts":[{"text":"Paris is sunny today."}]} diff --git a/tests/sessions/replay_consistency/replay_cases/04_state_update.jsonl b/tests/sessions/replay_consistency/replay_cases/04_state_update.jsonl new file mode 100644 index 00000000..ac306fa9 --- /dev/null +++ b/tests/sessions/replay_consistency/replay_cases/04_state_update.jsonl @@ -0,0 +1,5 @@ +{"record_type":"case","name":"state_update","app_name":"replay_app","user_id":"user-1","session_id":"session-state-update","session_config":{"store_historical_events":true}} +{"record_type":"event","id":"event-1","invocation_id":"inv-1","author":"user","timestamp_offset_ms":0,"parts":[{"text":"Remember that my name is Alice and my favorite color is blue."}]} +{"record_type":"event","id":"event-2","invocation_id":"inv-2","author":"agent","timestamp_offset_ms":1,"actions":{"state_delta":{"profile.name":"Alice","preference.color":"blue"}},"parts":[{"text":"I saved your name and favorite color."}]} +{"record_type":"event","id":"event-3","invocation_id":"inv-3","author":"user","timestamp_offset_ms":2,"parts":[{"text":"Actually my favorite color is green."}]} +{"record_type":"event","id":"event-4","invocation_id":"inv-4","author":"agent","timestamp_offset_ms":3,"actions":{"state_delta":{"preference.color":"green"}},"parts":[{"text":"Updated your favorite color to green."}]} diff --git a/tests/sessions/replay_consistency/replay_cases/05_memory_store_search.jsonl b/tests/sessions/replay_consistency/replay_cases/05_memory_store_search.jsonl new file mode 100644 index 00000000..cf182dfa --- /dev/null +++ b/tests/sessions/replay_consistency/replay_cases/05_memory_store_search.jsonl @@ -0,0 +1,12 @@ +{"record_type":"case","name":"memory_store_search","app_name":"replay_app","user_id":"user-1","session_id":"session-memory-store-search","session_config":{"store_historical_events":true},"memory_config":{"enabled":true}} +{"record_type":"event","id":"event-1","invocation_id":"inv-1","author":"user","timestamp_offset_ms":0,"parts":[{"text":"Remember that Alice likes green tea."}]} +{"record_type":"event","id":"event-2","invocation_id":"inv-2","author":"agent","timestamp_offset_ms":1,"parts":[{"text":"Noted."}]} +{"record_type":"event","id":"event-3","invocation_id":"inv-3","author":"user","timestamp_offset_ms":2,"parts":[{"text":"Alice prefers window seats when traveling."}]} +{"record_type":"event","id":"event-4","invocation_id":"inv-4","author":"agent","timestamp_offset_ms":3,"parts":[{"text":"Saved."}]} +{"record_type":"event","id":"event-5","invocation_id":"inv-5","author":"user","timestamp_offset_ms":4,"parts":[{"text":"Alice's passport country is Canada."}]} +{"record_type":"event","id":"event-6","invocation_id":"inv-6","author":"agent","timestamp_offset_ms":5,"parts":[{"text":"Saved."}]} +{"record_type":"event","id":"event-7","invocation_id":"inv-summary","author":"system","timestamp_offset_ms":6,"is_summary":true,"version":1,"custom_metadata":{"summary_id":"summary-session-memory-store-search-v1","session_id":"session-memory-store-search","summary_version":1,"updated_at_offset_ms":6,"source_event_ids":["event-1","event-3","event-5"]},"parts":[{"text":"Summary: Alice is preparing a Shanghai travel profile from earlier preferences and facts."}]} +{"record_type":"memory_search","query":"green","limit":10,"expected_texts":["Remember that Alice likes green tea."]} +{"record_type":"memory_search","query":"window","limit":10,"expected_texts":["Alice prefers window seats when traveling."]} +{"record_type":"memory_search","query":"passport","limit":10,"expected_texts":["Alice's passport country is Canada."]} +{"record_type":"memory_search","query":"Shanghai","limit":10,"expected_memories":[{"author":"system","text":"Summary: Alice is preparing a Shanghai travel profile from earlier preferences and facts."}]} diff --git a/tests/sessions/replay_consistency/replay_cases/06_summary_event_update.jsonl b/tests/sessions/replay_consistency/replay_cases/06_summary_event_update.jsonl new file mode 100644 index 00000000..d0423af5 --- /dev/null +++ b/tests/sessions/replay_consistency/replay_cases/06_summary_event_update.jsonl @@ -0,0 +1,7 @@ +{"record_type":"case","name":"summary_generation_update","app_name":"replay_app","user_id":"user-1","session_id":"session-summary-generation-update","session_config":{"store_historical_events":true},"summary_points":[3,5]} +{"record_type":"event","id":"event-1","invocation_id":"inv-1","author":"user","timestamp_offset_ms":0,"parts":[{"text":"Plan a replay launch for developers."}]} +{"record_type":"event","id":"event-2","invocation_id":"inv-2","author":"agent","timestamp_offset_ms":1,"parts":[{"text":"We need goals, audience, and timing."}]} +{"record_type":"event","id":"event-3","invocation_id":"inv-3","author":"user","timestamp_offset_ms":2,"parts":[{"text":"Audience is SDK developers and operators."}]} +{"record_type":"event","id":"event-4","invocation_id":"inv-4","author":"agent","timestamp_offset_ms":3,"parts":[{"text":"The first summary should include goals and audience."}]} +{"record_type":"event","id":"event-5","invocation_id":"inv-5","author":"user","timestamp_offset_ms":4,"parts":[{"text":"Add a release checklist and owner names."}]} +{"record_type":"event","id":"event-6","invocation_id":"inv-6","author":"agent","timestamp_offset_ms":5,"parts":[{"text":"Checklist and owners are now included."}]} diff --git a/tests/sessions/replay_consistency/replay_cases/07_summary_truncation.jsonl b/tests/sessions/replay_consistency/replay_cases/07_summary_truncation.jsonl new file mode 100644 index 00000000..af01a169 --- /dev/null +++ b/tests/sessions/replay_consistency/replay_cases/07_summary_truncation.jsonl @@ -0,0 +1,6 @@ +{"record_type":"case","name":"summary_truncation","app_name":"replay_app","user_id":"user-1","session_id":"session-summary-truncation","session_config":{"store_historical_events":true,"max_events":3}} +{"record_type":"event","id":"event-1","invocation_id":"inv-1","author":"user","timestamp_offset_ms":0,"parts":[{"text":"My name is Alice."}]} +{"record_type":"event","id":"event-2","invocation_id":"inv-2","author":"agent","timestamp_offset_ms":1,"parts":[{"text":"Nice to meet you, Alice."}]} +{"record_type":"event","id":"event-3","invocation_id":"inv-summary","author":"system","timestamp_offset_ms":2,"is_summary":true,"version":1,"custom_metadata":{"summary_id":"summary-session-summary-truncation-v1","session_id":"session-summary-truncation","summary_version":1,"updated_at_offset_ms":2,"compressed_event_ids":["event-1","event-2"]},"parts":[{"text":"Summary: Alice introduced herself."}]} +{"record_type":"event","id":"event-4","invocation_id":"inv-3","author":"user","timestamp_offset_ms":3,"parts":[{"text":"What is my name?"}]} +{"record_type":"event","id":"event-5","invocation_id":"inv-4","author":"agent","timestamp_offset_ms":4,"parts":[{"text":"Your name is Alice."}]} diff --git a/tests/sessions/replay_consistency/replay_cases/08_recovery_partial_event.jsonl b/tests/sessions/replay_consistency/replay_cases/08_recovery_partial_event.jsonl new file mode 100644 index 00000000..40c94df4 --- /dev/null +++ b/tests/sessions/replay_consistency/replay_cases/08_recovery_partial_event.jsonl @@ -0,0 +1,4 @@ +{"record_type":"case","name":"recovery_partial_event","app_name":"replay_app","user_id":"user-1","session_id":"session-recovery-partial","session_config":{"store_historical_events":true}} +{"record_type":"event","id":"event-1","invocation_id":"inv-1","author":"user","timestamp_offset_ms":0,"parts":[{"text":"Start a streaming answer."}]} +{"record_type":"event","id":"event-2","invocation_id":"inv-2","author":"agent","timestamp_offset_ms":1,"partial":true,"parts":[{"text":"stream chunk should not persist"}]} +{"record_type":"event","id":"event-3","invocation_id":"inv-2","author":"agent","timestamp_offset_ms":2,"parts":[{"text":"This is the completed answer."}]} diff --git a/tests/sessions/replay_consistency/replay_cases/09_duplicate_event_replay.jsonl b/tests/sessions/replay_consistency/replay_cases/09_duplicate_event_replay.jsonl new file mode 100644 index 00000000..a5d1ab6a --- /dev/null +++ b/tests/sessions/replay_consistency/replay_cases/09_duplicate_event_replay.jsonl @@ -0,0 +1,6 @@ +{"record_type":"case","name":"duplicate_event_replay","app_name":"replay_app","user_id":"user-1","session_id":"session-duplicate-event-replay","session_config":{"store_historical_events":true}} +{"record_type":"event","id":"event-1","invocation_id":"inv-1","author":"user","timestamp_offset_ms":0,"parts":[{"text":"Retry this request."}]} +{"record_type":"event","id":"event-2","invocation_id":"inv-2","author":"agent","timestamp_offset_ms":1,"parts":[{"text":"Working on the retry."}]} +{"record_type":"event","id":"event-3","invocation_id":"inv-1","author":"user","timestamp_offset_ms":2,"parts":[{"text":"Retry this request."}]} +{"record_type":"event","id":"event-4","invocation_id":"inv-2","author":"agent","timestamp_offset_ms":3,"parts":[{"text":"Working on the retry."}]} +{"record_type":"event","id":"event-5","invocation_id":"inv-3","author":"agent","timestamp_offset_ms":4,"parts":[{"text":"Final answer after retry."}]} diff --git a/tests/sessions/replay_consistency/replay_cases/10_branch_metadata_replay.jsonl b/tests/sessions/replay_consistency/replay_cases/10_branch_metadata_replay.jsonl new file mode 100644 index 00000000..fda99444 --- /dev/null +++ b/tests/sessions/replay_consistency/replay_cases/10_branch_metadata_replay.jsonl @@ -0,0 +1,5 @@ +{"record_type":"case","name":"branch_metadata_replay","app_name":"replay_app","user_id":"user-1","session_id":"session-branch-metadata-replay","session_config":{"store_historical_events":true}} +{"record_type":"event","id":"event-1","invocation_id":"inv-root","author":"user","branch":"main","timestamp_offset_ms":0,"custom_metadata":{"trace_id":"trace-branch-001","labels":["root","routing"],"attempt":1},"parts":[{"text":"Route this task to two branches."}]} +{"record_type":"event","id":"event-2","invocation_id":"inv-design","author":"agent","branch":"main.design","timestamp_offset_ms":1,"custom_metadata":{"trace_id":"trace-branch-001","branch_role":"design","score":{"quality":0.91,"rank":1}},"parts":[{"text":"Design branch proposes the UI flow."}]} +{"record_type":"event","id":"event-3","invocation_id":"inv-ops","author":"agent","branch":"main.ops","timestamp_offset_ms":2,"custom_metadata":{"trace_id":"trace-branch-001","branch_role":"ops","score":{"quality":0.87,"rank":2}},"parts":[{"text":"Ops branch checks deployment constraints."}]} +{"record_type":"event","id":"event-4","invocation_id":"inv-merge","author":"agent","branch":"main","timestamp_offset_ms":3,"custom_metadata":{"trace_id":"trace-branch-001","merged_from":["main.design","main.ops"],"decision":{"owner":"agent","accepted":true}},"parts":[{"text":"Merged branch findings into one plan."}]} diff --git a/tests/sessions/replay_consistency/reporters.py b/tests/sessions/replay_consistency/reporters.py new file mode 100644 index 00000000..6d60be79 --- /dev/null +++ b/tests/sessions/replay_consistency/reporters.py @@ -0,0 +1,418 @@ +# 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. +"""Report building functions for replay consistency tests.""" + +from __future__ import annotations + +import asyncio +from typing import Any + +from .backends import BASELINE_BACKEND_NAME +from .backends import DEFAULT_BACKEND_CONFIG +from .backends import ReplayBackendConfig +from .backends import comparison_backend_names +from .backends import run_memory_replay_case +from .backends import run_session_replay_case +from .comparators import allowed_diff_reason +from .comparators import diff_snapshots +from .comparators import extract_event_location +from .comparators import extract_summary_id +from .comparators import is_event_diff +from .comparators import is_session_metadata_diff +from .comparators import is_state_diff +from .comparators import is_summary_diff +from .constants import ALLOWED_DIFFS +from .loaders import REPLAY_CASES +from .models import ReplayCase +from .normalizers import all_normalized_events +from .normalizers import normalize_summary_text +from .normalizers import summary_metadata + + +def build_replay_diff_report( + backend_config: ReplayBackendConfig = DEFAULT_BACKEND_CONFIG, +) -> dict[str, Any]: + """Build the complete replay diff report for all cases.""" + cases = [build_case_diff_report(replay_case, backend_config) for replay_case in REPLAY_CASES] + return { + "schema_version": 2, + "backend_pairs": { + "session": actual_report_backend_names(cases, "session"), + "memory": actual_report_backend_names(cases, "memory"), + }, + "allowed_diffs": serialize_allowed_diffs(), + "cases": cases, + "totals": build_report_totals(cases), + } + + +def build_case_diff_report( + replay_case: ReplayCase, + backend_config: ReplayBackendConfig = DEFAULT_BACKEND_CONFIG, +) -> dict[str, Any]: + """Build diff report for a single case.""" + session_report = build_case_session_report(replay_case, backend_config) + memory_report = build_case_memory_report(replay_case, backend_config) + return { + "case_name": replay_case.name, + "app_name": replay_case.app_name, + "user_id": replay_case.user_id, + "session_id": replay_case.session_id, + "status": combined_status([session_report["status"], memory_report["status"]]), + "session": session_report, + "memory": memory_report, + } + + +def build_case_session_report( + replay_case: ReplayCase, + backend_config: ReplayBackendConfig = DEFAULT_BACKEND_CONFIG, +) -> dict[str, Any]: + """Build session report for a single case.""" + snapshots = asyncio.run(run_session_replay_case(replay_case, backend_config)) + backend_names = comparison_backend_names(snapshots) + if not backend_names: + return no_comparison_session_report() + + diffs = [] + summary_content_checks = [] + summary_metadata_checks = [] + for backend_name in backend_names: + diffs.extend(build_diff_report( + case_name=replay_case.name, + session_id=replay_case.session_id, + backend_expected=BASELINE_BACKEND_NAME, + backend_actual=backend_name, + diffs=diff_snapshots(snapshots[BASELINE_BACKEND_NAME], snapshots[backend_name]), + expected_snapshot=snapshots[BASELINE_BACKEND_NAME], + actual_snapshot=snapshots[backend_name], + )) + summary_content_checks.extend( + build_summary_content_checks(snapshots[BASELINE_BACKEND_NAME], snapshots[backend_name]) + ) + summary_metadata_checks.extend( + build_summary_metadata_checks(snapshots[BASELINE_BACKEND_NAME], snapshots[backend_name]) + ) + return { + "backend_expected": BASELINE_BACKEND_NAME, + "backend_actual": backend_names[0] if len(backend_names) == 1 else backend_names, + "backend_actuals": backend_names, + "status": session_report_status(diffs, summary_content_checks, summary_metadata_checks), + "event_diffs": [diff for diff in diffs if is_event_diff(diff) and not is_summary_diff(diff)], + "state_diffs": [diff for diff in diffs if is_state_diff(diff)], + "summary_diffs": [diff for diff in diffs if is_summary_diff(diff)], + "summary_content_checks": summary_content_checks, + "summary_metadata_checks": summary_metadata_checks, + "session_metadata_diffs": [diff for diff in diffs if is_session_metadata_diff(diff)], + } + + +def no_comparison_session_report() -> dict[str, Any]: + """Build a session report for InMemory-only light mode.""" + return { + "backend_expected": BASELINE_BACKEND_NAME, + "backend_actual": None, + "backend_actuals": [], + "status": "not_applicable", + "event_diffs": [], + "state_diffs": [], + "summary_diffs": [], + "summary_content_checks": [], + "summary_metadata_checks": [], + "session_metadata_diffs": [], + } + + +def build_summary_content_checks( + expected_snapshot: dict[str, Any], + actual_snapshot: dict[str, Any], +) -> list[dict[str, Any]]: + """Build summary content comparison checks.""" + expected_records = summary_comparison_records(expected_snapshot) + actual_records = summary_comparison_records(actual_snapshot) + checks = [] + for summary_id in sorted(set(expected_records) | set(actual_records)): + expected_record = expected_records.get(summary_id) + actual_record = actual_records.get(summary_id) + expected_text = expected_record["text"] if expected_record else None + actual_text = actual_record["text"] if actual_record else None + expected_normalized = normalize_summary_text(expected_text) + actual_normalized = normalize_summary_text(actual_text) + checks.append({ + "summary_id": summary_id, + "comparison": "normalized_text", + "expected_text": expected_text, + "actual_text": actual_text, + "expected_normalized_text": expected_normalized, + "actual_normalized_text": actual_normalized, + "matched": expected_normalized == actual_normalized, + }) + return checks + + +def build_summary_metadata_checks( + expected_snapshot: dict[str, Any], + actual_snapshot: dict[str, Any], +) -> list[dict[str, Any]]: + """Build summary metadata comparison checks.""" + expected_records = summary_comparison_records(expected_snapshot) + actual_records = summary_comparison_records(actual_snapshot) + checks = [] + for summary_id in sorted(set(expected_records) | set(actual_records)): + expected_metadata = expected_records.get(summary_id, {}).get("metadata", {}) + actual_metadata = actual_records.get(summary_id, {}).get("metadata", {}) + fields = [ + summary_metadata_field_check(field, expected_metadata, actual_metadata) + for field in sorted(set(expected_metadata) | set(actual_metadata)) + ] + checks.append({ + "summary_id": summary_id, + "comparison": "strict_metadata", + "matched": all(field["matched"] for field in fields), + "fields": fields, + }) + return checks + + +def summary_metadata_field_check( + field: str, + expected_metadata: dict[str, Any], + actual_metadata: dict[str, Any], +) -> dict[str, Any]: + """Check a single summary metadata field.""" + expected = expected_metadata.get(field) + actual = actual_metadata.get(field) + return { + "field": field, + "expected": expected, + "actual": actual, + "matched": expected == actual, + } + + +def summary_comparison_records(snapshot: dict[str, Any]) -> dict[str, dict[str, Any]]: + """Get summary comparison records, preferring cache records.""" + cache_records = summary_cache_records_by_id(snapshot) + return cache_records or summary_records_by_id(snapshot) + + +def summary_cache_records_by_id(snapshot: dict[str, Any]) -> dict[str, dict[str, Any]]: + """Get summary records from cache.""" + summary = snapshot.get("summary") + if not isinstance(summary, dict): + return {} + + metadata = summary.get("metadata") or {} + session_id = metadata.get("session_id") or snapshot.get("session_id") + summary_id = f"summary:{session_id}:latest" if session_id else "summary:latest" + return { + summary_id: { + "text": summary.get("text"), + "metadata": metadata, + } + } + + +def summary_records_by_id(snapshot: dict[str, Any]) -> dict[str, dict[str, Any]]: + """Get summary records from events by ID.""" + records = {} + for index, event in enumerate(all_normalized_events(snapshot)): + if not event.get("is_summary"): + continue + metadata = summary_metadata(event) + summary_id = metadata.get("summary_id") or f"summary_index:{index}" + records[summary_id] = { + "text": summary_text(event), + "metadata": metadata, + } + return records + + +def summary_text(event: dict[str, Any]) -> str: + """Extract text from a summary event.""" + return "".join(part["text"] for part in event.get("parts", []) if "text" in part) + + +def session_report_status( + diffs: list[dict[str, Any]], + summary_content_checks: list[dict[str, Any]], + summary_metadata_checks: list[dict[str, Any]], +) -> str: + """Determine session report status.""" + if any(not check["matched"] for check in summary_content_checks): + return "unallowed_diff" + if any(not check["matched"] for check in summary_metadata_checks): + return "unallowed_diff" + return diff_status(diffs) + + +def build_case_memory_report( + replay_case: ReplayCase, + backend_config: ReplayBackendConfig = DEFAULT_BACKEND_CONFIG, +) -> dict[str, Any]: + """Build memory report for a single case.""" + if not replay_case.memory_search_records: + return { + "backend_expected": None, + "backend_actual": None, + "status": "not_applicable", + "memory_diffs": [], + } + + snapshots = asyncio.run(run_memory_replay_case(replay_case, backend_config)) + backend_names = comparison_backend_names(snapshots) + if not backend_names: + return { + "backend_expected": BASELINE_BACKEND_NAME, + "backend_actual": None, + "backend_actuals": [], + "status": "not_applicable", + "memory_diffs": [], + } + + diffs = [] + for backend_name in backend_names: + diffs.extend(build_diff_report( + case_name=replay_case.name, + session_id=replay_case.session_id, + backend_expected=BASELINE_BACKEND_NAME, + backend_actual=backend_name, + diffs=diff_snapshots(snapshots[BASELINE_BACKEND_NAME], snapshots[backend_name]), + expected_snapshot=snapshots[BASELINE_BACKEND_NAME], + actual_snapshot=snapshots[backend_name], + )) + return { + "backend_expected": BASELINE_BACKEND_NAME, + "backend_actual": backend_names[0] if len(backend_names) == 1 else backend_names, + "backend_actuals": backend_names, + "status": diff_status(diffs), + "memory_diffs": diffs, + } + + +def build_report_totals(cases: list[dict[str, Any]]) -> dict[str, Any]: + """Build report totals.""" + session_event_diffs = count_case_diffs(cases, "session", "event_diffs") + session_state_diffs = count_case_diffs(cases, "session", "state_diffs") + session_summary_diffs = count_case_diffs(cases, "session", "summary_diffs") + session_metadata_diffs = count_case_diffs(cases, "session", "session_metadata_diffs") + summary_content_mismatches = count_summary_check_mismatches(cases, "summary_content_checks") + summary_metadata_mismatches = count_summary_check_mismatches(cases, "summary_metadata_checks") + memory_diffs = count_case_diffs(cases, "memory", "memory_diffs") + all_diffs = iter_case_diffs(cases) + return { + "cases": len(cases), + "session_event_diffs": session_event_diffs, + "session_state_diffs": session_state_diffs, + "session_summary_diffs": session_summary_diffs, + "session_metadata_diffs": session_metadata_diffs, + "summary_content_mismatches": summary_content_mismatches, + "summary_metadata_mismatches": summary_metadata_mismatches, + "memory_diffs": memory_diffs, + "allowed_diffs": sum(1 for diff in all_diffs if diff["allowed"]), + "unallowed_diffs": sum(1 for diff in iter_case_diffs(cases) if not diff["allowed"]), + } + + +def actual_report_backend_names(cases: list[dict[str, Any]], section: str) -> list[str]: + """Return backend names that actually appear in report cases.""" + names = [BASELINE_BACKEND_NAME] + for case in cases: + actuals = case[section].get("backend_actuals") + if actuals is None: + actual = case[section].get("backend_actual") + actuals = [actual] if actual else [] + for backend_name in actuals: + if backend_name not in names: + names.append(backend_name) + return names + + +def count_summary_check_mismatches(cases: list[dict[str, Any]], bucket: str) -> int: + """Count summary check mismatches.""" + return sum( + 1 + for case in cases + for check in case["session"][bucket] + if not check["matched"] + ) + + +def count_case_diffs(cases: list[dict[str, Any]], section: str, bucket: str) -> int: + """Count diffs in a specific section and bucket.""" + return sum(len(case[section][bucket]) for case in cases) + + +def iter_case_diffs(cases: list[dict[str, Any]]): + """Iterate over all diffs in all cases.""" + for case in cases: + yield from case["session"]["event_diffs"] + yield from case["session"]["state_diffs"] + yield from case["session"]["summary_diffs"] + yield from case["session"]["session_metadata_diffs"] + yield from case["memory"]["memory_diffs"] + + +def combined_status(statuses: list[str]) -> str: + """Combine multiple statuses into one.""" + active_statuses = [status for status in statuses if status != "not_applicable"] + if any(status == "unallowed_diff" for status in active_statuses): + return "unallowed_diff" + if any(status == "allowed_diff" for status in active_statuses): + return "allowed_diff" + return "matched" + + +def diff_status(diffs: list[dict[str, Any]]) -> str: + """Determine status from diffs.""" + if not diffs: + return "matched" + if any(not diff["allowed"] for diff in diffs): + return "unallowed_diff" + return "allowed_diff" + + +def serialize_allowed_diffs() -> list[dict[str, Any]]: + """Serialize allowed diffs for the report.""" + return [{ + "case_name": rule["case_name"], + "backend_expected": rule["backend_expected"], + "backend_actual": rule["backend_actual"], + "field_paths": sorted(rule["field_paths"]), + "reason": rule["reason"], + } for rule in ALLOWED_DIFFS] + + +def build_diff_report( + *, + case_name: str, + session_id: str, + backend_expected: str, + backend_actual: str, + diffs: list[dict[str, Any]], + expected_snapshot: dict[str, Any], + actual_snapshot: dict[str, Any], +) -> list[dict[str, Any]]: + """Build a detailed diff report.""" + report = [] + for diff in diffs: + allowed_reason = allowed_diff_reason(case_name, backend_expected, backend_actual, diff["field_path"]) + event_collection, event_index = extract_event_location(diff["field_path"]) + report.append({ + "case_name": case_name, + "session_id": session_id, + "backend_expected": backend_expected, + "backend_actual": backend_actual, + "field_path": diff["field_path"], + "event_collection": event_collection, + "event_index": event_index, + "summary_id": extract_summary_id(diff, expected_snapshot, actual_snapshot), + "expected": diff["expected"], + "actual": diff["actual"], + "allowed": allowed_reason is not None, + "allowed_reason": allowed_reason, + }) + return report diff --git a/tests/sessions/replay_consistency/session_memory_summary_diff_report.json b/tests/sessions/replay_consistency/session_memory_summary_diff_report.json new file mode 100644 index 00000000..574d483d --- /dev/null +++ b/tests/sessions/replay_consistency/session_memory_summary_diff_report.json @@ -0,0 +1,546 @@ +{ + "schema_version": 2, + "backend_pairs": { + "session": [ + "in_memory" + ], + "memory": [ + "in_memory" + ] + }, + "allowed_diffs": [ + { + "case_name": "summary_generation_update", + "backend_expected": "in_memory", + "backend_actual": "sqlite_sql", + "field_paths": [ + "events[0].author", + "events[0].invocation_id", + "events[0].is_summary", + "events[0].model_flags", + "events[0].partial", + "events[0].parts[0].text", + "events[1].author", + "events[1].invocation_id", + "events[1].parts[0].text", + "events[2].author", + "events[2].invocation_id", + "events[2].is_summary", + "events[2].model_flags", + "events[2].partial", + "events[2].parts[0].text", + "historical_events[2].author", + "historical_events[2].invocation_id", + "historical_events[2].is_summary", + "historical_events[2].model_flags", + "historical_events[2].partial", + "historical_events[2].parts[0].text", + "historical_events[3].author", + "historical_events[3].invocation_id", + "historical_events[3].parts[0].text", + "historical_events[4].author", + "historical_events[4].invocation_id", + "historical_events[4].is_summary", + "historical_events[4].model_flags", + "historical_events[4].partial", + "historical_events[4].parts[0].text" + ], + "reason": "After real summary compression, InMemory keeps the generated summary event before recent active events while SQLite reloads active and historical events in storage order. Summary cache content and strict summary metadata must still match." + }, + { + "case_name": "summary_generation_update", + "backend_expected": "in_memory", + "backend_actual": "mock_redis", + "field_paths": [ + "events[0].author", + "events[0].invocation_id", + "events[0].is_summary", + "events[0].model_flags", + "events[0].partial", + "events[0].parts[0].text", + "events[1].author", + "events[1].invocation_id", + "events[1].parts[0].text", + "events[2].author", + "events[2].invocation_id", + "events[2].is_summary", + "events[2].model_flags", + "events[2].partial", + "events[2].parts[0].text", + "historical_events[2].author", + "historical_events[2].invocation_id", + "historical_events[2].is_summary", + "historical_events[2].model_flags", + "historical_events[2].partial", + "historical_events[2].parts[0].text", + "historical_events[3].author", + "historical_events[3].invocation_id", + "historical_events[3].parts[0].text", + "historical_events[4].author", + "historical_events[4].invocation_id", + "historical_events[4].is_summary", + "historical_events[4].model_flags", + "historical_events[4].partial", + "historical_events[4].parts[0].text" + ], + "reason": "After real summary compression, InMemory keeps the generated summary event before recent active events while SQLite reloads active and historical events in storage order. Summary cache content and strict summary metadata must still match." + }, + { + "case_name": "summary_generation_update", + "backend_expected": "in_memory", + "backend_actual": "env_sql", + "field_paths": [ + "events[0].author", + "events[0].invocation_id", + "events[0].is_summary", + "events[0].model_flags", + "events[0].partial", + "events[0].parts[0].text", + "events[1].author", + "events[1].invocation_id", + "events[1].parts[0].text", + "events[2].author", + "events[2].invocation_id", + "events[2].is_summary", + "events[2].model_flags", + "events[2].partial", + "events[2].parts[0].text", + "historical_events[2].author", + "historical_events[2].invocation_id", + "historical_events[2].is_summary", + "historical_events[2].model_flags", + "historical_events[2].partial", + "historical_events[2].parts[0].text", + "historical_events[3].author", + "historical_events[3].invocation_id", + "historical_events[3].parts[0].text", + "historical_events[4].author", + "historical_events[4].invocation_id", + "historical_events[4].is_summary", + "historical_events[4].model_flags", + "historical_events[4].partial", + "historical_events[4].parts[0].text" + ], + "reason": "After real summary compression, InMemory keeps the generated summary event before recent active events while SQLite reloads active and historical events in storage order. Summary cache content and strict summary metadata must still match." + }, + { + "case_name": "summary_generation_update", + "backend_expected": "in_memory", + "backend_actual": "env_redis", + "field_paths": [ + "events[0].author", + "events[0].invocation_id", + "events[0].is_summary", + "events[0].model_flags", + "events[0].partial", + "events[0].parts[0].text", + "events[1].author", + "events[1].invocation_id", + "events[1].parts[0].text", + "events[2].author", + "events[2].invocation_id", + "events[2].is_summary", + "events[2].model_flags", + "events[2].partial", + "events[2].parts[0].text", + "historical_events[2].author", + "historical_events[2].invocation_id", + "historical_events[2].is_summary", + "historical_events[2].model_flags", + "historical_events[2].partial", + "historical_events[2].parts[0].text", + "historical_events[3].author", + "historical_events[3].invocation_id", + "historical_events[3].parts[0].text", + "historical_events[4].author", + "historical_events[4].invocation_id", + "historical_events[4].is_summary", + "historical_events[4].model_flags", + "historical_events[4].partial", + "historical_events[4].parts[0].text" + ], + "reason": "After real summary compression, InMemory keeps the generated summary event before recent active events while SQLite reloads active and historical events in storage order. Summary cache content and strict summary metadata must still match." + }, + { + "case_name": "summary_truncation", + "backend_expected": "in_memory", + "backend_actual": "sqlite_sql", + "field_paths": [ + "events[0].author", + "events[0].custom_metadata", + "events[0].invocation_id", + "events[0].is_summary", + "events[0].model_flags", + "events[0].parts[0].text", + "events[0].version", + "events[1].author", + "events[1].invocation_id", + "events[1].parts[0].text", + "events[2].author", + "events[2].custom_metadata", + "events[2].invocation_id", + "events[2].is_summary", + "events[2].model_flags", + "events[2].parts[0].text", + "events[2].version", + "events[3]", + "events[4]", + "historical_events[0]", + "historical_events[1]" + ], + "reason": "InMemorySessionService keeps max_events-trimmed events in active storage instead of moving them to historical_events; SQLite stores those filtered events in historical_events." + }, + { + "case_name": "summary_truncation", + "backend_expected": "in_memory", + "backend_actual": "mock_redis", + "field_paths": [ + "events[0].author", + "events[0].custom_metadata", + "events[0].invocation_id", + "events[0].is_summary", + "events[0].model_flags", + "events[0].parts[0].text", + "events[0].version", + "events[1].author", + "events[1].invocation_id", + "events[1].parts[0].text", + "events[2].author", + "events[2].custom_metadata", + "events[2].invocation_id", + "events[2].is_summary", + "events[2].model_flags", + "events[2].parts[0].text", + "events[2].version", + "events[3]", + "events[4]", + "historical_events[0]", + "historical_events[1]" + ], + "reason": "InMemorySessionService keeps max_events-trimmed events in active storage instead of moving them to historical_events; SQLite stores those filtered events in historical_events." + }, + { + "case_name": "summary_truncation", + "backend_expected": "in_memory", + "backend_actual": "env_sql", + "field_paths": [ + "events[0].author", + "events[0].custom_metadata", + "events[0].invocation_id", + "events[0].is_summary", + "events[0].model_flags", + "events[0].parts[0].text", + "events[0].version", + "events[1].author", + "events[1].invocation_id", + "events[1].parts[0].text", + "events[2].author", + "events[2].custom_metadata", + "events[2].invocation_id", + "events[2].is_summary", + "events[2].model_flags", + "events[2].parts[0].text", + "events[2].version", + "events[3]", + "events[4]", + "historical_events[0]", + "historical_events[1]" + ], + "reason": "InMemorySessionService keeps max_events-trimmed events in active storage instead of moving them to historical_events; SQLite stores those filtered events in historical_events." + }, + { + "case_name": "summary_truncation", + "backend_expected": "in_memory", + "backend_actual": "env_redis", + "field_paths": [ + "events[0].author", + "events[0].custom_metadata", + "events[0].invocation_id", + "events[0].is_summary", + "events[0].model_flags", + "events[0].parts[0].text", + "events[0].version", + "events[1].author", + "events[1].invocation_id", + "events[1].parts[0].text", + "events[2].author", + "events[2].custom_metadata", + "events[2].invocation_id", + "events[2].is_summary", + "events[2].model_flags", + "events[2].parts[0].text", + "events[2].version", + "events[3]", + "events[4]", + "historical_events[0]", + "historical_events[1]" + ], + "reason": "InMemorySessionService keeps max_events-trimmed events in active storage instead of moving them to historical_events; SQLite stores those filtered events in historical_events." + } + ], + "cases": [ + { + "case_name": "single_turn", + "app_name": "replay_app", + "user_id": "user-1", + "session_id": "session-1", + "status": "matched", + "session": { + "backend_expected": "in_memory", + "backend_actual": null, + "backend_actuals": [], + "status": "not_applicable", + "event_diffs": [], + "state_diffs": [], + "summary_diffs": [], + "summary_content_checks": [], + "summary_metadata_checks": [], + "session_metadata_diffs": [] + }, + "memory": { + "backend_expected": null, + "backend_actual": null, + "status": "not_applicable", + "memory_diffs": [] + } + }, + { + "case_name": "multi_turn", + "app_name": "replay_app", + "user_id": "user-1", + "session_id": "session-multi-turn", + "status": "matched", + "session": { + "backend_expected": "in_memory", + "backend_actual": null, + "backend_actuals": [], + "status": "not_applicable", + "event_diffs": [], + "state_diffs": [], + "summary_diffs": [], + "summary_content_checks": [], + "summary_metadata_checks": [], + "session_metadata_diffs": [] + }, + "memory": { + "backend_expected": null, + "backend_actual": null, + "status": "not_applicable", + "memory_diffs": [] + } + }, + { + "case_name": "tool_call", + "app_name": "replay_app", + "user_id": "user-1", + "session_id": "session-tool-call", + "status": "matched", + "session": { + "backend_expected": "in_memory", + "backend_actual": null, + "backend_actuals": [], + "status": "not_applicable", + "event_diffs": [], + "state_diffs": [], + "summary_diffs": [], + "summary_content_checks": [], + "summary_metadata_checks": [], + "session_metadata_diffs": [] + }, + "memory": { + "backend_expected": null, + "backend_actual": null, + "status": "not_applicable", + "memory_diffs": [] + } + }, + { + "case_name": "state_update", + "app_name": "replay_app", + "user_id": "user-1", + "session_id": "session-state-update", + "status": "matched", + "session": { + "backend_expected": "in_memory", + "backend_actual": null, + "backend_actuals": [], + "status": "not_applicable", + "event_diffs": [], + "state_diffs": [], + "summary_diffs": [], + "summary_content_checks": [], + "summary_metadata_checks": [], + "session_metadata_diffs": [] + }, + "memory": { + "backend_expected": null, + "backend_actual": null, + "status": "not_applicable", + "memory_diffs": [] + } + }, + { + "case_name": "memory_store_search", + "app_name": "replay_app", + "user_id": "user-1", + "session_id": "session-memory-store-search", + "status": "matched", + "session": { + "backend_expected": "in_memory", + "backend_actual": null, + "backend_actuals": [], + "status": "not_applicable", + "event_diffs": [], + "state_diffs": [], + "summary_diffs": [], + "summary_content_checks": [], + "summary_metadata_checks": [], + "session_metadata_diffs": [] + }, + "memory": { + "backend_expected": "in_memory", + "backend_actual": null, + "backend_actuals": [], + "status": "not_applicable", + "memory_diffs": [] + } + }, + { + "case_name": "summary_generation_update", + "app_name": "replay_app", + "user_id": "user-1", + "session_id": "session-summary-generation-update", + "status": "matched", + "session": { + "backend_expected": "in_memory", + "backend_actual": null, + "backend_actuals": [], + "status": "not_applicable", + "event_diffs": [], + "state_diffs": [], + "summary_diffs": [], + "summary_content_checks": [], + "summary_metadata_checks": [], + "session_metadata_diffs": [] + }, + "memory": { + "backend_expected": null, + "backend_actual": null, + "status": "not_applicable", + "memory_diffs": [] + } + }, + { + "case_name": "summary_truncation", + "app_name": "replay_app", + "user_id": "user-1", + "session_id": "session-summary-truncation", + "status": "matched", + "session": { + "backend_expected": "in_memory", + "backend_actual": null, + "backend_actuals": [], + "status": "not_applicable", + "event_diffs": [], + "state_diffs": [], + "summary_diffs": [], + "summary_content_checks": [], + "summary_metadata_checks": [], + "session_metadata_diffs": [] + }, + "memory": { + "backend_expected": null, + "backend_actual": null, + "status": "not_applicable", + "memory_diffs": [] + } + }, + { + "case_name": "recovery_partial_event", + "app_name": "replay_app", + "user_id": "user-1", + "session_id": "session-recovery-partial", + "status": "matched", + "session": { + "backend_expected": "in_memory", + "backend_actual": null, + "backend_actuals": [], + "status": "not_applicable", + "event_diffs": [], + "state_diffs": [], + "summary_diffs": [], + "summary_content_checks": [], + "summary_metadata_checks": [], + "session_metadata_diffs": [] + }, + "memory": { + "backend_expected": null, + "backend_actual": null, + "status": "not_applicable", + "memory_diffs": [] + } + }, + { + "case_name": "duplicate_event_replay", + "app_name": "replay_app", + "user_id": "user-1", + "session_id": "session-duplicate-event-replay", + "status": "matched", + "session": { + "backend_expected": "in_memory", + "backend_actual": null, + "backend_actuals": [], + "status": "not_applicable", + "event_diffs": [], + "state_diffs": [], + "summary_diffs": [], + "summary_content_checks": [], + "summary_metadata_checks": [], + "session_metadata_diffs": [] + }, + "memory": { + "backend_expected": null, + "backend_actual": null, + "status": "not_applicable", + "memory_diffs": [] + } + }, + { + "case_name": "branch_metadata_replay", + "app_name": "replay_app", + "user_id": "user-1", + "session_id": "session-branch-metadata-replay", + "status": "matched", + "session": { + "backend_expected": "in_memory", + "backend_actual": null, + "backend_actuals": [], + "status": "not_applicable", + "event_diffs": [], + "state_diffs": [], + "summary_diffs": [], + "summary_content_checks": [], + "summary_metadata_checks": [], + "session_metadata_diffs": [] + }, + "memory": { + "backend_expected": null, + "backend_actual": null, + "status": "not_applicable", + "memory_diffs": [] + } + } + ], + "totals": { + "cases": 10, + "session_event_diffs": 0, + "session_state_diffs": 0, + "session_summary_diffs": 0, + "session_metadata_diffs": 0, + "summary_content_mismatches": 0, + "summary_metadata_mismatches": 0, + "memory_diffs": 0, + "allowed_diffs": 0, + "unallowed_diffs": 0 + } +} diff --git a/tests/sessions/test_replay_consistency.py b/tests/sessions/test_replay_consistency.py new file mode 100644 index 00000000..0f91c97b --- /dev/null +++ b/tests/sessions/test_replay_consistency.py @@ -0,0 +1,455 @@ +# 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 consistency tests for session and memory backends.""" + +from __future__ import annotations + +import asyncio +import json +from pathlib import Path +from typing import Any + +import pytest + +from .replay_consistency import ( + BASELINE_BACKEND_NAME, + ENV_REDIS_BACKEND_NAME, + ENV_SQL_BACKEND_NAME, + MOCK_REDIS_BACKEND_NAME, + REPLAY_CASES, + MEMORY_REPLAY_CASES, + REDIS_URL_ENV, + SQLITE_BACKEND_NAME, + SQL_URL_ENV, + ReplayCase, + ReplayBackendConfig, + ReplayBackendUnavailable, + comparison_backend_names, + configured_memory_backend_names, + configured_session_backend_names, + default_backend_matrix_enabled, + run_session_replay_case, + run_memory_replay_case, + diff_snapshots, + build_replay_diff_report, + build_diff_report, + build_summary_content_checks, + build_summary_metadata_checks, + assert_all_diffs_allowed, + assert_session_replay_case_snapshot, + assert_memory_replay_case_snapshot, + assert_replay_case_fixtures_load, + uses_allowed_snapshot_variant, + assert_allowed_session_snapshot_variant, +) + + +def test_replay_case_fixtures_load(): + assert_replay_case_fixtures_load() + + +def test_replay_backend_matrix_defaults_to_in_memory(monkeypatch): + monkeypatch.delenv(SQL_URL_ENV, raising=False) + monkeypatch.delenv(REDIS_URL_ENV, raising=False) + + assert configured_session_backend_names() == [BASELINE_BACKEND_NAME] + assert configured_memory_backend_names() == [BASELINE_BACKEND_NAME] + + +def test_replay_backend_matrix_uses_configured_env_backends(monkeypatch): + monkeypatch.setenv(SQL_URL_ENV, "sqlite:///:memory:") + monkeypatch.setenv(REDIS_URL_ENV, "redis://localhost:6379/15") + + assert configured_session_backend_names() == [ + BASELINE_BACKEND_NAME, + ENV_SQL_BACKEND_NAME, + ENV_REDIS_BACKEND_NAME, + ] + assert configured_memory_backend_names() == [ + BASELINE_BACKEND_NAME, + ENV_SQL_BACKEND_NAME, + ENV_REDIS_BACKEND_NAME, + ] + + +def test_replay_backend_matrix_uses_single_configured_env_backend(monkeypatch): + monkeypatch.setenv(SQL_URL_ENV, "sqlite:///:memory:") + monkeypatch.delenv(REDIS_URL_ENV, raising=False) + + assert configured_session_backend_names() == [BASELINE_BACKEND_NAME, ENV_SQL_BACKEND_NAME] + assert configured_memory_backend_names() == [BASELINE_BACKEND_NAME, ENV_SQL_BACKEND_NAME] + + monkeypatch.delenv(SQL_URL_ENV, raising=False) + monkeypatch.setenv(REDIS_URL_ENV, "redis://localhost:6379/15") + + assert configured_session_backend_names() == [BASELINE_BACKEND_NAME, ENV_REDIS_BACKEND_NAME] + assert configured_memory_backend_names() == [BASELINE_BACKEND_NAME, ENV_REDIS_BACKEND_NAME] + + +def test_unconfigured_replay_runs_only_in_memory(monkeypatch): + monkeypatch.delenv(SQL_URL_ENV, raising=False) + monkeypatch.delenv(REDIS_URL_ENV, raising=False) + replay_case = next(replay_case for replay_case in REPLAY_CASES if replay_case.name == "single_turn") + + snapshots = asyncio.run(run_session_replay_case(replay_case)) + + assert list(snapshots) == [BASELINE_BACKEND_NAME] + + +def test_unavailable_sql_backend_falls_back_to_sqlite_sql(monkeypatch): + monkeypatch.delenv(SQL_URL_ENV, raising=False) + monkeypatch.delenv(REDIS_URL_ENV, raising=False) + replay_case = next(replay_case for replay_case in REPLAY_CASES if replay_case.name == "single_turn") + + snapshots = asyncio.run(run_session_replay_case( + replay_case, + backend_config=ReplayBackendConfig(sql_url="not-a-valid-sql-url"), + )) + + assert list(snapshots) == [BASELINE_BACKEND_NAME, SQLITE_BACKEND_NAME] + + +def test_unavailable_redis_backend_falls_back_to_mock_redis(monkeypatch): + monkeypatch.delenv(SQL_URL_ENV, raising=False) + monkeypatch.delenv(REDIS_URL_ENV, raising=False) + replay_case = next(replay_case for replay_case in REPLAY_CASES if replay_case.name == "single_turn") + + snapshots = asyncio.run(run_session_replay_case( + replay_case, + backend_config=ReplayBackendConfig(redis_url="redis://localhost:1/15"), + )) + + assert list(snapshots) == [BASELINE_BACKEND_NAME, MOCK_REDIS_BACKEND_NAME] + + +def test_unavailable_sql_memory_backend_falls_back_to_sqlite_sql(monkeypatch): + monkeypatch.delenv(SQL_URL_ENV, raising=False) + monkeypatch.delenv(REDIS_URL_ENV, raising=False) + replay_case = next(replay_case for replay_case in MEMORY_REPLAY_CASES if replay_case.name == "memory_store_search") + + snapshots = asyncio.run(run_memory_replay_case( + replay_case, + backend_config=ReplayBackendConfig(sql_url="not-a-valid-sql-url"), + )) + + assert list(snapshots) == [BASELINE_BACKEND_NAME, SQLITE_BACKEND_NAME] + + +def test_unavailable_redis_memory_backend_falls_back_to_mock_redis(monkeypatch): + monkeypatch.delenv(SQL_URL_ENV, raising=False) + monkeypatch.delenv(REDIS_URL_ENV, raising=False) + replay_case = next(replay_case for replay_case in MEMORY_REPLAY_CASES if replay_case.name == "memory_store_search") + + snapshots = asyncio.run(run_memory_replay_case( + replay_case, + backend_config=ReplayBackendConfig(redis_url="redis://localhost:1/15"), + )) + + assert list(snapshots) == [BASELINE_BACKEND_NAME, MOCK_REDIS_BACKEND_NAME] + + +@pytest.mark.parametrize("replay_case", REPLAY_CASES, ids=lambda replay_case: replay_case.name) +def test_session_replay_case_is_consistent_between_configured_backends(replay_case: ReplayCase): + try: + snapshots = asyncio.run(run_session_replay_case(replay_case)) + except ReplayBackendUnavailable as ex: + pytest.skip(str(ex)) + + backend_names = comparison_backend_names(snapshots) + if not backend_names: + pytest.skip("Replay backend config has no persistent session backend to compare") + + for backend_name in backend_names: + diffs = diff_snapshots(snapshots[BASELINE_BACKEND_NAME], snapshots[backend_name]) + if diffs: + report = build_diff_report( + case_name=replay_case.name, + session_id=replay_case.session_id, + backend_expected=BASELINE_BACKEND_NAME, + backend_actual=backend_name, + diffs=diffs, + expected_snapshot=snapshots[BASELINE_BACKEND_NAME], + actual_snapshot=snapshots[backend_name], + ) + assert_all_diffs_allowed(report) + continue + + assert diffs == [] + + +@pytest.mark.parametrize("replay_case", REPLAY_CASES, ids=lambda replay_case: replay_case.name) +def test_session_replay_case_preserves_expected_event_trace(replay_case: ReplayCase): + try: + snapshots = asyncio.run(run_session_replay_case(replay_case)) + except ReplayBackendUnavailable as ex: + pytest.skip(str(ex)) + + for backend_name, snapshot in snapshots.items(): + if uses_allowed_snapshot_variant(replay_case.name, backend_name): + assert_allowed_session_snapshot_variant(replay_case.name, snapshot) + continue + assert_session_replay_case_snapshot(replay_case.name, snapshot) + + +@pytest.mark.parametrize("replay_case", MEMORY_REPLAY_CASES, ids=lambda replay_case: replay_case.name) +def test_memory_replay_case_is_consistent_between_configured_backends(replay_case: ReplayCase): + try: + snapshots = asyncio.run(run_memory_replay_case(replay_case)) + except ReplayBackendUnavailable as ex: + pytest.skip(str(ex)) + + backend_names = comparison_backend_names(snapshots) + if not backend_names: + pytest.skip("Replay backend config has no persistent memory backend to compare") + + for backend_name in backend_names: + diffs = diff_snapshots(snapshots[BASELINE_BACKEND_NAME], snapshots[backend_name]) + assert diffs == [] + for snapshot in snapshots.values(): + assert_memory_replay_case_snapshot(replay_case, snapshot) + + +def test_diff_snapshots_reports_nested_event_text_mismatch(): + baseline = { + "session_id": "s1", + "events": [ + {"author": "user", "parts": [{"text": "hello"}]}, + {"author": "agent", "parts": [{"text": "hi"}]}, + ], + } + actual = { + "session_id": "s1", + "events": [ + {"author": "user", "parts": [{"text": "hello"}]}, + {"author": "agent", "parts": [{"text": "wrong"}]}, + ], + } + + diffs = diff_snapshots(baseline, actual) + + assert diffs == [{ + "field_path": "events[1].parts[0].text", + "expected": "hi", + "actual": "wrong", + }] + + +def test_diff_report_entries_include_context_and_allowed_diff_metadata(): + expected_snapshot = { + "session_id": "session-summary-truncation", + "events": [{"author": "user", "parts": [{"text": "My name is Alice."}]}], + "historical_events": [], + } + actual_snapshot = { + "session_id": "session-summary-truncation", + "events": [{ + "author": "system", + "is_summary": True, + "custom_metadata": {"summary_id": "summary-session-summary-truncation-v1"}, + "parts": [{"text": "Summary: Alice introduced herself."}], + }], + "historical_events": [], + } + diffs = diff_snapshots(expected_snapshot, actual_snapshot) + + report = build_diff_report( + case_name="summary_truncation", + session_id="session-summary-truncation", + backend_expected="in_memory", + backend_actual="sqlite_sql", + diffs=diffs, + expected_snapshot=expected_snapshot, + actual_snapshot=actual_snapshot, + ) + text_diff = next(entry for entry in report if entry["field_path"] == "events[0].parts[0].text") + + assert text_diff["case_name"] == "summary_truncation" + assert text_diff["session_id"] == "session-summary-truncation" + assert text_diff["backend_expected"] == "in_memory" + assert text_diff["backend_actual"] == "sqlite_sql" + assert text_diff["event_collection"] == "events" + assert text_diff["event_index"] == 0 + assert text_diff["summary_id"] == "summary-session-summary-truncation-v1" + assert text_diff["expected"] == "My name is Alice." + assert text_diff["actual"] == "Summary: Alice introduced herself." + assert text_diff["allowed"] is True + assert "InMemorySessionService" in text_diff["allowed_reason"] + + +def test_diff_report_marks_unlisted_backend_difference_as_unallowed(): + report = build_diff_report( + case_name="summary_truncation", + session_id="session-summary-truncation", + backend_expected="in_memory", + backend_actual="sqlite_sql", + diffs=[{"field_path": "state.profile.name", "expected": "Alice", "actual": "Bob"}], + expected_snapshot={"session_id": "session-summary-truncation", "events": [], "historical_events": []}, + actual_snapshot={"session_id": "session-summary-truncation", "events": [], "historical_events": []}, + ) + + assert report == [{ + "case_name": "summary_truncation", + "session_id": "session-summary-truncation", + "backend_expected": "in_memory", + "backend_actual": "sqlite_sql", + "field_path": "state.profile.name", + "event_collection": None, + "event_index": None, + "summary_id": None, + "expected": "Alice", + "actual": "Bob", + "allowed": False, + "allowed_reason": None, + }] + + +def test_replay_diff_report_records_every_case_and_diff_category(): + if not default_backend_matrix_enabled(): + pytest.skip("The fixture report assertions target the default InMemory-only matrix") + + report = build_replay_diff_report() + cases = report["cases"] + + assert report["backend_pairs"] == { + "session": [BASELINE_BACKEND_NAME], + "memory": [BASELINE_BACKEND_NAME], + } + assert [case["case_name"] for case in cases] == [replay_case.name for replay_case in REPLAY_CASES] + for case in cases: + assert case["session"]["backend_expected"] == BASELINE_BACKEND_NAME + assert case["session"]["backend_actual"] is None + assert case["session"]["backend_actuals"] == [] + assert set(case["session"]) >= { + "status", + "event_diffs", + "state_diffs", + "summary_diffs", + "summary_content_checks", + "summary_metadata_checks", + "session_metadata_diffs", + } + assert set(case["memory"]) >= {"status", "memory_diffs"} + + summary_case = next(case for case in cases if case["case_name"] == "summary_truncation") + assert summary_case["status"] == "matched" + assert summary_case["session"]["status"] == "not_applicable" + assert summary_case["session"]["event_diffs"] == [] + assert summary_case["session"]["summary_diffs"] == [] + assert summary_case["session"]["state_diffs"] == [] + + state_case = next(case for case in cases if case["case_name"] == "state_update") + assert state_case["session"]["state_diffs"] == [] + assert state_case["status"] == "matched" + + memory_case = next(case for case in cases if case["case_name"] == "memory_store_search") + assert memory_case["memory"]["backend_expected"] == BASELINE_BACKEND_NAME + assert memory_case["memory"]["backend_actual"] is None + assert memory_case["memory"]["backend_actuals"] == [] + assert memory_case["memory"]["status"] == "not_applicable" + assert memory_case["memory"]["memory_diffs"] == [] + + assert report["totals"]["summary_content_mismatches"] == 0 + assert report["totals"]["summary_metadata_mismatches"] == 0 + + non_memory_case = next(case for case in cases if case["case_name"] == "single_turn") + assert non_memory_case["memory"]["status"] == "not_applicable" + assert non_memory_case["memory"]["backend_expected"] is None + assert non_memory_case["memory"]["backend_actual"] is None + + +def test_summary_report_separates_content_semantics_from_strict_metadata(monkeypatch): + monkeypatch.delenv(SQL_URL_ENV, raising=False) + monkeypatch.delenv(REDIS_URL_ENV, raising=False) + + report = build_replay_diff_report(ReplayBackendConfig(sql_url="sqlite:///:memory:")) + summary_case = next(case for case in report["cases"] if case["case_name"] == "summary_generation_update") + + content_checks = summary_case["session"]["summary_content_checks"] + metadata_checks = summary_case["session"]["summary_metadata_checks"] + + assert [check["summary_id"] for check in content_checks] == [ + "summary:session-summary-generation-update:latest", + ] + assert all(check["matched"] for check in content_checks) + assert content_checks[0]["comparison"] == "normalized_text" + assert content_checks[0]["expected_normalized_text"].startswith( + "summary(session-summary-generation-update):" + ) + assert "facts=3-events" in content_checks[0]["expected_normalized_text"] + + summary_metadata = next( + check for check in metadata_checks + if check["summary_id"] == "summary:session-summary-generation-update:latest" + ) + metadata_by_field = {field["field"]: field for field in summary_metadata["fields"]} + assert summary_metadata["matched"] is True + assert metadata_by_field["session_id"]["expected"] == "session-summary-generation-update" + assert metadata_by_field["session_id"]["matched"] is True + assert metadata_by_field["manager_session_id"]["expected"] == "session-summary-generation-update" + assert metadata_by_field["has_summary"]["expected"] is True + assert metadata_by_field["has_summary_timestamp"]["expected"] is True + assert metadata_by_field["summary_event_count"]["expected"] == 1 + assert metadata_by_field["compressed_event_count"]["expected"] == 3 + assert metadata_by_field["historical_event_count"]["expected"] == 5 + assert metadata_by_field["original_event_count"]["expected"] == 5 + assert metadata_by_field["manager_compressed_event_count"]["expected"] == 3 + + +def test_summary_metadata_checks_do_not_hide_storage_mismatches(): + expected_snapshot = { + "session_id": "s1", + "events": [{ + "version": 2, + "is_summary": True, + "custom_metadata": { + "summary_id": "summary-1", + "session_id": "s1", + "summary_version": 2, + "supersedes": "summary-0", + "source_event_ids": ["event-1"], + }, + "parts": [{"text": "Summary: Alice likes green."}], + }], + "historical_events": [], + } + actual_snapshot = { + "session_id": "s2", + "events": [{ + "version": 3, + "is_summary": True, + "custom_metadata": { + "summary_id": "summary-1", + "session_id": "s2", + "summary_version": 3, + "supersedes": "summary-x", + "source_event_ids": ["event-2"], + }, + "parts": [{"text": "summary: alice likes green."}], + }], + "historical_events": [], + } + + content_checks = build_summary_content_checks(expected_snapshot, actual_snapshot) + metadata_checks = build_summary_metadata_checks(expected_snapshot, actual_snapshot) + fields = {field["field"]: field for field in metadata_checks[0]["fields"]} + + assert content_checks[0]["matched"] is True + assert metadata_checks[0]["matched"] is False + assert fields["session_id"]["matched"] is False + assert fields["event_version"]["matched"] is False + assert fields["summary_version"]["matched"] is False + assert fields["supersedes"]["matched"] is False + assert fields["source_event_ids"]["matched"] is False + + +def test_replay_diff_report_fixture_matches_current_output(): + if not default_backend_matrix_enabled(): + pytest.skip("The checked-in report fixture targets the default InMemory-only matrix") + + report_path = Path(__file__).parent / "replay_consistency" / "session_memory_summary_diff_report.json" + expected_report = json.loads(report_path.read_text(encoding="utf-8")) + + assert build_replay_diff_report() == expected_report diff --git a/tests/sessions/test_replay_mutations.py b/tests/sessions/test_replay_mutations.py new file mode 100644 index 00000000..361824a9 --- /dev/null +++ b/tests/sessions/test_replay_mutations.py @@ -0,0 +1,292 @@ +# 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. +"""Mutation tests for the replay consistency framework.""" + +from __future__ import annotations + +import asyncio +import copy +from typing import Any +from typing import Callable + +import pytest + +from .replay_consistency import ( + MEMORY_REPLAY_CASES, + REPLAY_CASES, + ReplayCase, + ReplayBackendUnavailable, + build_diff_report, + build_summary_content_checks, + build_summary_metadata_checks, + diff_snapshots, + run_memory_replay_case, + run_session_replay_case, +) + + +SnapshotMutator = Callable[[dict[str, Any]], None] + + +def _run_session_replay_case_or_skip(replay_case: ReplayCase) -> dict[str, dict[str, Any]]: + try: + return asyncio.run(run_session_replay_case(replay_case)) + except ReplayBackendUnavailable as ex: + pytest.skip(str(ex)) + + +def _run_memory_replay_case_or_skip(replay_case: ReplayCase) -> dict[str, dict[str, Any]]: + try: + return asyncio.run(run_memory_replay_case(replay_case)) + except ReplayBackendUnavailable as ex: + pytest.skip(str(ex)) + + +def _session_case(case_name: str) -> ReplayCase: + return next(replay_case for replay_case in REPLAY_CASES if replay_case.name == case_name) + + +def _first_text_part(snapshot: dict[str, Any]) -> dict[str, Any]: + for collection in ("historical_events", "events"): + for event in snapshot.get(collection, []): + for part in event.get("parts", []): + if "text" in part: + return part + raise AssertionError("Snapshot has no text part to mutate") + + +def _active_summary_event(snapshot: dict[str, Any]) -> dict[str, Any]: + for event in snapshot["events"]: + if event["is_summary"]: + return event + raise AssertionError("Snapshot has no active summary event") + + +def _historical_summary_text(snapshot: dict[str, Any]) -> str: + for event in snapshot["historical_events"]: + if event["is_summary"]: + return event["parts"][0]["text"] + raise AssertionError("Snapshot has no historical summary event") + + +def _session_unallowed_diffs( + replay_case: ReplayCase, + clean: dict[str, Any], + mutated: dict[str, Any], + *, + backend_actual: str = "mutated", +) -> list[dict[str, Any]]: + report = _session_diff_report( + replay_case, + clean, + mutated, + backend_actual=backend_actual, + ) + return [entry for entry in report if not entry["allowed"]] + + +def _session_diff_report( + replay_case: ReplayCase, + clean: dict[str, Any], + mutated: dict[str, Any], + *, + backend_actual: str = "mutated", +) -> list[dict[str, Any]]: + diffs = diff_snapshots(clean, mutated) + return build_diff_report( + case_name=replay_case.name, + session_id=replay_case.session_id, + backend_expected="in_memory", + backend_actual=backend_actual, + diffs=diffs, + expected_snapshot=clean, + actual_snapshot=mutated, + ) + + +def _summary_mismatches( + clean: dict[str, Any], + mutated: dict[str, Any], +) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: + content_checks = build_summary_content_checks(clean, mutated) + metadata_checks = build_summary_metadata_checks(clean, mutated) + content_mismatches = [check for check in content_checks if not check["matched"]] + metadata_mismatches = [check for check in metadata_checks if not check["matched"]] + return content_mismatches, metadata_mismatches + + +def _assert_session_mutation_detected( + replay_case: ReplayCase, + clean: dict[str, Any], + mutated: dict[str, Any], +) -> None: + assert _session_unallowed_diffs(replay_case, clean, mutated) + + +def _assert_summary_mutation_detected( + replay_case: ReplayCase, + clean: dict[str, Any], + mutated: dict[str, Any], +) -> None: + content_mismatches, metadata_mismatches = _summary_mismatches(clean, mutated) + assert ( + _session_unallowed_diffs(replay_case, clean, mutated) + or content_mismatches + or metadata_mismatches + ) + + +@pytest.mark.parametrize("replay_case", REPLAY_CASES, ids=lambda replay_case: replay_case.name) +def test_each_session_case_detects_a_basic_event_text_mutation(replay_case: ReplayCase): + snapshots = _run_session_replay_case_or_skip(replay_case) + clean = snapshots["in_memory"] + mutated = copy.deepcopy(clean) + + _first_text_part(mutated)["text"] = "MUTATED: event text changed" + + _assert_session_mutation_detected(replay_case, clean, mutated) + + +def test_state_mutation_is_detected(): + replay_case = _session_case("state_update") + snapshots = _run_session_replay_case_or_skip(replay_case) + clean = snapshots["in_memory"] + mutated = copy.deepcopy(clean) + + mutated["state"]["profile.name"] = "Mallory" + + _assert_session_mutation_detected(replay_case, clean, mutated) + + +def test_tool_call_mutation_is_detected(): + replay_case = _session_case("tool_call") + snapshots = _run_session_replay_case_or_skip(replay_case) + clean = snapshots["in_memory"] + mutated = copy.deepcopy(clean) + + mutated["events"][1]["parts"][0]["function_call"]["name"] = "get_stock_price" + + _assert_session_mutation_detected(replay_case, clean, mutated) + + +def _drop_first_memory(snapshot: dict[str, Any]) -> None: + snapshot["searches"][0]["memories"].pop(0) + + +def _change_memory_text(snapshot: dict[str, Any]) -> None: + snapshot["searches"][0]["memories"][0]["parts"][0]["text"] = "MUTATED: wrong memory" + + +def _change_memory_author(snapshot: dict[str, Any]) -> None: + snapshot["searches"][-1]["memories"][0]["author"] = "user" + + +@pytest.mark.parametrize( + "mutate", + [_drop_first_memory, _change_memory_text, _change_memory_author], + ids=["memory_missing", "memory_text_wrong", "memory_author_wrong"], +) +def test_memory_mutations_are_detected(mutate: SnapshotMutator): + replay_case = next(replay_case for replay_case in MEMORY_REPLAY_CASES if replay_case.name == "memory_store_search") + snapshots = _run_memory_replay_case_or_skip(replay_case) + clean = snapshots["in_memory"] + mutated = copy.deepcopy(clean) + + mutate(mutated) + + assert diff_snapshots(clean, mutated) + + +def _drop_summary_cache(snapshot: dict[str, Any]) -> None: + snapshot["summary"] = None + + +def _change_summary_text(snapshot: dict[str, Any]) -> None: + snapshot["summary"]["text"] = "MUTATED: wrong summary" + + +def _change_summary_session_id(snapshot: dict[str, Any]) -> None: + snapshot["summary"]["metadata"]["session_id"] = "wrong-session" + + +def _change_summary_manager_session_id(snapshot: dict[str, Any]) -> None: + snapshot["summary"]["metadata"]["manager_session_id"] = "wrong-session" + + +def _overwrite_latest_summary_with_stale_summary(snapshot: dict[str, Any]) -> None: + stale_summary_text = _historical_summary_text(snapshot) + latest_summary_event = _active_summary_event(snapshot) + snapshot["summary"]["text"] = stale_summary_text.removeprefix("Previous conversation summary: ") + snapshot["summary"]["metadata"]["summary_event_text"] = stale_summary_text + latest_summary_event["parts"][0]["text"] = stale_summary_text + + +def _drop_active_summary_event(snapshot: dict[str, Any]) -> None: + snapshot["events"] = [event for event in snapshot["events"] if not event["is_summary"]] + snapshot["summary"]["metadata"]["summary_event_count"] = 0 + snapshot["summary"]["metadata"]["summary_event_text"] = None + + +def _change_summary_compressed_count(snapshot: dict[str, Any]) -> None: + snapshot["summary"]["metadata"]["compressed_event_count"] += 1 + + +@pytest.mark.parametrize( + "mutate", + [ + _drop_summary_cache, + _change_summary_text, + _change_summary_session_id, + _change_summary_manager_session_id, + _overwrite_latest_summary_with_stale_summary, + _drop_active_summary_event, + _change_summary_compressed_count, + ], + ids=[ + "summary_missing", + "summary_text_wrong", + "summary_session_wrong", + "summary_manager_session_wrong", + "summary_overwritten_by_stale_summary", + "summary_event_missing", + "summary_compressed_count_wrong", + ], +) +def test_summary_mutations_are_detected(mutate: SnapshotMutator): + replay_case = _session_case("summary_generation_update") + snapshots = _run_session_replay_case_or_skip(replay_case) + clean = snapshots["in_memory"] + mutated = copy.deepcopy(clean) + + mutate(mutated) + + _assert_summary_mutation_detected(replay_case, clean, mutated) + + +def test_allowed_summary_event_order_diff_does_not_hide_summary_metadata_mismatch(): + replay_case = _session_case("summary_generation_update") + snapshots = _run_session_replay_case_or_skip(replay_case) + clean = snapshots["in_memory"] + mutated = copy.deepcopy(clean) + + latest_summary_event = _active_summary_event(mutated) + latest_summary_event["parts"][0]["text"] = "MUTATED: wrong summary event text" + mutated["summary"]["metadata"]["summary_event_text"] = "MUTATED: wrong summary event text" + + content_mismatches, metadata_mismatches = _summary_mismatches(clean, mutated) + report = _session_diff_report( + replay_case, + clean, + mutated, + backend_actual="sqlite_sql", + ) + allowed_paths = {entry["field_path"] for entry in report if entry["allowed"]} + unallowed_paths = {entry["field_path"] for entry in report if not entry["allowed"]} + + assert "events[0].parts[0].text" in allowed_paths + assert "summary.metadata.summary_event_text" in unallowed_paths + assert content_mismatches == [] + assert metadata_mismatches