RomanNum3ral
/
hermes-agent
mirror of https://github.com/NousResearch/hermes-agent.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fix(gateway): escalate the session-hygiene compaction cooldown on repeat failures 

A gateway session whose summary model keeps timing out no longer retries
compaction on the same fixed interval forever.

The in-agent compressor already escalates repeat summary timeouts
60 -> 300 -> 900s (ContextCompressor.record_timeout_failure), but that ladder
reads the in-memory _consecutive_timeout_failures counter and
bind_session_state() zeroes it (context_compressor.py:1645). Session hygiene
constructs a FRESH AIAgent for every run (gateway/run.py:16820) and re-binds
state each time, so from the gateway that streak is structurally always 0 --
only the flat hygiene_failure_cooldown_seconds (300s) could ever be recorded.
Issue #79624 reported exactly that steady state: an oversized session
(1053 messages, ~119.5k tokens) whose aux model always timed out, re-attempting
compaction every 300s across five days until the reporter deleted the session
by hand.

Track the streak on PersistentState instead, which outlives the per-run agent
and is not cleared by turn/boundary resets, so consecutive hygiene failures
climb 300 -> 900 -> 2700s and then saturate. Both failure sites (progress
timeout and aborted compression) feed it; a real compression resets it, so a
session that recovers starts from the first rung again. The ladder multiplies
the configured base, so operators who tuned
hygiene_failure_cooldown_seconds keep their first rung. Per-session, so one
wedged chat cannot penalize other conversations.

Deliberately NOT changed, since each is a maintainer policy call rather than a
defect (all three are written up on #79624):
  - no durable failure-streak column, so escalation still resets on restart
  - the gateway 30s / in-agent 120s / aux-client 300s-floor timeout mismatch
  - no `hermes doctor` check or `hermes sessions list` marker for a session
    stuck in a compression-failure cooldown

Note the reported exit(1) is NOT a crash: it is the deliberate
_signal_initiated_shutdown path (gateway/run.py:26746-26751, #5646) that lets
systemd Restart=on-failure revive the gateway after a bare SIGTERM, and it
fires on every `systemctl restart` independently of compaction. The compaction
log lines appear after the shutdown line because the gateway-owned executor is
torn down with shutdown(wait=False, cancel_futures=True) (run.py:21164), so an
in-flight turn keeps logging during teardown. Full analysis on the issue.

Post-review hardening (Phase 2c + /simplify-code found five real defects in the
first cut):
  - the recovery gate hand-rolled `_new_tokens < _approx_tokens` when a canonical
    predicate already existed: `compression_made_progress` (agent/turn_context.py,
    #39548). They disagree on 3 of 5 cases -- the hand-rolled form misses a
    row-count win when the summary keeps the token estimate flat, misses one
    where the summary is slightly MORE verbose (so a genuinely recovered session
    would keep escalating forever), and counts a sub-5% wobble as recovery. Now
    reuses the shared predicate, promoted from `_compression_made_progress` to a
    public name with the old private name kept as a back-compat alias so the
    existing importer (tests/agent/test_protected_tail_pressure_61932.py) and any
    patcher of that symbol keep working.
  - the reset was gated on "not aborted", but the degenerate "did not rotate or
    compact in place" branch (#21301) is NOT aborted and yields zero reduction,
    so a session wedged there reset its streak every run and could never
    escalate -- silently defeating the fix. Now gated on real progress.
  - no absolute ceiling: base * 9 reaches 9h at an operator base of 3600s,
    indistinguishable from "compaction switched off". Added
    _HYGIENE_COOLDOWN_MAX_SECONDS = 3600, mirroring the in-file
    _RECONNECT_BACKOFF_CAP precedent.
  - the reset used the get-or-create accessor to write a 0 that was already 0,
    materialising a _sessions entry (never evicted). Now peeks.
  - the abort verdict was probed twice, leaving the reset/record mutual
    exclusion implicit; a future await between the probes would have broken it
    silently. Computed once into _hyg_aborted.

Tests: 19 new in tests/gateway/test_hygiene_failure_cooldown_ladder.py --
ladder escalation, saturation, the absolute cap, per-session isolation,
reset-on-recovery, custom/zero base, PersistentState scoping (a mutation moving
the field to TurnState fails), degraded runners, the progress gate, the exact
progress-predicate semantics the gate depends on, and end-to-end that the
escalated value is what reaches the state DB. All 12 mutations caught, including
ones that restore the flat cooldown (the original bug), ungate the reset, swap
the canonical predicate back for the hand-rolled comparison, remove the cap, and
share the streak globally; the harness hard-errors when a mutation cannot be
applied, since a silently no-op mutation check is worse than none -- an earlier
version of it WAS silently no-opping after a refactor. The gate's contract test
slices by AST node span rather than a fixed character count, which had already
truncated once as the block grew. gateway hygiene + session-state + the three
touched agent compression suites: 50 passed; ruff clean.

E2E with real imports demonstrates the premise rather than asserting it:
bind_session_state zeroes the in-agent counter, and the recorded deadlines go
300 -> 900 -> 2700 -> 2700 -> 2700s where they were previously a flat 300s.

Reported by @yucezerey (#79624), whose state.db column dump and
"deleting the session fixed it" datapoint made the real mechanism findable.
This commit is contained in:
kshitij 2026-08-06 04:21:20 +05:30 committed by kshitij
parent 43fc86562c
commit c0d974b19f
4 changed files with 462 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
agent/turn_context.py
View File

@ -196,7 +196,7 @@ def reanchor_current_turn_user_idx(messages: List[Any], user_message: Any) -> in
return fallback
def _compression_made_progress(
def compression_made_progress(
orig_len: int, new_len: int, orig_tokens: int, new_tokens: int
) -> bool:
"""Return ``True`` if a compression pass materially reduced the request.
@ -219,6 +219,13 @@ def _compression_made_progress(
return orig_tokens > 0 and new_tokens < orig_tokens * 0.95
# Back-compat alias: this predicate was module-private until the gateway's
# session-hygiene recovery gate needed the same semantics (#79624). Keeping the
# old name bound means existing callers and any test that patches
# ``_compression_made_progress`` continue to work unchanged.
_compression_made_progress = compression_made_progress
def _compression_warrants_another_preflight_pass(
orig_tokens: int, new_tokens: int, threshold_tokens: int
) -> bool:

102
gateway/run.py
View File

@ -60,6 +60,9 @@ from agent.conversation_compression import (
from agent.conversation_loop import INTERRUPT_WAITING_FOR_MODEL_PREFIX
from agent.i18n import t
from agent.interrupt_compat import request_hard_interrupt
from agent.turn_context import (
compression_made_progress as _compression_made_progress,
)
from hermes_cli.config import cfg_get
from hermes_cli.fallback_config import get_fallback_chain
@ -120,6 +123,67 @@ _TELEGRAM_NOISY_STATUS_RE = re.compile(
)
_HYGIENE_COOLDOWN_LADDER_MULTIPLIERS = (1, 3, 9)
# Absolute ceiling on an escalated hygiene cooldown, mirroring
# _RECONNECT_BACKOFF_CAP above: with an operator-raised base the multiplier
# ladder alone would reach 9h (base 3600 -> 32400s), which is indistinguishable
# from "compaction silently switched off". 1h is well past the point where a
# retry is cheap and still recovers within a session.
_HYGIENE_COOLDOWN_MAX_SECONDS = 3600.0
def _hygiene_cooldown_for_failure(
gateway,
session_key: str,
base_cooldown_seconds: float,
) -> float:
"""Bump the hygiene failure streak and return the escalated cooldown.
The in-agent compressor escalates repeat summary timeouts 60 -> 300 -> 900s
(``ContextCompressor.record_timeout_failure``), but that ladder reads the
in-memory ``_consecutive_timeout_failures`` counter which
``bind_session_state`` zeroes. Session hygiene constructs a FRESH
``AIAgent`` per run and re-binds state every time, so from the gateway the
streak is structurally always 0 and only the flat
``hygiene_failure_cooldown_seconds`` could ever be recorded — a session
whose summary model always times out retried on that same fixed interval
forever (#79624).
The streak lives on ``PersistentState`` instead, which outlives the per-run
agent, so consecutive failures climb the ladder. Multiplies the configured
base so operators who tuned ``hygiene_failure_cooldown_seconds`` keep their
first rung, then clamps to ``_HYGIENE_COOLDOWN_MAX_SECONDS``.
"""
streak = 1
try:
state = gateway._session_state(session_key).persistent
state.hygiene_failure_streak += 1
streak = state.hygiene_failure_streak
except Exception as exc:
# The caller uses the return value to record the cooldown, so an
# escaping exception would mean NO cooldown at all (hot retry loop) —
# strictly worse than no escalation. Degrade to the base rung.
logger.debug("hygiene failure streak update failed: %s", exc)
multiplier = _HYGIENE_COOLDOWN_LADDER_MULTIPLIERS[
min(streak, len(_HYGIENE_COOLDOWN_LADDER_MULTIPLIERS)) - 1
]
return min(base_cooldown_seconds * multiplier, _HYGIENE_COOLDOWN_MAX_SECONDS)
def _reset_hygiene_failure_streak(gateway, session_key: str) -> None:
"""Clear the hygiene failure streak after a compression that reduced context.
Peeks rather than get-or-creates: writing a 0 that is already 0 must not
materialise a ``_sessions`` entry (those are never evicted).
"""
try:
state = gateway._peek_session_state(session_key)
if state is not None:
state.persistent.hygiene_failure_streak = 0
except Exception as exc:
logger.debug("hygiene failure streak reset failed: %s", exc)
def _record_hygiene_cooldown(gateway, session_id: str, cooldown_seconds: float) -> None:
"""Persist a session-hygiene compression-failure cooldown to the state DB.
@ -16912,7 +16976,10 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
if _hyg_failure_cooldown_seconds >= 0:
_record_hygiene_cooldown(
self, session_entry.session_id,
_hyg_failure_cooldown_seconds,
_hygiene_cooldown_for_failure(
self, session_key,
_hyg_failure_cooldown_seconds,
),
)
from agent.session_activity import (
ActivityProvenance,
@ -17103,11 +17170,40 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
# /compress to retry or /reset to start
# fresh.
_comp = getattr(_hyg_agent, "context_compressor", None)
if _comp is not None and getattr(_comp, "_last_compress_aborted", False):
_hyg_aborted = _comp is not None and getattr(
_comp, "_last_compress_aborted", False
)
if not _hyg_aborted:
# Only a run that materially reduced the
# request counts as recovery. The
# degenerate "did not rotate or compact
# in place" branch above leaves both
# counts equal and is NOT aborted, so
# gating on "not aborted" alone would
# clear the streak on every wedged run
# and the cooldown could never escalate
# (#79624). Reuse the canonical
# progress predicate rather than a
# hand-rolled token comparison: rows
# dropping is progress even when the
# summary keeps the token estimate flat,
# and a sub-5% token wobble is noise,
# not recovery (#39548).
if _compression_made_progress(
_msg_count, _new_count,
_approx_tokens, _new_tokens,
):
_reset_hygiene_failure_streak(
self, session_key
)
if _hyg_aborted:
if _hyg_failure_cooldown_seconds >= 0:
_record_hygiene_cooldown(
self, session_entry.session_id,
_hyg_failure_cooldown_seconds,
_hygiene_cooldown_for_failure(
self, session_key,
_hyg_failure_cooldown_seconds,
),
)
from agent.session_activity import (
ActivityProvenance,

9
gateway/session_state.py
View File

@ -148,6 +148,15 @@ class PersistentState:
# Monotonic run-generation counter (#28686). NEVER reset: clearing it
# would break stale-run detection.
run_generation: int = 0
# Consecutive session-hygiene compression failures for this session
# (#79624). The in-agent compressor escalates repeat timeouts via
# ContextCompressor._consecutive_timeout_failures, but hygiene builds a
# FRESH AIAgent per run and bind_session_state() zeroes that counter, so
# the in-agent ladder is structurally unreachable from the gateway.
# Tracking the streak here — outside the per-run agent — lets hygiene
# escalate its cooldown instead of retrying on a flat interval forever.
# Reset on a successful compression, not by turn/boundary resets.
hygiene_failure_streak: int = 0
@dataclass

346
tests/gateway/test_hygiene_failure_cooldown_ladder.py Normal file
View File

@ -0,0 +1,346 @@
"""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,
)
from gateway.session_state import PersistentState, SessionState
class _Runner:
"""Minimal gateway stand-in exposing just ``_session_state``."""
def __init__(self):
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 _peek_session_state(self, session_key):
return self._sessions.get(session_key)
BASE = 300.0
KEY = "agent:main:telegram:private:123"
# ---------------------------------------------------------------------------
# The state field
# ---------------------------------------------------------------------------
def test_persistent_state_tracks_hygiene_failure_streak():
"""The streak must live on PersistentState, not the per-run agent."""
assert PersistentState().hygiene_failure_streak == 0
def test_streak_survives_turn_and_conversation_resets():
"""PersistentState is not cleared wholesale by turn/boundary resets, which is
exactly why the streak lives there rather than on the hygiene agent."""
runner = _Runner()
_hygiene_cooldown_for_failure(runner, KEY, BASE)
state = runner._session_state(KEY)
# Simulate what a turn/boundary reset does: replace the turn + conversation
# scopes, leaving `persistent` alone.
state.turn = type(state.turn)()
state.conversation = type(state.conversation)()
assert state.persistent.hygiene_failure_streak == 1
# ---------------------------------------------------------------------------
# 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]
assert seen == [300.0, 900.0, 2700.0]
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_streak_is_monotonic_across_calls(self):
runner = _Runner()
for expected in (1, 2, 3, 4):
_hygiene_cooldown_for_failure(runner, KEY, BASE)
assert (
runner._session_state(KEY).persistent.hygiene_failure_streak
== expected
)
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_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_respects_a_custom_base(self):
runner = _Runner()
assert _hygiene_cooldown_for_failure(runner, KEY, 30.0) == 30.0
assert _hygiene_cooldown_for_failure(runner, KEY, 30.0) == 90.0
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_bare_runner_without_sessions_map_still_cools_down(self):
"""Many gateway tests build runners via object.__new__ with no _sessions.
``_sessions_map()`` self-heals, so this exercises the happy path on a
bare runner rather than the except branch — pinned because the
object.__new__ pattern is pervasive in gateway tests and must not raise.
"""
from gateway.run import GatewayRunner
bare = object.__new__(GatewayRunner)
assert _hygiene_cooldown_for_failure(bare, KEY, BASE) == BASE
def test_reset_on_bare_runner_is_a_noop(self):
from gateway.run import GatewayRunner
bare = object.__new__(GatewayRunner)
_reset_hygiene_failure_streak(bare, KEY) # must not raise
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
def test_absent_session_reset_is_a_noop(self):
"""Reset peeks rather than get-or-creates: a session with no state entry
must not materialise one just to write a 0 that is already 0
(_sessions entries are never evicted)."""
runner = _Runner()
_reset_hygiene_failure_streak(runner, "never-seen")
assert "never-seen" not in runner._sessions
# ---------------------------------------------------------------------------
# The reset gate in _handle_message_with_agent
# ---------------------------------------------------------------------------
class TestResetGate:
"""The reset must require ACTUAL context reduction, not merely 'not aborted'.
gateway/run.py has a degenerate branch ("did not rotate or compact in
place ... no session_db on the hygiene agent", #21301) that sets
``_new_tokens = _approx_tokens`` and is NOT aborted. Gating the reset on
'not aborted' alone cleared the streak on every such run, so a session
wedged there could never escalate — silently defeating the whole fix.
"""
@staticmethod
def _gate_source():
"""The `if not _hyg_aborted:` / `if _hyg_aborted:` pair and their bodies.
Sliced by AST node span rather than a fixed character count: a fixed
slice silently truncates when the block grows and the assertions then
pass or fail for the wrong reason.
"""
import ast
import inspect
import textwrap
import gateway.run as run_mod
src = textwrap.dedent(
inspect.getsource(run_mod.GatewayRunner._handle_message_with_agent)
)
tree = ast.parse(src)
lines = src.splitlines()
spans = []
for node in ast.walk(tree):
if isinstance(node, ast.Assign) and any(
isinstance(t, ast.Name) and t.id == "_hyg_aborted"
for t in node.targets
):
spans.append((node.lineno, node.end_lineno))
if isinstance(node, ast.If) and "_hyg_aborted" in ast.unparse(node.test):
spans.append((node.lineno, node.end_lineno))
assert spans, "could not locate the _hyg_aborted gate"
return "\n".join(lines[min(s[0] for s in spans) - 1:max(s[1] for s in spans)])
def test_reset_is_gated_on_the_canonical_progress_predicate(self):
gate = self._gate_source()
assert "_reset_hygiene_failure_streak" in gate
# Must reuse the shared predicate, not a hand-rolled comparison. A bare
# `_new_tokens < _approx_tokens` gets three cases wrong: it misses a
# row-count win when the summary keeps tokens flat, misses one where the
# summary is slightly more verbose, and counts a sub-5% wobble as
# recovery (#39548).
assert "_compression_made_progress(" in gate, (
"reset must use the canonical progress predicate"
)
assert "_new_tokens < _approx_tokens" not in gate, (
"hand-rolled token comparison disagrees with the canonical predicate"
)
def test_progress_predicate_semantics_the_gate_depends_on(self):
"""Pin the behaviour the gate is now relying on.
If these ever change, the hygiene recovery gate's meaning changes with
them — so bind them here rather than assuming.
"""
from agent.turn_context import compression_made_progress as prog
# Rows dropped is progress even when the token estimate stays flat
# (or rises slightly because the summary text is verbose).
assert prog(220, 100, 50_000, 50_000) is True
assert prog(220, 100, 50_000, 50_100) is True
# Size-only win with equal row counts is progress (#39548).
assert prog(220, 220, 288_000, 183_000) is True
# A sub-5% wobble is noise, not recovery.
assert prog(220, 220, 50_000, 49_900) is False
# The degenerate no-rotate branch: nothing moved (the #79624 wedge).
assert prog(220, 220, 50_000, 50_000) is False
def test_abort_probe_is_computed_once(self):
"""Mutual exclusion between reset and the failure record must be
explicit. Two separate getattr probes could disagree if a future edit
inserts an await between them."""
gate = self._gate_source()
assert gate.count("_last_compress_aborted") == 1, (
"compute the abort verdict once into _hyg_aborted and branch on it"
)
assert "if not _hyg_aborted:" in gate
assert "if _hyg_aborted:" in gate
# ---------------------------------------------------------------------------
# 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)