diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index c8fcd6e49..40bdd8dfe 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -77,6 +77,12 @@ jobs: # 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/ + - name: Run InMemory-only replay smoke test (not full matrix) + env: + TRPC_REPLAY_BACKENDS: in_memory + # The full cross-backend replay matrix runs in the coverage step above. + run: pytest -q tests/sessions/test_replay_consistency.py tests/sessions/test_replay_real_agent.py + - name: Upload coverage reports to Codecov uses: codecov/codecov-action@v4 env: diff --git a/requirements-test.txt b/requirements-test.txt index c54dad067..02b447dff 100644 --- a/requirements-test.txt +++ b/requirements-test.txt @@ -1,7 +1,7 @@ # Test framework pytest pytest-mock -pytest-asyncio +pytest-asyncio>=0.24.0 pyparsing>=2.4.2 pytest-cov unittest-xml-reporting diff --git a/tests/sessions/DESIGN.md b/tests/sessions/DESIGN.md new file mode 100644 index 000000000..51b12a6aa --- /dev/null +++ b/tests/sessions/DESIGN.md @@ -0,0 +1,3 @@ +# Session / Memory / Summary Replay Harness 设计说明 + +同一组轨迹驱动 InMemory、SQLite 和可选 Redis，快照覆盖事件、state、memory、summary。归一化只处理自动 ID、相对时间、Unicode 空白和无序容器；Summary 文本可规范化，但 session 归属、版本和覆盖关系必须一致。差异按后端对和字段路径精确匹配。SQLite 重开验证持久化；`TRPC_REPLAY_BACKENDS=in_memory` 为轻量模式。两份 JSON 均为测试生成产物，可删除后重跑再生，不是手写契约。 diff --git a/tests/sessions/conftest.py b/tests/sessions/conftest.py new file mode 100644 index 000000000..3defacdd2 --- /dev/null +++ b/tests/sessions/conftest.py @@ -0,0 +1,34 @@ +# +# 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 the Apache License Version 2.0. +# +"""Collection policy for the replay acceptance tests.""" + +import os + +import pytest + +BACKEND_MODE_ENVIRONMENT_VARIABLE = "TRPC_REPLAY_BACKENDS" +IN_MEMORY_BACKEND_MODE = "in_memory" +IN_MEMORY_TEST_NAME = "test_in_memory_only_lightweight_mode" +IN_MEMORY_REAL_AGENT_SAFE_TEST_NAME = "test_subtle_realistic_drift_is_detected" +REPLAY_TEST_FILES = frozenset({ + "test_replay_consistency.py", + "test_replay_real_agent.py", +}) +IN_MEMORY_SKIP_REASON = "disabled by TRPC_REPLAY_BACKENDS=in_memory" + + +def pytest_collection_modifyitems(items): + """Keep only the lightweight replay when InMemory-only mode is selected.""" + if os.getenv(BACKEND_MODE_ENVIRONMENT_VARIABLE) != IN_MEMORY_BACKEND_MODE: + return + skip = pytest.mark.skip(reason=IN_MEMORY_SKIP_REASON) + for item in items: + if (item.path.name == "test_replay_consistency.py" and item.name != IN_MEMORY_TEST_NAME) or ( + item.path.name == "test_replay_real_agent.py" + and item.name != IN_MEMORY_REAL_AGENT_SAFE_TEST_NAME): + item.add_marker(skip) diff --git a/tests/sessions/real_agent_replay_validation_report.json b/tests/sessions/real_agent_replay_validation_report.json new file mode 100644 index 000000000..82b3bab03 --- /dev/null +++ b/tests/sessions/real_agent_replay_validation_report.json @@ -0,0 +1,56 @@ +{ + "case_id": "real_agent_tool_memory", + "clean_comparison": { + "diffs": [], + "unexpected_diff_count": 0 + }, + "injected_anomalies": { + "memory_single_character_drift": { + "detected": true, + "diffs": [ + { + "allowed": false, + "backend_pair": [ + "in_memory", + "sqlite" + ], + "case_id": "real_agent_tool_memory", + "category": "memory", + "event_index": null, + "field_path": "/memory/Chinese/memories/0/content/parts/0/text", + "left": "Here are the preferences for **replay-user**:\n\n- **Preferred Language:** Chinese\n- **Notification Channel:** email", + "reason": null, + "right": "Here are the preferences for **replay-user**:\n\n- **Preferred Language:** Chinese\n- **Notification Channel:** emaiX", + "session_id": "replay-session", + "summary_id": null + } + ], + "unexpected_diff_count": 1 + }, + "tool_response_single_character_drift": { + "detected": true, + "diffs": [ + { + "allowed": false, + "backend_pair": [ + "in_memory", + "sqlite" + ], + "case_id": "real_agent_tool_memory", + "category": "events", + "event_index": 2, + "field_path": "/events/2/content/parts/0/function_response/response/preferred_language", + "left": "Chinese", + "reason": null, + "right": "ChinesX", + "session_id": "replay-session", + "summary_id": null + } + ], + "unexpected_diff_count": 1 + } + }, + "model": "deepseek-v4-flash", + "schema_version": 1, + "tool_round_trip": true +} diff --git a/tests/sessions/replay_cases.py b/tests/sessions/replay_cases.py new file mode 100644 index 000000000..a56311205 --- /dev/null +++ b/tests/sessions/replay_cases.py @@ -0,0 +1,303 @@ +# +# 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 the Apache License Version 2.0. +# +"""Standard replay traces for session, memory, and summary consistency tests.""" + +from __future__ import annotations + +from dataclasses import dataclass +from dataclasses import field +from enum import Enum +from typing import Any +from typing import Iterable +from typing import Mapping + +APP_NAME = "replay-app" +USER_ID = "replay-user" +SESSION_ID = "replay-session" +BASE_TIMESTAMP = 1_700_000_000.0 +TIMESTAMP_STEP_SECONDS = 1.0 +LARGE_INTEGER_VALUE = 9_007_199_254_740_993 + + +class OperationKind(str, Enum): + """Supported replay operation kinds.""" + + CREATE = "create" + APPEND = "append" + STORE_MEMORY = "store_memory" + SEARCH_MEMORY = "search_memory" + SUMMARY = "summary" + UNKNOWN_OUTCOME_RETRY = "unknown_outcome_retry" + UNKNOWN_MEMORY_RETRY = "unknown_memory_retry" + UNKNOWN_SUMMARY_RETRY = "unknown_summary_retry" + BEFORE_CALL_FAILURE = "before_call_failure" + + +@dataclass(frozen=True) +class ReplayOperation: + """One deterministic operation in a replay trace.""" + + kind: OperationKind + payload: Mapping[str, Any] = field(default_factory=dict) + + +@dataclass(frozen=True) +class ReplayCase: + """Named replay trace.""" + + case_id: str + operations: tuple[ReplayOperation, ...] + expected: "ExpectedOutcome" + + +@dataclass(frozen=True) +class ExpectedOutcome: + """Independent minimum business outcome for one trace.""" + + event_ids: tuple[str, ...] + state: Mapping[str, Any] = field(default_factory=dict) + memory_counts: Mapping[str, int] = field(default_factory=dict) + summary_generation: int = 0 + summary_fact: str = "" + minimum_historical_events: int = 0 + failure_count: int = 0 + + +def validate_replay_cases(cases: Iterable[ReplayCase]) -> None: + """Reject ambiguous case identifiers before matrix execution.""" + case_ids = [case.case_id for case in cases] + duplicates = sorted({case_id for case_id in case_ids if case_ids.count(case_id) > 1}) + if duplicates: + raise ValueError(f"duplicate replay case ids: {duplicates}") + + +def _create(state: Mapping[str, Any] | None = None) -> ReplayOperation: + return ReplayOperation(OperationKind.CREATE, {"state": dict(state or {})}) + + +def _text(event_id: str, author: str, text: str, *, state_delta: Mapping[str, Any] | None = None) -> ReplayOperation: + payload = { + "event_id": event_id, + "author": author, + "text": text, + } + if state_delta: + payload["state_delta"] = dict(state_delta) + return ReplayOperation(OperationKind.APPEND, payload) + + +def _tool(event_id: str, part_type: str, value: Mapping[str, Any]) -> ReplayOperation: + return ReplayOperation( + OperationKind.APPEND, + { + "event_id": event_id, + "author": "agent" if part_type == "function_call" else "tool", + "part_type": part_type, + "value": dict(value), + }, + ) + + +def _summary() -> ReplayOperation: + return ReplayOperation(OperationKind.SUMMARY) + + +def _memory(query: str) -> tuple[ReplayOperation, ReplayOperation]: + return ( + ReplayOperation(OperationKind.STORE_MEMORY), + ReplayOperation(OperationKind.SEARCH_MEMORY, {"query": query}), + ) + + +REPLAY_CASES = ( + ReplayCase( + "single_turn", + ( + _create(), + _text("single-user", "user", "Hello replay"), + _text("single-agent", "agent", "Hello user"), + ), + ExpectedOutcome(("single-user", "single-agent")), + ), + ReplayCase( + "multi_turn", + ( + _create(), + _text("multi-u1", "user", "First question"), + _text("multi-a1", "agent", "First answer"), + _text("multi-u2", "user", "Second question"), + _text("multi-a2", "agent", "Second answer"), + _text("multi-u3", "user", "Third question"), + _text("multi-a3", "agent", "Third answer"), + ), + ExpectedOutcome(("multi-u1", "multi-a1", "multi-u2", "multi-a2", "multi-u3", "multi-a3")), + ), + ReplayCase( + "tool_round_trip", + ( + _create(), + _text("tool-user", "user", "Find weather"), + _tool("tool-call", "function_call", { + "name": "weather", + "id": "call-weather", + "args": { + "city": "Shenzhen" + }, + }), + _tool("tool-response", "function_response", { + "name": "weather", + "id": "call-weather", + "response": { + "temperature": 28 + }, + }), + _text("tool-agent", "agent", "It is 28 C"), + ), + ExpectedOutcome(("tool-user", "tool-call", "tool-response", "tool-agent")), + ), + ReplayCase( + "state_overwrite", + ( + _create({ + "theme": "light", + "app:region": "cn", + "user:language": "en", + "large_counter": LARGE_INTEGER_VALUE, + }), + _text( + "state-1", + "agent", + "state one", + state_delta={ + "theme": "dark", + "app:region": "apac", + "user:language": "zh", + "temp:request_id": "ephemeral", + }, + ), + _text("state-2", "agent", "state two", state_delta={"theme": "system"}), + ), + ExpectedOutcome( + ("state-1", "state-2"), + { + "theme": "system", + "app:region": "apac", + "user:language": "zh", + "large_counter": LARGE_INTEGER_VALUE, + }, + ), + ), + ReplayCase( + "memory_preference", + ( + _create(), + _text("memory-pref", "user", "I prefer jasmine tea"), + _text("memory-fact", "agent", "Favorite drink is jasmine tea"), + *_memory("jasmine"), + ), + ExpectedOutcome(("memory-pref", "memory-fact"), memory_counts={"jasmine": 2}), + ), + ReplayCase( + "summary_create", + ( + _create(), + _text("summary-u1", "user", "Plan a trip"), + _text("summary-a1", "agent", "Choose Shenzhen"), + _text("summary-u2", "user", "Use the train"), + _summary(), + ), + ExpectedOutcome( + ("summary-a1", "summary-u2"), + summary_generation=1, + summary_fact="Plan a trip", + minimum_historical_events=1, + ), + ), + ReplayCase( + "summary_update", + ( + _create(), + _text("update-u1", "user", "Initial requirement"), + _text("update-a1", "agent", "Initial decision"), + _text("update-u2", "user", "Keep details"), + _summary(), + _text("update-u3", "user", "New requirement"), + _text("update-a2", "agent", "Updated decision"), + _summary(), + ), + ExpectedOutcome( + ("update-u3", "update-a2"), + summary_generation=2, + summary_fact="Initial requirement", + minimum_historical_events=3, + ), + ), + ReplayCase( + "summary_truncation", + ( + _create(), + _text("truncate-u1", "user", "Old context"), + _text("truncate-a1", "agent", "Old answer"), + _text("truncate-u2", "user", "Recent context"), + _text("truncate-a2", "agent", "Recent answer"), + _summary(), + _text("truncate-u3", "user", "Follow-up"), + _text("truncate-a3", "agent", "Follow-up answer"), + ), + ExpectedOutcome( + ("truncate-u2", "truncate-a2", "truncate-u3", "truncate-a3"), + summary_generation=1, + summary_fact="Old context", + minimum_historical_events=2, + ), + ), + ReplayCase( + "duplicate_retry", + ( + _create(), + ReplayOperation( + OperationKind.UNKNOWN_OUTCOME_RETRY, + { + "event_id": "retry-event", + "author": "user", + "text": "Write exactly once", + }, + ), + ReplayOperation(OperationKind.UNKNOWN_MEMORY_RETRY), + ReplayOperation(OperationKind.UNKNOWN_SUMMARY_RETRY), + ), + ExpectedOutcome(tuple(), summary_generation=1, summary_fact="Write exactly once"), + ), + ReplayCase( + "write_recovery", + ( + _create({"status": "clean"}), + ReplayOperation( + OperationKind.BEFORE_CALL_FAILURE, + { + "event_id": "failed-event", + "author": "agent", + "text": "must not persist", + "state_delta": { + "status": "dirty" + }, + }, + ), + _text("recovery-event", "agent", "recovered", state_delta={"status": "recovered"}), + *_memory("recovered"), + ), + ExpectedOutcome( + ("recovery-event", ), + {"status": "recovered"}, + {"recovered": 1}, + failure_count=1, + ), + ), +) + +REPLAY_CASE_BY_ID = {case.case_id: case for case in REPLAY_CASES} diff --git a/tests/sessions/replay_harness.py b/tests/sessions/replay_harness.py new file mode 100644 index 000000000..fa7310de0 --- /dev/null +++ b/tests/sessions/replay_harness.py @@ -0,0 +1,486 @@ +# +# 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 the Apache License Version 2.0. +# +"""Reusable execution harness for session, memory, and summary replay traces.""" + +from __future__ import annotations + +import asyncio +import copy +import time +from dataclasses import dataclass +from pathlib import Path +from typing import Any +from typing import AsyncIterator +from typing import Optional + +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.types import Content +from trpc_agent_sdk.types import EventActions +from trpc_agent_sdk.types import FunctionCall +from trpc_agent_sdk.types import FunctionResponse +from trpc_agent_sdk.types import Part + +from .replay_cases import APP_NAME +from .replay_cases import OperationKind +from .replay_cases import ReplayCase +from .replay_cases import ExpectedOutcome +from .replay_cases import ReplayOperation +from .replay_cases import SESSION_ID +from .replay_cases import TIMESTAMP_STEP_SECONDS +from .replay_cases import USER_ID + +OPERATION_TIMEOUT_SECONDS = 5.0 +SERVICE_CLOSE_TIMEOUT_SECONDS = 5.0 +SUMMARY_KEEP_RECENT_EVENTS = 2 +SUMMARY_PREFIX = "deterministic-summary-v" +SUMMARY_TEXT_LIMIT = 500 +SQLITE_BACKEND = "sqlite" +IN_MEMORY_BACKEND = "in_memory" +REDIS_BACKEND = "redis" +MEMORY_RESULT_SUFFIX_START = 2 + + +class DeterministicSummaryModel: + """Small deterministic LLM substitute used by the real summarizer.""" + + name = "replay-deterministic-summary" + + def __init__(self) -> None: + self._generation = 0 + + async def generate_async(self, request: Any, stream: bool = False, ctx: Any = None) -> AsyncIterator[LlmResponse]: + del stream, ctx + self._generation += 1 + prompt = request.contents[0].parts[0].text + conversation = prompt.split("Conversation:\n", maxsplit=1)[-1].rsplit("\n\nSummary:", maxsplit=1)[0] + text = f"{SUMMARY_PREFIX}{self._generation}: {conversation.strip()}"[:SUMMARY_TEXT_LIMIT] + yield LlmResponse(content=Content(parts=[Part.from_text(text=text)])) + + +class InjectedReplayFailure(RuntimeError): + """Expected failure used by replay recovery operations only.""" + + +@dataclass(frozen=True) +class ReplayIdentity: + """Storage identity used to isolate one replay run.""" + + app_name: str = APP_NAME + user_id: str = USER_ID + session_id: str = SESSION_ID + + +@dataclass +class BackendBundle: + """Session and memory services representing one replay backend.""" + + name: str + session_service: Any + memory_service: Any + db_url: Optional[str] = None + identity: ReplayIdentity = ReplayIdentity() + + async def close(self) -> None: + """Close both services even if the first close fails.""" + close_all = asyncio.gather(self.memory_service.close(), self.session_service.close(), return_exceptions=True) + results = await asyncio.wait_for(close_all, SERVICE_CLOSE_TIMEOUT_SECONDS) + errors = [result for result in results if isinstance(result, Exception)] + if errors: + raise errors[0] + + +def _session_config() -> SessionServiceConfig: + config = SessionServiceConfig(store_historical_events=True) + config.clean_ttl_config() + return config + + +def _memory_config() -> MemoryServiceConfig: + config = MemoryServiceConfig(enabled=True) + config.clean_ttl_config() + return config + + +def _summary_manager() -> SummarizerSessionManager: + model = DeterministicSummaryModel() + summarizer = SessionSummarizer( + model=model, + check_summarizer_functions=[lambda _: True], + keep_recent_count=SUMMARY_KEEP_RECENT_EVENTS, + start_by_user_turn=False, + ) + return SummarizerSessionManager(model=model, summarizer=summarizer) + + +async def create_in_memory_backend(identity: Optional[ReplayIdentity] = None) -> BackendBundle: + """Create lightweight in-memory services.""" + session_service = InMemorySessionService( + summarizer_manager=_summary_manager(), + session_config=_session_config(), + ) + memory_service = InMemoryMemoryService(memory_service_config=_memory_config()) + return BackendBundle(IN_MEMORY_BACKEND, session_service, memory_service, identity=identity or ReplayIdentity()) + + +async def create_sqlite_backend(db_path: Path, identity: Optional[ReplayIdentity] = None) -> BackendBundle: + """Create file-backed SQLite services and initialize their tables.""" + db_url = f"sqlite:///{db_path.as_posix()}" + session_service = SqlSessionService( + db_url=db_url, + summarizer_manager=_summary_manager(), + session_config=_session_config(), + is_async=False, + ) + memory_service = SqlMemoryService( + db_url=db_url, + memory_service_config=_memory_config(), + is_async=False, + ) + return BackendBundle(SQLITE_BACKEND, session_service, memory_service, db_url, identity or ReplayIdentity()) + + +async def create_redis_backend(db_url: str, identity: ReplayIdentity) -> BackendBundle: + """Create optional Redis integration services.""" + session_service = RedisSessionService( + db_url=db_url, + summarizer_manager=_summary_manager(), + session_config=_session_config(), + is_async=True, + decode_responses=True, + ) + memory_service = RedisMemoryService( + db_url=db_url, + memory_service_config=_memory_config(), + is_async=True, + decode_responses=True, + ) + return BackendBundle(REDIS_BACKEND, session_service, memory_service, db_url, identity) + + +def _event_from_operation(operation: ReplayOperation, operation_index: int, timestamp_origin: float) -> Event: + payload = operation.payload + part_type = payload.get("part_type", "text") + if part_type == "function_call": + part = Part(function_call=FunctionCall(**payload["value"])) + elif part_type == "function_response": + part = Part(function_response=FunctionResponse(**payload["value"])) + else: + part = Part.from_text(text=str(payload["text"])) + return Event( + id=str(payload["event_id"]), + invocation_id=f"replay-{operation_index}", + author=str(payload["author"]), + content=Content(parts=[part]), + actions=EventActions(state_delta=dict(payload.get("state_delta", {}))), + timestamp=timestamp_origin + operation_index * TIMESTAMP_STEP_SECONDS, + ) + + +class ReplayRunner: + """Execute one trace against one backend and capture its read model.""" + + def __init__(self, backend: BackendBundle) -> None: + self.backend = backend + self.session: Optional[Session] = None + self.memory_results: dict[str, Any] = {} + self.memory_query_by_key: dict[str, str] = {} + self.summary_generation = 0 + self.summary_checkpoints: list[dict[str, Any]] = [] + self.failures: list[dict[str, Any]] = [] + self.timestamp_origin = time.time() + + async def run(self, replay_case: ReplayCase) -> dict[str, Any]: + """Execute all operations and return a backend snapshot.""" + for index, operation in enumerate(replay_case.operations): + try: + isolated = _clone_operation(operation) + await asyncio.wait_for(self._execute(isolated, index), OPERATION_TIMEOUT_SECONDS) + except InjectedReplayFailure as error: + self.failures.append({ + "operation_index": index, + "error_type": type(error).__name__, + "message": str(error), + }) + self.session = await asyncio.wait_for(self._read_session(), OPERATION_TIMEOUT_SECONDS) + if self.session is None: + raise AssertionError(f"case {replay_case.case_id} did not create a session") + return await asyncio.wait_for(self.snapshot(replay_case.case_id), OPERATION_TIMEOUT_SECONDS) + + async def _execute(self, operation: ReplayOperation, operation_index: int) -> None: + handlers = { + OperationKind.CREATE: self._create, + OperationKind.APPEND: self._append, + OperationKind.STORE_MEMORY: self._store_memory, + OperationKind.SEARCH_MEMORY: self._search_memory, + OperationKind.SUMMARY: self._summarize, + OperationKind.UNKNOWN_OUTCOME_RETRY: self._unknown_outcome_retry, + OperationKind.UNKNOWN_MEMORY_RETRY: self._unknown_memory_retry, + OperationKind.UNKNOWN_SUMMARY_RETRY: self._unknown_summary_retry, + OperationKind.BEFORE_CALL_FAILURE: self._before_call_failure, + } + await handlers[operation.kind](operation, operation_index) + + async def _create(self, operation: ReplayOperation, operation_index: int) -> None: + del operation_index + identity = self.backend.identity + self.session = await self.backend.session_service.create_session( + app_name=identity.app_name, + user_id=identity.user_id, + session_id=identity.session_id, + state=dict(operation.payload.get("state", {})), + ) + _validate_session_scope(self.session, identity) + + async def _append(self, operation: ReplayOperation, operation_index: int) -> None: + session = self._require_session() + event = _event_from_operation(operation, operation_index, self.timestamp_origin) + await self.backend.session_service.append_event(session, event) + + async def _store_memory(self, operation: ReplayOperation, operation_index: int) -> None: + del operation, operation_index + session = await self._read_session() + if session is None: + raise AssertionError("memory operation requires a session") + await self.backend.memory_service.store_session(session) + + async def _search_memory(self, operation: ReplayOperation, operation_index: int) -> None: + del operation_index + session = self._require_session() + query = str(operation.payload["query"]) + result_key = _next_memory_result_key(self.memory_results, query) + self.memory_results[result_key] = await self.backend.memory_service.search_memory(session.save_key, query) + self.memory_query_by_key[result_key] = query + + async def _summarize(self, operation: ReplayOperation, operation_index: int) -> None: + del operation, operation_index + session = self._require_session() + await self.backend.session_service.create_session_summary(session) + self.summary_generation += 1 + self.session = await self._read_session() + await self._record_summary_checkpoint() + + async def _unknown_outcome_retry(self, operation: ReplayOperation, operation_index: int) -> None: + try: + await self._append(operation, operation_index) + raise InjectedReplayFailure("injected response loss after event commit") + except InjectedReplayFailure: + pass + stored = await self._read_session() + event_id = str(operation.payload["event_id"]) + if stored is None or not any(event.id == event_id for event in stored.events): + await self._append(operation, operation_index) + self.session = stored + + async def _unknown_memory_retry(self, operation: ReplayOperation, operation_index: int) -> None: + await self._store_memory(operation, operation_index) + await self._store_memory(operation, operation_index) + + async def _unknown_summary_retry(self, operation: ReplayOperation, operation_index: int) -> None: + try: + await self._summarize(operation, operation_index) + raise InjectedReplayFailure("injected response loss after summary commit") + except InjectedReplayFailure: + pass + stored = await self._read_session() + if stored is None or not any(event.is_summary_event() for event in stored.events): + await self._summarize(operation, operation_index) + self.session = stored + + async def _before_call_failure(self, operation: ReplayOperation, operation_index: int) -> None: + del operation, operation_index + raise InjectedReplayFailure("injected before-call failure") + + async def _read_session(self) -> Optional[Session]: + identity = self.backend.identity + session = await self.backend.session_service.get_session( + app_name=identity.app_name, + user_id=identity.user_id, + session_id=identity.session_id, + ) + if session is not None: + _validate_session_scope(session, identity) + return session + + def _require_session(self) -> Session: + if self.session is None: + raise AssertionError("operation requires a session") + return self.session + + async def snapshot(self, case_id: str) -> dict[str, Any]: + """Capture public session/memory data plus public summary metadata.""" + session = self._require_session() + manager = self.backend.session_service.summarizer_manager + summary = await manager.get_session_summary(session) if manager else None + summary_events = [event for event in session.events if event.is_summary_event()] + final_memory = await self._read_final_memory(session) + return { + "case_id": case_id, + "backend": self.backend.name, + "session_id": session.id, + "app_name": session.app_name, + "user_id": session.user_id, + "state": dict(session.state), + "events": [_event_to_dict(event) for event in session.events], + "historical_events": [_event_to_dict(event) for event in session.historical_events], + "memory": { + query: result.model_dump(mode="json") + for query, result in self.memory_results.items() + }, + "memory_final": final_memory, + "summary": _summary_to_dict(summary, summary_events, self.summary_generation, session.id), + "summary_checkpoints": [dict(checkpoint) for checkpoint in self.summary_checkpoints], + "failures": list(self.failures), + } + + async def _read_final_memory(self, session: Session) -> dict[str, Any]: + results = {} + for result_key, query in self.memory_query_by_key.items(): + response = await self.backend.memory_service.search_memory(session.save_key, query) + results[result_key] = response.model_dump(mode="json") + return results + + async def _record_summary_checkpoint(self) -> None: + session = self._require_session() + manager = self.backend.session_service.summarizer_manager + summary = await manager.get_session_summary(session) + anchors = [event for event in session.events if event.is_summary_event()] + checkpoint = _summary_to_dict(summary, anchors, self.summary_generation, session.id) + if checkpoint is None: + raise AssertionError("successful summary did not create metadata or anchor") + if checkpoint["text"] != _summary_text(anchors[0]) or checkpoint["anchor_count"] != 1: + raise AssertionError("summary cache and anchor disagree") + if self.summary_checkpoints and checkpoint["updated_at"] <= self.summary_checkpoints[-1]["updated_at"]: + raise AssertionError("summary timestamp is not monotonic") + self.summary_checkpoints.append(checkpoint) + + async def reload_snapshot(self, replay_case: ReplayCase) -> dict[str, Any]: + """Read a persisted backend after service reconstruction.""" + self.memory_results.clear() + self.memory_query_by_key.clear() + self.summary_checkpoints.clear() + self.failures.clear() + self.summary_generation = 0 + self.session = await asyncio.wait_for(self._read_session(), OPERATION_TIMEOUT_SECONDS) + if self.session is None: + raise AssertionError("persisted session is missing") + for operation in replay_case.operations: + if operation.kind == OperationKind.SEARCH_MEMORY: + query = str(operation.payload["query"]) + result_key = _next_memory_result_key(self.memory_query_by_key, query) + self.memory_query_by_key[result_key] = query + snapshot = await asyncio.wait_for(self.snapshot(replay_case.case_id), OPERATION_TIMEOUT_SECONDS) + snapshot["backend"] = f"{self.backend.name}_reloaded" + return snapshot + + +def _clone_operation(operation: ReplayOperation) -> ReplayOperation: + return copy.deepcopy(operation) + + +def _validate_session_scope(session: Session, identity: ReplayIdentity) -> None: + actual = (session.app_name, session.user_id, session.id) + expected = (identity.app_name, identity.user_id, identity.session_id) + if actual != expected: + raise AssertionError(f"physical session scope mismatch: expected {expected!r}, got {actual!r}") + + +def _next_memory_result_key(results: dict[str, Any], query: str) -> str: + if query not in results: + return query + suffix = MEMORY_RESULT_SUFFIX_START + while f"{query}#{suffix}" in results: + suffix += 1 + return f"{query}#{suffix}" + + +def _event_to_dict(event: Event) -> dict[str, Any]: + data = event.model_dump(mode="json", exclude_none=True) + data["is_summary"] = event.is_summary_event() + return data + + +def _summary_to_dict(summary: Any, summary_events: list[Event], generation: int, + session_id: str) -> Optional[dict[str, Any]]: + if summary is None and not summary_events: + return None + anchor = _event_to_dict(summary_events[0]) if summary_events else None + return { + "session_id": summary.session_id if summary else session_id, + "text": summary.summary_text if summary else _summary_text(summary_events[0]), + "generation": generation, + "updated_at": summary.summary_timestamp if summary else summary_events[0].timestamp, + "original_event_count": summary.original_event_count if summary else None, + "compressed_event_count": summary.compressed_event_count if summary else None, + "anchor": anchor, + "anchor_count": len(summary_events), + "cache_present": summary is not None, + } + + +def _summary_text(event: Event) -> str: + text = event.get_text() + marker = "Previous conversation summary: " + return text[len(marker):] if text.startswith(marker) else text + + +def validate_snapshot(expected: ExpectedOutcome, + snapshot: dict[str, Any], + require_cache: bool = True, + require_runtime: bool = True) -> list[str]: + """Validate one backend against case-owned business expectations.""" + errors = [] + event_ids = tuple(event["id"] for event in snapshot["events"] if not event["is_summary"]) + if event_ids != expected.event_ids: + errors.append(f"event ids: expected {expected.event_ids}, got {event_ids}") + for key, value in expected.state.items(): + if snapshot["state"].get(key) != value: + errors.append(f"state {key}: expected {value!r}, got {snapshot['state'].get(key)!r}") + if any(key.startswith("temp:") for key in snapshot["state"]): + errors.append("temporary state was persisted") + memory_field = "memory" if require_runtime else "memory_final" + for query, count in expected.memory_counts.items(): + actual = len(snapshot[memory_field].get(query, {}).get("memories", [])) + if actual != count: + errors.append(f"memory {query}: expected {count}, got {actual}") + errors.extend(_validate_summary(expected, snapshot, require_cache)) + if len(snapshot["historical_events"]) < expected.minimum_historical_events: + errors.append("historical event count is below expectation") + if require_runtime and len(snapshot["failures"]) != expected.failure_count: + errors.append(f"failure count: expected {expected.failure_count}, got {len(snapshot['failures'])}") + return errors + + +def _validate_summary(expected: ExpectedOutcome, snapshot: dict[str, Any], require_cache: bool) -> list[str]: + summary = snapshot["summary"] + if expected.summary_generation == 0: + return [] if summary is None else ["unexpected summary"] + if summary is None: + return ["expected summary is missing"] + errors = [] + if require_cache and summary["generation"] != expected.summary_generation: + errors.append("summary generation mismatch") + if expected.summary_fact not in summary["text"]: + errors.append(f"summary lost fact {expected.summary_fact!r}") + if summary["session_id"] != snapshot["session_id"] or summary["anchor_count"] != 1: + errors.append("summary ownership or anchor replacement mismatch") + if require_cache and (not summary["cache_present"] + or len(snapshot["summary_checkpoints"]) != expected.summary_generation): + errors.append("summary cache/checkpoint mismatch") + return errors diff --git a/tests/sessions/replay_report.py b/tests/sessions/replay_report.py new file mode 100644 index 000000000..0e806d65e --- /dev/null +++ b/tests/sessions/replay_report.py @@ -0,0 +1,422 @@ +# +# 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 the Apache License Version 2.0. +# +"""Normalization, comparison, metrics, and reporting for replay snapshots.""" + +from __future__ import annotations + +import copy +import json +import os +import unicodedata +from dataclasses import asdict +from dataclasses import dataclass +from pathlib import Path +from typing import Any +from typing import Callable +from typing import Iterable +from typing import Optional + +REPORT_SCHEMA_VERSION = 1 +SUMMARY_ANCHOR_ID = "" +STRUCTURAL_PATH_SEPARATOR = "/" +REPORT_DURATION_PRECISION = 6 +JSON_INDENT = 2 +NORMALIZATION_RULES = ( + "only generated summary IDs map to stable references; caller-provided event IDs remain exact", + "timestamps compare by relative rank so ordering violations remain visible", + "null and empty long_running_tool_ids normalize to an empty list", + "memory timestamps are excluded while duplicate counts remain", + "dictionary, set, and *_json field order is canonicalized", + "summary text alone receives Unicode NFKC, newline, and whitespace normalization", +) + + +@dataclass(frozen=True) +class AllowedDiff: + """One exact backend-specific difference rule.""" + + backend_pair: tuple[str, str] + field_path: str + reason: str + + +@dataclass(frozen=True) +class DiffItem: + """One path-level difference.""" + + case_id: str + session_id: str + backend_pair: tuple[str, str] + category: str + field_path: str + left: Any + right: Any + allowed: bool + reason: Optional[str] = None + event_index: Optional[int] = None + summary_id: Optional[str] = None + + +DEFAULT_ALLOWED_DIFFS = ( + AllowedDiff( + ("in_memory", "sqlite_reloaded"), + "/summary/cache_present", + "SessionSummary metadata is process-local; the persisted summary anchor remains authoritative after reload.", + ), + AllowedDiff(("in_memory", "sqlite_reloaded"), "/summary/original_event_count", + "Original event count exists only in the live summary cache."), + AllowedDiff(("in_memory", "sqlite_reloaded"), "/summary/compressed_event_count", + "Compressed event count exists only in the live summary cache."), + AllowedDiff(("in_memory", "sqlite_reloaded"), "/summary/generation", + "Summary generation is process-local and is unavailable after service reconstruction."), + AllowedDiff(("in_memory", "sqlite_reloaded"), "/summary_checkpoints/0", + "Summary checkpoints are process-local validation records."), + AllowedDiff(("in_memory", "sqlite_reloaded"), "/summary_checkpoints/1", + "Second-generation summary checkpoint is process-local."), + AllowedDiff(("in_memory", "sqlite_reloaded"), "/failures/0", + "Injected failure records belong to the replay process, not backend storage."), +) + + +def normalize_text(value: str) -> str: + """Normalize semantic text without fuzzy matching.""" + normalized = unicodedata.normalize("NFKC", value).replace("\r\n", "\n").replace("\r", "\n") + return " ".join(normalized.split()) + + +def normalize_snapshot(snapshot: dict[str, Any]) -> dict[str, Any]: + """Return a stable representation of one backend read model.""" + result = copy.deepcopy(snapshot) + result.pop("backend", None) + result["timeline_constraints"] = _timeline_constraints(result) + result["events"] = _normalize_events(result.get("events", [])) + result["historical_events"] = _normalize_events(result.get("historical_events", [])) + result["memory"] = _normalize_memory(result.get("memory", {})) + result["memory_final"] = _normalize_memory(result.get("memory_final", {})) + summary, checkpoints = _normalize_summary_series( + result.get("summary"), + result.get("summary_checkpoints", []), + ) + result["summary"] = summary + result["summary_checkpoints"] = checkpoints + return _normalize_value(result) + + +def _normalize_events(events: list[dict[str, Any]]) -> list[dict[str, Any]]: + timestamps = sorted({event.get("timestamp") for event in events if event.get("timestamp") is not None}) + timestamp_ranks = {value: rank for rank, value in enumerate(timestamps)} + normalized = [] + for event in events: + item = copy.deepcopy(event) + if item.get("is_summary"): + item["id"] = SUMMARY_ANCHOR_ID + item["long_running_tool_ids"] = sorted(item.get("long_running_tool_ids") or []) + if "timestamp" in item: + item["timestamp"] = timestamp_ranks[item["timestamp"]] + normalized.append(item) + return normalized + + +def _normalize_memory(memory: dict[str, Any]) -> dict[str, Any]: + normalized = {} + for query, response in memory.items(): + entries = [] + for entry in response.get("memories", []): + item = copy.deepcopy(entry) + item.pop("timestamp", None) + entries.append(_normalize_value(item)) + normalized[query] = {"memories": sorted(entries, key=_json_sort_key)} + return normalized + + +def _normalize_summary(summary: Optional[dict[str, Any]]) -> Optional[dict[str, Any]]: + if summary is None: + return None + result = copy.deepcopy(summary) + if result.get("text"): + result["text"] = normalize_text(result["text"]) + anchor = result.get("anchor") + if anchor: + result["anchor"] = _normalize_events([anchor])[0] + return result + + +def _normalize_summary_series( + summary: Optional[dict[str, Any]], + checkpoints: list[dict[str, Any]]) -> tuple[Optional[dict[str, Any]], list[dict[str, Any]]]: + normalized_summary = _normalize_summary(summary) + normalized_checkpoints = [_normalize_summary(checkpoint) for checkpoint in checkpoints] + checkpoint_times = [item.get("updated_at") for item in normalized_checkpoints] + checkpoint_times = [value for value in checkpoint_times if _valid_timestamp(value)] + if normalized_summary is not None: + summary_time = normalized_summary.get("updated_at") + normalized_summary["updated_at"] = { + "valid": + _valid_timestamp(summary_time), + "not_before_checkpoints": + _valid_timestamp(summary_time) + and all(summary_time >= checkpoint_time for checkpoint_time in checkpoint_times), + } + timestamps = sorted(set(checkpoint_times)) + ranks = {value: rank for rank, value in enumerate(timestamps)} + for item in normalized_checkpoints: + if item.get("updated_at") is not None: + item["updated_at"] = ranks[item["updated_at"]] + return normalized_summary, normalized_checkpoints + + +def _valid_timestamp(value: Any) -> bool: + return isinstance(value, (int, float)) and value > 0 + + +def _timeline_constraints(snapshot: dict[str, Any]) -> dict[str, bool]: + events = snapshot.get("events", []) + historical = snapshot.get("historical_events", []) + event_values = [item.get("timestamp") for item in events if not item.get("is_summary")] + history_values = [item.get("timestamp") for item in historical] + summary_values = [item.get("timestamp") for item in events if item.get("is_summary")] + current_times = [value for value in event_values if _valid_timestamp(value)] + history_times = [value for value in history_values if _valid_timestamp(value)] + summary_times = [value for value in summary_values if _valid_timestamp(value)] + return { + "events_all_valid": all(_valid_timestamp(value) for value in event_values), + "events_monotonic": current_times == sorted(current_times), + "history_all_valid": all(_valid_timestamp(value) for value in history_values), + "history_monotonic": history_times == sorted(history_times), + "history_before_current": not history_times or not current_times or max(history_times) <= min(current_times), + "summary_all_valid": all(_valid_timestamp(value) for value in summary_values), + "summary_not_before_history": not history_times or not summary_times + or max(history_times) <= min(summary_times), + } + + +def _normalize_value(value: Any, field_name: str = "") -> Any: + if isinstance(value, dict): + return {key: _normalize_value(value[key], key) for key in sorted(value)} + if isinstance(value, list): + return [_normalize_value(item, field_name) for item in value] + if isinstance(value, set): + return sorted((_normalize_value(item) for item in value), key=_json_sort_key) + if isinstance(value, str) and field_name.endswith("_json"): + try: + return _normalize_value(json.loads(value)) + except (TypeError, ValueError): + return value + return value + + +def _json_sort_key(value: Any) -> str: + return json.dumps(value, ensure_ascii=False, sort_keys=True, default=str) + + +def compare_snapshots(left: dict[str, Any], + right: dict[str, Any], + *, + allowed_diffs: Iterable[AllowedDiff] = DEFAULT_ALLOWED_DIFFS) -> list[DiffItem]: + """Normalize and compare two backend snapshots.""" + backend_pair = (left["backend"], right["backend"]) + raw_diffs: list[tuple[str, Any, Any]] = [] + _walk_diff(normalize_snapshot(left), normalize_snapshot(right), "", raw_diffs) + rules = {(rule.backend_pair, rule.field_path): rule.reason for rule in allowed_diffs} + reverse_rules = {((pair[1], pair[0]), path): reason for (pair, path), reason in rules.items()} + rules.update(reverse_rules) + return [_make_diff(left, backend_pair, raw_diff, rules.get((backend_pair, raw_diff[0]))) for raw_diff in raw_diffs] + + +def compare_persisted_snapshots(left: dict[str, Any], right: dict[str, Any]) -> list[DiffItem]: + """Compare persisted data with exact capability rules for process-local fields.""" + left_final = _with_final_memory(left) + right_final = _with_final_memory(right) + return compare_snapshots(left_final, right_final) + + +def _with_final_memory(snapshot: dict[str, Any]) -> dict[str, Any]: + result = copy.deepcopy(snapshot) + result["memory"] = result.pop("memory_final", {}) + return result + + +def _walk_diff(left: Any, right: Any, path: str, output: list[tuple[str, Any, Any]]) -> None: + if type(left) is not type(right): + output.append((path or STRUCTURAL_PATH_SEPARATOR, left, right)) + return + if isinstance(left, dict): + for key in sorted(set(left) | set(right)): + child_path = f"{path}/{key}" + if key not in left: + output.append((child_path, None, right[key])) + elif key not in right: + output.append((child_path, left[key], None)) + else: + _walk_diff(left[key], right[key], child_path, output) + return + if isinstance(left, list): + for index in range(max(len(left), len(right))): + child_path = f"{path}/{index}" + if index >= len(left): + output.append((child_path, None, right[index])) + elif index >= len(right): + output.append((child_path, left[index], None)) + else: + _walk_diff(left[index], right[index], child_path, output) + return + if left != right: + output.append((path or STRUCTURAL_PATH_SEPARATOR, left, right)) + + +def _make_diff(snapshot: dict[str, Any], backend_pair: tuple[str, str], raw_diff: tuple[str, Any, Any], + reason: Optional[str]) -> DiffItem: + path, left, right = raw_diff + category = path.strip(STRUCTURAL_PATH_SEPARATOR).split(STRUCTURAL_PATH_SEPARATOR)[0] or "snapshot" + event_index = _path_index(path, "events") + if event_index is None: + event_index = _path_index(path, "historical_events") + summary_id = _summary_reference(path, category) + return DiffItem( + case_id=snapshot["case_id"], + session_id=snapshot["session_id"], + backend_pair=backend_pair, + category=category, + field_path=path, + left=left, + right=right, + allowed=reason is not None, + reason=reason, + event_index=event_index, + summary_id=summary_id, + ) + + +def _path_index(path: str, category: str) -> Optional[int]: + parts = path.strip(STRUCTURAL_PATH_SEPARATOR).split(STRUCTURAL_PATH_SEPARATOR) + if len(parts) > 1 and parts[0] == category and parts[1].isdigit(): + return int(parts[1]) + return None + + +def _summary_reference(path: str, category: str) -> Optional[str]: + if category == "summary": + return SUMMARY_ANCHOR_ID + checkpoint_index = _path_index(path, "summary_checkpoints") + if checkpoint_index is not None: + return f"{SUMMARY_ANCHOR_ID}:{checkpoint_index}" + return None + + +def build_report(runs: list[dict[str, Any]], + comparisons: list[dict[str, Any]], + *, + skips: Optional[list[dict[str, str]]] = None, + detection_metrics: Optional[dict[str, Any]] = None) -> dict[str, Any]: + """Build complete stable report with clean-run metrics.""" + _validate_report_identities(runs, comparisons) + allowed_diff_audit = _allowed_diff_audit(comparisons) + _validate_allowed_diff_usage(comparisons, allowed_diff_audit) + unexpected = [diff for comparison in comparisons for diff in comparison["diffs"] if not diff["allowed"]] + false_positive_pairs = sum(1 for comparison in comparisons + if any(not diff["allowed"] for diff in comparison["diffs"])) + pair_count = len(comparisons) + metrics = { + "case_count": len({run["case_id"] + for run in runs}), + "run_count": len(runs), + "comparison_count": pair_count, + "unexpected_diff_count": len(unexpected), + "false_positive_case_pairs": false_positive_pairs, + "false_positive_rate": false_positive_pairs / pair_count if pair_count else 0.0, + } + metrics.update(detection_metrics or {}) + return { + "schema_version": REPORT_SCHEMA_VERSION, + "normalization_rules": list(NORMALIZATION_RULES), + "allowed_diff": [asdict(rule) for rule in DEFAULT_ALLOWED_DIFFS], + "allowed_diff_audit": allowed_diff_audit, + "runs": sorted(runs, key=lambda item: (item["case_id"], item["backend"])), + "comparisons": sorted(comparisons, key=lambda item: (item["case_id"], item["backend_pair"])), + "metrics": metrics, + "skips": sorted(skips or [], key=lambda item: item["backend"]), + } + + +def _allowed_diff_audit(comparisons: list[dict[str, Any]]) -> list[dict[str, Any]]: + audit = [] + for rule in DEFAULT_ALLOWED_DIFFS: + hits = sum(1 for comparison in comparisons for diff in comparison["diffs"] + if tuple(comparison["backend_pair"]) == rule.backend_pair and diff["field_path"] == rule.field_path + and diff["allowed"]) + item = asdict(rule) + item["hit_count"] = hits + audit.append(item) + return audit + + +def _validate_report_identities(runs: list[dict[str, Any]], comparisons: list[dict[str, Any]]) -> None: + run_keys = [(run["case_id"], run["backend"]) for run in runs] + comparison_keys = [(item["case_id"], tuple(item["backend_pair"])) for item in comparisons] + duplicate_runs = sorted({key for key in run_keys if run_keys.count(key) > 1}) + duplicate_comparisons = sorted({key for key in comparison_keys if comparison_keys.count(key) > 1}) + if duplicate_runs or duplicate_comparisons: + raise ValueError(f"duplicate report identities: runs={duplicate_runs}, comparisons={duplicate_comparisons}") + + +def _validate_allowed_diff_usage(comparisons: list[dict[str, Any]], audit: list[dict[str, Any]]) -> None: + executed_pairs = {tuple(item["backend_pair"]) for item in comparisons} + unused = [ + item["field_path"] for item in audit if tuple(item["backend_pair"]) in executed_pairs and item["hit_count"] == 0 + ] + if unused: + raise ValueError(f"unused allowed-diff rules: {sorted(unused)}") + + +def comparison_record(case_id: str, backend_pair: tuple[str, str], diffs: Iterable[DiffItem]) -> dict[str, Any]: + """Serialize one comparison, including zero-diff comparisons.""" + serialized = [asdict(diff) for diff in diffs] + serialized.sort(key=lambda item: (item["field_path"], _json_sort_key(item["left"]))) + return { + "case_id": case_id, + "backend_pair": list(backend_pair), + "status": "different" if any(not item["allowed"] for item in serialized) else "consistent", + "diffs": serialized, + } + + +def run_record(snapshot: dict[str, Any], duration_seconds: float) -> dict[str, Any]: + """Create compact per-case/backend report entry.""" + return { + "case_id": snapshot["case_id"], + "backend": snapshot["backend"], + "status": "passed", + "duration_seconds": round(duration_seconds, REPORT_DURATION_PRECISION), + "snapshot": { + "event_count": len(snapshot["events"]), + "historical_event_count": len(snapshot["historical_events"]), + "state_keys": sorted(snapshot["state"]), + "memory_queries": sorted(snapshot["memory"] or snapshot.get("memory_final", {})), + "summary_present": snapshot["summary"] is not None, + }, + } + + +def write_report(report: dict[str, Any], target: Path) -> None: + """Atomically write a stable UTF-8 JSON report.""" + temporary = target.with_suffix(f"{target.suffix}.tmp-{os.getpid()}") + temporary.write_text( + json.dumps(report, ensure_ascii=False, indent=JSON_INDENT, sort_keys=True) + "\n", + encoding="utf-8", + ) + temporary.replace(target) + + +Mutation = Callable[[dict[str, Any]], None] + + +def mutate_snapshot(snapshot: dict[str, Any], mutation: Mutation) -> dict[str, Any]: + """Apply a read-model mutation before normalization.""" + result = copy.deepcopy(snapshot) + mutation(result) + return result diff --git a/tests/sessions/session_memory_summary_diff_report.json b/tests/sessions/session_memory_summary_diff_report.json new file mode 100644 index 000000000..6c228e638 --- /dev/null +++ b/tests/sessions/session_memory_summary_diff_report.json @@ -0,0 +1 @@ +{"allowed_diff":[{"backend_pair":["in_memory","sqlite_reloaded"],"field_path":"/summary/cache_present","reason":"SessionSummary metadata is process-local; the persisted summary anchor remains authoritative after reload."},{"backend_pair":["in_memory","sqlite_reloaded"],"field_path":"/summary/original_event_count","reason":"Original event count exists only in the live summary cache."},{"backend_pair":["in_memory","sqlite_reloaded"],"field_path":"/summary/compressed_event_count","reason":"Compressed event count exists only in the live summary cache."},{"backend_pair":["in_memory","sqlite_reloaded"],"field_path":"/summary/generation","reason":"Summary generation is process-local and is unavailable after service reconstruction."},{"backend_pair":["in_memory","sqlite_reloaded"],"field_path":"/summary_checkpoints/0","reason":"Summary checkpoints are process-local validation records."},{"backend_pair":["in_memory","sqlite_reloaded"],"field_path":"/summary_checkpoints/1","reason":"Second-generation summary checkpoint is process-local."},{"backend_pair":["in_memory","sqlite_reloaded"],"field_path":"/failures/0","reason":"Injected failure records belong to the replay process, not backend storage."}],"allowed_diff_audit":[{"backend_pair":["in_memory","sqlite_reloaded"],"field_path":"/summary/cache_present","hit_count":4,"reason":"SessionSummary metadata is process-local; the persisted summary anchor remains authoritative after reload."},{"backend_pair":["in_memory","sqlite_reloaded"],"field_path":"/summary/original_event_count","hit_count":4,"reason":"Original event count exists only in the live summary cache."},{"backend_pair":["in_memory","sqlite_reloaded"],"field_path":"/summary/compressed_event_count","hit_count":4,"reason":"Compressed event count exists only in the live summary cache."},{"backend_pair":["in_memory","sqlite_reloaded"],"field_path":"/summary/generation","hit_count":4,"reason":"Summary generation is process-local and is unavailable after service reconstruction."},{"backend_pair":["in_memory","sqlite_reloaded"],"field_path":"/summary_checkpoints/0","hit_count":4,"reason":"Summary checkpoints are process-local validation records."},{"backend_pair":["in_memory","sqlite_reloaded"],"field_path":"/summary_checkpoints/1","hit_count":1,"reason":"Second-generation summary checkpoint is process-local."},{"backend_pair":["in_memory","sqlite_reloaded"],"field_path":"/failures/0","hit_count":1,"reason":"Injected failure records belong to the replay process, not backend storage."}],"comparisons":[{"backend_pair":["in_memory","sqlite"],"case_id":"duplicate_retry","diffs":[],"status":"consistent"},{"backend_pair":["in_memory","sqlite_reloaded"],"case_id":"duplicate_retry","diffs":[{"allowed":true,"backend_pair":["in_memory","sqlite_reloaded"],"case_id":"duplicate_retry","category":"summary","event_index":null,"field_path":"/summary/cache_present","left":true,"reason":"SessionSummary metadata is process-local; the persisted summary anchor remains authoritative after reload.","right":false,"session_id":"replay-session","summary_id":""},{"allowed":true,"backend_pair":["in_memory","sqlite_reloaded"],"case_id":"duplicate_retry","category":"summary","event_index":null,"field_path":"/summary/compressed_event_count","left":1,"reason":"Compressed event count exists only in the live summary cache.","right":null,"session_id":"replay-session","summary_id":""},{"allowed":true,"backend_pair":["in_memory","sqlite_reloaded"],"case_id":"duplicate_retry","category":"summary","event_index":null,"field_path":"/summary/generation","left":1,"reason":"Summary generation is process-local and is unavailable after service reconstruction.","right":0,"session_id":"replay-session","summary_id":""},{"allowed":true,"backend_pair":["in_memory","sqlite_reloaded"],"case_id":"duplicate_retry","category":"summary","event_index":null,"field_path":"/summary/original_event_count","left":1,"reason":"Original event count exists only in the live summary cache.","right":null,"session_id":"replay-session","summary_id":""},{"allowed":true,"backend_pair":["in_memory","sqlite_reloaded"],"case_id":"duplicate_retry","category":"summary_checkpoints","event_index":null,"field_path":"/summary_checkpoints/0","left":{"anchor":{"actions":{"artifact_delta":{},"state_delta":{}},"author":"system","content":{"parts":[{"text":"Previous conversation summary: deterministic-summary-v1: user: Write exactly once\nuser: Write exactly once"}],"role":"user"},"id":"","invocation_id":"summary","is_summary":true,"long_running_tool_ids":[],"model_flags":3,"requires_completion":false,"timestamp":0,"version":0,"visible":true},"anchor_count":1,"cache_present":true,"compressed_event_count":1,"generation":1,"original_event_count":1,"session_id":"replay-session","text":"deterministic-summary-v1: user: Write exactly once user: Write exactly once","updated_at":0},"reason":"Summary checkpoints are process-local validation records.","right":null,"session_id":"replay-session","summary_id":":0"}],"status":"consistent"},{"backend_pair":["in_memory","sqlite"],"case_id":"memory_preference","diffs":[],"status":"consistent"},{"backend_pair":["in_memory","sqlite_reloaded"],"case_id":"memory_preference","diffs":[],"status":"consistent"},{"backend_pair":["in_memory","sqlite"],"case_id":"multi_turn","diffs":[],"status":"consistent"},{"backend_pair":["in_memory","sqlite_reloaded"],"case_id":"multi_turn","diffs":[],"status":"consistent"},{"backend_pair":["in_memory","sqlite"],"case_id":"single_turn","diffs":[],"status":"consistent"},{"backend_pair":["in_memory","sqlite_reloaded"],"case_id":"single_turn","diffs":[],"status":"consistent"},{"backend_pair":["in_memory","sqlite"],"case_id":"state_overwrite","diffs":[],"status":"consistent"},{"backend_pair":["in_memory","sqlite_reloaded"],"case_id":"state_overwrite","diffs":[],"status":"consistent"},{"backend_pair":["in_memory","sqlite"],"case_id":"summary_create","diffs":[],"status":"consistent"},{"backend_pair":["in_memory","sqlite_reloaded"],"case_id":"summary_create","diffs":[{"allowed":true,"backend_pair":["in_memory","sqlite_reloaded"],"case_id":"summary_create","category":"summary","event_index":null,"field_path":"/summary/cache_present","left":true,"reason":"SessionSummary metadata is process-local; the persisted summary anchor remains authoritative after reload.","right":false,"session_id":"replay-session","summary_id":""},{"allowed":true,"backend_pair":["in_memory","sqlite_reloaded"],"case_id":"summary_create","category":"summary","event_index":null,"field_path":"/summary/compressed_event_count","left":3,"reason":"Compressed event count exists only in the live summary cache.","right":null,"session_id":"replay-session","summary_id":""},{"allowed":true,"backend_pair":["in_memory","sqlite_reloaded"],"case_id":"summary_create","category":"summary","event_index":null,"field_path":"/summary/generation","left":1,"reason":"Summary generation is process-local and is unavailable after service reconstruction.","right":0,"session_id":"replay-session","summary_id":""},{"allowed":true,"backend_pair":["in_memory","sqlite_reloaded"],"case_id":"summary_create","category":"summary","event_index":null,"field_path":"/summary/original_event_count","left":3,"reason":"Original event count exists only in the live summary cache.","right":null,"session_id":"replay-session","summary_id":""},{"allowed":true,"backend_pair":["in_memory","sqlite_reloaded"],"case_id":"summary_create","category":"summary_checkpoints","event_index":null,"field_path":"/summary_checkpoints/0","left":{"anchor":{"actions":{"artifact_delta":{},"state_delta":{}},"author":"system","content":{"parts":[{"text":"Previous conversation summary: deterministic-summary-v1: user: Plan a trip\nuser: Plan a trip"}],"role":"user"},"id":"","invocation_id":"summary","is_summary":true,"long_running_tool_ids":[],"model_flags":3,"requires_completion":false,"timestamp":0,"version":0,"visible":true},"anchor_count":1,"cache_present":true,"compressed_event_count":3,"generation":1,"original_event_count":3,"session_id":"replay-session","text":"deterministic-summary-v1: user: Plan a trip user: Plan a trip","updated_at":0},"reason":"Summary checkpoints are process-local validation records.","right":null,"session_id":"replay-session","summary_id":":0"}],"status":"consistent"},{"backend_pair":["in_memory","sqlite"],"case_id":"summary_truncation","diffs":[],"status":"consistent"},{"backend_pair":["in_memory","sqlite_reloaded"],"case_id":"summary_truncation","diffs":[{"allowed":true,"backend_pair":["in_memory","sqlite_reloaded"],"case_id":"summary_truncation","category":"summary","event_index":null,"field_path":"/summary/cache_present","left":true,"reason":"SessionSummary metadata is process-local; the persisted summary anchor remains authoritative after reload.","right":false,"session_id":"replay-session","summary_id":""},{"allowed":true,"backend_pair":["in_memory","sqlite_reloaded"],"case_id":"summary_truncation","category":"summary","event_index":null,"field_path":"/summary/compressed_event_count","left":3,"reason":"Compressed event count exists only in the live summary cache.","right":null,"session_id":"replay-session","summary_id":""},{"allowed":true,"backend_pair":["in_memory","sqlite_reloaded"],"case_id":"summary_truncation","category":"summary","event_index":null,"field_path":"/summary/generation","left":1,"reason":"Summary generation is process-local and is unavailable after service reconstruction.","right":0,"session_id":"replay-session","summary_id":""},{"allowed":true,"backend_pair":["in_memory","sqlite_reloaded"],"case_id":"summary_truncation","category":"summary","event_index":null,"field_path":"/summary/original_event_count","left":4,"reason":"Original event count exists only in the live summary cache.","right":null,"session_id":"replay-session","summary_id":""},{"allowed":true,"backend_pair":["in_memory","sqlite_reloaded"],"case_id":"summary_truncation","category":"summary_checkpoints","event_index":null,"field_path":"/summary_checkpoints/0","left":{"anchor":{"actions":{"artifact_delta":{},"state_delta":{}},"author":"system","content":{"parts":[{"text":"Previous conversation summary: deterministic-summary-v1: user: Old context\nuser: Old context\nagent: Old answer\nagent: Old answer"}],"role":"user"},"id":"","invocation_id":"summary","is_summary":true,"long_running_tool_ids":[],"model_flags":3,"requires_completion":false,"timestamp":0,"version":0,"visible":true},"anchor_count":1,"cache_present":true,"compressed_event_count":3,"generation":1,"original_event_count":4,"session_id":"replay-session","text":"deterministic-summary-v1: user: Old context user: Old context agent: Old answer agent: Old answer","updated_at":0},"reason":"Summary checkpoints are process-local validation records.","right":null,"session_id":"replay-session","summary_id":":0"}],"status":"consistent"},{"backend_pair":["in_memory","sqlite"],"case_id":"summary_update","diffs":[],"status":"consistent"},{"backend_pair":["in_memory","sqlite_reloaded"],"case_id":"summary_update","diffs":[{"allowed":true,"backend_pair":["in_memory","sqlite_reloaded"],"case_id":"summary_update","category":"summary","event_index":null,"field_path":"/summary/cache_present","left":true,"reason":"SessionSummary metadata is process-local; the persisted summary anchor remains authoritative after reload.","right":false,"session_id":"replay-session","summary_id":""},{"allowed":true,"backend_pair":["in_memory","sqlite_reloaded"],"case_id":"summary_update","category":"summary","event_index":null,"field_path":"/summary/compressed_event_count","left":3,"reason":"Compressed event count exists only in the live summary cache.","right":null,"session_id":"replay-session","summary_id":""},{"allowed":true,"backend_pair":["in_memory","sqlite_reloaded"],"case_id":"summary_update","category":"summary","event_index":null,"field_path":"/summary/generation","left":2,"reason":"Summary generation is process-local and is unavailable after service reconstruction.","right":0,"session_id":"replay-session","summary_id":""},{"allowed":true,"backend_pair":["in_memory","sqlite_reloaded"],"case_id":"summary_update","category":"summary","event_index":null,"field_path":"/summary/original_event_count","left":5,"reason":"Original event count exists only in the live summary cache.","right":null,"session_id":"replay-session","summary_id":""},{"allowed":true,"backend_pair":["in_memory","sqlite_reloaded"],"case_id":"summary_update","category":"summary_checkpoints","event_index":null,"field_path":"/summary_checkpoints/0","left":{"anchor":{"actions":{"artifact_delta":{},"state_delta":{}},"author":"system","content":{"parts":[{"text":"Previous conversation summary: deterministic-summary-v1: user: Initial requirement\nuser: Initial requirement"}],"role":"user"},"id":"","invocation_id":"summary","is_summary":true,"long_running_tool_ids":[],"model_flags":3,"requires_completion":false,"timestamp":0,"version":0,"visible":true},"anchor_count":1,"cache_present":true,"compressed_event_count":3,"generation":1,"original_event_count":3,"session_id":"replay-session","text":"deterministic-summary-v1: user: Initial requirement user: Initial requirement","updated_at":0},"reason":"Summary checkpoints are process-local validation records.","right":null,"session_id":"replay-session","summary_id":":0"},{"allowed":true,"backend_pair":["in_memory","sqlite_reloaded"],"case_id":"summary_update","category":"summary_checkpoints","event_index":null,"field_path":"/summary_checkpoints/1","left":{"anchor":{"actions":{"artifact_delta":{},"state_delta":{}},"author":"system","content":{"parts":[{"text":"Previous conversation summary: deterministic-summary-v2: system: Previous conversation summary: deterministic-summary-v1: user: Initial requirement\nuser: Initial requirement\nsystem: Previous conversation summary: deterministic-summary-v1: user: Initial requirement\nuser: Initial requirement\nagent: Initial decision\nagent: Initial decision\nuser: Keep details\nuser: Keep details"}],"role":"user"},"id":"","invocation_id":"summary","is_summary":true,"long_running_tool_ids":[],"model_flags":3,"requires_completion":false,"timestamp":0,"version":0,"visible":true},"anchor_count":1,"cache_present":true,"compressed_event_count":3,"generation":2,"original_event_count":5,"session_id":"replay-session","text":"deterministic-summary-v2: system: Previous conversation summary: deterministic-summary-v1: user: Initial requirement user: Initial requirement system: Previous conversation summary: deterministic-summary-v1: user: Initial requirement user: Initial requirement agent: Initial decision agent: Initial decision user: Keep details user: Keep details","updated_at":1},"reason":"Second-generation summary checkpoint is process-local.","right":null,"session_id":"replay-session","summary_id":":1"}],"status":"consistent"},{"backend_pair":["in_memory","sqlite"],"case_id":"tool_round_trip","diffs":[],"status":"consistent"},{"backend_pair":["in_memory","sqlite_reloaded"],"case_id":"tool_round_trip","diffs":[],"status":"consistent"},{"backend_pair":["in_memory","sqlite"],"case_id":"write_recovery","diffs":[],"status":"consistent"},{"backend_pair":["in_memory","sqlite_reloaded"],"case_id":"write_recovery","diffs":[{"allowed":true,"backend_pair":["in_memory","sqlite_reloaded"],"case_id":"write_recovery","category":"failures","event_index":null,"field_path":"/failures/0","left":{"error_type":"InjectedReplayFailure","message":"injected before-call failure","operation_index":1},"reason":"Injected failure records belong to the replay process, not backend storage.","right":null,"session_id":"replay-session","summary_id":null}],"status":"consistent"}],"metrics":{"case_count":10,"comparison_count":20,"false_positive_case_pairs":0,"false_positive_rate":0.0,"mutation_count":11,"mutation_detected":11,"mutation_detection_rate":1.0,"run_count":30,"summary_lost_detection_rate":1.0,"summary_overwrite_detection_rate":1.0,"summary_session_detection_rate":1.0,"unexpected_diff_count":0},"normalization_rules":["only generated summary IDs map to stable references; caller-provided event IDs remain exact","timestamps compare by relative rank so ordering violations remain visible","null and empty long_running_tool_ids normalize to an empty list","memory timestamps are excluded while duplicate counts remain","dictionary, set, and *_json field order is canonicalized","summary text alone receives Unicode NFKC, newline, and whitespace normalization"],"runs":[{"backend":"in_memory","case_id":"duplicate_retry","duration_seconds":0.001551,"snapshot":{"event_count":1,"historical_event_count":1,"memory_queries":[],"state_keys":[],"summary_present":true},"status":"passed"},{"backend":"sqlite","case_id":"duplicate_retry","duration_seconds":0.142272,"snapshot":{"event_count":1,"historical_event_count":1,"memory_queries":[],"state_keys":[],"summary_present":true},"status":"passed"},{"backend":"sqlite_reloaded","case_id":"duplicate_retry","duration_seconds":0.142272,"snapshot":{"event_count":1,"historical_event_count":1,"memory_queries":[],"state_keys":[],"summary_present":true},"status":"passed"},{"backend":"in_memory","case_id":"memory_preference","duration_seconds":0.001585,"snapshot":{"event_count":2,"historical_event_count":0,"memory_queries":["jasmine"],"state_keys":[],"summary_present":false},"status":"passed"},{"backend":"sqlite","case_id":"memory_preference","duration_seconds":0.098647,"snapshot":{"event_count":2,"historical_event_count":0,"memory_queries":["jasmine"],"state_keys":[],"summary_present":false},"status":"passed"},{"backend":"sqlite_reloaded","case_id":"memory_preference","duration_seconds":0.098647,"snapshot":{"event_count":2,"historical_event_count":0,"memory_queries":["jasmine"],"state_keys":[],"summary_present":false},"status":"passed"},{"backend":"in_memory","case_id":"multi_turn","duration_seconds":0.000596,"snapshot":{"event_count":6,"historical_event_count":0,"memory_queries":[],"state_keys":[],"summary_present":false},"status":"passed"},{"backend":"sqlite","case_id":"multi_turn","duration_seconds":0.106921,"snapshot":{"event_count":6,"historical_event_count":0,"memory_queries":[],"state_keys":[],"summary_present":false},"status":"passed"},{"backend":"sqlite_reloaded","case_id":"multi_turn","duration_seconds":0.106921,"snapshot":{"event_count":6,"historical_event_count":0,"memory_queries":[],"state_keys":[],"summary_present":false},"status":"passed"},{"backend":"in_memory","case_id":"single_turn","duration_seconds":0.00055,"snapshot":{"event_count":2,"historical_event_count":0,"memory_queries":[],"state_keys":[],"summary_present":false},"status":"passed"},{"backend":"sqlite","case_id":"single_turn","duration_seconds":0.078033,"snapshot":{"event_count":2,"historical_event_count":0,"memory_queries":[],"state_keys":[],"summary_present":false},"status":"passed"},{"backend":"sqlite_reloaded","case_id":"single_turn","duration_seconds":0.078033,"snapshot":{"event_count":2,"historical_event_count":0,"memory_queries":[],"state_keys":[],"summary_present":false},"status":"passed"},{"backend":"in_memory","case_id":"state_overwrite","duration_seconds":0.000468,"snapshot":{"event_count":2,"historical_event_count":0,"memory_queries":[],"state_keys":["app:region","large_counter","theme","user:language"],"summary_present":false},"status":"passed"},{"backend":"sqlite","case_id":"state_overwrite","duration_seconds":0.076003,"snapshot":{"event_count":2,"historical_event_count":0,"memory_queries":[],"state_keys":["app:region","large_counter","theme","user:language"],"summary_present":false},"status":"passed"},{"backend":"sqlite_reloaded","case_id":"state_overwrite","duration_seconds":0.076003,"snapshot":{"event_count":2,"historical_event_count":0,"memory_queries":[],"state_keys":["app:region","large_counter","theme","user:language"],"summary_present":false},"status":"passed"},{"backend":"in_memory","case_id":"summary_create","duration_seconds":0.001488,"snapshot":{"event_count":3,"historical_event_count":1,"memory_queries":[],"state_keys":[],"summary_present":true},"status":"passed"},{"backend":"sqlite","case_id":"summary_create","duration_seconds":0.108466,"snapshot":{"event_count":3,"historical_event_count":1,"memory_queries":[],"state_keys":[],"summary_present":true},"status":"passed"},{"backend":"sqlite_reloaded","case_id":"summary_create","duration_seconds":0.108466,"snapshot":{"event_count":3,"historical_event_count":1,"memory_queries":[],"state_keys":[],"summary_present":true},"status":"passed"},{"backend":"in_memory","case_id":"summary_truncation","duration_seconds":0.001233,"snapshot":{"event_count":5,"historical_event_count":2,"memory_queries":[],"state_keys":[],"summary_present":true},"status":"passed"},{"backend":"sqlite","case_id":"summary_truncation","duration_seconds":0.127522,"snapshot":{"event_count":5,"historical_event_count":2,"memory_queries":[],"state_keys":[],"summary_present":true},"status":"passed"},{"backend":"sqlite_reloaded","case_id":"summary_truncation","duration_seconds":0.127522,"snapshot":{"event_count":5,"historical_event_count":2,"memory_queries":[],"state_keys":[],"summary_present":true},"status":"passed"},{"backend":"in_memory","case_id":"summary_update","duration_seconds":0.001456,"snapshot":{"event_count":3,"historical_event_count":4,"memory_queries":[],"state_keys":[],"summary_present":true},"status":"passed"},{"backend":"sqlite","case_id":"summary_update","duration_seconds":0.153616,"snapshot":{"event_count":3,"historical_event_count":4,"memory_queries":[],"state_keys":[],"summary_present":true},"status":"passed"},{"backend":"sqlite_reloaded","case_id":"summary_update","duration_seconds":0.153616,"snapshot":{"event_count":3,"historical_event_count":4,"memory_queries":[],"state_keys":[],"summary_present":true},"status":"passed"},{"backend":"in_memory","case_id":"tool_round_trip","duration_seconds":0.000591,"snapshot":{"event_count":4,"historical_event_count":0,"memory_queries":[],"state_keys":[],"summary_present":false},"status":"passed"},{"backend":"sqlite","case_id":"tool_round_trip","duration_seconds":0.083496,"snapshot":{"event_count":4,"historical_event_count":0,"memory_queries":[],"state_keys":[],"summary_present":false},"status":"passed"},{"backend":"sqlite_reloaded","case_id":"tool_round_trip","duration_seconds":0.083496,"snapshot":{"event_count":4,"historical_event_count":0,"memory_queries":[],"state_keys":[],"summary_present":false},"status":"passed"},{"backend":"in_memory","case_id":"write_recovery","duration_seconds":0.000476,"snapshot":{"event_count":1,"historical_event_count":0,"memory_queries":["recovered"],"state_keys":["status"],"summary_present":false},"status":"passed"},{"backend":"sqlite","case_id":"write_recovery","duration_seconds":0.101335,"snapshot":{"event_count":1,"historical_event_count":0,"memory_queries":["recovered"],"state_keys":["status"],"summary_present":false},"status":"passed"},{"backend":"sqlite_reloaded","case_id":"write_recovery","duration_seconds":0.101335,"snapshot":{"event_count":1,"historical_event_count":0,"memory_queries":["recovered"],"state_keys":["status"],"summary_present":false},"status":"passed"}],"schema_version":1,"skips":[{"backend":"redis","reason":"TRPC_REPLAY_REDIS_URL is not set"}]} diff --git a/tests/sessions/test_replay_consistency.py b/tests/sessions/test_replay_consistency.py new file mode 100644 index 000000000..df66f5db7 --- /dev/null +++ b/tests/sessions/test_replay_consistency.py @@ -0,0 +1,746 @@ +# +# 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 the Apache License Version 2.0. +# +"""Acceptance tests for the Session / Memory / Summary replay harness.""" + +from __future__ import annotations + +import ast +import asyncio +import copy +import os +import time +import uuid +from dataclasses import dataclass +from pathlib import Path +from typing import Any +from typing import Callable +from unittest.mock import AsyncMock + +import pytest +import pytest_asyncio + +from trpc_agent_sdk.events import Event +from trpc_agent_sdk.sessions import InMemorySessionService +from trpc_agent_sdk.types import Content +from trpc_agent_sdk.types import Part + +from .replay_cases import APP_NAME +from .replay_cases import BASE_TIMESTAMP +from .replay_cases import ExpectedOutcome +from .replay_cases import LARGE_INTEGER_VALUE +from .replay_cases import OperationKind +from .replay_cases import REPLAY_CASES +from .replay_cases import ReplayCase +from .replay_cases import ReplayOperation +from .replay_cases import SESSION_ID +from .replay_cases import USER_ID +from .replay_cases import validate_replay_cases +from .replay_harness import ReplayRunner +from .replay_harness import ReplayIdentity +from .replay_harness import _clone_operation +from .replay_harness import create_in_memory_backend +from .replay_harness import create_redis_backend +from .replay_harness import create_sqlite_backend +from .replay_harness import validate_snapshot +from .replay_report import DiffItem +from .replay_report import build_report +from .replay_report import compare_snapshots +from .replay_report import compare_persisted_snapshots +from .replay_report import comparison_record +from .replay_report import mutate_snapshot +from .replay_report import run_record +from .replay_report import write_report + +LIGHTWEIGHT_TIMEOUT_SECONDS = 30.0 +CONCURRENT_SESSION_COUNT = 10 +CONCURRENT_ROUNDS = 5 +MINIMUM_CASE_COUNT = 10 +EXPECTED_MUTATION_COUNT = 11 +REPORT_PATH = Path(__file__).parent / "session_memory_summary_diff_report.json" +INJECTED_OPERATION_INDEX = 99 +BACKEND_RUNS_PER_CASE = 3 +COMPARISONS_PER_CASE = 2 +INVALID_FAILURE_COUNT = 2 +INVALID_SUMMARY_GENERATION = 2 +INVALID_ANCHOR_COUNT = 2 +INVALID_HISTORICAL_MINIMUM = 100 +MINIMUM_EXPECTATION_ERRORS = 8 +MAX_FALSE_POSITIVE_RATE = 0.05 +MAX_FUNCTION_LINES = 80 +MAX_FUNCTION_STATEMENTS = 60 +MAX_FUNCTION_PARAMETERS = 4 +MAX_FILE_LINES = 1000 +STRUCTURAL_NUMBERS = {-1, 0, 1} +REDIS_ENVIRONMENT_VARIABLE = "TRPC_REPLAY_REDIS_URL" +BACKEND_MODE_ENVIRONMENT_VARIABLE = "TRPC_REPLAY_BACKENDS" +IN_MEMORY_BACKEND_MODE = "in_memory" +QUALITY_FILES = ( + "replay_harness.py", + "replay_report.py", + "replay_cases.py", + "test_replay_consistency.py", + "test_replay_real_agent.py", +) + + +@pytest_asyncio.fixture(scope="module", loop_scope="module") +async def replay_matrix(tmp_path_factory): + """Run all public cases once and retain detached snapshots.""" + validate_replay_cases(REPLAY_CASES) + root = tmp_path_factory.mktemp("replay-matrix") + matrix: dict[str, dict[str, Any]] = {} + failures = [] + started = time.perf_counter() + for replay_case in REPLAY_CASES: + try: + matrix[replay_case.case_id] = await _run_backend_pair(replay_case, root) + except Exception as error: # pylint: disable=broad-except + failures.append(f"{replay_case.case_id}: {type(error).__name__}: {error}") + matrix["_duration"] = {"seconds": time.perf_counter() - started} + if failures: + pytest.fail("replay matrix case failures:\n" + "\n".join(failures), pytrace=False) + return matrix + + +async def _run_backend_pair(replay_case, root: Path) -> dict[str, Any]: + in_memory = None + sqlite = None + reopened = None + db_path = root / f"{replay_case.case_id}.db" + try: + in_memory = await create_in_memory_backend() + sqlite = await create_sqlite_backend(db_path) + in_memory_started = time.perf_counter() + in_memory_snapshot = await ReplayRunner(in_memory).run(replay_case) + in_memory_duration = time.perf_counter() - in_memory_started + sqlite_started = time.perf_counter() + sqlite_snapshot = await ReplayRunner(sqlite).run(replay_case) + await sqlite.close() + sqlite = None + reopened = await create_sqlite_backend(db_path) + reloaded_snapshot = await ReplayRunner(reopened).reload_snapshot(replay_case) + sqlite_duration = time.perf_counter() - sqlite_started + return { + "in_memory": in_memory_snapshot, + "sqlite": sqlite_snapshot, + "sqlite_reloaded": reloaded_snapshot, + "durations": { + "in_memory": in_memory_duration, + "sqlite": sqlite_duration, + }, + } + finally: + await asyncio.gather(*(backend.close() for backend in (in_memory, sqlite, reopened) if backend is not None)) + + +@pytest.mark.parametrize("case_id", [case.case_id for case in REPLAY_CASES]) +def test_public_replay_case_is_consistent(replay_matrix, case_id): + pair = replay_matrix[case_id] + live_diffs = compare_snapshots(pair["in_memory"], pair["sqlite"]) + persisted_diffs = compare_persisted_snapshots(pair["in_memory"], pair["sqlite_reloaded"]) + assert [diff for diff in [*live_diffs, *persisted_diffs] if not diff.allowed] == [] + + +@pytest.mark.parametrize("case_id", [case.case_id for case in REPLAY_CASES]) +def test_each_backend_matches_independent_expectation(replay_matrix, case_id): + replay_case = next(case for case in REPLAY_CASES if case.case_id == case_id) + pair = replay_matrix[case_id] + assert validate_snapshot(replay_case.expected, pair["in_memory"]) == [] + assert validate_snapshot(replay_case.expected, pair["sqlite"]) == [] + assert validate_snapshot( + replay_case.expected, + pair["sqlite_reloaded"], + require_cache=False, + require_runtime=False, + ) == [] + + +def test_public_case_count_and_lightweight_runtime(replay_matrix): + assert len(REPLAY_CASES) >= MINIMUM_CASE_COUNT + assert replay_matrix["_duration"]["seconds"] < LIGHTWEIGHT_TIMEOUT_SECONDS + + +async def test_in_memory_only_lightweight_mode(): + if os.getenv(BACKEND_MODE_ENVIRONMENT_VARIABLE) != IN_MEMORY_BACKEND_MODE: + pytest.skip(f"set {BACKEND_MODE_ENVIRONMENT_VARIABLE}={IN_MEMORY_BACKEND_MODE} to run the InMemory-only replay") + started = time.perf_counter() + failures = [] + for replay_case in REPLAY_CASES: + backend = await create_in_memory_backend() + try: + snapshot = await ReplayRunner(backend).run(replay_case) + failures.extend(f"{replay_case.case_id}: {error}" + for error in validate_snapshot(replay_case.expected, snapshot)) + finally: + await backend.close() + assert failures == [] + assert time.perf_counter() - started < LIGHTWEIGHT_TIMEOUT_SECONDS + + +def test_duplicate_replay_case_ids_are_rejected(): + duplicate = ReplayCase(REPLAY_CASES[0].case_id, (), ExpectedOutcome(())) + with pytest.raises(ValueError, match="duplicate replay case ids"): + validate_replay_cases((REPLAY_CASES[0], duplicate)) + + +def test_operation_clone_isolates_nested_payloads(): + original = ReplayOperation(OperationKind.APPEND, {"nested": {"values": ["original"]}}) + isolated = _clone_operation(original) + isolated.payload["nested"]["values"].append("mutated") + assert original.payload["nested"]["values"] == ["original"] + + +async def test_physical_session_scope_mismatch_is_rejected(): + backend = await create_in_memory_backend() + try: + session = await backend.session_service.create_session( + app_name=APP_NAME, + user_id=USER_ID, + session_id=SESSION_ID, + ) + backend.session_service.get_session = AsyncMock(return_value=session.model_copy( + update={"app_name": "wrong-app"})) + with pytest.raises(AssertionError, match="physical session scope mismatch"): + await ReplayRunner(backend)._read_session() + finally: + await backend.close() + + +async def test_repeated_memory_search_keeps_point_in_time_results(): + backend = await create_in_memory_backend() + runner = ReplayRunner(backend) + first_result = object() + second_result = object() + backend.memory_service.search_memory = AsyncMock(side_effect=(first_result, second_result)) + operation = ReplayOperation(OperationKind.SEARCH_MEMORY, {"query": "preference"}) + try: + runner.session = await backend.session_service.create_session( + app_name=APP_NAME, + user_id=USER_ID, + session_id=SESSION_ID, + ) + await runner._search_memory(operation, 0) + await runner._search_memory(operation, 1) + assert runner.memory_results == { + "preference": first_result, + "preference#2": second_result, + } + finally: + await backend.close() + + +async def test_reloaded_memory_uses_final_read_without_rewriting_trace_result(tmp_path): + replay_case = ReplayCase( + "point-in-time-memory", + ( + ReplayOperation(OperationKind.CREATE), + ReplayOperation( + OperationKind.APPEND, + { + "event_id": "memory-before", + "author": "user", + "text": "jasmine before search", + }, + ), + ReplayOperation(OperationKind.STORE_MEMORY), + ReplayOperation(OperationKind.SEARCH_MEMORY, {"query": "jasmine"}), + ReplayOperation( + OperationKind.APPEND, + { + "event_id": "memory-after", + "author": "user", + "text": "jasmine after search", + }, + ), + ReplayOperation(OperationKind.STORE_MEMORY), + ), + ExpectedOutcome(("memory-before", "memory-after")), + ) + db_path = tmp_path / "point-in-time-memory.db" + backend = await create_sqlite_backend(db_path) + live = await ReplayRunner(backend).run(replay_case) + await backend.close() + reopened = await create_sqlite_backend(db_path) + try: + reloaded = await ReplayRunner(reopened).reload_snapshot(replay_case) + assert len(live["memory"]["jasmine"]["memories"]) < len(live["memory_final"]["jasmine"]["memories"]) + assert reloaded["memory"] == {} + assert [diff for diff in compare_persisted_snapshots(live, reloaded) if not diff.allowed] == [] + finally: + await reopened.close() + + +def test_large_integer_survives_each_backend(replay_matrix): + snapshots = replay_matrix["state_overwrite"] + for backend in ("in_memory", "sqlite", "sqlite_reloaded"): + assert snapshots[backend]["state"]["large_counter"] == LARGE_INTEGER_VALUE + + +def test_report_rejects_duplicate_identities_and_unused_rules(): + duplicate_run = {"case_id": "duplicate", "backend": "in_memory"} + with pytest.raises(ValueError, match="duplicate report identities"): + build_report([duplicate_run, dict(duplicate_run)], []) + comparison = comparison_record("unused-rule", ("in_memory", "sqlite_reloaded"), []) + with pytest.raises(ValueError, match="unused allowed-diff rules"): + build_report([], [comparison]) + + +async def test_report_contains_every_backend_and_case(replay_matrix, tmp_path): + runs = [] + comparisons = [] + for replay_case in REPLAY_CASES: + pair = replay_matrix[replay_case.case_id] + for backend in ("in_memory", "sqlite", "sqlite_reloaded"): + duration_key = "sqlite" if backend == "sqlite_reloaded" else backend + runs.append(run_record(pair[backend], pair["durations"][duration_key])) + live_diffs = compare_snapshots(pair["in_memory"], pair["sqlite"]) + persisted_diffs = compare_persisted_snapshots(pair["in_memory"], pair["sqlite_reloaded"]) + comparisons.append(comparison_record(replay_case.case_id, ("in_memory", "sqlite"), live_diffs)) + comparisons.append(comparison_record(replay_case.case_id, ("in_memory", "sqlite_reloaded"), persisted_diffs)) + redis_enabled = bool(os.getenv(REDIS_ENVIRONMENT_VARIABLE)) + if redis_enabled: + await _append_redis_report_data(runs, comparisons) + detection = _mutation_results(replay_matrix) + report = build_report( + runs, + comparisons, + skips=[{ + "backend": "redis", + "reason": f"{REDIS_ENVIRONMENT_VARIABLE} is not set", + }] if not os.getenv(REDIS_ENVIRONMENT_VARIABLE) else [], + detection_metrics={ + "mutation_count": len(detection), + "mutation_detected": sum(detection.values()), + "mutation_detection_rate": sum(detection.values()) / len(detection), + "summary_lost_detection_rate": float(detection["summary_lost"]), + "summary_overwrite_detection_rate": float(detection["summary_overwrite"]), + "summary_session_detection_rate": float(detection["summary_session"]), + }, + ) + target = tmp_path / "session_memory_summary_diff_report.json" + write_report(report, target) + persisted = __import__("json").loads(target.read_text(encoding="utf-8")) + assert persisted["metrics"]["case_count"] == len(REPLAY_CASES) + assert persisted["metrics"]["false_positive_rate"] <= MAX_FALSE_POSITIVE_RATE + assert persisted["metrics"]["mutation_detection_rate"] == 1.0 + assert len(persisted["runs"]) == len(REPLAY_CASES) * (BACKEND_RUNS_PER_CASE + int(redis_enabled)) + assert len(persisted["comparisons"]) == len(REPLAY_CASES) * (COMPARISONS_PER_CASE + int(redis_enabled)) + assert persisted["allowed_diff"] + assert all(item["hit_count"] > 0 for item in persisted["allowed_diff_audit"]) + assert persisted["normalization_rules"] + if not redis_enabled: + root_report = __import__("json").loads(REPORT_PATH.read_text(encoding="utf-8")) + assert _report_contract(root_report) == _report_contract(persisted) + + +async def _append_redis_report_data(runs, comparisons): + for replay_case in REPLAY_CASES: + namespace = uuid.uuid4().hex + identity = ReplayIdentity( + app_name=f"{APP_NAME}-{namespace}", + user_id=f"{USER_ID}-{namespace}", + session_id=f"{SESSION_ID}-{namespace}", + ) + in_memory = None + backend = None + try: + in_memory = await create_in_memory_backend(identity) + backend = await create_redis_backend(os.environ[REDIS_ENVIRONMENT_VARIABLE], identity) + expected = await ReplayRunner(in_memory).run(replay_case) + started = time.perf_counter() + snapshot = await ReplayRunner(backend).run(replay_case) + runs.append(run_record(snapshot, time.perf_counter() - started)) + diffs = compare_snapshots(expected, snapshot) + comparisons.append(comparison_record(replay_case.case_id, ("in_memory", "redis"), diffs)) + finally: + await asyncio.gather(*(item.close() for item in (in_memory, backend) if item is not None)) + + +def _report_contract(report): + """Select stable acceptance fields without freezing snapshot structure.""" + schema_version = report["schema_version"] + metrics = report["metrics"] + contract = { + "metrics": { + "case_count": metrics["case_count"], + "comparison_count": metrics["comparison_count"], + "false_positive_rate": metrics["false_positive_rate"], + "mutation_detection_rate": metrics["mutation_detection_rate"], + "unexpected_diff_count": metrics["unexpected_diff_count"], + "summary_lost_detection_rate": metrics["summary_lost_detection_rate"], + "summary_overwrite_detection_rate": metrics["summary_overwrite_detection_rate"], + "summary_session_detection_rate": metrics["summary_session_detection_rate"], + }, + "allowed_diff_audit": [{ + "backend_pair": item["backend_pair"], + "field_path": item["field_path"], + "hit_count": item["hit_count"], + } for item in report["allowed_diff_audit"]], + } + contract["schema_version"] = schema_version + return contract + + +async def test_sqlite_summary_anchor_survives_close_and_reopen(tmp_path): + replay_case = next(case for case in REPLAY_CASES if case.case_id == "summary_update") + db_path = tmp_path / "reopen.db" + backend = await create_sqlite_backend(db_path) + try: + live = await ReplayRunner(backend).run(replay_case) + finally: + await backend.close() + reopened = await create_sqlite_backend(db_path) + try: + runner = ReplayRunner(reopened) + runner.session = await reopened.session_service.get_session( + app_name=APP_NAME, + user_id=USER_ID, + session_id=SESSION_ID, + ) + runner.summary_generation = live["summary"]["generation"] + restored = await runner.snapshot(replay_case.case_id) + finally: + await reopened.close() + assert restored["summary"]["cache_present"] is False + assert restored["summary"]["session_id"] == SESSION_ID + assert restored["summary"]["text"] == live["summary"]["text"] + assert restored["summary"]["anchor_count"] == 1 + + +@dataclass(frozen=True) +class MutationSpec: + """One injected inconsistency and its expected path.""" + + name: str + case_id: str + expected_path: str + mutate: Callable[[dict[str, Any]], None] + + +def _set_event_author(snapshot): + snapshot["events"][0]["author"] = "intruder" + + +def _reverse_events(snapshot): + snapshot["events"][0], snapshot["events"][1] = snapshot["events"][1], snapshot["events"][0] + + +def _change_state(snapshot): + snapshot["state"]["theme"] = "corrupted" + + +def _change_tool_name(snapshot): + snapshot["events"][1]["content"]["parts"][0]["function_call"]["name"] = "wrong-tool" + + +def _remove_memory(snapshot): + snapshot["memory"]["jasmine"]["memories"].pop() + + +def _lose_summary(snapshot): + snapshot["summary"] = None + + +def _restore_old_summary(snapshot): + old_summary = copy.deepcopy(snapshot["summary_checkpoints"][0]) + snapshot["summary"].update(old_summary) + snapshot["events"][0] = copy.deepcopy(old_summary["anchor"]) + + +def _move_summary_session(snapshot): + snapshot["summary"]["session_id"] = "wrong-session" + + +def _reverse_summary_time(snapshot): + snapshot["summary"]["updated_at"] = -1.0 + + +def _duplicate_event(snapshot): + snapshot["events"].append(copy.deepcopy(snapshot["events"][0])) + + +def _add_failure(snapshot): + snapshot["failures"].append({ + "operation_index": INJECTED_OPERATION_INDEX, + "error_type": "InjectedReplayFailure", + "message": "unexpected", + }) + + +MUTATIONS = ( + MutationSpec("event_content", "single_turn", "/events/0/author", _set_event_author), + MutationSpec("event_order", "multi_turn", "/events/0/author", _reverse_events), + MutationSpec("tool_name", "tool_round_trip", "/events/1/content/parts/0/function_call/name", _change_tool_name), + MutationSpec("state_value", "state_overwrite", "/state/theme", _change_state), + MutationSpec("memory_missing", "memory_preference", "/memory/jasmine/memories/1", _remove_memory), + MutationSpec("summary_lost", "summary_create", "/summary", _lose_summary), + MutationSpec("summary_overwrite", "summary_update", "/summary/generation", _restore_old_summary), + MutationSpec("summary_time", "summary_update", "/summary/updated_at/not_before_checkpoints", _reverse_summary_time), + MutationSpec("summary_session", "summary_truncation", "/summary/session_id", _move_summary_session), + MutationSpec("event_duplicate", "duplicate_retry", "/events/1", _duplicate_event), + MutationSpec("dirty_failure", "write_recovery", "/failures/1", _add_failure), +) + + +@pytest.mark.parametrize("spec", MUTATIONS, ids=lambda spec: spec.name) +def test_injected_inconsistency_is_detected(replay_matrix, spec): + pair = replay_matrix[spec.case_id] + mutated = mutate_snapshot(pair["sqlite"], spec.mutate) + diffs = compare_snapshots(pair["in_memory"], mutated) + unexpected_paths = {diff.field_path for diff in diffs if not diff.allowed} + assert spec.expected_path in unexpected_paths + + +def test_mutation_and_required_summary_detection_rates(replay_matrix): + detected = _mutation_results(replay_matrix) + assert len(MUTATIONS) == EXPECTED_MUTATION_COUNT + assert all(detected.values()) + required = {"summary_lost", "summary_overwrite", "summary_session"} + assert all(detected[name] for name in required) + + +def _mutation_results(replay_matrix) -> dict[str, bool]: + results = {} + for spec in MUTATIONS: + pair = replay_matrix[spec.case_id] + mutated = mutate_snapshot(pair["sqlite"], spec.mutate) + diffs = compare_snapshots(pair["in_memory"], mutated) + results[spec.name] = any(diff.field_path == spec.expected_path and not diff.allowed for diff in diffs) + return results + + +def test_diff_localizes_event_and_summary(replay_matrix): + event_pair = replay_matrix["single_turn"] + event_diffs = compare_snapshots( + event_pair["in_memory"], + mutate_snapshot(event_pair["sqlite"], _set_event_author), + ) + event_diff = _find_diff(event_diffs, "/events/0/author") + assert event_diff.event_index == 0 + assert event_diff.session_id == SESSION_ID + summary_pair = replay_matrix["summary_update"] + summary_diffs = compare_snapshots( + summary_pair["in_memory"], + mutate_snapshot(summary_pair["sqlite"], _move_summary_session), + ) + summary_diff = _find_diff(summary_diffs, "/summary/session_id") + assert summary_diff.summary_id == "" + assert summary_diff.left != summary_diff.right + checkpoint_diffs = compare_persisted_snapshots( + summary_pair["in_memory"], + summary_pair["sqlite_reloaded"], + ) + checkpoint_diff = _find_diff(checkpoint_diffs, "/summary_checkpoints/0") + assert checkpoint_diff.summary_id == ":0" + + +def test_normalization_preserves_caller_ids_and_event_text(replay_matrix): + pair = replay_matrix["single_turn"] + changed_id = mutate_snapshot(pair["sqlite"], lambda snapshot: snapshot["events"][0].update(id="wrong-id")) + changed_text = mutate_snapshot( + pair["sqlite"], + lambda snapshot: snapshot["events"][0]["content"]["parts"][0].update(text="Hello replay"), + ) + assert _find_diff(compare_snapshots(pair["in_memory"], changed_id), "/events/0/id") + assert _find_diff(compare_snapshots(pair["in_memory"], changed_text), "/events/0/content/parts/0/text") + + +def test_normalization_detects_invalid_event_timestamp(replay_matrix): + pair = replay_matrix["single_turn"] + mutated = mutate_snapshot(pair["sqlite"], lambda snapshot: snapshot["events"][0].update(timestamp=-1.0)) + diffs = compare_snapshots(pair["in_memory"], mutated) + assert _find_diff(diffs, "/timeline_constraints/events_all_valid") + + +def test_independent_expectation_reports_each_contract_failure(replay_matrix): + snapshot = copy.deepcopy(replay_matrix["summary_update"]["in_memory"]) + snapshot["events"] = [] + snapshot["state"]["temp:leak"] = True + snapshot["summary"]["generation"] = 0 + snapshot["summary"]["text"] = "missing" + snapshot["summary"]["session_id"] = "wrong" + snapshot["summary"]["anchor_count"] = INVALID_ANCHOR_COUNT + snapshot["summary"]["cache_present"] = False + snapshot["summary_checkpoints"] = [] + expected = ExpectedOutcome( + ("required-event", ), + {"required-state": "value"}, + {"required-query": 1}, + summary_generation=INVALID_SUMMARY_GENERATION, + summary_fact="required fact", + minimum_historical_events=INVALID_HISTORICAL_MINIMUM, + failure_count=INVALID_FAILURE_COUNT, + ) + errors = validate_snapshot(expected, snapshot) + assert len(errors) >= MINIMUM_EXPECTATION_ERRORS + assert "unexpected summary" in validate_snapshot(ExpectedOutcome(tuple()), snapshot) + + +def _find_diff(diffs: list[DiffItem], path: str) -> DiffItem: + match = next((diff for diff in diffs if diff.field_path == path), None) + assert match is not None, f"expected diff at {path!r}; got {[diff.field_path for diff in diffs]!r}" + return match + + +async def test_reload_snapshot_clears_reused_runner_state(tmp_path): + source = next(case for case in REPLAY_CASES if case.case_id == "write_recovery") + summary_source = next(case for case in REPLAY_CASES if case.case_id == "summary_create") + summary_events = tuple(operation for operation in summary_source.operations + if operation.kind == OperationKind.APPEND) + replay_case = ReplayCase( + "reload-reused-runner", + source.operations + summary_events + (ReplayOperation(OperationKind.SUMMARY), ), + source.expected, + ) + backend = await create_sqlite_backend(tmp_path / "reused-runner.db") + try: + runner = ReplayRunner(backend) + live = await runner.run(replay_case) + assert live["memory"] + assert live["summary_checkpoints"] + assert live["failures"] + reloaded = await runner.reload_snapshot(replay_case) + finally: + await backend.close() + assert live["memory"] + assert live["summary_checkpoints"] + assert live["failures"] + assert reloaded["memory"] == {} + assert reloaded["memory_final"] + assert reloaded["summary_checkpoints"] == [] + assert reloaded["failures"] == [] + assert reloaded["summary"]["generation"] == 0 + + +async def test_unknown_outcome_retry_writes_event_once(tmp_path): + replay_case = next(case for case in REPLAY_CASES if case.case_id == "duplicate_retry") + backend = await create_sqlite_backend(tmp_path / "unknown.db") + try: + append_event = AsyncMock(wraps=backend.session_service.append_event) + backend.session_service.append_event = append_event + runner = ReplayRunner(backend) + snapshot = await runner.run(replay_case) + memory = await backend.memory_service.search_memory(runner.session.save_key, "Write") + finally: + await backend.close() + ids = [event["id"] for event in [*snapshot["events"], *snapshot["historical_events"]]] + assert append_event.await_count == 1 + assert ids.count("retry-event") == 1 + assert snapshot["summary"]["anchor_count"] == 1 + assert len(memory.memories) == 1 + + +async def test_concurrent_sessions_remain_isolated(): + for _ in range(CONCURRENT_ROUNDS): + service = InMemorySessionService() + try: + await asyncio.gather( + *[_write_isolated_session(service, index) for index in range(CONCURRENT_SESSION_COUNT)]) + sessions = await service.list_sessions(app_name=APP_NAME, user_id=USER_ID) + assert len(sessions.sessions) == CONCURRENT_SESSION_COUNT + finally: + await service.close() + + +async def _write_isolated_session(service, index: int) -> None: + session_id = f"concurrent-{index}" + session = await service.create_session(app_name=APP_NAME, user_id=USER_ID, session_id=session_id) + event = Event( + id=f"event-{index}", + invocation_id=f"invocation-{index}", + author="user", + content=Content(parts=[Part.from_text(text=session_id)]), + timestamp=BASE_TIMESTAMP + index, + ) + await service.append_event(session, event) + stored = await service.get_session(app_name=APP_NAME, user_id=USER_ID, session_id=session_id) + assert stored.events[0].get_text() == session_id + + +def test_new_files_meet_size_and_function_limits(): + base = Path(__file__).parent + violations = [] + for filename in QUALITY_FILES: + path = base / filename + source = path.read_text(encoding="utf-8") + lines = source.splitlines() + if len(lines) > MAX_FILE_LINES: + violations.append(f"{filename}: file has {len(lines)} lines") + tree = ast.parse(source) + violations.extend(_function_violations(filename, tree)) + assert violations == [] + + +def _function_violations(filename: str, tree: ast.AST) -> list[str]: + violations = [] + function_types = (ast.FunctionDef, ast.AsyncFunctionDef) + for node in (item for item in ast.walk(tree) if isinstance(item, function_types)): + end_line = getattr(node, "end_lineno", node.lineno) + line_count = end_line - node.lineno + 1 + statement_count = sum(isinstance(item, ast.stmt) for item in ast.walk(node)) + parameter_count = len(node.args.posonlyargs) + len(node.args.args) + len(node.args.kwonlyargs) + parameter_count += int(node.args.vararg is not None) + int(node.args.kwarg is not None) + if line_count > MAX_FUNCTION_LINES: + violations.append(f"{filename}:{node.lineno} {node.name} has {line_count} lines") + if statement_count > MAX_FUNCTION_STATEMENTS: + violations.append(f"{filename}:{node.lineno} {node.name} has {statement_count} statements") + if parameter_count > MAX_FUNCTION_PARAMETERS: + violations.append(f"{filename}:{node.lineno} {node.name} has {parameter_count} parameters") + return violations + + +def test_harness_has_no_unbound_magic_numbers(): + base = Path(__file__).parent + violations = [] + for filename in QUALITY_FILES: + tree = ast.parse((base / filename).read_text(encoding="utf-8")) + parents = _parent_map(tree) + for node in (item for item in ast.walk(tree) + if isinstance(item, ast.Constant) and isinstance(item.value, (int, float))): + if node.value not in STRUCTURAL_NUMBERS and not _inside_named_constant(node, parents): + violations.append(f"{filename}:{node.lineno} literal {node.value}") + assert violations == [] + + +def _parent_map(tree: ast.AST) -> dict[ast.AST, ast.AST]: + return {child: parent for parent in ast.walk(tree) for child in ast.iter_child_nodes(parent)} + + +def _inside_named_constant(node: ast.AST, parents: dict[ast.AST, ast.AST]) -> bool: + current = node + while current in parents: + current = parents[current] + if isinstance(current, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): + return False + if isinstance(current, (ast.Assign, ast.AnnAssign)): + targets = current.targets if isinstance(current, ast.Assign) else [current.target] + return all(isinstance(target, ast.Name) and target.id.isupper() for target in targets) + return False + + +@pytest.mark.skipif(not os.getenv(REDIS_ENVIRONMENT_VARIABLE), reason=f"{REDIS_ENVIRONMENT_VARIABLE} is not configured") +@pytest.mark.parametrize("case_id", [case.case_id for case in REPLAY_CASES]) +async def test_redis_integration_uses_isolated_namespace(case_id): + namespace = uuid.uuid4().hex + identity = ReplayIdentity( + app_name=f"{APP_NAME}-{namespace}", + user_id=f"{USER_ID}-{namespace}", + session_id=f"{SESSION_ID}-{namespace}", + ) + replay_case = next(case for case in REPLAY_CASES if case.case_id == case_id) + in_memory = None + redis = None + try: + in_memory = await create_in_memory_backend(identity) + redis = await create_redis_backend(os.environ[REDIS_ENVIRONMENT_VARIABLE], identity) + expected = await ReplayRunner(in_memory).run(replay_case) + actual = await ReplayRunner(redis).run(replay_case) + assert [diff for diff in compare_snapshots(expected, actual) if not diff.allowed] == [] + assert validate_snapshot(replay_case.expected, actual) == [] + finally: + await asyncio.gather(*(backend.close() for backend in (in_memory, redis) if backend is not None)) diff --git a/tests/sessions/test_replay_real_agent.py b/tests/sessions/test_replay_real_agent.py new file mode 100644 index 000000000..4e2356661 --- /dev/null +++ b/tests/sessions/test_replay_real_agent.py @@ -0,0 +1,272 @@ +# +# 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 the Apache License Version 2.0. +# +"""Black-box replay validation with a real tool-calling agent.""" + +from __future__ import annotations + +import asyncio +import os +from dataclasses import asdict +from pathlib import Path +from typing import Any + +import pytest +from trpc_agent_sdk.agents import LlmAgent +from trpc_agent_sdk.models import OpenAIModel +from trpc_agent_sdk.runners import Runner +from trpc_agent_sdk.sessions import InMemorySessionService +from trpc_agent_sdk.tools import FunctionTool +from trpc_agent_sdk.types import Content +from trpc_agent_sdk.types import Part + +from tests.sessions.replay_cases import ExpectedOutcome +from tests.sessions.replay_cases import OperationKind +from tests.sessions.replay_cases import REPLAY_CASES +from tests.sessions.replay_cases import ReplayCase +from tests.sessions.replay_cases import ReplayOperation +from tests.sessions.replay_harness import ReplayRunner +from tests.sessions.replay_harness import create_in_memory_backend +from tests.sessions.replay_harness import create_sqlite_backend +from tests.sessions.replay_report import compare_snapshots +from tests.sessions.replay_report import mutate_snapshot +from tests.sessions.replay_report import write_report + +API_KEY_ENV = "TRPC_AGENT_API_KEY" +BASE_URL_ENV = "TRPC_AGENT_BASE_URL" +MODEL_NAME_ENV = "TRPC_AGENT_MODEL_NAME" +REPORT_PATH_ENV = "TRPC_REAL_REPLAY_REPORT_PATH" +REAL_AGENT_TIMEOUT_SECONDS = 60 +REAL_CASE_ID = "real_agent_tool_memory" +MEMORY_QUERY = "Chinese" +SOURCE_APP_NAME = "real-replay-source" +SOURCE_USER_ID = "real-replay-user" +SOURCE_SESSION_ID = "real-replay-session" + + +def lookup_user_preference(user_id: str) -> dict[str, str]: + """Return a deterministic business fact through a real tool call.""" + return { + "user_id": user_id, + "preferred_language": "Chinese", + "notification_channel": "email", + } + + +def _model_config() -> tuple[str, str, str]: + values = tuple(os.getenv(name, "") for name in (API_KEY_ENV, BASE_URL_ENV, MODEL_NAME_ENV)) + if not all(values): + pytest.skip(f"set {API_KEY_ENV}, {BASE_URL_ENV}, and {MODEL_NAME_ENV} to run the real-model replay") + return values + + +async def _capture_real_case() -> ReplayCase: + api_key, base_url, model_name = _model_config() + model = OpenAIModel(model_name=model_name, api_key=api_key, base_url=base_url) + agent = LlmAgent( + name="preference_agent", + model=model, + instruction=("Always call lookup_user_preference before answering. " + "Report the returned preferred language and notification channel exactly."), + tools=[FunctionTool(lookup_user_preference)], + ) + service = InMemorySessionService() + runner = Runner(app_name=SOURCE_APP_NAME, agent=agent, session_service=service) + await service.create_session( + app_name=SOURCE_APP_NAME, + user_id=SOURCE_USER_ID, + session_id=SOURCE_SESSION_ID, + ) + try: + await asyncio.wait_for(_consume_agent(runner), REAL_AGENT_TIMEOUT_SECONDS) + session = await service.get_session( + app_name=SOURCE_APP_NAME, + user_id=SOURCE_USER_ID, + session_id=SOURCE_SESSION_ID, + ) + if session is None: + raise AssertionError("real agent did not persist its session") + return _case_from_events(session.events) + finally: + await runner.close() + + +async def _consume_agent(runner: Runner) -> None: + message = Content( + parts=[Part.from_text(text="Look up the preferences for user replay-user, then answer with both values.", )]) + async for _ in runner.run_async( + user_id=SOURCE_USER_ID, + session_id=SOURCE_SESSION_ID, + new_message=message, + ): + pass + + +def _case_from_events(events: list[Any]) -> ReplayCase: + operations = [ReplayOperation(OperationKind.CREATE)] + event_ids: list[str] = [] + for event in events: + for part in event.content.parts if event.content else []: + operation = _part_operation(part, event.author, len(event_ids)) + if operation is not None: + operations.append(operation) + event_ids.append(str(operation.payload["event_id"])) + operations.extend(( + ReplayOperation(OperationKind.STORE_MEMORY), + ReplayOperation(OperationKind.SEARCH_MEMORY, {"query": MEMORY_QUERY}), + )) + part_types = {operation.payload.get("part_type") for operation in operations} + if not {"function_call", "function_response"}.issubset(part_types): + raise AssertionError("real model did not complete the required tool round trip") + return ReplayCase( + case_id=REAL_CASE_ID, + operations=tuple(operations), + expected=ExpectedOutcome(tuple(event_ids)), + ) + + +def _part_operation(part: Any, author: str, index: int) -> ReplayOperation | None: + payload: dict[str, Any] = { + "event_id": f"real-event-{index}", + "author": author, + } + if part.function_call: + payload.update(part_type="function_call", value=part.function_call.model_dump(exclude_none=True)) + elif part.function_response: + payload.update(part_type="function_response", value=part.function_response.model_dump(exclude_none=True)) + elif part.text and not part.thought: + payload["text"] = part.text + else: + return None + return ReplayOperation(OperationKind.APPEND, payload) + + +async def _replay_pair(replay_case: ReplayCase, db_path: Path) -> tuple[dict[str, Any], dict[str, Any]]: + in_memory = await create_in_memory_backend() + sqlite = await create_sqlite_backend(db_path) + try: + left, right = await asyncio.gather( + ReplayRunner(in_memory).run(replay_case), + ReplayRunner(sqlite).run(replay_case), + ) + return left, right + finally: + await asyncio.gather(in_memory.close(), sqlite.close()) + + +def _corrupt_first_string(value: Any) -> bool: + if isinstance(value, dict): + for key, child in value.items(): + if isinstance(child, str) and child: + value[key] = f"{child[:-1]}X" + return True + if _corrupt_first_string(child): + return True + elif isinstance(value, list): + return any(_corrupt_first_string(child) for child in value) + return False + + +def _corrupt_memory(snapshot: dict[str, Any]) -> None: + if not _corrupt_first_string(snapshot["memory"][MEMORY_QUERY]["memories"]): + raise AssertionError("no memory text was available to corrupt") + + +def _corrupt_tool_response(snapshot: dict[str, Any]) -> None: + for event in snapshot["events"]: + response = event["content"]["parts"][0].get("function_response") + if not response: + continue + payload = response.get("response", {}) + if isinstance(payload.get("preferred_language"), str): + payload["preferred_language"] = f"{payload['preferred_language'][:-1]}X" + return + if isinstance(payload.get("temperature"), int): + payload["temperature"] += 1 + return + if _corrupt_first_string(payload): + return + raise AssertionError("no function response was available to corrupt") + + +def _tool_memory_case() -> ReplayCase: + source = next(case for case in REPLAY_CASES if case.case_id == "tool_round_trip") + operations = source.operations + ( + ReplayOperation( + OperationKind.APPEND, + { + "event_id": "tool-memory-fact", + "author": "agent", + "text": "The user's preferred language is Chinese.", + }, + ), + ReplayOperation(OperationKind.STORE_MEMORY), + ReplayOperation(OperationKind.SEARCH_MEMORY, {"query": MEMORY_QUERY}), + ) + expected = ExpectedOutcome(source.expected.event_ids + ("tool-memory-fact", )) + return ReplayCase("tool_memory_semantic_drift", operations, expected) + + +def _write_real_report(replay_case: ReplayCase, clean_diffs: list[Any], injected: dict[str, list[Any]]) -> None: + report_path = os.getenv(REPORT_PATH_ENV) + if not report_path: + return + tool_operations = [ + operation.payload["part_type"] for operation in replay_case.operations + if operation.payload.get("part_type") in {"function_call", "function_response"} + ] + report = { + "schema_version": 1, + "case_id": replay_case.case_id, + "model": os.environ[MODEL_NAME_ENV], + "tool_round_trip": tool_operations == ["function_call", "function_response"], + "clean_comparison": { + "unexpected_diff_count": len(clean_diffs), + "diffs": [asdict(diff) for diff in clean_diffs], + }, + "injected_anomalies": { + name: { + "detected": bool(diffs), + "unexpected_diff_count": len(diffs), + "diffs": [asdict(diff) for diff in diffs], + } + for name, diffs in injected.items() + }, + } + write_report(report, Path(report_path)) + + +async def test_subtle_realistic_drift_is_detected(tmp_path: Path) -> None: + left, right = await _replay_pair(_tool_memory_case(), tmp_path / "mock-drift.db") + assert compare_snapshots(left, right) == [] + + tool_diffs = compare_snapshots(left, mutate_snapshot(right, _corrupt_tool_response)) + assert any(diff.category == "events" and "function_response" in diff.field_path for diff in tool_diffs) + + memory_diffs = compare_snapshots(left, mutate_snapshot(right, _corrupt_memory)) + assert any(diff.category == "memory" and diff.left != diff.right for diff in memory_diffs) + assert all(not diff.allowed for diff in tool_diffs + memory_diffs) + + +async def test_real_agent_tool_trace_replays_consistently(tmp_path: Path) -> None: + replay_case = await _capture_real_case() + left, right = await _replay_pair(replay_case, tmp_path / "real-agent.db") + clean_diffs = compare_snapshots(left, right) + assert clean_diffs == [] + + tool_diffs = compare_snapshots(left, mutate_snapshot(right, _corrupt_tool_response)) + memory_diffs = compare_snapshots(left, mutate_snapshot(right, _corrupt_memory)) + assert any(diff.category == "events" and not diff.allowed for diff in tool_diffs) + assert any(diff.category == "memory" and not diff.allowed for diff in memory_diffs) + _write_real_report( + replay_case, + clean_diffs, + { + "tool_response_single_character_drift": tool_diffs, + "memory_single_character_drift": memory_diffs, + }, + )