test(memory): sharpen compaction coverage

This commit is contained in:
chengyongru
2026-08-19 18:40:20 +08:00
committed by chengyongru
parent 6834d656a1
commit d4de0e4e3d
3 changed files with 36 additions and 38 deletions
+6 -15
View File
@@ -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."
)
# ---------------------------------------------------------------------------
+19 -19
View File
@@ -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
+11 -4
View File
@@ -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