tests/test_critical_fixes.py

"""Tests demonstrating critical fixes for TOIN/CCR implementation.

These tests verify the before/after behavior of critical bug fixes:
1. TOIN confidence math error (line 721)
2. TOIN double-count bug (lines 354-358)
3. compression_feedback.py race condition (lines 481-491)
4. Unbounded strategy dicts in compression_feedback.py
5. SmartCrusher integration with TOIN
"""

import time
from unittest.mock import patch

import pytest


class TestTOINConfidenceMathFix:
    """Test for CRITICAL: Confidence calculation math error in toin.py:721.

    BUG: `user_boost = min(0.3, pattern.user_count / 10 * 0.1)`
    Due to operator precedence: user_count / 10 * 0.1 = user_count * 0.01
    - 3 users: 0.03 boost (too small)
    - 10 users: 0.1 boost
    - 30 users needed to hit 0.3 cap!

    FIX: Should be `min(0.3, pattern.user_count * 0.03)` for meaningful boost
    - 3 users: 0.09 boost
    - 10 users: 0.3 boost (capped)
    """

    def test_confidence_user_boost_at_3_users(self):
        """With 3 users (min for network effect), boost should be meaningful."""
        from headroom.telemetry.toin import (
            TOINConfig,
            ToolIntelligenceNetwork,
            ToolPattern,
            reset_toin,
        )

        reset_toin()
        config = TOINConfig(min_users_for_network_effect=3)
        toin = ToolIntelligenceNetwork(config)

        # Create pattern with 3 users (correct API: tool_signature_hash is first arg)
        pattern = ToolPattern(
            tool_signature_hash="test123",
            user_count=3,
            sample_size=100,  # Good sample size
        )

        confidence = toin._calculate_confidence(pattern)

        # Sample confidence = min(0.7, 100/100) = 0.7
        # User boost for 3 users should be meaningful (>= 0.05)
        # FIX: With user_count * 0.03: boost = 0.09, total = 0.79
        # BUG: With user_count * 0.01: boost = 0.03, total = 0.73

        # After fix, confidence should be at least 0.75
        assert confidence >= 0.75, (
            f"Confidence {confidence} too low for 3 users - user boost not meaningful"
        )

    def test_confidence_user_boost_at_10_users(self):
        """With 10 users, boost should hit or approach cap."""
        from headroom.telemetry.toin import (
            TOINConfig,
            ToolIntelligenceNetwork,
            ToolPattern,
            reset_toin,
        )

        reset_toin()
        config = TOINConfig(min_users_for_network_effect=3)
        toin = ToolIntelligenceNetwork(config)

        pattern = ToolPattern(
            tool_signature_hash="test123",
            user_count=10,
            sample_size=100,
        )

        confidence = toin._calculate_confidence(pattern)

        # With 10 users, should be near cap (0.95)
        # Sample confidence = 0.7, user boost should be 0.3 (capped)
        # Total = min(0.95, 0.7 + 0.3) = 0.95
        # BUG: user_boost = 0.1, total = 0.8

        assert confidence >= 0.9, f"Confidence {confidence} too low for 10 users"


class TestTOINDoubleCountFix:
    """Test for CRITICAL: Double-count bug in toin.py:354-358.

    BUG: When _seen_instance_hashes hits cap (100), new instance_ids are NOT stored
    but user_count IS incremented. Next call with same instance_id:
    - `if self._instance_id not in pattern._seen_instance_hashes` → True (not stored!)
    - user_count incremented AGAIN → Double counting!

    FIX: Use a separate set to track ALL seen instances (no cap for lookup),
    OR check if we already tracked overflow for this instance.
    """

    def test_user_count_no_double_counting_after_cap(self):
        """Same instance shouldn't be counted twice even after cap hit."""
        from headroom.telemetry.models import ToolSignature
        from headroom.telemetry.toin import TOINConfig, ToolIntelligenceNetwork, reset_toin

        reset_toin()
        toin = ToolIntelligenceNetwork(TOINConfig())

        # Create a signature using the correct factory method
        items = [{"field1": "value1", "field2": 123}]
        sig = ToolSignature.from_items(items)

        # Simulate 101 unique instances (exceed the 100 cap)
        # First, fill up the cap with 100 unique instances
        original_instance_id = toin._instance_id
        for i in range(100):
            toin._instance_id = f"instance_{i}"
            toin.record_compression(sig, 100, 10, 1000, 100, strategy="test_strategy")

        # Now add one more instance (exceeds cap)
        toin._instance_id = "instance_100"
        toin.record_compression(sig, 100, 10, 1000, 100, strategy="test_strategy")

        # Get the pattern
        with toin._lock:
            pattern = toin._patterns[sig.structure_hash]
            user_count_after_101 = pattern.user_count

        # Now call again with same instance (instance_100)
        # BUG: This would increment user_count again because instance_100
        # was not stored (cap hit) so the check passes again
        toin.record_compression(sig, 100, 10, 1000, 100, strategy="test_strategy")

        with toin._lock:
            pattern = toin._patterns[sig.structure_hash]
            user_count_after_102 = pattern.user_count

        # Restore instance_id
        toin._instance_id = original_instance_id

        # User count should NOT increase for same instance
        assert user_count_after_102 == user_count_after_101, (
            f"Double-counting bug: user_count went from {user_count_after_101} to "
            f"{user_count_after_102} for same instance after cap hit"
        )


class TestCompressionFeedbackRaceCondition:
    """Test for CRITICAL: Race condition in compression_feedback.py:481-491.

    BUG: _last_event_timestamp is read (line 481) and written (line 491)
    WITHOUT holding the lock. Another thread calling record_retrieval()
    between these could cause events to be missed or double-counted.

    FIX: Move timestamp filtering and update inside the lock.
    """

    def test_analyze_from_store_thread_safety(self):
        """Concurrent analyze_from_store and record_retrieval should not lose events."""
        from headroom.cache.compression_feedback import (
            CompressionFeedback,
            reset_compression_feedback,
        )
        from headroom.cache.compression_store import CompressionStore, RetrievalEvent

        reset_compression_feedback()

        # Create store with mock events
        store = CompressionStore()
        feedback = CompressionFeedback(store=store, analysis_interval=0.0)  # No rate limiting

        # Pre-populate some events with correct API
        base_time = time.time()
        events_recorded = []

        def add_retrieval_event(tool_name: str, timestamp: float):
            event = RetrievalEvent(
                hash="test_hash",
                query=None,
                items_retrieved=10,
                total_items=100,
                tool_name=tool_name,
                timestamp=timestamp,
                retrieval_type="full",
            )
            # Directly add to feedback (simulating what analyze_from_store does)
            feedback.record_retrieval(event)
            events_recorded.append(event)

        # Record some events
        for i in range(10):
            add_retrieval_event(f"tool_{i % 3}", base_time + i)

        with feedback._lock:
            total_retrievals = feedback._total_retrievals
            patterns_count = len(feedback._tool_patterns)

        # All 10 events should be recorded
        assert total_retrievals == 10, f"Expected 10 retrievals, got {total_retrievals}"
        # Should have 3 unique tools (tool_0, tool_1, tool_2)
        assert patterns_count == 3, f"Expected 3 tool patterns, got {patterns_count}"

    def test_timestamp_filtering_inside_lock(self):
        """Verify that timestamp filtering happens atomically with update."""
        from headroom.cache.compression_feedback import (
            CompressionFeedback,
            reset_compression_feedback,
        )
        from headroom.cache.compression_store import CompressionStore, RetrievalEvent

        reset_compression_feedback()
        store = CompressionStore()
        feedback = CompressionFeedback(store=store, analysis_interval=0.0)

        # Manually set last event timestamp
        feedback._last_event_timestamp = 100.0

        # Create mock store with events (correct API)
        mock_events = [
            RetrievalEvent(
                hash="h1",
                query=None,
                items_retrieved=5,
                total_items=50,
                tool_name="tool_a",
                timestamp=99.0,
                retrieval_type="full",
            ),
            RetrievalEvent(
                hash="h2",
                query=None,
                items_retrieved=5,
                total_items=50,
                tool_name="tool_b",
                timestamp=101.0,
                retrieval_type="full",
            ),
            RetrievalEvent(
                hash="h3",
                query="test",
                items_retrieved=5,
                total_items=50,
                tool_name="tool_c",
                timestamp=102.0,
                retrieval_type="search",
            ),
        ]

        # Mock store.get_retrieval_events
        with patch.object(store, "get_retrieval_events", return_value=mock_events):
            feedback.analyze_from_store()

        # Only events with timestamp > 100.0 should be processed (h2, h3)
        with feedback._lock:
            total = feedback._total_retrievals
            # The timestamp should now be 102.0 (max of processed events)
            last_ts = feedback._last_event_timestamp

        assert total == 2, f"Expected 2 new events processed, got {total}"
        assert last_ts == 102.0, f"Expected last_event_timestamp=102.0, got {last_ts}"


class TestUnboundedStrategyDicts:
    """Test for HIGH: Unbounded strategy_compressions/strategy_retrievals dicts.

    BUG: Unlike common_queries (truncated at 100) and queried_fields (truncated at 50),
    the strategy dicts have no size limits and could grow unbounded.

    FIX: Add truncation logic similar to other dicts.
    """

    def test_strategy_dicts_have_size_limits(self):
        """Strategy dicts should be bounded to prevent memory leaks."""
        from headroom.cache.compression_feedback import (
            CompressionFeedback,
            reset_compression_feedback,
        )
        from headroom.cache.compression_store import CompressionStore

        reset_compression_feedback()
        store = CompressionStore()
        feedback = CompressionFeedback(store=store)

        # Record many compressions with different strategies
        for i in range(200):
            feedback.record_compression(
                tool_name="test_tool",
                original_count=100,
                compressed_count=10,
                strategy=f"strategy_{i}",  # 200 unique strategies
            )

        with feedback._lock:
            pattern = feedback._tool_patterns.get("test_tool")
            strategy_count = len(pattern.strategy_compressions) if pattern else 0

        # Strategy dict should be bounded (e.g., to 50 like queried_fields)
        assert strategy_count <= 50, (
            f"strategy_compressions has {strategy_count} entries, should be <= 50"
        )


class TestSmartCrusherTOINIntegration:
    """Test for CRITICAL: SmartCrusher not calling toin.record_compression().

    BUG: SmartCrusher calls feedback.record_compression() but never calls
    toin.record_compression(). This means TOIN only learns from retrieval events,
    not from compression events - breaking the feedback loop.

    FIX: Add toin.record_compression() call after compression in SmartCrusher.
    """

    def test_smart_crusher_records_to_toin(self):
        """SmartCrusher should record compression events to TOIN."""
        from headroom.telemetry.models import ToolSignature
        from headroom.telemetry.toin import get_toin, reset_toin
        from headroom.transforms.smart_crusher import SmartCrusher, SmartCrusherConfig

        reset_toin()

        config = SmartCrusherConfig(
            min_items_to_analyze=5,
            max_items_after_crush=10,
            use_feedback_hints=True,
        )
        crusher = SmartCrusher(config)

        # Create test items that look like search results with a clear score field
        # This pattern is crushable because:
        # 1. Has a clear numeric score field in BOUNDED range [0,1]
        # 2. Has repeated structure with some constant fields (type, language)
        # 3. Score values vary within the bounded range
        items = [
            {
                "name": f"repo_{i}",
                "relevance_score": (50 - i) / 50.0,  # Bounded [0,1] - descending order
                "type": "repository",  # Constant field
                "language": "python" if i % 3 == 0 else "javascript",  # Low cardinality
                "description": f"Description {i % 5}",  # Low cardinality
            }
            for i in range(50)
        ]

        # Get TOIN instance and check initial state
        toin = get_toin()
        len(toin._patterns)

        # Crush the array
        result, info, markers, _summary = crusher._crush_array(
            items, query_context="test query", tool_name="test_tool"
        )

        # Verify compression happened (not skipped)
        assert "skip" not in info.lower(), (
            f"Compression was skipped: {info}. Test needs crushable data."
        )

        # Get the signature that would have been created
        sig = ToolSignature.from_items(items)

        # Check TOIN was notified
        with toin._lock:
            pattern = toin._patterns.get(sig.structure_hash)

        # After fix, TOIN should have a pattern for this tool's signature
        assert pattern is not None, (
            f"TOIN should have recorded the compression event. "
            f"Info: {info}, pattern count: {len(toin._patterns)}"
        )
        if pattern:
            assert pattern.total_compressions >= 1, (
                f"Pattern should have at least 1 compression recorded, got {pattern.total_compressions}"
            )


class TestAllFixesIntegrated:
    """Integration tests ensuring all fixes work together."""

    def test_full_feedback_loop(self):
        """Test complete feedback loop: compress -> store -> retrieve -> learn."""
        from headroom.cache.compression_feedback import (
            reset_compression_feedback,
        )
        from headroom.cache.compression_store import reset_compression_store
        from headroom.telemetry.models import ToolSignature
        from headroom.telemetry.toin import get_toin, reset_toin
        from headroom.transforms.smart_crusher import SmartCrusher, SmartCrusherConfig

        # Reset all singletons
        reset_toin()
        reset_compression_store()
        reset_compression_feedback()

        # Setup
        config = SmartCrusherConfig(
            min_items_to_analyze=5,
            max_items_after_crush=10,
            use_feedback_hints=True,
        )
        crusher = SmartCrusher(config)

        # Create test items that look like API responses with scoring
        # This pattern is crushable because:
        # 1. Has a clear numeric score field in BOUNDED range [0,1]
        # 2. Has constant fields (status, type)
        # 3. Has enough items for compression (100)
        items = [
            {
                "priority": (100 - i) / 100.0,  # Bounded [0,1] - descending order
                "status": "ok",  # Constant field
                "type": "response",  # Constant field
                "data": f"content_{i % 10}",  # Low cardinality (only 10 unique values)
            }
            for i in range(100)
        ]

        # Step 1: Compress
        result, info, markers, _summary = crusher._crush_array(
            items, query_context="find status", tool_name="api_response"
        )

        # Verify compression happened (not skipped)
        assert "skip" not in info.lower(), (
            f"Compression was skipped: {info}. Test needs crushable data."
        )

        # Step 2: Check TOIN was notified (after fix)
        toin = get_toin()
        sig = ToolSignature.from_items(items)

        with toin._lock:
            toin_pattern = toin._patterns.get(sig.structure_hash)

        # After fix, TOIN should have the pattern
        assert toin_pattern is not None, (
            f"TOIN should have learned from the compression event. Info: {info}"
        )
        assert toin_pattern.total_compressions >= 1, (
            f"TOIN pattern should have recorded compression, got {toin_pattern.total_compressions}"
        )


# Run specific test to verify fix
if __name__ == "__main__":
    pytest.main([__file__, "-v", "--tb=short"])