mirror of
https://github.com/HKUDS/nanobot.git
synced 2026-04-13 22:49:47 +00:00
1cb28b39a3
Keep a legal recent suffix in idle auto-compacted sessions so resumed chats preserve their freshest live context while older messages are summarized. Recover persisted summaries even when retained messages remain, and document the new behavior.
898 lines
35 KiB
Python
898 lines
35 KiB
Python
"""Tests for auto compact (idle TTL) feature."""
|
|
|
|
import asyncio
|
|
from datetime import datetime, timedelta
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from nanobot.agent.loop import AgentLoop
|
|
from nanobot.bus.events import InboundMessage
|
|
from nanobot.bus.queue import MessageBus
|
|
from nanobot.config.schema import AgentDefaults
|
|
from nanobot.command import CommandContext
|
|
from nanobot.providers.base import LLMResponse
|
|
|
|
|
|
def _make_loop(tmp_path: Path, session_ttl_minutes: int = 15) -> AgentLoop:
|
|
"""Create a minimal AgentLoop for testing."""
|
|
bus = MessageBus()
|
|
provider = MagicMock()
|
|
provider.get_default_model.return_value = "test-model"
|
|
provider.estimate_prompt_tokens.return_value = (10_000, "test")
|
|
provider.chat_with_retry = AsyncMock(return_value=LLMResponse(content="ok", tool_calls=[]))
|
|
provider.generation.max_tokens = 4096
|
|
loop = AgentLoop(
|
|
bus=bus,
|
|
provider=provider,
|
|
workspace=tmp_path,
|
|
model="test-model",
|
|
context_window_tokens=128_000,
|
|
session_ttl_minutes=session_ttl_minutes,
|
|
)
|
|
loop.tools.get_definitions = MagicMock(return_value=[])
|
|
return loop
|
|
|
|
|
|
def _add_turns(session, turns: int, *, prefix: str = "msg") -> None:
|
|
"""Append simple user/assistant turns to a session."""
|
|
for i in range(turns):
|
|
session.add_message("user", f"{prefix} user {i}")
|
|
session.add_message("assistant", f"{prefix} assistant {i}")
|
|
|
|
|
|
class TestSessionTTLConfig:
|
|
"""Test session TTL configuration."""
|
|
|
|
def test_default_ttl_is_zero(self):
|
|
"""Default TTL should be 0 (disabled)."""
|
|
defaults = AgentDefaults()
|
|
assert defaults.session_ttl_minutes == 0
|
|
|
|
def test_custom_ttl(self):
|
|
"""Custom TTL should be stored correctly."""
|
|
defaults = AgentDefaults(session_ttl_minutes=30)
|
|
assert defaults.session_ttl_minutes == 30
|
|
|
|
|
|
class TestAgentLoopTTLParam:
|
|
"""Test that AutoCompact receives and stores session_ttl_minutes."""
|
|
|
|
def test_loop_stores_ttl(self, tmp_path):
|
|
"""AutoCompact should store the TTL value."""
|
|
loop = _make_loop(tmp_path, session_ttl_minutes=25)
|
|
assert loop.auto_compact._ttl == 25
|
|
|
|
def test_loop_default_ttl_zero(self, tmp_path):
|
|
"""AutoCompact default TTL should be 0 (disabled)."""
|
|
loop = _make_loop(tmp_path, session_ttl_minutes=0)
|
|
assert loop.auto_compact._ttl == 0
|
|
|
|
|
|
class TestAutoCompact:
|
|
"""Test the _archive method."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_is_expired_boundary(self, tmp_path):
|
|
"""Exactly at TTL boundary should be expired (>= not >)."""
|
|
loop = _make_loop(tmp_path, session_ttl_minutes=15)
|
|
ts = datetime.now() - timedelta(minutes=15)
|
|
assert loop.auto_compact._is_expired(ts) is True
|
|
ts2 = datetime.now() - timedelta(minutes=14, seconds=59)
|
|
assert loop.auto_compact._is_expired(ts2) is False
|
|
await loop.close_mcp()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_is_expired_string_timestamp(self, tmp_path):
|
|
"""_is_expired should parse ISO string timestamps."""
|
|
loop = _make_loop(tmp_path, session_ttl_minutes=15)
|
|
ts = (datetime.now() - timedelta(minutes=20)).isoformat()
|
|
assert loop.auto_compact._is_expired(ts) is True
|
|
assert loop.auto_compact._is_expired(None) is False
|
|
assert loop.auto_compact._is_expired("") is False
|
|
await loop.close_mcp()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_expired_only_archives_expired_sessions(self, tmp_path):
|
|
"""With multiple sessions, only the expired one should be archived."""
|
|
loop = _make_loop(tmp_path, session_ttl_minutes=15)
|
|
# Expired session
|
|
s1 = loop.sessions.get_or_create("cli:expired")
|
|
s1.add_message("user", "old")
|
|
s1.updated_at = datetime.now() - timedelta(minutes=20)
|
|
loop.sessions.save(s1)
|
|
# Active session
|
|
s2 = loop.sessions.get_or_create("cli:active")
|
|
s2.add_message("user", "recent")
|
|
loop.sessions.save(s2)
|
|
|
|
async def _fake_archive(messages):
|
|
return "Summary."
|
|
|
|
loop.consolidator.archive = _fake_archive
|
|
loop.auto_compact.check_expired(loop._schedule_background)
|
|
await asyncio.sleep(0.1)
|
|
|
|
active_after = loop.sessions.get_or_create("cli:active")
|
|
assert len(active_after.messages) == 1
|
|
assert active_after.messages[0]["content"] == "recent"
|
|
await loop.close_mcp()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_auto_compact_archives_prefix_and_keeps_recent_suffix(self, tmp_path):
|
|
"""_archive should summarize the old prefix and keep a recent legal suffix."""
|
|
loop = _make_loop(tmp_path, session_ttl_minutes=15)
|
|
session = loop.sessions.get_or_create("cli:test")
|
|
_add_turns(session, 6)
|
|
loop.sessions.save(session)
|
|
|
|
archived_messages = []
|
|
|
|
async def _fake_archive(messages):
|
|
archived_messages.extend(messages)
|
|
return "Summary."
|
|
|
|
loop.consolidator.archive = _fake_archive
|
|
|
|
await loop.auto_compact._archive("cli:test")
|
|
|
|
assert len(archived_messages) == 4
|
|
session_after = loop.sessions.get_or_create("cli:test")
|
|
assert len(session_after.messages) == loop.auto_compact._RECENT_SUFFIX_MESSAGES
|
|
assert session_after.messages[0]["content"] == "msg user 2"
|
|
assert session_after.messages[-1]["content"] == "msg assistant 5"
|
|
await loop.close_mcp()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_auto_compact_stores_summary(self, tmp_path):
|
|
"""_archive should store the summary in _summaries."""
|
|
loop = _make_loop(tmp_path, session_ttl_minutes=15)
|
|
session = loop.sessions.get_or_create("cli:test")
|
|
_add_turns(session, 6, prefix="hello")
|
|
loop.sessions.save(session)
|
|
|
|
async def _fake_archive(messages):
|
|
return "User said hello."
|
|
|
|
loop.consolidator.archive = _fake_archive
|
|
|
|
await loop.auto_compact._archive("cli:test")
|
|
|
|
entry = loop.auto_compact._summaries.get("cli:test")
|
|
assert entry is not None
|
|
assert entry[0] == "User said hello."
|
|
session_after = loop.sessions.get_or_create("cli:test")
|
|
assert len(session_after.messages) == loop.auto_compact._RECENT_SUFFIX_MESSAGES
|
|
await loop.close_mcp()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_auto_compact_empty_session(self, tmp_path):
|
|
"""_archive on empty session should not archive."""
|
|
loop = _make_loop(tmp_path, session_ttl_minutes=15)
|
|
session = loop.sessions.get_or_create("cli:test")
|
|
|
|
archive_called = False
|
|
|
|
async def _fake_archive(messages):
|
|
nonlocal archive_called
|
|
archive_called = True
|
|
return "Summary."
|
|
|
|
loop.consolidator.archive = _fake_archive
|
|
|
|
await loop.auto_compact._archive("cli:test")
|
|
|
|
assert not archive_called
|
|
session_after = loop.sessions.get_or_create("cli:test")
|
|
assert len(session_after.messages) == 0
|
|
await loop.close_mcp()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_auto_compact_respects_last_consolidated(self, tmp_path):
|
|
"""_archive should only archive un-consolidated messages."""
|
|
loop = _make_loop(tmp_path, session_ttl_minutes=15)
|
|
session = loop.sessions.get_or_create("cli:test")
|
|
_add_turns(session, 14)
|
|
session.last_consolidated = 18
|
|
loop.sessions.save(session)
|
|
|
|
archived_count = 0
|
|
|
|
async def _fake_archive(messages):
|
|
nonlocal archived_count
|
|
archived_count = len(messages)
|
|
return "Summary."
|
|
|
|
loop.consolidator.archive = _fake_archive
|
|
|
|
await loop.auto_compact._archive("cli:test")
|
|
|
|
assert archived_count == 2
|
|
await loop.close_mcp()
|
|
|
|
|
|
class TestAutoCompactIdleDetection:
|
|
"""Test idle detection triggers auto-new in _process_message."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_no_auto_compact_when_ttl_disabled(self, tmp_path):
|
|
"""No auto-new should happen when TTL is 0 (disabled)."""
|
|
loop = _make_loop(tmp_path, session_ttl_minutes=0)
|
|
session = loop.sessions.get_or_create("cli:test")
|
|
session.add_message("user", "old message")
|
|
session.updated_at = datetime.now() - timedelta(minutes=30)
|
|
loop.sessions.save(session)
|
|
|
|
msg = InboundMessage(channel="cli", sender_id="user", chat_id="test", content="new msg")
|
|
await loop._process_message(msg)
|
|
|
|
session_after = loop.sessions.get_or_create("cli:test")
|
|
assert any(m["content"] == "old message" for m in session_after.messages)
|
|
await loop.close_mcp()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_auto_compact_triggers_on_idle(self, tmp_path):
|
|
"""Proactive auto-new archives expired session; _process_message reloads it."""
|
|
loop = _make_loop(tmp_path, session_ttl_minutes=15)
|
|
session = loop.sessions.get_or_create("cli:test")
|
|
_add_turns(session, 6, prefix="old")
|
|
session.updated_at = datetime.now() - timedelta(minutes=20)
|
|
loop.sessions.save(session)
|
|
|
|
archived_messages = []
|
|
|
|
async def _fake_archive(messages):
|
|
archived_messages.extend(messages)
|
|
return "Summary."
|
|
|
|
loop.consolidator.archive = _fake_archive
|
|
|
|
# Simulate proactive archive completing before message arrives
|
|
await loop.auto_compact._archive("cli:test")
|
|
|
|
msg = InboundMessage(channel="cli", sender_id="user", chat_id="test", content="new msg")
|
|
await loop._process_message(msg)
|
|
|
|
session_after = loop.sessions.get_or_create("cli:test")
|
|
assert len(archived_messages) == 4
|
|
assert not any(m["content"] == "old user 0" for m in session_after.messages)
|
|
assert any(m["content"] == "new msg" for m in session_after.messages)
|
|
await loop.close_mcp()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_no_auto_compact_when_active(self, tmp_path):
|
|
"""No auto-new should happen when session is recently active."""
|
|
loop = _make_loop(tmp_path, session_ttl_minutes=15)
|
|
session = loop.sessions.get_or_create("cli:test")
|
|
session.add_message("user", "recent message")
|
|
loop.sessions.save(session)
|
|
|
|
msg = InboundMessage(channel="cli", sender_id="user", chat_id="test", content="new msg")
|
|
await loop._process_message(msg)
|
|
|
|
session_after = loop.sessions.get_or_create("cli:test")
|
|
assert any(m["content"] == "recent message" for m in session_after.messages)
|
|
await loop.close_mcp()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_auto_compact_does_not_affect_priority_commands(self, tmp_path):
|
|
"""Priority commands (/stop, /restart) bypass _process_message entirely via run()."""
|
|
loop = _make_loop(tmp_path, session_ttl_minutes=15)
|
|
session = loop.sessions.get_or_create("cli:test")
|
|
session.add_message("user", "old message")
|
|
session.updated_at = datetime.now() - timedelta(minutes=20)
|
|
loop.sessions.save(session)
|
|
|
|
# Priority commands are dispatched in run() before _process_message is called.
|
|
# Simulate that path directly via dispatch_priority.
|
|
raw = "/stop"
|
|
msg = InboundMessage(channel="cli", sender_id="user", chat_id="test", content=raw)
|
|
ctx = CommandContext(msg=msg, session=session, key="cli:test", raw=raw, loop=loop)
|
|
result = await loop.commands.dispatch_priority(ctx)
|
|
assert result is not None
|
|
assert "stopped" in result.content.lower() or "no active task" in result.content.lower()
|
|
|
|
# Session should be untouched since priority commands skip _process_message
|
|
session_after = loop.sessions.get_or_create("cli:test")
|
|
assert any(m["content"] == "old message" for m in session_after.messages)
|
|
await loop.close_mcp()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_auto_compact_with_slash_new(self, tmp_path):
|
|
"""Auto-new fires before /new dispatches; session is cleared twice but idempotent."""
|
|
loop = _make_loop(tmp_path, session_ttl_minutes=15)
|
|
session = loop.sessions.get_or_create("cli:test")
|
|
for i in range(4):
|
|
session.add_message("user", f"msg{i}")
|
|
session.add_message("assistant", f"resp{i}")
|
|
session.updated_at = datetime.now() - timedelta(minutes=20)
|
|
loop.sessions.save(session)
|
|
|
|
async def _fake_archive(messages):
|
|
return "Summary."
|
|
|
|
loop.consolidator.archive = _fake_archive
|
|
|
|
msg = InboundMessage(channel="cli", sender_id="user", chat_id="test", content="/new")
|
|
response = await loop._process_message(msg)
|
|
|
|
assert response is not None
|
|
assert "new session started" in response.content.lower()
|
|
|
|
session_after = loop.sessions.get_or_create("cli:test")
|
|
# Session is empty (auto-new archived and cleared, /new cleared again)
|
|
assert len(session_after.messages) == 0
|
|
await loop.close_mcp()
|
|
|
|
|
|
class TestAutoCompactSystemMessages:
|
|
"""Test that auto-new also works for system messages."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_auto_compact_triggers_for_system_messages(self, tmp_path):
|
|
"""Proactive auto-new archives expired session; system messages reload it."""
|
|
loop = _make_loop(tmp_path, session_ttl_minutes=15)
|
|
session = loop.sessions.get_or_create("cli:test")
|
|
_add_turns(session, 6, prefix="old")
|
|
session.updated_at = datetime.now() - timedelta(minutes=20)
|
|
loop.sessions.save(session)
|
|
|
|
async def _fake_archive(messages):
|
|
return "Summary."
|
|
|
|
loop.consolidator.archive = _fake_archive
|
|
|
|
# Simulate proactive archive completing before system message arrives
|
|
await loop.auto_compact._archive("cli:test")
|
|
|
|
msg = InboundMessage(
|
|
channel="system", sender_id="subagent", chat_id="cli:test",
|
|
content="subagent result",
|
|
)
|
|
await loop._process_message(msg)
|
|
|
|
session_after = loop.sessions.get_or_create("cli:test")
|
|
assert not any(
|
|
m["content"] == "old user 0"
|
|
for m in session_after.messages
|
|
)
|
|
await loop.close_mcp()
|
|
|
|
|
|
class TestAutoCompactEdgeCases:
|
|
"""Edge cases for auto session new."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_auto_compact_with_nothing_summary(self, tmp_path):
|
|
"""Auto-new should not inject when archive produces '(nothing)'."""
|
|
loop = _make_loop(tmp_path, session_ttl_minutes=15)
|
|
session = loop.sessions.get_or_create("cli:test")
|
|
_add_turns(session, 6, prefix="thanks")
|
|
session.updated_at = datetime.now() - timedelta(minutes=20)
|
|
loop.sessions.save(session)
|
|
|
|
loop.provider.chat_with_retry = AsyncMock(
|
|
return_value=LLMResponse(content="(nothing)", tool_calls=[])
|
|
)
|
|
|
|
await loop.auto_compact._archive("cli:test")
|
|
|
|
session_after = loop.sessions.get_or_create("cli:test")
|
|
assert len(session_after.messages) == loop.auto_compact._RECENT_SUFFIX_MESSAGES
|
|
# "(nothing)" summary should not be stored
|
|
assert "cli:test" not in loop.auto_compact._summaries
|
|
|
|
await loop.close_mcp()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_auto_compact_archive_failure_still_keeps_recent_suffix(self, tmp_path):
|
|
"""Auto-new should keep the recent suffix even if LLM archive falls back to raw dump."""
|
|
loop = _make_loop(tmp_path, session_ttl_minutes=15)
|
|
session = loop.sessions.get_or_create("cli:test")
|
|
_add_turns(session, 6, prefix="important")
|
|
session.updated_at = datetime.now() - timedelta(minutes=20)
|
|
loop.sessions.save(session)
|
|
|
|
loop.provider.chat_with_retry = AsyncMock(side_effect=Exception("API down"))
|
|
|
|
# Should not raise
|
|
await loop.auto_compact._archive("cli:test")
|
|
|
|
session_after = loop.sessions.get_or_create("cli:test")
|
|
assert len(session_after.messages) == loop.auto_compact._RECENT_SUFFIX_MESSAGES
|
|
|
|
await loop.close_mcp()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_auto_compact_preserves_runtime_checkpoint_before_check(self, tmp_path):
|
|
"""Short expired sessions keep recent messages; checkpoint restore still works on resume."""
|
|
loop = _make_loop(tmp_path, session_ttl_minutes=15)
|
|
session = loop.sessions.get_or_create("cli:test")
|
|
session.metadata[AgentLoop._RUNTIME_CHECKPOINT_KEY] = {
|
|
"assistant_message": {"role": "assistant", "content": "interrupted response"},
|
|
"completed_tool_results": [],
|
|
"pending_tool_calls": [],
|
|
}
|
|
session.add_message("user", "previous message")
|
|
session.updated_at = datetime.now() - timedelta(minutes=20)
|
|
loop.sessions.save(session)
|
|
|
|
archived_messages = []
|
|
|
|
async def _fake_archive(messages):
|
|
archived_messages.extend(messages)
|
|
return "Summary."
|
|
|
|
loop.consolidator.archive = _fake_archive
|
|
|
|
# Simulate proactive archive completing before message arrives
|
|
await loop.auto_compact._archive("cli:test")
|
|
|
|
msg = InboundMessage(channel="cli", sender_id="user", chat_id="test", content="continue")
|
|
await loop._process_message(msg)
|
|
|
|
session_after = loop.sessions.get_or_create("cli:test")
|
|
assert archived_messages == []
|
|
assert any(m["content"] == "previous message" for m in session_after.messages)
|
|
assert any(m["content"] == "interrupted response" for m in session_after.messages)
|
|
|
|
await loop.close_mcp()
|
|
|
|
|
|
class TestAutoCompactIntegration:
|
|
"""End-to-end test of auto session new feature."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_full_lifecycle(self, tmp_path):
|
|
"""
|
|
Full lifecycle: messages -> idle -> auto-new -> archive -> clear -> summary injected as runtime context.
|
|
"""
|
|
loop = _make_loop(tmp_path, session_ttl_minutes=15)
|
|
session = loop.sessions.get_or_create("cli:test")
|
|
|
|
# Phase 1: User has a conversation longer than the retained recent suffix
|
|
session.add_message("user", "I'm learning English, teach me past tense")
|
|
session.add_message("assistant", "Past tense is used for actions completed in the past...")
|
|
session.add_message("user", "Give me an example")
|
|
session.add_message("assistant", '"I walked to the store yesterday."')
|
|
session.add_message("user", "Give me another example")
|
|
session.add_message("assistant", '"She visited Paris last year."')
|
|
session.add_message("user", "Quiz me")
|
|
session.add_message("assistant", "What is the past tense of go?")
|
|
session.add_message("user", "I think it is went")
|
|
session.add_message("assistant", "Correct.")
|
|
loop.sessions.save(session)
|
|
|
|
# Phase 2: Time passes (simulate idle)
|
|
session.updated_at = datetime.now() - timedelta(minutes=20)
|
|
loop.sessions.save(session)
|
|
|
|
# Phase 3: User returns with a new message
|
|
loop.provider.chat_with_retry = AsyncMock(
|
|
return_value=LLMResponse(
|
|
content="User is learning English past tense. Example: 'I walked to the store yesterday.'",
|
|
tool_calls=[],
|
|
)
|
|
)
|
|
|
|
msg = InboundMessage(
|
|
channel="cli", sender_id="user", chat_id="test",
|
|
content="Let's continue, teach me present perfect",
|
|
)
|
|
response = await loop._process_message(msg)
|
|
|
|
# Phase 4: Verify
|
|
session_after = loop.sessions.get_or_create("cli:test")
|
|
|
|
# The oldest messages should be trimmed from live session history
|
|
assert not any(
|
|
"past tense is used" in str(m.get("content", "")) for m in session_after.messages
|
|
)
|
|
|
|
# Summary should NOT be persisted in session (ephemeral, one-shot)
|
|
assert not any(
|
|
"[Resumed Session]" in str(m.get("content", "")) for m in session_after.messages
|
|
)
|
|
# Runtime context end marker should NOT be persisted
|
|
assert not any(
|
|
"[/Runtime Context]" in str(m.get("content", "")) for m in session_after.messages
|
|
)
|
|
|
|
# Pending summary should be consumed (one-shot)
|
|
assert "cli:test" not in loop.auto_compact._summaries
|
|
|
|
# The new message should be processed (response exists)
|
|
assert response is not None
|
|
|
|
await loop.close_mcp()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_runtime_context_markers_not_persisted_for_multi_paragraph_turn(self, tmp_path):
|
|
"""Auto-compact resume context must not leak runtime markers into persisted session history."""
|
|
loop = _make_loop(tmp_path, session_ttl_minutes=15)
|
|
session = loop.sessions.get_or_create("cli:test")
|
|
session.add_message("user", "old message")
|
|
session.updated_at = datetime.now() - timedelta(minutes=20)
|
|
loop.sessions.save(session)
|
|
|
|
async def _fake_archive(messages):
|
|
return "Summary."
|
|
|
|
loop.consolidator.archive = _fake_archive
|
|
|
|
# Simulate proactive archive completing before message arrives
|
|
await loop.auto_compact._archive("cli:test")
|
|
|
|
msg = InboundMessage(
|
|
channel="cli", sender_id="user", chat_id="test",
|
|
content="Paragraph one\n\nParagraph two\n\nParagraph three",
|
|
)
|
|
await loop._process_message(msg)
|
|
|
|
session_after = loop.sessions.get_or_create("cli:test")
|
|
assert any(m.get("content") == "old message" for m in session_after.messages)
|
|
for persisted in session_after.messages:
|
|
content = str(persisted.get("content", ""))
|
|
assert "[Runtime Context" not in content
|
|
assert "[/Runtime Context]" not in content
|
|
await loop.close_mcp()
|
|
|
|
|
|
class TestProactiveAutoCompact:
|
|
"""Test proactive auto-new on idle ticks (TimeoutError path in run loop)."""
|
|
|
|
@staticmethod
|
|
async def _run_check_expired(loop):
|
|
"""Helper: run check_expired via callback and wait for background tasks."""
|
|
loop.auto_compact.check_expired(loop._schedule_background)
|
|
await asyncio.sleep(0.1)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_no_check_when_ttl_disabled(self, tmp_path):
|
|
"""check_expired should be a no-op when TTL is 0."""
|
|
loop = _make_loop(tmp_path, session_ttl_minutes=0)
|
|
session = loop.sessions.get_or_create("cli:test")
|
|
session.add_message("user", "old message")
|
|
session.updated_at = datetime.now() - timedelta(minutes=30)
|
|
loop.sessions.save(session)
|
|
|
|
await self._run_check_expired(loop)
|
|
|
|
session_after = loop.sessions.get_or_create("cli:test")
|
|
assert len(session_after.messages) == 1
|
|
await loop.close_mcp()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_proactive_archive_on_idle_tick(self, tmp_path):
|
|
"""Expired session should be archived during idle tick."""
|
|
loop = _make_loop(tmp_path, session_ttl_minutes=15)
|
|
session = loop.sessions.get_or_create("cli:test")
|
|
_add_turns(session, 5, prefix="old")
|
|
session.updated_at = datetime.now() - timedelta(minutes=20)
|
|
loop.sessions.save(session)
|
|
|
|
archived_messages = []
|
|
|
|
async def _fake_archive(messages):
|
|
archived_messages.extend(messages)
|
|
return "User chatted about old things."
|
|
|
|
loop.consolidator.archive = _fake_archive
|
|
|
|
await self._run_check_expired(loop)
|
|
|
|
session_after = loop.sessions.get_or_create("cli:test")
|
|
assert len(session_after.messages) == loop.auto_compact._RECENT_SUFFIX_MESSAGES
|
|
assert len(archived_messages) == 2
|
|
entry = loop.auto_compact._summaries.get("cli:test")
|
|
assert entry is not None
|
|
assert entry[0] == "User chatted about old things."
|
|
await loop.close_mcp()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_no_proactive_archive_when_active(self, tmp_path):
|
|
"""Recently active session should NOT be archived on idle tick."""
|
|
loop = _make_loop(tmp_path, session_ttl_minutes=15)
|
|
session = loop.sessions.get_or_create("cli:test")
|
|
session.add_message("user", "recent message")
|
|
loop.sessions.save(session)
|
|
|
|
await self._run_check_expired(loop)
|
|
|
|
session_after = loop.sessions.get_or_create("cli:test")
|
|
assert len(session_after.messages) == 1
|
|
await loop.close_mcp()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_no_duplicate_archive(self, tmp_path):
|
|
"""Should not archive the same session twice if already in progress."""
|
|
loop = _make_loop(tmp_path, session_ttl_minutes=15)
|
|
session = loop.sessions.get_or_create("cli:test")
|
|
_add_turns(session, 6, prefix="old")
|
|
session.updated_at = datetime.now() - timedelta(minutes=20)
|
|
loop.sessions.save(session)
|
|
|
|
archive_count = 0
|
|
started = asyncio.Event()
|
|
block_forever = asyncio.Event()
|
|
|
|
async def _slow_archive(messages):
|
|
nonlocal archive_count
|
|
archive_count += 1
|
|
started.set()
|
|
await block_forever.wait()
|
|
return "Summary."
|
|
|
|
loop.consolidator.archive = _slow_archive
|
|
|
|
# First call starts archiving via callback
|
|
loop.auto_compact.check_expired(loop._schedule_background)
|
|
await started.wait()
|
|
assert archive_count == 1
|
|
|
|
# Second call should skip (key is in _archiving)
|
|
loop.auto_compact.check_expired(loop._schedule_background)
|
|
await asyncio.sleep(0.05)
|
|
assert archive_count == 1
|
|
|
|
# Clean up
|
|
block_forever.set()
|
|
await asyncio.sleep(0.1)
|
|
await loop.close_mcp()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_proactive_archive_error_does_not_block(self, tmp_path):
|
|
"""Proactive archive failure should be caught and not block future ticks."""
|
|
loop = _make_loop(tmp_path, session_ttl_minutes=15)
|
|
session = loop.sessions.get_or_create("cli:test")
|
|
_add_turns(session, 6, prefix="old")
|
|
session.updated_at = datetime.now() - timedelta(minutes=20)
|
|
loop.sessions.save(session)
|
|
|
|
async def _failing_archive(messages):
|
|
raise RuntimeError("LLM down")
|
|
|
|
loop.consolidator.archive = _failing_archive
|
|
|
|
# Should not raise
|
|
await self._run_check_expired(loop)
|
|
|
|
# Key should be removed from _archiving (finally block)
|
|
assert "cli:test" not in loop.auto_compact._archiving
|
|
await loop.close_mcp()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_proactive_archive_skips_empty_sessions(self, tmp_path):
|
|
"""Proactive archive should not call LLM for sessions with no un-consolidated messages."""
|
|
loop = _make_loop(tmp_path, session_ttl_minutes=15)
|
|
session = loop.sessions.get_or_create("cli:test")
|
|
session.updated_at = datetime.now() - timedelta(minutes=20)
|
|
loop.sessions.save(session)
|
|
|
|
archive_called = False
|
|
|
|
async def _fake_archive(messages):
|
|
nonlocal archive_called
|
|
archive_called = True
|
|
return "Summary."
|
|
|
|
loop.consolidator.archive = _fake_archive
|
|
|
|
await self._run_check_expired(loop)
|
|
|
|
assert not archive_called
|
|
await loop.close_mcp()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_no_reschedule_after_successful_archive(self, tmp_path):
|
|
"""Already-archived session should NOT be re-scheduled on subsequent ticks."""
|
|
loop = _make_loop(tmp_path, session_ttl_minutes=15)
|
|
session = loop.sessions.get_or_create("cli:test")
|
|
_add_turns(session, 5, prefix="old")
|
|
session.updated_at = datetime.now() - timedelta(minutes=20)
|
|
loop.sessions.save(session)
|
|
|
|
archive_count = 0
|
|
|
|
async def _fake_archive(messages):
|
|
nonlocal archive_count
|
|
archive_count += 1
|
|
return "Summary."
|
|
|
|
loop.consolidator.archive = _fake_archive
|
|
|
|
# First tick: archives the session
|
|
await self._run_check_expired(loop)
|
|
assert archive_count == 1
|
|
|
|
# Second tick: should NOT re-schedule (updated_at is fresh after clear)
|
|
await self._run_check_expired(loop)
|
|
assert archive_count == 1 # Still 1, not re-scheduled
|
|
await loop.close_mcp()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_empty_skip_refreshes_updated_at_prevents_reschedule(self, tmp_path):
|
|
"""Empty session skip refreshes updated_at, preventing immediate re-scheduling."""
|
|
loop = _make_loop(tmp_path, session_ttl_minutes=15)
|
|
session = loop.sessions.get_or_create("cli:test")
|
|
session.updated_at = datetime.now() - timedelta(minutes=20)
|
|
loop.sessions.save(session)
|
|
|
|
archive_count = 0
|
|
|
|
async def _fake_archive(messages):
|
|
nonlocal archive_count
|
|
archive_count += 1
|
|
return "Summary."
|
|
|
|
loop.consolidator.archive = _fake_archive
|
|
|
|
# First tick: skips (no messages), refreshes updated_at
|
|
await self._run_check_expired(loop)
|
|
assert archive_count == 0
|
|
|
|
# Second tick: should NOT re-schedule because updated_at is fresh
|
|
await self._run_check_expired(loop)
|
|
assert archive_count == 0
|
|
await loop.close_mcp()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_session_can_be_compacted_again_after_new_messages(self, tmp_path):
|
|
"""After successful compact + user sends new messages + idle again, should compact again."""
|
|
loop = _make_loop(tmp_path, session_ttl_minutes=15)
|
|
session = loop.sessions.get_or_create("cli:test")
|
|
_add_turns(session, 5, prefix="first")
|
|
session.updated_at = datetime.now() - timedelta(minutes=20)
|
|
loop.sessions.save(session)
|
|
|
|
archive_count = 0
|
|
|
|
async def _fake_archive(messages):
|
|
nonlocal archive_count
|
|
archive_count += 1
|
|
return "Summary."
|
|
|
|
loop.consolidator.archive = _fake_archive
|
|
|
|
# First compact cycle
|
|
await loop.auto_compact._archive("cli:test")
|
|
assert archive_count == 1
|
|
|
|
# User returns, sends new messages
|
|
msg = InboundMessage(channel="cli", sender_id="user", chat_id="test", content="second topic")
|
|
await loop._process_message(msg)
|
|
|
|
# Simulate idle again
|
|
loop.sessions.invalidate("cli:test")
|
|
session2 = loop.sessions.get_or_create("cli:test")
|
|
session2.updated_at = datetime.now() - timedelta(minutes=20)
|
|
loop.sessions.save(session2)
|
|
|
|
# Second compact cycle should succeed
|
|
await loop.auto_compact._archive("cli:test")
|
|
assert archive_count == 2
|
|
await loop.close_mcp()
|
|
|
|
|
|
class TestSummaryPersistence:
|
|
"""Test that summary survives restart via session metadata."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_summary_persisted_in_session_metadata(self, tmp_path):
|
|
"""After archive, _last_summary should be in session metadata."""
|
|
loop = _make_loop(tmp_path, session_ttl_minutes=15)
|
|
session = loop.sessions.get_or_create("cli:test")
|
|
_add_turns(session, 6, prefix="hello")
|
|
session.updated_at = datetime.now() - timedelta(minutes=20)
|
|
loop.sessions.save(session)
|
|
|
|
async def _fake_archive(messages):
|
|
return "User said hello."
|
|
|
|
loop.consolidator.archive = _fake_archive
|
|
|
|
await loop.auto_compact._archive("cli:test")
|
|
|
|
# Summary should be persisted in session metadata
|
|
session_after = loop.sessions.get_or_create("cli:test")
|
|
meta = session_after.metadata.get("_last_summary")
|
|
assert meta is not None
|
|
assert meta["text"] == "User said hello."
|
|
assert "last_active" in meta
|
|
await loop.close_mcp()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_summary_recovered_after_restart(self, tmp_path):
|
|
"""Summary should be recovered from metadata when _summaries is empty (simulates restart)."""
|
|
loop = _make_loop(tmp_path, session_ttl_minutes=15)
|
|
session = loop.sessions.get_or_create("cli:test")
|
|
_add_turns(session, 6, prefix="hello")
|
|
last_active = datetime.now() - timedelta(minutes=20)
|
|
session.updated_at = last_active
|
|
loop.sessions.save(session)
|
|
|
|
async def _fake_archive(messages):
|
|
return "User said hello."
|
|
|
|
loop.consolidator.archive = _fake_archive
|
|
|
|
# Archive
|
|
await loop.auto_compact._archive("cli:test")
|
|
|
|
# Simulate restart: clear in-memory state
|
|
loop.auto_compact._summaries.clear()
|
|
loop.sessions.invalidate("cli:test")
|
|
|
|
# prepare_session should recover summary from metadata
|
|
reloaded = loop.sessions.get_or_create("cli:test")
|
|
assert len(reloaded.messages) == loop.auto_compact._RECENT_SUFFIX_MESSAGES
|
|
_, summary = loop.auto_compact.prepare_session(reloaded, "cli:test")
|
|
|
|
assert summary is not None
|
|
assert "User said hello." in summary
|
|
assert "Inactive for" in summary
|
|
# Metadata should be cleaned up after consumption
|
|
assert "_last_summary" not in reloaded.metadata
|
|
await loop.close_mcp()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_metadata_cleanup_no_leak(self, tmp_path):
|
|
"""_last_summary should be removed from metadata after being consumed."""
|
|
loop = _make_loop(tmp_path, session_ttl_minutes=15)
|
|
session = loop.sessions.get_or_create("cli:test")
|
|
_add_turns(session, 6, prefix="hello")
|
|
session.updated_at = datetime.now() - timedelta(minutes=20)
|
|
loop.sessions.save(session)
|
|
|
|
async def _fake_archive(messages):
|
|
return "Summary."
|
|
|
|
loop.consolidator.archive = _fake_archive
|
|
|
|
await loop.auto_compact._archive("cli:test")
|
|
|
|
# Clear in-memory to force metadata path
|
|
loop.auto_compact._summaries.clear()
|
|
loop.sessions.invalidate("cli:test")
|
|
reloaded = loop.sessions.get_or_create("cli:test")
|
|
|
|
# First call: consumes from metadata
|
|
_, summary = loop.auto_compact.prepare_session(reloaded, "cli:test")
|
|
assert summary is not None
|
|
|
|
# Second call: no summary (already consumed)
|
|
_, summary2 = loop.auto_compact.prepare_session(reloaded, "cli:test")
|
|
assert summary2 is None
|
|
assert "_last_summary" not in reloaded.metadata
|
|
await loop.close_mcp()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_metadata_cleanup_on_inmemory_path(self, tmp_path):
|
|
"""In-memory _summaries path should also clean up _last_summary from metadata."""
|
|
loop = _make_loop(tmp_path, session_ttl_minutes=15)
|
|
session = loop.sessions.get_or_create("cli:test")
|
|
_add_turns(session, 6, prefix="hello")
|
|
session.updated_at = datetime.now() - timedelta(minutes=20)
|
|
loop.sessions.save(session)
|
|
|
|
async def _fake_archive(messages):
|
|
return "Summary."
|
|
|
|
loop.consolidator.archive = _fake_archive
|
|
|
|
await loop.auto_compact._archive("cli:test")
|
|
|
|
# Both _summaries and metadata have the summary
|
|
assert "cli:test" in loop.auto_compact._summaries
|
|
loop.sessions.invalidate("cli:test")
|
|
reloaded = loop.sessions.get_or_create("cli:test")
|
|
assert "_last_summary" in reloaded.metadata
|
|
|
|
# In-memory path is taken (no restart)
|
|
_, summary = loop.auto_compact.prepare_session(reloaded, "cli:test")
|
|
assert summary is not None
|
|
# Metadata should also be cleaned up
|
|
assert "_last_summary" not in reloaded.metadata
|
|
await loop.close_mcp()
|