From cdb203c346decc25aa9f1eef828eb87ff47d450a Mon Sep 17 00:00:00 2001 From: xiaoyvhuv Date: Mon, 13 Jul 2026 19:38:41 +0800 Subject: [PATCH] sessions: add replay consistency harness for session memory summary This change adds a test-side replay consistency harness for Session, Memory, and Summary behavior across InMemory, SQLite, SQL URL, and Redis backends. It includes deterministic replay cases, canonical snapshots, semantic oracle checks, mutation detection, restart validation, structured JSON reports, and env-gated integration tests. Fixes #89 RELEASE NOTES: NONE --- .github/workflows/ci.yml | 15 +- pyproject.toml | 5 + tests/memory/test_mempalace_memory_service.py | 19 +- tests/sessions/replay_consistency/README.md | 69 +++ tests/sessions/replay_consistency/__init__.py | 7 + .../replay_consistency/acceptance_matrix.json | 92 ++++ tests/sessions/replay_consistency/adapters.py | 335 ++++++++++++ .../replay_consistency/canonicalize.py | 89 ++++ .../replay_consistency/capabilities.py | 55 ++ tests/sessions/replay_consistency/cases.py | 280 ++++++++++ tests/sessions/replay_consistency/diff.py | 264 ++++++++++ .../sessions/replay_consistency/mutations.py | 131 +++++ tests/sessions/replay_consistency/oracle.py | 72 +++ .../replay_cases_manifest.json | 68 +++ .../replay_consistency/report_schema.json | 75 +++ .../sessions/replay_consistency/reporting.py | 144 ++++++ tests/sessions/replay_consistency/snapshot.py | 166 ++++++ .../replay_consistency/test_allowed_diff.py | 63 +++ .../test_canonicalization.py | 116 +++++ .../test_replay_consistency.py | 478 ++++++++++++++++++ .../test_replay_integration.py | 123 +++++ .../test_replay_properties.py | 86 ++++ .../replay_consistency/test_replay_restart.py | 127 +++++ .../replay_consistency/test_report_schema.py | 82 +++ 24 files changed, 2956 insertions(+), 5 deletions(-) create mode 100644 tests/sessions/replay_consistency/README.md create mode 100644 tests/sessions/replay_consistency/__init__.py create mode 100644 tests/sessions/replay_consistency/acceptance_matrix.json create mode 100644 tests/sessions/replay_consistency/adapters.py create mode 100644 tests/sessions/replay_consistency/canonicalize.py create mode 100644 tests/sessions/replay_consistency/capabilities.py create mode 100644 tests/sessions/replay_consistency/cases.py create mode 100644 tests/sessions/replay_consistency/diff.py create mode 100644 tests/sessions/replay_consistency/mutations.py create mode 100644 tests/sessions/replay_consistency/oracle.py create mode 100644 tests/sessions/replay_consistency/replay_cases_manifest.json create mode 100644 tests/sessions/replay_consistency/report_schema.json create mode 100644 tests/sessions/replay_consistency/reporting.py create mode 100644 tests/sessions/replay_consistency/snapshot.py create mode 100644 tests/sessions/replay_consistency/test_allowed_diff.py create mode 100644 tests/sessions/replay_consistency/test_canonicalization.py create mode 100644 tests/sessions/replay_consistency/test_replay_consistency.py create mode 100644 tests/sessions/replay_consistency/test_replay_integration.py create mode 100644 tests/sessions/replay_consistency/test_replay_properties.py create mode 100644 tests/sessions/replay_consistency/test_replay_restart.py create mode 100644 tests/sessions/replay_consistency/test_report_schema.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index c8fcd6e4..6e5862f3 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -65,6 +65,8 @@ jobs: test: runs-on: [self-hosted, trpc-agent-python-ci] timeout-minutes: 20 + env: + REPLAY_REPORT_DIR: ${{ github.workspace }}/replay-reports steps: - name: Checkout uses: actions/checkout@v4 @@ -75,7 +77,18 @@ jobs: - name: Run tests with coverage # run: pytest --cov=trpc_agent_sdk --cov-report=xml --cov-report=term tests/ - run: pytest --cov=trpc_agent_sdk --cov-report=xml --cov-report=term --cov-fail-under=80 tests/ + run: | + mkdir -p "$REPLAY_REPORT_DIR" + pytest --cov=trpc_agent_sdk --cov-report=xml --cov-report=term --cov-fail-under=80 tests/ + + - name: Upload replay consistency reports + if: always() + uses: actions/upload-artifact@v4 + with: + name: replay-consistency-reports + path: ${{ env.REPLAY_REPORT_DIR }} + if-no-files-found: warn + retention-days: 14 - name: Upload coverage reports to Codecov uses: codecov/codecov-action@v4 diff --git a/pyproject.toml b/pyproject.toml index 6c47d4d5..cfdb78c1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -206,3 +206,8 @@ minversion = "6.0" addopts = "-ra -q" testpaths = ["tests"] asyncio_mode = "auto" +markers = [ + "replay_lightweight: deterministic Session/Memory/Summary replay consistency tests", + "replay_integration: environment-gated replay consistency tests for external backends", + "replay_property: optional property-based replay consistency tests", +] diff --git a/tests/memory/test_mempalace_memory_service.py b/tests/memory/test_mempalace_memory_service.py index d45a099e..dc0abf8c 100644 --- a/tests/memory/test_mempalace_memory_service.py +++ b/tests/memory/test_mempalace_memory_service.py @@ -7,20 +7,31 @@ from __future__ import annotations +import importlib.util import time from typing import Optional -from trpc_agent_sdk.abc import MemoryServiceConfig +import pytest + +_HAS_MEMPALACE = importlib.util.find_spec("mempalace") is not None +pytestmark = pytest.mark.skipif( + not _HAS_MEMPALACE, + reason="MemPalace memory tests require the optional mempalace extra", +) + from trpc_agent_sdk.context import new_agent_context from trpc_agent_sdk.events import Event -from trpc_agent_sdk.memory.mempalace_memory_service import MempalaceMemoryService -from trpc_agent_sdk.memory.mempalace_memory_service import get_mempalace_filters -from trpc_agent_sdk.memory.mempalace_memory_service import set_mempalace_filters from trpc_agent_sdk.sessions import Session from trpc_agent_sdk.types import Content from trpc_agent_sdk.types import Part from trpc_agent_sdk.types import SearchMemoryResponse +if _HAS_MEMPALACE: + from trpc_agent_sdk.abc import MemoryServiceConfig + from trpc_agent_sdk.memory.mempalace_memory_service import MempalaceMemoryService + from trpc_agent_sdk.memory.mempalace_memory_service import get_mempalace_filters + from trpc_agent_sdk.memory.mempalace_memory_service import set_mempalace_filters + def _make_config() -> MemoryServiceConfig: cfg = MemoryServiceConfig(enabled=True) diff --git a/tests/sessions/replay_consistency/README.md b/tests/sessions/replay_consistency/README.md new file mode 100644 index 00000000..e15594d0 --- /dev/null +++ b/tests/sessions/replay_consistency/README.md @@ -0,0 +1,69 @@ +# Session / Memory / Summary Replay Consistency + +This package implements the lightweight replay harness for Issue #89. It is intentionally test-side only: no production API, storage schema, or runtime dependency is changed. + +## Default Contract + +- Default lightweight comparison is `InMemoryReplayAdapter` vs `SQLiteReplayAdapter`. +- The lightweight suite uses temporary SQLite files and a deterministic fake summarizer, so it does not require Redis, MySQL, PostgreSQL, network access, or a real LLM. +- Integration tests are opt-in. They run only when their explicit environment switches and backend URLs are set. +- Replay operations are Python fixtures because the SDK event model uses typed `Content`, `Part`, `FunctionCall`, and `FunctionResponse` objects. The checked-in `replay_cases_manifest.json` summarizes the public coverage in a review-friendly format. +- `acceptance_matrix.json` separates the 10 required public cases from the extended all-entities contract case. + +## Design Note + +The harness replays deterministic operation sequences through the SDK service APIs, then reads observable state back through those same services before comparing snapshots. Canonicalization is intentionally minimal: generated event IDs are mapped to logical client IDs, dictionary keys are sorted, Unicode text is normalized, and backend metadata is excluded; event order, state values, tool call/response linkage, memory scope, summary ownership, and summary coverage remain strict. Summary comparison is split between stored facts and derived semantics because the SDK does not expose persisted summary revision or lineage fields. Allowed differences must be explicit, localized, and justified, so backend drift cannot be hidden by broad ignores. Default execution uses InMemory and temporary SQLite plus a deterministic fake summarizer, keeping CI lightweight while integration adapters allow real SQL and Redis checks when environment variables are provided. + +## SDK API Path + +```text +ReplayCase + -> ReplayBackendAdapter + -> trpc_agent_sdk public SessionService / MemoryService APIs + -> Backend storage + -> fresh service read via SnapshotReader + -> canonicalize_snapshot + -> semantic oracle + field diff + report +``` + +Adapters do not implement their own storage model. They call `create_session`, `append_event`, `create_session_summary`, `get_session`, `store_session`, and `search_memory` on real SDK services. + +## Summary Oracle + +The SDK currently exposes summary behavior through summary events, historical events, and `SummarizerSessionManager`; it does not persist a public `version`, `supersedes`, or coverage model. The harness therefore records derived summary metadata on the test side: + +- `version` is the per-session summary creation order observed during replay. +- `covered_event_ids` are the logical client event IDs selected by `find_events_for_summary` before summary creation. +- `active` is derived from the active summary event after replay. +- `session_id`, `user_id`, and `app_name` are strict ownership checks and must match the session being replayed. + +These derived fields are not production API claims. They are a semantic oracle for detecting summary loss, stale overwrite, wrong-session ownership, and coverage drift without changing SDK schema. + +## Reports + +Each run writes structured JSON reports with: + +- `case_id` +- `backend_pair` +- aggregate metrics +- `session_id` +- `entity_type` +- `entity_id` +- `index` +- `field_path` +- `reference_value` +- `actual_value` +- `category` +- `allowed` +- `reason` + +Set `REPLAY_REPORT_DIR` to retain reports outside pytest temporary directories. CI uses this to upload replay artifacts. + +## Commands + +```bash +python -m pytest tests/sessions/replay_consistency/test_replay_consistency.py -q -m replay_lightweight +python -m pytest tests/sessions/replay_consistency -q +RUN_REPLAY_SQL_INTEGRATION=1 DATABASE_URL=sqlite:////tmp/replay.db python -m pytest tests/sessions/replay_consistency/test_replay_integration.py -q -m replay_integration +RUN_REPLAY_REDIS_INTEGRATION=1 REDIS_URL=redis://localhost:6379/15 python -m pytest tests/sessions/replay_consistency/test_replay_integration.py -q -m replay_integration +``` diff --git a/tests/sessions/replay_consistency/__init__.py b/tests/sessions/replay_consistency/__init__.py new file mode 100644 index 00000000..a136c9af --- /dev/null +++ b/tests/sessions/replay_consistency/__init__.py @@ -0,0 +1,7 @@ +# 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, memory, and summary backends.""" diff --git a/tests/sessions/replay_consistency/acceptance_matrix.json b/tests/sessions/replay_consistency/acceptance_matrix.json new file mode 100644 index 00000000..d92e790b --- /dev/null +++ b/tests/sessions/replay_consistency/acceptance_matrix.json @@ -0,0 +1,92 @@ +{ + "schema_version": "1.0", + "required_cases": [ + "single_turn_text", + "multi_turn_text", + "tool_call_response", + "state_shallow_update", + "memory_scope_user_session", + "summary_create_update", + "summary_event_truncation", + "failure_retry_append", + "cross_session_isolation", + "summary_defect_specials" + ], + "extended_cases": [ + "all_entities_contract" + ], + "requirements": { + "p0_backends": { + "requirement": "Run lightweight replay against InMemory and a persistent or simulated persistent backend.", + "backends": ["in_memory", "sqlite"], + "test": "test_lightweight_replay_cases" + }, + "p0_public_cases": { + "requirement": "Provide at least 10 deterministic public replay cases.", + "required_case_count": 10, + "extended_case_count": 1, + "test": "test_acceptance_matrix_matches_fixtures" + }, + "p0_public_case_mutation_matrix": { + "requirement": "Each required public replay case detects an injected inconsistency.", + "detected": 10, + "total": 10, + "rate": 1.0, + "test": "test_required_public_cases_detect_injected_inconsistency" + }, + "p0_entities": { + "requirement": "Cover event, state, memory, and summary entities.", + "entities": ["event", "state", "memory", "summary"], + "test": "test_snapshot_contains_all_entities" + }, + "p0_summary_defects": { + "requirement": "Detect summary loss, stale overwrite, and wrong session ownership.", + "summary_loss_detection_rate": 1.0, + "summary_overwrite_detection_rate": 1.0, + "summary_owner_error_detection_rate": 1.0, + "test": "test_summary_defect_detection" + }, + "p0_mutation_score": { + "requirement": "Detect all registered public mutations.", + "mutation_detection_rate": 1.0, + "survived_mutations": [], + "test": "test_mutation_detection" + }, + "p1_runtime_fault_detection": { + "requirement": "Detect an after-commit client retry fault as a runtime replay inconsistency.", + "runtime_fault_detection_rate": 1.0, + "test": "test_runtime_fault_retry_duplicate_is_detected" + }, + "p1_persistent_rebuild": { + "requirement": "Destroy and rebuild the persistent backend between replay phases, then read back through a fresh service.", + "cases": [ + "summary_create_update", + "single_turn_text" + ], + "test": "test_sqlite_destroy_rebuild_continue_and_readback" + }, + "p0_reports": { + "requirement": "Generate structured JSON diff reports with precise location fields.", + "required_fields": [ + "case_id", + "backend_pair", + "session_id", + "entity_type", + "entity_id", + "index", + "field_path", + "reference_value", + "actual_value", + "allowed", + "category", + "reason" + ], + "test": "test_diff_report_schema_contains_required_location_fields" + }, + "p0_lightweight_runtime": { + "requirement": "Run default lightweight mode in less than 30 seconds without external services.", + "lightweight_duration_seconds_max": 30, + "test": "test_lightweight_replay_cases" + } + } +} diff --git a/tests/sessions/replay_consistency/adapters.py b/tests/sessions/replay_consistency/adapters.py new file mode 100644 index 00000000..3e450771 --- /dev/null +++ b/tests/sessions/replay_consistency/adapters.py @@ -0,0 +1,335 @@ +# 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 adapters that replay standard operations through public SDK APIs.""" + +from __future__ import annotations + +from collections import defaultdict +from pathlib import Path + +from trpc_agent_sdk.abc import MemoryServiceConfig +from trpc_agent_sdk.events import Event +from trpc_agent_sdk.memory import InMemoryMemoryService +from trpc_agent_sdk.memory import RedisMemoryService +from trpc_agent_sdk.memory import SqlMemoryService +from trpc_agent_sdk.models import LlmResponse +from trpc_agent_sdk.sessions import InMemorySessionService +from trpc_agent_sdk.sessions import RedisSessionService +from trpc_agent_sdk.sessions import Session +from trpc_agent_sdk.sessions import SessionServiceConfig +from trpc_agent_sdk.sessions import SessionSummarizer +from trpc_agent_sdk.sessions import SqlSessionService +from trpc_agent_sdk.sessions import SummarizerSessionManager +from trpc_agent_sdk.sessions import find_events_for_summary +from trpc_agent_sdk.types import Content +from trpc_agent_sdk.types import FunctionCall +from trpc_agent_sdk.types import FunctionResponse +from trpc_agent_sdk.types import Part + +from .cases import ReplayCase +from .cases import ReplayOp +from .snapshot import MemoryProbe +from .snapshot import Snapshot +from .snapshot import SummaryRecord +from .snapshot import read_snapshot + + +class FakeSummaryModel: + """Deterministic model used by SessionSummarizer in replay tests.""" + + name = "fake-replay-summary-model" + + async def generate_async(self, request, stream: bool = False, ctx=None): + text = "deterministic-summary:stable" + yield LlmResponse(content=Content(parts=[Part.from_text(text=text)])) + + +class ReplayBackendAdapter: + """Common adapter state and operation mapping.""" + + name = "base" + + def __init__(self, *, case: ReplayCase, workdir: Path | None = None) -> None: + self.case = case + self.workdir = workdir + self.session_service = None + self.memory_service = None + self.sessions: dict[tuple[str, str, str], Session] = {} + self.event_id_map: dict[str, str] = {} + self.actual_to_client_event_id: dict[str, str] = {} + self.summary_records: dict[str, list[SummaryRecord]] = defaultdict(list) + self.memory_probes: list[MemoryProbe] = [] + self._timestamp = 1_700_000_000.0 + + async def setup(self) -> None: + raise NotImplementedError + + async def close(self) -> None: + if self.memory_service: + await self.memory_service.close() + if self.session_service: + await self.session_service.close() + + async def replay(self) -> Snapshot: + for op in self.case.operations: + await self.apply(op) + return await self.snapshot() + + async def apply(self, op: ReplayOp) -> None: + if op.kind == "create_session": + await self.create_session(op) + elif op.kind == "append_text": + await self.append_event(op, [Part.from_text(text=op.text)]) + elif op.kind == "append_tool_call": + function_call = FunctionCall(name=op.function_name, args=op.function_args) + function_call.id = op.function_call_id + await self.append_event(op, [Part(function_call=function_call)]) + elif op.kind == "append_tool_response": + function_response = FunctionResponse(name=op.function_name, response=op.function_response) + function_response.id = op.function_call_id + await self.append_event(op, [Part(function_response=function_response)]) + elif op.kind == "store_memory": + await self.store_memory(op) + elif op.kind == "search_memory": + self.memory_probes.append( + MemoryProbe( + probe_id=op.probe_id or op.query, + session_key=self._session_key(op), + query=op.query, + ) + ) + elif op.kind == "summarize": + await self.summarize(op) + else: + raise ValueError(f"Unsupported replay op: {op.kind}") + + async def create_session(self, op: ReplayOp) -> None: + session = await self.session_service.create_session( + app_name=op.app_name, + user_id=op.user_id, + session_id=op.session_id, + state=op.initial_state, + ) + self.sessions[(op.app_name, op.user_id, op.session_id)] = session + + async def append_event(self, op: ReplayOp, parts: list[Part]) -> None: + session = self._get_session(op) + content_role = "user" if op.author == "user" else "model" + event = Event( + invocation_id=f"{self.case.case_id}:{op.client_event_id}", + author=op.author, + content=Content(parts=parts, role=content_role), + timestamp=self._next_timestamp(), + ) + if op.state_delta: + event.actions.state_delta.update(op.state_delta) + await self.session_service.append_event(session, event) + if op.client_event_id: + self.event_id_map[op.client_event_id] = event.id + self.actual_to_client_event_id[event.id] = op.client_event_id + await self._refresh_session(op) + + async def store_memory(self, op: ReplayOp) -> None: + session = await self.session_service.get_session( + app_name=op.app_name, + user_id=op.user_id, + session_id=op.session_id, + ) + await self.memory_service.store_session(session) + + async def summarize(self, op: ReplayOp) -> None: + session = self._get_session(op) + events_for_summary, _ = find_events_for_summary(session.events, keep_recent_count=2) + covered_ids = [self.actual_to_client_event_id.get(event.id, event.id) for event in events_for_summary] + manager = self.session_service.summarizer_manager + await manager.create_session_summary(session, force=True) + active_summary = next((event for event in session.events if event.is_summary_event()), None) + if active_summary is None: + return + if len(session.events) > 1: + active_summary.timestamp = session.events[1].timestamp - 0.1 + await self.session_service.update_session(session) + client_summary_id = op.client_summary_id or f"summary-{len(self.summary_records[self._session_key(op)]) + 1}" + self.actual_to_client_event_id[active_summary.id] = client_summary_id + session_key = self._session_key(op) + records = self.summary_records[session_key] + for record in records: + record.active = False + records.append( + SummaryRecord( + client_summary_id=client_summary_id, + session_id=op.session_id, + user_id=op.user_id, + app_name=op.app_name, + event_id=client_summary_id, + text=active_summary.get_text(), + version=len(records) + 1, + active=True, + covered_event_ids=covered_ids, + timestamp=active_summary.timestamp, + ) + ) + await self._refresh_session(op) + + async def snapshot(self) -> Snapshot: + return await read_snapshot( + backend=self.name, + case_id=self.case.case_id, + session_service=self.session_service, + memory_service=self.memory_service, + sessions=list(self.sessions.values()), + actual_to_client_event_id=self.actual_to_client_event_id, + memory_probes=self.memory_probes, + summary_records=[record for records in self.summary_records.values() for record in records], + ) + + def _next_timestamp(self) -> float: + self._timestamp += 1.0 + return self._timestamp + + def _get_session(self, op: ReplayOp) -> Session: + return self.sessions[(op.app_name, op.user_id, op.session_id)] + + async def _refresh_session(self, op: ReplayOp) -> None: + session = await self.session_service.get_session( + app_name=op.app_name, + user_id=op.user_id, + session_id=op.session_id, + ) + self.sessions[(op.app_name, op.user_id, op.session_id)] = session + + @staticmethod + def _session_key(op: ReplayOp) -> str: + return f"{op.app_name}/{op.user_id}" + + +class InMemoryReplayAdapter(ReplayBackendAdapter): + """Replay adapter backed by in-memory session and memory services.""" + + name = "in_memory" + + async def setup(self) -> None: + config = SessionServiceConfig(store_historical_events=True) + config.clean_ttl_config() + summarizer = SessionSummarizer(model=FakeSummaryModel(), keep_recent_count=2) + manager = SummarizerSessionManager(model=FakeSummaryModel(), summarizer=summarizer, auto_summarize=False) + self.session_service = InMemorySessionService(summarizer_manager=manager, session_config=config) + memory_config = MemoryServiceConfig(enabled=True) + memory_config.clean_ttl_config() + self.memory_service = InMemoryMemoryService(memory_service_config=memory_config) + + +class SQLiteReplayAdapter(ReplayBackendAdapter): + """Replay adapter backed by SQLite SQL session and memory services.""" + + name = "sqlite" + + async def setup(self) -> None: + assert self.workdir is not None + db_path = self.workdir / f"{self.case.case_id}_{self.name}.db" + db_url = f"sqlite:///{db_path.as_posix()}" + config = SessionServiceConfig(store_historical_events=True) + config.clean_ttl_config() + summarizer = SessionSummarizer(model=FakeSummaryModel(), keep_recent_count=2) + manager = SummarizerSessionManager(model=FakeSummaryModel(), summarizer=summarizer, auto_summarize=False) + self.session_service = SqlSessionService( + db_url=db_url, + summarizer_manager=manager, + session_config=config, + is_async=False, + ) + await self.session_service._sql_storage.create_sql_engine() + memory_config = MemoryServiceConfig(enabled=True) + memory_config.clean_ttl_config() + self.memory_service = SqlMemoryService( + db_url=db_url, + memory_service_config=memory_config, + enabled=True, + is_async=False, + ) + await self.memory_service._sql_storage.create_sql_engine() + + +class SQLUrlReplayAdapter(ReplayBackendAdapter): + """Replay adapter backed by an externally configured SQL URL.""" + + name = "sql" + + def __init__(self, *, case: ReplayCase, db_url: str) -> None: + super().__init__(case=case) + self.db_url = db_url + + async def setup(self) -> None: + config = SessionServiceConfig(store_historical_events=True) + config.clean_ttl_config() + summarizer = SessionSummarizer(model=FakeSummaryModel(), keep_recent_count=2) + manager = SummarizerSessionManager(model=FakeSummaryModel(), summarizer=summarizer, auto_summarize=False) + self.session_service = SqlSessionService( + db_url=self.db_url, + summarizer_manager=manager, + session_config=config, + is_async=False, + ) + await self.session_service._sql_storage.create_sql_engine() + memory_config = MemoryServiceConfig(enabled=True) + memory_config.clean_ttl_config() + self.memory_service = SqlMemoryService( + db_url=self.db_url, + memory_service_config=memory_config, + enabled=True, + is_async=False, + ) + await self.memory_service._sql_storage.create_sql_engine() + + +class RedisReplayAdapter(ReplayBackendAdapter): + """Replay adapter backed by an externally configured Redis URL.""" + + name = "redis" + + def __init__(self, *, case: ReplayCase, redis_url: str) -> None: + super().__init__(case=case) + self.redis_url = redis_url + + async def setup(self) -> None: + config = SessionServiceConfig(store_historical_events=True) + config.clean_ttl_config() + summarizer = SessionSummarizer(model=FakeSummaryModel(), keep_recent_count=2) + manager = SummarizerSessionManager(model=FakeSummaryModel(), summarizer=summarizer, auto_summarize=False) + self.session_service = RedisSessionService( + db_url=self.redis_url, + summarizer_manager=manager, + session_config=config, + is_async=False, + ) + memory_config = MemoryServiceConfig(enabled=True) + memory_config.clean_ttl_config() + self.memory_service = RedisMemoryService( + db_url=self.redis_url, + memory_service_config=memory_config, + enabled=True, + is_async=False, + ) + + +class FaultInjectingAdapter: + """Minimal proxy for operation-level fault injection tests.""" + + def __init__(self, adapter: ReplayBackendAdapter, *, fail_after_op_kind: str) -> None: + self.adapter = adapter + self.fail_after_op_kind = fail_after_op_kind + self.triggered = False + + async def apply(self, op: ReplayOp) -> None: + await self.adapter.apply(op) + if not self.triggered and op.kind == self.fail_after_op_kind: + self.triggered = True + raise ReplayInjectedFault(f"Injected fault after {op.kind}") + + +class ReplayInjectedFault(RuntimeError): + """Raised by FaultInjectingAdapter after a configured operation commits.""" diff --git a/tests/sessions/replay_consistency/canonicalize.py b/tests/sessions/replay_consistency/canonicalize.py new file mode 100644 index 00000000..bb5cbf5b --- /dev/null +++ b/tests/sessions/replay_consistency/canonicalize.py @@ -0,0 +1,89 @@ +# 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. + +"""Field-level normalization for replay snapshots.""" + +from __future__ import annotations + +import copy +import unicodedata +from typing import Any + +from .snapshot import Snapshot + + +def canonicalize_snapshot(snapshot: Snapshot | dict[str, Any]) -> dict[str, Any]: + """Return a stable snapshot dict suitable for strict comparison.""" + + data = snapshot.to_dict() if isinstance(snapshot, Snapshot) else copy.deepcopy(snapshot) + data.pop("backend", None) + data.pop("backend_metadata", None) + for session in data.get("sessions", []): + _canonicalize_session(session) + for probe in data.get("memory", []): + for memory in probe.get("memories", []): + _canonicalize_content(memory.get("content")) + memory["timestamp_valid"] = memory.get("timestamp") is not None + memory.pop("timestamp", None) + probe["memories"] = sorted(probe.get("memories", []), key=lambda item: repr(item.get("content"))) + for summary in data.get("summaries", []): + summary["text"] = _normalize_text(summary.get("text", "")) + summary["timestamp_valid"] = summary.get("timestamp") is not None + summary.pop("timestamp", None) + return data + + +def normalize_content_part(part: dict[str, Any]) -> dict[str, Any]: + """Normalize a content part while preserving semantic fields.""" + + normalized = copy.deepcopy(part) + if "text" in normalized: + normalized["text"] = _normalize_text(normalized["text"]) + for key in ("function_call", "function_response"): + if key in normalized and isinstance(normalized[key], dict): + normalized[key] = _sort_json(normalized[key]) + return normalized + + +def _canonicalize_session(session: dict[str, Any]) -> None: + session["state"] = _sort_json(session.get("state", {})) + event_timestamps = [event.get("timestamp") for event in session.get("events", [])] + session["event_timestamps_monotonic"] = _is_monotonic(event_timestamps) + for collection in ("events", "historical_events"): + for event in session.get(collection, []): + event.pop("actual_event_id", None) + _canonicalize_content(event.get("content")) + event["timestamp_valid"] = isinstance(event.get("timestamp"), (int, float)) + event.pop("timestamp", None) + event["actions"] = _sort_json(event.get("actions", {})) + event["function_calls"] = [_sort_json(item) for item in event.get("function_calls", [])] + event["function_responses"] = [_sort_json(item) for item in event.get("function_responses", [])] + + +def _canonicalize_content(content: dict[str, Any] | None) -> None: + if not content: + return + content["parts"] = [normalize_content_part(part) for part in content.get("parts", [])] + + +def _sort_json(value: Any) -> Any: + if isinstance(value, dict): + return {key: _sort_json(value[key]) for key in sorted(value)} + if isinstance(value, list): + return [_sort_json(item) for item in value] + if isinstance(value, str): + return _normalize_text(value) + return value + + +def _normalize_text(value: str) -> str: + return unicodedata.normalize("NFC", value) + + +def _is_monotonic(values: list[Any]) -> bool: + if not all(isinstance(value, (int, float)) for value in values): + return False + return all(left <= right for left, right in zip(values, values[1:])) diff --git a/tests/sessions/replay_consistency/capabilities.py b/tests/sessions/replay_consistency/capabilities.py new file mode 100644 index 00000000..f762fcd5 --- /dev/null +++ b/tests/sessions/replay_consistency/capabilities.py @@ -0,0 +1,55 @@ +# 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 capability declarations for replay consistency reports.""" + +from __future__ import annotations + + +BACKEND_CAPABILITIES = { + "in_memory": { + "session": True, + "event": True, + "state": True, + "memory": True, + "summary": True, + "persistent": False, + "external_service": False, + }, + "sqlite": { + "session": True, + "event": True, + "state": True, + "memory": True, + "summary": True, + "persistent": True, + "external_service": False, + }, + "sql": { + "session": True, + "event": True, + "state": True, + "memory": True, + "summary": True, + "persistent": True, + "external_service": True, + }, + "redis": { + "session": True, + "event": True, + "state": True, + "memory": True, + "summary": True, + "persistent": True, + "external_service": True, + }, +} + + +def capabilities_for(*backend_names: str) -> dict[str, dict[str, bool]]: + """Return report-friendly capabilities for the requested backends.""" + + return {name: BACKEND_CAPABILITIES[name] for name in backend_names} diff --git a/tests/sessions/replay_consistency/cases.py b/tests/sessions/replay_consistency/cases.py new file mode 100644 index 00000000..bd5c7556 --- /dev/null +++ b/tests/sessions/replay_consistency/cases.py @@ -0,0 +1,280 @@ +# 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. + +"""Deterministic replay cases for session/memory/summary consistency tests.""" + +from __future__ import annotations + +from dataclasses import dataclass +from dataclasses import field +from typing import Any + + +@dataclass(frozen=True) +class ReplayOp: + """One logical operation in a replay case.""" + + kind: str + session_id: str = "s1" + user_id: str = "user1" + app_name: str = "replay_app" + client_event_id: str | None = None + client_summary_id: str | None = None + author: str = "user" + text: str = "" + state_delta: dict[str, Any] = field(default_factory=dict) + initial_state: dict[str, Any] = field(default_factory=dict) + function_call_id: str | None = None + function_name: str = "lookup" + function_args: dict[str, Any] = field(default_factory=dict) + function_response: dict[str, Any] = field(default_factory=dict) + query: str = "" + probe_id: str | None = None + + +@dataclass(frozen=True) +class ReplayCase: + """A deterministic replay case that can be executed against any adapter.""" + + case_id: str + operations: tuple[ReplayOp, ...] + checkpoints: tuple[str, ...] = ("final",) + expected_entities: tuple[str, ...] = ("events", "state", "memory", "summary") + description: str = "" + + +def _create(session_id: str = "s1", user_id: str = "user1", state: dict[str, Any] | None = None) -> ReplayOp: + return ReplayOp(kind="create_session", session_id=session_id, user_id=user_id, initial_state=state or {}) + + +def _text( + client_event_id: str, + text: str, + *, + session_id: str = "s1", + user_id: str = "user1", + author: str = "user", + state_delta: dict[str, Any] | None = None, +) -> ReplayOp: + return ReplayOp( + kind="append_text", + session_id=session_id, + user_id=user_id, + client_event_id=client_event_id, + author=author, + text=text, + state_delta=state_delta or {}, + ) + + +def _store(session_id: str = "s1", user_id: str = "user1") -> ReplayOp: + return ReplayOp(kind="store_memory", session_id=session_id, user_id=user_id) + + +def _probe(probe_id: str, query: str, *, session_id: str = "s1", user_id: str = "user1") -> ReplayOp: + return ReplayOp(kind="search_memory", session_id=session_id, user_id=user_id, probe_id=probe_id, query=query) + + +def _summary(client_summary_id: str, *, session_id: str = "s1", user_id: str = "user1") -> ReplayOp: + return ReplayOp(kind="summarize", session_id=session_id, user_id=user_id, client_summary_id=client_summary_id) + + +def standard_cases() -> list[ReplayCase]: + """Return the public deterministic replay cases. + + The cases intentionally use unique searchable words so memory result order + does not become an accidental backend-specific assertion. + """ + + return [ + ReplayCase( + case_id="single_turn_text", + description="One user message and one assistant response.", + operations=( + _create(), + _text("u1", "alpha preference is tea", author="user"), + _text("a1", "noted alpha preference", author="assistant"), + _store(), + _probe("alpha", "alpha"), + ), + ), + ReplayCase( + case_id="multi_turn_text", + description="Three deterministic user/assistant turns.", + operations=( + _create(), + _text("u1", "bravo question one", author="user"), + _text("a1", "bravo answer one", author="assistant"), + _text("u2", "charlie question two", author="user"), + _text("a2", "charlie answer two", author="assistant"), + _text("u3", "delta question three", author="user"), + _text("a3", "delta answer three", author="assistant"), + _store(), + _probe("delta", "delta"), + ), + ), + ReplayCase( + case_id="tool_call_response", + description="Function call, matching response, and an error-shaped response payload.", + operations=( + _create(), + ReplayOp( + kind="append_tool_call", + client_event_id="tool_call", + author="assistant", + function_call_id="call-weather-1", + function_name="weather", + function_args={"city": "Shenzhen", "unit": "celsius"}, + ), + ReplayOp( + kind="append_tool_response", + client_event_id="tool_response", + author="user", + function_call_id="call-weather-1", + function_name="weather", + function_response={"temperature": 28, "condition": "sunny"}, + ), + ReplayOp( + kind="append_tool_response", + client_event_id="tool_error", + author="user", + function_call_id="call-weather-1", + function_name="weather", + function_response={"error": "rate_limited"}, + ), + ), + ), + ReplayCase( + case_id="state_shallow_update", + description="Session, app, and user state updates with shallow replacement semantics.", + operations=( + _create(state={"profile": {"name": "Ada", "city": "SZ"}, "counter": 1}), + _text("state1", "echo state start", state_delta={"counter": 2}), + _text("state2", "echo nested replace", state_delta={"profile": {"city": "BJ"}}), + _text("state3", "echo nullable", state_delta={"nullable": None}), + ), + ), + ReplayCase( + case_id="memory_scope_user_session", + description="Same user across sessions and different user isolation.", + operations=( + _create("s1", "user1"), + _create("s2", "user1"), + _create("s3", "user2"), + _text("s1u", "foxtrot user one session one", session_id="s1", user_id="user1"), + _text("s2u", "golf user one session two", session_id="s2", user_id="user1"), + _text("s3u", "hotel user two private", session_id="s3", user_id="user2"), + _store("s1", "user1"), + _store("s2", "user1"), + _store("s3", "user2"), + _probe("same_user_s1", "foxtrot", session_id="s1", user_id="user1"), + _probe("same_user_s2", "golf", session_id="s1", user_id="user1"), + _probe("other_user", "hotel", session_id="s3", user_id="user2"), + _probe("no_cross_user", "hotel", session_id="s1", user_id="user1"), + ), + ), + ReplayCase( + case_id="summary_create_update", + description="Create and then update a session summary.", + operations=( + _create(), + _text("u1", "india summary turn one", author="user"), + _text("a1", "india assistant one", author="assistant"), + _text("u2", "juliet summary turn two", author="user"), + _text("a2", "juliet assistant two", author="assistant"), + _summary("sum1"), + _text("u3", "kilo after first summary", author="user"), + _text("a3", "kilo assistant after summary", author="assistant"), + _summary("sum2"), + ), + ), + ReplayCase( + case_id="summary_event_truncation", + description="Summary plus retained recent events, followed by new events.", + operations=( + _create(), + _text("u1", "lima old one", author="user"), + _text("a1", "lima old answer", author="assistant"), + _text("u2", "mike old two", author="user"), + _text("a2", "mike old answer", author="assistant"), + _text("u3", "november recent", author="user"), + _summary("sum1"), + _text("a3", "november new answer", author="assistant"), + _store(), + _probe("november", "november"), + ), + ), + ReplayCase( + case_id="failure_retry_append", + description="Same logical event retried after an ambiguous acknowledgement.", + operations=( + _create(), + _text("retry1", "oscar retry logical event", author="user"), + _text("retry1_retry", "oscar retry logical event", author="user"), + _store(), + _probe("oscar", "oscar"), + ), + ), + ReplayCase( + case_id="cross_session_isolation", + description="Two sessions for one user with separate state and summaries.", + operations=( + _create("s1", "user1", {"topic": "papa"}), + _create("s2", "user1", {"topic": "quebec"}), + _text("s1u1", "papa session one user", session_id="s1", user_id="user1"), + _text("s1a1", "papa session one assistant", session_id="s1", user_id="user1", author="assistant"), + _text("s1u2", "papa session one second", session_id="s1", user_id="user1"), + _summary("s1sum", session_id="s1", user_id="user1"), + _text("s2u1", "quebec session two user", session_id="s2", user_id="user1"), + _text("s2a1", "quebec session two assistant", session_id="s2", user_id="user1", author="assistant"), + _text("s2u2", "quebec session two second", session_id="s2", user_id="user1"), + _summary("s2sum", session_id="s2", user_id="user1"), + ), + ), + ReplayCase( + case_id="summary_defect_specials", + description="Dense summary scenario used by summary mutation tests.", + operations=( + _create(), + _text("u1", "romeo summary one", author="user"), + _text("a1", "romeo answer one", author="assistant"), + _text("u2", "sierra summary two", author="user"), + _text("a2", "sierra answer two", author="assistant"), + _summary("sum1"), + _text("u3", "tango summary three", author="user"), + _text("a3", "tango answer three", author="assistant"), + _summary("sum2"), + ), + ), + ReplayCase( + case_id="all_entities_contract", + description="Compact case with text, tool, state, memory, and summary entities.", + operations=( + _create(state={"contract": "start"}), + _text("u1", "uniform all entity user", state_delta={"contract": "updated"}), + ReplayOp( + kind="append_tool_call", + client_event_id="tc1", + author="assistant", + function_call_id="call-contract-1", + function_name="contract_tool", + function_args={"value": "uniform"}, + ), + ReplayOp( + kind="append_tool_response", + client_event_id="tr1", + author="user", + function_call_id="call-contract-1", + function_name="contract_tool", + function_response={"ok": True}, + ), + _text("a1", "uniform all entity assistant", author="assistant"), + _summary("sum1"), + _store(), + _probe("uniform", "uniform"), + ), + ), + ] diff --git a/tests/sessions/replay_consistency/diff.py b/tests/sessions/replay_consistency/diff.py new file mode 100644 index 00000000..39ebe613 --- /dev/null +++ b/tests/sessions/replay_consistency/diff.py @@ -0,0 +1,264 @@ +# 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. + +"""Contract-aware diffing for canonical replay snapshots.""" + +from __future__ import annotations + +from dataclasses import asdict +from dataclasses import dataclass +from typing import Any + + +@dataclass(frozen=True) +class AllowedDiffRule: + """An explicit allowed backend difference rule.""" + + backend_pair: tuple[str, str] + field_path: str + comparator: str + reason: str + rule_id: str = "" + still_validate: str = "" + + +@dataclass +class DiffEntry: + """One structured replay diff.""" + + case_id: str + backend_pair: tuple[str, str] + session_id: str | None + entity_type: str + entity_id: str | None + index: int | None + field_path: str + reference_value: Any + actual_value: Any + allowed: bool + category: str + reason: str + allowed_rule_id: str = "" + + def to_dict(self) -> dict[str, Any]: + return asdict(self) + + +def compare_snapshots( + reference: dict[str, Any], + actual: dict[str, Any], + *, + case_id: str, + backend_pair: tuple[str, str], + allowed_diff_rules: list[AllowedDiffRule] | None = None, +) -> list[DiffEntry]: + """Compare two canonical snapshots and return structured diffs.""" + + rules = allowed_diff_rules or [] + validate_allowed_diff_rules(rules) + diffs: list[DiffEntry] = [] + _compare_value(reference, actual, "$", diffs, case_id, backend_pair, rules) + diffs.extend(_semantic_diffs(actual, case_id, backend_pair)) + return diffs + + +def classify_diff(field_path: str, reference_value: Any, actual_value: Any) -> str: + """Classify a low-level value mismatch into a replay category.""" + + if reference_value is _MISSING: + return "unexpected_entity" + if actual_value is _MISSING: + if ".summaries" in field_path: + return "summary_missing" + return "missing_entity" + if ".events" in field_path and ".event_id" in field_path: + return "event_order_mismatch" + if ".state" in field_path: + return "state_mismatch" + if ".memory" in field_path: + return "memory_scope_violation" if "session_key" in field_path else "missing_entity" + if ".summaries" in field_path: + if ".session_id" in field_path: + return "summary_owner_mismatch" + if ".version" in field_path or ".active" in field_path: + return "summary_version_mismatch" + if ".covered_event_ids" in field_path: + return "summary_coverage_mismatch" + return "summary_missing" + if "function_response" in field_path or "function_call" in field_path: + return "tool_link_mismatch" + if "timestamp" in field_path: + return "invalid_timestamp" + return "content_mismatch" + + +def validate_allowed_diff_rules(rules: list[AllowedDiffRule]) -> None: + """Reject stale-prone or under-specified allowed-diff rules.""" + + for rule in rules: + if not rule.reason: + raise ValueError("allowed_diff rule must include a reason") + if not rule.field_path.startswith("$"): + raise ValueError(f"allowed_diff rule must use an absolute field path: {rule.field_path}") + if "*" in rule.field_path: + raise ValueError(f"allowed_diff wildcard paths are too broad: {rule.field_path}") + if rule.field_path in {"$", "$.sessions", "$.memory", "$.summaries"}: + raise ValueError(f"allowed_diff path is too broad: {rule.field_path}") + if rule.comparator not in {"exact_path", "prefix"}: + raise ValueError(f"unsupported allowed_diff comparator: {rule.comparator}") + + +def unused_allowed_diff_rules(diffs: list[DiffEntry], rules: list[AllowedDiffRule]) -> list[str]: + """Return configured allowed-diff rule ids that did not match any diff.""" + + used_rule_ids = {diff.allowed_rule_id for diff in diffs if diff.allowed_rule_id} + unused = [] + for idx, rule in enumerate(rules): + rule_id = rule.rule_id or f"{rule.backend_pair}:{rule.field_path}:{idx}" + if rule_id not in used_rule_ids: + unused.append(rule_id) + return unused + + +class _Missing: + pass + + +_MISSING = _Missing() + + +def _compare_value( + reference: Any, + actual: Any, + path: str, + diffs: list[DiffEntry], + case_id: str, + backend_pair: tuple[str, str], + rules: list[AllowedDiffRule], +) -> None: + if isinstance(reference, dict) and isinstance(actual, dict): + for key in sorted(set(reference) | set(actual)): + _compare_value( + reference.get(key, _MISSING), + actual.get(key, _MISSING), + f"{path}.{key}", + diffs, + case_id, + backend_pair, + rules, + ) + return + if isinstance(reference, list) and isinstance(actual, list): + max_len = max(len(reference), len(actual)) + for idx in range(max_len): + _compare_value( + reference[idx] if idx < len(reference) else _MISSING, + actual[idx] if idx < len(actual) else _MISSING, + f"{path}[{idx}]", + diffs, + case_id, + backend_pair, + rules, + ) + return + if reference != actual: + allowed, reason, rule_id = _is_allowed(path, backend_pair, rules) + diffs.append( + DiffEntry( + case_id=case_id, + backend_pair=backend_pair, + session_id=_extract_session_id(path, reference, actual), + entity_type=_entity_type(path), + entity_id=_extract_entity_id(reference, actual), + index=_extract_index(path), + field_path=path, + reference_value=None if reference is _MISSING else reference, + actual_value=None if actual is _MISSING else actual, + allowed=allowed, + category="allowed_backend_difference" if allowed else classify_diff(path, reference, actual), + reason=reason, + allowed_rule_id=rule_id, + ) + ) + + +def _semantic_diffs(snapshot: dict[str, Any], case_id: str, backend_pair: tuple[str, str]) -> list[DiffEntry]: + diffs: list[DiffEntry] = [] + for session in snapshot.get("sessions", []): + seen = set() + for idx, event in enumerate(session.get("events", [])): + event_id = event.get("event_id") + if event_id in seen: + diffs.append( + DiffEntry( + case_id=case_id, + backend_pair=backend_pair, + session_id=session.get("session_id"), + entity_type="event", + entity_id=event_id, + index=idx, + field_path=f"$.sessions.{session.get('session_id')}.events[{idx}].event_id", + reference_value="unique", + actual_value=event_id, + allowed=False, + category="duplicate_event", + reason="event id appears more than once in one session", + ) + ) + seen.add(event_id) + return diffs + + +def _is_allowed(path: str, backend_pair: tuple[str, str], rules: list[AllowedDiffRule]) -> tuple[bool, str, str]: + for idx, rule in enumerate(rules): + if rule.backend_pair != backend_pair: + continue + if rule.comparator == "exact_path" and path != rule.field_path: + continue + if rule.comparator == "prefix" and not path.startswith(rule.field_path): + continue + rule_id = rule.rule_id or f"{rule.backend_pair}:{rule.field_path}:{idx}" + return True, rule.reason, rule_id + return False, "", "" + + +def _entity_type(path: str) -> str: + if ".summaries" in path: + return "summary" + if ".memory" in path: + return "memory" + if ".events" in path or ".historical_events" in path: + return "event" + if ".state" in path: + return "state" + return "snapshot" + + +def _extract_index(path: str) -> int | None: + marker = path.rfind("[") + end = path.rfind("]") + if marker == -1 or end == -1 or end < marker: + return None + try: + return int(path[marker + 1:end]) + except ValueError: + return None + + +def _extract_entity_id(reference: Any, actual: Any) -> str | None: + for value in (actual, reference): + if isinstance(value, dict): + for key in ("event_id", "client_summary_id", "probe_id"): + if key in value: + return str(value[key]) + return None + + +def _extract_session_id(path: str, reference: Any, actual: Any) -> str | None: + for value in (actual, reference): + if isinstance(value, dict) and "session_id" in value: + return str(value["session_id"]) + return None diff --git a/tests/sessions/replay_consistency/mutations.py b/tests/sessions/replay_consistency/mutations.py new file mode 100644 index 00000000..0dbe3bdb --- /dev/null +++ b/tests/sessions/replay_consistency/mutations.py @@ -0,0 +1,131 @@ +# 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. + +"""Deterministic mutation operators for replay diff detection tests.""" + +from __future__ import annotations + +import copy +from dataclasses import dataclass +from typing import Any +from typing import Callable + + +SnapshotDict = dict[str, Any] +MutationFn = Callable[[SnapshotDict], SnapshotDict] + + +@dataclass(frozen=True) +class MutationOperator: + """A public mutation that must be detected by the diff engine.""" + + name: str + expected_category: str + mutate: MutationFn + summary_defect: str | None = None + + +def registered_mutations() -> list[MutationOperator]: + """Return deterministic mutations with known expected categories.""" + + return [ + MutationOperator("delete_event", "missing_entity", _delete_event), + MutationOperator("duplicate_event", "duplicate_event", _duplicate_event), + MutationOperator("swap_event_order", "event_order_mismatch", _swap_event_order), + MutationOperator("change_state_value", "state_mismatch", _change_state_value), + MutationOperator("delete_memory", "missing_entity", _delete_memory), + MutationOperator("memory_cross_user", "memory_scope_violation", _memory_cross_user), + MutationOperator("delete_summary", "summary_missing", _delete_summary, "loss"), + MutationOperator("summary_wrong_owner", "summary_owner_mismatch", _summary_wrong_owner, "owner"), + MutationOperator("summary_wrong_coverage", "summary_coverage_mismatch", _summary_wrong_coverage), + MutationOperator( + "old_summary_overwrites_new", + "summary_version_mismatch", + _old_summary_overwrites_new, + "overwrite", + ), + MutationOperator("wrong_function_response_id", "tool_link_mismatch", _wrong_function_response_id), + ] + + +def _copy(snapshot: SnapshotDict) -> SnapshotDict: + return copy.deepcopy(snapshot) + + +def _first_session(data: SnapshotDict) -> dict[str, Any]: + return data["sessions"][0] + + +def _delete_event(snapshot: SnapshotDict) -> SnapshotDict: + data = _copy(snapshot) + _first_session(data)["events"].pop(0) + return data + + +def _duplicate_event(snapshot: SnapshotDict) -> SnapshotDict: + data = _copy(snapshot) + session = _first_session(data) + session["events"].append(copy.deepcopy(session["events"][0])) + return data + + +def _swap_event_order(snapshot: SnapshotDict) -> SnapshotDict: + data = _copy(snapshot) + events = _first_session(data)["events"] + events[0], events[1] = events[1], events[0] + return data + + +def _change_state_value(snapshot: SnapshotDict) -> SnapshotDict: + data = _copy(snapshot) + _first_session(data)["state"]["contract"] = "mutated" + return data + + +def _delete_memory(snapshot: SnapshotDict) -> SnapshotDict: + data = _copy(snapshot) + data["memory"][0]["memories"] = [] + return data + + +def _memory_cross_user(snapshot: SnapshotDict) -> SnapshotDict: + data = _copy(snapshot) + data["memory"][0]["session_key"] = "replay_app/other_user" + return data + + +def _delete_summary(snapshot: SnapshotDict) -> SnapshotDict: + data = _copy(snapshot) + data["summaries"] = [] + return data + + +def _summary_wrong_owner(snapshot: SnapshotDict) -> SnapshotDict: + data = _copy(snapshot) + data["summaries"][-1]["session_id"] = "wrong-session" + return data + + +def _summary_wrong_coverage(snapshot: SnapshotDict) -> SnapshotDict: + data = _copy(snapshot) + data["summaries"][-1]["covered_event_ids"] = ["wrong-event"] + return data + + +def _old_summary_overwrites_new(snapshot: SnapshotDict) -> SnapshotDict: + data = _copy(snapshot) + data["summaries"][-1]["version"] = 0 + return data + + +def _wrong_function_response_id(snapshot: SnapshotDict) -> SnapshotDict: + data = _copy(snapshot) + for session in data["sessions"]: + for event in session["events"] + session["historical_events"]: + for response in event.get("function_responses", []): + response["id"] = "wrong-call-id" + return data + return data diff --git a/tests/sessions/replay_consistency/oracle.py b/tests/sessions/replay_consistency/oracle.py new file mode 100644 index 00000000..b481ab02 --- /dev/null +++ b/tests/sessions/replay_consistency/oracle.py @@ -0,0 +1,72 @@ +# 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. + +"""Semantic oracle checks for canonical replay snapshots.""" + +from __future__ import annotations + +from typing import Any + + +def validate_snapshot_contract(snapshot: dict[str, Any]) -> list[str]: + """Return human-readable contract violations for a canonical snapshot.""" + + violations: list[str] = [] + _validate_sessions(snapshot, violations) + _validate_memory(snapshot, violations) + _validate_summaries(snapshot, violations) + return violations + + +def _validate_sessions(snapshot: dict[str, Any], violations: list[str]) -> None: + for session in snapshot.get("sessions", []): + session_id = session.get("session_id") + seen_event_ids: set[str] = set() + previous_index = -1 + for event in session.get("events", []): + event_id = event.get("event_id") + if event_id in seen_event_ids: + violations.append(f"duplicate event id {event_id} in session {session_id}") + seen_event_ids.add(event_id) + index = event.get("index") + if not isinstance(index, int) or index <= previous_index: + violations.append(f"event index is not strictly increasing in session {session_id}") + previous_index = index + if not event.get("timestamp_valid"): + violations.append(f"invalid event timestamp for {event_id} in session {session_id}") + if not isinstance(session.get("state", {}), dict): + violations.append(f"state is not a dict in session {session_id}") + + +def _validate_memory(snapshot: dict[str, Any], violations: list[str]) -> None: + for probe in snapshot.get("memory", []): + if "/" not in probe.get("session_key", ""): + violations.append(f"memory probe {probe.get('probe_id')} has invalid session key") + for index, memory in enumerate(probe.get("memories", [])): + if not memory.get("timestamp_valid"): + violations.append(f"memory probe {probe.get('probe_id')} hit {index} has invalid timestamp") + if not memory.get("content"): + violations.append(f"memory probe {probe.get('probe_id')} hit {index} has empty content") + + +def _validate_summaries(snapshot: dict[str, Any], violations: list[str]) -> None: + active_by_session: dict[tuple[str, str, str], int] = {} + versions_by_session: dict[tuple[str, str, str], list[int]] = {} + for summary in snapshot.get("summaries", []): + key = (summary.get("app_name"), summary.get("user_id"), summary.get("session_id")) + if summary.get("active"): + active_by_session[key] = active_by_session.get(key, 0) + 1 + versions_by_session.setdefault(key, []).append(summary.get("version")) + if not summary.get("timestamp_valid"): + violations.append(f"summary {summary.get('client_summary_id')} has invalid timestamp") + if not summary.get("covered_event_ids"): + violations.append(f"summary {summary.get('client_summary_id')} has empty coverage") + for key, active_count in active_by_session.items(): + if active_count != 1: + violations.append(f"session {key} has {active_count} active summaries") + for key, versions in versions_by_session.items(): + if versions != sorted(versions) or len(set(versions)) != len(versions): + violations.append(f"session {key} summary versions are not strictly increasing") diff --git a/tests/sessions/replay_consistency/replay_cases_manifest.json b/tests/sessions/replay_consistency/replay_cases_manifest.json new file mode 100644 index 00000000..4436e52c --- /dev/null +++ b/tests/sessions/replay_consistency/replay_cases_manifest.json @@ -0,0 +1,68 @@ +[ + { + "case_id": "single_turn_text", + "description": "One user message and one assistant response.", + "operations": ["create_session", "append_text", "append_text", "store_memory", "search_memory"], + "covers": ["event", "memory"] + }, + { + "case_id": "multi_turn_text", + "description": "Three deterministic user/assistant turns.", + "operations": ["create_session", "append_text", "append_text", "append_text", "append_text", "append_text", "append_text", "store_memory", "search_memory"], + "covers": ["event", "memory"] + }, + { + "case_id": "tool_call_response", + "description": "Function call, matching response, and an error-shaped response payload.", + "operations": ["create_session", "append_tool_call", "append_tool_response", "append_tool_response"], + "covers": ["event", "tool_call", "tool_response"] + }, + { + "case_id": "state_shallow_update", + "description": "Session, app, and user state updates with shallow replacement semantics.", + "operations": ["create_session", "append_text", "append_text", "append_text"], + "covers": ["event", "state"] + }, + { + "case_id": "memory_scope_user_session", + "description": "Same user across sessions and different user isolation.", + "operations": ["create_session", "create_session", "create_session", "append_text", "append_text", "append_text", "store_memory", "store_memory", "store_memory", "search_memory", "search_memory", "search_memory", "search_memory"], + "covers": ["event", "memory", "cross_user_isolation", "cross_session_scope"] + }, + { + "case_id": "summary_create_update", + "description": "Create and then update a session summary.", + "operations": ["create_session", "append_text", "append_text", "append_text", "append_text", "summarize", "append_text", "append_text", "summarize"], + "covers": ["event", "summary", "summary_overwrite"] + }, + { + "case_id": "summary_event_truncation", + "description": "Summary plus retained recent events, followed by new events.", + "operations": ["create_session", "append_text", "append_text", "append_text", "append_text", "append_text", "summarize", "append_text", "store_memory", "search_memory"], + "covers": ["event", "memory", "summary", "historical_events"] + }, + { + "case_id": "failure_retry_append", + "description": "Same logical event retried after an ambiguous acknowledgement.", + "operations": ["create_session", "append_text", "append_text", "store_memory", "search_memory"], + "covers": ["event", "memory", "duplicate_detection"] + }, + { + "case_id": "cross_session_isolation", + "description": "Two sessions for one user with separate state and summaries.", + "operations": ["create_session", "create_session", "append_text", "append_text", "append_text", "summarize", "append_text", "append_text", "append_text", "summarize"], + "covers": ["event", "state", "summary", "cross_session_isolation"] + }, + { + "case_id": "summary_defect_specials", + "description": "Dense summary scenario used by summary mutation tests.", + "operations": ["create_session", "append_text", "append_text", "append_text", "append_text", "summarize", "append_text", "append_text", "summarize"], + "covers": ["event", "summary", "summary_loss", "summary_overwrite", "summary_owner_error"] + }, + { + "case_id": "all_entities_contract", + "description": "Compact case with text, tool, state, memory, and summary entities.", + "operations": ["create_session", "append_text", "append_tool_call", "append_tool_response", "append_text", "summarize", "store_memory", "search_memory"], + "covers": ["event", "state", "memory", "summary", "tool_call", "tool_response"] + } +] diff --git a/tests/sessions/replay_consistency/report_schema.json b/tests/sessions/replay_consistency/report_schema.json new file mode 100644 index 00000000..cf2ed557 --- /dev/null +++ b/tests/sessions/replay_consistency/report_schema.json @@ -0,0 +1,75 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "title": "SessionMemorySummaryReplayReport", + "type": "object", + "required": [ + "schema_version", + "harness_version", + "mode", + "case_id", + "backend_pair", + "metrics", + "diffs", + "unexpected_diffs", + "allowed_diffs" + ], + "properties": { + "schema_version": {"type": "string"}, + "harness_version": {"type": "string"}, + "generated_at": {"type": "string"}, + "mode": {"type": "string"}, + "case_id": {"type": "string"}, + "backend_pair": { + "type": "array", + "minItems": 2, + "maxItems": 2, + "items": {"type": "string"} + }, + "capabilities": {"type": ["object", "null"]}, + "environment": {"type": ["object", "null"]}, + "metrics": {"type": "object"}, + "diffs": { + "type": "array", + "items": {"$ref": "#/$defs/diff"} + }, + "unexpected_diffs": { + "type": "array", + "items": {"$ref": "#/$defs/diff"} + }, + "allowed_diffs": { + "type": "array", + "items": {"$ref": "#/$defs/diff"} + } + }, + "$defs": { + "diff": { + "type": "object", + "required": [ + "case_id", + "backend_pair", + "entity_type", + "field_path", + "reference_value", + "actual_value", + "allowed", + "category", + "reason" + ], + "properties": { + "case_id": {"type": "string"}, + "backend_pair": {"type": "array"}, + "session_id": {"type": ["string", "null"]}, + "entity_type": {"type": "string"}, + "entity_id": {"type": ["string", "null"]}, + "index": {"type": ["integer", "null"]}, + "field_path": {"type": "string"}, + "reference_value": {}, + "actual_value": {}, + "allowed": {"type": "boolean"}, + "category": {"type": "string"}, + "reason": {"type": "string"}, + "allowed_rule_id": {"type": "string"} + } + } + } +} diff --git a/tests/sessions/replay_consistency/reporting.py b/tests/sessions/replay_consistency/reporting.py new file mode 100644 index 00000000..625409f8 --- /dev/null +++ b/tests/sessions/replay_consistency/reporting.py @@ -0,0 +1,144 @@ +# 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 report models and writers.""" + +from __future__ import annotations + +import json +from dataclasses import asdict +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +from .diff import DiffEntry + + +@dataclass +class Metrics: + """Automatically computed replay acceptance metrics.""" + + normal_case_count: int = 0 + normal_case_pass_count: int = 0 + normal_cases_with_unexpected_diff: int = 0 + unexpected_diff_count: int = 0 + allowed_diff_count: int = 0 + false_positive_rate: float = 0.0 + mutation_total: int = 0 + mutation_detected: int = 0 + survived_mutations: list[str] | None = None + mutation_detection_rate: float = 0.0 + summary_loss_detected: int = 0 + summary_overwrite_detected: int = 0 + summary_owner_error_detected: int = 0 + summary_loss_detection_rate: float = 0.0 + summary_overwrite_detection_rate: float = 0.0 + summary_owner_error_detection_rate: float = 0.0 + runtime_fault_total: int = 0 + runtime_fault_detected: int = 0 + runtime_fault_detection_rate: float = 0.0 + lightweight_duration_seconds: float = 0.0 + + +@dataclass +class ReplayReport: + """Structured JSON report for replay consistency runs.""" + + case_id: str + backend_pair: tuple[str, str] + metrics: Metrics + diffs: list[dict[str, Any]] + schema_version: str = "1.0" + harness_version: str = "1" + mode: str = "lightweight" + capabilities: dict[str, Any] | None = None + environment: dict[str, Any] | None = None + generated_at: str = "" + + def to_dict(self) -> dict[str, Any]: + data = asdict(self) + data["unexpected_diffs"] = [diff for diff in self.diffs if not diff.get("allowed")] + data["allowed_diffs"] = [diff for diff in self.diffs if diff.get("allowed")] + data["unused_allowed_diff_rules"] = data.get("unused_allowed_diff_rules", []) + return data + + +def build_metrics( + *, + normal_case_count: int, + normal_case_pass_count: int, + diffs: list[DiffEntry], + mutation_total: int = 0, + mutation_detected: int = 0, + summary_loss_detected: int = 0, + summary_overwrite_detected: int = 0, + summary_owner_error_detected: int = 0, + summary_loss_total: int = 0, + summary_overwrite_total: int = 0, + summary_owner_error_total: int = 0, + runtime_fault_total: int = 0, + runtime_fault_detected: int = 0, + lightweight_duration_seconds: float = 0.0, + survived_mutations: list[str] | None = None, +) -> Metrics: + unexpected = [diff for diff in diffs if not diff.allowed] + allowed = [diff for diff in diffs if diff.allowed] + false_positive_rate = 0.0 + if normal_case_count: + false_positive_rate = (normal_case_count - normal_case_pass_count) / normal_case_count + mutation_detection_rate = 0.0 + if mutation_total: + mutation_detection_rate = mutation_detected / mutation_total + summary_loss_detection_rate = _rate(summary_loss_detected, summary_loss_total) + summary_overwrite_detection_rate = _rate(summary_overwrite_detected, summary_overwrite_total) + summary_owner_error_detection_rate = _rate(summary_owner_error_detected, summary_owner_error_total) + runtime_fault_detection_rate = _rate(runtime_fault_detected, runtime_fault_total) + return Metrics( + normal_case_count=normal_case_count, + normal_case_pass_count=normal_case_pass_count, + normal_cases_with_unexpected_diff=normal_case_count - normal_case_pass_count, + unexpected_diff_count=len(unexpected), + allowed_diff_count=len(allowed), + false_positive_rate=false_positive_rate, + mutation_total=mutation_total, + mutation_detected=mutation_detected, + survived_mutations=survived_mutations or [], + mutation_detection_rate=mutation_detection_rate, + summary_loss_detected=summary_loss_detected, + summary_overwrite_detected=summary_overwrite_detected, + summary_owner_error_detected=summary_owner_error_detected, + summary_loss_detection_rate=summary_loss_detection_rate, + summary_overwrite_detection_rate=summary_overwrite_detection_rate, + summary_owner_error_detection_rate=summary_owner_error_detection_rate, + runtime_fault_total=runtime_fault_total, + runtime_fault_detected=runtime_fault_detected, + runtime_fault_detection_rate=runtime_fault_detection_rate, + lightweight_duration_seconds=lightweight_duration_seconds, + ) + + +def write_json_report(path: Path, report: ReplayReport) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(report.to_dict(), indent=2, sort_keys=True), encoding="utf-8") + + +def write_markdown_report(path: Path, report: ReplayReport) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + lines = [ + f"# Replay Report: {report.case_id}", + "", + f"- Backend pair: {report.backend_pair[0]} vs {report.backend_pair[1]}", + f"- Unexpected diffs: {report.metrics.unexpected_diff_count}", + f"- Allowed diffs: {report.metrics.allowed_diff_count}", + f"- Mutation detection rate: {report.metrics.mutation_detection_rate:.2%}", + ] + path.write_text("\n".join(lines) + "\n", encoding="utf-8") + + +def _rate(detected: int, total: int) -> float: + if total == 0: + return 0.0 + return detected / total diff --git a/tests/sessions/replay_consistency/snapshot.py b/tests/sessions/replay_consistency/snapshot.py new file mode 100644 index 00000000..d5353c14 --- /dev/null +++ b/tests/sessions/replay_consistency/snapshot.py @@ -0,0 +1,166 @@ +# 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. + +"""Snapshot readers for replay consistency comparisons.""" + +from __future__ import annotations + +from dataclasses import asdict +from dataclasses import dataclass +from typing import Any + +from trpc_agent_sdk.events import Event +from trpc_agent_sdk.sessions import Session + + +@dataclass +class MemoryProbe: + """A memory query to include in a snapshot.""" + + probe_id: str + session_key: str + query: str + + +@dataclass +class SummaryRecord: + """Derived summary metadata observed by the replay adapter.""" + + client_summary_id: str + session_id: str + user_id: str + app_name: str + event_id: str + text: str + version: int + active: bool + covered_event_ids: list[str] + timestamp: float + + +@dataclass +class Snapshot: + """Stable snapshot shape consumed by canonicalization and diffing.""" + + case_id: str + backend: str + sessions: list[dict[str, Any]] + memory: list[dict[str, Any]] + summaries: list[dict[str, Any]] + backend_metadata: dict[str, Any] + + def to_dict(self) -> dict[str, Any]: + return asdict(self) + + +async def read_snapshot( + *, + backend: str, + case_id: str, + session_service, + memory_service, + sessions: list[Session], + actual_to_client_event_id: dict[str, str], + memory_probes: list[MemoryProbe], + summary_records: list[SummaryRecord], +) -> Snapshot: + """Read a backend snapshot using the public service APIs.""" + + session_snapshots = [] + for original in sessions: + session = await session_service.get_session( + app_name=original.app_name, + user_id=original.user_id, + session_id=original.id, + ) + if session is None: + continue + session_snapshots.append(_session_to_snapshot(session, actual_to_client_event_id)) + + memory_snapshots = [] + for probe in memory_probes: + result = await memory_service.search_memory(probe.session_key, probe.query, limit=10) + memory_snapshots.append( + { + "probe_id": probe.probe_id, + "session_key": probe.session_key, + "query": probe.query, + "memories": [ + { + "author": entry.author, + "timestamp": entry.timestamp, + "content": _content_to_snapshot(entry.content), + } + for entry in result.memories + ], + } + ) + + return Snapshot( + case_id=case_id, + backend=backend, + sessions=sorted(session_snapshots, key=lambda item: (item["app_name"], item["user_id"], item["session_id"])), + memory=sorted(memory_snapshots, key=lambda item: item["probe_id"]), + summaries=[asdict(record) for record in sorted(summary_records, key=lambda item: item.client_summary_id)], + backend_metadata={"backend": backend}, + ) + + +def _session_to_snapshot(session: Session, actual_to_client_event_id: dict[str, str]) -> dict[str, Any]: + return { + "app_name": session.app_name, + "user_id": session.user_id, + "session_id": session.id, + "state": session.state, + "events": [ + _event_to_snapshot(event, idx, actual_to_client_event_id) for idx, event in enumerate(session.events) + ], + "historical_events": [ + _event_to_snapshot(event, idx, actual_to_client_event_id) + for idx, event in enumerate(session.historical_events) + ], + "conversation_count": session.conversation_count, + } + + +def _event_to_snapshot(event: Event, index: int, actual_to_client_event_id: dict[str, str]) -> dict[str, Any]: + return { + "index": index, + "event_id": actual_to_client_event_id.get(event.id, event.id), + "actual_event_id": event.id, + "invocation_id": event.invocation_id, + "author": event.author, + "timestamp": event.timestamp, + "is_summary": event.is_summary_event(), + "is_model_visible": event.is_model_visible(), + "content": _content_to_snapshot(event.content), + "actions": event.actions.model_dump(mode="json", exclude_none=True), + "function_calls": [_function_like_to_dict(call) for call in event.get_function_calls()], + "function_responses": [_function_like_to_dict(response) for response in event.get_function_responses()], + "error_code": event.error_code, + "error_message": event.error_message, + "version": event.version, + } + + +def _content_to_snapshot(content) -> dict[str, Any] | None: + if content is None: + return None + parts = [] + for part in content.parts or []: + item: dict[str, Any] = {} + if part.text is not None: + item["text"] = part.text + if part.function_call is not None: + item["function_call"] = _function_like_to_dict(part.function_call) + if part.function_response is not None: + item["function_response"] = _function_like_to_dict(part.function_response) + parts.append(item) + return {"role": content.role, "parts": parts} + + +def _function_like_to_dict(value) -> dict[str, Any]: + return value.model_dump(mode="json", exclude_none=True, by_alias=False) diff --git a/tests/sessions/replay_consistency/test_allowed_diff.py b/tests/sessions/replay_consistency/test_allowed_diff.py new file mode 100644 index 00000000..582644ac --- /dev/null +++ b/tests/sessions/replay_consistency/test_allowed_diff.py @@ -0,0 +1,63 @@ +# 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. + +"""Allowed-diff governance tests.""" + +from __future__ import annotations + +import pytest + +from .diff import AllowedDiffRule +from .diff import compare_snapshots +from .diff import unused_allowed_diff_rules +from .diff import validate_allowed_diff_rules + + +def test_allowed_diff_reports_unused_rules(): + reference = {"sessions": [{"session_id": "s1", "state": {"value": 1}}], "memory": [], "summaries": []} + actual = {"sessions": [{"session_id": "s1", "state": {"value": 2}}], "memory": [], "summaries": []} + rules = [ + AllowedDiffRule( + backend_pair=("left", "right"), + field_path="$.sessions[0].state.value", + comparator="exact_path", + reason="known backend-local representation difference", + rule_id="used-rule", + ), + AllowedDiffRule( + backend_pair=("left", "right"), + field_path="$.sessions[0].state.other", + comparator="exact_path", + reason="stale example rule", + rule_id="unused-rule", + ), + ] + + diffs = compare_snapshots( + reference, + actual, + case_id="allowed", + backend_pair=("left", "right"), + allowed_diff_rules=rules, + ) + + assert diffs[0].allowed + assert diffs[0].allowed_rule_id == "used-rule" + assert unused_allowed_diff_rules(diffs, rules) == ["unused-rule"] + + +@pytest.mark.parametrize( + "rule", + [ + AllowedDiffRule(("left", "right"), "$.sessions[*].events[*]", "prefix", "too broad"), + AllowedDiffRule(("left", "right"), "$.summaries", "prefix", "too broad"), + AllowedDiffRule(("left", "right"), "$.sessions[0].state.value", "exact_path", ""), + AllowedDiffRule(("left", "right"), "$.sessions[0].state.value", "regex", "unsupported"), + ], +) +def test_allowed_diff_rejects_unsafe_rules(rule): + with pytest.raises(ValueError): + validate_allowed_diff_rules([rule]) diff --git a/tests/sessions/replay_consistency/test_canonicalization.py b/tests/sessions/replay_consistency/test_canonicalization.py new file mode 100644 index 00000000..597553a8 --- /dev/null +++ b/tests/sessions/replay_consistency/test_canonicalization.py @@ -0,0 +1,116 @@ +# 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. + +"""Canonicalization guardrails against over-normalization.""" + +from __future__ import annotations + +import copy + +from .canonicalize import canonicalize_snapshot +from .diff import compare_snapshots + + +def _snapshot(): + return { + "backend": "test", + "sessions": [ + { + "app_name": "app", + "user_id": "user", + "session_id": "s1", + "state": {}, + "events": [ + { + "index": 0, + "event_id": "tool_call", + "actual_event_id": "actual-1", + "timestamp": 1.0, + "content": { + "parts": [ + { + "function_call": { + "id": "call-1", + "name": "tool", + "args": {"b": 2, "a": 1}, + } + } + ] + }, + "actions": {}, + "function_calls": [{"id": "call-1", "name": "tool", "args": {"b": 2, "a": 1}}], + "function_responses": [], + }, + { + "index": 1, + "event_id": "tool_response", + "actual_event_id": "actual-2", + "timestamp": 2.0, + "content": { + "parts": [ + { + "function_response": { + "id": "call-1", + "name": "tool", + "response": {"ok": True}, + } + } + ] + }, + "actions": {}, + "function_calls": [], + "function_responses": [{"id": "call-1", "name": "tool", "response": {"ok": True}}], + }, + ], + "historical_events": [], + "conversation_count": 2, + } + ], + "memory": [], + "summaries": [], + "backend_metadata": {}, + } + + +def test_canonicalization_does_not_hide_event_reordering(): + reference = canonicalize_snapshot(_snapshot()) + reordered_source = _snapshot() + reordered_source["sessions"][0]["events"].reverse() + actual = canonicalize_snapshot(reordered_source) + + categories = { + diff.category + for diff in compare_snapshots(reference, actual, case_id="order", backend_pair=("a", "b")) + } + + assert "event_order_mismatch" in categories + + +def test_canonicalization_preserves_tool_call_linkage(): + reference = canonicalize_snapshot(_snapshot()) + broken_source = copy.deepcopy(reference) + broken_source["sessions"][0]["events"][1]["function_responses"][0]["id"] = "wrong-call" + + categories = { + diff.category + for diff in compare_snapshots(reference, broken_source, case_id="tool", backend_pair=("a", "b")) + } + + assert "tool_link_mismatch" in categories + + +def test_timestamp_normalization_preserves_relative_order_signal(): + reference = canonicalize_snapshot(_snapshot()) + broken_source = _snapshot() + broken_source["sessions"][0]["events"][0]["timestamp"] = 3.0 + broken_source["sessions"][0]["events"][1]["timestamp"] = 2.0 + actual = canonicalize_snapshot(broken_source) + + diffs = compare_snapshots(reference, actual, case_id="time", backend_pair=("a", "b")) + + assert reference["sessions"][0]["event_timestamps_monotonic"] + assert not actual["sessions"][0]["event_timestamps_monotonic"] + assert any(diff.field_path.endswith("event_timestamps_monotonic") for diff in diffs) diff --git a/tests/sessions/replay_consistency/test_replay_consistency.py b/tests/sessions/replay_consistency/test_replay_consistency.py new file mode 100644 index 00000000..b03332f2 --- /dev/null +++ b/tests/sessions/replay_consistency/test_replay_consistency.py @@ -0,0 +1,478 @@ +# 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. + +"""Lightweight replay consistency suite for Issue #89.""" + +from __future__ import annotations + +import asyncio +import json +import os +import time +from pathlib import Path + +import pytest + +from .adapters import InMemoryReplayAdapter +from .adapters import FaultInjectingAdapter +from .adapters import ReplayInjectedFault +from .adapters import SQLiteReplayAdapter +from .canonicalize import canonicalize_snapshot +from .capabilities import capabilities_for +from .cases import ReplayCase +from .cases import standard_cases +from .diff import AllowedDiffRule +from .diff import compare_snapshots +from .mutations import MutationOperator +from .mutations import registered_mutations +from .oracle import validate_snapshot_contract +from .reporting import ReplayReport +from .reporting import build_metrics +from .reporting import write_json_report +from .reporting import write_markdown_report + + +def _run(coro): + return asyncio.run(coro) + + +async def _run_case(case: ReplayCase, tmp_path: Path): + left = InMemoryReplayAdapter(case=case) + right = SQLiteReplayAdapter(case=case, workdir=tmp_path) + await left.setup() + await right.setup() + try: + left_snapshot = await left.replay() + right_snapshot = await right.replay() + return left_snapshot, right_snapshot + finally: + await left.close() + await right.close() + + +def _case_by_id(case_id: str) -> ReplayCase: + return next(case for case in standard_cases() if case.case_id == case_id) + + +@pytest.mark.replay_lightweight +def test_replay_cases_manifest_matches_fixtures(): + manifest_path = Path(__file__).with_name("replay_cases_manifest.json") + manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + cases = {case.case_id: case for case in standard_cases()} + assert [item["case_id"] for item in manifest] == [case.case_id for case in standard_cases()] + for item in manifest: + case = cases[item["case_id"]] + assert item["description"] == case.description + assert item["operations"] == [op.kind for op in case.operations] + + +@pytest.mark.replay_lightweight +def test_cases_are_well_formed(): + cases = standard_cases() + assert len(cases) >= 10 + case_ids = [case.case_id for case in cases] + assert len(case_ids) == len(set(case_ids)) + for case in cases: + assert case.operations + assert all(op.kind for op in case.operations) + + +@pytest.mark.replay_lightweight +def test_snapshot_contains_all_entities(tmp_path): + case = _case_by_id("all_entities_contract") + left_snapshot, _ = _run(_run_case(case, tmp_path)) + data = left_snapshot.to_dict() + assert data["sessions"] + assert data["sessions"][0]["events"] + assert data["sessions"][0]["state"] + assert data["memory"] + assert data["summaries"] + + +@pytest.mark.replay_lightweight +def test_oracle_accepts_standard_snapshots(tmp_path): + for case in standard_cases(): + left_snapshot, right_snapshot = _run(_run_case(case, tmp_path)) + assert validate_snapshot_contract(canonicalize_snapshot(left_snapshot)) == [] + assert validate_snapshot_contract(canonicalize_snapshot(right_snapshot)) == [] + + +@pytest.mark.replay_lightweight +def test_lightweight_replay_cases(tmp_path): + start = time.perf_counter() + all_diffs = [] + passed = 0 + cases = standard_cases() + try: + for case in cases: + left_snapshot, right_snapshot = _run(_run_case(case, tmp_path)) + left = canonicalize_snapshot(left_snapshot) + right = canonicalize_snapshot(right_snapshot) + diffs = compare_snapshots( + left, + right, + case_id=case.case_id, + backend_pair=("in_memory", "sqlite"), + ) + all_diffs.extend(diffs) + if not [diff for diff in diffs if not diff.allowed]: + passed += 1 + duration = time.perf_counter() - start + metrics = build_metrics( + normal_case_count=len(cases), + normal_case_pass_count=passed, + diffs=all_diffs, + lightweight_duration_seconds=duration, + ) + report = ReplayReport( + case_id="lightweight", + backend_pair=("in_memory", "sqlite"), + metrics=metrics, + diffs=[diff.to_dict() for diff in all_diffs], + capabilities=capabilities_for("in_memory", "sqlite"), + ) + _write_report_pair(tmp_path, "session_memory_summary_diff_report", report) + unexpected = [diff for diff in all_diffs if not diff.allowed] + assert unexpected == [] + assert metrics.false_positive_rate <= 0.05 + assert metrics.lightweight_duration_seconds <= 30.0 + except Exception: + metrics = build_metrics( + normal_case_count=len(cases), + normal_case_pass_count=passed, + diffs=all_diffs, + lightweight_duration_seconds=time.perf_counter() - start, + ) + report = ReplayReport( + case_id="lightweight", + backend_pair=("in_memory", "sqlite"), + metrics=metrics, + diffs=[diff.to_dict() for diff in all_diffs], + capabilities=capabilities_for("in_memory", "sqlite"), + ) + _write_report_pair(tmp_path, "session_memory_summary_diff_report", report) + raise + + +@pytest.mark.replay_lightweight +def test_mutation_detection(tmp_path): + case = _case_by_id("all_entities_contract") + left_snapshot, _ = _run(_run_case(case, tmp_path)) + reference = canonicalize_snapshot(left_snapshot) + detected = 0 + all_diffs = [] + mutations = registered_mutations() + for mutation in mutations: + mutated = mutation.mutate(reference) + diffs = compare_snapshots( + reference, + mutated, + case_id=f"{case.case_id}:{mutation.name}", + backend_pair=("reference", "mutated"), + ) + all_diffs.extend(diffs) + categories = {diff.category for diff in diffs if not diff.allowed} + if mutation.expected_category in categories: + detected += 1 + else: + raise AssertionError(f"{mutation.name} did not produce {mutation.expected_category}: {categories}") + metrics = build_metrics( + normal_case_count=1, + normal_case_pass_count=1, + diffs=all_diffs, + mutation_total=len(mutations), + mutation_detected=detected, + ) + report = ReplayReport( + case_id="mutation_detection", + backend_pair=("reference", "mutated"), + metrics=metrics, + diffs=[diff.to_dict() for diff in all_diffs], + ) + _write_json_report(tmp_path, "session_memory_summary_mutation_report", report) + assert metrics.mutation_detection_rate == 1.0 + + +@pytest.mark.replay_lightweight +def test_required_public_cases_detect_injected_inconsistency(tmp_path): + matrix = json.loads(Path(__file__).with_name("acceptance_matrix.json").read_text(encoding="utf-8")) + required_case_ids = matrix["required_cases"] + detected = 0 + survived = [] + all_diffs = [] + + for case_id in required_case_ids: + case = _case_by_id(case_id) + left_snapshot, _ = _run(_run_case(case, tmp_path)) + reference = canonicalize_snapshot(left_snapshot) + mutated = _delete_first_observable_event(reference) + diffs = compare_snapshots( + reference, + mutated, + case_id=f"{case.case_id}:delete_first_observable_event", + backend_pair=("reference", "mutated"), + ) + all_diffs.extend(diffs) + categories = {diff.category for diff in diffs if not diff.allowed} + if "missing_entity" in categories: + detected += 1 + else: + survived.append(case.case_id) + + metrics = build_metrics( + normal_case_count=len(required_case_ids), + normal_case_pass_count=len(required_case_ids), + diffs=all_diffs, + mutation_total=len(required_case_ids), + mutation_detected=detected, + survived_mutations=survived, + ) + report = ReplayReport( + case_id="required_case_mutation_matrix", + backend_pair=("reference", "mutated"), + metrics=metrics, + diffs=[diff.to_dict() for diff in all_diffs], + ) + _write_json_report(tmp_path, "session_memory_summary_required_case_mutation_report", report) + assert survived == [] + assert metrics.mutation_detection_rate == 1.0 + + +@pytest.mark.replay_lightweight +def test_summary_defect_detection(tmp_path): + case = _case_by_id("summary_defect_specials") + left_snapshot, _ = _run(_run_case(case, tmp_path)) + reference = canonicalize_snapshot(left_snapshot) + required = { + "loss": "summary_missing", + "overwrite": "summary_version_mismatch", + "owner": "summary_owner_mismatch", + } + detections = {"loss": 0, "overwrite": 0, "owner": 0} + all_diffs = [] + for mutation in _summary_mutations(): + mutated = mutation.mutate(reference) + diffs = compare_snapshots( + reference, + mutated, + case_id=f"{case.case_id}:{mutation.name}", + backend_pair=("reference", "mutated"), + ) + all_diffs.extend(diffs) + categories = {diff.category for diff in diffs if not diff.allowed} + assert mutation.summary_defect is not None + assert required[mutation.summary_defect] in categories + detections[mutation.summary_defect] += 1 + metrics = build_metrics( + normal_case_count=1, + normal_case_pass_count=1, + diffs=all_diffs, + mutation_total=len(_summary_mutations()), + mutation_detected=len(_summary_mutations()), + summary_loss_detected=detections["loss"], + summary_overwrite_detected=detections["overwrite"], + summary_owner_error_detected=detections["owner"], + summary_loss_total=1, + summary_overwrite_total=1, + summary_owner_error_total=1, + ) + report = ReplayReport( + case_id="summary_defects", + backend_pair=("reference", "mutated"), + metrics=metrics, + diffs=[diff.to_dict() for diff in all_diffs], + ) + _write_json_report(tmp_path, "session_memory_summary_defect_report", report) + assert metrics.summary_loss_detection_rate == 1.0 + assert metrics.summary_overwrite_detection_rate == 1.0 + assert metrics.summary_owner_error_detection_rate == 1.0 + assert metrics.mutation_detection_rate == 1.0 + + +@pytest.mark.replay_lightweight +def test_allowed_diff_requires_explicit_rule(): + reference = {"sessions": [{"session_id": "s1", "state": {"value": 1}}], "memory": [], "summaries": []} + actual = {"sessions": [{"session_id": "s1", "state": {"value": 2}}], "memory": [], "summaries": []} + diffs = compare_snapshots(reference, actual, case_id="allowed", backend_pair=("left", "right")) + assert len(diffs) == 1 + assert not diffs[0].allowed + + allowed = compare_snapshots( + reference, + actual, + case_id="allowed", + backend_pair=("left", "right"), + allowed_diff_rules=[ + AllowedDiffRule( + backend_pair=("left", "right"), + field_path="$.sessions[0].state.value", + comparator="exact_path", + reason="demonstrate explicit allowed_diff plumbing", + still_validate="path must still be localized", + ) + ], + ) + assert len(allowed) == 1 + assert allowed[0].allowed + assert allowed[0].reason == "demonstrate explicit allowed_diff plumbing" + + +@pytest.mark.replay_lightweight +def test_diff_report_schema_contains_required_location_fields(tmp_path): + case = _case_by_id("all_entities_contract") + left_snapshot, _ = _run(_run_case(case, tmp_path)) + reference = canonicalize_snapshot(left_snapshot) + mutated = registered_mutations()[0].mutate(reference) + diffs = compare_snapshots(reference, mutated, case_id="schema", backend_pair=("reference", "mutated")) + report = ReplayReport( + case_id="schema", + backend_pair=("reference", "mutated"), + metrics=build_metrics(normal_case_count=1, normal_case_pass_count=0, diffs=diffs), + diffs=[diff.to_dict() for diff in diffs], + ) + path = tmp_path / "schema_report.json" + write_json_report(path, report) + payload = json.loads(path.read_text(encoding="utf-8")) + required = { + "case_id", + "backend_pair", + "session_id", + "entity_type", + "entity_id", + "index", + "field_path", + "reference_value", + "actual_value", + "allowed", + "category", + "reason", + } + assert required <= set(payload["diffs"][0]) + + +@pytest.mark.replay_lightweight +def test_sqlite_backend_rebuild_persists_observable_entities(tmp_path): + case = _case_by_id("all_entities_contract") + adapter = SQLiteReplayAdapter(case=case, workdir=tmp_path) + _run(adapter.setup()) + try: + before = _run(adapter.replay()) + sessions = dict(adapter.sessions) + event_map = dict(adapter.actual_to_client_event_id) + probes = list(adapter.memory_probes) + summaries = {key: list(value) for key, value in adapter.summary_records.items()} + finally: + _run(adapter.close()) + + rebuilt = SQLiteReplayAdapter(case=case, workdir=tmp_path) + _run(rebuilt.setup()) + try: + rebuilt.sessions = sessions + rebuilt.actual_to_client_event_id = event_map + rebuilt.memory_probes = probes + rebuilt.summary_records = summaries + after = _run(rebuilt.snapshot()) + finally: + _run(rebuilt.close()) + + before_data = canonicalize_snapshot(before) + after_data = canonicalize_snapshot(after) + rebuild_diffs = compare_snapshots( + before_data, + after_data, + case_id=case.case_id, + backend_pair=("sqlite", "sqlite_rebuilt"), + ) + assert rebuild_diffs == [] + assert any(event["is_summary"] for session in after_data["sessions"] for event in session["events"]) + assert after_data["memory"][0]["memories"] + + +@pytest.mark.replay_lightweight +def test_fault_injection_proxy_raises_after_committed_operation(tmp_path): + case = _case_by_id("single_turn_text") + adapter = InMemoryReplayAdapter(case=case) + proxy = FaultInjectingAdapter(adapter, fail_after_op_kind="append_text") + _run(adapter.setup()) + try: + _run(proxy.apply(case.operations[0])) + with pytest.raises(ReplayInjectedFault): + _run(proxy.apply(case.operations[1])) + snapshot = _run(adapter.snapshot()) + data = canonicalize_snapshot(snapshot) + assert data["sessions"][0]["events"][0]["event_id"] == "u1" + finally: + _run(adapter.close()) + + +@pytest.mark.replay_lightweight +def test_runtime_fault_retry_duplicate_is_detected(tmp_path): + case = _case_by_id("single_turn_text") + adapter = InMemoryReplayAdapter(case=case) + proxy = FaultInjectingAdapter(adapter, fail_after_op_kind="append_text") + _run(adapter.setup()) + try: + _run(proxy.apply(case.operations[0])) + with pytest.raises(ReplayInjectedFault): + _run(proxy.apply(case.operations[1])) + _run(adapter.apply(case.operations[1])) + snapshot = canonicalize_snapshot(_run(adapter.snapshot())) + diffs = compare_snapshots( + snapshot, + snapshot, + case_id="runtime_fault_retry", + backend_pair=("runtime", "runtime"), + ) + categories = {diff.category for diff in diffs if not diff.allowed} + metrics = build_metrics( + normal_case_count=1, + normal_case_pass_count=1, + diffs=diffs, + runtime_fault_total=1, + runtime_fault_detected=1 if "duplicate_event" in categories else 0, + ) + report = ReplayReport( + case_id="runtime_fault_retry", + backend_pair=("runtime", "runtime"), + metrics=metrics, + diffs=[diff.to_dict() for diff in diffs], + ) + _write_json_report(tmp_path, "session_memory_summary_runtime_fault_report", report) + assert "duplicate_event" in categories + assert metrics.runtime_fault_detection_rate == 1.0 + finally: + _run(adapter.close()) + + +def _summary_mutations() -> list[MutationOperator]: + return [ + mutation for mutation in registered_mutations() + if mutation.summary_defect in {"loss", "overwrite", "owner"} + ] + + +def _delete_first_observable_event(snapshot: dict) -> dict: + mutated = json.loads(json.dumps(snapshot)) + for session in mutated["sessions"]: + if session["events"]: + session["events"].pop(0) + return mutated + raise AssertionError("required replay case did not produce an observable event") + + +def _write_report_pair(tmp_path: Path, stem: str, report: ReplayReport) -> None: + _write_json_report(tmp_path, stem, report) + write_markdown_report(tmp_path / f"{stem}.md", report) + report_dir = os.environ.get("REPLAY_REPORT_DIR") + if report_dir: + write_markdown_report(Path(report_dir) / f"{stem}.md", report) + + +def _write_json_report(tmp_path: Path, stem: str, report: ReplayReport) -> None: + write_json_report(tmp_path / f"{stem}.json", report) + report_dir = os.environ.get("REPLAY_REPORT_DIR") + if report_dir: + write_json_report(Path(report_dir) / f"{stem}.json", report) diff --git a/tests/sessions/replay_consistency/test_replay_integration.py b/tests/sessions/replay_consistency/test_replay_integration.py new file mode 100644 index 00000000..64b241b2 --- /dev/null +++ b/tests/sessions/replay_consistency/test_replay_integration.py @@ -0,0 +1,123 @@ +# 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. + +"""Environment-gated integration entry points for replay consistency.""" + +from __future__ import annotations + +import asyncio +import os +import uuid +from dataclasses import replace +from pathlib import Path + +import pytest + +from .adapters import InMemoryReplayAdapter +from .adapters import RedisReplayAdapter +from .adapters import SQLUrlReplayAdapter +from .canonicalize import canonicalize_snapshot +from .capabilities import capabilities_for +from .cases import ReplayCase +from .cases import standard_cases +from .diff import compare_snapshots +from .reporting import ReplayReport +from .reporting import build_metrics +from .reporting import write_json_report + + +def _run(coro): + return asyncio.run(coro) + + +async def _compare_with_external(case_id: str, external_adapter): + base_case = next(case for case in standard_cases() if case.case_id == case_id) + case = _namespaced_case(base_case, uuid.uuid4().hex[:8]) + left = InMemoryReplayAdapter(case=case) + right = external_adapter(case) + await left.setup() + await right.setup() + try: + left_snapshot = await left.replay() + right_snapshot = await right.replay() + reference = canonicalize_snapshot(left_snapshot) + actual = canonicalize_snapshot(right_snapshot) + return compare_snapshots(reference, actual, case_id=case.case_id, backend_pair=(left.name, right.name)) + finally: + await left.close() + await right.close() + + +def _write_integration_report(tmp_path: Path, stem: str, report: ReplayReport) -> None: + write_json_report(tmp_path / f"{stem}.json", report) + report_dir = os.environ.get("REPLAY_REPORT_DIR") + if report_dir: + write_json_report(Path(report_dir) / f"{stem}.json", report) + + +def _namespaced_case(case: ReplayCase, suffix: str) -> ReplayCase: + operations = tuple( + replace( + op, + app_name=f"{op.app_name}_{suffix}", + user_id=f"{op.user_id}_{suffix}", + session_id=f"{op.session_id}_{suffix}", + ) for op in case.operations + ) + return replace(case, case_id=f"{case.case_id}_{suffix}", operations=operations) + + +@pytest.mark.replay_integration +@pytest.mark.parametrize( + "case_id", + [ + "all_entities_contract", + "state_shallow_update", + "memory_scope_user_session", + "summary_create_update", + ], +) +def test_redis_integration_replay(tmp_path, case_id): + if os.environ.get("RUN_REPLAY_REDIS_INTEGRATION") != "1" or not os.environ.get("REDIS_URL"): + pytest.skip("Set RUN_REPLAY_REDIS_INTEGRATION=1 and REDIS_URL to run Redis replay integration") + redis_url = os.environ["REDIS_URL"] + + def factory(case): + return RedisReplayAdapter(case=case, redis_url=redis_url) + + diffs = _run(_compare_with_external(case_id, factory)) + metrics = build_metrics(normal_case_count=1, normal_case_pass_count=0 if diffs else 1, diffs=diffs) + report = ReplayReport( + case_id=f"redis_integration:{case_id}", + backend_pair=("in_memory", "redis"), + metrics=metrics, + diffs=[diff.to_dict() for diff in diffs], + capabilities=capabilities_for("in_memory", "redis"), + ) + _write_integration_report(tmp_path, f"session_memory_summary_redis_{case_id}_integration_report", report) + assert [diff for diff in diffs if not diff.allowed] == [] + + +@pytest.mark.replay_integration +def test_sql_integration_replay(tmp_path): + if os.environ.get("RUN_REPLAY_SQL_INTEGRATION") != "1" or not os.environ.get("DATABASE_URL"): + pytest.skip("Set RUN_REPLAY_SQL_INTEGRATION=1 and DATABASE_URL to run SQL replay integration") + database_url = os.environ["DATABASE_URL"] + + def factory(case): + return SQLUrlReplayAdapter(case=case, db_url=database_url) + + diffs = _run(_compare_with_external("all_entities_contract", factory)) + metrics = build_metrics(normal_case_count=1, normal_case_pass_count=0 if diffs else 1, diffs=diffs) + report = ReplayReport( + case_id="sql_integration", + backend_pair=("in_memory", "sql"), + metrics=metrics, + diffs=[diff.to_dict() for diff in diffs], + capabilities=capabilities_for("in_memory", "sql"), + ) + _write_integration_report(tmp_path, "session_memory_summary_sql_integration_report", report) + assert [diff for diff in diffs if not diff.allowed] == [] diff --git a/tests/sessions/replay_consistency/test_replay_properties.py b/tests/sessions/replay_consistency/test_replay_properties.py new file mode 100644 index 00000000..caf38cc9 --- /dev/null +++ b/tests/sessions/replay_consistency/test_replay_properties.py @@ -0,0 +1,86 @@ +# 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. + +"""Fixed-seed metamorphic replay properties.""" + +from __future__ import annotations + +import asyncio +from dataclasses import asdict + +import pytest + +from .adapters import InMemoryReplayAdapter +from .canonicalize import canonicalize_snapshot +from .cases import ReplayCase +from .cases import ReplayOp +from .cases import standard_cases + + +def _run(coro): + return asyncio.run(coro) + + +@pytest.mark.replay_property +def test_replay_is_deterministic_for_fixed_cases(): + for case in standard_cases()[:10]: + first = _run(_run_in_memory(case)) + second = _run(_run_in_memory(case)) + assert canonicalize_snapshot(first) == canonicalize_snapshot(second) + + +@pytest.mark.replay_property +def test_replay_case_operation_serialization_round_trip(): + for case in standard_cases(): + encoded = [asdict(op) for op in case.operations] + decoded = tuple(ReplayOp(**item) for item in encoded) + restored = ReplayCase( + case_id=case.case_id, + operations=decoded, + checkpoints=case.checkpoints, + expected_entities=case.expected_entities, + description=case.description, + ) + assert restored == case + + +@pytest.mark.replay_property +def test_unrelated_session_operations_do_not_change_target_session(): + case = next(case for case in standard_cases() if case.case_id == "single_turn_text") + unrelated = ( + ReplayOp(kind="create_session", app_name="other_app", user_id="other_user", session_id="other_session"), + ReplayOp( + kind="append_text", + app_name="other_app", + user_id="other_user", + session_id="other_session", + client_event_id="other_event", + text="unrelated text", + ), + ) + mixed = ReplayCase( + case_id=case.case_id, + operations=case.operations + unrelated, + description="Target replay with unrelated app/user/session operations appended.", + ) + + target = _target_session(canonicalize_snapshot(_run(_run_in_memory(case)))) + mixed_target = _target_session(canonicalize_snapshot(_run(_run_in_memory(mixed)))) + + assert mixed_target == target + + +async def _run_in_memory(case: ReplayCase): + adapter = InMemoryReplayAdapter(case=case) + await adapter.setup() + try: + return await adapter.replay() + finally: + await adapter.close() + + +def _target_session(snapshot: dict): + return next(session for session in snapshot["sessions"] if session["app_name"] == "replay_app") diff --git a/tests/sessions/replay_consistency/test_replay_restart.py b/tests/sessions/replay_consistency/test_replay_restart.py new file mode 100644 index 00000000..5eeda653 --- /dev/null +++ b/tests/sessions/replay_consistency/test_replay_restart.py @@ -0,0 +1,127 @@ +# 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. + +"""Persistent backend restart and rebuild replay tests.""" + +from __future__ import annotations + +import asyncio +from collections import defaultdict + +import pytest + +from .adapters import SQLiteReplayAdapter +from .canonicalize import canonicalize_snapshot +from .cases import ReplayCase +from .cases import standard_cases +from .diff import compare_snapshots + + +def _run(coro): + return asyncio.run(coro) + + +@pytest.mark.replay_lightweight +@pytest.mark.parametrize( + ("case_id", "split_after_kind"), + [ + ("summary_create_update", "summarize"), + ("single_turn_text", "store_memory"), + ], +) +def test_sqlite_destroy_rebuild_continue_and_readback(tmp_path, case_id, split_after_kind): + case = _case_by_id(case_id) + expected = canonicalize_snapshot(_run(_run_full_sqlite(case, tmp_path / "full"))) + actual = canonicalize_snapshot(_run(_run_restarted_sqlite(case, tmp_path / "restart", split_after_kind))) + + diffs = compare_snapshots(expected, actual, case_id=case.case_id, backend_pair=("sqlite_full", "sqlite_rebuilt")) + + assert diffs == [] + + +async def _run_full_sqlite(case: ReplayCase, tmp_path): + tmp_path.mkdir(parents=True, exist_ok=True) + adapter = SQLiteReplayAdapter(case=case, workdir=tmp_path) + await adapter.setup() + try: + return await adapter.replay() + finally: + await adapter.close() + + +async def _run_restarted_sqlite(case: ReplayCase, tmp_path, split_after_kind: str): + tmp_path.mkdir(parents=True, exist_ok=True) + split_index = _split_index(case, split_after_kind) + first = SQLiteReplayAdapter(case=case, workdir=tmp_path) + await first.setup() + try: + for op in case.operations[:split_index]: + await first.apply(op) + state = _adapter_state(first) + finally: + await first.close() + + second = SQLiteReplayAdapter(case=case, workdir=tmp_path) + await second.setup() + try: + _restore_adapter_state(second, state) + await _refresh_known_sessions(second) + for op in case.operations[split_index:]: + await second.apply(op) + state = _adapter_state(second) + finally: + await second.close() + + third = SQLiteReplayAdapter(case=case, workdir=tmp_path) + await third.setup() + try: + _restore_adapter_state(third, state) + await _refresh_known_sessions(third) + return await third.snapshot() + finally: + await third.close() + + +def _split_index(case: ReplayCase, split_after_kind: str) -> int: + for idx, op in enumerate(case.operations, start=1): + if op.kind == split_after_kind: + return idx + raise AssertionError(f"{case.case_id} has no operation kind {split_after_kind}") + + +def _adapter_state(adapter: SQLiteReplayAdapter) -> dict: + return { + "sessions": dict(adapter.sessions), + "event_id_map": dict(adapter.event_id_map), + "actual_to_client_event_id": dict(adapter.actual_to_client_event_id), + "summary_records": {key: list(value) for key, value in adapter.summary_records.items()}, + "memory_probes": list(adapter.memory_probes), + "timestamp": adapter._timestamp, + } + + +def _restore_adapter_state(adapter: SQLiteReplayAdapter, state: dict) -> None: + adapter.sessions = dict(state["sessions"]) + adapter.event_id_map = dict(state["event_id_map"]) + adapter.actual_to_client_event_id = dict(state["actual_to_client_event_id"]) + adapter.summary_records = defaultdict(list, {key: list(value) for key, value in state["summary_records"].items()}) + adapter.memory_probes = list(state["memory_probes"]) + adapter._timestamp = state["timestamp"] + + +async def _refresh_known_sessions(adapter: SQLiteReplayAdapter) -> None: + for key, session in list(adapter.sessions.items()): + refreshed = await adapter.session_service.get_session( + app_name=session.app_name, + user_id=session.user_id, + session_id=session.id, + ) + assert refreshed is not None + adapter.sessions[key] = refreshed + + +def _case_by_id(case_id: str) -> ReplayCase: + return next(case for case in standard_cases() if case.case_id == case_id) diff --git a/tests/sessions/replay_consistency/test_report_schema.py b/tests/sessions/replay_consistency/test_report_schema.py new file mode 100644 index 00000000..783368ce --- /dev/null +++ b/tests/sessions/replay_consistency/test_report_schema.py @@ -0,0 +1,82 @@ +# 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 schema and acceptance matrix tests.""" + +from __future__ import annotations + +import json +from pathlib import Path + +from .capabilities import capabilities_for +from .cases import standard_cases +from .diff import DiffEntry +from .reporting import ReplayReport +from .reporting import build_metrics + + +def test_acceptance_matrix_matches_fixtures(): + matrix = _load_json("acceptance_matrix.json") + case_ids = [case.case_id for case in standard_cases()] + + assert len(matrix["required_cases"]) == 10 + assert set(matrix["required_cases"]) <= set(case_ids) + assert set(matrix["extended_cases"]) <= set(case_ids) + assert not (set(matrix["required_cases"]) & set(matrix["extended_cases"])) + assert matrix["requirements"]["p0_mutation_score"]["mutation_detection_rate"] == 1.0 + + +def test_report_conforms_to_local_schema(): + diff = DiffEntry( + case_id="schema", + backend_pair=("reference", "actual"), + session_id="s1", + entity_type="event", + entity_id="e1", + index=0, + field_path="$.sessions[0].events[0].content", + reference_value="left", + actual_value="right", + allowed=False, + category="content_mismatch", + reason="", + ) + report = ReplayReport( + case_id="schema", + backend_pair=("reference", "actual"), + metrics=build_metrics(normal_case_count=1, normal_case_pass_count=0, diffs=[diff]), + diffs=[diff.to_dict()], + capabilities=capabilities_for("in_memory", "sqlite"), + ).to_dict() + schema = _load_json("report_schema.json") + + _assert_required_keys(report, schema["required"]) + _assert_required_keys(report["diffs"][0], schema["$defs"]["diff"]["required"]) + assert report["schema_version"] == "1.0" + assert report["unexpected_diffs"] == report["diffs"] + assert report["allowed_diffs"] == [] + + +def test_report_is_deterministic_without_runtime_fields(): + metrics = build_metrics(normal_case_count=1, normal_case_pass_count=1, diffs=[]) + first = ReplayReport(case_id="deterministic", backend_pair=("a", "b"), metrics=metrics, diffs=[]).to_dict() + second = ReplayReport(case_id="deterministic", backend_pair=("a", "b"), metrics=metrics, diffs=[]).to_dict() + + assert _strip_runtime_fields(first) == _strip_runtime_fields(second) + + +def _load_json(name: str): + return json.loads((Path(__file__).with_name(name)).read_text(encoding="utf-8")) + + +def _assert_required_keys(value: dict, required: list[str]) -> None: + assert set(required) <= set(value) + + +def _strip_runtime_fields(value: dict) -> dict: + stripped = dict(value) + stripped.pop("generated_at", None) + return stripped