diff --git a/agent/agent_init.py b/agent/agent_init.py index 722088106b0f..65c897c0c833 100644 --- a/agent/agent_init.py +++ b/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) diff --git a/agent/conversation_loop.py b/agent/conversation_loop.py index 0cc8c03ad923..4cea0eb989a0 100644 --- a/agent/conversation_loop.py +++ b/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 diff --git a/agent/turn_context.py b/agent/turn_context.py index 89952495b381..669d615847cf 100644 --- a/agent/turn_context.py +++ b/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( diff --git a/tests/agent/test_compression_max_attempts_config.py b/tests/agent/test_compression_max_attempts_config.py index cd9429cae471..8fb23f89d0a7 100644 --- a/tests/agent/test_compression_max_attempts_config.py +++ b/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 ): diff --git a/tests/run_agent/test_post_tool_compression_attempt_cap.py b/tests/run_agent/test_post_tool_compression_attempt_cap.py index 5d3a9b69e533..a15c13eb4c5a 100644 --- a/tests/run_agent/test_post_tool_compression_attempt_cap.py +++ b/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 diff --git a/tests/run_agent/test_preflight_compression_cap_e2e.py b/tests/run_agent/test_preflight_compression_cap_e2e.py new file mode 100644 index 000000000000..e29b1ecc44f5 --- /dev/null +++ b/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