apps/coordinator/tests/test_context_compaction.py

@@ -0,0 +1,330 @@
+"""Tests for context compaction functionality.
+
+Context compaction reduces memory usage by:
+1. Requesting a summary from the agent of completed work
+2. Replacing conversation history with concise summary
+3. Measuring context reduction achieved
+"""
+
+import pytest
+from unittest.mock import AsyncMock, MagicMock
+
+from src.context_compaction import ContextCompactor, CompactionResult
+from src.models import ContextUsage
+
+
+class TestContextCompactor:
+    """Test ContextCompactor class."""
+
+    @pytest.fixture
+    def mock_api_client(self) -> AsyncMock:
+        """Mock Claude API client."""
+        mock = AsyncMock()
+        return mock
+
+    @pytest.fixture
+    def compactor(self, mock_api_client: AsyncMock) -> ContextCompactor:
+        """Create ContextCompactor instance with mocked API."""
+        return ContextCompactor(api_client=mock_api_client)
+
+    @pytest.mark.asyncio
+    async def test_generate_summary_prompt(
+        self, compactor: ContextCompactor, mock_api_client: AsyncMock
+    ) -> None:
+        """Should generate prompt asking agent to summarize work."""
+        # Mock API response
+        mock_api_client.send_message.return_value = {
+            "content": "Completed task X. Found pattern Y. Decision: use approach Z.",
+            "usage": {"input_tokens": 150000, "output_tokens": 100},
+        }
+
+        summary = await compactor.request_summary("agent-1")
+
+        # Verify API was called with summarization prompt
+        mock_api_client.send_message.assert_called_once()
+        call_args = mock_api_client.send_message.call_args
+        assert call_args[0][0] == "agent-1"  # agent_id
+        prompt = call_args[0][1]  # message
+
+        # Verify prompt asks for summary
+        assert "summarize" in prompt.lower() or "summary" in prompt.lower()
+        assert "completed work" in prompt.lower() or "work completed" in prompt.lower()
+        assert summary == "Completed task X. Found pattern Y. Decision: use approach Z."
+
+    @pytest.mark.asyncio
+    async def test_compact_conversation_history(
+        self, compactor: ContextCompactor, mock_api_client: AsyncMock
+    ) -> None:
+        """Should replace conversation history with summary."""
+        # Mock getting context before and after compaction
+        mock_api_client.get_context_usage.side_effect = [
+            {"used_tokens": 160000, "total_tokens": 200000},  # Before
+            {"used_tokens": 80000, "total_tokens": 200000},   # After
+        ]
+
+        # Mock getting summary
+        mock_api_client.send_message.return_value = {
+            "content": "Work summary here",
+            "usage": {"input_tokens": 160000, "output_tokens": 50},
+        }
+
+        # Mock replacing conversation history
+        mock_api_client.replace_history.return_value = None
+
+        result = await compactor.compact("agent-1")
+
+        # Verify history was replaced
+        mock_api_client.replace_history.assert_called_once_with(
+            "agent-1", "Work summary here"
+        )
+
+        # Verify result contains before/after metrics
+        assert isinstance(result, CompactionResult)
+        assert result.agent_id == "agent-1"
+        assert result.before_tokens == 160000
+        assert result.after_tokens == 80000
+
+    @pytest.mark.asyncio
+    async def test_measure_context_reduction(
+        self, compactor: ContextCompactor, mock_api_client: AsyncMock
+    ) -> None:
+        """Should measure context reduction achieved."""
+        # Mock context before compaction (80%)
+        mock_api_client.get_context_usage.side_effect = [
+            {"used_tokens": 160000, "total_tokens": 200000},  # Before
+            {"used_tokens": 80000, "total_tokens": 200000},  # After
+        ]
+
+        mock_api_client.send_message.return_value = {
+            "content": "Summary",
+            "usage": {"input_tokens": 160000, "output_tokens": 50},
+        }
+
+        mock_api_client.replace_history.return_value = {
+            "used_tokens": 80000,
+            "total_tokens": 200000,
+        }
+
+        result = await compactor.compact("agent-1")
+
+        # Verify reduction metrics
+        assert result.before_tokens == 160000
+        assert result.after_tokens == 80000
+        assert result.tokens_freed == 80000
+        assert result.reduction_percent == 50.0  # 50% reduction
+
+    @pytest.mark.asyncio
+    async def test_compaction_achieves_target_reduction(
+        self, compactor: ContextCompactor, mock_api_client: AsyncMock
+    ) -> None:
+        """Should achieve 40-50% context reduction target."""
+        # Mock 80% usage before compaction
+        mock_api_client.get_context_usage.side_effect = [
+            {"used_tokens": 160000, "total_tokens": 200000},  # 80% before
+            {"used_tokens": 88000, "total_tokens": 200000},  # 45% reduction (target)
+        ]
+
+        mock_api_client.send_message.return_value = {
+            "content": "Summary of work",
+            "usage": {"input_tokens": 160000, "output_tokens": 75},
+        }
+
+        mock_api_client.replace_history.return_value = {
+            "used_tokens": 88000,
+            "total_tokens": 200000,
+        }
+
+        result = await compactor.compact("agent-1")
+
+        # Verify target reduction achieved
+        assert result.reduction_percent >= 40.0
+        assert result.reduction_percent <= 50.0
+        assert result.success is True
+
+    @pytest.mark.asyncio
+    async def test_log_compaction_metrics(
+        self, compactor: ContextCompactor, mock_api_client: AsyncMock
+    ) -> None:
+        """Should log before/after metrics."""
+        mock_api_client.get_context_usage.side_effect = [
+            {"used_tokens": 160000, "total_tokens": 200000},
+            {"used_tokens": 90000, "total_tokens": 200000},
+        ]
+
+        mock_api_client.send_message.return_value = {
+            "content": "Summary",
+            "usage": {"input_tokens": 160000, "output_tokens": 50},
+        }
+
+        mock_api_client.replace_history.return_value = {
+            "used_tokens": 90000,
+            "total_tokens": 200000,
+        }
+
+        result = await compactor.compact("agent-1")
+
+        # Verify logging information present in result
+        assert result.before_tokens == 160000
+        assert result.after_tokens == 90000
+        assert result.before_percent == 80.0
+        assert result.after_percent == 45.0
+
+    @pytest.mark.asyncio
+    async def test_compaction_handles_api_errors(
+        self, compactor: ContextCompactor, mock_api_client: AsyncMock
+    ) -> None:
+        """Should handle API errors gracefully."""
+        # Mock API error during summary request
+        mock_api_client.get_context_usage.return_value = {
+            "used_tokens": 160000,
+            "total_tokens": 200000,
+        }
+
+        mock_api_client.send_message.side_effect = Exception("API timeout")
+
+        result = await compactor.compact("agent-1")
+
+        # Should return failed result, not crash
+        assert result.success is False
+        assert "API timeout" in result.error_message
+
+    @pytest.mark.asyncio
+    async def test_compaction_validates_reduction_achieved(
+        self, compactor: ContextCompactor, mock_api_client: AsyncMock
+    ) -> None:
+        """Should validate that context reduction was actually achieved."""
+        # Mock insufficient reduction (only 10% freed)
+        mock_api_client.get_context_usage.side_effect = [
+            {"used_tokens": 160000, "total_tokens": 200000},  # Before: 80%
+            {"used_tokens": 144000, "total_tokens": 200000},  # After: 72% (only 10% freed)
+        ]
+
+        mock_api_client.send_message.return_value = {
+            "content": "Brief summary",
+            "usage": {"input_tokens": 160000, "output_tokens": 30},
+        }
+
+        mock_api_client.replace_history.return_value = {
+            "used_tokens": 144000,
+            "total_tokens": 200000,
+        }
+
+        result = await compactor.compact("agent-1")
+
+        # Should still succeed but report low reduction
+        assert result.success is True
+        assert result.reduction_percent == 10.0
+        assert result.tokens_freed == 16000
+
+    @pytest.mark.asyncio
+    async def test_generate_concise_summary(
+        self, compactor: ContextCompactor, mock_api_client: AsyncMock
+    ) -> None:
+        """Should generate concise summary of completed work."""
+        mock_api_client.send_message.return_value = {
+            "content": (
+                "Implemented feature X using pattern Y. "
+                "Key decisions: chose approach Z over W because of performance. "
+                "Discovered issue with dependency A, fixed by upgrading to version B."
+            ),
+            "usage": {"input_tokens": 150000, "output_tokens": 80},
+        }
+
+        summary = await compactor.request_summary("agent-1")
+
+        # Verify summary contains key information
+        assert "Implemented" in summary
+        assert "pattern" in summary
+        assert "decisions" in summary or "Decision" in summary
+        assert len(summary) > 50  # Should have substance
+
+    @pytest.mark.asyncio
+    async def test_summary_prompt_includes_context(
+        self, compactor: ContextCompactor, mock_api_client: AsyncMock
+    ) -> None:
+        """Should include context about what to summarize."""
+        mock_api_client.send_message.return_value = {
+            "content": "Summary text",
+            "usage": {"input_tokens": 100, "output_tokens": 50},
+        }
+
+        await compactor.request_summary("agent-1")
+
+        call_args = mock_api_client.send_message.call_args
+        prompt = call_args[0][1]
+
+        # Verify prompt asks for specific things
+        assert any(
+            word in prompt.lower()
+            for word in ["pattern", "decision", "approach", "key finding"]
+        )
+
+
+class TestCompactionResult:
+    """Test CompactionResult data class."""
+
+    def test_calculate_reduction_percent(self) -> None:
+        """Should calculate reduction percentage correctly."""
+        result = CompactionResult(
+            agent_id="agent-1",
+            before_tokens=160000,
+            after_tokens=80000,
+            before_percent=80.0,
+            after_percent=40.0,
+            tokens_freed=80000,
+            reduction_percent=50.0,
+            success=True,
+        )
+
+        assert result.reduction_percent == 50.0
+        assert result.tokens_freed == 80000
+
+    def test_success_flag_true_on_good_reduction(self) -> None:
+        """Should mark success=True when reduction is achieved."""
+        result = CompactionResult(
+            agent_id="agent-1",
+            before_tokens=160000,
+            after_tokens=88000,
+            before_percent=80.0,
+            after_percent=44.0,
+            tokens_freed=72000,
+            reduction_percent=45.0,
+            success=True,
+        )
+
+        assert result.success is True
+
+    def test_success_flag_false_on_error(self) -> None:
+        """Should mark success=False on errors."""
+        result = CompactionResult(
+            agent_id="agent-1",
+            before_tokens=160000,
+            after_tokens=160000,  # No reduction
+            before_percent=80.0,
+            after_percent=80.0,
+            tokens_freed=0,
+            reduction_percent=0.0,
+            success=False,
+            error_message="API timeout",
+        )
+
+        assert result.success is False
+        assert result.error_message == "API timeout"
+
+    def test_repr_includes_key_metrics(self) -> None:
+        """Should provide readable string representation."""
+        result = CompactionResult(
+            agent_id="agent-1",
+            before_tokens=160000,
+            after_tokens=80000,
+            before_percent=80.0,
+            after_percent=40.0,
+            tokens_freed=80000,
+            reduction_percent=50.0,
+            success=True,
+        )
+
+        repr_str = repr(result)
+        assert "agent-1" in repr_str
+        assert "50.0%" in repr_str or "50%" in repr_str
+        assert "success" in repr_str.lower()

apps/coordinator/tests/test_context_monitor.py

@@ -319,6 +319,115 @@ class TestContextMonitor:
         # Should not have increased
         assert len(callback_calls) == initial_count
 
+    @pytest.mark.asyncio
+    async def test_perform_compaction_when_triggered(
+        self, mock_claude_api: AsyncMock
+    ) -> None:
+        """Should perform compaction when COMPACT action is triggered."""
+        from unittest.mock import patch
+        from src.context_compaction import CompactionResult
+
+        # Mock compaction result
+        mock_compaction_result = CompactionResult(
+            agent_id="agent-1",
+            before_tokens=164000,
+            after_tokens=90000,
+            before_percent=82.0,
+            after_percent=45.0,
+            tokens_freed=74000,
+            reduction_percent=45.1,
+            success=True,
+        )
+
+        with patch("src.context_monitor.ContextCompactor") as mock_compactor_class:
+            mock_compactor = mock_compactor_class.return_value
+            mock_compactor.compact = AsyncMock(return_value=mock_compaction_result)
+
+            # Create monitor with patched compactor
+            monitor = ContextMonitor(api_client=mock_claude_api, poll_interval=0.1)
+
+            # Mock 82% usage (triggers COMPACT)
+            mock_claude_api.get_context_usage.return_value = {
+                "used_tokens": 164000,
+                "total_tokens": 200000,
+            }
+
+            # Trigger compaction
+            compaction_result = await monitor.trigger_compaction("agent-1")
+
+            # Verify compactor was called
+            mock_compactor.compact.assert_called_once_with("agent-1")
+            assert compaction_result == mock_compaction_result
+
+    @pytest.mark.asyncio
+    async def test_compaction_logs_metrics(
+        self, mock_claude_api: AsyncMock
+    ) -> None:
+        """Should log compaction metrics when compaction is performed."""
+        from unittest.mock import patch
+        from src.context_compaction import CompactionResult
+
+        mock_compaction_result = CompactionResult(
+            agent_id="agent-1",
+            before_tokens=164000,
+            after_tokens=82000,
+            before_percent=82.0,
+            after_percent=41.0,
+            tokens_freed=82000,
+            reduction_percent=50.0,
+            success=True,
+        )
+
+        with patch("src.context_monitor.ContextCompactor") as mock_compactor_class:
+            mock_compactor = mock_compactor_class.return_value
+            mock_compactor.compact = AsyncMock(return_value=mock_compaction_result)
+
+            monitor = ContextMonitor(api_client=mock_claude_api, poll_interval=0.1)
+
+            mock_claude_api.get_context_usage.return_value = {
+                "used_tokens": 164000,
+                "total_tokens": 200000,
+            }
+
+            result = await monitor.trigger_compaction("agent-1")
+
+            # Verify result contains metrics
+            assert result.reduction_percent == 50.0
+            assert result.tokens_freed == 82000
+            assert result.success is True
+
+    @pytest.mark.asyncio
+    async def test_compaction_handles_failure(
+        self, mock_claude_api: AsyncMock
+    ) -> None:
+        """Should handle compaction failure and log error."""
+        from unittest.mock import patch
+        from src.context_compaction import CompactionResult
+
+        mock_compaction_result = CompactionResult(
+            agent_id="agent-1",
+            before_tokens=0,
+            after_tokens=0,
+            before_percent=0.0,
+            after_percent=0.0,
+            tokens_freed=0,
+            reduction_percent=0.0,
+            success=False,
+            error_message="API timeout during compaction",
+        )
+
+        with patch("src.context_monitor.ContextCompactor") as mock_compactor_class:
+            mock_compactor = mock_compactor_class.return_value
+            mock_compactor.compact = AsyncMock(return_value=mock_compaction_result)
+
+            monitor = ContextMonitor(api_client=mock_claude_api, poll_interval=0.1)
+
+            result = await monitor.trigger_compaction("agent-1")
+
+            # Verify failure is reported
+            assert result.success is False
+            assert result.error_message == "API timeout during compaction"
+
 
 class TestIssueMetadata:
     """Test IssueMetadata model."""