fix(core): scope stall reaping to the caller's subtree and commit failures atomically

Reap only descendants of the waiting agent, re-check status and silence
under the coordinator lock at the moment the failure is written, and stop
claiming the terminal notice is already queued in the wait result.

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

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

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,56 @@ 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, *, under: str) -> list[dict[str, Any]]:
+        """Fail the descendants of ``under`` that were silent for ``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()
+        reaped: list[dict[str, Any]] = []
+        tasks: list[asyncio.Task[Any]] = []
+        async with self._lock:
+            for aid in self._subtree_order_locked(under):
+                runtime = self.runtimes.get(aid)
+                if aid == under or runtime is None or self.statuses.get(aid) != "running":
+                    continue
+                silence = now - runtime.last_activity
+                if silence < max_silence:
+                    continue
+                error = f"agent produced no event for {silence:.0f}s; marked failed as stalled"
+                self.statuses[aid] = "failed"
+                self.errors[aid] = error
+                runtime.user_wake_required = True
+                runtime.last_activity = now
+                runtime.wake.set()
+                if runtime.task is not None and not runtime.task.done():
+                    tasks.append(runtime.task)
+                reaped.append({"agent_id": aid, "name": self.names.get(aid, aid), "error": error})
+        for entry in reaped:
+            logger.warning("agent %s stalled: %s", entry["agent_id"], entry["error"])
+            logger.info("agent.status %s=failed", entry["agent_id"])
+        for task in tasks:
+            task.cancel()
+        if reaped:
+            await self._maybe_snapshot()
+        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 +317,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()

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)

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, under=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; treat this list as "
+                    "their failure notice. Do not wait on them again — continue work, "
+                    "respawn what is still needed, or call agent_finish."
+                ),
             },
             ensure_ascii=False,
             default=str,

tests/test_agent_stall_guard.py

@@ -0,0 +1,258 @@
+"""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, under=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, under="root") == []
+    assert coordinator.statuses["child"] == "running"
+
+
+@pytest.mark.asyncio
+async def test_reaping_stays_inside_the_callers_subtree() -> None:
+    coordinator = AgentCoordinator()
+    await coordinator.register("root", "strix", parent_id=None)
+    await coordinator.register("a", "A", parent_id="root")
+    await coordinator.register("a1", "A1", parent_id="a")
+    await coordinator.register("b", "B", parent_id="root")
+    for aid in ("root", "a1", "b"):
+        coordinator.runtimes[aid].last_activity -= 10_000
+
+    reaped = await coordinator.reap_stalled(1, under="a")
+
+    assert [r["agent_id"] for r in reaped] == ["a1"]
+    assert coordinator.statuses["a1"] == "failed"
+    assert coordinator.statuses["b"] == "running"
+    assert coordinator.statuses["root"] == "running"