From d4de0e4e3dc2fcda959b89e7d992e168b175f9e2 Mon Sep 17 00:00:00 2001 From: chengyongru Date: Wed, 19 Aug 2026 15:50:48 +0800 Subject: [PATCH] test(memory): sharpen compaction coverage --- tests/agent/test_autocompact_unit.py | 21 ++++--------- tests/agent/test_consolidator.py | 38 ++++++++++++------------ tests/agent/test_context_prompt_cache.py | 15 +++++++--- 3 files changed, 36 insertions(+), 38 deletions(-) diff --git a/tests/agent/test_autocompact_unit.py b/tests/agent/test_autocompact_unit.py index 7ceefcf7e..d9dafc8df 100644 --- a/tests/agent/test_autocompact_unit.py +++ b/tests/agent/test_autocompact_unit.py @@ -184,23 +184,14 @@ class TestIsExpired: class TestSessionSummary: """Test prompt rendering for the structured summary value.""" - def test_contains_isoformat_timestamp(self): - """Output should contain last_active as isoformat.""" + def test_formats_prompt(self): last_active = datetime(2026, 5, 13, 14, 30, 0) - result = SessionSummary("Some text", last_active).for_prompt() - assert "2026-05-13T14:30:00" in result + summary = SessionSummary("User discussed Python.", last_active) - def test_contains_summary_text(self): - """Output should contain the provided text verbatim.""" - last_active = datetime(2026, 1, 1) - result = SessionSummary("User discussed Python.", last_active).for_prompt() - assert "User discussed Python." in result - - def test_output_starts_with_label(self): - """Output should start with the standard prefix.""" - last_active = datetime(2026, 1, 1) - result = SessionSummary("text", last_active).for_prompt() - assert result.startswith("Previous conversation summary (last active ") + assert summary.for_prompt() == ( + "Previous conversation summary (last active 2026-05-13T14:30:00):\n" + "User discussed Python." + ) # --------------------------------------------------------------------------- diff --git a/tests/agent/test_consolidator.py b/tests/agent/test_consolidator.py index cef7420c5..1d109b33a 100644 --- a/tests/agent/test_consolidator.py +++ b/tests/agent/test_consolidator.py @@ -21,6 +21,7 @@ from nanobot.runtime_context import ( RuntimeContextBlock, append_runtime_context, ) +from nanobot.session.keys import UNIFIED_SESSION_KEY, remember_last_channel from nanobot.session.manager import Session from nanobot.utils.llm_runtime import LLMRuntime from nanobot.utils.prompt_templates import render_template @@ -274,19 +275,6 @@ class TestConsolidatorPromptContract: assert "check context below" not in prompt.lower() assert "Do not mark something [skip] merely because it might already exist" in prompt - def test_archive_prompt_scopes_idle_overview_when_message_count_is_provided(self): - prompt = render_template( - "agent/consolidator_archive.md", - strip=True, - archive_count=4, - ) - - assert "only the final 4 conversation messages" in prompt - assert "Earlier messages are context" in prompt - assert "Do not call tools" in prompt - assert "Only SNIP facts" in prompt - - class TestConsolidatorArchiveErrorHandling: """archive() must fall back to raw_archive when the LLM returns an error response (finish_reason == 'error'), e.g. overloaded / quota exceeded. @@ -1033,7 +1021,7 @@ class TestCompactIdleSession: assert "user-14" in sent_content @pytest.mark.asyncio - async def test_reuses_model_prefix_without_persisting_temporary_turn( + async def test_preserves_tool_history_and_persists_only_overview( self, real_consolidator, mock_provider, @@ -1210,7 +1198,7 @@ class TestCompactIdleSession: assert "final 2 conversation messages" in sent[-1]["content"] @pytest.mark.asyncio - async def test_uses_persisted_workspace_scope_for_system_prefix( + async def test_reuses_real_prefix_for_unified_session_workspace( self, loop_factory, mock_provider, @@ -1220,13 +1208,14 @@ class TestCompactIdleSession: project.mkdir() (tmp_path / "AGENTS.md").write_text("GLOBAL_WORKSPACE_MARKER", encoding="utf-8") (project / "AGENTS.md").write_text("PROJECT_WORKSPACE_MARKER", encoding="utf-8") - loop = loop_factory(provider=mock_provider) + loop = loop_factory(provider=mock_provider, unified_session=True) runtime = loop.llm_runtime() runtime.provider.chat_with_retry.return_value = LLMResponse( content="Summary.", finish_reason="stop", ) - session = loop.sessions.get_or_create("websocket:scope") + session = loop.sessions.get_or_create(UNIFIED_SESSION_KEY) + remember_last_channel(session.metadata, "websocket", "scope") session.metadata["workspace_scope"] = { "project_path": str(project), "access_mode": "restricted", @@ -1234,13 +1223,24 @@ class TestCompactIdleSession: session.add_message("user", "project question") session.add_message("assistant", "project answer") loop.sessions.save(session) + ordinary_messages = loop.context.build_messages( + history=session.get_history(max_messages=0), + current_message="next project question", + channel="websocket", + workspace=project, + session_key=session.key, + unified_session=True, + ) await loop.consolidator.compact_idle_session( - "websocket:scope", + session.key, runtime=runtime, ) - system = runtime.provider.chat_with_retry.call_args.kwargs["messages"][0]["content"] + sent_messages = runtime.provider.chat_with_retry.call_args.kwargs["messages"] + assert sent_messages[:-1] == ordinary_messages[:-1] + assert "final 2 conversation messages" in sent_messages[-1]["content"] + system = sent_messages[0]["content"] assert "PROJECT_WORKSPACE_MARKER" in system assert "GLOBAL_WORKSPACE_MARKER" not in system diff --git a/tests/agent/test_context_prompt_cache.py b/tests/agent/test_context_prompt_cache.py index 39dc8fb1b..e3e7f25e1 100644 --- a/tests/agent/test_context_prompt_cache.py +++ b/tests/agent/test_context_prompt_cache.py @@ -113,30 +113,37 @@ def test_recent_history_injection_is_session_scoped(tmp_path) -> None: assert "legacy entry without session" not in prompt -def test_session_summary_replaces_matching_recent_history_entry(tmp_path) -> None: +def test_session_summary_replaces_interleaved_recent_history_entry(tmp_path) -> None: workspace = _make_workspace(tmp_path) builder = ContextBuilder(workspace) - session_key = "telegram:chat-1" + session_key = "unified:default" overview = "CURRENT_SESSION_OVERVIEW_MARKER" builder.memory.append_history("another session event", session_key=session_key) - summary_cursor = builder.memory.append_history(overview, session_key=session_key) + builder.memory.append_history(overview, session_key=session_key) + latest_cursor = builder.memory.append_history( + "later telegram event", + session_key="telegram:chat-1", + ) summary = SessionSummary(overview, real_datetime(2026, 8, 19, 10, 0)) prompt = builder.build_system_prompt( session_key=session_key, session_summary=summary, + unified_session=True, ) assert "# Recent History" in prompt assert "another session event" in prompt + assert "later telegram event" in prompt assert "[Archived Context Summary]" in prompt assert prompt.count(overview) == 1 - builder.memory.set_last_dream_cursor(summary_cursor) + builder.memory.set_last_dream_cursor(latest_cursor) processed_prompt = builder.build_system_prompt( session_key=session_key, session_summary=summary, + unified_session=True, ) assert "# Recent History" not in processed_prompt assert processed_prompt.count(overview) == 1