From 62bec4b3f83ea61881bb526f6c3efcc17f1a8575 Mon Sep 17 00:00:00 2001 From: teknium1 <127238744+teknium1@users.noreply.github.com> Date: Fri, 24 Jul 2026 12:47:13 -0700 Subject: [PATCH] fix(compression): add recovery path to anti-thrash auto-compaction block MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When two consecutive compactions each failed to clear the threshold, the anti-thrashing breaker blocked automatic compaction PERMANENTLY for the life of the session: nothing decremented _ineffective_compression_count (or _fallback_compression_streak) while blocked, so a session whose middle region was briefly too small to compact never auto-compacted again — it grew unbounded until the provider's hard context limit, and only /new or /reset recovered it. Recovery is a probation probe, not amnesty: after _ANTI_THRASH_RECOVERY_SECONDS (300s) of continuous block the gate grants exactly ONE attempt by dropping tripped counters to 1 strike (persisted, so sibling agents on the same session row — gateway hygiene — unblock too). An ineffective probe re-trips the guard on the next real-usage verdict and the next recovery waits a full fresh window, so the worst case in a truly incompressible session is one compaction attempt per window — bounded, not thrash. The recovery clock is armed lazily on the first BLOCKED evaluation and is deliberately not durable: a restart that loads a durable tripped counter (#69872) starts a full fresh window blocked, preserving the restart-must-never-disarm contract (#54923). Fixes #14694 --- agent/context_compressor.py | 69 ++++++- .../test_compression_anti_thrash_recovery.py | 193 ++++++++++++++++++ 2 files changed, 259 insertions(+), 3 deletions(-) create mode 100644 tests/agent/test_compression_anti_thrash_recovery.py diff --git a/agent/context_compressor.py b/agent/context_compressor.py index 8fe7c85bf69d..73fa1e36f215 100644 --- a/agent/context_compressor.py +++ b/agent/context_compressor.py @@ -1245,6 +1245,7 @@ class ContextCompressor(ContextEngine): self._last_aux_model_failure_model = None self._last_compression_savings_pct = 100.0 self._ineffective_compression_count = 0 + self._anti_thrash_recovery_deadline = 0.0 self._fallback_compression_streak = 0 self._verify_compaction_cleared_threshold = False self._last_compression_made_progress = False @@ -1382,6 +1383,7 @@ class ContextCompressor(ContextEngine): self._last_aux_model_failure_model = None self._last_compression_savings_pct = 100.0 self._ineffective_compression_count = 0 + self._anti_thrash_recovery_deadline = 0.0 self._fallback_compression_streak = 0 self._verify_compaction_cleared_threshold = False self._last_compression_made_progress = False @@ -1408,6 +1410,7 @@ class ContextCompressor(ContextEngine): self._consecutive_timeout_failures = 0 self._fallback_compression_streak = 0 self._ineffective_compression_count = 0 + self._anti_thrash_recovery_deadline = 0.0 self.get_active_compression_failure_cooldown() self._load_fallback_compression_streak() self._load_ineffective_compression_count() @@ -1779,6 +1782,15 @@ class ContextCompressor(ContextEngine): # rationale as the gpt-5.5/Codex 85% autoraise. _MIN_CTX_TRIGGER_RATIO = 0.85 + # Anti-thrash recovery window (#14694): once the ineffective/fallback + # breaker trips, automatic compaction stays blocked for this long, then + # ONE probe attempt is allowed (counters drop to 1 strike, so another + # ineffective pass re-trips immediately). Long enough that a genuinely + # incompressible session isn't compacting in a loop; short enough that a + # session which has since grown real compressible material recovers well + # before it rides into the provider's hard context limit. + _ANTI_THRASH_RECOVERY_SECONDS = 300.0 + @staticmethod def _coerce_max_tokens(value: Any) -> int | None: """Normalize a max_tokens value to a positive int or None. @@ -2048,6 +2060,12 @@ 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 + # Monotonic deadline after which a tripped anti-thrash guard grants + # one probation probe (#14694). 0.0 = clock not armed. Armed lazily on + # the first blocked evaluation; deliberately NOT durable, so a process + # restart with a persisted tripped counter (#69872) waits a full fresh + # window before probing (#54923: restart must never disarm a guard). + self._anti_thrash_recovery_deadline: float = 0.0 # Consecutive completed deterministic-fallback boundaries. Unlike the # real-usage effectiveness counter, ordinary fitting responses must not # reset this breaker; only a healthy completed summary does. @@ -2338,21 +2356,66 @@ class ContextCompressor(ContextEngine): _cooldown_remaining, ) return True - # Anti-thrashing: back off if recent compressions were ineffective + # Anti-thrashing: back off if recent compressions were ineffective. + # The back-off must not be permanent (#14694): the tripped state was + # judged against the transcript as it existed THEN (e.g. a middle + # region too small to matter), but the conversation keeps growing and + # can accumulate plenty of compressible material later. Without a + # recovery path the session never auto-compacts again and rides into + # the provider's hard context limit. Recovery is a probation probe: + # after _ANTI_THRASH_RECOVERY_SECONDS of continuous block, allow ONE + # attempt by dropping the tripped counter(s) to 1 strike (persisted, + # so sibling agents on the same session row unblock too). If the probe + # is ineffective again the very next verdict re-trips the guard, so + # the worst case in the truly-incompressible state is one compaction + # attempt per recovery window — bounded, not thrash. + # + # The clock is armed lazily on the first BLOCKED evaluation rather + # than persisted at trip time: a fresh process that loads a durable + # tripped counter (#69872) therefore starts a full window blocked, + # preserving the restart-must-not-disarm contract (#54923). if ( self._ineffective_compression_count >= 2 or self._fallback_compression_streak >= 2 ): + _now = time.monotonic() + if self._anti_thrash_recovery_deadline <= 0.0: + self._anti_thrash_recovery_deadline = ( + _now + self._ANTI_THRASH_RECOVERY_SECONDS + ) + elif _now >= self._anti_thrash_recovery_deadline: + self._anti_thrash_recovery_deadline = 0.0 + if self._ineffective_compression_count >= 2: + self._record_ineffective_compression_verdict(1) + if self._fallback_compression_streak >= 2: + self._fallback_compression_streak = 1 + self._persist_fallback_compression_streak() + if not self.quiet_mode: + logger.info( + "Anti-thrashing recovery: %.0fs elapsed since the " + "guard tripped — allowing one compaction probe " + "(ineffective=%d fallback=%d).", + self._ANTI_THRASH_RECOVERY_SECONDS, + self._ineffective_compression_count, + self._fallback_compression_streak, + ) + return False if not self.quiet_mode: logger.warning( "Compression skipped — repeated compaction attempts did not " "restore healthy context. ineffective=%d fallback=%d. " - "Consider /new to start fresh, or /compress for " - "focused compression.", + "Auto-compaction will retry once in %.0fs. Consider /new " + "to start fresh, or /compress for focused " + "compression.", self._ineffective_compression_count, self._fallback_compression_streak, + max(0.0, self._anti_thrash_recovery_deadline - _now), ) return True + # Guard not tripped (counters were cleared by an effective compaction + # or a fitting real-usage reading) — disarm any pending recovery clock + # so a LATER trip starts its own full window. + self._anti_thrash_recovery_deadline = 0.0 return False # ------------------------------------------------------------------ diff --git a/tests/agent/test_compression_anti_thrash_recovery.py b/tests/agent/test_compression_anti_thrash_recovery.py new file mode 100644 index 000000000000..06af4076bb94 --- /dev/null +++ b/tests/agent/test_compression_anti_thrash_recovery.py @@ -0,0 +1,193 @@ +"""Anti-thrash recovery: the tripped guard must not be permanent (#14694). + +When two consecutive compactions each fail to clear the threshold, the +anti-thrashing breaker blocks automatic compaction. Before this fix the block +was permanent for the life of the session: nothing ever decremented +``_ineffective_compression_count`` (or ``_fallback_compression_streak``) +while blocked, so a session whose middle region was briefly too small to +compact never auto-compacted again — it grew unbounded until the provider's +hard context limit, and only ``/new`` or ``/reset`` recovered it. + +The recovery contract pinned here: + +* After ``_ANTI_THRASH_RECOVERY_SECONDS`` of continuous block, the gate + grants exactly ONE probation probe: tripped counters drop to 1 strike + (persisted) and the gate reports unblocked once. +* An ineffective probe re-trips the guard on the very next verdict, and the + next recovery waits a FULL fresh window (no immediate re-probe loop). +* An effective probe (or any fitting real-usage reading) fully clears the + counters through the existing ``update_from_response`` path. +* The recovery clock is armed lazily on the first blocked evaluation and is + NOT durable: a process restart that loads a durable tripped counter + (#69872) starts a full fresh window blocked — a restart must never disarm + or shorten the guard (#54923). +* The protection itself is preserved: inside the window the gate stays + blocked exactly as before. +""" + +from unittest.mock import patch + +from agent.context_compressor import ContextCompressor +from hermes_state import SessionDB + + +def _compressor(threshold_tokens: int = 10_000) -> ContextCompressor: + cc = ContextCompressor( + model="test-model", + threshold_percent=0.75, + protect_first_n=3, + protect_last_n=20, + quiet_mode=True, + config_context_length=40960, + provider="test", + ) + cc.threshold_tokens = threshold_tokens + return cc + + +def _trip(cc: ContextCompressor) -> None: + """Arm the breaker exactly as two ineffective real-usage verdicts do.""" + cc._record_ineffective_compression_verdict(2) + + +class TestRecoveryWindow: + def test_blocked_within_window_unblocked_after(self): + cc = _compressor() + _trip(cc) + base = 1000.0 + with patch("agent.context_compressor.time.monotonic", return_value=base): + # First blocked evaluation arms the clock and stays blocked. + assert cc.should_compress(cc.threshold_tokens + 1) is False + with patch( + "agent.context_compressor.time.monotonic", + return_value=base + cc._ANTI_THRASH_RECOVERY_SECONDS - 1, + ): + # Still inside the window: protection intact. + assert cc.should_compress(cc.threshold_tokens + 1) is False + assert cc._ineffective_compression_count == 2 + with patch( + "agent.context_compressor.time.monotonic", + return_value=base + cc._ANTI_THRASH_RECOVERY_SECONDS + 1, + ): + # Window elapsed: exactly one probe is granted. + assert cc.should_compress(cc.threshold_tokens + 1) is True + # Probation, not amnesty: one strike remains armed. + assert cc._ineffective_compression_count == 1 + + def test_ineffective_probe_re_trips_and_waits_a_full_fresh_window(self): + cc = _compressor() + _trip(cc) + base = 1000.0 + with patch("agent.context_compressor.time.monotonic", return_value=base): + assert cc.should_compress(cc.threshold_tokens + 1) is False + probe_time = base + cc._ANTI_THRASH_RECOVERY_SECONDS + 1 + with patch( + "agent.context_compressor.time.monotonic", return_value=probe_time + ): + assert cc.should_compress(cc.threshold_tokens + 1) is True + # The probe compaction completes but does not clear the threshold. + cc._verify_compaction_cleared_threshold = True + cc.update_from_response({"prompt_tokens": cc.threshold_tokens + 1}) + assert cc._ineffective_compression_count == 2 + # Re-tripped: blocked again immediately (arms a new clock). + assert cc.should_compress(cc.threshold_tokens + 1) is False + with patch( + "agent.context_compressor.time.monotonic", + return_value=probe_time + cc._ANTI_THRASH_RECOVERY_SECONDS - 5, + ): + # No immediate re-probe loop: the second window is full length, + # measured from the re-trip, not the original trip. + assert cc.should_compress(cc.threshold_tokens + 1) is False + with patch( + "agent.context_compressor.time.monotonic", + return_value=probe_time + cc._ANTI_THRASH_RECOVERY_SECONDS + 5, + ): + assert cc.should_compress(cc.threshold_tokens + 1) is True + + def test_effective_probe_clears_the_guard_completely(self): + cc = _compressor() + _trip(cc) + base = 1000.0 + with patch("agent.context_compressor.time.monotonic", return_value=base): + assert cc.should_compress(cc.threshold_tokens + 1) is False + with patch( + "agent.context_compressor.time.monotonic", + return_value=base + cc._ANTI_THRASH_RECOVERY_SECONDS + 1, + ): + assert cc.should_compress(cc.threshold_tokens + 1) is True + cc._verify_compaction_cleared_threshold = True + cc.update_from_response({"prompt_tokens": cc.threshold_tokens - 500}) + assert cc._ineffective_compression_count == 0 + assert cc._anti_thrash_recovery_deadline == 0.0 + + def test_fallback_streak_breaker_recovers_too(self): + cc = _compressor() + cc._fallback_compression_streak = 2 + base = 1000.0 + with patch("agent.context_compressor.time.monotonic", return_value=base): + assert cc.should_compress(cc.threshold_tokens + 1) is False + with patch( + "agent.context_compressor.time.monotonic", + return_value=base + cc._ANTI_THRASH_RECOVERY_SECONDS + 1, + ): + assert cc.should_compress(cc.threshold_tokens + 1) is True + assert cc._fallback_compression_streak == 1 + + def test_under_threshold_never_arms_the_clock(self): + cc = _compressor() + _trip(cc) + base = 1000.0 + with patch("agent.context_compressor.time.monotonic", return_value=base): + # Under threshold: gate never evaluated, clock untouched. + assert cc.should_compress(cc.threshold_tokens - 1) is False + assert cc._anti_thrash_recovery_deadline == 0.0 + + def test_untripped_guard_disarms_a_stale_clock(self): + cc = _compressor() + _trip(cc) + base = 1000.0 + with patch("agent.context_compressor.time.monotonic", return_value=base): + assert cc.should_compress(cc.threshold_tokens + 1) is False + assert cc._anti_thrash_recovery_deadline > 0.0 + # A fitting real-usage reading clears the counter mid-window. + cc.update_from_response({"prompt_tokens": cc.threshold_tokens - 500}) + with patch("agent.context_compressor.time.monotonic", return_value=base + 1): + assert cc.should_compress(cc.threshold_tokens + 1) is True + # The stale clock was disarmed, so a LATER trip starts a full window. + assert cc._anti_thrash_recovery_deadline == 0.0 + + +class TestRestartSemantics: + def test_restart_with_durable_tripped_counter_waits_a_full_window(self, tmp_path): + """#69872 x #14694: a restart must not disarm OR shorten the guard.""" + db = SessionDB(db_path=tmp_path / "state.db") + db.create_session(session_id="sess-1", source="cli") + db.set_compression_ineffective_count("sess-1", 2) + + cc = _compressor() + cc.bind_session_state(session_db=db, session_id="sess-1") + assert cc._ineffective_compression_count == 2 + # The recovery clock is process-local and must come up disarmed. + assert cc._anti_thrash_recovery_deadline == 0.0 + base = 5000.0 + with patch("agent.context_compressor.time.monotonic", return_value=base): + assert cc.should_compress(cc.threshold_tokens + 1) is False + with patch( + "agent.context_compressor.time.monotonic", + return_value=base + cc._ANTI_THRASH_RECOVERY_SECONDS + 1, + ): + assert cc.should_compress(cc.threshold_tokens + 1) is True + # The probation reset is durable, so sibling agents on the same + # session row (gateway hygiene) unblock too. + assert db.get_compression_ineffective_count("sess-1") == 1 + + def test_session_reset_disarms_the_recovery_clock(self): + cc = _compressor() + _trip(cc) + base = 1000.0 + with patch("agent.context_compressor.time.monotonic", return_value=base): + assert cc.should_compress(cc.threshold_tokens + 1) is False + assert cc._anti_thrash_recovery_deadline > 0.0 + cc.on_session_reset() + assert cc._anti_thrash_recovery_deadline == 0.0 + assert cc._ineffective_compression_count == 0