diff --git a/agent/chat_completion_helpers.py b/agent/chat_completion_helpers.py index 1ee1702b45e8..5ee8ee02d43b 100644 --- a/agent/chat_completion_helpers.py +++ b/agent/chat_completion_helpers.py @@ -28,7 +28,10 @@ from hermes_cli.timeouts import get_provider_request_timeout, get_provider_stale_timeout from hermes_constants import PARTIAL_STREAM_STUB_ID, FINISH_REASON_LENGTH from agent.error_classifier import FailoverReason -from agent.model_metadata import is_local_endpoint +from agent.model_metadata import ( + detect_local_server_type_cached, + is_local_endpoint, +) from agent.message_sanitization import ( _sanitize_surrogates, _repair_tool_call_arguments, @@ -103,6 +106,27 @@ def _message_chars(messages: Any) -> int: return _chars(api_payload) // 4 +def _is_detected_local_engine(base_url: Any, api_key: str = "") -> bool: + """True only for a *genuine* local inference engine at ``base_url``. + + A bare loopback proxy (e.g. the cursor-openai-api proxy on 127.0.0.1:8080) + is loopback-by-address but fronts a REMOTE backend, so it must NOT get the + unbounded local-engine treatment. ``is_local_endpoint`` is a cheap pre-gate + that keeps remote cloud URLs from ever being probed; ``detect_local_server_ + type_cached`` then probes engine-specific endpoints (Ollama /api/tags, LM + Studio, vLLM, llama.cpp) and returns ``None`` for a bare proxy. + + Shared by the stale-stream detector and the httpx read-timeout gates so the + two conditions can never silently diverge. ``api_key`` is forwarded so an + auth-gated local engine probes with the same credentials as the live stream. + """ + return ( + bool(base_url) + and is_local_endpoint(base_url) + and detect_local_server_type_cached(base_url, api_key) is not None + ) + + def _is_openai_codex_backend(agent) -> bool: base_url_lower = str(getattr(agent, "_base_url_lower", "") or "") base_url_hostname = str(getattr(agent, "_base_url_hostname", "") or "") @@ -1773,7 +1797,20 @@ def _call_chat_completions(): # prefill on large contexts before producing the first token. # Auto-increase the httpx read timeout unless the user explicitly # overrode HERMES_STREAM_READ_TIMEOUT. - if _stream_read_timeout == 120.0 and agent.base_url and is_local_endpoint(agent.base_url): + # + # Gate on a *detected* local engine, not a loopback address: a + # loopback proxy fronting a remote backend (cursor-openai-api) must + # not get the unbounded local treatment — it falls through to the + # cloud-reasoning branch below, which keeps the socket read timeout + # in step with the (finite) stale detector. See the stale-timeout + # block for the full rationale; detect_local_server_type_cached is + # memoized to stay off the hot path. + if ( + _stream_read_timeout == 120.0 + and _is_detected_local_engine( + agent.base_url, getattr(agent, "api_key", "") + ) + ): _stream_read_timeout = _base_timeout logger.debug( "Local provider detected (%s) — stream read timeout raised to %.0fs", @@ -1787,7 +1824,7 @@ def _call_chat_completions(): ): # Cloud reasoning models (e.g. Opus) routinely pause mid-stream # for minutes during extended thinking. The stale-stream - # detector is deliberately scaled up to tolerate this (180–300s, + # detector is deliberately scaled up to tolerate this (180–900s, # see the stale-timeout block below), but the raw httpx socket # read timeout defaulted to a flat 120s and fired *first* — # tearing down a healthy reasoning stream before the stale @@ -2512,7 +2549,24 @@ def _call(): # 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 agent.base_url and is_local_endpoint(agent.base_url): + # + # Gate on a *detected* local engine rather than a loopback address: the + # cursor-openai-api proxy is loopback-by-address (127.0.0.1) but fronts the + # REMOTE Cursor backend, so is_local_endpoint() alone misclassifies it as a + # slow local model server and disables the detector — turning a slow-but- + # healthy remote call into something indistinguishable from a hang. + # detect_local_server_type_cached() probes for engine-specific endpoints + # (Ollama /api/tags, LM Studio, vLLM, llama.cpp) and returns None for a bare + # proxy, so only a genuine local engine disables the detector; a loopback + # proxy falls through to the context-scaled finite timeout below. We keep + # is_local_endpoint() as a cheap pre-gate so remote cloud URLs are never + # probed, and the probe result is memoized to stay off the hot path. + if ( + _stream_stale_timeout_base == 180.0 + and _is_detected_local_engine( + agent.base_url, getattr(agent, "api_key", "") + ) + ): _stream_stale_timeout = float("inf") logger.debug("Local provider detected (%s) — stale stream timeout disabled", agent.base_url) else: @@ -2523,7 +2577,12 @@ def _call(): # spurious RemoteProtocolError ("peer closed connection"). _est_tokens = estimate_request_context_tokens(api_kwargs) if _est_tokens > 100_000: - _stream_stale_timeout = max(_stream_stale_timeout_base, 300.0) + # Very large contexts routed through the loopback cursor-openai-api + # proxy (cloud reasoning, e.g. Opus) can spend 10+ minutes in + # prefill before the first chunk — a healthy ~589s prefill was being + # killed by the old 300s ceiling and re-prefilling forever. Allow + # up to 15 min so such a prefill survives with comfortable margin. + _stream_stale_timeout = max(_stream_stale_timeout_base, 900.0) elif _est_tokens > 50_000: _stream_stale_timeout = max(_stream_stale_timeout_base, 240.0) else: diff --git a/agent/model_metadata.py b/agent/model_metadata.py index 4493eae5f1f8..9f8e46b8ec27 100644 --- a/agent/model_metadata.py +++ b/agent/model_metadata.py @@ -599,6 +599,35 @@ def detect_local_server_type(base_url: str, api_key: str = "") -> Optional[str]: return None +# Memoized results of detect_local_server_type, keyed on normalized base_url. +# The probe issues up to four blocking HTTP GETs, which is far too expensive to +# run on the streaming hot path (it would fire once per request). A given +# base_url's server identity is effectively static for the process lifetime, so +# the first probe pays the network cost and every later call is a dict lookup. +_local_server_type_cache: Dict[str, Optional[str]] = {} + + +def detect_local_server_type_cached(base_url: str, api_key: str = "") -> Optional[str]: + """Cached wrapper around :func:`detect_local_server_type`. + + Returns one of ``"ollama"|"lm-studio"|"vllm"|"llamacpp"`` when a real local + inference engine is detected at ``base_url``, else ``None`` (e.g. a loopback + proxy that merely fronts a remote backend, like the cursor-openai-api proxy, + which exposes none of the engine-specific probe endpoints). + + ``None`` results are cached too: a structural non-engine (a proxy) stays a + non-engine, and re-probing it on every stream request would defeat the + purpose. This trades a small risk of stale negatives for a momentarily + unreachable engine — acceptable, since a down engine isn't streaming anyway. + """ + normalized = _normalize_base_url(base_url) or base_url or "" + if normalized in _local_server_type_cache: + return _local_server_type_cache[normalized] + result = detect_local_server_type(base_url, api_key) + _local_server_type_cache[normalized] = result + return result + + def _iter_nested_dicts(value: Any): if isinstance(value, dict): yield value diff --git a/tests/agent/test_proxy_stale_timeout.py b/tests/agent/test_proxy_stale_timeout.py new file mode 100644 index 000000000000..b229d548f1cf --- /dev/null +++ b/tests/agent/test_proxy_stale_timeout.py @@ -0,0 +1,155 @@ +"""Tests for loopback-proxy vs real-local-engine stale-stream gating. + +The stream stale-timeout (and the matching httpx read-timeout) is disabled +only for a *detected* local inference engine. A loopback proxy that merely +fronts a remote backend — e.g. the cursor-openai-api proxy on 127.0.0.1:8080, +which exposes none of the engine-specific probe endpoints — is loopback by +address but must keep a finite, context-scaled stale timeout so a slow-but- +healthy remote call stays distinguishable from a hang. + +These pin: +- detect_local_server_type_cached memoizes the (expensive) probe per base_url +- the production gate (real is_local_endpoint + cached probe) disables the + detector only for a real engine, never for a bare loopback proxy +""" + +import pytest +from unittest.mock import patch + +import agent.model_metadata as model_metadata +from agent.model_metadata import detect_local_server_type_cached, is_local_endpoint + + +@pytest.fixture(autouse=True) +def _clear_probe_cache(): + """Reset the memoized probe results around each test.""" + model_metadata._local_server_type_cache.clear() + yield + model_metadata._local_server_type_cache.clear() + + +def _stale_detector_disabled(base_url, base_timeout=180.0): + """Mirror the production gate in chat_completion_helpers (both the stale + and read-timeout branches share this exact condition).""" + return ( + base_timeout == 180.0 + and bool(base_url) + and is_local_endpoint(base_url) + and detect_local_server_type_cached(base_url) is not None + ) + + +class TestDetectLocalServerTypeCached: + """The cached wrapper must memoize so the hot path never re-probes.""" + + def test_caches_per_base_url(self): + calls = [] + + def _fake_probe(base_url, api_key=""): + calls.append(base_url) + return "ollama" + + with patch.object(model_metadata, "detect_local_server_type", _fake_probe): + first = detect_local_server_type_cached("http://127.0.0.1:11434/v1") + second = detect_local_server_type_cached("http://127.0.0.1:11434/v1") + + assert first == "ollama" + assert second == "ollama" + assert len(calls) == 1 # probed once, served from cache thereafter + + def test_caches_none_results(self): + """A bare proxy probes to None; that negative is cached too, otherwise + the proxy would re-probe on every single stream request.""" + calls = [] + + def _fake_probe(base_url, api_key=""): + calls.append(base_url) + return None + + with patch.object(model_metadata, "detect_local_server_type", _fake_probe): + assert detect_local_server_type_cached("http://127.0.0.1:8080/v1") is None + assert detect_local_server_type_cached("http://127.0.0.1:8080/v1") is None + + assert len(calls) == 1 + + +class TestStaleDetectorGate: + """The detector is disabled only for a real detected local engine.""" + + def test_loopback_proxy_keeps_finite_timeout(self): + """cursor-openai-api proxy: loopback address but probes to None.""" + with patch.object(model_metadata, "detect_local_server_type", lambda u, api_key="": None): + assert is_local_endpoint("http://127.0.0.1:8080/v1") is True + assert _stale_detector_disabled("http://127.0.0.1:8080/v1") is False + + def test_real_local_engine_disables_detector(self): + """A detected Ollama engine keeps the long-prefill behavior (disabled).""" + with patch.object(model_metadata, "detect_local_server_type", lambda u, api_key="": "ollama"): + assert _stale_detector_disabled("http://127.0.0.1:11434/v1") is True + + def test_remote_endpoint_never_probed(self): + """Remote cloud URLs short-circuit on is_local_endpoint, so the probe + (which would error against a cloud host) is never reached.""" + def _boom(base_url, api_key=""): + raise AssertionError("detect_local_server_type must not run for remote URLs") + + with patch.object(model_metadata, "detect_local_server_type", _boom): + assert _stale_detector_disabled("https://api.openai.com/v1") is False + + def test_explicit_override_keeps_detector_finite(self): + """A non-default base timeout (user set HERMES_STREAM_STALE_TIMEOUT) + never disables the detector, even for a real engine.""" + with patch.object(model_metadata, "detect_local_server_type", lambda u, api_key="": "ollama"): + assert _stale_detector_disabled("http://127.0.0.1:11434/v1", base_timeout=240.0) is False + + +class TestRealProductionGate: + """Exercise the ACTUAL shipping predicate, not a hand-mirrored copy. + + Both the stale-stream and httpx read-timeout gates in + chat_completion_helpers now route through the shared + ``_is_detected_local_engine`` helper, so pinning that helper directly pins + the production behavior: a loopback proxy stays finite (returns False), a + detected local engine disables the detector (returns True), and remote URLs + are never probed. + """ + + def test_loopback_proxy_returns_false(self): + """cursor-openai-api proxy: loopback address, probes to None -> finite.""" + from agent.chat_completion_helpers import _is_detected_local_engine + + with patch.object(model_metadata, "detect_local_server_type", lambda u, api_key="": None): + assert _is_detected_local_engine("http://127.0.0.1:8080/v1") is False + + def test_real_local_engine_returns_true(self): + """A detected Ollama engine -> long-prefill behavior (detector off).""" + from agent.chat_completion_helpers import _is_detected_local_engine + + with patch.object(model_metadata, "detect_local_server_type", lambda u, api_key="": "ollama"): + assert _is_detected_local_engine("http://127.0.0.1:11434/v1") is True + + def test_remote_endpoint_never_probed(self): + """Remote cloud URLs short-circuit on is_local_endpoint, so the probe + (which would error against a cloud host) is never reached.""" + from agent.chat_completion_helpers import _is_detected_local_engine + + def _boom(base_url, api_key=""): + raise AssertionError("detect_local_server_type must not run for remote URLs") + + with patch.object(model_metadata, "detect_local_server_type", _boom): + assert _is_detected_local_engine("https://api.openai.com/v1") is False + + def test_api_key_forwarded_to_probe(self): + """The live credentials are forwarded so an auth-gated local engine + probes with parity to the real stream connection.""" + from agent.chat_completion_helpers import _is_detected_local_engine + + seen = {} + + def _probe(base_url, api_key=""): + seen["api_key"] = api_key + return "vllm" + + with patch.object(model_metadata, "detect_local_server_type", _probe): + assert _is_detected_local_engine("http://127.0.0.1:8000/v1", "secret-key") is True + assert seen["api_key"] == "secret-key" diff --git a/tests/tools/test_heartbeat_stale_interrupt.py b/tests/tools/test_heartbeat_stale_interrupt.py new file mode 100644 index 000000000000..833742e0c84f --- /dev/null +++ b/tests/tools/test_heartbeat_stale_interrupt.py @@ -0,0 +1,94 @@ +"""Tests that the delegate heartbeat staleness monitor interrupts an idle, +wedged subagent at the idle stale limit. + +Previously the monitor only logged and stopped touching the parent at the idle +threshold, relying on the gateway inactivity timeout to fire — which never +happens for a live (non-drain) session, leaving a wedged child with no +recovery. Now an *idle* stale child is interrupted via the same path the hard +timeout branch uses. A child sitting inside a long-running tool hit the much +more generous in-tool threshold instead and is NOT interrupted (it may still be +doing legitimate work). +""" + +import threading + +from unittest.mock import MagicMock + + +class _IdleStaleChild: + """Subagent stub that hangs idle (no tool, no iteration progress) until + the staleness monitor interrupts it.""" + + def __init__(self, *, current_tool=None): + self._subagent_id = None # skip the live-agent registry + self.model = "test/model" + self.max_iterations = 30 + self._current_tool = current_tool + self._hang = threading.Event() + self.interrupt_called = threading.Event() + + def get_activity_summary(self): + # api_call_count and current_tool never advance → counts as stale. + return { + "api_call_count": 0, + "max_iterations": self.max_iterations, + "current_tool": self._current_tool, + "last_activity_desc": "", + } + + def run_conversation(self, user_message, task_id=None, stream_callback=None): + # Block until interrupted (or a generous safety timeout) so the test + # finishes promptly when the monitor interrupts. + self._hang.wait(10.0) + return {"final_response": "", "completed": False, "api_calls": 0} + + def interrupt(self): + self.interrupt_called.set() + self._hang.set() + + +def _run(child, monkeypatch): + from tools import delegate_tool + + # Tighten the heartbeat so the stale window elapses in milliseconds, and + # leave the hard child timeout off so only the staleness monitor can act. + monkeypatch.setattr(delegate_tool, "_HEARTBEAT_INTERVAL", 0.01) + monkeypatch.setattr(delegate_tool, "_HEARTBEAT_STALE_CYCLES_IDLE", 2) + monkeypatch.setattr(delegate_tool, "_HEARTBEAT_STALE_CYCLES_IN_TOOL", 2) + monkeypatch.setattr(delegate_tool, "_get_child_timeout", lambda: None) + + parent = MagicMock() + parent._touch_activity = MagicMock() + parent._current_task_id = None + return delegate_tool._run_single_child( + task_index=0, + goal="test goal", + child=child, + parent_agent=parent, + ) + + +class TestHeartbeatStaleInterrupt: + + def test_idle_stale_child_is_interrupted(self, monkeypatch): + child = _IdleStaleChild(current_tool=None) + _run(child, monkeypatch) + assert child.interrupt_called.is_set() + + def test_in_tool_stale_child_not_interrupted(self, monkeypatch): + """A child stuck inside a tool hit the in-tool threshold; the monitor + stops touching the parent but must NOT interrupt mid-tool. We release + the child ourselves so the test can complete.""" + child = _IdleStaleChild(current_tool="terminal") + + def _release_soon(): + # Give the heartbeat loop time to reach the in-tool stale limit and + # break without interrupting, then unblock the child. + threading.Event().wait(0.2) + child._hang.set() + + releaser = threading.Thread(target=_release_soon, daemon=True) + releaser.start() + _run(child, monkeypatch) + releaser.join(timeout=2.0) + assert not child.interrupt_called.is_set() diff --git a/tools/delegate_tool.py b/tools/delegate_tool.py index 451a4748bcc0..438c5186d57e 100644 --- a/tools/delegate_tool.py +++ b/tools/delegate_tool.py @@ -1544,13 +1544,41 @@ def _heartbeat_loop(): else _HEARTBEAT_STALE_CYCLES_IDLE ) if _stale_count[0] >= stale_limit: - logger.warning( - "Subagent %d appears stale (no progress for %d " - "heartbeat cycles, tool=%s) — stopping heartbeat", - task_index, - _stale_count[0], - child_tool or "", - ) + # An *idle* child (no active tool) that hasn't advanced for + # the full idle window is genuinely wedged — nothing will + # move it again — so signal it to stop using the same + # interrupt path as the hard-timeout branch below. Without + # this the monitor only stopped touching the parent and + # relied on the gateway timeout, which never fires for a + # live (non-drain) session, leaving no recovery. + # + # A child sitting *inside* a long-running tool hit the much + # more generous in-tool threshold instead; that may still be + # legitimate work (terminal command, web fetch), so we don't + # interrupt mid-tool — just stop touching the parent and let + # the gateway timeout adjudicate, as before. + if not child_tool: + logger.warning( + "Subagent %d appears stale (no progress for %d " + "heartbeat cycles, idle) — interrupting subagent", + task_index, + _stale_count[0], + ) + try: + if hasattr(child, "interrupt"): + child.interrupt() + elif hasattr(child, "_interrupt_requested"): + child._interrupt_requested = True + except Exception: + pass + else: + logger.warning( + "Subagent %d appears stale (no progress for %d " + "heartbeat cycles, tool=%s) — stopping heartbeat", + task_index, + _stale_count[0], + child_tool, + ) break # stop touching parent, let gateway timeout fire if child_tool: