diff --git a/agent/agent_init.py b/agent/agent_init.py index 62de3f2c540f3..bad0bf6d46fc7 100644 --- a/agent/agent_init.py +++ b/agent/agent_init.py @@ -1318,6 +1318,9 @@ def init_agent( compression_abort_on_summary_failure = str( _compression_cfg.get("abort_on_summary_failure", False) ).lower() in {"true", "1", "yes"} + compression_wall_clock_cap_seconds = _compression_cfg.get( + "wall_clock_cap_seconds", 0 + ) # Read optional explicit context_length override for the auxiliary # compression model. Custom endpoints often cannot report this via @@ -1535,6 +1538,7 @@ def init_agent( provider=agent.provider, api_mode=agent.api_mode, abort_on_summary_failure=compression_abort_on_summary_failure, + wall_clock_cap_seconds=compression_wall_clock_cap_seconds, ) agent.compression_enabled = compression_enabled diff --git a/agent/context_compressor.py b/agent/context_compressor.py index 079c4b0b5603c..76779a5432d11 100644 --- a/agent/context_compressor.py +++ b/agent/context_compressor.py @@ -19,11 +19,12 @@ import hashlib import json import logging +import math import re import time from typing import Any, Dict, List, Optional -from agent.auxiliary_client import call_llm, _is_connection_error +from agent.auxiliary_client import call_llm, _get_task_timeout, _is_connection_error from agent.context_engine import ContextEngine from agent.model_metadata import ( MINIMUM_CONTEXT_LENGTH, @@ -596,6 +597,7 @@ def __init__( provider: str = "", api_mode: str = "", abort_on_summary_failure: bool = False, + wall_clock_cap_seconds: Any = 0, ): self.model = model self.base_url = base_url @@ -612,6 +614,10 @@ def __init__( # When False (default = historical behavior), insert a # deterministic "summary unavailable" handoff and drop the middle window. self.abort_on_summary_failure = abort_on_summary_failure + self.wall_clock_cap_seconds = self._normalize_wall_clock_cap_seconds( + wall_clock_cap_seconds + ) + self._compression_deadline: Optional[float] = None self.context_length = get_model_context_length( model, base_url=base_url, api_key=api_key, @@ -681,6 +687,87 @@ def __init__( self._last_aux_model_failure_error: Optional[str] = None self._last_aux_model_failure_model: Optional[str] = None + @staticmethod + def _normalize_wall_clock_cap_seconds(value: Any) -> float: + """Normalize optional compression wall-clock cap seconds. + + ``0``/missing/false/negative/non-finite/invalid values disable the cap. + Plain numeric strings are accepted so YAML/env-style config such as + ``"420"`` works, but mixed strings like ``"420 seconds"`` are rejected. + """ + if value is None or isinstance(value, bool): + return 0.0 + if isinstance(value, str): + value = value.strip() + if not value: + return 0.0 + try: + seconds = float(value) + except (TypeError, ValueError): + return 0.0 + if not math.isfinite(seconds) or seconds <= 0: + return 0.0 + return seconds + + def _begin_wall_clock_deadline(self) -> None: + if self.wall_clock_cap_seconds > 0: + self._compression_deadline = time.monotonic() + self.wall_clock_cap_seconds + else: + self._compression_deadline = None + + def _clear_wall_clock_deadline(self) -> None: + self._compression_deadline = None + + def _ensure_wall_clock_deadline(self) -> None: + if self.wall_clock_cap_seconds > 0 and self._compression_deadline is None: + self._begin_wall_clock_deadline() + + def _remaining_wall_clock_seconds(self) -> Optional[float]: + if self.wall_clock_cap_seconds <= 0: + return None + self._ensure_wall_clock_deadline() + if self._compression_deadline is None: + return None + return self._compression_deadline - time.monotonic() + + def _wall_clock_cap_exceeded(self) -> bool: + remaining = self._remaining_wall_clock_seconds() + return remaining is not None and remaining <= 0 + + def _effective_summary_timeout(self) -> Optional[float]: + """Return an explicit timeout only when wall-clock cap is active. + + With no cap, omit the timeout kwarg so ``call_llm`` keeps its existing + auxiliary.compression.timeout resolution behavior. With a cap, bound the + call by the smaller of configured aux timeout and remaining wall-clock. + """ + remaining = self._remaining_wall_clock_seconds() + if remaining is None: + return None + if remaining <= 0: + raise TimeoutError("compression wall-clock cap exceeded") + try: + base_timeout = self._normalize_wall_clock_cap_seconds( + _get_task_timeout("compression") + ) + except Exception: + base_timeout = 0.0 + effective = min(base_timeout, remaining) if base_timeout > 0 else remaining + return max(0.001, effective) + + def _build_wall_clock_cap_fallback_summary( + self, turns_to_summarize: List[Dict[str, Any]] + ) -> str: + """Use the local deterministic fallback when the cap is exhausted.""" + error = "compression wall-clock cap exceeded" + self._last_summary_error = error + self._last_summary_fallback_used = True + self._last_summary_dropped_count = len(turns_to_summarize) + return self._build_static_fallback_summary( + turns_to_summarize, + reason=error, + ) + def update_from_response(self, usage: Dict[str, Any]): """Update tracked token usage from API response.""" self.last_prompt_tokens = usage.get("prompt_tokens", 0) @@ -1389,7 +1476,11 @@ def _generate_summary( "messages": [{"role": "user", "content": prompt}], "max_tokens": int(summary_budget * 1.3), # timeout resolved from auxiliary.compression.timeout config by call_llm + # unless a wall-clock cap must bound the current stage. } + effective_timeout = self._effective_summary_timeout() + if effective_timeout is not None: + call_kwargs["timeout"] = effective_timeout if self.summary_model: call_kwargs["model"] = self.summary_model response = call_llm(**call_kwargs) @@ -1407,6 +1498,8 @@ def _generate_summary( self._last_summary_error = None return self._with_summary_prefix(summary) except RuntimeError: + if self._wall_clock_cap_exceeded(): + return self._build_wall_clock_cap_fallback_summary(turns_to_summarize) # No provider configured — long cooldown, unlikely to self-resolve self._summary_failure_cooldown_until = time.monotonic() + _SUMMARY_FAILURE_COOLDOWN_SECONDS self._last_summary_error = "no auxiliary LLM provider configured" @@ -1452,6 +1545,8 @@ def _generate_summary( # back to the main model instead of entering a 60-second cooldown. # See issue #18458. _is_streaming_closed = _is_connection_error(e) + if self._wall_clock_cap_exceeded(): + return self._build_wall_clock_cap_fallback_summary(turns_to_summarize) if _is_json_decode and not _is_model_not_found and not _is_timeout: logger.error( "Context compression failed: auxiliary LLM returned a " @@ -1855,6 +1950,24 @@ def compress(self, messages: List[Dict[str, Any]], current_tokens: int = None, f self._last_aux_model_failure_model = None self._last_compress_aborted = False + self._begin_wall_clock_deadline() + try: + return self._compress_with_deadline( + messages, + current_tokens=current_tokens, + focus_topic=focus_topic, + force=force, + ) + finally: + self._clear_wall_clock_deadline() + + def _compress_with_deadline( + self, + messages: List[Dict[str, Any]], + current_tokens: int = None, + focus_topic: str = None, + force: bool = False, + ) -> List[Dict[str, Any]]: # Manual /compress (force=True) bypasses the failure cooldown so the # user can retry immediately after an auto-compress abort. Without # this, /compress would silently no-op for 30-60s after a failure. diff --git a/hermes_cli/config.py b/hermes_cli/config.py index 8dc3b291f4c0e..4e37924fb5f3e 100644 --- a/hermes_cli/config.py +++ b/hermes_cli/config.py @@ -1139,6 +1139,11 @@ def _ensure_hermes_home_managed(home: Path): # Default False matches historical behavior; set to # True if you'd rather pause than silently lose # context turns when your aux model is flaky. + "wall_clock_cap_seconds": 0, # Optional end-to-end compression LLM cap. + # 0/false/missing/invalid disables. When enabled, + # each summary LLM call gets timeout=min(aux timeout, + # remaining cap). If the cap is exhausted, Hermes + # uses the local deterministic fallback. "codex_gpt55_autoraise": True, # When True, gpt-5.5 on the ChatGPT Codex OAuth # route raises its compaction trigger to 85% (vs the # global `threshold` above). Codex hard-caps gpt-5.5 diff --git a/tests/agent/test_context_compressor.py b/tests/agent/test_context_compressor.py index 1b4242e0e017b..d4766a089c893 100644 --- a/tests/agent/test_context_compressor.py +++ b/tests/agent/test_context_compressor.py @@ -1,5 +1,7 @@ """Tests for agent/context_compressor.py — compression logic, thresholds, truncation fallback.""" +import time + import pytest from unittest.mock import patch, MagicMock @@ -361,6 +363,195 @@ def test_summary_failure_enters_cooldown_and_skips_retry(self): assert mock_call.call_count == 1 +class TestCompressionWallClockCap: + def _new_compressor(self, **kwargs): + with patch("agent.context_compressor.get_model_context_length", return_value=100000): + return ContextCompressor(model="test/model", quiet_mode=True, **kwargs) + + def _mock_summary_response(self, content="ok summary"): + mock_response = MagicMock() + mock_response.choices = [MagicMock()] + mock_response.choices[0].message.content = content + return mock_response + + @pytest.mark.parametrize( + ("value", "expected"), + [ + (None, 0.0), + (0, 0.0), + (False, 0.0), + (True, 0.0), + (-1, 0.0), + ("", 0.0), + ("420 seconds", 0.0), + (float("nan"), 0.0), + (float("inf"), 0.0), + ({"seconds": 420}, 0.0), + (6, 6.0), + (6.5, 6.5), + ("420", 420.0), + (" 7.25 ", 7.25), + ], + ) + def test_normalizes_wall_clock_cap_seconds(self, value, expected): + result = ContextCompressor._normalize_wall_clock_cap_seconds(value) + assert result == expected + + def test_summary_call_omits_timeout_when_cap_disabled(self): + c = self._new_compressor(wall_clock_cap_seconds=0) + with patch( + "agent.context_compressor.call_llm", + return_value=self._mock_summary_response(), + ) as mock_call: + c._generate_summary([{"role": "user", "content": "summarize me"}]) + + assert "timeout" not in mock_call.call_args.kwargs + + def test_summary_call_bounds_timeout_to_remaining_cap(self): + c = self._new_compressor(wall_clock_cap_seconds=10) + c._begin_wall_clock_deadline() + try: + with ( + patch("agent.context_compressor._get_task_timeout", return_value=300), + patch( + "agent.context_compressor.call_llm", + return_value=self._mock_summary_response(), + ) as mock_call, + ): + c._generate_summary([{"role": "user", "content": "summarize me"}]) + finally: + c._clear_wall_clock_deadline() + + timeout = mock_call.call_args.kwargs["timeout"] + assert 0 < timeout <= 10 + + def test_summary_call_preserves_lower_aux_timeout(self): + c = self._new_compressor(wall_clock_cap_seconds=300) + c._begin_wall_clock_deadline() + try: + with ( + patch("agent.context_compressor._get_task_timeout", return_value=12), + patch( + "agent.context_compressor.call_llm", + return_value=self._mock_summary_response(), + ) as mock_call, + ): + c._generate_summary([{"role": "user", "content": "summarize me"}]) + finally: + c._clear_wall_clock_deadline() + + assert 0 < mock_call.call_args.kwargs["timeout"] <= 12 + + def test_expired_cap_skips_llm_and_uses_static_fallback(self): + c = self._new_compressor(wall_clock_cap_seconds=1) + c._compression_deadline = time.monotonic() - 1 + messages = [ + {"role": "user", "content": "important task"}, + {"role": "assistant", "content": "did work"}, + ] + try: + with patch("agent.context_compressor.call_llm") as mock_call: + summary = c._generate_summary(messages) + finally: + c._clear_wall_clock_deadline() + + mock_call.assert_not_called() + assert summary.startswith(SUMMARY_PREFIX) + assert "compression wall-clock cap exceeded" in summary + assert c._last_summary_fallback_used is True + assert c._last_summary_dropped_count == len(messages) + + def test_public_compress_pre_expired_cap_skips_llm_and_uses_fallback(self): + c = self._new_compressor( + wall_clock_cap_seconds=1, + protect_first_n=1, + protect_last_n=2, + ) + messages = [{"role": "system", "content": "System"}] + [ + {"role": "user" if i % 2 == 0 else "assistant", "content": f"msg {i}"} + for i in range(8) + ] + + def _expire_before_summary(*_args, **_kwargs): + c._compression_deadline = time.monotonic() - 1 + return 6 + + with ( + patch.object(c, "_find_tail_cut_by_tokens", side_effect=_expire_before_summary), + patch("agent.context_compressor.call_llm") as mock_call, + ): + result = c.compress(messages) + + mock_call.assert_not_called() + assert len(result) < len(messages) + assert c._last_summary_fallback_used is True + assert c._last_summary_dropped_count == 4 + assert c._compression_deadline is None + assert any( + "compression wall-clock cap exceeded" in str(msg.get("content", "")) + for msg in result + ) + + def test_compress_clears_deadline_on_too_few_messages(self): + c = self._new_compressor( + wall_clock_cap_seconds=1, + protect_first_n=2, + protect_last_n=2, + ) + messages = [ + {"role": "user", "content": "one"}, + {"role": "assistant", "content": "two"}, + ] + assert c.compress(messages) == messages + assert c._compression_deadline is None + + def test_compress_clears_deadline_on_no_middle_window(self): + c = self._new_compressor( + wall_clock_cap_seconds=1, + protect_first_n=1, + protect_last_n=2, + ) + messages = [ + {"role": "system", "content": "System"}, + {"role": "user", "content": "one"}, + {"role": "assistant", "content": "two"}, + {"role": "user", "content": "three"}, + {"role": "assistant", "content": "four"}, + {"role": "user", "content": "five"}, + ] + with patch.object(c, "_find_tail_cut_by_tokens", return_value=1): + assert c.compress(messages) == messages + assert c._compression_deadline is None + + def test_wall_clock_expired_after_llm_error_uses_fallback_without_main_retry(self): + c = self._new_compressor( + wall_clock_cap_seconds=1, + summary_model_override="slow-aux", + protect_first_n=1, + protect_last_n=2, + ) + messages = [{"role": "system", "content": "System"}] + [ + {"role": "user" if i % 2 == 0 else "assistant", "content": f"msg {i}"} + for i in range(8) + ] + + def _expire_and_raise(*_args, **_kwargs): + c._compression_deadline = time.monotonic() - 1 + raise Exception("summary call timed out") + + with ( + patch.object(c, "_find_tail_cut_by_tokens", return_value=6), + patch("agent.context_compressor.call_llm", side_effect=_expire_and_raise) as mock_call, + ): + result = c.compress(messages) + + assert mock_call.call_count == 1 + assert len(result) < len(messages) + assert c._last_summary_fallback_used is True + assert c._last_aux_model_failure_model is None + assert c._compression_deadline is None + + class TestSummaryFallbackToMainModel: """When ``summary_model`` differs from the main model and the summary LLM call fails, the compressor should retry once on the main model before diff --git a/tests/run_agent/test_413_compression.py b/tests/run_agent/test_413_compression.py index 939c3682b886c..d1e69ad89e720 100644 --- a/tests/run_agent/test_413_compression.py +++ b/tests/run_agent/test_413_compression.py @@ -932,3 +932,33 @@ def test_413_still_compresses_when_enabled(self, agent): mock_compress.assert_called_once() assert result["completed"] is True assert result.get("compaction_disabled") is not True + +def test_agent_init_wires_compression_wall_clock_cap_from_config(): + cfg = { + "compression": { + "enabled": True, + "wall_clock_cap_seconds": "6", + }, + "model": {"context_length": 100000}, + } + with ( + patch("hermes_cli.config.load_config", return_value=cfg), + 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"), + patch("agent.context_compressor.get_model_context_length", return_value=100000), + ): + a = AIAgent( + api_key="test-key-1234567890", + base_url="https://openrouter.ai/api/v1", + quiet_mode=True, + skip_context_files=True, + skip_memory=True, + ) + + assert a.context_compressor.wall_clock_cap_seconds == 6.0 + +def test_default_config_disables_compression_wall_clock_cap(): + from hermes_cli.config import DEFAULT_CONFIG + + assert DEFAULT_CONFIG["compression"]["wall_clock_cap_seconds"] == 0