fix(gateway): keep silent in-flight tools alive

agent/tool_executor.py

@@ -196,6 +196,24 @@ def _resolve_concurrent_tool_timeout() -> float | None:
     )
 
 
+def _clear_current_tool_if_idle_locked(agent) -> None:
+    """Clear the marker when called with the worker lock already held."""
+    if not agent._tool_worker_threads:
+        agent._current_tool = None
+        agent._current_tool_started_at = None
+
+
+def _clear_current_tool_if_idle(agent) -> None:
+    """Clear the activity marker only after the last concurrent worker exits."""
+    worker_lock = getattr(agent, "_tool_worker_threads_lock", None)
+    worker_threads = getattr(agent, "_tool_worker_threads", None)
+    if worker_lock is None or worker_threads is None:
+        agent._current_tool = None
+        return
+    with worker_lock:
+        _clear_current_tool_if_idle_locked(agent)
+
+
 def _flush_session_db_after_tool_progress(
     agent,
     messages: list,
@@ -1008,7 +1026,17 @@ def _begin_tool_execution(
                 f"{args_preview}"
             )
 
-    agent._current_tool = function_name
+    tool_started_at = time.monotonic()
+    worker_lock = getattr(agent, "_tool_worker_threads_lock", None)
+    if worker_lock is None:
+        agent._current_tool = function_name
+        agent._current_tool_started_at = tool_started_at
+    else:
+        with worker_lock:
+            agent._current_tool = function_name
+            current_started_at = getattr(agent, "_current_tool_started_at", None)
+            if current_started_at is None or tool_started_at < current_started_at:
+                agent._current_tool_started_at = tool_started_at
     agent._touch_activity(f"executing tool: {function_name}")
     try:
         from tools.environments.base import set_activity_callback
@@ -1292,6 +1320,12 @@ def _run_tool(
         _worker_tid = threading.current_thread().ident
         with agent._tool_worker_threads_lock:
             agent._tool_worker_threads.add(_worker_tid)
+            if not agent._current_tool:
+                agent._current_tool = tool_names_str
+            if getattr(agent, "_current_tool_started_at", None) is None:
+                # The watchdog must have a wall-time anchor even when a worker
+                # reaches dispatch without going through tool preflight.
+                agent._current_tool_started_at = time.monotonic()
         # Race: if the agent was interrupted between fan-out (which
         # snapshotted an empty/earlier set) and our registration, apply
         # the interrupt to our own tid now so is_interrupted() inside
@@ -1451,6 +1485,7 @@ def _execute(next_args: dict[str, Any]) -> Any:
             # into _interrupted_threads, poisoning the recycled thread.
             with agent._tool_worker_threads_lock:
                 agent._tool_worker_threads.discard(_worker_tid)
+                _clear_current_tool_if_idle_locked(agent)
             try:
                 _ra()._set_interrupt(False, _worker_tid)
             except Exception:
@@ -1775,7 +1810,7 @@ def _execute(next_args: dict[str, Any]) -> Any:
                 logging.debug("Tool %s completed in %.2fs", function_name, tool_duration)
                 logging.debug("Tool result (%d chars): %s", len(function_result), function_result)
 
-        agent._current_tool = None
+        _clear_current_tool_if_idle(agent)
         _status_suffix = " (error)" if is_error else ""
         agent._touch_activity(f"tool completed: {name} ({tool_duration:.1f}s){_status_suffix}")
 
@@ -2691,7 +2726,7 @@ def _execute(next_args: dict) -> Any:
             except Exception as _ver_err:
                 logging.debug("file-mutation verifier record failed: %s", _ver_err)
 
-        agent._current_tool = None
+        _clear_current_tool_if_idle(agent)
         _status_suffix = " (error)" if _is_error_result else ""
         agent._touch_activity(f"tool completed: {function_name} ({tool_duration:.1f}s){_status_suffix}")
 

gateway/run.py

@@ -3255,6 +3255,24 @@ def _dequeue_pending_event(adapter, session_key: str) -> MessageEvent | None:
 _INTERRUPT_REASON_SSE_DISCONNECT = "SSE client disconnected"
 _INTERRUPT_REASON_GATEWAY_SHUTDOWN = "Gateway shutting down"
 _INTERRUPT_REASON_GATEWAY_RESTART = "Gateway restarting"
+_DEFAULT_IN_FLIGHT_TOOL_TIMEOUT = 420.0
+
+
+def _resolve_in_flight_tool_timeout() -> Optional[float]:
+    """Resolve the independent wall-time ceiling for an executing tool."""
+    raw = os.getenv("HERMES_CONCURRENT_TOOL_TIMEOUT_S", "").strip()
+    if not raw:
+        return _DEFAULT_IN_FLIGHT_TOOL_TIMEOUT
+    try:
+        value = float(raw)
+    except ValueError:
+        logger.warning(
+            "invalid HERMES_CONCURRENT_TOOL_TIMEOUT_S=%r; using %.0fs",
+            raw,
+            _DEFAULT_IN_FLIGHT_TOOL_TIMEOUT,
+        )
+        return _DEFAULT_IN_FLIGHT_TOOL_TIMEOUT
+    return value if value > 0 else None
 
 
 def _reap_gateway_turn_processes(
@@ -3441,6 +3459,7 @@ def _watch_gateway_turn_inactivity(
     task_id: str,
     process_baseline,
     timeout: float,
+    in_flight_tool_timeout: Optional[float] = _DEFAULT_IN_FLIGHT_TOOL_TIMEOUT,
     worker_done: threading.Event,
     timeout_fired: threading.Event,
     cleanup_lock: threading.Lock,
@@ -3453,13 +3472,27 @@ def _watch_gateway_turn_inactivity(
         if agent is None or not hasattr(agent, "get_activity_summary"):
             continue
         try:
-            idle_seconds = float(
-                agent.get_activity_summary().get("seconds_since_activity", 0.0)
-            )
+            activity = agent.get_activity_summary()
+            current_tool = activity.get("current_tool")
+            if current_tool:
+                started_at = activity.get("tool_started_at")
+                if (
+                    in_flight_tool_timeout is None
+                    or started_at is None
+                    or time.monotonic() - float(started_at) < in_flight_tool_timeout
+                ):
+                    continue
+                logger.warning(
+                    "In-flight tool %s exceeded its %.0fs wall-time ceiling",
+                    current_tool,
+                    in_flight_tool_timeout,
+                )
+            else:
+                idle_seconds = float(activity.get("seconds_since_activity", 0.0))
+                if idle_seconds < timeout:
+                    continue
         except Exception:
             continue
-        if idle_seconds < timeout:
-            continue
         _abandon_timed_out_gateway_turn(
             agent_holder=agent_holder,
             task_id=task_id,
@@ -28939,6 +28972,7 @@ def _stream_confirmed_final_delivery(
             # Default 1800s (30 min inactivity).  0 = unlimited.
             _agent_timeout_raw = _float_env("HERMES_AGENT_TIMEOUT", 1800)
             _agent_timeout = _agent_timeout_raw if _agent_timeout_raw > 0 else None
+            _in_flight_tool_timeout = _resolve_in_flight_tool_timeout()
             _agent_warning_raw = _float_env("HERMES_AGENT_TIMEOUT_WARNING", 900)
             _agent_warning = _agent_warning_raw if _agent_warning_raw > 0 else None
             _warning_fired = False
@@ -28999,6 +29033,7 @@ def _run_sync_with_timeout_lifecycle():
                         "task_id": _turn_task_id,
                         "process_baseline": _turn_process_baseline,
                         "timeout": _agent_timeout,
+                        "in_flight_tool_timeout": _in_flight_tool_timeout,
                         "worker_done": _turn_worker_done,
                         "timeout_fired": _turn_timeout_fired,
                         "cleanup_lock": _turn_cleanup_lock,
@@ -29091,6 +29126,8 @@ def _run_sync_with_timeout_lifecycle():
                         try:
                             _act = _agent_ref.get_activity_summary()
                             _idle_secs = _act.get("seconds_since_activity", 0.0)
+                            if _act.get("current_tool"):
+                                _idle_secs = 0.0
                         except Exception:
                             pass
                     # Staged warning: fire once before escalating to full timeout.

run_agent.py

@@ -4343,6 +4343,7 @@ def get_activity_summary(self) -> dict:
             last_activity_provenance=provenance,
             extra={
             "current_tool": self._current_tool,
+            "tool_started_at": getattr(self, "_current_tool_started_at", None),
             "api_call_count": self._api_call_count,
             "max_iterations": self.max_iterations,
             "budget_used": self.iteration_budget.used,

tests/gateway/test_abandoned_turn_process_cleanup.py

@@ -11,12 +11,27 @@
 
 
 class _IdleAgent:
-    def __init__(self, idle_seconds=60.0):
+    def __init__(
+        self,
+        idle_seconds=60.0,
+        current_tool=None,
+        activity_observed=None,
+        tool_started_at=None,
+    ):
         self.idle_seconds = idle_seconds
+        self.current_tool = current_tool
+        self.activity_observed = activity_observed
+        self.tool_started_at = tool_started_at
         self.interrupts = []
 
     def get_activity_summary(self):
-        return {"seconds_since_activity": self.idle_seconds}
+        if self.activity_observed is not None:
+            self.activity_observed.set()
+        return {
+            "seconds_since_activity": self.idle_seconds,
+            "current_tool": self.current_tool,
+            "tool_started_at": self.tool_started_at,
+        }
 
     def interrupt(self, reason):
         self.interrupts.append(reason)
@@ -67,6 +82,80 @@ def test_thread_watchdog_reaps_only_processes_created_by_timed_out_turn(monkeypa
     ]
 
 
+def test_thread_watchdog_does_not_abandon_silent_in_flight_tool(monkeypatch):
+    activity_observed = threading.Event()
+    agent = _IdleAgent(
+        current_tool="silent_tool",
+        activity_observed=activity_observed,
+    )
+    worker_done, timeout_fired, cleanup_lock = _state()
+    monkeypatch.setattr(
+        process_registry,
+        "kill_started_since",
+        lambda *_args, **_kwargs: (_ for _ in ()).throw(
+            AssertionError("an active tool must not be reaped")
+        ),
+    )
+
+    watchdog = threading.Thread(
+        target=_watch_gateway_turn_inactivity,
+        kwargs={
+            "agent_holder": [agent],
+            "task_id": "session-a",
+            "process_baseline": frozenset(),
+            "timeout": 30.0,
+            "worker_done": worker_done,
+            "timeout_fired": timeout_fired,
+            "cleanup_lock": cleanup_lock,
+            "poll_interval": 0.01,
+        },
+    )
+    watchdog.start()
+    assert activity_observed.wait(5)
+    assert not timeout_fired.is_set()
+    worker_done.set()
+    watchdog.join(timeout=1)
+
+    assert not timeout_fired.is_set()
+    assert agent.interrupts == []
+
+
+def test_thread_watchdog_abandons_hung_in_flight_tool(monkeypatch):
+    agent = _IdleAgent(current_tool="hung_tool", tool_started_at=0.0)
+    worker_done, timeout_fired, cleanup_lock = _state()
+    calls = []
+    monkeypatch.setattr(
+        process_registry,
+        "kill_started_since",
+        lambda task_id, baseline, *, source: calls.append(
+            (task_id, baseline, source)
+        )
+        or 1,
+    )
+
+    watchdog = threading.Thread(
+        target=_watch_gateway_turn_inactivity,
+        kwargs={
+            "agent_holder": [agent],
+            "task_id": "session-a",
+            "process_baseline": frozenset(),
+            "timeout": 30.0,
+            "in_flight_tool_timeout": 0.01,
+            "worker_done": worker_done,
+            "timeout_fired": timeout_fired,
+            "cleanup_lock": cleanup_lock,
+            "poll_interval": 0.01,
+        },
+    )
+    watchdog.start()
+    watchdog.join(timeout=1)
+
+    assert not watchdog.is_alive()
+    assert timeout_fired.is_set()
+    assert agent.interrupts == ["Execution timed out (inactivity)"]
+    assert calls == [("session-a", frozenset(), "gateway_turn_timeout")]
+
+
 def test_completed_worker_wins_race_and_preserves_background_process(monkeypatch):
     agent = _IdleAgent()
     worker_done, timeout_fired, cleanup_lock = _state()

tests/run_agent/test_run_agent.py

@@ -2058,6 +2058,46 @@ def fake_handle(name, args, task_id, **kwargs):
         assert messages[1]["tool_call_id"] == "c2"
         assert "result_fast" in messages[1]["content"]
 
+    def test_concurrent_activity_marker_survives_timed_out_sibling(
+        self, agent, monkeypatch
+    ):
+        """A detached sibling keeps the activity marker until it exits."""
+        tc1 = _mock_tool_call(name="web_search", arguments='{"q":"fast"}', call_id="c1")
+        tc2 = _mock_tool_call(name="web_search", arguments='{"q":"slow"}', call_id="c2")
+        mock_msg = _mock_assistant_msg(content="", tool_calls=[tc1, tc2])
+        messages = []
+        slow_started = threading.Event()
+        fast_done = threading.Event()
+        slow_done = threading.Event()
+        release_slow = threading.Event()
+
+        def fake_handle(_name, args, _task_id, **_kwargs):
+            if args["q"] == "slow":
+                slow_started.set()
+                release_slow.wait(5)
+                slow_done.set()
+            else:
+                fast_done.set()
+            return "ok"
+
+        monkeypatch.setenv("HERMES_CONCURRENT_TOOL_TIMEOUT_S", "0.05")
+        with patch("run_agent.handle_function_call", side_effect=fake_handle):
+            agent._execute_tool_calls_concurrent(mock_msg, messages, "task-1")
+
+        assert slow_started.is_set()
+        assert fast_done.is_set()
+        assert not slow_done.is_set()
+        assert agent.get_activity_summary()["current_tool"]
+
+        release_slow.set()
+        deadline = time.monotonic() + 5
+        while not slow_done.is_set() and time.monotonic() < deadline:
+            time.sleep(0.01)
+        assert slow_done.is_set()
+        while agent.get_activity_summary()["current_tool"] and time.monotonic() < deadline:
+            time.sleep(0.01)
+        assert not agent.get_activity_summary()["current_tool"]
+
 
     def test_concurrent_submit_shutdown_error_returns_tool_errors(self, agent):
         """Submit-time interpreter shutdown should not escape the outer loop."""

tests/run_agent/test_tool_activity_heartbeat.py

@@ -242,6 +242,8 @@ def test_concurrent_tool_call_heartbeat(monkeypatch):
     )
     touches: list = []
     agent._touch_activity = lambda desc: touches.append(time.time())
+    started_at: list[float | None] = []
+    monkeypatch.setattr(te, "_begin_tool_execution", lambda *args, **kwargs: None)
 
     agent._execute_tool_calls_concurrent = (
         __import__("run_agent").AIAgent._execute_tool_calls_concurrent.__get__(agent)
@@ -258,6 +260,7 @@ def __init__(self, tool_calls):
             self.tool_calls = tool_calls
 
     def _invoke(name, *a, **kw):
+        started_at.append(agent._current_tool_started_at)
         time.sleep(0.25)
         return json.dumps({"ok": name})
 
@@ -268,3 +271,4 @@ def _invoke(name, *a, **kw):
     agent._execute_tool_calls_concurrent(msg, messages, "task")
 
     assert len(touches) >= 3, f"expected mid-call heartbeats, got {len(touches)}"
+    assert started_at and started_at[0] is not None

website/docs/reference/environment-variables.md

@@ -812,6 +812,7 @@ Advanced per-platform knobs for throttling the outbound message batcher. Most us
 | `HERMES_STREAM_RETRIES` | Number of mid-stream reconnect attempts on transient network errors (default: `3`). |
 | `HERMES_STREAM_STALE_GIVEUP` | Cross-turn circuit breaker: after this many consecutive stale kills (streaming or non-streaming) with no completed response, abort each call immediately with an actionable error instead of re-waiting out the stale timeout (default: `5`, `0` disables). Resets on any completed response, `/model` switch, fallback activation, or turn-start primary restore. |
 | `HERMES_AGENT_TIMEOUT` | Gateway inactivity timeout for a running agent in seconds (default: `1800`, 30 minutes). Resets on every tool call and streamed token. Set to `0` to disable. |
+| `HERMES_CONCURRENT_TOOL_TIMEOUT_S` | Wall-time ceiling for an in-flight tool, including silent tools and concurrent batches (default: `420` seconds). Set to `0` to disable. Healthy silent tools are also stopped when this ceiling expires. |
 | `HERMES_GATEWAY_MAX_STARTS` | Respawn-storm circuit breaker: maximum gateway (re)starts allowed within the window before an exponential backoff is slept to break the storm (default: `5`, `0` disables). Also configurable via `gateway.respawn_storm.max_starts` in `config.yaml`. |
 | `HERMES_GATEWAY_START_WINDOW_S` | Respawn-storm breaker window in seconds (default: `120`). Also configurable via `gateway.respawn_storm.window_seconds` in `config.yaml`. |
 | `HERMES_AGENT_TIMEOUT_WARNING` | Gateway: send a warning message after this many seconds of inactivity (default: 75% of `HERMES_AGENT_TIMEOUT`). |