fix(compaction): anti-thrashing guard never fired; score against the threshold

`should_compress()` documents anti-thrashing protection ("if the last two
compressions each saved less than 10%, skip compression to avoid infinite
loops"). In practice `_ineffective_compression_count` reset on every pass,
so the guard was dead code and a mis-sized context window presented as a
hung CLI instead of a warning.

Two defects:

1. Mixed measurement bases. Effectiveness was
   `(current_tokens - estimate(compressed)) / current_tokens`, where
   `current_tokens` is the provider's FULL prompt (system prompt + tool
   schemas + messages) but `estimate(compressed)` covers messages only.
   Every compaction therefore reported ~96% savings and reset the counter.
   Savings is now scored messages-vs-messages.

2. Message shrinkage is the wrong yardstick. `should_compress()` trips on
   the full prompt, but compaction can only shrink messages -- the system
   prompt and tool schemas are an incompressible floor. When that floor
   alone meets the threshold, each pass shrinks messages by a healthy
   margin, legitimately resets the counter, and still leaves the prompt
   over the line; the next turn compacts again, forever. Observed in the
   wild: 45+ consecutive compactions, one auxiliary-LLM call each, zero
   progress. Effectiveness is now scored against the goal -- did the
   projected prompt get under the threshold? -- and a futile pass warns
   with the numbers that prove it.

Also record an ineffective pass on the "only N messages (need > M)" early
return, which previously returned the transcript unchanged without moving
any anti-thrash state -- the same class of bug the neighbouring
"no compressable window" branch was already fixed for.

Tests: 4 of the 5 new cases fail on main and pass here; the fifth pins
that effective compaction still resets the counter (121 -> 15 messages,
88.9% savings).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

agent/context_compressor.py

@@ -2840,10 +2840,21 @@ This compaction should PRIORITISE preserving all information related to the focu
         # Only need head + 3 tail messages minimum (token budget decides the real tail size)
         _min_for_compress = self._protect_head_size(messages) + 3 + 1
         if n_messages <= _min_for_compress:
+            # Record the no-op, exactly as the sibling "no compressable window"
+            # branch below does (#40803). Returning without touching the
+            # anti-thrashing counter leaves should_compress() saying True on a
+            # transcript that can never shrink: when the prompt sits above the
+            # threshold because of the incompressible floor (system prompt +
+            # tool schemas), every subsequent turn re-fires a compaction that
+            # returns here unchanged, and the CLI appears frozen.
+            self._ineffective_compression_count += 1
+            self._last_compression_savings_pct = 0.0
             if not self.quiet_mode:
                 logger.warning(
-                    "Cannot compress: only %d messages (need > %d)",
+                    "Cannot compress: only %d messages (need > %d). "
+                    "ineffective_compression_count=%d",
                     n_messages, _min_for_compress,
+                    self._ineffective_compression_count,
                 )
             return messages
 
@@ -3139,12 +3150,43 @@ This compaction should PRIORITISE preserving all information related to the focu
         compressed = _strip_historical_media(compressed)
 
         new_estimate = estimate_messages_tokens_rough(compressed)
-        saved_estimate = display_tokens - new_estimate
 
-        # Anti-thrashing: track compression effectiveness
-        savings_pct = (saved_estimate / display_tokens * 100) if display_tokens > 0 else 0
+        # Anti-thrashing: measure effectiveness on a like-for-like basis.
+        #
+        # ``display_tokens`` is usually ``current_tokens`` — the provider's real
+        # prompt count, which includes the system prompt and tool schemas.
+        # ``new_estimate`` covers the messages ONLY. Comparing the two makes a
+        # compaction that freed almost nothing look like it saved ~96%, so the
+        # counter below resets every pass and the anti-thrashing guard is dead
+        # code. Compaction can only shrink messages, so score it against the
+        # messages it was given.
+        pre_estimate = estimate_messages_tokens_rough(messages)
+        saved_estimate = pre_estimate - new_estimate
+        savings_pct = (saved_estimate / pre_estimate * 100) if pre_estimate > 0 else 0
         self._last_compression_savings_pct = savings_pct
-        if savings_pct < 10:
+
+        # Effectiveness is "did we get the prompt under the threshold?", not
+        # "did the message list shrink?". ``should_compress()`` trips on the
+        # FULL prompt, but compaction can only shrink messages: the system
+        # prompt and tool schemas are an incompressible floor. When that floor
+        # alone meets the threshold — a too-small context_length makes this
+        # trivially true — every pass can shrink messages by a healthy margin,
+        # reset the counter here, and still leave the prompt over the line, so
+        # the next turn compacts again, forever. Score against the goal.
+        incompressible_floor = max(0, display_tokens - pre_estimate)
+        projected_prompt = incompressible_floor + new_estimate
+        if projected_prompt >= self.threshold_tokens:
+            self._ineffective_compression_count += 1
+            if not self.quiet_mode:
+                logger.warning(
+                    "Compaction cannot clear the threshold: ~%d incompressible "
+                    "tokens (system prompt + tool schemas) + ~%d compressed "
+                    "messages >= %d threshold. Shrinking messages further "
+                    "cannot help. ineffective_compression_count=%d",
+                    incompressible_floor, new_estimate, self.threshold_tokens,
+                    self._ineffective_compression_count,
+                )
+        elif savings_pct < 10:
             self._ineffective_compression_count += 1
         else:
             self._ineffective_compression_count = 0

tests/agent/test_compaction_anti_thrash.py

@@ -0,0 +1,125 @@
+"""The anti-thrashing guard must actually fire.
+
+Two defects made it dead code:
+
+1. Effectiveness was scored as ``(current_tokens - estimate(compressed)) / current_tokens``.
+   ``current_tokens`` is the provider's FULL prompt (system prompt + tool schemas +
+   messages); ``estimate(compressed)`` covers messages only. The mixed basis reported
+   ~96% savings on every pass, so ``_ineffective_compression_count`` reset each time.
+
+2. Even scored correctly, message shrinkage is the wrong yardstick. ``should_compress()``
+   trips on the full prompt, but compaction can only shrink messages. When the
+   incompressible floor alone meets the threshold, each pass can shrink messages by a
+   healthy margin, reset the counter, and still leave the prompt over the line -- so the
+   next turn compacts again, forever.
+
+Together these made a mis-sized context window present as a hung CLI rather than a
+warning. Effectiveness is now scored against the goal: did the prompt get under the
+threshold?
+"""
+import pytest
+
+from agent.context_compressor import ContextCompressor
+
+
+def _compressor(threshold_tokens: int) -> ContextCompressor:
+    cc = ContextCompressor(
+        model="test-model",
+        threshold_percent=0.75,
+        protect_first_n=5,
+        protect_last_n=20,
+        quiet_mode=True,
+        config_context_length=40960,
+        provider="test",
+    )
+    cc.threshold_tokens = threshold_tokens  # pin; don't couple to window math
+    cc._generate_summary = lambda *a, **k: "Summary of earlier turns."
+    return cc
+
+
+def _messages(n: int, size: int = 1500) -> list:
+    msgs = [{"role": "system", "content": "sys"}]
+    for i in range(n):
+        role = "user" if i % 2 == 0 else "assistant"
+        msgs.append({"role": role, "content": f"m{i} " + "z" * size})
+    return msgs
+
+
+class TestSavingsBasis:
+    def test_savings_does_not_depend_on_current_tokens(self):
+        """Savings is a property of the messages, not of the caller's token count.
+
+        Before the fix, passing the full-prompt count made an identical compaction
+        report a wildly higher savings percentage.
+        """
+        msgs = _messages(14)
+
+        a = _compressor(threshold_tokens=24_576)
+        a.compress([m.copy() for m in msgs], current_tokens=100_000)
+        with_full_prompt = a._last_compression_savings_pct
+
+        b = _compressor(threshold_tokens=24_576)
+        b.compress([m.copy() for m in msgs], current_tokens=None)
+        messages_only = b._last_compression_savings_pct
+
+        assert with_full_prompt == pytest.approx(messages_only, abs=0.01), (
+            "savings must be scored on a messages-vs-messages basis; got "
+            f"{with_full_prompt:.1f}% with a full-prompt count vs "
+            f"{messages_only:.1f}% without"
+        )
+
+    def test_no_op_compaction_is_not_reported_as_huge_savings(self):
+        cc = _compressor(threshold_tokens=24_576)
+        msgs = _messages(4, size=10)  # below the minimum compressible size
+        cc.compress(msgs, current_tokens=100_000)
+        assert cc._last_compression_savings_pct < 50
+
+
+class TestFutilityGuard:
+    def test_stops_when_floor_alone_meets_threshold(self):
+        """Incompressible floor >= threshold -> shrinking messages cannot help."""
+        cc = _compressor(threshold_tokens=24_576)
+        msgs = _messages(14)
+        # Full prompt is far above threshold and dominated by system + tools.
+        real_prompt_tokens = 33_564
+
+        fired = 0
+        for _ in range(6):
+            if cc.should_compress(real_prompt_tokens):
+                fired += 1
+                msgs = cc.compress(msgs, current_tokens=real_prompt_tokens)
+                msgs.append({"role": "user", "content": "next " + "w" * 4000})
+
+        assert cc._ineffective_compression_count >= 2
+        assert not cc.should_compress(real_prompt_tokens), (
+            "compaction that cannot clear the threshold must stop and warn"
+        )
+        assert fired <= 3, f"expected the loop to break early, compacted {fired}x"
+
+    def test_effective_compaction_still_resets_the_counter(self):
+        """A compaction that gets the prompt under the threshold is not thrashing."""
+        cc = _compressor(threshold_tokens=750_000)
+        msgs = _messages(120, size=2000)
+        before = len(msgs)
+        out = cc.compress(msgs, current_tokens=None)
+
+        assert len(out) < before, "a long transcript must still compact"
+        assert cc._last_compression_savings_pct > 10
+        assert cc._ineffective_compression_count == 0
+
+
+class TestMinimumMessagesBranch:
+    def test_too_few_messages_records_an_ineffective_pass(self):
+        """Returning the transcript unchanged must move the anti-thrash state.
+
+        Otherwise should_compress() keeps saying True about a transcript that can
+        never shrink, and every turn re-enters a no-op compaction.
+        """
+        cc = _compressor(threshold_tokens=1)
+        msgs = _messages(3, size=10)
+        before = cc._ineffective_compression_count
+
+        out = cc.compress(msgs, current_tokens=100_000)
+
+        assert len(out) == len(msgs), "nothing should have been compressed"
+        assert cc._ineffective_compression_count == before + 1