"""Session-hygiene compression must escalate its cooldown for repeat failures. Issue #79624: a gateway session whose summary model always times out retried compaction on a flat ``hygiene_failure_cooldown_seconds`` interval forever. The in-agent compressor already escalates repeat timeouts 60 -> 300 -> 900s via ``ContextCompressor.record_timeout_failure``, but that ladder reads the in-memory ``_consecutive_timeout_failures`` counter, and: * session hygiene constructs a FRESH ``AIAgent`` for every run (``gateway/run.py`` ~16820), and * ``ContextCompressor.bind_session_state`` zeroes that counter. so the in-agent ladder is *structurally unreachable* from the gateway — the streak is always 0 there. These tests pin the streak to ``PersistentState`` (which outlives the per-run agent) and assert the ladder actually climbs. """ from __future__ import annotations import pytest from gateway.run import ( _HYGIENE_COOLDOWN_LADDER_MULTIPLIERS, _hygiene_cooldown_for_failure, _record_hygiene_cooldown, _reset_hygiene_failure_streak, hygiene_compaction_recovered, hygiene_wait_should_extend, ) from gateway.run import GatewayRunner from gateway.session_state import SessionState def _Runner(): """A real ``GatewayRunner`` with no ``__init__`` side effects. Deliberately NOT a hand-written stub: an earlier version reimplemented ``_session_state`` and ``_peek_session_state``, which meant the tests exercised the copies rather than the production accessors and could drift from them silently (the real ``_peek_session_state`` returns ``None`` on a falsy ``_sessions``, and ``_session_state`` goes through ``_sessions_map()`` self-healing). ``object.__new__`` is already the idiom elsewhere in this file, and the self-healing map means no attribute setup is needed. """ return object.__new__(GatewayRunner) BASE = 300.0 KEY = "agent:main:telegram:private:123" # --------------------------------------------------------------------------- # The state field # --------------------------------------------------------------------------- # --------------------------------------------------------------------------- # The ladder # --------------------------------------------------------------------------- class TestCooldownLadder: def test_first_failure_uses_the_configured_base(self): """Operators who tuned hygiene_failure_cooldown_seconds keep rung 1.""" runner = _Runner() assert _hygiene_cooldown_for_failure(runner, KEY, BASE) == BASE def test_consecutive_failures_escalate(self): runner = _Runner() seen = [ _hygiene_cooldown_for_failure(runner, KEY, BASE) for _ in range(3) ] assert seen == [BASE * m for m in _HYGIENE_COOLDOWN_LADDER_MULTIPLIERS] def test_consecutive_failures_escalate_across_gateway_restart(self, tmp_path): from hermes_state import SessionDB db = SessionDB(db_path=tmp_path / "state.db") try: db.create_session("before-rotation", "telegram", session_key=KEY) first_runner = _Runner() first_runner._session_db = db assert _hygiene_cooldown_for_failure(first_runner, KEY, BASE) == BASE db.create_session( "after-rotation", "telegram", session_key=KEY, parent_session_id="before-rotation", ) restarted_runner = _Runner() restarted_runner._session_db = db assert _hygiene_cooldown_for_failure( restarted_runner, KEY, BASE ) == BASE * 3 other_chat_runner = _Runner() other_chat_runner._session_db = db assert _hygiene_cooldown_for_failure( other_chat_runner, "agent:main:telegram:private:999", BASE ) == BASE finally: db.close() def test_ladder_saturates_at_the_top_rung(self): """A permanently un-compactable session must not grow without bound.""" runner = _Runner() for _ in range(3): _hygiene_cooldown_for_failure(runner, KEY, BASE) top = BASE * _HYGIENE_COOLDOWN_LADDER_MULTIPLIERS[-1] for _ in range(10): assert _hygiene_cooldown_for_failure(runner, KEY, BASE) == top def test_reset_returns_to_the_first_rung(self): """A session that recovers must start over, not stay pinned at the top.""" runner = _Runner() for _ in range(3): _hygiene_cooldown_for_failure(runner, KEY, BASE) _reset_hygiene_failure_streak(runner, KEY) assert runner._session_state(KEY).persistent.hygiene_failure_streak == 0 assert _hygiene_cooldown_for_failure(runner, KEY, BASE) == BASE def test_reset_returns_restarted_gateway_to_first_rung(self, tmp_path): from hermes_state import SessionDB db = SessionDB(db_path=tmp_path / "state.db") try: runner = _Runner() runner._session_db = db assert _hygiene_cooldown_for_failure(runner, KEY, BASE) == BASE _reset_hygiene_failure_streak(runner, KEY) restarted_runner = _Runner() restarted_runner._session_db = db assert _hygiene_cooldown_for_failure( restarted_runner, KEY, BASE ) == BASE finally: db.close() def test_streaks_are_per_session(self): """One wedged session must not penalize every other chat.""" runner = _Runner() other = "agent:main:telegram:private:999" for _ in range(3): _hygiene_cooldown_for_failure(runner, KEY, BASE) assert _hygiene_cooldown_for_failure(runner, other, BASE) == BASE def test_absolute_cap_bounds_a_large_operator_base(self): """The multiplier ladder alone would reach 9h at base=3600, which is indistinguishable from 'compaction silently switched off'.""" from gateway.run import _HYGIENE_COOLDOWN_MAX_SECONDS runner = _Runner() seen = [ _hygiene_cooldown_for_failure(runner, KEY, 3600.0) for _ in range(4) ] assert max(seen) == _HYGIENE_COOLDOWN_MAX_SECONDS assert all(v <= _HYGIENE_COOLDOWN_MAX_SECONDS for v in seen) def test_cap_does_not_shrink_the_configured_base(self): """A base already above the cap must still be honoured on rung 1 — clamping must never hand back a SHORTER cooldown than configured.""" from gateway.run import _HYGIENE_COOLDOWN_MAX_SECONDS runner = _Runner() big = _HYGIENE_COOLDOWN_MAX_SECONDS * 2 assert _hygiene_cooldown_for_failure(runner, KEY, big) == pytest.approx( _HYGIENE_COOLDOWN_MAX_SECONDS ) def test_zero_base_stays_zero(self): """A 0 base is 'cool down for no time'; escalation must not invent one.""" runner = _Runner() assert _hygiene_cooldown_for_failure(runner, KEY, 0.0) == 0.0 assert _hygiene_cooldown_for_failure(runner, KEY, 0.0) == 0.0 # --------------------------------------------------------------------------- # Degraded runners (the gateway test-double pitfall) # --------------------------------------------------------------------------- class TestDegradedRunners: def test_runner_whose_session_state_raises_still_cools_down(self): """The real degraded case: a stand-in whose _session_state blows up. A missing streak must degrade to 'no escalation'. It must NEVER let the exception escape, because the caller uses the return value to record the cooldown — losing it would mean no cooldown at all and a hot retry loop. """ class _Exploding: def _session_state(self, session_key): raise RuntimeError("no sessions map") gw = _Exploding() assert _hygiene_cooldown_for_failure(gw, KEY, BASE) == BASE _reset_hygiene_failure_streak(gw, KEY) # must not raise # --------------------------------------------------------------------------- # The failure reason reaches the state DB # --------------------------------------------------------------------------- class TestFailureReasonForwarded: """`record_compression_failure_cooldown` writes compression_failure_error UNCONDITIONALLY, so omitting the reason clobbers to NULL whatever the in-conversation path recorded — and readers then show the user "unknown error". Matters more now that a cooldown can last an hour.""" def _capture(self, *args): seen = {} class _DB: def record_compression_failure_cooldown(self, sid, until, error=None): seen.update(sid=sid, until=until, error=error) class _GW: _session_db = _DB() _record_hygiene_cooldown(_GW(), "sid-1", 300.0, *args) return seen def test_reason_is_forwarded_when_supplied(self): seen = self._capture("summary model timed out") assert seen["error"] == "summary model timed out" # --------------------------------------------------------------------------- # The recovery predicate (extracted from _handle_message_with_agent) # --------------------------------------------------------------------------- class TestHygieneCompactionRecovered: """Direct unit tests for the recovery decision. This replaces three source-reading tests that asserted on ``inspect.getsource`` text. AGENTS.md bans reading source in tests outright and names this file's module as the case where the right answer is to extract the logic — which is what ``hygiene_compaction_recovered`` is. The old tests were also actively wrong: one asserted the buggy ``_new_tokens < _approx_tokens`` substring was present, so it passed while the gate was broken and had to be edited when the gate was fixed. """ BASE = dict( aborted=False, rotated=True, in_place=False, msg_count=220, new_count=100, approx_tokens=50_000, new_tokens=30_000, ) def _call(self, **over): return hygiene_compaction_recovered(**{**self.BASE, **over}) def test_real_rotation_with_reduction_is_recovery(self): assert self._call() is True def test_abort_is_never_recovery(self): assert self._call(aborted=True) is False def test_no_rewrite_is_never_recovery_even_when_counts_look_good(self): """The degenerate #21301 path: not aborted, but nothing was rewritten. Deliberately passes counts that WOULD read as progress, so this binds the rotated/in_place guard specifically. Using equal counts here would pass vacuously — the progress predicate already rejects those, so the guard could be deleted and the test would still pass. """ # Sanity: these counts do read as progress on their own. from agent.turn_context import compression_made_progress assert compression_made_progress(220, 100, 50_000, 30_000) is True # ...but with nothing rewritten it must still not count as recovery. assert self._call(rotated=False, in_place=False) is False def test_in_place_compaction_counts(self): assert self._call(rotated=False, in_place=True) is True def test_row_drop_with_flat_tokens_is_recovery(self): """Rows dropping is progress even when the summary keeps tokens flat. A bare token comparison misses this and would keep a recovered session escalating to the cap forever. """ assert self._call(new_count=100, new_tokens=50_000) is True def test_size_only_win_is_recovery(self): """Equal rows, large token reduction (#39548).""" assert self._call( msg_count=220, new_count=220, approx_tokens=288_000, new_tokens=183_000, ) is True def test_sub_five_percent_wobble_is_not_recovery(self): """Noise must not clear the streak, or escalation is defeated again.""" assert self._call( msg_count=220, new_count=220, approx_tokens=50_000, new_tokens=49_900, ) is False class TestHygieneWaitShouldExtend: """Host must not keep waiting after the commit fence is already cancelled.""" def test_extends_while_idle_and_under_ceiling(self): assert hygiene_wait_should_extend( idle=1.0, timeout=30.0, waited=10.0, ceiling=600.0, ) is True def test_stops_when_idle_budget_exhausted(self): assert hygiene_wait_should_extend( idle=30.0, timeout=30.0, waited=10.0, ceiling=600.0, ) is False def test_stops_at_ceiling(self): assert hygiene_wait_should_extend( idle=1.0, timeout=30.0, waited=600.0, ceiling=600.0, ) is False def test_fence_cancel_stops_even_with_fresh_progress(self): assert hygiene_wait_should_extend( idle=0.0, timeout=30.0, waited=0.1, ceiling=600.0, fence_cancelled=True, ) is False # --------------------------------------------------------------------------- # Integration with the persist helper # --------------------------------------------------------------------------- class TestRecordedCooldownEscalates: """The escalated value must be what actually lands in the state DB.""" class _DB: def __init__(self): self.calls = [] def record_compression_failure_cooldown(self, sid, until, error=None): self.calls.append((sid, until)) class _GW: def __init__(self, db): self._session_db = db self._sessions = {} def _session_state(self, session_key): state = self._sessions.get(session_key) if state is None: state = SessionState() self._sessions[session_key] = state return state def test_persisted_deadlines_grow(self, monkeypatch): import time as real_time db = self._DB() gw = self._GW(db) monkeypatch.setattr( "gateway.run.logger", __import__("logging").getLogger("test") ) now = real_time.time() for _ in range(3): _record_hygiene_cooldown( gw, "sess-1", _hygiene_cooldown_for_failure(gw, KEY, BASE) ) assert len(db.calls) == 3 waits = [until - now for _, until in db.calls] # Strictly increasing, and each close to its ladder rung. assert waits[0] < waits[1] < waits[2] for wait, mult in zip(waits, _HYGIENE_COOLDOWN_LADDER_MULTIPLIERS): assert wait == pytest.approx(BASE * mult, abs=5.0)