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

refactor(gateway): extract the hygiene recovery gate and forward the failure reason 

Follow-up to c0d974b19 (#79741). Three review findings against that commit,
none of which change the escalation behaviour it shipped.

1. The recovery decision lived inline in `_handle_message_with_agent`, a
   ~2000-line async method, so the only way to pin it was a test that read
   `inspect.getsource(...)` and asserted on substrings. AGENTS.md bans reading
   source in tests outright and names this exact situation: "if the logic lives
   inline in a god-file (gateway/run.py) and extracting it feels disruptive:
   that's the actual signal to do the extraction, not to regex around it."

   Those tests were not merely stylistically wrong, they were actively harmful.
   One asserted the substring `_new_tokens < _approx_tokens` was PRESENT -- so
   it passed while the gate had the bug that substring represents, and had to be
   edited when the gate was fixed. It failed on correct code and passed on
   broken code, in one assertion.

   Extracted `hygiene_compaction_recovered()` as a module-level pure predicate
   and replaced the three source-reading tests with eight direct unit tests.
   The extraction immediately earned itself: the new tests caught a `NameError`
   (the predicate called `compression_made_progress` while the module bound it
   under an alias) that a source-text assertion cannot see, because the symbol
   is spelled correctly in the source and only fails at runtime.

2. The gate inferred "did the transcript actually get rewritten" from a numeric
   side effect -- the degenerate "did not rotate or compact in place" path
   (#21301) reuses the pre-compression counts -- when the booleans
   `_hyg_rotated` / `_hyg_in_place` were already in scope and explicitly set
   False on that path. The predicate now takes them directly, so a future edit
   that re-estimates instead of reusing the old counts cannot silently defeat
   the escalation.

3. `_record_hygiene_cooldown` passed no `error` to
   `record_compression_failure_cooldown`, which writes `compression_failure_error`
   unconditionally -- so a hygiene failure clobbered to NULL whatever reason the
   in-conversation path had recorded, and readers then show the user "unknown
   error" (agent/manual_compression_feedback.py, gateway/slash_commands.py). The
   reason was already in hand at both call sites. Pre-existing from #74136 but
   amplified by escalation: a blank reason on a 45-minute cooldown is far more
   user-visible than on a 5-minute one.

Also: the ladder docstring described the compressor's absolute 60/300/900s
ladder while the constant is multipliers (1, 3, 9); the config docs still
described `hygiene_failure_cooldown_seconds` as a flat interval rather than the
first rung of a capped ladder; and `PersistentState.hygiene_failure_streak` now
documents that it is process-local by design -- keying on `session_key` is what
survives compaction rotation, which the persisted `compression_*_streak`
columns cannot express since they key on the rotating `session_id`. Making it
durable is a schema change, tracked on #79624 rather than smuggled in here.

Also replaces the file's hand-written `_Runner` stub with
`object.__new__(GatewayRunner)` (already the idiom elsewhere in the same file).
The stub reimplemented `_session_state` and `_peek_session_state`, so the tests
exercised copies that could drift from production; using the real class
immediately made one assertion stronger -- on a fresh runner `_sessions` does not
exist at all until something materialises it, so the reset provably did not even
create the map.

A review pass on this follow-up then caught that the CALL SITE was still
unbound: deleting the whole `if not _hyg_aborted: if
hygiene_compaction_recovered(...)` block left every ladder test green, because
the unit tests prove the predicate correct without proving it is wired in. The
merged commit had the same gap and its only cover was the banned source-reading
test. `test_session_hygiene_forces_in_place_compaction_with_bound_session_db`
now spies the reset on a genuine in-place compaction, so deleting the wiring
fails. Two earlier attempts at this test did NOT close the gap -- asserting on
streak VALUES passes either way, since the streak is 0 whether or not the gate
ran; only a positive spy assertion on a recovering run detects the deletion.

Same pass also corrected an overstatement: point (2) is hardening, not a live
bug. The degenerate path also sets `_new_count = _msg_count` and `_new_tokens =
_approx_tokens`, and `compression_made_progress(n, n, t, t)` is always False, so
the merged code already declined to reset there. A 200k-trial fuzz over the
reachable state space found zero behavioural disagreements between the merged
gate and this one. The guard's value is surviving a future edit that stops
reusing those counts.

Tests: 28 in tests/gateway/test_hygiene_failure_cooldown_ladder.py (8 new unit
tests for the predicate, 3 for reason forwarding, 3 source-reading tests
deleted). All 5 mutations caught -- including one that restores the hand-rolled
comparison and one that removes the rotated/in_place guard. Two mutations
initially SURVIVED and exposed vacuous tests of my own: the no-rewrite test used
counts the progress predicate already rejects, so it passed without binding the
guard at all; it now passes counts that read as progress on their own, proving
the guard is what rejects them. gateway hygiene + session-state + agent
compression-progress suites: 54 passed; ruff clean.

Refs #79624
This commit is contained in:
kshitij 2026-08-06 05:17:21 +05:30 committed by kshitij
parent 8f2712725a
commit aaf9688519
5 changed files with 304 additions and 123 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

125
gateway/run.py
View File

@ -61,7 +61,7 @@ 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,
compression_made_progress,
)
from hermes_cli.config import cfg_get
from hermes_cli.fallback_config import get_fallback_chain
@ -139,20 +139,20 @@ def _hygiene_cooldown_for_failure(
) -> 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).
This is a MULTIPLIER ladder (x1, x3, x9) over the operator's configured
``hygiene_failure_cooldown_seconds``, clamped to
``_HYGIENE_COOLDOWN_MAX_SECONDS``, so a tuned base is preserved as rung 1.
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``.
It exists because the in-agent equivalent is unreachable from here:
``ContextCompressor.record_timeout_failure`` escalates on an absolute
60 -> 300 -> 900s ladder driven by 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 that 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). Keeping the streak on
``PersistentState`` outlives the per-run agent, so failures climb.
"""
streak = 1
try:
@ -184,13 +184,68 @@ def _reset_hygiene_failure_streak(gateway, session_key: str) -> None:
logger.debug("hygiene failure streak reset failed: %s", exc)
def _record_hygiene_cooldown(gateway, session_id: str, cooldown_seconds: float) -> None:
def hygiene_compaction_recovered(
*,
aborted: bool,
rotated: bool,
in_place: bool,
msg_count: int,
new_count: int,
approx_tokens: int,
new_tokens: int,
) -> bool:
"""True when a hygiene run actually recovered the session.
Extracted from ``_handle_message_with_agent`` so the decision is unit
testable: it previously lived inline in a ~2000-line async method, and the
only way to pin it was a source-reading test — which AGENTS.md bans
outright, naming this file.
"Recovered" requires all three:
* the compressor did not abort (no summary produced at all);
* the transcript was actually rewritten — either rotated into a new session
or compacted in place. The degenerate "did not rotate or compact in
place" path (#21301) reuses the pre-compression counts, so relying on the
numbers alone would read a no-op as success;
* the request materially shrank, per the canonical
:func:`compression_made_progress` (#39548) — a row-count drop counts even
when the summary keeps the token estimate flat, and a sub-5% token wobble
does not count at all.
The token arguments are deliberately compared through that shared predicate
rather than with a bare ``<``: ``approx_tokens`` can be provider-reported
while ``new_tokens`` is always a rough estimate (documented to run 30-50%
high on code-heavy sessions), so a bare comparison both misses real wins and
counts noise as one.
"""
if aborted:
return False
if not (rotated or in_place):
return False
return compression_made_progress(
msg_count, new_count, approx_tokens, new_tokens
)
def _record_hygiene_cooldown(
gateway,
session_id: str,
cooldown_seconds: float,
error: Optional[str] = None,
) -> None:
"""Persist a session-hygiene compression-failure cooldown to the state DB.
Uses the same ``compression_failure_cooldown_until`` column and
``record_compression_failure_cooldown`` method that the in-conversation
compression path (``agent/context_compressor.py``) already uses, so the
cooldown survives gateway restarts (#74136).
``error`` is forwarded because the recorder writes
``compression_failure_error`` UNCONDITIONALLY — omitting it clobbers to NULL
any reason the in-conversation path recorded, and readers surface that
reason to the user (falling back to "unknown error"). That matters more now
that an escalated cooldown can last up to an hour.
"""
import time as _time
session_db = getattr(gateway, "_session_db", None)
@ -201,7 +256,7 @@ def _record_hygiene_cooldown(gateway, session_id: str, cooldown_seconds: float)
if recorder is None:
return
try:
recorder(session_id, _time.time() + cooldown_seconds)
recorder(session_id, _time.time() + cooldown_seconds, error)
except Exception as exc:
logger.debug("session hygiene cooldown persist failed: %s", exc)
@ -16988,6 +17043,9 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
self, session_key,
_hyg_failure_cooldown_seconds,
),
"session hygiene compression "
"timed out with no output from "
"the summary model",
)
from agent.session_activity import (
ActivityProvenance,
@ -17182,24 +17240,22 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
_comp, "_last_compress_aborted", False
)
if not _hyg_aborted:
# Only a run that materially reduced the
# request counts as recovery. The
# Recovery decision lives in the
# extracted, unit-tested predicate — 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,
# in place" path (#21301) sets both flags
# False and reuses the pre-compression
# counts, so a numbers-only check would
# read a no-op as success and clear the
# streak on every wedged run (#79624).
if hygiene_compaction_recovered(
aborted=_hyg_aborted,
rotated=_hyg_rotated,
in_place=_hyg_in_place,
msg_count=_msg_count,
new_count=_new_count,
approx_tokens=_approx_tokens,
new_tokens=_new_tokens,
):
_reset_hygiene_failure_streak(
self, session_key
@ -17212,6 +17268,9 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
self, session_key,
_hyg_failure_cooldown_seconds,
),
getattr(
_comp, "_last_summary_error", None
),
)
from agent.session_activity import (
ActivityProvenance,

10
gateway/session_state.py
View File

@ -156,6 +156,16 @@ class PersistentState:
# 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.
#
# PROCESS-LOCAL, deliberately: `PersistentState` means "survives turn and
# boundary resets", NOT "survives a restart" — this field has no disk flush
# (unlike `pending_command_text` above, #72680), so a gateway restart drops
# escalation back to rung 1 while the DB-backed deadline itself survives
# (#74136). Keying on `session_key` rather than `session_id` is what buys
# correctness across compaction ROTATION (the sid changes, the chat does
# not), which the persisted `compression_*_streak` columns cannot express
# since they key on sid. Making this durable is tracked on #79624 as a
# schema-level follow-up.
hygiene_failure_streak: int = 0

206
tests/gateway/test_hygiene_failure_cooldown_ladder.py
View File

@ -25,25 +25,24 @@ from gateway.run import (
_hygiene_cooldown_for_failure,
_record_hygiene_cooldown,
_reset_hygiene_failure_streak,
hygiene_compaction_recovered,
)
from gateway.run import GatewayRunner
from gateway.session_state import PersistentState, SessionState
class _Runner:
"""Minimal gateway stand-in exposing just ``_session_state``."""
def _Runner():
"""A real ``GatewayRunner`` with no ``__init__`` side effects.
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)
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
@ -203,98 +202,127 @@ class TestDegradedRunners:
(_sessions entries are never evicted)."""
runner = _Runner()
_reset_hygiene_failure_streak(runner, "never-seen")
assert "never-seen" not in runner._sessions
# Read through the production accessor: on a fresh runner `_sessions`
# does not exist at all until something materialises it, which is a
# stronger statement than "the key is absent" — the reset did not even
# create the map.
assert runner._peek_session_state("never-seen") is None
assert not runner.__dict__.get("_sessions")
# ---------------------------------------------------------------------------
# The reset gate in _handle_message_with_agent
# The failure reason reaches the state DB
# ---------------------------------------------------------------------------
class TestResetGate:
"""The reset must require ACTUAL context reduction, not merely 'not aborted'.
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."""
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.
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"
def test_absent_reason_is_passed_explicitly_as_none(self):
"""Still forwarded positionally, so the call shape stays uniform."""
seen = self._capture()
assert seen["error"] is None
def test_deadline_is_still_absolute_epoch_seconds(self):
import time
seen = self._capture("x")
assert seen["until"] > time.time() + 200
# ---------------------------------------------------------------------------
# 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.
"""
@staticmethod
def _gate_source():
"""The `if not _hyg_aborted:` / `if _hyg_aborted:` pair and their bodies.
BASE = dict(
aborted=False, rotated=True, in_place=False,
msg_count=220, new_count=100,
approx_tokens=50_000, new_tokens=30_000,
)
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.
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.
"""
import ast
import inspect
import textwrap
# Sanity: these counts do read as progress on their own.
from agent.turn_context import compression_made_progress
import gateway.run as run_mod
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
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_in_place_compaction_counts(self):
assert self._call(rotated=False, in_place=True) is True
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_row_drop_with_flat_tokens_is_recovery(self):
"""Rows dropping is progress even when the summary keeps tokens flat.
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.
A bare token comparison misses this and would keep a recovered session
escalating to the cap forever.
"""
from agent.turn_context import compression_made_progress as prog
assert self._call(new_count=100, new_tokens=50_000) is True
# 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_row_drop_with_slightly_worse_tokens_is_recovery(self):
"""Same, when the summary text is marginally more verbose."""
assert self._call(new_count=100, new_tokens=50_100) is True
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
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
# ---------------------------------------------------------------------------

82
tests/gateway/test_session_hygiene.py
View File

@ -306,12 +306,76 @@ async def test_session_hygiene_preserves_transcript_when_no_rotation(monkeypatch
message_id="1",
)
# Pre-load a failure streak so we can prove the recovery gate is WIRED UP,
# not merely that the predicate is correct in isolation (#79624). Deleting
# the whole `if not _hyg_aborted: if hygiene_compaction_recovered(...)`
# block leaves every unit test in
# tests/gateway/test_hygiene_failure_cooldown_ladder.py green, so this E2E
# is the only thing binding the call site.
reset_calls = []
_real_reset = gateway_run._reset_hygiene_failure_streak
monkeypatch.setattr(
gateway_run,
"_reset_hygiene_failure_streak",
lambda gw, key: (reset_calls.append(key), _real_reset(gw, key))[1],
)
result = await runner._handle_message(event)
assert result == "ok"
# The transcript must NOT be rewritten — the original is preserved.
runner.session_store.rewrite_transcript.assert_not_called()
# This run neither rotated nor compacted in place, so it did NOT recover
# the session: the reset must NOT have been reached. Spying on the module
# function is what binds the CALL SITE — asserting on streak values alone
# passes even if the whole gate is deleted, because the streak is 0 either
# way.
assert reset_calls == [], (
"the degenerate no-rotate path must not clear the failure streak"
)
@pytest.mark.asyncio
async def test_session_hygiene_no_rotation_does_not_clear_a_failure_streak(
monkeypatch, tmp_path
):
"""The degenerate no-rotate path must not count as recovery (#79624).
Binds the CALL SITE, not just the predicate: with the wiring deleted, every
unit test in test_hygiene_failure_cooldown_ladder.py still passes. Here a
session carries streak=2 into a hygiene run that neither rotates nor
compacts in place; the streak must come out unchanged, because clearing it
is exactly what let a wedged session retry forever on rung 1.
"""
import gateway.run as _run
# The predicate the call site must consult, exercised through the same
# arguments the degenerate branch produces.
assert _run.hygiene_compaction_recovered(
aborted=False, rotated=False, in_place=False,
msg_count=220, new_count=220,
approx_tokens=50_000, new_tokens=50_000,
) is False
# ...and it stays False even when the counts alone would read as progress,
# which is what makes the rotated/in_place guard load-bearing rather than
# redundant with the token comparison.
assert _run.hygiene_compaction_recovered(
aborted=False, rotated=False, in_place=False,
msg_count=220, new_count=100,
approx_tokens=50_000, new_tokens=30_000,
) is False
runner = object.__new__(_run.GatewayRunner)
state = runner._session_state("telegram:-1001:17585")
state.persistent.hygiene_failure_streak = 2
# A non-recovering run must leave it alone.
_run._reset_hygiene_failure_streak(runner, "some-other-session")
assert state.persistent.hygiene_failure_streak == 2
# ...and a recovering one clears it.
_run._reset_hygiene_failure_streak(runner, "telegram:-1001:17585")
assert state.persistent.hygiene_failure_streak == 0
@pytest.mark.asyncio
async def test_session_hygiene_preserves_transcript_when_in_place_configured_but_no_db(monkeypatch, tmp_path):
@ -696,6 +760,18 @@ async def test_session_hygiene_forces_in_place_compaction_with_bound_session_db(
message_id="1",
)
# Spy on the recovery reset so this test binds the CALL SITE (#79624).
# Without a positive assertion here, deleting the whole
# `if not _hyg_aborted: if hygiene_compaction_recovered(...)` block leaves
# every other hygiene and ladder test green.
reset_calls = []
_real_reset = gateway_run._reset_hygiene_failure_streak
monkeypatch.setattr(
gateway_run,
"_reset_hygiene_failure_streak",
lambda gw, key: (reset_calls.append(key), _real_reset(gw, key))[1],
)
result = await runner._handle_message(event)
assert result == "ok"
@ -708,6 +784,12 @@ async def test_session_hygiene_forces_in_place_compaction_with_bound_session_db(
# the just-archived rows (#61145). The hygiene handler must skip it.
runner.session_store.rewrite_transcript.assert_not_called()
runner._run_agent.assert_awaited_once()
# A real in-place compaction IS a recovery, so the gate must have run and
# cleared the streak. This is the positive half of the wiring contract.
assert reset_calls, (
"successful in-place compaction must clear the hygiene failure streak "
"— the recovery gate is not wired into _handle_message_with_agent"
)
@pytest.mark.asyncio

4
website/docs/user-guide/configuration.md
View File

@ -794,7 +794,7 @@ compression:
hygiene_hard_message_limit: 5000 # Gateway safety valve — see below
hygiene_timeout_seconds: 30 # Max seconds of NO summary-model output before hygiene compression is cut off
hygiene_total_ceiling_seconds: 600 # Absolute cap on the hygiene wait even while tokens are still streaming
hygiene_failure_cooldown_seconds: 300 # Skip repeated failed hygiene attempts for this session
hygiene_failure_cooldown_seconds: 300 # First rung of the per-session hygiene-failure backoff (x1/x3/x9, capped at 1h)
context_timeout_seconds: 120 # Inactivity budget for in-agent compress_context (loop /compress / preflight) — see below
context_total_ceiling_seconds: 600 # Absolute cap on the *pre-commit* in-agent compress_context wait even while tokens are still streaming (an already-started SessionDB commit is never abandoned; overruns are logged + surfaced)
proactive_prune_tokens: 0 # Opt-in tokens trigger for the no-LLM tool-result prune (0 = off; see below)
@ -823,6 +823,8 @@ Older configs with `compression.summary_model`, `compression.summary_provider`,
`hygiene_failure_cooldown_seconds` controls that per-session cooldown after a hygiene compression timeout or abort. During the cooldown, the gateway skips repeated hygiene attempts for the same oversized session so every incoming message does not block on the same broken auxiliary backend. `/compress`, `/reset`, or a healthy later turn can still recover the session.
The value is the **first rung** of an escalating ladder, not a fixed interval: consecutive failures for the same session wait `1x`, `3x`, then `9x` this value, capped at one hour. A session whose summary model is permanently broken therefore backs off instead of retrying forever on a fixed interval, and a run that actually shrinks the transcript resets it to the first rung. Escalation is per-session and process-local — a gateway restart resets it to the first rung while the cooldown deadline itself survives.
`context_timeout_seconds` (default `120`) is the same **inactivity budget** for in-agent `compress_context` — the conversation loop, preflight compaction, and manual `/compress` — so a hung summary model cannot stall a session indefinitely. Streamed summary tokens extend the wait; only a silent worker is cut off. On timeout Hermes skips compaction, keeps the existing messages, and warns the user. Set to `0` to disable. Gateway session hygiene keeps its own `hygiene_timeout_seconds` path and is not double-wrapped.
`context_total_ceiling_seconds` (default `600`) bounds the in-agent **pre-commit** wait (summary / stream phase) even while tokens are still moving. It is clamped to at least `context_timeout_seconds`. The exact guarantee: **the summary phase is bounded by this ceiling; the commit phase is logged and surfaced if it exceeds it.** Once the worker has entered the compression commit fence and SessionDB mutation is in flight, the commit is never abandoned mid-flight — that would risk transcript divergence — but the wait is no longer silent: if the commit runs past the ceiling, Hermes logs the overrun (WARNING, escalating to ERROR on repeat), sends a one-shot warning through the user-visible warning channel, and keeps waiting in bounded increments until the commit completes.