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/core/agents.py

@@ -215,8 +215,8 @@ class AgentCoordinator:
         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.
+    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
@@ -225,25 +225,32 @@ class AgentCoordinator:
         if max_silence <= 0:
             return []
         now = time.monotonic()
-        stalled: list[tuple[str, asyncio.Task[Any] | None, float]] = []
+        reaped: list[dict[str, Any]] = []
+        tasks: list[asyncio.Task[Any]] = []
         async with self._lock:
-            for aid, status in self.statuses.items():
-                if aid == exclude or status != "running":
-                    continue
+            for aid in self._subtree_order_locked(under):
                 runtime = self.runtimes.get(aid)
-                if runtime is None:
+                if aid == under or runtime is None or self.statuses.get(aid) != "running":
                     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})
+                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:

strix/tools/agents_graph/tools.py

@@ -301,7 +301,7 @@ async def _reap_stalled_agents(coordinator: AgentCoordinator, me: str) -> list[d
     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)
+    return await coordinator.reap_stalled(stall_timeout + _STALL_REAP_GRACE_S, under=me)
 
 
 @function_tool(timeout=_WAIT_HARD_CEILING_S)
@@ -478,9 +478,9 @@ async def wait_for_agents(  # noqa: PLR0911
                     "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."
+                    "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,

tests/test_agent_stall_guard.py

@@ -158,7 +158,7 @@ async def _call_wait(coordinator: AgentCoordinator, args: dict[str, Any]) -> dic
 @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)
+        return await coordinator.reap_stalled(0.3, under=me)
 
     monkeypatch.setattr(graph_tools, "_reap_stalled_agents", _reap)
     yield
@@ -236,5 +236,23 @@ async def test_reaping_is_off_when_disabled() -> 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 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"