fix(compression): unify the attempt cap across every compression site

Follow-up to the salvaged #64010 (Kenmege) and #63870 (dombejar) commits,
making one resolved compression.max_attempts cap govern ALL per-turn
compression attempt sites:

- conversation_loop: resolve max_compression_attempts ONCE at turn start
  (it was previously re-resolved inside the API-call loop) and route the
  pre-API pressure gate through it — that gate still hardcoded
  'compression_attempts < 3' and logged 'attempt=%s/3'.
- conversation_loop: the salvaged post-tool compaction gate now uses the
  resolved cap instead of a hardcoded 3.
- turn_context: the preflight compaction loop was 'for _pass in range(3)';
  it now sizes itself from the same resolved cap.
- agent_init: harden the max_attempts parser — reject booleans (bool
  subclasses int; 'true' would coerce to 1), reject fractional floats
  instead of truncating them, keep accepting integral floats and numeric
  strings; anything else falls back to 3 (floor 1, ceiling 10 unchanged).
- tests: replace #63870's inspect.getsource source-shape test with
  behavioral loop tests (post-tool compaction fires <= cap times per turn,
  shares its budget with the pre-API gate, resets between turns); add an
  e2e test proving a 4th preflight pass runs at config cap=6 while the
  unset default still stops at 3; extend the #64010 config tests with the
  bool/float parser semantics.

Salvages #64010 by @Kenmege and #63870 by @dombejar.

agent/agent_init.py

@@ -1788,11 +1788,24 @@ def init_agent(
     # estimate above the threshold even though the messages compress fine
     # (the #62605 failure class).  Default 3 preserves current behavior, so
     # an unset key is behavior-neutral; validated >= 1, hard-capped at 10,
-    # and a non-integer value falls back to 3.
-    try:
-        compression_max_attempts = int(_compression_cfg.get("max_attempts", 3))
-    except (TypeError, ValueError):
+    # and any non-int-like value falls back to 3.  Booleans are rejected
+    # (bool subclasses int, so int(True) would silently become 1) and
+    # fractional floats are rejected rather than truncated — "4.7 attempts"
+    # is a config mistake, not a request for 4.
+    _raw_max_attempts = _compression_cfg.get("max_attempts", 3)
+    if isinstance(_raw_max_attempts, bool):
         compression_max_attempts = 3
+    elif isinstance(_raw_max_attempts, int):
+        compression_max_attempts = _raw_max_attempts
+    elif isinstance(_raw_max_attempts, float):
+        compression_max_attempts = (
+            int(_raw_max_attempts) if _raw_max_attempts.is_integer() else 3
+        )
+    else:
+        try:
+            compression_max_attempts = int(str(_raw_max_attempts).strip())
+        except (TypeError, ValueError):
+            compression_max_attempts = 3
     if compression_max_attempts < 1:
         compression_max_attempts = 3
     compression_max_attempts = min(compression_max_attempts, 10)

agent/conversation_loop.py

@@ -685,6 +685,13 @@ def run_conversation(
     truncated_tool_call_retries = 0
     truncated_response_parts: List[str] = []
     compression_attempts = 0
+    # One resolved per-turn compression attempt cap, shared by every site that
+    # consumes ``compression_attempts``: the pre-API pressure gate, the
+    # overflow/413 retry handlers, and the post-tool compaction gate.
+    # Config-driven via compression.max_attempts (parsed + validated in
+    # agent_init); default 3 preserves the prior hardcoded behavior for
+    # objects without the attribute (older pickles / minimal stubs).
+    max_compression_attempts = getattr(agent, "max_compression_attempts", 3)
     _last_preflight_pressure: Optional[int] = None
     _preflight_compression_blocked = _ctx.preflight_compression_blocked
     _turn_exit_reason = "unknown"  # Diagnostic: why the loop ended
@@ -1168,7 +1175,7 @@ def run_conversation(
         if (
             agent.compression_enabled
             and len(messages) > 1
-            and compression_attempts < 3
+            and compression_attempts < max_compression_attempts
             and not _preflight_compression_blocked
             and not _defer_preflight(request_pressure_tokens)
             and not _compression_cooldown
@@ -1177,12 +1184,13 @@ def run_conversation(
             compression_attempts += 1
             logger.info(
                 "Pre-API compression: ~%s request tokens >= %s threshold "
-                "(context=%s, attempt=%s/3)",
+                "(context=%s, attempt=%s/%s)",
                 f"{request_pressure_tokens:,}",
                 f"{int(getattr(_compressor, 'threshold_tokens', 0) or 0):,}",
                 f"{int(getattr(_compressor, 'context_length', 0) or 0):,}"
                 if getattr(_compressor, "context_length", 0) else "unknown",
                 compression_attempts,
+                max_compression_attempts,
             )
             agent._emit_status(
                 f"📦 Pre-API compression: ~{request_pressure_tokens:,} tokens "
@@ -1252,9 +1260,6 @@ def run_conversation(
         retry_count = 0
         max_retries = agent._api_max_retries
         _retry = TurnRetryState()
-        # Config-driven via compression.max_attempts (parsed + validated in
-        # agent_init).  Default 3 preserves the prior hardcoded behavior.
-        max_compression_attempts = getattr(agent, "max_compression_attempts", 3)
 
         finish_reason = "stop"
         response = None  # Guard against UnboundLocalError if all retries fail
@@ -5179,7 +5184,7 @@ def run_conversation(
 
                 if (
                     agent.compression_enabled
-                    and compression_attempts < 3
+                    and compression_attempts < max_compression_attempts
                     and _compressor.should_compress(_real_tokens)
                 ):
                     compression_attempts += 1

agent/turn_context.py

@@ -665,7 +665,13 @@ def build_turn_context(
                 f">= {_compressor.threshold_tokens:,} threshold. "
                 "This may take a moment."
             )
-            for _pass in range(3):
+            # Preflight passes honor the same configured per-turn cap
+            # (compression.max_attempts) as the loop's compression sites;
+            # default 3 preserves the prior hardcoded behavior.
+            _max_preflight_passes = max(
+                1, int(getattr(agent, "max_compression_attempts", 3) or 3)
+            )
+            for _pass in range(_max_preflight_passes):
                 _orig_len = len(messages)
                 _orig_tokens = _preflight_tokens
                 messages, active_system_prompt = agent._compress_context(

tests/agent/test_compression_max_attempts_config.py

@@ -86,6 +86,27 @@ class TestCompressionMaxAttemptsConfig:
         agent = _make_agent(monkeypatch, tmp_path, max_attempts="lots")
         assert agent.max_compression_attempts == 3
 
+    def test_boolean_is_rejected_not_coerced(self, monkeypatch, tmp_path):
+        # bool subclasses int: int(True) == 1 would silently near-disable
+        # compression retries. YAML `max_attempts: true` must fall back to 3.
+        agent = _make_agent(monkeypatch, tmp_path, max_attempts=True)
+        assert agent.max_compression_attempts == 3
+        agent = _make_agent(monkeypatch, tmp_path, max_attempts=False)
+        assert agent.max_compression_attempts == 3
+
+    def test_fractional_float_is_rejected_not_truncated(self, monkeypatch, tmp_path):
+        # "4.7 attempts" is a config mistake, not a request for 4.
+        agent = _make_agent(monkeypatch, tmp_path, max_attempts=4.7)
+        assert agent.max_compression_attempts == 3
+
+    def test_integral_float_and_numeric_string_are_accepted(
+        self, monkeypatch, tmp_path
+    ):
+        agent = _make_agent(monkeypatch, tmp_path, max_attempts=5.0)
+        assert agent.max_compression_attempts == 5
+        agent = _make_agent(monkeypatch, tmp_path, max_attempts="6")
+        assert agent.max_compression_attempts == 6
+
     def test_loop_pickup_degrades_to_default_when_attribute_missing(
         self, monkeypatch, tmp_path
     ):

tests/run_agent/test_post_tool_compression_attempt_cap.py

@@ -1,30 +1,206 @@
-"""Regression guard for the post-tool automatic-compression attempt cap.
+"""Behavioral regression tests for the post-tool compression attempt cap.
 
-The pre-API and overflow compression paths share ``compression_attempts`` as a
-three-pass per-turn backstop. The post-tool path must use the same counter;
-otherwise a long tool loop can compact after every tool response for the
-lifetime of the turn.
+The pre-API pressure gate, the overflow/413 error handlers, and the post-tool
+compaction gate all share ``compression_attempts`` as a per-turn backstop,
+bounded by the resolved ``compression.max_attempts`` cap (default 3).  Before
+the fix the post-tool path neither checked nor incremented the counter, so a
+long tool loop could compact after every tool response for the lifetime of
+the turn.
+
+These tests drive ``run_conversation()`` through real tool iterations with a
+compressor that always demands compression and assert ``_compress_context``
+fires at most ``max_compression_attempts`` times per turn — no source
+inspection, only observable behavior.
 """
+
 from __future__ import annotations
 
-import inspect
+import json
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from run_agent import AIAgent
 
 
-def _post_tool_compression_block() -> str:
-    from agent import conversation_loop
-
-    source = inspect.getsource(conversation_loop.run_conversation)
-    start = source.index("# Use real token counts from the API response")
-    end = source.index("# Save session log incrementally", start)
-    return source[start:end]
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
 
 
-def test_post_tool_compression_uses_shared_per_turn_attempt_cap():
-    block = _post_tool_compression_block()
-
-    assert "compression_attempts < 3" in block, (
-        "post-tool compression must stop after the shared three attempts per turn"
+def _tool_call(i: int):
+    return SimpleNamespace(
+        id=f"call_{i}",
+        type="function",
+        function=SimpleNamespace(name="web_search", arguments='{"query": "x"}'),
     )
-    assert "compression_attempts += 1" in block, (
-        "post-tool compression must consume one shared per-turn attempt"
+
+
+def _tool_response(i: int):
+    msg = SimpleNamespace(
+        content=None,
+        reasoning_content=None,
+        reasoning=None,
+        tool_calls=[_tool_call(i)],
     )
+    choice = SimpleNamespace(message=msg, finish_reason="tool_calls")
+    return SimpleNamespace(choices=[choice], model="test/model", usage=None)
+
+
+def _stop_response():
+    msg = SimpleNamespace(
+        content="done",
+        reasoning_content=None,
+        reasoning=None,
+        tool_calls=None,
+    )
+    choice = SimpleNamespace(message=msg, finish_reason="stop")
+    return SimpleNamespace(choices=[choice], model="test/model", usage=None)
+
+
+def _make_tool_defs(*names: str) -> list:
+    return [
+        {
+            "type": "function",
+            "function": {
+                "name": n,
+                "description": f"{n} tool",
+                "parameters": {"type": "object", "properties": {}},
+            },
+        }
+        for n in names
+    ]
+
+
+def _pressured_compressor() -> MagicMock:
+    """A compressor that always reports context pressure after tools run.
+
+    ``should_defer_preflight_to_real_usage`` returns True so the turn-start
+    preflight and the pre-API pressure gate stand down — isolating the
+    post-tool gate as the only compression site under test.
+    """
+    compressor = MagicMock()
+    compressor.protect_first_n = 3
+    compressor.protect_last_n = 20
+    compressor.threshold_tokens = 10_000
+    compressor.context_length = 200_000
+    compressor.last_prompt_tokens = 150_000
+    compressor.should_compress.return_value = True
+    compressor.should_defer_preflight_to_real_usage.return_value = True
+    compressor.get_active_compression_failure_cooldown.return_value = None
+    return compressor
+
+
+@pytest.fixture()
+def agent():
+    with (
+        patch("run_agent.get_tool_definitions", return_value=_make_tool_defs("web_search")),
+        patch("run_agent.check_toolset_requirements", return_value={}),
+        patch("run_agent.OpenAI"),
+    ):
+        a = AIAgent(
+            api_key="test-key-1234567890",
+            base_url="https://openrouter.ai/api/v1",
+            quiet_mode=True,
+            skip_context_files=True,
+            skip_memory=True,
+            max_iterations=10,
+        )
+    a.client = MagicMock()
+    a._cached_system_prompt = "You are helpful."
+    a._use_prompt_caching = False
+    a._disable_streaming = True
+    a.tool_delay = 0
+    a.save_trajectories = False
+    a.compression_enabled = True
+    a.context_compressor = _pressured_compressor()
+    return a
+
+
+def _run_tool_loop(agent, n_tool_iterations: int):
+    """Drive one turn: ``n_tool_iterations`` tool calls, then a stop."""
+    responses = [_tool_response(i) for i in range(n_tool_iterations)]
+    responses.append(_stop_response())
+    agent.client.chat.completions.create.side_effect = responses
+
+    compress_calls = []
+
+    def _fake_compress(messages, system_message, **_kwargs):
+        compress_calls.append(len(messages))
+        return messages, "compressed prompt"
+
+    with (
+        patch.object(agent, "_compress_context", side_effect=_fake_compress),
+        patch.object(agent, "_persist_session"),
+        patch.object(agent, "_save_trajectory"),
+        patch.object(agent, "_cleanup_task_resources"),
+        patch(
+            "run_agent.handle_function_call",
+            lambda name, args, task_id=None, **kwargs: json.dumps({"ok": True}),
+        ),
+    ):
+        result = agent.run_conversation("do a lot of tool work")
+
+    return result, compress_calls
+
+
+# ---------------------------------------------------------------------------
+# Tests
+# ---------------------------------------------------------------------------
+
+
+class TestPostToolCompressionAttemptCap:
+    def test_post_tool_compression_capped_at_default_three(self, agent):
+        """7 tool iterations under constant pressure → exactly 3 compactions.
+
+        Before the fix the post-tool gate re-fired after every tool response
+        (7 compactions here); the shared per-turn counter caps it at the
+        resolved default of 3.
+        """
+        assert agent.max_compression_attempts == 3  # config default
+        result, compress_calls = _run_tool_loop(agent, n_tool_iterations=7)
+
+        assert result["completed"] is True
+        assert len(compress_calls) == 3, (
+            f"post-tool compression must stop at the per-turn cap (3), "
+            f"got {len(compress_calls)} compactions"
+        )
+
+    def test_post_tool_compression_honors_configured_cap(self, agent):
+        """A raised compression.max_attempts cap lets more rounds run."""
+        agent.max_compression_attempts = 5
+        result, compress_calls = _run_tool_loop(agent, n_tool_iterations=8)
+
+        assert result["completed"] is True
+        assert len(compress_calls) == 5
+
+    def test_post_tool_compression_shares_counter_with_pre_api_gate(self, agent):
+        """Pre-API compactions consume the same per-turn budget.
+
+        Let the pre-API pressure gate fire once (defer disabled for the first
+        check), then keep the pressure on through tool iterations: the
+        combined total must still respect the cap.
+        """
+        # First pre-API check does not defer → pre-API gate fires once;
+        # afterwards defer again so only the post-tool gate keeps firing.
+        defers = iter([False])
+        agent.context_compressor.should_defer_preflight_to_real_usage.side_effect = (
+            lambda _t: next(defers, True)
+        )
+        result, compress_calls = _run_tool_loop(agent, n_tool_iterations=7)
+
+        assert result["completed"] is True
+        assert len(compress_calls) == 3, (
+            "pre-API and post-tool compactions must share one per-turn "
+            f"attempt budget, got {len(compress_calls)} total compactions"
+        )
+
+    def test_cap_is_per_turn_not_per_session(self, agent):
+        """A fresh turn gets a fresh attempt budget."""
+        _result, first = _run_tool_loop(agent, n_tool_iterations=5)
+        agent.client.chat.completions.create.side_effect = None
+        _result, second = _run_tool_loop(agent, n_tool_iterations=5)
+
+        assert len(first) == 3
+        assert len(second) == 3

tests/run_agent/test_preflight_compression_cap_e2e.py

@@ -0,0 +1,186 @@
+"""E2E: compression.max_attempts=6 drives a 4th+ preflight compaction pass.
+
+The turn-start preflight loop in ``agent/turn_context.py`` was hardcoded to
+``range(3)``: even when every pass made real progress and the request stayed
+over threshold, the 4th pass never ran, regardless of configuration.  The
+loop now sizes itself from the same resolved ``compression.max_attempts`` cap
+as the conversation loop's compression sites.
+
+This test builds a real ``AIAgent`` from a config with
+``compression.max_attempts: 6`` (the config-driven path through
+``agent_init``), then drives a full ``run_conversation()`` turn in which the
+estimated request size keeps shrinking ~10% per compaction but stays above
+threshold — the exact "progress, but not enough yet" shape that legitimately
+needs more than three rounds.  With cap=6 the preflight must run a 4th pass
+(and ultimately all six).
+"""
+
+from __future__ import annotations
+
+import contextlib
+import io
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+from hermes_state import SessionDB
+from run_agent import AIAgent
+
+
+def _config(max_attempts) -> dict:
+    return {
+        "compression": {
+            "enabled": True,
+            "threshold": 0.50,
+            "target_ratio": 0.20,
+            "protect_first_n": 3,
+            "protect_last_n": 20,
+            "max_attempts": max_attempts,
+        },
+        "prompt_caching": {"cache_ttl": "5m"},
+        "sessions": {},
+        "bedrock": {},
+    }
+
+
+def _stop_response():
+    msg = SimpleNamespace(
+        content="done",
+        reasoning_content=None,
+        reasoning=None,
+        tool_calls=None,
+    )
+    choice = SimpleNamespace(message=msg, finish_reason="stop")
+    return SimpleNamespace(choices=[choice], model="test/model", usage=None)
+
+
+def _make_agent(monkeypatch, tmp_path: Path, *, max_attempts) -> AIAgent:
+    from hermes_cli import config as config_mod
+
+    monkeypatch.setattr(
+        config_mod, "load_config", lambda: _config(max_attempts)
+    )
+    db = SessionDB(db_path=tmp_path / "state.db")
+    with (
+        contextlib.redirect_stdout(io.StringIO()),
+        patch("run_agent.get_tool_definitions", return_value=[]),
+        patch("run_agent.check_toolset_requirements", return_value={}),
+        patch("run_agent.OpenAI"),
+    ):
+        agent = AIAgent(
+            base_url="https://openrouter.ai/api/v1",
+            api_key="test-key",
+            model="test/model",
+            enabled_toolsets=[],
+            disabled_toolsets=[],
+            quiet_mode=True,
+            skip_memory=True,
+            skip_context_files=True,
+            session_db=db,
+            session_id="preflight-cap-e2e",
+        )
+    agent.client = MagicMock()
+    agent._cached_system_prompt = "You are helpful."
+    agent._use_prompt_caching = False
+    agent._disable_streaming = True
+    agent.tool_delay = 0
+    agent.save_trajectories = False
+    return agent
+
+
+def test_preflight_runs_fourth_compaction_pass_at_cap_six(monkeypatch, tmp_path):
+    agent = _make_agent(monkeypatch, tmp_path, max_attempts=6)
+    # Config-driven attach seam (agent_init) resolved the raised cap.
+    assert agent.max_compression_attempts == 6
+
+    # Keep the request permanently over threshold while every compaction
+    # makes material (~10% > the 5% progress floor) headway.
+    compressor = agent.context_compressor
+    compressor.threshold_tokens = 50_000
+
+    estimate_state = {"tokens": 1_000_000.0, "calls": 0}
+
+    def _shrinking_estimate(*_args, **_kwargs):
+        if estimate_state["calls"]:
+            estimate_state["tokens"] *= 0.9
+        estimate_state["calls"] += 1
+        return int(estimate_state["tokens"])
+
+    compress_calls = []
+
+    def _fake_compress(messages, system_message, **_kwargs):
+        compress_calls.append(len(messages))
+        return messages, "compressed prompt"
+
+    # 60 messages > protect_first_n + protect_last_n + 1, so the cheap
+    # preflight count gate opens without patching internals.
+    history = [
+        {"role": "user" if i % 2 == 0 else "assistant", "content": f"msg {i}"}
+        for i in range(60)
+    ]
+    agent.client.chat.completions.create.return_value = _stop_response()
+
+    with (
+        patch(
+            "agent.turn_context.estimate_request_tokens_rough",
+            side_effect=_shrinking_estimate,
+        ),
+        patch.object(agent, "_compress_context", side_effect=_fake_compress),
+        patch.object(agent, "_persist_session"),
+        patch.object(agent, "_save_trajectory"),
+        patch.object(agent, "_cleanup_task_resources"),
+    ):
+        result = agent.run_conversation("hello", conversation_history=history)
+
+    assert result["completed"] is True
+    # The old hardcoded range(3) made a 4th pass impossible; cap=6 must
+    # deliver it (and, with steady progress over threshold, all six).
+    assert len(compress_calls) >= 4, (
+        f"expected a 4th preflight compaction pass at cap=6, "
+        f"got {len(compress_calls)} passes"
+    )
+    assert len(compress_calls) == 6
+
+
+def test_preflight_still_stops_at_default_three(monkeypatch, tmp_path):
+    """Unset compression.max_attempts keeps the historical 3-pass behavior."""
+    agent = _make_agent(monkeypatch, tmp_path, max_attempts=None)
+    assert agent.max_compression_attempts == 3
+
+    compressor = agent.context_compressor
+    compressor.threshold_tokens = 50_000
+
+    estimate_state = {"tokens": 1_000_000.0, "calls": 0}
+
+    def _shrinking_estimate(*_args, **_kwargs):
+        if estimate_state["calls"]:
+            estimate_state["tokens"] *= 0.9
+        estimate_state["calls"] += 1
+        return int(estimate_state["tokens"])
+
+    compress_calls = []
+
+    def _fake_compress(messages, system_message, **_kwargs):
+        compress_calls.append(len(messages))
+        return messages, "compressed prompt"
+
+    history = [
+        {"role": "user" if i % 2 == 0 else "assistant", "content": f"msg {i}"}
+        for i in range(60)
+    ]
+    agent.client.chat.completions.create.return_value = _stop_response()
+
+    with (
+        patch(
+            "agent.turn_context.estimate_request_tokens_rough",
+            side_effect=_shrinking_estimate,
+        ),
+        patch.object(agent, "_compress_context", side_effect=_fake_compress),
+        patch.object(agent, "_persist_session"),
+        patch.object(agent, "_save_trajectory"),
+        patch.object(agent, "_cleanup_task_resources"),
+    ):
+        result = agent.run_conversation("hello", conversation_history=history)
+
+    assert result["completed"] is True
+    assert len(compress_calls) == 3