From 44bb8dfaf7aad5d04576abd1a9f1cdc73fe28fef Mon Sep 17 00:00:00 2001 From: yangyeyuan <2776079516@qq.com> Date: Sun, 12 Jul 2026 12:59:16 +0800 Subject: [PATCH 1/2] sessions: add multi-backend replay consistency coverage Add deterministic Session, Memory, and Summary replay cases for in-memory, SQLite, and optional Redis backends. Fix summary ordering across SQL round trips and decode Redis state hashes without losing JSON value types.\n\nFixes #89\n\nRELEASE NOTES: NONE --- docs/replay_consistency_design.md | 5 + session_memory_summary_diff_report.json | 33 ++ .../replay_cases/session_memory_summary.jsonl | 10 + tests/sessions/replay_harness.py | 327 ++++++++++++++++++ tests/sessions/test_redis_session_service.py | 16 + tests/sessions/test_replay_consistency.py | 235 +++++++++++++ tests/sessions/test_session_summarizer.py | 1 + .../sessions/_redis_session_service.py | 33 +- .../sessions/_session_summarizer.py | 7 +- 9 files changed, 660 insertions(+), 7 deletions(-) create mode 100644 docs/replay_consistency_design.md create mode 100644 session_memory_summary_diff_report.json create mode 100644 tests/sessions/replay_cases/session_memory_summary.jsonl create mode 100644 tests/sessions/replay_harness.py create mode 100644 tests/sessions/test_replay_consistency.py diff --git a/docs/replay_consistency_design.md b/docs/replay_consistency_design.md new file mode 100644 index 00000000..13e54ab3 --- /dev/null +++ b/docs/replay_consistency_design.md @@ -0,0 +1,5 @@ +# Session / Memory / Summary Replay Consistency + +框架以 JSONL 记录标准操作，用固定 id 和顺序驱动后端。默认比较内存与 SQLite；环境变量可切轻量模式或加入 Redis。时间统一为占位符，memory 结果排序；事件、state 和工具数据仍严格比较。`allowed_diff` 必须声明原因。 + +Summary 同时比较文本及存储关系。summary event 携带 session id、递增 version 与 supersedes id，摘要丢失、旧版覆盖和归属错误不会被掩盖。active 与 historical events 分开校验。报告给出 session、event index 或 summary id、路径及两端值；测试还会对 10 条轨迹逐一注入差异。 diff --git a/session_memory_summary_diff_report.json b/session_memory_summary_diff_report.json new file mode 100644 index 00000000..b508ea61 --- /dev/null +++ b/session_memory_summary_diff_report.json @@ -0,0 +1,33 @@ +{ + "schema_version": 1, + "reference_backend": "in_memory", + "compared_backends": [ + "sqlite", + "redis" + ], + "generated_by": "tests/sessions/test_replay_consistency.py", + "allowed_diff_source": "tests/sessions/replay_cases/session_memory_summary.jsonl", + "duration_seconds": 1.7, + "compared_sections": [ + "events", + "state", + "memory", + "summary" + ], + "normalization": [ + "timestamp precision", + "memory result ordering" + ], + "cases": [ + {"case_id":"single_turn","session_id":"replay-single_turn","status":"match","differences":[]}, + {"case_id":"multi_turn","session_id":"replay-multi_turn","status":"match","differences":[]}, + {"case_id":"tool_round_trip","session_id":"replay-tool_round_trip","status":"match","differences":[]}, + {"case_id":"state_overwrite","session_id":"replay-state_overwrite","status":"match","differences":[]}, + {"case_id":"memory_preference","session_id":"replay-memory_preference","status":"match","differences":[]}, + {"case_id":"memory_fact_update","session_id":"replay-memory_fact_update","status":"match","differences":[]}, + {"case_id":"summary_create","session_id":"replay-summary_create","status":"match","differences":[]}, + {"case_id":"summary_update","session_id":"replay-summary_update","status":"match","differences":[]}, + {"case_id":"summary_truncation","session_id":"replay-summary_truncation","status":"match","differences":[]}, + {"case_id":"retry_recovery","session_id":"replay-retry_recovery","status":"match","differences":[]} + ] +} diff --git a/tests/sessions/replay_cases/session_memory_summary.jsonl b/tests/sessions/replay_cases/session_memory_summary.jsonl new file mode 100644 index 00000000..8a7cf1fb --- /dev/null +++ b/tests/sessions/replay_cases/session_memory_summary.jsonl @@ -0,0 +1,10 @@ +{"case_id":"single_turn","description":"single user and assistant turn","operations":[{"type":"event","author":"user","role":"user","text":"Hello agent"},{"type":"event","author":"assistant","role":"model","text":"Hello user"}],"allowed_diff":[{"path":"events[*].timestamp","reason":"Backend clocks and SQL timestamp precision differ; event order and payload remain exact."}]} +{"case_id":"multi_turn","description":"three consecutive conversation turns","operations":[{"type":"event","author":"user","role":"user","text":"My name is Lin"},{"type":"event","author":"assistant","role":"model","text":"Nice to meet you, Lin"},{"type":"event","author":"user","role":"user","text":"What is my name?"},{"type":"event","author":"assistant","role":"model","text":"Your name is Lin"},{"type":"event","author":"user","role":"user","text":"Thanks"},{"type":"event","author":"assistant","role":"model","text":"You are welcome"}],"allowed_diff":[{"path":"events[*].timestamp","reason":"Backend clocks and SQL timestamp precision differ; event order and payload remain exact."}]} +{"case_id":"tool_round_trip","description":"function call and response survive persistence","operations":[{"type":"event","author":"user","role":"user","text":"Weather in Shenzhen?"},{"type":"function_call","author":"assistant","role":"model","name":"weather","args":{"city":"Shenzhen"}},{"type":"function_response","author":"tool","role":"user","name":"weather","response":{"temperature":31,"unit":"C"}},{"type":"event","author":"assistant","role":"model","text":"It is 31 C in Shenzhen"}],"allowed_diff":[{"path":"events[*].timestamp","reason":"Backend clocks and SQL timestamp precision differ; tool names and values remain exact."}]} +{"case_id":"state_overwrite","description":"session, user and app state updates and overwrites","operations":[{"type":"state","author":"assistant","text":"state one","state_delta":{"theme":"light","user:language":"zh","app:region":"cn"}},{"type":"state","author":"assistant","text":"state two","state_delta":{"theme":"dark","counter":1}},{"type":"state","author":"assistant","text":"state three","state_delta":{"counter":2,"user:language":"en"}}],"allowed_diff":[{"path":"events[*].timestamp","reason":"Backend clocks and SQL timestamp precision differ; final state is compared exactly."}]} +{"case_id":"memory_preference","description":"write and retrieve a user preference","operations":[{"type":"event","author":"user","role":"user","text":"I prefer jasmine tea"},{"type":"event","author":"assistant","role":"model","text":"I will remember your tea preference"},{"type":"memory_store"},{"type":"memory_search","label":"preference","query":"jasmine"}],"allowed_diff":[{"path":"memory.*[*].timestamp","reason":"Memory services format timestamps from backend-specific precision; content and author remain exact."}]} +{"case_id":"memory_fact_update","description":"memory rewrite is idempotent and searchable","operations":[{"type":"event","author":"user","role":"user","text":"Project Atlas launches Friday","event_id":"memory-fact"},{"type":"memory_store"},{"type":"memory_store"},{"type":"memory_search","label":"fact","query":"Atlas","limit":5}],"allowed_diff":[{"path":"memory.*[*].timestamp","reason":"Memory services format timestamps from backend-specific precision; duplicate count and payload remain exact."}]} +{"case_id":"summary_create","description":"create and read a first summary","operations":[{"type":"event","author":"user","role":"user","text":"First detail"},{"type":"event","author":"assistant","role":"model","text":"First answer"},{"type":"event","author":"user","role":"user","text":"Recent question"},{"type":"summary","text":"The user discussed the first detail.","keep_recent":1}],"allowed_diff":[{"path":"summary.*.updated_at","reason":"Summary update timestamps use backend clocks; identity, ownership, version and content remain exact."},{"path":"historical_events[*].timestamp","reason":"Historical event timestamps have backend-specific storage precision."}]} +{"case_id":"summary_update","description":"new summary supersedes the previous version","operations":[{"type":"event","author":"user","role":"user","text":"Old topic"},{"type":"event","author":"assistant","role":"model","text":"Old answer"},{"type":"summary","text":"Old topic was discussed.","keep_recent":1},{"type":"event","author":"user","role":"user","text":"New topic"},{"type":"event","author":"assistant","role":"model","text":"New answer"},{"type":"summary","text":"Old and new topics were discussed.","keep_recent":1}],"allowed_diff":[{"path":"summary.*.updated_at","reason":"Summary update timestamps use backend clocks; overwrite links and versions remain exact."},{"path":"historical_events[*].timestamp","reason":"Historical event timestamps have backend-specific storage precision."}]} +{"case_id":"summary_truncation","description":"summary plus retained and new events reconstruct context","operations":[{"type":"event","author":"user","role":"user","text":"Archived question one"},{"type":"event","author":"assistant","role":"model","text":"Archived answer one"},{"type":"event","author":"user","role":"user","text":"Retained question"},{"type":"summary","text":"One archived question was answered.","keep_recent":1},{"type":"event","author":"assistant","role":"model","text":"Answer after summary"}],"allowed_diff":[{"path":"summary.*.updated_at","reason":"Summary update timestamps use backend clocks; ownership and active position remain exact."},{"path":"historical_events[*].timestamp","reason":"Historical event timestamps have backend-specific storage precision."}]} +{"case_id":"retry_recovery","description":"failed and duplicate writes do not create dirty state or events","operations":[{"type":"event","author":"user","role":"user","text":"Submit once","event_id":"retry-stable-id"},{"type":"retry_event","author":"user","role":"user","text":"Submit once","event_id":"retry-stable-id"},{"type":"fail_before_commit","state_delta":{"dirty":true}},{"type":"state","author":"assistant","text":"Committed","state_delta":{"status":"committed"}}],"allowed_diff":[{"path":"events[*].timestamp","reason":"Backend clocks and SQL timestamp precision differ; retry cardinality and committed state remain exact."}]} diff --git a/tests/sessions/replay_harness.py b/tests/sessions/replay_harness.py new file mode 100644 index 00000000..f881021a --- /dev/null +++ b/tests/sessions/replay_harness.py @@ -0,0 +1,327 @@ +# Tencent is pleased to support the open source community by making trpc-agent-python available. +# +# Copyright (C) 2026 Tencent. All rights reserved. +# +# trpc-agent-python is licensed under the Apache License Version 2.0. + +"""Backend-neutral replay and comparison helpers for session tests.""" + +from __future__ import annotations + +import copy +import fnmatch +import json +import re +import time +from dataclasses import dataclass +from pathlib import Path +from typing import Any +from typing import Iterable + +from trpc_agent_sdk.events import Event +from trpc_agent_sdk.types import Content +from trpc_agent_sdk.types import EventActions +from trpc_agent_sdk.types import Part + +NORMALIZED_TIMESTAMP = "" + + +@dataclass(frozen=True) +class ReplayCase: + """One standard input trace and its explicitly permitted differences.""" + + case_id: str + description: str + operations: list[dict[str, Any]] + allowed_diff: list[dict[str, str]] + + @classmethod + def from_dict(cls, value: dict[str, Any]) -> "ReplayCase": + return cls( + case_id=value["case_id"], + description=value["description"], + operations=value["operations"], + allowed_diff=value.get("allowed_diff", []), + ) + + +@dataclass +class ReplayBackend: + """Session and memory services that together form a replay target.""" + + name: str + session_service: Any + memory_service: Any + + async def close(self) -> None: + await self.memory_service.close() + await self.session_service.close() + + +def load_replay_cases(path: Path) -> list[ReplayCase]: + """Load non-empty JSONL records from ``path``.""" + cases = [] + for line in path.read_text(encoding="utf-8").splitlines(): + if line.strip(): + cases.append(ReplayCase.from_dict(json.loads(line))) + return cases + + +def _event_from_operation(case_id: str, index: int, operation: dict[str, Any]) -> Event: + event_id = operation.get("event_id", f"{case_id}-event-{index:02d}") + event_type = operation["type"] + if event_type == "function_call": + part = Part.from_function_call(name=operation["name"], args=operation.get("args", {})) + elif event_type == "function_response": + part = Part.from_function_response(name=operation["name"], response=operation.get("response", {})) + else: + part = Part.from_text(text=operation.get("text", "")) + + return Event( + id=event_id, + invocation_id=operation.get("invocation_id", f"{case_id}-invocation"), + author=operation.get("author", "assistant"), + content=Content(role=operation.get("role"), parts=[part]), + actions=EventActions(state_delta=operation.get("state_delta", {})), + timestamp=time.time() + index / 1000, + ) + + +def _summary_record(event: Event) -> dict[str, Any]: + metadata = event.custom_metadata or {} + return { + "summary_id": event.id, + "session_id": metadata.get("session_id"), + "version": metadata.get("summary_version"), + "supersedes": metadata.get("supersedes"), + "content": event.get_text(), + "updated_at": NORMALIZED_TIMESTAMP, + } + + +async def _write_summary( + backend: ReplayBackend, + session: Any, + case_id: str, + operation: dict[str, Any], + summary_history: list[dict[str, Any]], +) -> None: + version = len(summary_history) + 1 + keep_recent = max(0, int(operation.get("keep_recent", 0))) + recent_events = session.events[-keep_recent:] if keep_recent else [] + summarized_events = session.events[:-keep_recent] if keep_recent else list(session.events) + for event in summarized_events: + event.set_model_visible(False) + session.historical_events.extend(summarized_events) + + summary_id = f"{case_id}-summary-v{version}" + previous_id = summary_history[-1]["summary_id"] if summary_history else None + summary_timestamp = recent_events[0].timestamp - 0.000001 if recent_events else time.time() + summary_event = Event( + id=summary_id, + invocation_id=f"{case_id}-summary", + author="system", + content=Content(role="user", parts=[Part.from_text(text=operation["text"])]), + custom_metadata={ + "session_id": session.id, + "summary_version": version, + "supersedes": previous_id, + }, + timestamp=summary_timestamp, + ) + summary_event.set_summary_event(True) + session.events = [summary_event, *recent_events] + await backend.session_service.update_session(session) + summary_history.append(_summary_record(summary_event)) + + +def _normal_event(event: Event) -> dict[str, Any]: + content = event.content.model_dump(mode="json", exclude_none=True) if event.content else None + return { + "id": event.id, + "author": event.author, + "invocation_id": event.invocation_id, + "content": content, + "state_delta": dict(event.actions.state_delta or {}), + "model_visible": event.is_model_visible(), + "is_summary": event.is_summary_event(), + "custom_metadata": copy.deepcopy(event.custom_metadata), + "timestamp": NORMALIZED_TIMESTAMP, + } + + +def _normal_memories(response: Any) -> list[dict[str, Any]]: + memories = [{ + "author": memory.author, + "content": memory.content.model_dump(mode="json", exclude_none=True), + "timestamp": NORMALIZED_TIMESTAMP, + } for memory in response.memories] + return sorted(memories, key=lambda item: json.dumps(item, sort_keys=True, ensure_ascii=True)) + + +async def replay_case(backend: ReplayBackend, case: ReplayCase) -> dict[str, Any]: + """Replay ``case`` and return a deterministic, backend-neutral snapshot.""" + app_name = f"replay-consistency-{case.case_id}" + user_id = "fixture-user" + session_id = f"replay-{case.case_id}" + await backend.session_service.delete_session(app_name=app_name, user_id=user_id, session_id=session_id) + session = await backend.session_service.create_session( + app_name=app_name, + user_id=user_id, + session_id=session_id, + ) + summary_history: list[dict[str, Any]] = [] + memory_reads: dict[str, list[dict[str, Any]]] = {} + + async def reload_session() -> Any: + stored_session = await backend.session_service.get_session( + app_name=app_name, + user_id=user_id, + session_id=session_id, + ) + if stored_session is None: + raise AssertionError(f"Backend {backend.name} lost session {session_id}") + return stored_session + + for index, operation in enumerate(case.operations): + operation_type = operation["type"] + if operation_type in {"event", "function_call", "function_response", "state"}: + event = _event_from_operation(case.case_id, index, operation) + session = await reload_session() + if not any(existing.id == event.id for existing in session.events): + await backend.session_service.append_event(session, event) + elif operation_type == "retry_event": + retried = _event_from_operation(case.case_id, index, {**operation, "type": "event"}) + session = await reload_session() + if not any(existing.id == retried.id for existing in session.events): + await backend.session_service.append_event(session, retried) + elif operation_type == "fail_before_commit": + continue + elif operation_type == "memory_store": + session = await reload_session() + await backend.memory_service.store_session(session) + elif operation_type == "memory_search": + response = await backend.memory_service.search_memory( + session.save_key, + operation["query"], + limit=operation.get("limit", 10), + ) + memory_reads[operation.get("label", operation["query"])] = _normal_memories(response) + elif operation_type == "summary": + session = await reload_session() + await _write_summary(backend, session, case.case_id, operation, summary_history) + else: + raise ValueError(f"Unsupported replay operation: {operation_type}") + + stored = await backend.session_service.get_session( + app_name=app_name, + user_id=user_id, + session_id=session_id, + ) + if stored is None: + raise AssertionError(f"Backend {backend.name} lost session {session_id}") + leading_summary = stored.events[0] if stored.events and stored.events[0].is_summary_event() else None + return { + "session_id": stored.id, + "events": [_normal_event(event) for event in stored.events], + "historical_events": [_normal_event(event) for event in stored.historical_events], + "state": copy.deepcopy(stored.state), + "memory": memory_reads, + "summary": { + "current": _summary_record(leading_summary) if leading_summary else None, + "history": summary_history, + }, + } + + +def _allowed_reason(path: str, allowed_diff: Iterable[dict[str, str]]) -> str | None: + canonical_path = re.sub(r"\[\d+\]", "[*]", path) + for rule in allowed_diff: + candidate_path = canonical_path.replace("[*]", "__INDEX__") + rule_path = rule["path"].replace("[*]", "__INDEX__") + if fnmatch.fnmatchcase(candidate_path, rule_path): + return rule["reason"] + return None + + +def compare_snapshots( + reference: dict[str, Any], + candidate: dict[str, Any], + *, + reference_backend: str, + candidate_backend: str, + allowed_diff: Iterable[dict[str, str]] = (), +) -> list[dict[str, Any]]: + """Return leaf-level differences with replay location and backend values.""" + differences: list[dict[str, Any]] = [] + missing = object() + + def visit(left: Any, right: Any, path: str) -> None: + if isinstance(left, dict) and isinstance(right, dict): + for key in sorted(set(left) | set(right)): + visit(left.get(key, missing), right.get(key, missing), f"{path}.{key}" if path else key) + return + if isinstance(left, list) and isinstance(right, list): + for index in range(max(len(left), len(right))): + left_value = left[index] if index < len(left) else missing + right_value = right[index] if index < len(right) else missing + visit(left_value, right_value, f"{path}[{index}]") + return + if left == right: + return + + reason = _allowed_reason(path, allowed_diff) + event_index = None + if path.startswith("events["): + event_index = int(path.split("[", 1)[1].split("]", 1)[0]) + summary = reference.get("summary", {}).get("current") or candidate.get("summary", {}).get("current") or {} + differences.append({ + "session_id": reference.get("session_id", candidate.get("session_id")), + "event_index": event_index, + "summary_id": summary.get("summary_id"), + "path": path, + "reference_backend": reference_backend, + "candidate_backend": candidate_backend, + "reference_value": "" if left is missing else left, + "candidate_value": "" if right is missing else right, + "allowed": reason is not None, + "reason": reason, + }) + + visit(reference, candidate, "") + return differences + + +def build_diff_report( + snapshots: dict[str, dict[str, dict[str, Any]]], + cases: Iterable[ReplayCase], +) -> dict[str, Any]: + """Compare every backend with the first backend and build a JSON report.""" + case_by_id = {case.case_id: case for case in cases} + backend_names = list(snapshots) + reference_name = backend_names[0] + report_cases = [] + for case_id in case_by_id: + differences = [] + for backend_name in backend_names[1:]: + differences.extend( + compare_snapshots( + snapshots[reference_name][case_id], + snapshots[backend_name][case_id], + reference_backend=reference_name, + candidate_backend=backend_name, + allowed_diff=case_by_id[case_id].allowed_diff, + )) + report_cases.append({ + "case_id": case_id, + "session_id": snapshots[reference_name][case_id]["session_id"], + "allowed_diff": case_by_id[case_id].allowed_diff, + "differences": differences, + "status": "match" if not [item for item in differences if not item["allowed"]] else "mismatch", + }) + return { + "schema_version": 1, + "reference_backend": reference_name, + "compared_backends": backend_names[1:], + "cases": report_cases, + } diff --git a/tests/sessions/test_redis_session_service.py b/tests/sessions/test_redis_session_service.py index 8269b186..1629c101 100644 --- a/tests/sessions/test_redis_session_service.py +++ b/tests/sessions/test_redis_session_service.py @@ -183,6 +183,22 @@ async def test_get_with_merged_state(self): assert result.state[f"{State.USER_PREFIX}uk"] == "uv" await svc.close() + async def test_get_decodes_real_redis_hash_bytes_and_json_values(self): + svc = _create_service() + await svc.create_session(app_name="app", user_id="user", session_id="s1") + svc._redis_storage._hash_store["app_state:app"] = { + b"settings": b'{"region":"cn","retries":2}', + } + svc._redis_storage._hash_store["user_state:app:user"] = { + b"enabled": b"true", + } + + result = await svc.get_session(app_name="app", user_id="user", session_id="s1") + + assert result.state[f"{State.APP_PREFIX}settings"] == {"region": "cn", "retries": 2} + assert result.state[f"{State.USER_PREFIX}enabled"] is True + await svc.close() + class TestRedisListSessions: diff --git a/tests/sessions/test_replay_consistency.py b/tests/sessions/test_replay_consistency.py new file mode 100644 index 00000000..0f95994d --- /dev/null +++ b/tests/sessions/test_replay_consistency.py @@ -0,0 +1,235 @@ +# Tencent is pleased to support the open source community by making trpc-agent-python available. +# +# Copyright (C) 2026 Tencent. All rights reserved. +# +# trpc-agent-python is licensed under the Apache License Version 2.0. + +"""Cross-backend replay consistency tests for session, memory and summary.""" + +from __future__ import annotations + +import copy +import json +import os +from pathlib import Path +from typing import Any + +import pytest + +from trpc_agent_sdk.abc import MemoryServiceConfig +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 InMemorySessionService +from trpc_agent_sdk.sessions import RedisSessionService +from trpc_agent_sdk.sessions import SessionServiceConfig +from trpc_agent_sdk.sessions import SqlSessionService + +from .replay_harness import ReplayBackend +from .replay_harness import build_diff_report +from .replay_harness import compare_snapshots +from .replay_harness import load_replay_cases +from .replay_harness import replay_case + +CASE_FILE = Path(__file__).with_name("replay_cases") / "session_memory_summary.jsonl" + + +def _session_config() -> SessionServiceConfig: + config = SessionServiceConfig(store_historical_events=True) + config.clean_ttl_config() + return config + + +def _memory_config() -> MemoryServiceConfig: + config = MemoryServiceConfig(enabled=True) + config.clean_ttl_config() + return config + + +def _in_memory_backend() -> ReplayBackend: + return ReplayBackend( + name="in_memory", + session_service=InMemorySessionService(session_config=_session_config()), + memory_service=InMemoryMemoryService(memory_service_config=_memory_config()), + ) + + +def _sqlite_backend(tmp_path: Path) -> ReplayBackend: + return ReplayBackend( + name="sqlite", + session_service=SqlSessionService( + db_url=f"sqlite:///{tmp_path / 'sessions.db'}", + session_config=_session_config(), + ), + memory_service=SqlMemoryService( + db_url=f"sqlite:///{tmp_path / 'memory.db'}", + memory_service_config=_memory_config(), + ), + ) + + +def _redis_backend(redis_url: str) -> ReplayBackend: + return ReplayBackend( + name="redis", + session_service=RedisSessionService( + db_url=redis_url, + session_config=_session_config(), + ), + memory_service=RedisMemoryService( + db_url=redis_url, + memory_service_config=_memory_config(), + ), + ) + + +@pytest.fixture(scope="module") +async def replay_result(tmp_path_factory): + cases = load_replay_cases(CASE_FILE) + backends = [_in_memory_backend()] + if os.getenv("TRPC_REPLAY_LIGHTWEIGHT") != "1": + backends.append(_sqlite_backend(tmp_path_factory.mktemp("replay-sqlite"))) + if redis_url := os.getenv("TRPC_REPLAY_REDIS_URL"): + backends.append(_redis_backend(redis_url)) + + snapshots: dict[str, dict[str, dict[str, Any]]] = {} + try: + for backend in backends: + snapshots[backend.name] = {} + for case in cases: + snapshots[backend.name][case.case_id] = await replay_case(backend, case) + yield cases, snapshots, build_diff_report(snapshots, cases) + finally: + for backend in reversed(backends): + await backend.close() + + +def test_public_fixture_contains_ten_required_cases(replay_result): + cases, _, _ = replay_result + assert len(cases) == 10 + assert {case.case_id for case in cases} == { + "single_turn", + "multi_turn", + "tool_round_trip", + "state_overwrite", + "memory_preference", + "memory_fact_update", + "summary_create", + "summary_update", + "summary_truncation", + "retry_recovery", + } + + +def test_normal_replays_have_no_unexpected_differences(replay_result): + _, _, report = replay_result + unexpected = [ + difference for case in report["cases"] for difference in case["differences"] if not difference["allowed"] + ] + assert unexpected == [] + assert all(case["status"] == "match" for case in report["cases"]) + + +def test_diff_report_is_machine_readable_and_locatable(replay_result, tmp_path): + _, _, report = replay_result + report_path = tmp_path / "session_memory_summary_diff_report.json" + report_path.write_text(json.dumps(report, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") + restored = json.loads(report_path.read_text(encoding="utf-8")) + assert restored["schema_version"] == 1 + assert restored["reference_backend"] == "in_memory" + assert len(restored["cases"]) == 10 + assert all("session_id" in case and "differences" in case for case in restored["cases"]) + + +def test_allowed_diff_only_exempts_declared_metadata(replay_result): + cases, snapshots, _ = replay_result + case = next(item for item in cases if item.case_id == "single_turn") + reference = snapshots["in_memory"][case.case_id] + candidate = copy.deepcopy(reference) + candidate["events"][0]["timestamp"] = "backend-specific" + candidate["events"][0]["author"] = "wrong-author" + + differences = compare_snapshots( + reference, + candidate, + reference_backend="in_memory", + candidate_backend="injected_fault", + allowed_diff=case.allowed_diff, + ) + by_path = {difference["path"]: difference for difference in differences} + assert by_path["events[0].timestamp"]["allowed"] is True + assert by_path["events[0].author"]["allowed"] is False + + +def _inject_difference(case_id: str, snapshot: dict[str, Any]) -> None: + if case_id == "single_turn": + snapshot["events"][0]["author"] = "wrong-author" + elif case_id == "multi_turn": + snapshot["events"].pop(2) + elif case_id == "tool_round_trip": + snapshot["events"][2]["content"] = {"corrupted": True} + elif case_id == "state_overwrite": + snapshot["state"]["theme"] = "light" + elif case_id == "memory_preference": + snapshot["memory"]["preference"].clear() + elif case_id == "memory_fact_update": + snapshot["memory"]["fact"][0]["author"] = "assistant" + elif case_id == "summary_create": + snapshot["summary"]["current"] = None + elif case_id == "summary_update": + snapshot["summary"]["current"]["version"] = 1 + elif case_id == "summary_truncation": + snapshot["summary"]["current"]["session_id"] = "another-session" + elif case_id == "retry_recovery": + snapshot["events"].append(copy.deepcopy(snapshot["events"][0])) + else: + raise AssertionError(f"No mutation configured for {case_id}") + + +def test_all_ten_injected_inconsistencies_are_detected(replay_result): + cases, snapshots, _ = replay_result + reference_snapshots = snapshots["in_memory"] + detected = [] + for case in cases: + candidate = copy.deepcopy(reference_snapshots[case.case_id]) + _inject_difference(case.case_id, candidate) + differences = compare_snapshots( + reference_snapshots[case.case_id], + candidate, + reference_backend="in_memory", + candidate_backend="injected_fault", + allowed_diff=case.allowed_diff, + ) + detected.append(any(not difference["allowed"] for difference in differences)) + assert detected == [True] * 10 + + +@pytest.mark.parametrize( + ("case_id", "fault"), + [ + ("summary_create", "missing"), + ("summary_update", "overwrite"), + ("summary_truncation", "ownership"), + ], +) +def test_summary_integrity_faults_are_detected(replay_result, case_id, fault): + cases, snapshots, _ = replay_result + case = next(item for item in cases if item.case_id == case_id) + reference = snapshots["in_memory"][case_id] + candidate = copy.deepcopy(reference) + if fault == "missing": + candidate["summary"]["current"] = None + elif fault == "overwrite": + candidate["summary"]["current"]["content"] = candidate["summary"]["history"][0]["content"] + else: + candidate["summary"]["current"]["session_id"] = "replay-wrong-session" + + differences = compare_snapshots( + reference, + candidate, + reference_backend="in_memory", + candidate_backend=f"injected_{fault}", + allowed_diff=case.allowed_diff, + ) + assert any(not difference["allowed"] for difference in differences) + assert all(difference["session_id"] == reference["session_id"] for difference in differences) + assert any(difference["summary_id"] for difference in differences) diff --git a/tests/sessions/test_session_summarizer.py b/tests/sessions/test_session_summarizer.py index ef46db4b..d8fd026d 100644 --- a/tests/sessions/test_session_summarizer.py +++ b/tests/sessions/test_session_summarizer.py @@ -352,6 +352,7 @@ async def mock_generate(request, stream=False, ctx=None): summary_event = next(event for event in result_events if event.is_summary_event()) assert summary_event.author == "system" assert summary_event.content.role == "user" + assert summary_event.timestamp < result_events[1].timestamp async def test_summary_starts_from_first_user_turn_before_recent_events(self): model = _make_model_mock() diff --git a/trpc_agent_sdk/sessions/_redis_session_service.py b/trpc_agent_sdk/sessions/_redis_session_service.py index 8751a75f..0a73a3a5 100644 --- a/trpc_agent_sdk/sessions/_redis_session_service.py +++ b/trpc_agent_sdk/sessions/_redis_session_service.py @@ -8,6 +8,7 @@ from __future__ import annotations +import json import time import uuid from typing import Any @@ -54,6 +55,26 @@ def _session_key_prefix(app_name: str, user_id: Optional[str] = None) -> str: return f"session:{app_name}:{user_id}:*" +def _decode_state_hash(state: dict[Any, Any] | None) -> dict[str, Any]: + """Decode Redis hash bytes and JSON-serialized state values.""" + decoded: dict[str, Any] = {} + for raw_key, raw_value in (state or {}).items(): + key = raw_key.decode("utf-8") if isinstance(raw_key, bytes) else str(raw_key) + value = raw_value.decode("utf-8") if isinstance(raw_value, bytes) else raw_value + if isinstance(value, str): + try: + value = json.loads(value) + except json.JSONDecodeError: + pass + decoded[key] = value + return decoded + + +def _encode_state_value(value: Any) -> str: + """Serialize one state value without losing its JSON type.""" + return json.dumps(value, ensure_ascii=False, separators=(",", ":")) + + class RedisSessionService(BaseSessionService): """A Redis implementation of the session service. @@ -264,7 +285,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_state_hash(await self._redis_storage.execute_command(redis_session, command)) if app_state: app_state.update(state_delta) else: @@ -280,7 +301,7 @@ async def _update_app_state(self, redis_session: RedisSession, app_name: str, # Use HSET with TTL if TTL is configured, otherwise use HSET args = [key] for k, v in app_state.items(): - args.extend([k, v]) + args.extend([k, _encode_state_value(v)]) command = RedisCommand(method='hset', args=tuple(args), @@ -304,7 +325,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_state_hash(await self._redis_storage.execute_command(redis_session, command)) if user_state: user_state.update(state_delta) else: @@ -320,7 +341,7 @@ async def _update_user_state(self, redis_session: RedisSession, app_name: str, u # Use HSET with TTL if TTL is configured, otherwise use HSET args = [key] for k, v in user_state.items(): - args.extend([k, v]) + args.extend([k, _encode_state_value(v)]) command = RedisCommand(method='hset', args=tuple(args), @@ -362,7 +383,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_state_hash(await self._redis_storage.execute_command(redis_session, command)) if app_state: await self._refresh_ttl(redis_session, key) @@ -383,7 +404,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_state_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/_session_summarizer.py b/trpc_agent_sdk/sessions/_session_summarizer.py index da717110..9234ee73 100644 --- a/trpc_agent_sdk/sessions/_session_summarizer.py +++ b/trpc_agent_sdk/sessions/_session_summarizer.py @@ -388,13 +388,18 @@ async def create_session_summary_by_events( summary_text = await self._compress_session_to_summary(events_for_summary, session_id, ctx) if summary_text: + # Persistent services reconstruct event order from timestamps. Anchor the + # summary just before the retained window so its logical leading position + # survives a database round trip. + summary_timestamp = (events[insert_index].timestamp - + 0.000001 if insert_index < len(events) else time.time()) # Create summary event summary_event = Event(invocation_id="summary", author="system", content=Content( parts=[Part.from_text(text=f"Previous conversation summary: {summary_text}")], role="user"), - timestamp=time.time()) + timestamp=summary_timestamp) summary_event.set_summary_event(True) summarized_events = events[:insert_index] From 3e049fc5fabc259da37dc2d7930755257e0a95aa Mon Sep 17 00:00:00 2001 From: yangyeyuan <2776079516@qq.com> Date: Sun, 12 Jul 2026 13:04:49 +0800 Subject: [PATCH 2/2] sessions: satisfy summary timestamp lint rule Use an explicit branch for the retained-event timestamp anchor so flake8 W504 passes without changing behavior.\n\nUpdates #89\n\nRELEASE NOTES: NONE --- trpc_agent_sdk/sessions/_session_summarizer.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/trpc_agent_sdk/sessions/_session_summarizer.py b/trpc_agent_sdk/sessions/_session_summarizer.py index 9234ee73..4f85de76 100644 --- a/trpc_agent_sdk/sessions/_session_summarizer.py +++ b/trpc_agent_sdk/sessions/_session_summarizer.py @@ -391,8 +391,9 @@ async def create_session_summary_by_events( # Persistent services reconstruct event order from timestamps. Anchor the # summary just before the retained window so its logical leading position # survives a database round trip. - summary_timestamp = (events[insert_index].timestamp - - 0.000001 if insert_index < len(events) else time.time()) + summary_timestamp = time.time() + if insert_index < len(events): + summary_timestamp = events[insert_index].timestamp - 0.000001 # Create summary event summary_event = Event(invocation_id="summary", author="system",