mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-07-25 17:18:11 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 6

fix(compression): add recovery path to anti-thrash auto-compaction block

Browse source 
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
This commit is contained in:
teknium1 2026-07-24 12:47:13 -07:00 • committed by Teknium
parent 5a0f51325c
commit 62bec4b3f8
2 changed files with 259 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

69
agent/context_compressor.py
View file

@ -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 <topic> for "
"focused compression.",
"Auto-compaction will retry once in %.0fs. Consider /new "
"to start fresh, or /compress <topic> 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
# ------------------------------------------------------------------

193
tests/agent/test_compression_anti_thrash_recovery.py Normal file
View file

@ -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