test: add unit tests for 8 modules (batch 2)

tests/agent/test_context_compressor.py

@@ -0,0 +1,136 @@
+"""Tests for agent/context_compressor.py — compression logic, thresholds, truncation fallback."""
+
+import pytest
+from unittest.mock import patch, MagicMock
+
+from agent.context_compressor import ContextCompressor
+
+
+@pytest.fixture()
+def compressor():
+    """Create a ContextCompressor with mocked dependencies."""
+    with patch("agent.context_compressor.get_model_context_length", return_value=100000), \
+         patch("agent.context_compressor.get_text_auxiliary_client", return_value=(None, None)):
+        c = ContextCompressor(
+            model="test/model",
+            threshold_percent=0.85,
+            protect_first_n=2,
+            protect_last_n=2,
+            quiet_mode=True,
+        )
+        return c
+
+
+class TestShouldCompress:
+    def test_below_threshold(self, compressor):
+        compressor.last_prompt_tokens = 50000
+        assert compressor.should_compress() is False
+
+    def test_above_threshold(self, compressor):
+        compressor.last_prompt_tokens = 90000
+        assert compressor.should_compress() is True
+
+    def test_exact_threshold(self, compressor):
+        compressor.last_prompt_tokens = 85000
+        assert compressor.should_compress() is True
+
+    def test_explicit_tokens(self, compressor):
+        assert compressor.should_compress(prompt_tokens=90000) is True
+        assert compressor.should_compress(prompt_tokens=50000) is False
+
+
+class TestShouldCompressPreflight:
+    def test_short_messages(self, compressor):
+        msgs = [{"role": "user", "content": "short"}]
+        assert compressor.should_compress_preflight(msgs) is False
+
+    def test_long_messages(self, compressor):
+        # Each message ~100k chars / 4 = 25k tokens, need >85k threshold
+        msgs = [{"role": "user", "content": "x" * 400000}]
+        assert compressor.should_compress_preflight(msgs) is True
+
+
+class TestUpdateFromResponse:
+    def test_updates_fields(self, compressor):
+        compressor.update_from_response({
+            "prompt_tokens": 5000,
+            "completion_tokens": 1000,
+            "total_tokens": 6000,
+        })
+        assert compressor.last_prompt_tokens == 5000
+        assert compressor.last_completion_tokens == 1000
+        assert compressor.last_total_tokens == 6000
+
+    def test_missing_fields_default_zero(self, compressor):
+        compressor.update_from_response({})
+        assert compressor.last_prompt_tokens == 0
+
+
+class TestGetStatus:
+    def test_returns_expected_keys(self, compressor):
+        status = compressor.get_status()
+        assert "last_prompt_tokens" in status
+        assert "threshold_tokens" in status
+        assert "context_length" in status
+        assert "usage_percent" in status
+        assert "compression_count" in status
+
+    def test_usage_percent_calculation(self, compressor):
+        compressor.last_prompt_tokens = 50000
+        status = compressor.get_status()
+        assert status["usage_percent"] == 50.0
+
+
+class TestCompress:
+    def _make_messages(self, n):
+        return [{"role": "user" if i % 2 == 0 else "assistant", "content": f"msg {i}"} for i in range(n)]
+
+    def test_too_few_messages_returns_unchanged(self, compressor):
+        msgs = self._make_messages(4)  # protect_first=2 + protect_last=2 + 1 = 5 needed
+        result = compressor.compress(msgs)
+        assert result == msgs
+
+    def test_truncation_fallback_no_client(self, compressor):
+        # compressor has client=None, so should use truncation fallback
+        msgs = [{"role": "system", "content": "System prompt"}] + self._make_messages(10)
+        result = compressor.compress(msgs)
+        assert len(result) < len(msgs)
+        # Should keep system message and last N
+        assert result[0]["role"] == "system"
+        assert compressor.compression_count == 1
+
+    def test_compression_increments_count(self, compressor):
+        msgs = self._make_messages(10)
+        compressor.compress(msgs)
+        assert compressor.compression_count == 1
+        compressor.compress(msgs)
+        assert compressor.compression_count == 2
+
+    def test_protects_first_and_last(self, compressor):
+        msgs = self._make_messages(10)
+        result = compressor.compress(msgs)
+        # First 2 messages should be preserved (protect_first_n=2)
+        # Last 2 messages should be preserved (protect_last_n=2)
+        assert result[-1]["content"] == msgs[-1]["content"]
+        assert result[-2]["content"] == msgs[-2]["content"]
+
+
+class TestCompressWithClient:
+    def test_summarization_path(self):
+        mock_client = MagicMock()
+        mock_response = MagicMock()
+        mock_response.choices = [MagicMock()]
+        mock_response.choices[0].message.content = "[CONTEXT SUMMARY]: stuff happened"
+        mock_client.chat.completions.create.return_value = mock_response
+
+        with patch("agent.context_compressor.get_model_context_length", return_value=100000), \
+             patch("agent.context_compressor.get_text_auxiliary_client", return_value=(mock_client, "test-model")):
+            c = ContextCompressor(model="test", quiet_mode=True)
+
+        msgs = [{"role": "user" if i % 2 == 0 else "assistant", "content": f"msg {i}"} for i in range(10)]
+        result = c.compress(msgs)
+
+        # Should have summary message in the middle
+        contents = [m.get("content", "") for m in result]
+        assert any("CONTEXT SUMMARY" in c for c in contents)
+        assert len(result) < len(msgs)

tests/agent/test_prompt_caching.py

@@ -0,0 +1,128 @@
+"""Tests for agent/prompt_caching.py — Anthropic cache control injection."""
+
+import copy
+import pytest
+
+from agent.prompt_caching import (
+    _apply_cache_marker,
+    apply_anthropic_cache_control,
+)
+
+
+MARKER = {"type": "ephemeral"}
+
+
+class TestApplyCacheMarker:
+    def test_tool_message_gets_top_level_marker(self):
+        msg = {"role": "tool", "content": "result"}
+        _apply_cache_marker(msg, MARKER)
+        assert msg["cache_control"] == MARKER
+
+    def test_none_content_gets_top_level_marker(self):
+        msg = {"role": "assistant", "content": None}
+        _apply_cache_marker(msg, MARKER)
+        assert msg["cache_control"] == MARKER
+
+    def test_string_content_wrapped_in_list(self):
+        msg = {"role": "user", "content": "Hello"}
+        _apply_cache_marker(msg, MARKER)
+        assert isinstance(msg["content"], list)
+        assert len(msg["content"]) == 1
+        assert msg["content"][0]["type"] == "text"
+        assert msg["content"][0]["text"] == "Hello"
+        assert msg["content"][0]["cache_control"] == MARKER
+
+    def test_list_content_last_item_gets_marker(self):
+        msg = {
+            "role": "user",
+            "content": [
+                {"type": "text", "text": "First"},
+                {"type": "text", "text": "Second"},
+            ],
+        }
+        _apply_cache_marker(msg, MARKER)
+        assert "cache_control" not in msg["content"][0]
+        assert msg["content"][1]["cache_control"] == MARKER
+
+    def test_empty_list_content_no_crash(self):
+        msg = {"role": "user", "content": []}
+        # Should not crash on empty list
+        _apply_cache_marker(msg, MARKER)
+
+
+class TestApplyAnthropicCacheControl:
+    def test_empty_messages(self):
+        result = apply_anthropic_cache_control([])
+        assert result == []
+
+    def test_returns_deep_copy(self):
+        msgs = [{"role": "user", "content": "Hello"}]
+        result = apply_anthropic_cache_control(msgs)
+        assert result is not msgs
+        assert result[0] is not msgs[0]
+        # Original should be unmodified
+        assert "cache_control" not in msgs[0].get("content", "")
+
+    def test_system_message_gets_marker(self):
+        msgs = [
+            {"role": "system", "content": "You are helpful"},
+            {"role": "user", "content": "Hi"},
+        ]
+        result = apply_anthropic_cache_control(msgs)
+        # System message should have cache_control
+        sys_content = result[0]["content"]
+        assert isinstance(sys_content, list)
+        assert sys_content[0]["cache_control"]["type"] == "ephemeral"
+
+    def test_last_3_non_system_get_markers(self):
+        msgs = [
+            {"role": "system", "content": "System"},
+            {"role": "user", "content": "msg1"},
+            {"role": "assistant", "content": "msg2"},
+            {"role": "user", "content": "msg3"},
+            {"role": "assistant", "content": "msg4"},
+        ]
+        result = apply_anthropic_cache_control(msgs)
+        # System (index 0) + last 3 non-system (indices 2, 3, 4) = 4 breakpoints
+        # Index 1 (msg1) should NOT have marker
+        content_1 = result[1]["content"]
+        if isinstance(content_1, str):
+            assert True  # No marker applied (still a string)
+        else:
+            assert "cache_control" not in content_1[0]
+
+    def test_no_system_message(self):
+        msgs = [
+            {"role": "user", "content": "Hello"},
+            {"role": "assistant", "content": "Hi"},
+        ]
+        result = apply_anthropic_cache_control(msgs)
+        # Both should get markers (4 slots available, only 2 messages)
+        assert len(result) == 2
+
+    def test_1h_ttl(self):
+        msgs = [{"role": "system", "content": "System prompt"}]
+        result = apply_anthropic_cache_control(msgs, cache_ttl="1h")
+        sys_content = result[0]["content"]
+        assert isinstance(sys_content, list)
+        assert sys_content[0]["cache_control"]["ttl"] == "1h"
+
+    def test_max_4_breakpoints(self):
+        msgs = [
+            {"role": "system", "content": "System"},
+        ] + [
+            {"role": "user" if i % 2 == 0 else "assistant", "content": f"msg{i}"}
+            for i in range(10)
+        ]
+        result = apply_anthropic_cache_control(msgs)
+        # Count how many messages have cache_control
+        count = 0
+        for msg in result:
+            content = msg.get("content")
+            if isinstance(content, list):
+                for item in content:
+                    if isinstance(item, dict) and "cache_control" in item:
+                        count += 1
+            elif "cache_control" in msg:
+                count += 1
+        assert count <= 4

tests/cron/test_scheduler.py

@@ -0,0 +1,36 @@
+"""Tests for cron/scheduler.py — origin resolution and delivery routing."""
+
+import pytest
+
+from cron.scheduler import _resolve_origin
+
+
+class TestResolveOrigin:
+    def test_full_origin(self):
+        job = {
+            "origin": {
+                "platform": "telegram",
+                "chat_id": "123456",
+                "chat_name": "Test Chat",
+            }
+        }
+        result = _resolve_origin(job)
+        assert result is not None
+        assert result["platform"] == "telegram"
+        assert result["chat_id"] == "123456"
+
+    def test_no_origin(self):
+        assert _resolve_origin({}) is None
+        assert _resolve_origin({"origin": None}) is None
+
+    def test_missing_platform(self):
+        job = {"origin": {"chat_id": "123"}}
+        assert _resolve_origin(job) is None
+
+    def test_missing_chat_id(self):
+        job = {"origin": {"platform": "telegram"}}
+        assert _resolve_origin(job) is None
+
+    def test_empty_origin(self):
+        job = {"origin": {}}
+        assert _resolve_origin(job) is None