From c83f12797d845736c55978585dee601f01c36c61 Mon Sep 17 00:00:00 2001 From: Cai Zichen <3234912331@qq.com> Date: Sat, 11 Jul 2026 19:13:27 +0800 Subject: [PATCH 1/4] test(sessions): add replay consistency harness Add reusable replay cases and backend adapters for session, memory, and summary consistency checks across InMemory, SQLite, and optional Redis. Cover cross-session memory, mid-replay restart, and state namespace round-trips, and fix summary lineage persistence plus Redis app/user state compatibility uncovered by the new suite. --- tests/sessions/replay_acceptance_design.md | 28 + tests/sessions/replay_cases.py | 611 ++++++++ tests/sessions/replay_harness.py | 1258 +++++++++++++++++ tests/sessions/replay_models.py | 262 ++++ tests/sessions/replay_summary.py | 55 + tests/sessions/test_replay_consistency.py | 200 +++ .../sessions/_in_memory_session_service.py | 4 + .../sessions/_redis_session_service.py | 38 +- .../sessions/_sql_session_service.py | 31 +- .../sessions/_summarizer_manager.py | 136 +- 10 files changed, 2585 insertions(+), 38 deletions(-) create mode 100644 tests/sessions/replay_acceptance_design.md create mode 100644 tests/sessions/replay_cases.py create mode 100644 tests/sessions/replay_harness.py create mode 100644 tests/sessions/replay_models.py create mode 100644 tests/sessions/replay_summary.py create mode 100644 tests/sessions/test_replay_consistency.py diff --git a/tests/sessions/replay_acceptance_design.md b/tests/sessions/replay_acceptance_design.md new file mode 100644 index 00000000..e649a19f --- /dev/null +++ b/tests/sessions/replay_acceptance_design.md @@ -0,0 +1,28 @@ +# Replay Consistency Design Note + +## 设计说明 +本框架以统一 `ReplayCase` 协议驱动 `InMemory`、文件型 `SQLite` 与可选 `Redis` 后端，先生成 `session/state/memory/summary` 四类快照，再做归一化比较。归一化只移除非业务噪声，不掩盖事件顺序、state 最终值、memory 检索结果及 summary 归属与覆盖关系；其中普通时间戳做精度收敛，`summary_timestamp` 使用 case 级确定性时钟参与比较。summary 比较分为两层：一层比较摘要文本与压缩后的事件窗口，另一层比较 `summary_id/version/replaces/session_id` 等 lineage 元数据。为验证保存/读取语义，summary 元数据会随 summary event 一起持久化，`SQLite/Redis` 适配器在抓取快照前执行一次关服务后重开读回，并支持在 case 中显式插入中途重启步骤验证“重启后继续写”。协议还支持同一 case 内用 `session_alias` 驱动多个 session，以覆盖跨 session 的 memory 聚合和 `app:/user:/temp:` 状态可见性差异。负例分为 snapshot mutation 与 runtime fault 两类，前者验证比较器精度，后者验证重复写入、中途失败和运行时污染场景。轻量模式默认运行 `InMemory + SQLite`，集成模式通过环境变量启用 `Redis`。 + +## 官方 10 条验收 Case +| Case ID | 场景 | 验收点 | +| --- | --- | --- | +| `single_turn_text` | 单轮普通对话 | 基础事件一致性 | +| `multi_turn_dialogue` | 多轮对话 | 事件顺序与多轮回放 | +| `tool_call_and_response` | 工具调用对话 | `function_call/response` 一致性 | +| `state_and_memory_roundtrip` | 多次 state 更新覆盖 + memory 检索 | state 覆盖语义与 memory 一致性 | +| `summary_compaction_with_history` | summary 生成与历史事件压缩 | summary 与 historical events | +| `summary_version_rolls_forward` | summary 更新 | version / replaces 递增 | +| `summary_binding_mismatch_injection` | summary 归属错误 | `summary.session_id` 检出 | +| `summary_missing_injection` | summary 丢失 | `summary` 缺失检出 | +| `runtime_summary_overwrite_fault` | summary 覆盖错误 | `summary.replaces` 检出 | +| `partial_failure_event_loss_fault` | 写入中途失败 | 事件窗口异常检出 | + +## 扩展 Case +- `state_corruption_injection`：补充 state 字段级污染检测。 +- `summary_lineage_corruption_injection`：补充 snapshot 层 lineage 篡改。 +- `duplicate_event_runtime_fault`：补充重复写入异常。 +- `runtime_state_corruption_fault`：补充运行时 state 污染。 +- `runtime_summary_loss_fault`：补充运行时 summary 丢失。 +- `cross_session_memory_aggregation`：补充同一 app/user 下跨 session 的 memory 聚合语义。 +- `restart_mid_replay_after_summary`：补充 summary 持久化后中途重启再续写的恢复语义。 +- `state_namespace_roundtrip`：补充 `app:/user:/temp:` 状态命名空间在跨 session 和重启后的可见性语义。 diff --git a/tests/sessions/replay_cases.py b/tests/sessions/replay_cases.py new file mode 100644 index 00000000..901d4604 --- /dev/null +++ b/tests/sessions/replay_cases.py @@ -0,0 +1,611 @@ +"""Replay cases for acceptance and extended consistency tests.""" + +from __future__ import annotations + +from .replay_models import EventSpec +from .replay_models import FunctionCallSpec +from .replay_models import FunctionResponseSpec +from .replay_models import ReplayCase +from .replay_models import ReplayStep +from .replay_models import RuntimeFault +from .replay_models import RuntimeFaultOperation +from .replay_models import SnapshotMutation +from .replay_models import SnapshotMutationOperation + + +_PERSISTENT_BACKEND = "persistent" + + +_BASELINE_CASES: tuple[ReplayCase, ...] = ( + ReplayCase( + case_id="single_turn_text", + description="One user turn followed by one assistant text response.", + session_id="replay-single-turn", + steps=( + ReplayStep.create_session(), + ReplayStep.append_event( + EventSpec(author="user", role="user", text="Hello, what can you do?"), + ), + ReplayStep.append_event( + EventSpec(author="assistant", role="model", text="I can help answer questions."), + ), + ), + ), + ReplayCase( + case_id="multi_turn_dialogue", + description="Multiple user and assistant turns should preserve event ordering across backends.", + session_id="replay-multi-turn", + steps=( + ReplayStep.create_session(), + ReplayStep.append_event( + EventSpec(author="user", role="user", text="Hello assistant."), + ), + ReplayStep.append_event( + EventSpec(author="assistant", role="model", text="Hello, how can I help?"), + ), + ReplayStep.append_event( + EventSpec(author="user", role="user", text="Please remember my travel plan."), + ), + ReplayStep.append_event( + EventSpec(author="assistant", role="model", text="I will remember your travel plan."), + ), + ), + ), + ReplayCase( + case_id="tool_call_and_response", + description="Assistant tool call followed by tool response.", + session_id="replay-tool-call", + steps=( + ReplayStep.create_session(), + ReplayStep.append_event( + EventSpec(author="user", role="user", text="Check the Beijing weather."), + ), + ReplayStep.append_event( + EventSpec( + author="assistant", + role="model", + function_calls=( + FunctionCallSpec( + name="get_weather", + args={"city": "Beijing"}, + call_id="call-weather-1", + ), + ), + ), + ), + ReplayStep.append_event( + EventSpec( + author="assistant", + role="user", + function_responses=( + FunctionResponseSpec( + name="get_weather", + response={"temperature": "25C", "condition": "Sunny"}, + call_id="call-weather-1", + ), + ), + ), + ), + ), + ), + ReplayCase( + case_id="state_and_memory_roundtrip", + description="Repeated session state updates should preserve overwrite semantics before memory persistence.", + session_id="replay-memory-state", + steps=( + ReplayStep.create_session(initial_state={"user_name": "alice"}), + ReplayStep.append_event( + EventSpec(author="user", role="user", text="Please remember that I prefer tea."), + ), + ReplayStep.append_event( + EventSpec( + author="assistant", + role="model", + text="Noted. You prefer tea over coffee.", + state_delta={"preference": "tea"}, + ), + ), + ReplayStep.append_event( + EventSpec(author="user", role="user", text="Actually update that to green tea."), + ), + ReplayStep.append_event( + EventSpec( + author="assistant", + role="model", + text="Updated. You now prefer green tea.", + state_delta={"preference": "green tea", "drink_temperature": "hot"}, + ), + ), + ReplayStep.store_memory(), + ReplayStep.search_memory(name="preference_search", query="green tea", limit=5), + ), + ), + ReplayCase( + case_id="summary_compaction_with_history", + description="Deterministic summary compaction keeps recent events and stores historical events.", + session_id="replay-summary-history", + enable_summary=True, + summary_keep_recent_count=2, + store_historical_events=True, + steps=( + ReplayStep.create_session(), + ReplayStep.append_event( + EventSpec(author="user", role="user", text="I am planning a weekend trip."), + ), + ReplayStep.append_event( + EventSpec(author="assistant", role="model", text="Great, where would you like to go?"), + ), + ReplayStep.append_event( + EventSpec(author="user", role="user", text="I want to visit Hangzhou."), + ), + ReplayStep.append_event( + EventSpec(author="assistant", role="model", text="Hangzhou is known for West Lake."), + ), + ReplayStep.create_summary(force=True), + ), + ), + ReplayCase( + case_id="summary_version_rolls_forward", + description="A second summary should replace the first one with incremented lineage metadata.", + session_id="replay-summary-version", + enable_summary=True, + summary_keep_recent_count=2, + store_historical_events=True, + steps=( + ReplayStep.create_session(), + ReplayStep.append_event( + EventSpec(author="user", role="user", text="Remember my project is called Atlas."), + ), + ReplayStep.append_event( + EventSpec(author="assistant", role="model", text="Got it, your project is Atlas."), + ), + ReplayStep.append_event( + EventSpec(author="user", role="user", text="It uses Redis and SQL backends."), + ), + ReplayStep.append_event( + EventSpec(author="assistant", role="model", text="Atlas uses Redis and SQL backends."), + ), + ReplayStep.create_summary(force=True), + ReplayStep.append_event( + EventSpec(author="user", role="user", text="Also note that replay consistency is critical."), + ), + ReplayStep.append_event( + EventSpec(author="assistant", role="model", text="I will keep replay consistency in mind."), + ), + ReplayStep.create_summary(force=True), + ), + ), +) + + +_NEGATIVE_CASES: tuple[ReplayCase, ...] = ( + ReplayCase( + case_id="summary_binding_mismatch_injection", + description="Injected summary ownership mismatch must be reported at the exact summary field path.", + session_id="replay-negative-summary-binding", + enable_summary=True, + summary_keep_recent_count=2, + store_historical_events=True, + expected_diff_paths=("summary.session_id",), + snapshot_mutations=( + SnapshotMutation( + backend_name=_PERSISTENT_BACKEND, + path="summary.session_id", + value="wrong-session-id", + ), + ), + steps=( + ReplayStep.create_session(), + ReplayStep.append_event( + EventSpec(author="user", role="user", text="Please summarize my travel plan."), + ), + ReplayStep.append_event( + EventSpec(author="assistant", role="model", text="Sure, tell me the route."), + ), + ReplayStep.append_event( + EventSpec(author="user", role="user", text="Shanghai to Hangzhou by train."), + ), + ReplayStep.append_event( + EventSpec(author="assistant", role="model", text="That route is short and convenient."), + ), + ReplayStep.create_summary(force=True), + ), + ), + ReplayCase( + case_id="summary_missing_injection", + description="Injected summary loss must be detected as a summary-level mismatch.", + session_id="replay-negative-summary-missing", + enable_summary=True, + summary_keep_recent_count=2, + store_historical_events=True, + expected_diff_paths=("summary",), + snapshot_mutations=( + SnapshotMutation( + backend_name=_PERSISTENT_BACKEND, + path="summary", + value=None, + ), + ), + steps=( + ReplayStep.create_session(), + ReplayStep.append_event( + EventSpec(author="user", role="user", text="Track my preferences for black coffee."), + ), + ReplayStep.append_event( + EventSpec(author="assistant", role="model", text="I noted your coffee preference."), + ), + ReplayStep.append_event( + EventSpec(author="user", role="user", text="Also remember I dislike sugary drinks."), + ), + ReplayStep.append_event( + EventSpec(author="assistant", role="model", text="I will avoid sugary drink suggestions."), + ), + ReplayStep.create_summary(force=True), + ), + ), + ReplayCase( + case_id="state_corruption_injection", + description="Injected state corruption must surface at the exact state field path.", + session_id="replay-negative-state-corruption", + expected_diff_paths=("state.preference",), + snapshot_mutations=( + SnapshotMutation( + backend_name=_PERSISTENT_BACKEND, + path="state.preference", + value="coffee", + ), + ), + steps=( + ReplayStep.create_session(initial_state={"user_name": "alice"}), + ReplayStep.append_event( + EventSpec(author="user", role="user", text="Please remember that I prefer tea."), + ), + ReplayStep.append_event( + EventSpec( + author="assistant", + role="model", + text="Noted. You prefer tea over coffee.", + state_delta={"preference": "tea"}, + ), + ), + ), + ), + ReplayCase( + case_id="summary_lineage_corruption_injection", + description="Injected summary lineage corruption must be detected via the replaces field.", + session_id="replay-negative-summary-lineage", + enable_summary=True, + summary_keep_recent_count=2, + store_historical_events=True, + expected_diff_paths=("summary.replaces",), + snapshot_mutations=( + SnapshotMutation( + backend_name=_PERSISTENT_BACKEND, + path="summary.replaces", + value="wrong-summary-id", + ), + ), + steps=( + ReplayStep.create_session(), + ReplayStep.append_event( + EventSpec(author="user", role="user", text="Remember my project codename is Northstar."), + ), + ReplayStep.append_event( + EventSpec(author="assistant", role="model", text="The codename is Northstar."), + ), + ReplayStep.append_event( + EventSpec(author="user", role="user", text="It runs on both SQLite and Redis."), + ), + ReplayStep.append_event( + EventSpec(author="assistant", role="model", text="Northstar runs on SQLite and Redis."), + ), + ReplayStep.create_summary(force=True), + ReplayStep.append_event( + EventSpec(author="user", role="user", text="Replay consistency matters a lot."), + ), + ReplayStep.append_event( + EventSpec(author="assistant", role="model", text="I will keep replay consistency as a priority."), + ), + ReplayStep.create_summary(force=True), + ), + ), + ReplayCase( + case_id="duplicate_event_runtime_fault", + description="A duplicated event injected during replay must be detected as an event-count mismatch.", + session_id="replay-negative-duplicate-event", + expected_diff_paths=("session.events.length",), + runtime_faults=( + RuntimeFault( + backend_name=_PERSISTENT_BACKEND, + after_step=2, + operation=RuntimeFaultOperation.DUPLICATE_LAST_EVENT, + ), + ), + steps=( + ReplayStep.create_session(), + ReplayStep.append_event( + EventSpec(author="user", role="user", text="Please store this reminder."), + ), + ReplayStep.append_event( + EventSpec(author="assistant", role="model", text="I stored the reminder."), + ), + ), + ), + ReplayCase( + case_id="runtime_state_corruption_fault", + description="A runtime state corruption must be detected on the precise state path.", + session_id="replay-negative-runtime-state", + expected_diff_paths=("state.preference",), + runtime_faults=( + RuntimeFault( + backend_name=_PERSISTENT_BACKEND, + after_step=2, + operation=RuntimeFaultOperation.SET_SESSION_VALUE, + path="state.preference", + value="coffee", + ), + ), + steps=( + ReplayStep.create_session(initial_state={"user_name": "alice"}), + ReplayStep.append_event( + EventSpec(author="user", role="user", text="Please remember that I prefer tea."), + ), + ReplayStep.append_event( + EventSpec( + author="assistant", + role="model", + text="Noted. You prefer tea over coffee.", + state_delta={"preference": "tea"}, + ), + ), + ), + ), + ReplayCase( + case_id="runtime_summary_loss_fault", + description="A runtime summary deletion must be detected as a missing summary.", + session_id="replay-negative-runtime-summary-loss", + enable_summary=True, + summary_keep_recent_count=2, + store_historical_events=True, + expected_diff_paths=("summary",), + runtime_faults=( + RuntimeFault( + backend_name=_PERSISTENT_BACKEND, + after_step=5, + operation=RuntimeFaultOperation.DELETE_SUMMARY, + ), + ), + steps=( + ReplayStep.create_session(), + ReplayStep.append_event( + EventSpec(author="user", role="user", text="Please summarize my sprint notes."), + ), + ReplayStep.append_event( + EventSpec(author="assistant", role="model", text="Sure, continue."), + ), + ReplayStep.append_event( + EventSpec(author="user", role="user", text="We fixed replay ordering bugs."), + ), + ReplayStep.append_event( + EventSpec(author="assistant", role="model", text="The replay ordering bugs were fixed."), + ), + ReplayStep.create_summary(force=True), + ), + ), + ReplayCase( + case_id="runtime_summary_overwrite_fault", + description="A runtime summary overwrite must be detected through lineage replacement fields.", + session_id="replay-negative-runtime-summary-overwrite", + enable_summary=True, + summary_keep_recent_count=2, + store_historical_events=True, + expected_diff_paths=( + "summary.replaces", + "summary.metadata.replaces", + ), + runtime_faults=( + RuntimeFault( + backend_name=_PERSISTENT_BACKEND, + after_step=8, + operation=RuntimeFaultOperation.SET_SUMMARY_VALUE, + path="metadata.replaces", + value="wrong-summary-id", + ), + ), + steps=( + ReplayStep.create_session(), + ReplayStep.append_event( + EventSpec(author="user", role="user", text="Remember the codename is Aurora."), + ), + ReplayStep.append_event( + EventSpec(author="assistant", role="model", text="The codename is Aurora."), + ), + ReplayStep.append_event( + EventSpec(author="user", role="user", text="Aurora runs on SQL and Redis."), + ), + ReplayStep.append_event( + EventSpec(author="assistant", role="model", text="Aurora runs on SQL and Redis."), + ), + ReplayStep.create_summary(force=True), + ReplayStep.append_event( + EventSpec(author="user", role="user", text="Replay correctness matters a lot."), + ), + ReplayStep.append_event( + EventSpec(author="assistant", role="model", text="I will prioritize replay correctness."), + ), + ReplayStep.create_summary(force=True), + ), + ), + ReplayCase( + case_id="partial_failure_event_loss_fault", + description="A partial failure that loses the final event but keeps state must be detected as an event-window mismatch.", + session_id="replay-negative-partial-failure", + expected_diff_paths=("session.events.length",), + runtime_faults=( + RuntimeFault( + backend_name=_PERSISTENT_BACKEND, + after_step=2, + operation=RuntimeFaultOperation.DROP_LAST_EVENT_KEEP_STATE, + ), + ), + steps=( + ReplayStep.create_session(initial_state={"user_name": "alice"}), + ReplayStep.append_event( + EventSpec(author="user", role="user", text="Please remember that I prefer tea."), + ), + ReplayStep.append_event( + EventSpec( + author="assistant", + role="model", + text="Noted. You prefer tea over coffee.", + state_delta={"preference": "tea"}, + ), + ), + ), + ), +) + + +_ROBUSTNESS_CASES: tuple[ReplayCase, ...] = ( + ReplayCase( + case_id="cross_session_memory_aggregation", + description="Memory written by one session should remain searchable from another session under the same app/user scope.", + session_id="replay-memory-target", + steps=( + ReplayStep.create_session(session_alias="source", session_id="replay-memory-source"), + ReplayStep.append_event( + EventSpec(author="user", role="user", text="Please remember that I prefer oolong tea."), + session_alias="source", + ), + ReplayStep.append_event( + EventSpec(author="assistant", role="model", text="Noted. You prefer oolong tea."), + session_alias="source", + ), + ReplayStep.store_memory(session_alias="source"), + ReplayStep.create_session(session_alias="default", session_id="replay-memory-target"), + ReplayStep.append_event( + EventSpec(author="user", role="user", text="What drink did I say I prefer?"), + ), + ReplayStep.search_memory( + name="cross_session_preference_search", + query="oolong", + limit=5, + ), + ), + ), + ReplayCase( + case_id="restart_mid_replay_after_summary", + description="Persistent backends should restore summary state correctly after a restart and continue replaying later turns.", + session_id="replay-restart-summary", + enable_summary=True, + summary_keep_recent_count=2, + store_historical_events=True, + steps=( + ReplayStep.create_session(), + ReplayStep.append_event( + EventSpec(author="user", role="user", text="I am planning a Hangzhou trip."), + ), + ReplayStep.append_event( + EventSpec(author="assistant", role="model", text="Great, what should I remember?"), + ), + ReplayStep.append_event( + EventSpec(author="user", role="user", text="Please remember I need a hotel near West Lake."), + ), + ReplayStep.append_event( + EventSpec(author="assistant", role="model", text="I will remember the hotel preference."), + ), + ReplayStep.create_summary(force=True), + ReplayStep.restart_services(), + ReplayStep.append_event( + EventSpec(author="user", role="user", text="Also note that I will arrive next Friday."), + ), + ReplayStep.append_event( + EventSpec(author="assistant", role="model", text="Arrival next Friday is recorded."), + ), + ReplayStep.restart_services(), + ReplayStep.append_event( + EventSpec(author="user", role="user", text="I prefer morning check-in if available."), + ), + ReplayStep.append_event( + EventSpec(author="assistant", role="model", text="I will keep the morning check-in preference."), + ), + ReplayStep.create_summary(force=True), + ), + ), + ReplayCase( + case_id="state_namespace_roundtrip", + description="App, user, session, and temp state should preserve their intended visibility across sessions and restarts.", + session_id="replay-state-namespace-b", + steps=( + ReplayStep.create_session( + session_alias="writer", + session_id="replay-state-namespace-a", + initial_state={ + "app:locale": "zh-CN", + "user:timezone": "Asia/Shanghai", + "temp:request_id": "req-1", + "draft": "first-session", + }, + ), + ReplayStep.append_event( + EventSpec( + author="assistant", + role="model", + text="Updating shared and session state.", + state_delta={ + "app:release": "2026.07", + "user:tone": "concise", + "temp:trace_id": "trace-1", + "draft": "writer-updated", + }, + ), + session_alias="writer", + ), + ReplayStep.restart_services(), + ReplayStep.create_session( + session_alias="default", + session_id="replay-state-namespace-b", + initial_state={ + "temp:request_id": "req-2", + "draft": "second-session", + }, + ), + ReplayStep.append_event( + EventSpec(author="assistant", role="model", text="Second session should inherit shared state only."), + ), + ), + ), +) + + +# The acceptance suite is the public, fixed set of 10 replay cases used to +# demonstrate baseline correctness and inconsistency detection. +REPLAY_ACCEPTANCE_CASES: tuple[ReplayCase, ...] = ( + _BASELINE_CASES[0], # single_turn_text + _BASELINE_CASES[1], # multi_turn_dialogue + _BASELINE_CASES[2], # tool_call_and_response + _BASELINE_CASES[3], # state_and_memory_roundtrip + _BASELINE_CASES[4], # summary_compaction_with_history + _BASELINE_CASES[5], # summary_version_rolls_forward + _NEGATIVE_CASES[0], # summary_binding_mismatch_injection + _NEGATIVE_CASES[1], # summary_missing_injection + _NEGATIVE_CASES[7], # runtime_summary_overwrite_fault + _NEGATIVE_CASES[8], # partial_failure_event_loss_fault +) + + +# Extra cases extend coverage beyond the fixed acceptance set while reusing the +# same harness and reporting pipeline. +REPLAY_EXTRA_CASES: tuple[ReplayCase, ...] = ( + _NEGATIVE_CASES[2], # state_corruption_injection + _NEGATIVE_CASES[3], # summary_lineage_corruption_injection + _NEGATIVE_CASES[4], # duplicate_event_runtime_fault + _NEGATIVE_CASES[5], # runtime_state_corruption_fault + _NEGATIVE_CASES[6], # runtime_summary_loss_fault + _ROBUSTNESS_CASES[0], # cross_session_memory_aggregation + _ROBUSTNESS_CASES[1], # restart_mid_replay_after_summary + _ROBUSTNESS_CASES[2], # state_namespace_roundtrip +) + + +REPLAY_ALL_CASES: tuple[ReplayCase, ...] = REPLAY_ACCEPTANCE_CASES + REPLAY_EXTRA_CASES diff --git a/tests/sessions/replay_harness.py b/tests/sessions/replay_harness.py new file mode 100644 index 00000000..5262d40d --- /dev/null +++ b/tests/sessions/replay_harness.py @@ -0,0 +1,1258 @@ +"""Replay harness for session/memory consistency tests. + +This module provides: +- backend adapters for replaying the same case on multiple backends, +- snapshot extraction helpers, +- normalization helpers for backend-specific noise, +- structured diff generation, +- JSON report writing. + +The first iteration intentionally focuses on deterministic, LLM-free smoke +cases. More complex summary and failure-injection cases can be added on top of +the same protocol without changing the adapter boundary. +""" + +from __future__ import annotations + +from abc import ABC +from abc import abstractmethod +from contextlib import contextmanager +from dataclasses import replace +import hashlib +import os +from pathlib import Path +import json +import re +import tempfile +import time +import uuid +from typing import Any +from typing import Iterator +from typing import Optional +from urllib.parse import urlparse + +from google.genai.types import Content +from google.genai.types import FunctionCall +from google.genai.types import FunctionResponse +from google.genai.types import Part +from trpc_agent_sdk.abc import MemoryServiceConfig +from trpc_agent_sdk.events import Event +from trpc_agent_sdk.memory import BaseMemoryService +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.sessions import BaseSessionService +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 SqlSessionService + +from .replay_models import BackendSnapshot +from .replay_models import DiffEntry +from .replay_models import EventSpec +from .replay_models import FunctionCallSpec +from .replay_models import FunctionResponseSpec +from .replay_models import MemoryQuerySpec +from .replay_models import ReplayCase +from .replay_models import ReplayStep +from .replay_models import ReplayStepKind +from .replay_models import RuntimeFault +from .replay_models import RuntimeFaultOperation +from .replay_models import SnapshotMutation +from .replay_models import SnapshotMutationOperation +from .replay_models import SummarySnapshot +from .replay_summary import build_replay_summarizer_manager + + +DEFAULT_REPORT_PATH = Path(__file__).with_name("session_memory_summary_diff_report.json") +_EVENT_INDEX_RE = re.compile(r"\[(\d+)\]") +_SUMMARY_EVENT_METADATA_KEY = "session_summary" +_CASE_TIME_BASES: dict[str, float] = {} +_BASELINE_BACKEND_NAME = "inmemory" +_PERSISTENT_BACKEND_TARGETS = {"persistent", "secondary", "non_baseline"} +_BASELINE_BACKEND_TARGETS = {"baseline", "primary", "inmemory"} +_CASE_TIME_FUTURE_SKEW_SECONDS = 60.0 +_REPLAY_CLOCK_MODE_ENV = "TRPC_AGENT_REPLAY_CLOCK_MODE" +_REPLAY_FIXED_EPOCH_ENV = "TRPC_AGENT_REPLAY_FIXED_EPOCH" +_REPLAY_CLOCK_MODE_FRESHNESS_SAFE = "freshness_safe" +_REPLAY_CLOCK_MODE_FIXED_SAFE = "fixed_safe" +_DEFAULT_FIXED_SAFE_EPOCH = 4_102_444_800.0 + + +def _make_memory_config() -> MemoryServiceConfig: + config = MemoryServiceConfig(enabled=True) + config.clean_ttl_config() + return config + + +def _make_session_config() -> SessionServiceConfig: + config = SessionServiceConfig() + config.clean_ttl_config() + return config + + +class ReplayBackendAdapter(ABC): + """Shared backend replay logic. + + Concrete adapters only need to build the concrete session and memory + services. The replay protocol itself stays backend-agnostic. + """ + + name: str + + def __init__(self) -> None: + self._session_service: BaseSessionService | None = None + self._memory_service: BaseMemoryService | None = None + self._session: Session | None = None + self._sessions: dict[str, Session] = {} + self._session_ids: dict[str, str] = {} + self._active_session_alias = "default" + self._memory_results: dict[str, Any] = {} + self._case: ReplayCase | None = None + self._event_sequence = 0 + self._event_time_base = 0.0 + + @property + def session_service(self) -> BaseSessionService: + if self._session_service is None: + raise RuntimeError("Session service is not initialized.") + return self._session_service + + @property + def memory_service(self) -> BaseMemoryService: + if self._memory_service is None: + raise RuntimeError("Memory service is not initialized.") + return self._memory_service + + @property + def session(self) -> Session: + if self._session is None: + raise RuntimeError("Replay session has not been created.") + return self._session + + @property + def case(self) -> ReplayCase: + if self._case is None: + raise RuntimeError("Replay case is not initialized.") + return self._case + + async def setup(self, case: ReplayCase) -> None: + """Initialize backend services for one replay run.""" + + self._case = case + self._memory_results = {} + self._session = None + self._sessions = {} + self._session_ids = {} + self._active_session_alias = "default" + self._event_sequence = 0 + self._event_time_base = _deterministic_time_base(case.case_id) + self._session_service = await self._build_session_service() + self._memory_service = await self._build_memory_service() + + async def _close_services(self) -> None: + """Dispose backend services while preserving replay state.""" + if self._memory_service is not None: + await self._memory_service.close() + self._memory_service = None + if self._session_service is not None: + await self._session_service.close() + self._session_service = None + + async def close(self) -> None: + """Dispose backend services.""" + + await self._close_services() + self._session = None + self._sessions = {} + self._session_ids = {} + self._active_session_alias = "default" + self._memory_results = {} + self._case = None + self._event_sequence = 0 + self._event_time_base = 0.0 + + async def run_case(self, case: ReplayCase) -> BackendSnapshot: + """Replay all steps in a case and collect a snapshot.""" + + for step_index, step in enumerate(case.steps): + await self._run_step(case, step) + await self._apply_runtime_faults(step_index) + if self.should_restart_before_snapshot(): + await self._restart_services() + return await self.collect_snapshot(case) + + def should_restart_before_snapshot(self) -> bool: + """Return whether the adapter should validate persisted read-back.""" + + return False + + def get_runtime_metadata(self) -> dict[str, Any]: + return { + "backend_name": self.name, + "storage_kind": self.name, + } + + def get_report_metadata(self) -> dict[str, Any]: + return self.get_runtime_metadata() + + async def _restart_services(self) -> None: + """Restart backend services and rebuild the read snapshot from persistence.""" + + await self._close_services() + self._session_service = await self._build_session_service() + self._memory_service = await self._build_memory_service() + reopened_sessions: dict[str, Session] = {} + for session_alias, session_id in self._session_ids.items(): + reopened_session = await self.session_service.get_session( + app_name=self.case.app_name, + user_id=self.case.user_id, + session_id=session_id, + ) + if reopened_session is not None: + reopened_sessions[session_alias] = reopened_session + self._sessions = reopened_sessions + self._session = self._sessions.get(self._active_session_alias) + if self._session is None and self._sessions: + self._active_session_alias = next(iter(self._sessions)) + self._session = self._sessions[self._active_session_alias] + if self._session is None: + return + + reopened_memory_results: dict[str, Any] = {} + for step in self.case.steps: + if step.kind == ReplayStepKind.SEARCH_MEMORY and step.memory_query is not None: + query_session = self._resolve_session(step.session_alias) + reopened_memory_results[step.memory_query.name] = await self._search_memory_for_session( + query_session, + step.memory_query, + ) + if reopened_memory_results: + self._memory_results = reopened_memory_results + + def should_restart_during_replay(self) -> bool: + """Return whether RESTART_SERVICES steps should perform a real restart.""" + + return False + + async def _run_step(self, case: ReplayCase, step: ReplayStep) -> None: + if step.kind == ReplayStepKind.CREATE_SESSION: + session_id = step.session_id or case.session_id + session = await self.session_service.create_session( + app_name=case.app_name, + user_id=case.user_id, + session_id=session_id, + state=step.initial_state, + ) + self._session = session + self._sessions[step.session_alias] = session + self._session_ids[step.session_alias] = session.id + self._active_session_alias = step.session_alias + return + + if step.kind in {ReplayStepKind.APPEND_EVENT, ReplayStepKind.APPEND_STATE}: + if step.event is None: + raise ValueError(f"Replay step {step.kind} requires an event.") + target_session = self._resolve_session(step.session_alias) + self._event_sequence += 1 + event = build_event( + step.event, + sequence=self._event_sequence, + timestamp=self._event_time_base + (self._event_sequence / 1000.0), + ) + await self.session_service.append_event(target_session, event) + self._session = target_session + return + + if step.kind == ReplayStepKind.STORE_MEMORY: + target_session = self._resolve_session(step.session_alias) + await self.memory_service.store_session(target_session) + self._session = target_session + return + + if step.kind == ReplayStepKind.SEARCH_MEMORY: + if step.memory_query is None: + raise ValueError("SEARCH_MEMORY step requires a query specification.") + target_session = self._resolve_session(step.session_alias) + self._memory_results[step.memory_query.name] = await self._search_memory_for_session( + target_session, + step.memory_query, + ) + self._session = target_session + return + + if step.kind == ReplayStepKind.CREATE_SUMMARY: + target_session = self._resolve_session(step.session_alias) + self._event_sequence += 1 + summary_timestamp = self._event_time_base + (self._event_sequence / 1000.0) + manager = getattr(self.session_service, "summarizer_manager", None) + with _patched_time_time(summary_timestamp): + if manager is not None and step.force_summary: + await manager.create_session_summary(target_session, force=True) + else: + await self.session_service.create_session_summary(target_session) + self._session = target_session + return + + if step.kind == ReplayStepKind.RESTART_SERVICES: + if self.should_restart_during_replay(): + await self._restart_services() + return + + raise ValueError(f"Unsupported replay step kind: {step.kind}") + + async def _search_memory(self, query: MemoryQuerySpec) -> list[dict[str, Any]]: + return await self._search_memory_for_session(self.session, query) + + async def _search_memory_for_session( + self, + session: Session, + query: MemoryQuerySpec, + ) -> list[dict[str, Any]]: + response = await self.memory_service.search_memory( + key=session.save_key, + query=query.query, + limit=query.limit, + ) + return [_memory_entry_to_snapshot(entry) for entry in response.memories] + + async def collect_snapshot(self, case: ReplayCase) -> BackendSnapshot: + """Collect a backend snapshot after replay.""" + + session = self._session + if session is None: + raise RuntimeError(f"Replay case '{case.case_id}' did not produce an active session.") + session = await self.session_service.get_session( + app_name=session.app_name, + user_id=session.user_id, + session_id=session.id, + ) + if session is None: + raise RuntimeError(f"Replay session '{self.session.id}' was not found.") + + summary = await self._get_summary_snapshot(session) + return BackendSnapshot( + backend_name=self.name, + case_id=case.case_id, + app_name=session.app_name, + user_id=session.user_id, + session_id=session.id, + session={ + "conversation_count": session.conversation_count, + "events": [_event_to_snapshot(event) for event in session.events], + "historical_events": [_event_to_snapshot(event) for event in session.historical_events], + }, + state=dict(session.state), + memory=dict(self._memory_results), + summary=summary, + ) + + def _resolve_session(self, session_alias: str) -> Session: + session = self._sessions.get(session_alias) + if session is None: + raise RuntimeError(f"Replay session alias '{session_alias}' has not been created.") + self._active_session_alias = session_alias + self._session = session + return session + + async def _get_summary_snapshot(self, session: Session) -> Optional[SummarySnapshot]: + manager = getattr(self.session_service, "summarizer_manager", None) + if manager is None: + return None + + summary = await manager.get_session_summary(session) + if summary is None: + return None + + return SummarySnapshot( + session_id=summary.session_id, + summary_text=summary.summary_text, + original_event_count=summary.original_event_count, + compressed_event_count=summary.compressed_event_count, + summary_id=(summary.metadata or {}).get("summary_id"), + version=(summary.metadata or {}).get("version"), + replaces=(summary.metadata or {}).get("replaces"), + summarized_event_count=(summary.metadata or {}).get("summarized_event_count"), + summary_timestamp=getattr(summary, "summary_timestamp", None), + metadata=dict(getattr(summary, "metadata", {}) or {}), + ) + + async def _apply_runtime_faults(self, step_index: int) -> None: + for fault in self.case.runtime_faults: + if not _backend_target_matches(fault.backend_name, self.name) or fault.after_step != step_index: + continue + await self._apply_runtime_fault(fault) + + async def _apply_runtime_fault(self, fault: RuntimeFault) -> None: + if fault.operation == RuntimeFaultOperation.DUPLICATE_LAST_EVENT: + if not self.session.events: + raise RuntimeError("Cannot duplicate the last event of an empty session.") + duplicated_event = self.session.events[-1].model_copy(deep=True) + duplicated_event.id = f"{duplicated_event.id}-duplicate" + duplicated_event.invocation_id = f"{duplicated_event.invocation_id}-duplicate" + duplicated_event.timestamp += 0.0001 + self.session.events.append(duplicated_event) + await self.session_service.update_session(self.session) + return + + if fault.operation == RuntimeFaultOperation.DROP_LAST_EVENT_KEEP_STATE: + if not self.session.events: + raise RuntimeError("Cannot drop the last event of an empty session.") + self.session.events.pop() + await self.session_service.update_session(self.session) + return + + if fault.operation == RuntimeFaultOperation.SET_SESSION_VALUE: + if not fault.path: + raise ValueError("SET_SESSION_VALUE fault requires a target path.") + _set_path_value(self.session, fault.path, fault.value) + await self.session_service.update_session(self.session) + return + + manager = getattr(self.session_service, "summarizer_manager", None) + if manager is None: + raise RuntimeError("Runtime summary fault requires a summarizer manager.") + + if fault.operation == RuntimeFaultOperation.DELETE_SUMMARY: + summary_cache = getattr(manager, "_summarizer_cache", {}) + summary_cache.get(self.session.app_name, {}).get(self.session.user_id, {}).pop(self.session.id, None) + summary_event = _get_summary_event(self.session) + if summary_event is not None: + custom_metadata = dict(summary_event.custom_metadata or {}) + custom_metadata.pop(_SUMMARY_EVENT_METADATA_KEY, None) + summary_event.custom_metadata = custom_metadata or None + await self.session_service.update_session(self.session) + return + + if fault.operation == RuntimeFaultOperation.SET_SUMMARY_VALUE: + if not fault.path: + raise ValueError("SET_SUMMARY_VALUE fault requires a target path.") + summary = await manager.get_session_summary(self.session) + if summary is None: + raise RuntimeError("Cannot mutate summary value because no cached summary exists.") + _set_path_value(summary, fault.path, fault.value) + summary_event = _get_summary_event(self.session) + if summary_event is not None and summary_event.custom_metadata: + persisted_summary = summary_event.custom_metadata.get(_SUMMARY_EVENT_METADATA_KEY) + if isinstance(persisted_summary, dict): + _set_path_value(persisted_summary, fault.path, fault.value) + await self.session_service.update_session(self.session) + return + + raise ValueError(f"Unsupported runtime fault operation: {fault.operation}") + + @abstractmethod + async def _build_session_service(self) -> BaseSessionService: + """Create the session service used by this adapter.""" + + @abstractmethod + async def _build_memory_service(self) -> BaseMemoryService: + """Create the memory service used by this adapter.""" + + +class InMemoryReplayAdapter(ReplayBackendAdapter): + """Replay adapter backed by in-memory session and memory services.""" + + name = "inmemory" + + async def _build_session_service(self) -> BaseSessionService: + config = _make_session_config() + config.store_historical_events = self.case.store_historical_events + summarizer_manager = None + if self.case.enable_summary: + summarizer_manager = build_replay_summarizer_manager( + keep_recent_count=self.case.summary_keep_recent_count, + ) + return InMemorySessionService( + summarizer_manager=summarizer_manager, + session_config=config, + ) + + async def _build_memory_service(self) -> BaseMemoryService: + return InMemoryMemoryService(memory_service_config=_make_memory_config()) + + +class SqliteReplayAdapter(ReplayBackendAdapter): + """Replay adapter backed by SQLite-based session and memory services.""" + + name = "sqlite" + + def __init__(self) -> None: + super().__init__() + self._temp_dir = tempfile.TemporaryDirectory(prefix="trpc-replay-sqlite-") + self._session_db_url = self._sqlite_url("sessions.sqlite") + self._memory_db_url = self._sqlite_url("memory.sqlite") + + def _sqlite_url(self, filename: str) -> str: + return f"sqlite:///{(Path(self._temp_dir.name) / filename).as_posix()}" + + def should_restart_before_snapshot(self) -> bool: + return True + + def should_restart_during_replay(self) -> bool: + return True + + async def _build_session_service(self) -> BaseSessionService: + config = _make_session_config() + config.store_historical_events = self.case.store_historical_events + summarizer_manager = None + if self.case.enable_summary: + summarizer_manager = build_replay_summarizer_manager( + keep_recent_count=self.case.summary_keep_recent_count, + ) + service = SqlSessionService( + db_url=self._session_db_url, + summarizer_manager=summarizer_manager, + is_async=False, + session_config=config, + ) + await service._sql_storage.create_sql_engine() + return service + + async def _build_memory_service(self) -> BaseMemoryService: + service = SqlMemoryService( + db_url=self._memory_db_url, + is_async=False, + memory_service_config=_make_memory_config(), + ) + await service._sql_storage.create_sql_engine() + return service + + async def close(self) -> None: + await super().close() + self._temp_dir.cleanup() + + def get_runtime_metadata(self) -> dict[str, Any]: + return { + "backend_name": self.name, + "storage_kind": "sqlite", + "connection": { + "scheme": "sqlite", + "session_db": Path(urlparse(self._session_db_url).path).name, + "memory_db": Path(urlparse(self._memory_db_url).path).name, + }, + } + + +class RedisReplayAdapter(ReplayBackendAdapter): + """Replay adapter backed by Redis session and memory services.""" + + name = "redis" + + def __init__(self, redis_url: str) -> None: + super().__init__() + self._redis_url = redis_url + self._logical_case: ReplayCase | None = None + self._storage_namespace = f"replay-{uuid.uuid4().hex[:8]}" + + @property + def logical_case(self) -> ReplayCase: + if self._logical_case is None: + raise RuntimeError("Logical replay case is not initialized.") + return self._logical_case + + async def setup(self, case: ReplayCase) -> None: + self._logical_case = case + await super().setup(self._namespaced_case(case)) + + def should_restart_before_snapshot(self) -> bool: + return True + + def should_restart_during_replay(self) -> bool: + return True + + async def run_case(self, case: ReplayCase) -> BackendSnapshot: + snapshot = await super().run_case(self.case) + summary = snapshot.summary + if summary is not None: + summary = replace( + summary, + session_id=self.logical_case.session_id, + summary_id=self._project_identifier(summary.summary_id), + replaces=self._project_identifier(summary.replaces), + metadata=self._project_scalar(summary.metadata), + ) + return replace( + snapshot, + app_name=self.logical_case.app_name, + user_id=self.logical_case.user_id, + session_id=self.logical_case.session_id, + summary=summary, + ) + + async def _build_session_service(self) -> BaseSessionService: + config = _make_session_config() + config.store_historical_events = self.case.store_historical_events + summarizer_manager = None + if self.case.enable_summary: + summarizer_manager = build_replay_summarizer_manager( + keep_recent_count=self.case.summary_keep_recent_count, + ) + return RedisSessionService( + db_url=self._redis_url, + summarizer_manager=summarizer_manager, + is_async=False, + session_config=config, + ) + + async def _build_memory_service(self) -> BaseMemoryService: + return RedisMemoryService( + db_url=self._redis_url, + enabled=True, + is_async=False, + memory_service_config=_make_memory_config(), + ) + + async def close(self) -> None: + await super().close() + self._logical_case = None + + def _namespaced_case(self, case: ReplayCase) -> ReplayCase: + namespaced_steps = tuple( + replace( + step, + session_id=(f"{step.session_id}:{self._storage_namespace}" if step.session_id is not None else None), + ) + for step in case.steps + ) + return replace( + case, + app_name=f"{case.app_name}:{self._storage_namespace}", + user_id=f"{case.user_id}:{self._storage_namespace}", + session_id=f"{case.session_id}:{self._storage_namespace}", + steps=namespaced_steps, + ) + + def _project_identifier(self, value: Optional[str]) -> Optional[str]: + if value is None: + return None + return value.replace(self.case.session_id, self.logical_case.session_id) + + def _project_scalar(self, value: Any) -> Any: + if isinstance(value, dict): + return {key: self._project_scalar(item) for key, item in value.items()} + if isinstance(value, list): + return [self._project_scalar(item) for item in value] + if isinstance(value, tuple): + return [self._project_scalar(item) for item in value] + if isinstance(value, str): + return self._project_identifier(value) + return value + + def get_runtime_metadata(self) -> dict[str, Any]: + return { + "backend_name": self.name, + "storage_kind": "redis", + "connection": _summarize_connection_url(self._redis_url), + "storage_namespace": self._storage_namespace, + "storage_identity": { + "app_name": self.case.app_name, + "user_id": self.case.user_id, + "session_id": self.case.session_id, + }, + "logical_identity": { + "app_name": self.logical_case.app_name, + "user_id": self.logical_case.user_id, + "session_id": self.logical_case.session_id, + }, + } + + def get_report_metadata(self) -> dict[str, Any]: + return { + "backend_name": self.name, + "storage_kind": "redis", + "connection": _summarize_connection_url(self._redis_url), + "namespace_strategy": "per_case_random", + } + + +def build_event(spec: EventSpec, *, sequence: int, timestamp: float) -> Event: + """Materialize an Event from an EventSpec.""" + + parts: list[Part] = [] + if spec.text is not None: + parts.append(Part.from_text(text=spec.text)) + + for function_call in spec.function_calls: + parts.append(Part(function_call=_build_function_call(function_call))) + + for function_response in spec.function_responses: + parts.append(Part(function_response=_build_function_response(function_response))) + + content: Content | None = None + if parts: + content = Content(role=spec.role, parts=parts) + + event = Event( + id=spec.event_id or f"replay-event-{sequence}", + invocation_id=f"replay-invocation-{sequence}", + author=spec.author, + branch=spec.branch, + content=content, + visible=spec.visible, + partial=spec.partial, + timestamp=timestamp, + ) + if spec.state_delta: + event.actions.state_delta = dict(spec.state_delta) + if spec.is_summary_event: + event.set_summary_event(True) + return event + + +def _build_function_call(spec: FunctionCallSpec) -> FunctionCall: + return FunctionCall(name=spec.name, args=dict(spec.args), id=spec.call_id) + + +def _build_function_response(spec: FunctionResponseSpec) -> FunctionResponse: + return FunctionResponse(name=spec.name, response=dict(spec.response), id=spec.call_id) + + +def _event_to_snapshot(event: Event) -> dict[str, Any]: + return { + "author": event.author, + "branch": event.branch, + "visible": event.visible, + "partial": event.partial, + "error_code": event.error_code, + "error_message": event.error_message, + "text": event.get_text(), + "is_summary_event": event.is_summary_event(), + "model_visible": event.is_model_visible(), + "state_delta": dict(event.actions.state_delta or {}), + "function_calls": [ + { + "name": call.name, + "args": _normalize_scalar(call.args), + } + for call in event.get_function_calls() + ], + "function_responses": [ + { + "name": response.name, + "response": _normalize_scalar(response.response), + } + for response in event.get_function_responses() + ], + } + + +def _memory_entry_to_snapshot(entry: Any) -> dict[str, Any]: + parts = getattr(getattr(entry, "content", None), "parts", None) or [] + text = "".join(part.text for part in parts if getattr(part, "text", None)) + role = getattr(getattr(entry, "content", None), "role", None) + return { + "author": getattr(entry, "author", None), + "role": role, + "text": text, + "timestamp": getattr(entry, "timestamp", None), + } + + +def normalize_backend_snapshot(snapshot: BackendSnapshot) -> dict[str, Any]: + """Normalize backend noise before diffing.""" + + return { + "backend_name": snapshot.backend_name, + "case_id": snapshot.case_id, + "app_name": snapshot.app_name, + "user_id": snapshot.user_id, + "session_id": snapshot.session_id, + "session": _normalize_scalar(snapshot.session), + "state": _normalize_scalar(snapshot.state), + "memory": _normalize_memory(snapshot.memory), + "summary": _normalize_summary(snapshot.summary), + } + + +def _normalize_summary(summary: Optional[SummarySnapshot]) -> Optional[dict[str, Any]]: + if summary is None: + return None + return { + "session_id": summary.session_id, + "summary_text": summary.summary_text.strip(), + "original_event_count": summary.original_event_count, + "compressed_event_count": summary.compressed_event_count, + "summary_id": summary.summary_id, + "version": summary.version, + "replaces": summary.replaces, + "summarized_event_count": summary.summarized_event_count, + "summary_timestamp": _normalize_timestamp(summary.summary_timestamp), + "metadata": _normalize_scalar(summary.metadata), + } + + +def _normalize_memory(memory_results: dict[str, Any]) -> dict[str, Any]: + normalized: dict[str, Any] = {} + for query_name, entries in memory_results.items(): + normalized_entries = [] + for entry in entries: + normalized_entries.append( + { + "author": entry.get("author"), + "role": entry.get("role"), + "text": (entry.get("text") or "").strip(), + }) + normalized[query_name] = sorted( + normalized_entries, + key=lambda item: (item["text"], item["author"] or "", item["role"] or ""), + ) + return normalized + + +def _normalize_scalar(value: Any) -> Any: + if isinstance(value, dict): + return {key: _normalize_scalar(value[key]) for key in sorted(value)} + if isinstance(value, list): + return [_normalize_scalar(item) for item in value] + if isinstance(value, tuple): + return [_normalize_scalar(item) for item in value] + if isinstance(value, str): + return value.strip() + return value + + +def _normalize_timestamp(value: Optional[float]) -> Optional[float]: + if value is None: + return None + return round(float(value), 6) + + +def get_replay_clock_metadata() -> dict[str, Any]: + mode = (os.getenv(_REPLAY_CLOCK_MODE_ENV) or _REPLAY_CLOCK_MODE_FRESHNESS_SAFE).strip().lower() + if mode == _REPLAY_CLOCK_MODE_FIXED_SAFE: + configured_epoch = _parse_optional_float(os.getenv(_REPLAY_FIXED_EPOCH_ENV)) + effective_epoch = configured_epoch if configured_epoch is not None else _DEFAULT_FIXED_SAFE_EPOCH + return { + "mode": _REPLAY_CLOCK_MODE_FIXED_SAFE, + "future_skew_seconds": _CASE_TIME_FUTURE_SKEW_SECONDS, + "configured_fixed_epoch": configured_epoch, + "default_fixed_epoch": _DEFAULT_FIXED_SAFE_EPOCH, + "effective_fixed_epoch": effective_epoch, + "freshness_safe": True, + } + return { + "mode": _REPLAY_CLOCK_MODE_FRESHNESS_SAFE, + "future_skew_seconds": _CASE_TIME_FUTURE_SKEW_SECONDS, + "freshness_safe": True, + } + + +def _parse_optional_float(raw: Optional[str]) -> Optional[float]: + if raw is None or not raw.strip(): + return None + return float(raw.strip()) + + +def _summarize_connection_url(url: str) -> dict[str, Any]: + parsed = urlparse(url) + database = parsed.path.lstrip("/") or None + if database and database.isdigit(): + database = int(database) + return { + "scheme": parsed.scheme, + "host": parsed.hostname, + "port": parsed.port, + "database": database, + "has_auth": bool(parsed.username or parsed.password), + } + + +def _backend_target_matches(target_name: str, backend_name: str) -> bool: + normalized_target = target_name.strip().lower() + normalized_backend = backend_name.strip().lower() + if normalized_target == normalized_backend: + return True + if normalized_target in _PERSISTENT_BACKEND_TARGETS: + return normalized_backend != _BASELINE_BACKEND_NAME + if normalized_target in _BASELINE_BACKEND_TARGETS: + return normalized_backend == _BASELINE_BACKEND_NAME + return False + + +def expected_diff_paths_for_backend_pair( + case: ReplayCase, + *, + backend_a: str, + backend_b: str, +) -> tuple[str, ...]: + if not case.expected_diff_paths: + return () + + for target_name in [mutation.backend_name for mutation in case.snapshot_mutations] + [ + fault.backend_name for fault in case.runtime_faults + ]: + if _backend_target_matches(target_name, backend_a) or _backend_target_matches(target_name, backend_b): + return case.expected_diff_paths + return () + + +def diff_backend_snapshots( + *, + case: ReplayCase, + left: BackendSnapshot, + right: BackendSnapshot, +) -> list[DiffEntry]: + """Generate structured diffs for two snapshots.""" + + left_view = normalize_backend_snapshot(left) + right_view = normalize_backend_snapshot(right) + _apply_snapshot_mutations(left_view, case.snapshot_mutations, left.backend_name) + _apply_snapshot_mutations(right_view, case.snapshot_mutations, right.backend_name) + diffs: list[DiffEntry] = [] + + for scope in ("session", "state", "memory", "summary"): + scope_summary_id = _resolve_summary_id(left_view.get(scope), right_view.get(scope)) if scope == "summary" else None + _diff_values( + case=case, + backend_a=left.backend_name, + backend_b=right.backend_name, + scope=scope, + path=scope, + left=left_view[scope], + right=right_view[scope], + out=diffs, + session_id=_resolve_session_id(left_view, right_view, case.session_id), + summary_id=scope_summary_id, + ) + return diffs + + +def _apply_snapshot_mutations( + snapshot_view: dict[str, Any], + mutations: tuple[SnapshotMutation, ...], + backend_name: str, +) -> None: + for mutation in mutations: + if not _backend_target_matches(mutation.backend_name, backend_name): + continue + _apply_snapshot_mutation(snapshot_view, mutation) + + +def _apply_snapshot_mutation(snapshot_view: dict[str, Any], mutation: SnapshotMutation) -> None: + tokens = _parse_path_tokens(mutation.path) + if not tokens: + raise ValueError(f"Invalid mutation path: {mutation.path}") + + parent, last_token = _resolve_parent(snapshot_view, tokens) + if mutation.operation == SnapshotMutationOperation.SET: + _set_token(parent, last_token, mutation.value) + return + if mutation.operation == SnapshotMutationOperation.DELETE: + _delete_token(parent, last_token) + return + raise ValueError(f"Unsupported snapshot mutation operation: {mutation.operation}") + + +def _parse_path_tokens(path: str) -> list[Any]: + tokens: list[Any] = [] + for part in path.split("."): + if not part: + continue + cursor = part + while cursor: + match = re.match(r"^([^\[]+)(\[(\d+)\])?(.*)$", cursor) + if not match: + raise ValueError(f"Invalid path segment: {cursor}") + key, _, index, rest = match.groups() + if key: + tokens.append(key) + if index is not None: + tokens.append(int(index)) + cursor = rest + return tokens + + +def _resolve_parent(root: dict[str, Any], tokens: list[Any]) -> tuple[Any, Any]: + target = root + for token in tokens[:-1]: + target = _get_token(target, token) + return target, tokens[-1] + + +def _set_token(container: Any, token: Any, value: Any) -> None: + if isinstance(token, int): + container[token] = value + elif isinstance(container, dict): + container[token] = value + else: + setattr(container, token, value) + + +def _delete_token(container: Any, token: Any) -> None: + if isinstance(token, int): + del container[token] + elif isinstance(container, dict): + container.pop(token, None) + else: + delattr(container, token) + + +def _get_token(container: Any, token: Any) -> Any: + if isinstance(token, int): + return container[token] + if isinstance(container, dict): + return container[token] + return getattr(container, token) + + +def _set_path_value(root: Any, path: str, value: Any) -> None: + tokens = _parse_path_tokens(path) + if not tokens: + raise ValueError(f"Invalid mutation path: {path}") + parent, last_token = _resolve_parent(root, tokens) + _set_token(parent, last_token, value) + + +def _diff_values( + *, + case: ReplayCase, + backend_a: str, + backend_b: str, + scope: str, + path: str, + left: Any, + right: Any, + out: list[DiffEntry], + session_id: str, + summary_id: Optional[str], +) -> None: + if type(left) is not type(right): + out.append(_make_diff(case, backend_a, backend_b, scope, path, left, right, session_id, summary_id)) + return + + if isinstance(left, dict): + for key in sorted(set(left) | set(right)): + next_path = f"{path}.{key}" + if key not in left or key not in right: + out.append( + _make_diff( + case, + backend_a, + backend_b, + scope, + next_path, + left.get(key), + right.get(key), + session_id, + summary_id, + )) + continue + _diff_values( + case=case, + backend_a=backend_a, + backend_b=backend_b, + scope=scope, + path=next_path, + left=left[key], + right=right[key], + out=out, + session_id=session_id, + summary_id=summary_id, + ) + return + + if isinstance(left, list): + if len(left) != len(right): + out.append(_make_diff(case, backend_a, backend_b, scope, f"{path}.length", len(left), len(right), + session_id, summary_id)) + return + for index, (left_item, right_item) in enumerate(zip(left, right)): + _diff_values( + case=case, + backend_a=backend_a, + backend_b=backend_b, + scope=scope, + path=f"{path}[{index}]", + left=left_item, + right=right_item, + out=out, + session_id=session_id, + summary_id=summary_id, + ) + return + + if left != right: + out.append(_make_diff(case, backend_a, backend_b, scope, path, left, right, session_id, summary_id)) + + +def _make_diff( + case: ReplayCase, + backend_a: str, + backend_b: str, + scope: str, + path: str, + left: Any, + right: Any, + session_id: str, + summary_id: Optional[str], +) -> DiffEntry: + allowed = path in set(case.allowed_diff_paths) + event_index = _extract_event_index(path) + reason = "Allowed backend-specific difference." if allowed else None + return DiffEntry( + case_id=case.case_id, + backend_a=backend_a, + backend_b=backend_b, + scope=scope, + path=path, + left=left, + right=right, + allowed=allowed, + session_id=session_id, + event_index=event_index, + summary_id=summary_id, + reason=reason, + ) + + +def _resolve_summary_id(left: Any, right: Any) -> Optional[str]: + for value in (left, right): + if isinstance(value, dict): + summary_id = value.get("summary_id") + if summary_id is not None: + return str(summary_id) + return None + + +def _resolve_session_id(left: dict[str, Any], right: dict[str, Any], fallback: str) -> str: + left_session_id = left.get("session_id") + right_session_id = right.get("session_id") + if isinstance(left_session_id, str) and left_session_id == right_session_id: + return left_session_id + return fallback + + +def _deterministic_time_base(case_id: str) -> float: + if case_id not in _CASE_TIME_BASES: + digest = hashlib.sha1(case_id.encode("utf-8")).digest() + offset_millis = int.from_bytes(digest[:6], "big") % 1_000 + clock_metadata = get_replay_clock_metadata() + if clock_metadata["mode"] == _REPLAY_CLOCK_MODE_FIXED_SAFE: + base_epoch = float(clock_metadata["effective_fixed_epoch"]) + else: + base_epoch = round(time.time(), 3) + _CASE_TIME_FUTURE_SKEW_SECONDS + _CASE_TIME_BASES[case_id] = base_epoch + (offset_millis / 1000.0) + return _CASE_TIME_BASES[case_id] + + +@contextmanager +def _patched_time_time(timestamp: float) -> Iterator[None]: + original_time = time.time + time.time = lambda: timestamp + try: + yield + finally: + time.time = original_time + + +def _get_summary_event(session: Session) -> Optional[Event]: + for event in session.events: + if event.is_summary_event(): + return event + return None + + +def _extract_event_index(path: str) -> Optional[int]: + match = _EVENT_INDEX_RE.search(path) + if not match: + return None + return int(match.group(1)) + + +def format_diffs(diffs: list[DiffEntry]) -> str: + """Render human-readable diffs for assertion failures.""" + + if not diffs: + return "No differences detected." + lines = [] + for diff in diffs: + prefix = "ALLOWED" if diff.allowed else "DIFF" + lines.append(f"{prefix} {diff.path}: {diff.left!r} != {diff.right!r}") + return "\n".join(lines) + + +def build_case_report(case: ReplayCase, diffs: list[DiffEntry]) -> dict[str, Any]: + """Build a grouped report for one replay case.""" + + expected_paths = set(case.expected_diff_paths) + allowed_paths = set(case.allowed_diff_paths) + detected_paths = {diff.path for diff in diffs if not diff.allowed} + return { + "case_id": case.case_id, + "description": case.description, + "expects_diffs": bool(expected_paths), + "expected_diff_paths": sorted(expected_paths), + "allowed_diff_paths": sorted(allowed_paths), + "detected_diff_paths": sorted(detected_paths), + "missing_expected_paths": sorted(expected_paths - detected_paths), + "unexpected_diff_paths": sorted(detected_paths - expected_paths), + "diffs": [diff.to_dict() for diff in diffs], + } + + +def build_comparison_report( + case: ReplayCase, + *, + backend_a: str, + backend_b: str, + diffs: list[DiffEntry], + runtime_context: Optional[dict[str, Any]] = None, +) -> dict[str, Any]: + expected_paths = set(expected_diff_paths_for_backend_pair(case, backend_a=backend_a, backend_b=backend_b)) + allowed_paths = set(case.allowed_diff_paths) + detected_paths = {diff.path for diff in diffs if not diff.allowed} + return { + "backend_a": backend_a, + "backend_b": backend_b, + "expected_diff_paths": sorted(expected_paths), + "allowed_diff_paths": sorted(allowed_paths), + "detected_diff_paths": sorted(detected_paths), + "missing_expected_paths": sorted(expected_paths - detected_paths), + "unexpected_diff_paths": sorted(detected_paths - expected_paths), + "runtime_context": runtime_context or {}, + "diffs": [diff.to_dict() for diff in diffs], + } + + +def build_case_matrix_report(case: ReplayCase, comparisons: list[dict[str, Any]]) -> dict[str, Any]: + expected_paths = set(case.expected_diff_paths) + allowed_paths = set(case.allowed_diff_paths) + detected_paths = { + path + for comparison in comparisons + for path in comparison.get("detected_diff_paths", []) + } + all_diffs = [ + diff + for comparison in comparisons + for diff in comparison.get("diffs", []) + ] + return { + "case_id": case.case_id, + "description": case.description, + "expects_diffs": bool(expected_paths), + "expected_diff_paths": sorted(expected_paths), + "allowed_diff_paths": sorted(allowed_paths), + "detected_diff_paths": sorted(detected_paths), + "missing_expected_paths": sorted(expected_paths - detected_paths), + "unexpected_diff_paths": sorted(detected_paths - expected_paths), + "comparison_count": len(comparisons), + "comparisons": comparisons, + "diffs": all_diffs, + } + + +def write_diff_report( + report_path: Path, + case_reports: list[dict[str, Any]], + metadata: Optional[dict[str, Any]] = None, +) -> None: + """Write grouped replay reports to JSON.""" + + payload = { + "meta": metadata or {}, + "cases": case_reports, + } + report_path.write_text(json.dumps(payload, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") diff --git a/tests/sessions/replay_models.py b/tests/sessions/replay_models.py new file mode 100644 index 00000000..9a590f63 --- /dev/null +++ b/tests/sessions/replay_models.py @@ -0,0 +1,262 @@ +"""Typed models for session/memory replay consistency tests. + +The replay harness uses a small protocol instead of hard-coding backend calls +inside each test. This keeps the tests readable and makes it easy to add new +backends or new replay cases over time. +""" + +from __future__ import annotations + +from dataclasses import asdict +from dataclasses import dataclass +from dataclasses import field +from enum import Enum +from typing import Any +from typing import Optional + + +class ReplayStepKind(str, Enum): + """Supported operations in a replay case.""" + + CREATE_SESSION = "create_session" + APPEND_EVENT = "append_event" + APPEND_STATE = "append_state" + STORE_MEMORY = "store_memory" + SEARCH_MEMORY = "search_memory" + CREATE_SUMMARY = "create_summary" + RESTART_SERVICES = "restart_services" + + +class SnapshotMutationOperation(str, Enum): + """Supported snapshot mutation operations for negative cases.""" + + SET = "set" + DELETE = "delete" + + +class RuntimeFaultOperation(str, Enum): + """Supported runtime fault injections for negative replay cases.""" + + DUPLICATE_LAST_EVENT = "duplicate_last_event" + DROP_LAST_EVENT_KEEP_STATE = "drop_last_event_keep_state" + SET_SESSION_VALUE = "set_session_value" + DELETE_SUMMARY = "delete_summary" + SET_SUMMARY_VALUE = "set_summary_value" + + +@dataclass(frozen=True) +class FunctionCallSpec: + """Declarative function call content for an event.""" + + name: str + args: dict[str, Any] = field(default_factory=dict) + call_id: Optional[str] = None + + +@dataclass(frozen=True) +class FunctionResponseSpec: + """Declarative function response content for an event.""" + + name: str + response: dict[str, Any] = field(default_factory=dict) + call_id: Optional[str] = None + + +@dataclass(frozen=True) +class EventSpec: + """Business-level event description used by replay steps.""" + + author: str + text: Optional[str] = None + role: Optional[str] = None + state_delta: dict[str, Any] = field(default_factory=dict) + function_calls: tuple[FunctionCallSpec, ...] = field(default_factory=tuple) + function_responses: tuple[FunctionResponseSpec, ...] = field(default_factory=tuple) + branch: Optional[str] = None + visible: bool = True + partial: bool = False + is_summary_event: bool = False + event_id: Optional[str] = None + + +@dataclass(frozen=True) +class MemoryQuerySpec: + """Search query to run after replaying a case.""" + + name: str + query: str + limit: int = 10 + + +@dataclass(frozen=True) +class ReplayStep: + """One replay operation.""" + + kind: ReplayStepKind + event: Optional[EventSpec] = None + initial_state: dict[str, Any] = field(default_factory=dict) + memory_query: Optional[MemoryQuerySpec] = None + force_summary: bool = False + session_alias: str = "default" + session_id: Optional[str] = None + + @classmethod + def create_session( + cls, + *, + initial_state: Optional[dict[str, Any]] = None, + session_alias: str = "default", + session_id: Optional[str] = None, + ) -> "ReplayStep": + return cls( + kind=ReplayStepKind.CREATE_SESSION, + initial_state=initial_state or {}, + session_alias=session_alias, + session_id=session_id, + ) + + @classmethod + def append_event(cls, event: EventSpec, *, session_alias: str = "default") -> "ReplayStep": + return cls(kind=ReplayStepKind.APPEND_EVENT, event=event, session_alias=session_alias) + + @classmethod + def append_state( + cls, + *, + author: str, + state_delta: dict[str, Any], + session_alias: str = "default", + ) -> "ReplayStep": + return cls( + kind=ReplayStepKind.APPEND_STATE, + event=EventSpec(author=author, state_delta=state_delta), + session_alias=session_alias, + ) + + @classmethod + def store_memory(cls, *, session_alias: str = "default") -> "ReplayStep": + return cls(kind=ReplayStepKind.STORE_MEMORY, session_alias=session_alias) + + @classmethod + def search_memory( + cls, + *, + name: str, + query: str, + limit: int = 10, + session_alias: str = "default", + ) -> "ReplayStep": + return cls( + kind=ReplayStepKind.SEARCH_MEMORY, + memory_query=MemoryQuerySpec(name=name, query=query, limit=limit), + session_alias=session_alias, + ) + + @classmethod + def create_summary(cls, *, force: bool = False, session_alias: str = "default") -> "ReplayStep": + return cls(kind=ReplayStepKind.CREATE_SUMMARY, force_summary=force, session_alias=session_alias) + + @classmethod + def restart_services(cls) -> "ReplayStep": + return cls(kind=ReplayStepKind.RESTART_SERVICES) + + +@dataclass(frozen=True) +class ReplayCase: + """A deterministic replay scenario.""" + + case_id: str + description: str + app_name: str = "replay_app" + user_id: str = "replay_user" + session_id: str = "replay_session" + enable_summary: bool = False + summary_keep_recent_count: int = 2 + store_historical_events: bool = False + steps: tuple[ReplayStep, ...] = field(default_factory=tuple) + allowed_diff_paths: tuple[str, ...] = field(default_factory=tuple) + expected_diff_paths: tuple[str, ...] = field(default_factory=tuple) + snapshot_mutations: tuple["SnapshotMutation", ...] = field(default_factory=tuple) + runtime_faults: tuple["RuntimeFault", ...] = field(default_factory=tuple) + + +@dataclass(frozen=True) +class SnapshotMutation: + """A deterministic post-replay mutation applied to one backend snapshot.""" + + backend_name: str + path: str + operation: SnapshotMutationOperation = SnapshotMutationOperation.SET + value: Any = None + + +@dataclass(frozen=True) +class RuntimeFault: + """A deterministic fault injected during replay execution.""" + + backend_name: str + after_step: int + operation: RuntimeFaultOperation + path: Optional[str] = None + value: Any = None + + +@dataclass(frozen=True) +class SummarySnapshot: + """Comparable summary view extracted from a backend.""" + + session_id: str + summary_text: str + original_event_count: int + compressed_event_count: int + summary_id: Optional[str] = None + version: Optional[int] = None + replaces: Optional[str] = None + summarized_event_count: Optional[int] = None + summary_timestamp: Optional[float] = None + metadata: dict[str, Any] = field(default_factory=dict) + + def to_dict(self) -> dict[str, Any]: + return asdict(self) + + +@dataclass(frozen=True) +class BackendSnapshot: + """Full business snapshot collected after replaying one case.""" + + backend_name: str + case_id: str + app_name: str + user_id: str + session_id: str + session: dict[str, Any] + state: dict[str, Any] + memory: dict[str, Any] + summary: Optional[SummarySnapshot] + + def to_dict(self) -> dict[str, Any]: + data = asdict(self) + if self.summary is not None: + data["summary"] = self.summary.to_dict() + return data + + +@dataclass(frozen=True) +class DiffEntry: + """One structured difference between two backend snapshots.""" + + case_id: str + backend_a: str + backend_b: str + scope: str + path: str + left: Any + right: Any + allowed: bool + session_id: Optional[str] = None + event_index: Optional[int] = None + summary_id: Optional[str] = None + reason: Optional[str] = None + + def to_dict(self) -> dict[str, Any]: + return asdict(self) diff --git a/tests/sessions/replay_summary.py b/tests/sessions/replay_summary.py new file mode 100644 index 00000000..950ec269 --- /dev/null +++ b/tests/sessions/replay_summary.py @@ -0,0 +1,55 @@ +"""Deterministic summary helpers for replay tests. + +The replay harness must not depend on a real LLM. This module provides a tiny +deterministic summarizer that reuses the framework's session compaction logic +while producing stable summary text and metadata. +""" + +from __future__ import annotations + +import re + +from trpc_agent_sdk.sessions import Session +from trpc_agent_sdk.sessions import SessionSummarizer +from trpc_agent_sdk.sessions import SummarizerSessionManager + + +_WHITESPACE_RE = re.compile(r"\s+") + + +class _ReplaySummaryModel: + """Minimal model-like object used only for summary metadata.""" + + name = "replay-summary-model" + + +class DeterministicSessionSummarizer(SessionSummarizer): + """A SessionSummarizer that produces deterministic text without LLM calls.""" + + def __init__(self, *, keep_recent_count: int = 2, start_by_user_turn: bool = True) -> None: + super().__init__( + model=_ReplaySummaryModel(), # type: ignore[arg-type] + keep_recent_count=keep_recent_count, + start_by_user_turn=start_by_user_turn, + ) + self._minimum_events = max(keep_recent_count + 1, 2) + + async def should_summarize(self, session: Session) -> bool: + return len(session.events) >= self._minimum_events + + async def _generate_summary(self, conversation_text: str, ctx=None) -> str: + normalized = _WHITESPACE_RE.sub(" ", conversation_text).strip() + if not normalized: + return "" + return f"deterministic-summary[{normalized}]" + + +def build_replay_summarizer_manager(*, keep_recent_count: int) -> SummarizerSessionManager: + """Create a deterministic summarizer manager for replay tests.""" + + summarizer = DeterministicSessionSummarizer(keep_recent_count=keep_recent_count) + return SummarizerSessionManager( + model=summarizer.model, + summarizer=summarizer, + auto_summarize=True, + ) diff --git a/tests/sessions/test_replay_consistency.py b/tests/sessions/test_replay_consistency.py new file mode 100644 index 00000000..fc40b01c --- /dev/null +++ b/tests/sessions/test_replay_consistency.py @@ -0,0 +1,200 @@ +"""Replay consistency smoke tests across session and memory backends.""" + +from __future__ import annotations + +import asyncio +import os +from time import perf_counter +from typing import Callable +from typing import Type + +import pytest + +from .replay_cases import REPLAY_ACCEPTANCE_CASES +from .replay_cases import REPLAY_ALL_CASES +from .replay_cases import REPLAY_EXTRA_CASES +from .replay_harness import DEFAULT_REPORT_PATH +from .replay_harness import build_case_matrix_report +from .replay_harness import InMemoryReplayAdapter +from .replay_harness import RedisReplayAdapter +from .replay_harness import ReplayBackendAdapter +from .replay_harness import SqliteReplayAdapter +from .replay_harness import build_comparison_report +from .replay_harness import diff_backend_snapshots +from .replay_harness import expected_diff_paths_for_backend_pair +from .replay_harness import format_diffs +from .replay_harness import get_replay_clock_metadata +from .replay_harness import write_diff_report +from .replay_models import BackendSnapshot +from .replay_models import DiffEntry +from .replay_models import ReplayCase + + +ADAPTER_TYPES: tuple[Type[ReplayBackendAdapter], ...] = ( + InMemoryReplayAdapter, + SqliteReplayAdapter, +) +REDIS_REPLAY_URL_ENV = "TRPC_AGENT_REPLAY_REDIS_URL" +AdapterFactory = Callable[[], ReplayBackendAdapter] + + +async def _run_case_on_backend( + adapter_factory: AdapterFactory, + case: ReplayCase, +) -> tuple[BackendSnapshot, dict[str, object], dict[str, object]]: + adapter = adapter_factory() + await adapter.setup(case) + try: + snapshot = await adapter.run_case(case) + return snapshot, adapter.get_runtime_metadata(), adapter.get_report_metadata() + finally: + await adapter.close() + + +async def _run_replay_cases( + adapter_factories: tuple[AdapterFactory, ...], + *, + mode_name: str, + write_report: bool = True, +) -> tuple[list[DiffEntry], list[dict[str, object]], float]: + all_diffs: list[DiffEntry] = [] + case_reports: list[dict[str, object]] = [] + backend_report_metadata: dict[str, dict[str, object]] = {} + start_time = perf_counter() + + for case in REPLAY_ALL_CASES: + backend_runs = [await _run_case_on_backend(adapter_factory, case) for adapter_factory in adapter_factories] + snapshots = [run[0] for run in backend_runs] + runtime_metadata = { + snapshot.backend_name: runtime_info + for snapshot, runtime_info, _ in backend_runs + } + for snapshot, _, report_metadata in backend_runs: + backend_report_metadata.setdefault(snapshot.backend_name, report_metadata) + baseline_snapshot = snapshots[0] + comparisons: list[dict[str, object]] = [] + for other_snapshot in snapshots[1:]: + diffs = diff_backend_snapshots(case=case, left=baseline_snapshot, right=other_snapshot) + all_diffs.extend(diffs) + comparisons.append( + build_comparison_report( + case, + backend_a=baseline_snapshot.backend_name, + backend_b=other_snapshot.backend_name, + diffs=diffs, + runtime_context={ + baseline_snapshot.backend_name: runtime_metadata[baseline_snapshot.backend_name], + other_snapshot.backend_name: runtime_metadata[other_snapshot.backend_name], + }, + )) + case_reports.append(build_case_matrix_report(case, comparisons)) + + elapsed_seconds = perf_counter() - start_time + if write_report: + write_diff_report( + DEFAULT_REPORT_PATH, + case_reports, + metadata={ + "mode": mode_name, + "elapsed_seconds": round(elapsed_seconds, 3), + "backend_names": [factory.name for factory in adapter_factories], + "baseline_backend": adapter_factories[0].name, + "comparison_mode": "baseline_vs_all", + "backend_summaries": [backend_report_metadata[name] for name in sorted(backend_report_metadata)], + "clock_strategy": get_replay_clock_metadata(), + "acceptance_case_count": len(REPLAY_ACCEPTANCE_CASES), + "all_case_count": len(REPLAY_ALL_CASES), + }, + ) + return all_diffs, case_reports, elapsed_seconds + + +def _make_redis_adapter_factory() -> AdapterFactory: + redis_url = os.getenv(REDIS_REPLAY_URL_ENV) + if not redis_url: + raise RuntimeError("Redis replay URL is not configured.") + + class _ConfiguredRedisReplayAdapter(RedisReplayAdapter): + name = "redis" + + def __init__(self) -> None: + super().__init__(redis_url=redis_url) + + return _ConfiguredRedisReplayAdapter + + +def _assert_case_expectations( + all_diffs: list[DiffEntry], + case_set: tuple[ReplayCase, ...], + case_reports: list[dict[str, object]], +) -> None: + case_map = {case.case_id: case for case in case_set} + report_map = {str(report["case_id"]): report for report in case_reports} + + for case_id, case in case_map.items(): + case_report = report_map[case_id] + comparisons = case_report.get("comparisons", []) + for comparison in comparisons: + backend_a = str(comparison["backend_a"]) + backend_b = str(comparison["backend_b"]) + expected_paths = set(expected_diff_paths_for_backend_pair(case, backend_a=backend_a, backend_b=backend_b)) + case_diffs = [ + diff + for diff in all_diffs + if diff.case_id == case_id + and diff.backend_a == backend_a + and diff.backend_b == backend_b + and not diff.allowed + ] + detected_paths = {diff.path for diff in case_diffs} + + if expected_paths: + missing_paths = sorted(expected_paths - detected_paths) + unexpected_paths = sorted(detected_paths - expected_paths) + assert not missing_paths, ( + f"{case_id} missing expected diffs for {backend_a} vs {backend_b}: {missing_paths}" + ) + assert not unexpected_paths, ( + f"{case_id} produced unexpected diffs for {backend_a} vs {backend_b}: {unexpected_paths}\n" + f"{format_diffs(case_diffs)}" + ) + continue + + assert not case_diffs, ( + f"{case_id} produced unexpected diffs for {backend_a} vs {backend_b}:\n" + f"{format_diffs(case_diffs)}" + ) + + +def test_replay_consistency_smoke_cases() -> None: + """Ensure acceptance and extended replay cases behave as expected.""" + + all_diffs, case_reports, elapsed_seconds = asyncio.run(_run_replay_cases(ADAPTER_TYPES, mode_name="lightweight")) + _assert_case_expectations(all_diffs, REPLAY_ALL_CASES, case_reports) + + assert elapsed_seconds <= 30.0, f"lightweight replay mode exceeded 30s: {elapsed_seconds:.3f}s" + + +def test_acceptance_case_count() -> None: + """Keep the public acceptance suite fixed at 10 cases.""" + + assert len(REPLAY_ACCEPTANCE_CASES) == 10 + assert len(REPLAY_ALL_CASES) >= len(REPLAY_ACCEPTANCE_CASES) + assert len(REPLAY_EXTRA_CASES) >= 1 + + +def test_replay_consistency_redis_integration_mode() -> None: + """Run an optional Redis-backed integration comparison when configured.""" + + redis_url = os.getenv(REDIS_REPLAY_URL_ENV) + if not redis_url: + pytest.skip(f"{REDIS_REPLAY_URL_ENV} is not set") + + redis_adapter_factory = _make_redis_adapter_factory() + all_diffs, case_reports, _ = asyncio.run( + _run_replay_cases( + (InMemoryReplayAdapter, SqliteReplayAdapter, redis_adapter_factory), + mode_name="integration", + write_report=True, + )) + _assert_case_expectations(all_diffs, REPLAY_ALL_CASES, case_reports) diff --git a/trpc_agent_sdk/sessions/_in_memory_session_service.py b/trpc_agent_sdk/sessions/_in_memory_session_service.py index a2c542ed..40e31709 100644 --- a/trpc_agent_sdk/sessions/_in_memory_session_service.py +++ b/trpc_agent_sdk/sessions/_in_memory_session_service.py @@ -469,6 +469,10 @@ def _set_session(self, app_name: str, user_id: str, session_id: str, session: Se if user_id not in self._sessions[app_name]: self._sessions[app_name][user_id] = {} + # Storage should own its own session object. Reusing the caller's + # instance can create aliasing after update_session(), which makes + # later append_event() mutate the same object twice. + session = session.model_copy(deep=True) if not self._session_config.store_historical_events: session = session.model_copy(update={"historical_events": []}) diff --git a/trpc_agent_sdk/sessions/_redis_session_service.py b/trpc_agent_sdk/sessions/_redis_session_service.py index 01f77182..6abf4a60 100644 --- a/trpc_agent_sdk/sessions/_redis_session_service.py +++ b/trpc_agent_sdk/sessions/_redis_session_service.py @@ -36,6 +36,20 @@ from ._utils import user_state_key +def _decode_redis_hash(raw_state: Optional[dict[Any, Any]]) -> dict[str, Any]: + """Normalize Redis hash responses to plain Python strings when possible.""" + + if not raw_state: + return {} + + normalized: dict[str, Any] = {} + for raw_key, raw_value in raw_state.items(): + key = raw_key.decode("utf-8") if isinstance(raw_key, bytes) else raw_key + value = raw_value.decode("utf-8") if isinstance(raw_value, bytes) else raw_value + normalized[str(key)] = value + return normalized + + class RedisSessionService(BaseSessionService): """A Redis implementation of the session service. @@ -246,7 +260,7 @@ async def _update_app_state(self, redis_session: RedisSession, app_name: str, key = app_state_key(app_name) command = RedisCommand(method='hgetall', args=(key, )) - app_state: dict[str, Any] = await self._redis_storage.execute_command(redis_session, command) + app_state = _decode_redis_hash(await self._redis_storage.execute_command(redis_session, command)) if app_state: app_state.update(state_delta) else: @@ -259,13 +273,9 @@ async def _update_app_state(self, redis_session: RedisSession, app_name: str, await self._refresh_ttl(redis_session, key) return app_state - # Use HSET with TTL if TTL is configured, otherwise use HSET - args = [key] - for k, v in app_state.items(): - args.extend([k, v]) - command = RedisCommand(method='hset', - args=tuple(args), + args=(key, ), + kwargs={"mapping": app_state}, expire=RedisExpire(key=key, ttl=self._session_config.ttl)) await self._redis_storage.execute_command(redis_session, command) @@ -286,7 +296,7 @@ async def _update_user_state(self, redis_session: RedisSession, app_name: str, u key = user_state_key(app_name, user_id) command = RedisCommand(method='hgetall', args=(key, )) - user_state: dict[str, Any] = await self._redis_storage.execute_command(redis_session, command) + user_state = _decode_redis_hash(await self._redis_storage.execute_command(redis_session, command)) if user_state: user_state.update(state_delta) else: @@ -299,13 +309,9 @@ async def _update_user_state(self, redis_session: RedisSession, app_name: str, u await self._refresh_ttl(redis_session, key) return user_state - # Use HSET with TTL if TTL is configured, otherwise use HSET - args = [key] - for k, v in user_state.items(): - args.extend([k, v]) - command = RedisCommand(method='hset', - args=tuple(args), + args=(key, ), + kwargs={"mapping": user_state}, expire=RedisExpire(key=key, ttl=self._session_config.ttl)) await self._redis_storage.execute_command(redis_session, command) @@ -344,7 +350,7 @@ async def _get_app_state(self, redis_session: RedisSession, app_name: str) -> di """ key = app_state_key(app_name) command = RedisCommand(method='hgetall', args=(key, )) - app_state = await self._redis_storage.execute_command(redis_session, command) + app_state = _decode_redis_hash(await self._redis_storage.execute_command(redis_session, command)) if app_state: await self._refresh_ttl(redis_session, key) @@ -365,7 +371,7 @@ async def _get_user_state(self, redis_session: RedisSession, app_name: str, user """ key = user_state_key(app_name, user_id) command = RedisCommand(method='hgetall', args=(key, )) - user_state = await self._redis_storage.execute_command(redis_session, command) + user_state = _decode_redis_hash(await self._redis_storage.execute_command(redis_session, command)) if user_state: await self._refresh_ttl(redis_session, key) return user_state or {} diff --git a/trpc_agent_sdk/sessions/_sql_session_service.py b/trpc_agent_sdk/sessions/_sql_session_service.py index d5ad5a08..6f0c9e0d 100644 --- a/trpc_agent_sdk/sessions/_sql_session_service.py +++ b/trpc_agent_sdk/sessions/_sql_session_service.py @@ -105,6 +105,24 @@ def _events_from_storage(events: Optional[list[dict[str, Any]]]) -> list[Event]: return [Event.model_validate(event) for event in (events or [])] +def _normalize_active_event_order(events: list[Event]) -> list[Event]: + """Keep a summary anchor at the front of the active event window. + + Persistent backends reconstruct events from storage using timestamp order, + but a summary event is a logical anchor inserted at the beginning of the + active context window even though it is created later in time. Moving the + first summary anchor to the front keeps SQL behavior aligned with the + in-memory session semantics. + """ + + for index, event in enumerate(events): + if event.is_summary_event(): + if index == 0: + return events + return [event] + events[:index] + events[index + 1:] + return events + + class SessionStorageBase(DeclarativeBase): """Base class for SqlSessionService tables only. @@ -482,7 +500,7 @@ async def get_session( user_state_delta=user_state, session_state=storage_session.state)) - events = [e.to_event() for e in reversed(storage_events)] + events = _normalize_active_event_order([e.to_event() for e in reversed(storage_events)]) historical_events = (_events_from_storage(storage_session.historical_events) if self._session_config.store_historical_events else []) session = storage_session.to_session(state=merged_state, events=events, historical_events=historical_events) @@ -569,7 +587,7 @@ async def append_event(self, session: Session, event: Event) -> Event: session.last_update_time = storage_session.update_timestamp_tz session.state = storage_session.state session.conversation_count = storage_session.conversation_count - session.events = [e.to_event() for e in reversed(storage_events)] + session.events = _normalize_active_event_order([e.to_event() for e in reversed(storage_events)]) session.historical_events = (_events_from_storage(storage_session.historical_events) if self._session_config.store_historical_events else []) @@ -788,12 +806,15 @@ async def _cleanup_expired_async(self) -> None: async def _cleanup_loop(self) -> None: """Background task for periodic cleanup of expired sessions and states.""" logger.debug("Cleanup task started with interval: %ss", self._session_config.ttl.cleanup_interval_seconds) + if self.__cleanup_stop_event is None: + logger.debug("Cleanup loop exited because stop event was not initialized") + return try: - while not self.__cleanup_stop_event.is_set(): + stop_event = self.__cleanup_stop_event + while not stop_event.is_set(): try: - await asyncio.wait_for(self.__cleanup_stop_event.wait(), - timeout=self._session_config.ttl.cleanup_interval_seconds) + await asyncio.wait_for(stop_event.wait(), timeout=self._session_config.ttl.cleanup_interval_seconds) break except asyncio.TimeoutError: try: diff --git a/trpc_agent_sdk/sessions/_summarizer_manager.py b/trpc_agent_sdk/sessions/_summarizer_manager.py index 2a07f58c..1226cdff 100644 --- a/trpc_agent_sdk/sessions/_summarizer_manager.py +++ b/trpc_agent_sdk/sessions/_summarizer_manager.py @@ -42,6 +42,9 @@ from ._session_summarizer import SessionSummary +_SUMMARY_EVENT_METADATA_KEY = "session_summary" + + class SummarizerSessionManager: """Session service with automatic summarization capabilities. @@ -71,6 +74,109 @@ def __init__( self._auto_summarize = auto_summarize self._summarizer_cache: Dict[str, Dict[str, Dict[str, SessionSummary]]] = {} + def _get_cached_summary(self, session: Session) -> Optional[SessionSummary]: + """Return cached summary for a session if available.""" + + app_name = session.app_name + user_id = session.user_id + cached_summary = self._summarizer_cache.get(app_name, {}).get(user_id, {}).get(session.id) + if cached_summary is not None: + return cached_summary + + restored_summary = self._restore_summary_from_session(session) + if restored_summary is not None: + self._cache_summary(session, restored_summary) + return restored_summary + + def _cache_summary(self, session: Session, summary: SessionSummary) -> None: + """Cache a session summary for fast same-process reads.""" + + app_name = session.app_name + user_id = session.user_id + if app_name not in self._summarizer_cache: + self._summarizer_cache[app_name] = {} + if user_id not in self._summarizer_cache[app_name]: + self._summarizer_cache[app_name][user_id] = {} + self._summarizer_cache[app_name][user_id][session.id] = summary + + def _get_summary_event(self, session: Session): + """Return the persisted summary anchor event if present.""" + + for event in session.events: + if event.is_summary_event(): + return event + return None + + def _serialize_summary(self, summary: SessionSummary) -> Dict[str, Any]: + """Serialize a summary into event metadata for persistence.""" + + return { + "session_id": summary.session_id, + "summary_text": summary.summary_text, + "original_event_count": summary.original_event_count, + "compressed_event_count": summary.compressed_event_count, + "summary_timestamp": summary.summary_timestamp, + "metadata": dict(summary.metadata or {}), + } + + def _persist_summary_to_session(self, session: Session, summary: SessionSummary) -> None: + """Attach summary metadata to the persisted summary event.""" + + summary_event = self._get_summary_event(session) + if summary_event is None: + return + custom_metadata = dict(summary_event.custom_metadata or {}) + custom_metadata[_SUMMARY_EVENT_METADATA_KEY] = self._serialize_summary(summary) + summary_event.custom_metadata = custom_metadata + summary_event.timestamp = summary.summary_timestamp + + def _restore_summary_from_session(self, session: Session) -> Optional[SessionSummary]: + """Rebuild a SessionSummary from persisted summary-event metadata.""" + + summary_event = self._get_summary_event(session) + if summary_event is None or not summary_event.custom_metadata: + return None + + persisted_summary = summary_event.custom_metadata.get(_SUMMARY_EVENT_METADATA_KEY) + if not isinstance(persisted_summary, dict): + return None + + try: + return SessionSummary( + session_id=str(persisted_summary["session_id"]), + summary_text=str(persisted_summary["summary_text"]), + original_event_count=int(persisted_summary["original_event_count"]), + compressed_event_count=int(persisted_summary["compressed_event_count"]), + summary_timestamp=float(persisted_summary["summary_timestamp"]), + metadata=dict(persisted_summary.get("metadata") or {}), + ) + except (KeyError, TypeError, ValueError) as ex: + logger.warning("Failed to restore persisted summary for session %s: %s", session.id, ex) + return None + + def _build_summary_metadata( + self, + *, + session: Session, + original_event_count: int, + compressed_event_count: int, + previous_summary: Optional[SessionSummary] = None, + ) -> Dict[str, Any]: + """Build stable lineage metadata for a session summary.""" + if previous_summary is None: + previous_summary = self._get_cached_summary(session) + previous_metadata = dict(previous_summary.metadata) if previous_summary and previous_summary.metadata else {} + previous_version = int(previous_metadata.get("version", 0) or 0) + version = previous_version + 1 + summary_id = f"{session.id}:summary:v{version}" + summarized_event_count = max(0, original_event_count - compressed_event_count + 1) + return { + "summary_id": summary_id, + "version": version, + "replaces": previous_metadata.get("summary_id"), + "summarized_event_count": summarized_event_count, + } + def set_session_service(self, session_service: SessionServiceABC, force: bool = False) -> None: """Set the session service to use. @@ -104,6 +210,7 @@ async def create_session_summary(self, # Check if session should be summarized if is_should_summarize: logger.debug("Summarizing session %s", session.id) + previous_summary = self._get_cached_summary(session) # Compress the session so the active events list contains only # model-visible summary/recent events. Raw events are retained only @@ -116,19 +223,21 @@ async def create_session_summary(self, summary_text = await self._summarizer.create_session_summary( session, ctx, store_historical_events=store_historical_events) if summary_text: - app_name = session.app_name - user_id = session.user_id - if app_name not in self._summarizer_cache: - self._summarizer_cache[app_name] = {} - if user_id not in self._summarizer_cache[app_name]: - self._summarizer_cache[app_name][user_id] = {} - self._summarizer_cache[app_name][user_id][session.id] = SessionSummary( + summary = SessionSummary( session_id=session.id, summary_text=summary_text, original_event_count=original_event_count, compressed_event_count=len(session.events), summary_timestamp=time.time(), + metadata=self._build_summary_metadata( + session=session, + original_event_count=original_event_count, + compressed_event_count=len(session.events), + previous_summary=previous_summary, + ), ) + self._cache_summary(session, summary) + self._persist_summary_to_session(session, summary) # Update the stored session if self._base_service: await self._base_service.update_session(session) @@ -142,16 +251,9 @@ async def get_session_summary(self, session: Session) -> Optional[SessionSummary Returns: SessionSummary if successful, None otherwise """ - if not self._summarizer or not self._summarizer_cache: - return None - app_name = session.app_name - user_id = session.user_id - - if app_name not in self._summarizer_cache or user_id not in self._summarizer_cache[ - app_name] or session.id not in self._summarizer_cache[app_name][user_id]: - return None - - return self._summarizer_cache[app_name][user_id][session.id] + if not self._summarizer: + return self._restore_summary_from_session(session) + return self._get_cached_summary(session) def get_summarizer_metadata(self) -> Dict[str, Any]: """Get metadata about the summarizer configuration. From 2d5d4331f5c724e1e933c257506356463a8bcfd0 Mon Sep 17 00:00:00 2001 From: Cai Zichen <3234912331@qq.com> Date: Sat, 11 Jul 2026 19:34:19 +0800 Subject: [PATCH 2/4] test(sessions): tighten replay harness coverage Extend replay snapshots to retain all session aliases, preserve memory query observations across restarts, and allow runtime faults to target non-active sessions. Add cross-user isolation coverage plus targeted harness tests for duplicate query names and restart-stable memory observations. --- tests/sessions/replay_acceptance_design.md | 6 +- tests/sessions/replay_cases.py | 135 +++++++++ tests/sessions/replay_harness.py | 323 ++++++++++++++++----- tests/sessions/replay_models.py | 32 ++ tests/sessions/test_replay_consistency.py | 61 ++++ 5 files changed, 478 insertions(+), 79 deletions(-) diff --git a/tests/sessions/replay_acceptance_design.md b/tests/sessions/replay_acceptance_design.md index e649a19f..fbff01fc 100644 --- a/tests/sessions/replay_acceptance_design.md +++ b/tests/sessions/replay_acceptance_design.md @@ -1,7 +1,7 @@ # Replay Consistency Design Note ## 设计说明 -本框架以统一 `ReplayCase` 协议驱动 `InMemory`、文件型 `SQLite` 与可选 `Redis` 后端，先生成 `session/state/memory/summary` 四类快照，再做归一化比较。归一化只移除非业务噪声，不掩盖事件顺序、state 最终值、memory 检索结果及 summary 归属与覆盖关系；其中普通时间戳做精度收敛，`summary_timestamp` 使用 case 级确定性时钟参与比较。summary 比较分为两层：一层比较摘要文本与压缩后的事件窗口，另一层比较 `summary_id/version/replaces/session_id` 等 lineage 元数据。为验证保存/读取语义，summary 元数据会随 summary event 一起持久化，`SQLite/Redis` 适配器在抓取快照前执行一次关服务后重开读回，并支持在 case 中显式插入中途重启步骤验证“重启后继续写”。协议还支持同一 case 内用 `session_alias` 驱动多个 session，以覆盖跨 session 的 memory 聚合和 `app:/user:/temp:` 状态可见性差异。负例分为 snapshot mutation 与 runtime fault 两类，前者验证比较器精度，后者验证重复写入、中途失败和运行时污染场景。轻量模式默认运行 `InMemory + SQLite`，集成模式通过环境变量启用 `Redis`。 +本框架以统一 `ReplayCase` 协议驱动 `InMemory`、文件型 `SQLite` 与可选 `Redis` 后端，先生成 `session/state/memory/summary` 四类快照，再做归一化比较。归一化只移除非业务噪声，不掩盖事件顺序、state 最终值、memory 检索结果及 summary 归属与覆盖关系；其中普通时间戳做精度收敛，`summary_timestamp` 使用 case 级确定性时钟参与比较。summary 比较分为两层：一层比较摘要文本与压缩后的事件窗口，另一层比较 `summary_id/version/replaces/session_id` 等 lineage 元数据。为验证保存/读取语义，summary 元数据会随 summary event 一起持久化，`SQLite/Redis` 适配器在抓取快照前执行一次关服务后重开读回，并支持在 case 中显式插入中途重启步骤验证“重启后继续写”。协议支持用 `session_alias` 在同一 case 内驱动多 session / 多 user，并在最终快照中同时保留 `active session` 与 `sessions_by_alias` 视图，避免非活跃 session 损坏被漏检。memory 检索按 step 记录为独立 observation，不会在重启后被重算覆盖；同名 query 可跨 session 重复使用。负例分为 snapshot mutation 与 runtime fault 两类，二者都支持 alias 级注入，前者验证比较器精度，后者验证重复写入、中途失败、非活跃 session 污染和运行时 summary 破坏。轻量模式默认运行 `InMemory + SQLite`，集成模式通过环境变量启用 `Redis`。 ## 官方 10 条验收 Case | Case ID | 场景 | 验收点 | @@ -23,6 +23,10 @@ - `duplicate_event_runtime_fault`：补充重复写入异常。 - `runtime_state_corruption_fault`：补充运行时 state 污染。 - `runtime_summary_loss_fault`：补充运行时 summary 丢失。 +- `non_active_session_summary_loss_fault`：补充非活跃 session 的 summary 损坏检测。 - `cross_session_memory_aggregation`：补充同一 app/user 下跨 session 的 memory 聚合语义。 - `restart_mid_replay_after_summary`：补充 summary 持久化后中途重启再续写的恢复语义。 - `state_namespace_roundtrip`：补充 `app:/user:/temp:` 状态命名空间在跨 session 和重启后的可见性语义。 +- `cross_user_memory_isolation`：补充同一 app 下不同 user 的长期记忆隔离语义。 +- `duplicate_memory_query_name_across_sessions`：用定向测试覆盖跨 alias 的同名 memory query 不应互相覆盖。 +- `memory_query_observation_survives_restart`：用定向测试覆盖重启后 memory query 观测不得被后续结果回填。 diff --git a/tests/sessions/replay_cases.py b/tests/sessions/replay_cases.py index 901d4604..16e2673e 100644 --- a/tests/sessions/replay_cases.py +++ b/tests/sessions/replay_cases.py @@ -463,6 +463,47 @@ ), ), ), + ReplayCase( + case_id="non_active_session_summary_loss_fault", + description="A runtime summary deletion on a non-active session alias must still be detected through alias-scoped snapshots.", + session_id="replay-negative-target-session", + enable_summary=True, + summary_keep_recent_count=2, + store_historical_events=True, + expected_diff_paths=("sessions_by_alias.source.summary",), + runtime_faults=( + RuntimeFault( + backend_name=_PERSISTENT_BACKEND, + after_step=6, + operation=RuntimeFaultOperation.DELETE_SUMMARY, + session_alias="source", + ), + ), + steps=( + ReplayStep.create_session(session_alias="source", session_id="replay-negative-source-session"), + ReplayStep.append_event( + EventSpec(author="user", role="user", text="Please summarize my architecture notes."), + session_alias="source", + ), + ReplayStep.append_event( + EventSpec(author="assistant", role="model", text="Sure, continue with the details."), + session_alias="source", + ), + ReplayStep.append_event( + EventSpec(author="user", role="user", text="We run SQLite for local replay and Redis in production."), + session_alias="source", + ), + ReplayStep.append_event( + EventSpec(author="assistant", role="model", text="I will summarize the backend setup."), + session_alias="source", + ), + ReplayStep.create_summary(force=True, session_alias="source"), + ReplayStep.create_session(session_alias="default", session_id="replay-negative-target-session"), + ReplayStep.append_event( + EventSpec(author="user", role="user", text="Switch focus to the current session."), + ), + ), + ), ) @@ -575,6 +616,92 @@ ), ), ), + ReplayCase( + case_id="duplicate_memory_query_name_across_sessions", + description="Two sessions may reuse the same query name without overwriting earlier memory observations.", + session_id="replay-duplicate-query-target", + steps=( + ReplayStep.create_session(session_alias="source", session_id="replay-duplicate-query-source"), + ReplayStep.append_event( + EventSpec(author="user", role="user", text="Please remember that I prefer dragon well tea."), + session_alias="source", + ), + ReplayStep.append_event( + EventSpec(author="assistant", role="model", text="Dragon well tea preference recorded."), + session_alias="source", + ), + ReplayStep.store_memory(session_alias="source"), + ReplayStep.search_memory( + name="shared_preference_search", + query="dragon well", + limit=5, + session_alias="source", + ), + ReplayStep.restart_services(), + ReplayStep.create_session(session_alias="default", session_id="replay-duplicate-query-target"), + ReplayStep.search_memory( + name="shared_preference_search", + query="dragon well", + limit=5, + ), + ), + ), + ReplayCase( + case_id="memory_query_observation_survives_restart", + description="Memory query observations should retain their original step-time results across restarts and later writes.", + session_id="replay-memory-observation", + steps=( + ReplayStep.create_session(), + ReplayStep.append_event( + EventSpec(author="user", role="user", text="Please remember that my favorite tea is oolong."), + ), + ReplayStep.append_event( + EventSpec(author="assistant", role="model", text="I will remember your oolong preference."), + ), + ReplayStep.store_memory(), + ReplayStep.search_memory(name="tea_preference", query="oolong", limit=5), + ReplayStep.restart_services(), + ReplayStep.append_event( + EventSpec(author="user", role="user", text="Also remember that I enjoy jasmine tea."), + ), + ReplayStep.append_event( + EventSpec(author="assistant", role="model", text="I will remember the jasmine preference too."), + ), + ReplayStep.store_memory(), + ReplayStep.search_memory(name="tea_preference", query="jasmine", limit=5), + ), + ), + ReplayCase( + case_id="cross_user_memory_isolation", + description="Memory saved for one user must not become visible to another user in the same application.", + session_id="replay-cross-user-target", + steps=( + ReplayStep.create_session( + session_alias="source", + session_id="replay-cross-user-source", + user_id="replay-user-a", + ), + ReplayStep.append_event( + EventSpec(author="user", role="user", text="Please remember that I prefer matcha desserts."), + session_alias="source", + ), + ReplayStep.append_event( + EventSpec(author="assistant", role="model", text="Matcha dessert preference recorded."), + session_alias="source", + ), + ReplayStep.store_memory(session_alias="source"), + ReplayStep.create_session( + session_alias="default", + session_id="replay-cross-user-target", + user_id="replay-user-b", + ), + ReplayStep.search_memory( + name="cross_user_matcha_search", + query="matcha", + limit=5, + ), + ), + ), ) @@ -602,9 +729,17 @@ _NEGATIVE_CASES[4], # duplicate_event_runtime_fault _NEGATIVE_CASES[5], # runtime_state_corruption_fault _NEGATIVE_CASES[6], # runtime_summary_loss_fault + _NEGATIVE_CASES[9], # non_active_session_summary_loss_fault _ROBUSTNESS_CASES[0], # cross_session_memory_aggregation _ROBUSTNESS_CASES[1], # restart_mid_replay_after_summary _ROBUSTNESS_CASES[2], # state_namespace_roundtrip + _ROBUSTNESS_CASES[5], # cross_user_memory_isolation +) + + +REPLAY_TARGETED_CASES: tuple[ReplayCase, ...] = ( + _ROBUSTNESS_CASES[3], # duplicate_memory_query_name_across_sessions + _ROBUSTNESS_CASES[4], # memory_query_observation_survives_restart ) diff --git a/tests/sessions/replay_harness.py b/tests/sessions/replay_harness.py index 5262d40d..5f4bb547 100644 --- a/tests/sessions/replay_harness.py +++ b/tests/sessions/replay_harness.py @@ -59,6 +59,7 @@ from .replay_models import ReplayStepKind from .replay_models import RuntimeFault from .replay_models import RuntimeFaultOperation +from .replay_models import SessionSnapshot from .replay_models import SnapshotMutation from .replay_models import SnapshotMutationOperation from .replay_models import SummarySnapshot @@ -106,7 +107,7 @@ def __init__(self) -> None: self._memory_service: BaseMemoryService | None = None self._session: Session | None = None self._sessions: dict[str, Session] = {} - self._session_ids: dict[str, str] = {} + self._session_identities: dict[str, dict[str, str]] = {} self._active_session_alias = "default" self._memory_results: dict[str, Any] = {} self._case: ReplayCase | None = None @@ -144,7 +145,7 @@ async def setup(self, case: ReplayCase) -> None: self._memory_results = {} self._session = None self._sessions = {} - self._session_ids = {} + self._session_identities = {} self._active_session_alias = "default" self._event_sequence = 0 self._event_time_base = _deterministic_time_base(case.case_id) @@ -166,7 +167,7 @@ async def close(self) -> None: await self._close_services() self._session = None self._sessions = {} - self._session_ids = {} + self._session_identities = {} self._active_session_alias = "default" self._memory_results = {} self._case = None @@ -177,7 +178,7 @@ async def run_case(self, case: ReplayCase) -> BackendSnapshot: """Replay all steps in a case and collect a snapshot.""" for step_index, step in enumerate(case.steps): - await self._run_step(case, step) + await self._run_step(case, step, step_index) await self._apply_runtime_faults(step_index) if self.should_restart_before_snapshot(): await self._restart_services() @@ -204,11 +205,11 @@ async def _restart_services(self) -> None: self._session_service = await self._build_session_service() self._memory_service = await self._build_memory_service() reopened_sessions: dict[str, Session] = {} - for session_alias, session_id in self._session_ids.items(): + for session_alias, identity in self._session_identities.items(): reopened_session = await self.session_service.get_session( - app_name=self.case.app_name, - user_id=self.case.user_id, - session_id=session_id, + app_name=identity["app_name"], + user_id=identity["user_id"], + session_id=identity["session_id"], ) if reopened_session is not None: reopened_sessions[session_alias] = reopened_session @@ -220,34 +221,23 @@ async def _restart_services(self) -> None: if self._session is None: return - reopened_memory_results: dict[str, Any] = {} - for step in self.case.steps: - if step.kind == ReplayStepKind.SEARCH_MEMORY and step.memory_query is not None: - query_session = self._resolve_session(step.session_alias) - reopened_memory_results[step.memory_query.name] = await self._search_memory_for_session( - query_session, - step.memory_query, - ) - if reopened_memory_results: - self._memory_results = reopened_memory_results - def should_restart_during_replay(self) -> bool: """Return whether RESTART_SERVICES steps should perform a real restart.""" return False - async def _run_step(self, case: ReplayCase, step: ReplayStep) -> None: + async def _run_step(self, case: ReplayCase, step: ReplayStep, step_index: int) -> None: if step.kind == ReplayStepKind.CREATE_SESSION: session_id = step.session_id or case.session_id session = await self.session_service.create_session( - app_name=case.app_name, - user_id=case.user_id, + app_name=step.app_name or case.app_name, + user_id=step.user_id or case.user_id, session_id=session_id, state=step.initial_state, ) self._session = session self._sessions[step.session_alias] = session - self._session_ids[step.session_alias] = session.id + self._session_identities[step.session_alias] = self._make_session_identity(session) self._active_session_alias = step.session_alias return @@ -275,10 +265,16 @@ async def _run_step(self, case: ReplayCase, step: ReplayStep) -> None: if step.memory_query is None: raise ValueError("SEARCH_MEMORY step requires a query specification.") target_session = self._resolve_session(step.session_alias) - self._memory_results[step.memory_query.name] = await self._search_memory_for_session( - target_session, - step.memory_query, - ) + entries = await self._search_memory_for_session(target_session, step.memory_query) + self._memory_results[self._memory_observation_key(step_index, step)] = { + "query_name": step.memory_query.name, + "session_alias": step.session_alias, + "app_name": target_session.app_name, + "user_id": target_session.user_id, + "session_id": target_session.id, + "step_index": step_index, + "entries": entries, + } self._session = target_session return @@ -320,32 +316,42 @@ async def _search_memory_for_session( async def collect_snapshot(self, case: ReplayCase) -> BackendSnapshot: """Collect a backend snapshot after replay.""" - session = self._session - if session is None: + if self._session is None: + raise RuntimeError(f"Replay case '{case.case_id}' did not produce an active session.") + sessions_by_alias: dict[str, SessionSnapshot] = {} + for session_alias, identity in self._session_identities.items(): + session = await self.session_service.get_session( + app_name=identity["app_name"], + user_id=identity["user_id"], + session_id=identity["session_id"], + ) + if session is None: + raise RuntimeError(f"Replay session '{identity['session_id']}' was not found.") + self._sessions[session_alias] = session + sessions_by_alias[session_alias] = await self._build_session_snapshot(session_alias, session) + + active_alias = self._active_session_alias + active_snapshot = sessions_by_alias.get(active_alias) + if active_snapshot is None and sessions_by_alias: + active_alias = next(iter(sessions_by_alias)) + active_snapshot = sessions_by_alias[active_alias] + self._active_session_alias = active_alias + self._session = self._sessions[active_alias] + if active_snapshot is None: raise RuntimeError(f"Replay case '{case.case_id}' did not produce an active session.") - session = await self.session_service.get_session( - app_name=session.app_name, - user_id=session.user_id, - session_id=session.id, - ) - if session is None: - raise RuntimeError(f"Replay session '{self.session.id}' was not found.") - summary = await self._get_summary_snapshot(session) return BackendSnapshot( backend_name=self.name, case_id=case.case_id, - app_name=session.app_name, - user_id=session.user_id, - session_id=session.id, - session={ - "conversation_count": session.conversation_count, - "events": [_event_to_snapshot(event) for event in session.events], - "historical_events": [_event_to_snapshot(event) for event in session.historical_events], - }, - state=dict(session.state), + app_name=active_snapshot.app_name, + user_id=active_snapshot.user_id, + session_id=active_snapshot.session_id, + active_session_alias=active_alias, + session=active_snapshot.session, + state=active_snapshot.state, memory=dict(self._memory_results), - summary=summary, + summary=active_snapshot.summary, + sessions_by_alias=sessions_by_alias, ) def _resolve_session(self, session_alias: str) -> Session: @@ -385,29 +391,30 @@ async def _apply_runtime_faults(self, step_index: int) -> None: await self._apply_runtime_fault(fault) async def _apply_runtime_fault(self, fault: RuntimeFault) -> None: + target_session = self._get_fault_session(fault) if fault.operation == RuntimeFaultOperation.DUPLICATE_LAST_EVENT: - if not self.session.events: + if not target_session.events: raise RuntimeError("Cannot duplicate the last event of an empty session.") - duplicated_event = self.session.events[-1].model_copy(deep=True) + duplicated_event = target_session.events[-1].model_copy(deep=True) duplicated_event.id = f"{duplicated_event.id}-duplicate" duplicated_event.invocation_id = f"{duplicated_event.invocation_id}-duplicate" duplicated_event.timestamp += 0.0001 - self.session.events.append(duplicated_event) - await self.session_service.update_session(self.session) + target_session.events.append(duplicated_event) + await self.session_service.update_session(target_session) return if fault.operation == RuntimeFaultOperation.DROP_LAST_EVENT_KEEP_STATE: - if not self.session.events: + if not target_session.events: raise RuntimeError("Cannot drop the last event of an empty session.") - self.session.events.pop() - await self.session_service.update_session(self.session) + target_session.events.pop() + await self.session_service.update_session(target_session) return if fault.operation == RuntimeFaultOperation.SET_SESSION_VALUE: if not fault.path: raise ValueError("SET_SESSION_VALUE fault requires a target path.") - _set_path_value(self.session, fault.path, fault.value) - await self.session_service.update_session(self.session) + _set_path_value(target_session, fault.path, fault.value) + await self.session_service.update_session(target_session) return manager = getattr(self.session_service, "summarizer_manager", None) @@ -416,32 +423,65 @@ async def _apply_runtime_fault(self, fault: RuntimeFault) -> None: if fault.operation == RuntimeFaultOperation.DELETE_SUMMARY: summary_cache = getattr(manager, "_summarizer_cache", {}) - summary_cache.get(self.session.app_name, {}).get(self.session.user_id, {}).pop(self.session.id, None) - summary_event = _get_summary_event(self.session) + summary_cache.get(target_session.app_name, {}).get(target_session.user_id, {}).pop(target_session.id, None) + summary_event = _get_summary_event(target_session) if summary_event is not None: custom_metadata = dict(summary_event.custom_metadata or {}) custom_metadata.pop(_SUMMARY_EVENT_METADATA_KEY, None) summary_event.custom_metadata = custom_metadata or None - await self.session_service.update_session(self.session) + await self.session_service.update_session(target_session) return if fault.operation == RuntimeFaultOperation.SET_SUMMARY_VALUE: if not fault.path: raise ValueError("SET_SUMMARY_VALUE fault requires a target path.") - summary = await manager.get_session_summary(self.session) + summary = await manager.get_session_summary(target_session) if summary is None: raise RuntimeError("Cannot mutate summary value because no cached summary exists.") _set_path_value(summary, fault.path, fault.value) - summary_event = _get_summary_event(self.session) + summary_event = _get_summary_event(target_session) if summary_event is not None and summary_event.custom_metadata: persisted_summary = summary_event.custom_metadata.get(_SUMMARY_EVENT_METADATA_KEY) if isinstance(persisted_summary, dict): _set_path_value(persisted_summary, fault.path, fault.value) - await self.session_service.update_session(self.session) + await self.session_service.update_session(target_session) return raise ValueError(f"Unsupported runtime fault operation: {fault.operation}") + def _make_session_identity(self, session: Session) -> dict[str, str]: + return { + "app_name": session.app_name, + "user_id": session.user_id, + "session_id": session.id, + } + + def _memory_observation_key(self, step_index: int, step: ReplayStep) -> str: + if step.memory_query is None: + raise ValueError("Memory observation key requires a query specification.") + return f"step_{step_index:03d}:{step.session_alias}:{step.memory_query.name}" + + def _get_fault_session(self, fault: RuntimeFault) -> Session: + session = self._sessions.get(fault.session_alias) + if session is None: + raise RuntimeError(f"Runtime fault session alias '{fault.session_alias}' has not been created.") + return session + + async def _build_session_snapshot(self, session_alias: str, session: Session) -> SessionSnapshot: + return SessionSnapshot( + session_alias=session_alias, + app_name=session.app_name, + user_id=session.user_id, + session_id=session.id, + session={ + "conversation_count": session.conversation_count, + "events": [_event_to_snapshot(event) for event in session.events], + "historical_events": [_event_to_snapshot(event) for event in session.historical_events], + }, + state=dict(session.state), + summary=await self._get_summary_snapshot(session), + ) + @abstractmethod async def _build_session_service(self) -> BaseSessionService: """Create the session service used by this adapter.""" @@ -544,6 +584,7 @@ def __init__(self, redis_url: str) -> None: super().__init__() self._redis_url = redis_url self._logical_case: ReplayCase | None = None + self._logical_session_identities: dict[str, dict[str, str]] = {} self._storage_namespace = f"replay-{uuid.uuid4().hex[:8]}" @property @@ -554,6 +595,7 @@ def logical_case(self) -> ReplayCase: async def setup(self, case: ReplayCase) -> None: self._logical_case = case + self._logical_session_identities = _collect_logical_session_identities(case) await super().setup(self._namespaced_case(case)) def should_restart_before_snapshot(self) -> bool: @@ -564,21 +606,26 @@ def should_restart_during_replay(self) -> bool: async def run_case(self, case: ReplayCase) -> BackendSnapshot: snapshot = await super().run_case(self.case) - summary = snapshot.summary - if summary is not None: - summary = replace( - summary, - session_id=self.logical_case.session_id, - summary_id=self._project_identifier(summary.summary_id), - replaces=self._project_identifier(summary.replaces), - metadata=self._project_scalar(summary.metadata), - ) + sessions_by_alias = { + alias: self._project_session_snapshot(alias, session_snapshot) + for alias, session_snapshot in snapshot.sessions_by_alias.items() + } + active_snapshot = sessions_by_alias[snapshot.active_session_alias] + memory = { + key: self._project_memory_observation(value) + for key, value in snapshot.memory.items() + } + summary = active_snapshot.summary return replace( snapshot, - app_name=self.logical_case.app_name, - user_id=self.logical_case.user_id, - session_id=self.logical_case.session_id, + app_name=active_snapshot.app_name, + user_id=active_snapshot.user_id, + session_id=active_snapshot.session_id, + session=active_snapshot.session, + state=active_snapshot.state, + memory=memory, summary=summary, + sessions_by_alias=sessions_by_alias, ) async def _build_session_service(self) -> BaseSessionService: @@ -607,11 +654,14 @@ async def _build_memory_service(self) -> BaseMemoryService: async def close(self) -> None: await super().close() self._logical_case = None + self._logical_session_identities = {} def _namespaced_case(self, case: ReplayCase) -> ReplayCase: namespaced_steps = tuple( replace( step, + app_name=(f"{step.app_name}:{self._storage_namespace}" if step.app_name is not None else None), + user_id=(f"{step.user_id}:{self._storage_namespace}" if step.user_id is not None else None), session_id=(f"{step.session_id}:{self._storage_namespace}" if step.session_id is not None else None), ) for step in case.steps @@ -627,7 +677,7 @@ def _namespaced_case(self, case: ReplayCase) -> ReplayCase: def _project_identifier(self, value: Optional[str]) -> Optional[str]: if value is None: return None - return value.replace(self.case.session_id, self.logical_case.session_id) + return self._project_scalar(value) def _project_scalar(self, value: Any) -> Any: if isinstance(value, dict): @@ -637,9 +687,48 @@ def _project_scalar(self, value: Any) -> Any: if isinstance(value, tuple): return [self._project_scalar(item) for item in value] if isinstance(value, str): - return self._project_identifier(value) + for logical_identity in self._logical_session_identities.values(): + value = value.replace(f"{logical_identity['app_name']}:{self._storage_namespace}", logical_identity["app_name"]) + value = value.replace(f"{logical_identity['user_id']}:{self._storage_namespace}", logical_identity["user_id"]) + value = value.replace(f"{logical_identity['session_id']}:{self._storage_namespace}", logical_identity["session_id"]) + value = value.replace(self.case.app_name, self.logical_case.app_name) + value = value.replace(self.case.user_id, self.logical_case.user_id) + value = value.replace(self.case.session_id, self.logical_case.session_id) + return value return value + def _project_session_snapshot(self, session_alias: str, snapshot: SessionSnapshot) -> SessionSnapshot: + logical_identity = self._logical_session_identities.get(session_alias) + projected_summary = snapshot.summary + if projected_summary is not None: + projected_summary = replace( + projected_summary, + session_id=logical_identity["session_id"] if logical_identity is not None else self._project_scalar(projected_summary.session_id), + summary_id=self._project_identifier(projected_summary.summary_id), + replaces=self._project_identifier(projected_summary.replaces), + metadata=self._project_scalar(projected_summary.metadata), + ) + return replace( + snapshot, + app_name=logical_identity["app_name"] if logical_identity is not None else self._project_scalar(snapshot.app_name), + user_id=logical_identity["user_id"] if logical_identity is not None else self._project_scalar(snapshot.user_id), + session_id=logical_identity["session_id"] if logical_identity is not None else self._project_scalar(snapshot.session_id), + state=self._project_scalar(snapshot.state), + summary=projected_summary, + ) + + def _project_memory_observation(self, observation: Any) -> Any: + if not isinstance(observation, dict): + return self._project_scalar(observation) + projected = self._project_scalar(observation) + session_alias = projected.get("session_alias") + logical_identity = self._logical_session_identities.get(session_alias) + if logical_identity is not None: + projected["app_name"] = logical_identity["app_name"] + projected["user_id"] = logical_identity["user_id"] + projected["session_id"] = logical_identity["session_id"] + return projected + def get_runtime_metadata(self) -> dict[str, Any]: return { "backend_name": self.name, @@ -759,10 +848,12 @@ def normalize_backend_snapshot(snapshot: BackendSnapshot) -> dict[str, Any]: "app_name": snapshot.app_name, "user_id": snapshot.user_id, "session_id": snapshot.session_id, + "active_session_alias": snapshot.active_session_alias, "session": _normalize_scalar(snapshot.session), "state": _normalize_scalar(snapshot.state), "memory": _normalize_memory(snapshot.memory), "summary": _normalize_summary(snapshot.summary), + "sessions_by_alias": _normalize_sessions_by_alias(snapshot.sessions_by_alias, snapshot.active_session_alias), } @@ -785,7 +876,20 @@ def _normalize_summary(summary: Optional[SummarySnapshot]) -> Optional[dict[str, def _normalize_memory(memory_results: dict[str, Any]) -> dict[str, Any]: normalized: dict[str, Any] = {} - for query_name, entries in memory_results.items(): + for observation_key, observation in memory_results.items(): + if isinstance(observation, dict): + entries = observation.get("entries", []) + normalized_observation = { + "query_name": observation.get("query_name"), + "session_alias": observation.get("session_alias"), + "app_name": observation.get("app_name"), + "user_id": observation.get("user_id"), + "session_id": observation.get("session_id"), + "step_index": observation.get("step_index"), + } + else: + entries = observation + normalized_observation = {} normalized_entries = [] for entry in entries: normalized_entries.append( @@ -794,10 +898,31 @@ def _normalize_memory(memory_results: dict[str, Any]) -> dict[str, Any]: "role": entry.get("role"), "text": (entry.get("text") or "").strip(), }) - normalized[query_name] = sorted( + normalized_observation["entries"] = sorted( normalized_entries, key=lambda item: (item["text"], item["author"] or "", item["role"] or ""), ) + normalized[observation_key] = _normalize_scalar(normalized_observation) + return normalized + + +def _normalize_sessions_by_alias( + sessions_by_alias: dict[str, SessionSnapshot], + active_session_alias: str, +) -> dict[str, Any]: + normalized: dict[str, Any] = {} + for session_alias, snapshot in sessions_by_alias.items(): + if session_alias == active_session_alias: + continue + normalized[session_alias] = { + "session_alias": snapshot.session_alias, + "app_name": snapshot.app_name, + "user_id": snapshot.user_id, + "session_id": snapshot.session_id, + "session": _normalize_scalar(snapshot.session), + "state": _normalize_scalar(snapshot.state), + "summary": _normalize_summary(snapshot.summary), + } return normalized @@ -859,6 +984,19 @@ def _summarize_connection_url(url: str) -> dict[str, Any]: } +def _collect_logical_session_identities(case: ReplayCase) -> dict[str, dict[str, str]]: + identities: dict[str, dict[str, str]] = {} + for step in case.steps: + if step.kind != ReplayStepKind.CREATE_SESSION: + continue + identities[step.session_alias] = { + "app_name": step.app_name or case.app_name, + "user_id": step.user_id or case.user_id, + "session_id": step.session_id or case.session_id, + } + return identities + + def _backend_target_matches(target_name: str, backend_name: str) -> bool: normalized_target = target_name.strip().lower() normalized_backend = backend_name.strip().lower() @@ -916,6 +1054,26 @@ def diff_backend_snapshots( session_id=_resolve_session_id(left_view, right_view, case.session_id), summary_id=scope_summary_id, ) + left_aliases = left_view.get("sessions_by_alias", {}) + right_aliases = right_view.get("sessions_by_alias", {}) + for session_alias in sorted(set(left_aliases) | set(right_aliases)): + left_alias_snapshot = left_aliases.get(session_alias) + right_alias_snapshot = right_aliases.get(session_alias) + _diff_values( + case=case, + backend_a=left.backend_name, + backend_b=right.backend_name, + scope="sessions_by_alias", + path=f"sessions_by_alias.{session_alias}", + left=left_alias_snapshot, + right=right_alias_snapshot, + out=diffs, + session_id=_resolve_value_session_id(left_alias_snapshot, right_alias_snapshot, case.session_id), + summary_id=_resolve_summary_id( + left_alias_snapshot.get("summary") if isinstance(left_alias_snapshot, dict) else None, + right_alias_snapshot.get("summary") if isinstance(right_alias_snapshot, dict) else None, + ), + ) return diffs @@ -1124,6 +1282,15 @@ def _resolve_session_id(left: dict[str, Any], right: dict[str, Any], fallback: s return fallback +def _resolve_value_session_id(left: Any, right: Any, fallback: str) -> str: + for value in (left, right): + if isinstance(value, dict): + session_id = value.get("session_id") + if isinstance(session_id, str): + return session_id + return fallback + + def _deterministic_time_base(case_id: str) -> float: if case_id not in _CASE_TIME_BASES: digest = hashlib.sha1(case_id.encode("utf-8")).digest() diff --git a/tests/sessions/replay_models.py b/tests/sessions/replay_models.py index 9a590f63..f8e18a8c 100644 --- a/tests/sessions/replay_models.py +++ b/tests/sessions/replay_models.py @@ -99,6 +99,8 @@ class ReplayStep: force_summary: bool = False session_alias: str = "default" session_id: Optional[str] = None + app_name: Optional[str] = None + user_id: Optional[str] = None @classmethod def create_session( @@ -107,12 +109,16 @@ def create_session( initial_state: Optional[dict[str, Any]] = None, session_alias: str = "default", session_id: Optional[str] = None, + app_name: Optional[str] = None, + user_id: Optional[str] = None, ) -> "ReplayStep": return cls( kind=ReplayStepKind.CREATE_SESSION, initial_state=initial_state or {}, session_alias=session_alias, session_id=session_id, + app_name=app_name, + user_id=user_id, ) @classmethod @@ -197,6 +203,7 @@ class RuntimeFault: backend_name: str after_step: int operation: RuntimeFaultOperation + session_alias: str = "default" path: Optional[str] = None value: Any = None @@ -220,6 +227,25 @@ def to_dict(self) -> dict[str, Any]: return asdict(self) +@dataclass(frozen=True) +class SessionSnapshot: + """Comparable view of one resolved session alias.""" + + session_alias: str + app_name: str + user_id: str + session_id: str + session: dict[str, Any] + state: dict[str, Any] + summary: Optional[SummarySnapshot] + + def to_dict(self) -> dict[str, Any]: + data = asdict(self) + if self.summary is not None: + data["summary"] = self.summary.to_dict() + return data + + @dataclass(frozen=True) class BackendSnapshot: """Full business snapshot collected after replaying one case.""" @@ -229,15 +255,21 @@ class BackendSnapshot: app_name: str user_id: str session_id: str + active_session_alias: str session: dict[str, Any] state: dict[str, Any] memory: dict[str, Any] summary: Optional[SummarySnapshot] + sessions_by_alias: dict[str, SessionSnapshot] = field(default_factory=dict) def to_dict(self) -> dict[str, Any]: data = asdict(self) if self.summary is not None: data["summary"] = self.summary.to_dict() + data["sessions_by_alias"] = { + alias: snapshot.to_dict() + for alias, snapshot in self.sessions_by_alias.items() + } return data diff --git a/tests/sessions/test_replay_consistency.py b/tests/sessions/test_replay_consistency.py index fc40b01c..d6b07e56 100644 --- a/tests/sessions/test_replay_consistency.py +++ b/tests/sessions/test_replay_consistency.py @@ -13,6 +13,7 @@ from .replay_cases import REPLAY_ACCEPTANCE_CASES from .replay_cases import REPLAY_ALL_CASES from .replay_cases import REPLAY_EXTRA_CASES +from .replay_cases import REPLAY_TARGETED_CASES from .replay_harness import DEFAULT_REPORT_PATH from .replay_harness import build_case_matrix_report from .replay_harness import InMemoryReplayAdapter @@ -38,6 +39,13 @@ AdapterFactory = Callable[[], ReplayBackendAdapter] +def _find_case(case_id: str) -> ReplayCase: + for case in REPLAY_ALL_CASES + REPLAY_TARGETED_CASES: + if case.case_id == case_id: + return case + raise KeyError(f"Unknown replay case: {case_id}") + + async def _run_case_on_backend( adapter_factory: AdapterFactory, case: ReplayCase, @@ -183,6 +191,59 @@ def test_acceptance_case_count() -> None: assert len(REPLAY_EXTRA_CASES) >= 1 +def test_replay_harness_collects_all_session_alias_snapshots() -> None: + """Non-active sessions should remain visible in the final snapshot.""" + + snapshot, _, _ = asyncio.run( + _run_case_on_backend(InMemoryReplayAdapter, _find_case("cross_session_memory_aggregation")) + ) + + assert snapshot.active_session_alias == "default" + assert set(snapshot.sessions_by_alias) == {"source", "default"} + assert snapshot.sessions_by_alias["source"].session_id == "replay-memory-source" + assert snapshot.sessions_by_alias["source"].session["events"][0]["text"] == "Please remember that I prefer oolong tea." + assert snapshot.sessions_by_alias["default"].session_id == "replay-memory-target" + + +def test_replay_harness_preserves_memory_query_observations_across_restart() -> None: + """Repeated query names should preserve separate observations before and after restart.""" + + snapshot, _, _ = asyncio.run( + _run_case_on_backend(SqliteReplayAdapter, _find_case("memory_query_observation_survives_restart")) + ) + + observations = sorted(snapshot.memory.values(), key=lambda item: item["step_index"]) + assert [item["query_name"] for item in observations] == ["tea_preference", "tea_preference"] + assert [item["session_alias"] for item in observations] == ["default", "default"] + assert len(observations) == 2 + first_texts = {entry["text"] for entry in observations[0]["entries"]} + second_texts = {entry["text"] for entry in observations[1]["entries"]} + assert "Please remember that my favorite tea is oolong." in first_texts + assert "I will remember your oolong preference." in first_texts + assert "Also remember that I enjoy jasmine tea." in second_texts + assert "I will remember the jasmine preference too." in second_texts + assert "Also remember that I enjoy jasmine tea." not in first_texts + + +def test_replay_harness_keeps_duplicate_query_names_per_session_alias() -> None: + """Query names may repeat across aliases without overwriting previous observations.""" + + snapshot, _, _ = asyncio.run( + _run_case_on_backend(SqliteReplayAdapter, _find_case("duplicate_memory_query_name_across_sessions")) + ) + + observations = sorted(snapshot.memory.values(), key=lambda item: item["step_index"]) + assert len(observations) == 2 + assert [item["query_name"] for item in observations] == ["shared_preference_search", "shared_preference_search"] + assert [item["session_alias"] for item in observations] == ["source", "default"] + assert observations[0]["step_index"] < observations[1]["step_index"] + assert observations[0]["entries"] != observations[1]["entries"] + first_texts = {entry["text"] for entry in observations[0]["entries"]} + second_texts = {entry["text"] for entry in observations[1]["entries"]} + assert any("dragon well" in text.lower() for text in first_texts) + assert any("dragon well" in text.lower() for text in second_texts) + + def test_replay_consistency_redis_integration_mode() -> None: """Run an optional Redis-backed integration comparison when configured.""" From b774b2e41964b95b2c19e3941e1ec3a5065f07bc Mon Sep 17 00:00:00 2001 From: Cai Zichen <3234912331@qq.com> Date: Sat, 11 Jul 2026 20:15:21 +0800 Subject: [PATCH 3/4] style(sessions): fix summarizer manager formatting --- trpc_agent_sdk/sessions/_summarizer_manager.py | 1 - 1 file changed, 1 deletion(-) diff --git a/trpc_agent_sdk/sessions/_summarizer_manager.py b/trpc_agent_sdk/sessions/_summarizer_manager.py index 1226cdff..07e68d84 100644 --- a/trpc_agent_sdk/sessions/_summarizer_manager.py +++ b/trpc_agent_sdk/sessions/_summarizer_manager.py @@ -41,7 +41,6 @@ from ._session_summarizer import SessionSummarizer from ._session_summarizer import SessionSummary - _SUMMARY_EVENT_METADATA_KEY = "session_summary" From 901555853b4485971eaed27a919b2dca6b640591 Mon Sep 17 00:00:00 2001 From: Cai Zichen <3234912331@qq.com> Date: Sat, 11 Jul 2026 20:28:31 +0800 Subject: [PATCH 4/4] test(redis): support hset mapping in redis session mocks --- tests/sessions/test_redis_session_service.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/tests/sessions/test_redis_session_service.py b/tests/sessions/test_redis_session_service.py index 8269b186..9aaace3c 100644 --- a/tests/sessions/test_redis_session_service.py +++ b/tests/sessions/test_redis_session_service.py @@ -83,9 +83,13 @@ async def execute_command(self, session, command): return [k for k in self._store.keys() if k.startswith(prefix)] elif method == 'hset': key = args[0] - pairs = args[1:] if key not in self._hash_store: self._hash_store[key] = {} + mapping = getattr(command, "kwargs", {}).get("mapping", {}) + if mapping: + self._hash_store[key].update(mapping) + return True + pairs = args[1:] for i in range(0, len(pairs), 2): self._hash_store[key][pairs[i]] = pairs[i + 1] return True