fix(delegate): keep finite stale-timeout for loopback proxy; interrupt idle-stale children

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:

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

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"

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()