diff --git a/run_agent.py b/run_agent.py index 86d185724390..47a7151c7599 100644 --- a/run_agent.py +++ b/run_agent.py @@ -5130,18 +5130,35 @@ def _call(): # httpx timeout (default 1800s) with zero feedback. The stale # detector kills the connection early so the main retry loop can # apply richer recovery (credential rotation, provider fallback). + # Stale-call timeout selection. Sliding scale based on context + # size; user can override via HERMES_API_CALL_STALE_TIMEOUT (which + # acts as a floor — larger contexts still get bumped above it). + # + # NOTE: prior versions disabled this watchdog entirely + # (`_stale_timeout = float("inf")`) for any base_url that + # `is_local_endpoint()` recognised as loopback / RFC-1918, on the + # theory that a local LLM might legitimately take longer than 5 + # min to first token. In practice that branch also matches + # **local proxies** that forward to a cloud provider (e.g. the + # OCPlatform billing proxy on 127.0.0.1:18801 fronting Anthropic). + # When the upstream stream stalls, the proxy has nothing to + # forward, the watchdog never fires, and the call wedges + # indefinitely while the anthropic SDK accumulates hundreds of + # internal retries. Sub-agents on the local proxy were the most + # affected — see hermes-patches/subagent-stale-call-local-bypass.md + # for the full incident. + # + # Operators running an actual local LLM that needs longer than + # the bumped ceiling should set HERMES_API_CALL_STALE_TIMEOUT + # explicitly (e.g. 1800 for slow llama.cpp prefill). _stale_base = float(os.getenv("HERMES_API_CALL_STALE_TIMEOUT", 300.0)) - _base_url = getattr(self, "_base_url", None) or "" - if _stale_base == 300.0 and _base_url and is_local_endpoint(_base_url): - _stale_timeout = float("inf") + _est_tokens = sum(len(str(v)) for v in api_kwargs.get("messages", [])) // 4 + if _est_tokens > 100_000: + _stale_timeout = max(_stale_base, 600.0) + elif _est_tokens > 50_000: + _stale_timeout = max(_stale_base, 450.0) else: - _est_tokens = sum(len(str(v)) for v in api_kwargs.get("messages", [])) // 4 - if _est_tokens > 100_000: - _stale_timeout = max(_stale_base, 600.0) - elif _est_tokens > 50_000: - _stale_timeout = max(_stale_base, 450.0) - else: - _stale_timeout = _stale_base + _stale_timeout = _stale_base _call_start = time.time() self._touch_activity("waiting for non-streaming API response") @@ -5838,25 +5855,29 @@ def _call(): self._close_request_openai_client(request_client, reason="stream_request_complete") _stream_stale_timeout_base = float(os.getenv("HERMES_STREAM_STALE_TIMEOUT", 180.0)) - # Local providers (Ollama, oMLX, llama-cpp) can take 300+ seconds - # for prefill on large contexts. Disable the stale detector unless - # the user explicitly set HERMES_STREAM_STALE_TIMEOUT. - if _stream_stale_timeout_base == 180.0 and self.base_url and is_local_endpoint(self.base_url): - _stream_stale_timeout = float("inf") - logger.debug("Local provider detected (%s) — stale stream timeout disabled", self.base_url) + # Stale-stream detector. Sliding scale on context size; the env + # var acts as a floor. + # + # NOTE: prior versions disabled this entirely + # (`_stream_stale_timeout = float("inf")`) for any base_url that + # `is_local_endpoint()` recognised, on the assumption that a local + # LLM might legitimately take minutes for prefill. That same + # branch silently disables the watchdog for **local proxies** that + # forward to a cloud provider (e.g. OpenClaw routing layer on + # 127.0.0.1:18801 -> Anthropic), which can leave a streaming call + # wedged forever when the upstream stalls. See + # hermes-patches/subagent-stale-call-local-bypass.md. + # + # Operators running an actual local LLM that needs longer than + # the bumped ceiling should set HERMES_STREAM_STALE_TIMEOUT + # explicitly (e.g. 600 for slow llama.cpp prefill). + _est_tokens = sum(len(str(v)) for v in api_kwargs.get("messages", [])) // 4 + if _est_tokens > 100_000: + _stream_stale_timeout = max(_stream_stale_timeout_base, 300.0) + elif _est_tokens > 50_000: + _stream_stale_timeout = max(_stream_stale_timeout_base, 240.0) else: - # Scale the stale timeout for large contexts: slow models (like Opus) - # can legitimately think for minutes before producing the first token - # when the context is large. Without this, the stale detector kills - # healthy connections during the model's thinking phase, producing - # spurious RemoteProtocolError ("peer closed connection"). - _est_tokens = sum(len(str(v)) for v in api_kwargs.get("messages", [])) // 4 - if _est_tokens > 100_000: - _stream_stale_timeout = max(_stream_stale_timeout_base, 300.0) - elif _est_tokens > 50_000: - _stream_stale_timeout = max(_stream_stale_timeout_base, 240.0) - else: - _stream_stale_timeout = _stream_stale_timeout_base + _stream_stale_timeout = _stream_stale_timeout_base t = threading.Thread(target=_call, daemon=True) t.start() diff --git a/tests/agent/test_anthropic_adapter.py b/tests/agent/test_anthropic_adapter.py index 737db01a352d..2b25f8d6b502 100644 --- a/tests/agent/test_anthropic_adapter.py +++ b/tests/agent/test_anthropic_adapter.py @@ -7,7 +7,7 @@ import pytest -from agent.prompt_caching import apply_anthropic_cache_control +from agent.prompt_caching import apply_anthropic_cache_control, make_cache_marker from agent.anthropic_adapter import ( _is_oauth_token, _refresh_oauth_token, @@ -730,7 +730,9 @@ def test_assistant_cache_control_blocks_are_preserved(self): assert assistant_blocks[0]["type"] == "text" assert assistant_blocks[0]["text"] == "Hello from assistant" - assert assistant_blocks[0]["cache_control"] == {"type": "ephemeral"} + # W1 (cache-ttl-1h-default): default marker now carries ttl='1h'. + # Use make_cache_marker() so the test tracks CACHE_TTL changes. + assert assistant_blocks[0]["cache_control"] == make_cache_marker() def test_tool_cache_control_is_preserved_on_tool_result_block(self): messages = apply_anthropic_cache_control([ @@ -752,7 +754,8 @@ def test_tool_cache_control_is_preserved_on_tool_result_block(self): assert tool_block["type"] == "tool_result" assert tool_block["tool_use_id"] == "tc_1" assert tool_block["content"] == "result" - assert tool_block["cache_control"] == {"type": "ephemeral"} + # W1 (cache-ttl-1h-default): default marker now carries ttl='1h'. + assert tool_block["cache_control"] == make_cache_marker() def test_preserved_thinking_blocks_are_rehydrated_before_tool_use(self): messages = [ diff --git a/tests/agent/test_stale_call_local_bypass.py b/tests/agent/test_stale_call_local_bypass.py new file mode 100644 index 000000000000..695ae74bed81 --- /dev/null +++ b/tests/agent/test_stale_call_local_bypass.py @@ -0,0 +1,199 @@ +"""Tests for stale-call/stream timeout selection on local-endpoint base URLs. + +Covers the regression fixed in +hermes-patches/subagent-stale-call-local-bypass.md. + +Prior versions set `_stale_timeout = float("inf")` whenever +`is_local_endpoint(base_url)` returned True, which silently disabled the +watchdog for local *proxies* that forward to a cloud provider (e.g. the +OpenClaw billing proxy on 127.0.0.1:18801 fronting Anthropic). +Sub-agents calling through such a proxy could wedge for 40+ minutes. + +The fix removes the `is_local_endpoint`-keyed bypass and applies the same +sliding scale for every base_url. Operators running a real local LLM +that needs longer than the bumped ceiling can override via +HERMES_API_CALL_STALE_TIMEOUT (non-streaming) or +HERMES_STREAM_STALE_TIMEOUT (streaming). +""" + +import math +import os + +import pytest +from unittest.mock import patch + + +# --------------------------------------------------------------------------- +# Helpers — replicate the tiny selection blocks from run_agent.py. Tested +# via direct logic so we don't have to instantiate AIAgent (and pull in +# the entire CLI/provider stack) for what is fundamentally an arithmetic +# decision. +# --------------------------------------------------------------------------- + + +def _select_non_streaming_stale_timeout(messages, base_url, env=None): + """Mirror of the block at run_agent.py around L5152.""" + env = env if env is not None else os.environ + _stale_base = float(env.get("HERMES_API_CALL_STALE_TIMEOUT", 300.0)) + _est_tokens = sum(len(str(v)) for v in (messages or [])) // 4 + if _est_tokens > 100_000: + return max(_stale_base, 600.0) + if _est_tokens > 50_000: + return max(_stale_base, 450.0) + return _stale_base + + +def _select_stream_stale_timeout(messages, base_url, env=None): + """Mirror of the block at run_agent.py around L5857.""" + env = env if env is not None else os.environ + _base = float(env.get("HERMES_STREAM_STALE_TIMEOUT", 180.0)) + _est_tokens = sum(len(str(v)) for v in (messages or [])) // 4 + if _est_tokens > 100_000: + return max(_base, 300.0) + if _est_tokens > 50_000: + return max(_base, 240.0) + return _base + + +# --------------------------------------------------------------------------- +# Non-streaming stale-call timeout +# --------------------------------------------------------------------------- + + +class TestNonStreamingStaleTimeoutLocalProxy: + """Local proxy must NOT get an infinite stale timeout (the regression).""" + + @pytest.mark.parametrize("base_url", [ + "http://127.0.0.1:18801", # OpenClaw billing proxy + "http://localhost:11434", # Ollama-style port, but hermes treats + # it the same now (no inf bypass) + "http://192.168.1.5:8000", # RFC-1918 local LAN + "http://host.docker.internal:11434", + ]) + def test_local_endpoint_does_not_disable_watchdog(self, base_url): + with patch.dict(os.environ, {}, clear=False): + os.environ.pop("HERMES_API_CALL_STALE_TIMEOUT", None) + t = _select_non_streaming_stale_timeout( + messages=[{"role": "user", "content": "hi"}], + base_url=base_url, + ) + assert math.isfinite(t), \ + f"local endpoint {base_url} got infinite stale timeout" + assert t == 300.0, \ + f"small-context default should be 300s, got {t}" + + def test_local_endpoint_large_context_bumps_to_600(self): + # 100k-token-ish message — we estimate as len(str(msg))//4 + big = "x" * 500_000 # roughly 125k "tokens" by the heuristic + with patch.dict(os.environ, {}, clear=False): + os.environ.pop("HERMES_API_CALL_STALE_TIMEOUT", None) + t = _select_non_streaming_stale_timeout( + messages=[{"role": "user", "content": big}], + base_url="http://127.0.0.1:18801", + ) + assert t == 600.0 + + +class TestNonStreamingStaleTimeoutEnvOverride: + """User-supplied env var still wins, even for local endpoints.""" + + def test_user_set_lower_floor_respected(self): + with patch.dict(os.environ, + {"HERMES_API_CALL_STALE_TIMEOUT": "120"}, + clear=False): + t = _select_non_streaming_stale_timeout( + messages=[{"role": "user", "content": "hi"}], + base_url="http://127.0.0.1:18801", + ) + # Small context → returns env value verbatim. + assert t == 120.0 + + def test_user_set_higher_floor_respected_for_local_llm(self): + # Operators running a slow local LLM bump the ceiling. + with patch.dict(os.environ, + {"HERMES_API_CALL_STALE_TIMEOUT": "1800"}, + clear=False): + t = _select_non_streaming_stale_timeout( + messages=[{"role": "user", "content": "hi"}], + base_url="http://localhost:11434", + ) + assert t == 1800.0 + + def test_env_override_floors_large_context_bumping(self): + # Env var is a floor — large-context bump only kicks in if it's + # higher than the env value. + big = "x" * 500_000 + with patch.dict(os.environ, + {"HERMES_API_CALL_STALE_TIMEOUT": "900"}, + clear=False): + t = _select_non_streaming_stale_timeout( + messages=[{"role": "user", "content": big}], + base_url="http://127.0.0.1:18801", + ) + # max(900, 600) == 900 + assert t == 900.0 + + +class TestNonStreamingStaleTimeoutRemote: + """Remote endpoints behave identically to local — no special-casing.""" + + @pytest.mark.parametrize("base_url", [ + "https://api.anthropic.com", + "https://api.openai.com", + "https://openrouter.ai/api/v1", + ]) + def test_remote_uses_same_sliding_scale(self, base_url): + with patch.dict(os.environ, {}, clear=False): + os.environ.pop("HERMES_API_CALL_STALE_TIMEOUT", None) + t = _select_non_streaming_stale_timeout( + messages=[{"role": "user", "content": "hi"}], + base_url=base_url, + ) + assert t == 300.0 + + +# --------------------------------------------------------------------------- +# Streaming stale timeout (parallel regression) +# --------------------------------------------------------------------------- + + +class TestStreamStaleTimeoutLocalProxy: + """Same regression existed in the streaming path — verify it's gone.""" + + @pytest.mark.parametrize("base_url", [ + "http://127.0.0.1:18801", + "http://localhost:11434", + "http://10.0.0.5:1234", + ]) + def test_local_endpoint_does_not_disable_stream_watchdog(self, base_url): + with patch.dict(os.environ, {}, clear=False): + os.environ.pop("HERMES_STREAM_STALE_TIMEOUT", None) + t = _select_stream_stale_timeout( + messages=[{"role": "user", "content": "hi"}], + base_url=base_url, + ) + assert math.isfinite(t), \ + f"local endpoint {base_url} got infinite stream stale timeout" + assert t == 180.0 + + def test_local_large_context_bumps_to_300(self): + big = "x" * 500_000 + with patch.dict(os.environ, {}, clear=False): + os.environ.pop("HERMES_STREAM_STALE_TIMEOUT", None) + t = _select_stream_stale_timeout( + messages=[{"role": "user", "content": big}], + base_url="http://127.0.0.1:18801", + ) + assert t == 300.0 + + +class TestStreamStaleTimeoutEnvOverride: + def test_local_llm_can_extend_via_env(self): + with patch.dict(os.environ, + {"HERMES_STREAM_STALE_TIMEOUT": "600"}, + clear=False): + t = _select_stream_stale_timeout( + messages=[{"role": "user", "content": "hi"}], + base_url="http://localhost:11434", + ) + assert t == 600.0 diff --git a/tests/run_agent/test_run_agent.py b/tests/run_agent/test_run_agent.py index 46eec2cf71df..10fb9babb39d 100644 --- a/tests/run_agent/test_run_agent.py +++ b/tests/run_agent/test_run_agent.py @@ -942,7 +942,9 @@ def test_qwen_portal_formats_messages_and_metadata(self, agent): assert kwargs["metadata"]["sessionId"] == "sess-123" assert kwargs["extra_body"]["vl_high_resolution_images"] is True assert isinstance(kwargs["messages"][0]["content"], list) - assert kwargs["messages"][0]["content"][0]["cache_control"] == {"type": "ephemeral"} + # W1 (cache-ttl-1h-default): default marker now carries ttl='1h'. + from agent.prompt_caching import make_cache_marker + assert kwargs["messages"][0]["content"][0]["cache_control"] == make_cache_marker() assert kwargs["messages"][2]["content"][0]["text"] == "hi" def test_qwen_portal_normalizes_bare_string_content_parts(self, agent):