fix(agent): add time-based recovery to compression anti-thrashing protection (#14694)

The anti-thrashing guard permanently disabled auto-compression after 2
ineffective compressions (<10% savings each), with no way to recover
during a long session. Add a 300-second cooldown: once enough time has
elapsed since the last compression, reset the ineffective counter so
the session gets another chance. This preserves the rapid-fire guard
while allowing recovery when significant new context has accumulated.

agent/context_compressor.py

@@ -297,6 +297,7 @@ class ContextCompressor(ContextEngine):
         self._last_summary_error = None
         self._last_compression_savings_pct = 100.0
         self._ineffective_compression_count = 0
+        self._last_compression_time = 0.0
 
     def update_model(
         self,
@@ -389,6 +390,7 @@ class ContextCompressor(ContextEngine):
         # Anti-thrashing: track whether last compression was effective
         self._last_compression_savings_pct: float = 100.0
         self._ineffective_compression_count: int = 0
+        self._last_compression_time: float = 0.0
         self._summary_failure_cooldown_until: float = 0.0
         self._last_summary_error: Optional[str] = None
 
@@ -407,6 +409,18 @@ class ContextCompressor(ContextEngine):
         tokens = prompt_tokens if prompt_tokens is not None else self.last_prompt_tokens
         if tokens < self.threshold_tokens:
             return False
+        # Anti-thrashing: time-based recovery — if enough time has passed since
+        # the last compression, reset the ineffective counter so the session
+        # gets another chance.  300s is enough for significant new context to
+        # accumulate, making another attempt worthwhile.
+        if (self._ineffective_compression_count >= 2
+                and self._last_compression_time > 0
+                and time.monotonic() - self._last_compression_time >= 300):
+            self._ineffective_compression_count = 0
+            if not self.quiet_mode:
+                logger.info(
+                    "Anti-thrashing cooldown expired — re-enabling auto-compression"
+                )
         # Anti-thrashing: back off if recent compressions were ineffective
         if self._ineffective_compression_count >= 2:
             if not self.quiet_mode:
@@ -1279,6 +1293,7 @@ The user has requested that this compaction PRIORITISE preserving all informatio
         saved_estimate = display_tokens - new_estimate
 
         # Anti-thrashing: track compression effectiveness
+        self._last_compression_time = time.monotonic()
         savings_pct = (saved_estimate / display_tokens * 100) if display_tokens > 0 else 0
         self._last_compression_savings_pct = savings_pct
         if savings_pct < 10:

tests/agent/test_context_compressor.py

@@ -1,5 +1,7 @@
 """Tests for agent/context_compressor.py — compression logic, thresholds, truncation fallback."""
 
+import time
+
 import pytest
 from unittest.mock import patch, MagicMock
 
@@ -38,6 +40,47 @@ class TestShouldCompress:
         assert compressor.should_compress(prompt_tokens=50000) is False
 
 
+class TestAntiThrashingRecovery:
+    """Anti-thrashing protection should recover after a cooldown period (#14694)."""
+
+    def test_anti_thrash_blocks_after_two_ineffective(self, compressor):
+        """Two ineffective compressions should block further auto-compression."""
+        compressor._ineffective_compression_count = 2
+        compressor.last_prompt_tokens = 90000
+        assert compressor.should_compress() is False
+
+    def test_anti_thrash_recovers_after_cooldown(self, compressor):
+        """After 300s cooldown, the anti-thrashing counter resets and
+        compression is re-enabled.  This test fails without the fix."""
+        compressor._ineffective_compression_count = 2
+        compressor._last_compression_time = time.monotonic() - 301
+        compressor.last_prompt_tokens = 90000
+        assert compressor.should_compress() is True
+        assert compressor._ineffective_compression_count == 0
+
+    def test_anti_thrash_still_blocks_within_cooldown(self, compressor):
+        """Within 300s the block should remain active."""
+        compressor._ineffective_compression_count = 2
+        compressor._last_compression_time = time.monotonic() - 60
+        compressor.last_prompt_tokens = 90000
+        assert compressor.should_compress() is False
+
+    def test_anti_thrash_no_recovery_without_timestamp(self, compressor):
+        """If no compression has ever run (_last_compression_time == 0),
+        the time-based recovery should not trigger."""
+        compressor._ineffective_compression_count = 2
+        compressor._last_compression_time = 0.0
+        compressor.last_prompt_tokens = 90000
+        assert compressor.should_compress() is False
+
+    def test_session_reset_clears_compression_time(self, compressor):
+        """on_session_reset() should reset _last_compression_time."""
+        compressor._last_compression_time = 12345.0
+        compressor._ineffective_compression_count = 2
+        compressor.on_session_reset()
+        assert compressor._last_compression_time == 0.0
+        assert compressor._ineffective_compression_count == 0
+
 
 class TestUpdateFromResponse:
     def test_updates_fields(self, compressor):