diff --git a/agent/tool_executor.py b/agent/tool_executor.py index 3f0d64fbb1382..b5b497a3b7bd6 100644 --- a/agent/tool_executor.py +++ b/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}") diff --git a/gateway/run.py b/gateway/run.py index 9e9adaf5a5921..74153f2e2517d 100644 --- a/gateway/run.py +++ b/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. diff --git a/run_agent.py b/run_agent.py index 6704ad7cbdd8e..c2cb155b0294b 100644 --- a/run_agent.py +++ b/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, diff --git a/tests/gateway/test_abandoned_turn_process_cleanup.py b/tests/gateway/test_abandoned_turn_process_cleanup.py index da0fcdaa0c338..e43facab6b8eb 100644 --- a/tests/gateway/test_abandoned_turn_process_cleanup.py +++ b/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() diff --git a/tests/run_agent/test_run_agent.py b/tests/run_agent/test_run_agent.py index 88c18fa12e578..8e37b7276d997 100644 --- a/tests/run_agent/test_run_agent.py +++ b/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.""" diff --git a/tests/run_agent/test_tool_activity_heartbeat.py b/tests/run_agent/test_tool_activity_heartbeat.py index 6a93f4a05162d..2122da1cefd85 100644 --- a/tests/run_agent/test_tool_activity_heartbeat.py +++ b/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 diff --git a/website/docs/reference/environment-variables.md b/website/docs/reference/environment-variables.md index e35fbcaf0a2f8..2d4903b619e1c 100644 --- a/website/docs/reference/environment-variables.md +++ b/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`). |