diff --git a/strix/config/settings.py b/strix/config/settings.py index 9309ac39..e9e60791 100644 --- a/strix/config/settings.py +++ b/strix/config/settings.py @@ -112,6 +112,10 @@ class RuntimeSettings(BaseSettings): backend: str = Field(default="docker", alias="STRIX_RUNTIME_BACKEND") # Max screenshot/image tool outputs kept live per agent context (0 = none). max_context_images: int = Field(default=3, ge=0, alias="STRIX_MAX_CONTEXT_IMAGES") + # Seconds a running agent may go without emitting a single run event before + # its turn is abandoned and, if it still does not recover, a waiting parent + # marks it failed (0 = never). + agent_stall_timeout: int = Field(default=1800, ge=0, alias="STRIX_AGENT_STALL_TIMEOUT") class TelemetrySettings(BaseSettings): diff --git a/strix/core/agents.py b/strix/core/agents.py index 4d3d65cc..bd5cf86f 100644 --- a/strix/core/agents.py +++ b/strix/core/agents.py @@ -6,6 +6,7 @@ import asyncio import json import logging import tempfile +import time from dataclasses import dataclass, field from pathlib import Path from typing import TYPE_CHECKING, Any, Literal, cast @@ -45,6 +46,8 @@ class AgentRuntime: wake: asyncio.Event = field(default_factory=asyncio.Event) mailbox: list[dict[str, Any]] = field(default_factory=list) user_wake_required: bool = False + # Monotonic time of the agent's last sign of life: a run event or a status change. + last_activity: float = field(default_factory=time.monotonic) class AgentCoordinator: @@ -170,7 +173,7 @@ class AgentCoordinator: "task": task or "", "skills": list(skills or []), } - self.runtimes.setdefault(agent_id, AgentRuntime()) + self.runtimes.setdefault(agent_id, AgentRuntime()).last_activity = time.monotonic() logger.info("agent.register %s (%s) parent=%s", agent_id, name, parent_id or "-") await self._maybe_snapshot() @@ -200,10 +203,49 @@ class AgentCoordinator: self.statuses[agent_id] = "running" self.errors.pop(agent_id, None) self.wait_kinds.pop(agent_id, None) - self.runtimes.setdefault(agent_id, AgentRuntime()).user_wake_required = False + runtime = self.runtimes.setdefault(agent_id, AgentRuntime()) + runtime.user_wake_required = False + runtime.last_activity = time.monotonic() self._parent_notified.discard(agent_id) await self._maybe_snapshot() + def touch(self, agent_id: str) -> None: + """Record a sign of life; called on every run event, so it takes no lock.""" + runtime = self.runtimes.get(agent_id) + if runtime is not None: + runtime.last_activity = time.monotonic() + + async def reap_stalled(self, max_silence: float, *, exclude: str) -> list[dict[str, Any]]: + """Fail every running agent silent for longer than ``max_silence`` seconds. + + A turn wedged somewhere no timeout covers never reaches a terminal status, + so whoever waits on it would wait forever. Cancelling the task ends the + wedged turn; the agent's own loop then delivers the terminal notice. + """ + if max_silence <= 0: + return [] + now = time.monotonic() + stalled: list[tuple[str, asyncio.Task[Any] | None, float]] = [] + async with self._lock: + for aid, status in self.statuses.items(): + if aid == exclude or status != "running": + continue + runtime = self.runtimes.get(aid) + if runtime is None: + continue + silence = now - runtime.last_activity + if silence >= max_silence: + stalled.append((aid, runtime.task, silence)) + reaped: list[dict[str, Any]] = [] + for aid, task, silence in stalled: + error = f"agent produced no event for {silence:.0f}s; marked failed as stalled" + logger.warning("agent %s stalled: %s", aid, error) + await self.set_status(aid, "failed", error=error) + if task is not None and not task.done(): + task.cancel() + reaped.append({"agent_id": aid, "name": self.names.get(aid, aid), "error": error}) + return reaped + async def park_waiting(self, agent_id: str, *, wait_kind: WaitKind) -> None: """Park an agent, recording what it is waiting on so the driver can time it.""" async with self._lock: @@ -268,6 +310,7 @@ class AgentCoordinator: self._parent_notified.discard(agent_id) runtime = self.runtimes.setdefault(agent_id, AgentRuntime()) runtime.user_wake_required = status in {"failed", "crashed"} + runtime.last_activity = time.monotonic() runtime.wake.set() logger.info("agent.status %s=%s", agent_id, status) await self._maybe_snapshot() diff --git a/strix/core/execution.py b/strix/core/execution.py index dfcd39fa..4417492f 100644 --- a/strix/core/execution.py +++ b/strix/core/execution.py @@ -6,7 +6,7 @@ import asyncio import contextlib import logging import uuid -from collections.abc import Callable +from collections.abc import AsyncIterator, Callable from functools import cache from typing import TYPE_CHECKING, Any, cast @@ -19,7 +19,7 @@ from openai import ( APITimeoutError, ) -from strix.config import codex +from strix.config import codex, load_settings from strix.core.hooks import ( BudgetExceededError, BudgetPausedError, @@ -149,6 +149,38 @@ def _transient_model_retry_delay(attempt: int) -> float: return min(delay, _TRANSIENT_MODEL_RETRY_MAX_DELAY_S) +def _agent_stall_timeout() -> float: + return float(load_settings().runtime.agent_stall_timeout) + + +async def _with_stall_guard(stream: Any, timeout: float, agent_id: str) -> AsyncIterator[Any]: + """Yield run events, abandoning the turn if none arrives for ``timeout`` seconds. + + The model-stream idle guard only covers the model call itself. Everything + else the run loop awaits between events (tool transport, session writes, + a re-issued request that never opens) has no bound of its own, so a hang + there would keep the agent "running" forever. + """ + events = stream.stream_events() + if timeout <= 0: + async for event in events: + yield event + return + iterator = events.__aiter__() + while True: + try: + event = await asyncio.wait_for(iterator.__anext__(), timeout) + except StopAsyncIteration: + return + except TimeoutError: + message = f"agent turn produced no event for {timeout:.0f}s" + logger.warning("%s for %s; abandoning the turn", message, agent_id) + with contextlib.suppress(Exception): + stream.cancel(mode="immediate") + raise TimeoutError(message) from None + yield event + + async def _salvage_stream_to_session( session: Session, pre_run_items: list[Any], @@ -658,6 +690,7 @@ async def _run_cycle( # noqa: PLR0912, PLR0915 image_strips = 0 compactions = 0 model_retries = 0 + stall_timeout = _agent_stall_timeout() while True: stream: Any = None pre_run_items: list[Any] = [] @@ -688,7 +721,8 @@ async def _run_cycle( # noqa: PLR0912, PLR0915 await coordinator.attach_stream(agent_id, stream) try: try: - async for event in stream.stream_events(): + async for event in _with_stall_guard(stream, stall_timeout, agent_id): + coordinator.touch(agent_id) if event_sink is not None: try: event_sink(agent_id, event) diff --git a/strix/tools/agents_graph/tools.py b/strix/tools/agents_graph/tools.py index da05bb1e..d7c102ab 100644 --- a/strix/tools/agents_graph/tools.py +++ b/strix/tools/agents_graph/tools.py @@ -12,7 +12,8 @@ from typing import Any, Literal, get_args from agents import RunContextWrapper, function_tool -from strix.core.agents import Status, coordinator_from_context +from strix.config import load_settings +from strix.core.agents import AgentCoordinator, Status, coordinator_from_context from strix.core.execution import notify_parent_on_terminal from strix.core.hooks import LLM_TURN_KEY from strix.report.state import get_global_report_state @@ -291,6 +292,16 @@ _WAIT_DEFAULT_TIMEOUT_S = 300 # tool's own timeout fire first and return a clean result. _WAIT_HARD_CEILING_S = _WAIT_DEFAULT_TIMEOUT_S + 1 _WAITED_TURN_KEY = "waited_llm_turn" +# Headroom past the per-turn stall guard so a wedged agent gets to abandon and +# replay its own turn before a waiting parent gives up on it. +_STALL_REAP_GRACE_S = 300.0 + + +async def _reap_stalled_agents(coordinator: AgentCoordinator, me: str) -> list[dict[str, Any]]: + stall_timeout = float(load_settings().runtime.agent_stall_timeout) + if stall_timeout <= 0: + return [] + return await coordinator.reap_stalled(stall_timeout + _STALL_REAP_GRACE_S, exclude=me) @function_tool(timeout=_WAIT_HARD_CEILING_S) @@ -422,6 +433,8 @@ async def wait_for_agents( # noqa: PLR0911 default=str, ) + stalled = await _reap_stalled_agents(coordinator, me) + # Non-interactive agents cannot be woken once terminal, so with nobody # running or waiting there is no message left to wait for. if not await coordinator.active_agents_except(me): @@ -431,6 +444,7 @@ async def wait_for_agents( # noqa: PLR0911 "success": True, "wait_outcome": "no_active_agents", "reason": reason, + "stalled_agents": stalled, "agents": [ {"agent_id": aid, "name": names.get(aid, aid), "status": status} for aid, status in statuses.items() @@ -452,13 +466,22 @@ async def wait_for_agents( # noqa: PLR0911 await asyncio.wait_for(coordinator.wait_for_message(me), timeout_seconds) except TimeoutError: await coordinator.mark_running(me) + stalled = await _reap_stalled_agents(coordinator, me) return json.dumps( { "success": True, "wait_outcome": "timeout", "timeout_seconds": timeout_seconds, "reason": reason, - "note": "No messages within timeout — continue work or call agent_finish.", + "stalled_agents": stalled, + "note": ( + "No messages within timeout — continue work or call agent_finish." + if not stalled + else "No messages within timeout. The agents in stalled_agents produced " + "no output for too long and were marked failed; their terminal notices " + "are queued for you. Do not wait on them again — continue work, respawn " + "what is still needed, or call agent_finish." + ), }, ensure_ascii=False, default=str, diff --git a/tests/test_agent_stall_guard.py b/tests/test_agent_stall_guard.py new file mode 100644 index 00000000..6952a196 --- /dev/null +++ b/tests/test_agent_stall_guard.py @@ -0,0 +1,240 @@ +"""Tests for the per-agent stall guard. + +The model-stream idle watchdog only covers the model call. A turn can also +wedge in what the run loop awaits *between* events — a re-issued request that +never opens, a tool transport that never answers — and nothing bounded that, +so the agent stayed ``running`` forever and its parent waited on it forever. +Two layers now cover it: the run cycle abandons a turn that emits no event for +the stall timeout, and a parent's ``wait_for_agents`` fails any agent that has +been silent for longer still. +""" + +from __future__ import annotations + +import asyncio +import json +import time +from typing import TYPE_CHECKING, Any, cast + +import pytest +from agents import RunConfig, Runner +from agents.tool_context import ToolContext + +from strix.core import execution +from strix.core.agents import AgentCoordinator +from strix.tools.agents_graph import tools as graph_tools +from strix.tools.agents_graph.tools import wait_for_agents + + +if TYPE_CHECKING: + from collections.abc import Iterator + + +class _HangingStream: + """Emits one event, then never produces another.""" + + def __init__(self) -> None: + self.run_loop_exception: BaseException | None = None + self.cancelled = False + + async def stream_events(self) -> Any: + yield "first" + await asyncio.Event().wait() + + def cancel(self, mode: str = "immediate") -> None: # noqa: ARG002 + self.cancelled = True + + +class _HealthyStream: + def __init__(self) -> None: + self.run_loop_exception: BaseException | None = None + + async def stream_events(self) -> Any: + for i in range(3): + await asyncio.sleep(0.01) + yield f"event-{i}" + + +@pytest.fixture +def _fast_stall(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]: + monkeypatch.setattr(execution, "_agent_stall_timeout", lambda: 0.2) + monkeypatch.setattr(execution, "_TRANSIENT_MODEL_RETRY_BASE_DELAY_S", 0.0) + monkeypatch.setattr(execution, "_TRANSIENT_MODEL_RETRY_MAX_DELAY_S", 0.0) + yield + + +def _serve(monkeypatch: pytest.MonkeyPatch, streams: list[Any]) -> dict[str, int]: + calls = {"n": 0} + + def _fake_run_streamed(*_args: Any, **_kwargs: Any) -> Any: + stream = streams[calls["n"]] + calls["n"] += 1 + return stream + + monkeypatch.setattr(Runner, "run_streamed", _fake_run_streamed) + return calls + + +async def _run_cycle(coordinator: AgentCoordinator) -> Any: + return await execution._run_cycle( + object(), + coordinator, + "root", + input_data="task", + run_config=cast("RunConfig", object()), + context={}, + max_turns=5, + session=None, + interactive=False, + event_sink=None, + hooks=None, + ) + + +@pytest.mark.asyncio +@pytest.mark.usefixtures("_fast_stall") +async def test_hung_turn_is_abandoned_and_replayed(monkeypatch: pytest.MonkeyPatch) -> None: + hung = _HangingStream() + healthy = _HealthyStream() + calls = _serve(monkeypatch, [hung, healthy]) + coordinator = AgentCoordinator() + await coordinator.register("root", "strix", parent_id=None) + + started = time.monotonic() + result = await _run_cycle(coordinator) + + assert result is healthy + assert calls["n"] == 2 + assert hung.cancelled is True + assert time.monotonic() - started < 2 + + +@pytest.mark.asyncio +@pytest.mark.usefixtures("_fast_stall") +async def test_agent_that_never_recovers_is_marked_failed( + monkeypatch: pytest.MonkeyPatch, +) -> None: + _serve( + monkeypatch, + [_HangingStream() for _ in range(execution._MAX_TRANSIENT_MODEL_RETRIES + 1)], + ) + coordinator = AgentCoordinator() + await coordinator.register("root", "strix", parent_id=None) + + with pytest.raises(TimeoutError, match="produced no event"): + await _run_cycle(coordinator) + + assert coordinator.statuses["root"] == "crashed" + + +@pytest.mark.asyncio +async def test_stall_guard_is_off_when_disabled(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(execution, "_agent_stall_timeout", lambda: 0.0) + hung = _HangingStream() + _serve(monkeypatch, [hung]) + coordinator = AgentCoordinator() + await coordinator.register("root", "strix", parent_id=None) + + with pytest.raises(TimeoutError): + await asyncio.wait_for(_run_cycle(coordinator), timeout=0.5) + + assert hung.cancelled is False + + +# --- wait_for_agents reaps silent children ------------------------------------------ + + +async def _call_wait(coordinator: AgentCoordinator, args: dict[str, Any]) -> dict[str, Any]: + ctx = ToolContext( + context={"coordinator": coordinator, "agent_id": "root"}, + tool_name=wait_for_agents.name, + tool_call_id="call-1", + tool_arguments="{}", + ) + raw: str = await wait_for_agents.on_invoke_tool(ctx, json.dumps(args)) + return cast("dict[str, Any]", json.loads(raw)) + + +@pytest.fixture +def _reap_after(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]: + async def _reap(coordinator: AgentCoordinator, me: str) -> list[dict[str, Any]]: + return await coordinator.reap_stalled(0.3, exclude=me) + + monkeypatch.setattr(graph_tools, "_reap_stalled_agents", _reap) + yield + + +@pytest.mark.asyncio +@pytest.mark.usefixtures("_reap_after") +async def test_waiting_parent_fails_a_child_silent_too_long() -> None: + coordinator = AgentCoordinator() + await coordinator.register("root", "strix", parent_id=None) + await coordinator.register("child", "ATO-Chaining", parent_id="root") + await coordinator.attach_runtime("root", resumable=False) + child_task = asyncio.create_task(asyncio.Event().wait()) + await coordinator.attach_runtime("child", task=child_task, resumable=False) + + result = await _call_wait(coordinator, {"timeout_seconds": 1}) + + assert result["wait_outcome"] == "timeout" + assert [a["agent_id"] for a in result["stalled_agents"]] == ["child"] + assert coordinator.statuses["child"] == "failed" + assert "stalled" in coordinator.errors["child"] + await asyncio.sleep(0) + assert child_task.cancelled() + + # Nothing is left to wait on, so the next wait returns at once. + result = await _call_wait(coordinator, {"timeout_seconds": 60}) + assert result["wait_outcome"] == "no_active_agents" + + +@pytest.mark.asyncio +@pytest.mark.usefixtures("_reap_after") +async def test_child_that_keeps_emitting_events_is_left_alone() -> None: + coordinator = AgentCoordinator() + await coordinator.register("root", "strix", parent_id=None) + await coordinator.register("child", "DeepFuzz", parent_id="root") + await coordinator.attach_runtime("root", resumable=False) + await coordinator.attach_runtime("child", resumable=False) + + async def _heartbeat() -> None: + for _ in range(12): + await asyncio.sleep(0.1) + coordinator.touch("child") + + beat = asyncio.create_task(_heartbeat()) + result = await _call_wait(coordinator, {"timeout_seconds": 1}) + await beat + + assert result["wait_outcome"] == "timeout" + assert result["stalled_agents"] == [] + assert coordinator.statuses["child"] == "running" + + +@pytest.mark.asyncio +@pytest.mark.usefixtures("_reap_after") +async def test_waiting_children_are_not_reaped() -> None: + # A child parked in its own wait emits no events by design. + coordinator = AgentCoordinator() + await coordinator.register("root", "strix", parent_id=None) + await coordinator.register("child", "Coordinator", parent_id="root") + await coordinator.attach_runtime("root", resumable=False) + await coordinator.attach_runtime("child", resumable=False) + await coordinator.park_waiting("child", wait_kind="agents") + + result = await _call_wait(coordinator, {"timeout_seconds": 1}) + + assert result["wait_outcome"] == "timeout" + assert result["stalled_agents"] == [] + assert coordinator.statuses["child"] == "waiting" + + +@pytest.mark.asyncio +async def test_reaping_is_off_when_disabled() -> None: + coordinator = AgentCoordinator() + await coordinator.register("root", "strix", parent_id=None) + await coordinator.register("child", "X", parent_id="root") + coordinator.runtimes["child"].last_activity -= 10_000 + + assert await coordinator.reap_stalled(0, exclude="root") == [] + assert coordinator.statuses["child"] == "running"