fix(tools): tell a waiting parent when stop_agent stops its child

strix/core/agents.py

@@ -383,12 +383,15 @@ class AgentCoordinator:
         if tasks:
             await asyncio.gather(*tasks, return_exceptions=True)
 
-    async def cancel_descendants_graceful(self, agent_id: str) -> None:
+    async def cancel_descendants_graceful(self, agent_id: str) -> list[str]:
+        """Stop a subtree leaves-first and report which agents were stopped."""
         async with self._lock:
             order = self._subtree_order_locked(agent_id)
-        for aid in reversed(order):
+        stopped = list(reversed(order))
+        for aid in stopped:
             await self.request_stop(aid)
         await self._maybe_snapshot()
+        return stopped
 
     async def attach_stream(
         self,

strix/core/execution.py

@@ -866,9 +866,9 @@ _TERMINAL_NOTICE = {
         "message it again."
     ),
     "stopped": (
-        "[Agent capped] {name} ({agent_id}) hit its turn limit and was stopped "
-        "before finishing. It will not send a completion report, so stop waiting "
-        "on this child; account for its capped subtask and continue."
+        "[Agent stopped] {name} ({agent_id}) was stopped before finishing (turn limit "
+        "or an explicit stop). It will not send a completion report, so stop waiting "
+        "on this child; account for its unfinished subtask and continue."
     ),
 }
 

strix/tools/agents_graph/tools.py

@@ -685,9 +685,16 @@ async def stop_agent(
         )
 
     if cascade:
-        await coordinator.cancel_descendants_graceful(target_agent_id)
+        stopped = await coordinator.cancel_descendants_graceful(target_agent_id)
     else:
         await coordinator.request_stop(target_agent_id)
+        stopped = [target_agent_id]
+
+    # The stopper knows what it just did; anyone else waiting on those agents does not.
+    async with coordinator._lock:
+        orphaned = [aid for aid in stopped if coordinator.parent_of.get(aid) not in (None, me)]
+    for aid in orphaned:
+        await notify_parent_on_terminal(coordinator, aid, "stopped")
 
     logger.info(
         "stop_agent: target=%s cascade=%s reason=%r",

tests/test_execution.py

@@ -22,7 +22,7 @@ from strix.core.execution import (
     notify_parent_on_terminal,
 )
 from strix.core.sessions import seed_initial_input
-from strix.tools.agents_graph.tools import agent_finish
+from strix.tools.agents_graph.tools import agent_finish, stop_agent
 from strix.tools.finish.tool import finish_scan
 
 
@@ -89,6 +89,22 @@ async def _call_agent_finish(
     return parsed
 
 
+async def _call_stop_agent(
+    coordinator: AgentCoordinator, agent_id: str, target_agent_id: str
+) -> dict[str, Any]:
+    ctx = ToolContext(
+        context={"coordinator": coordinator, "agent_id": agent_id},
+        tool_name="stop_agent",
+        tool_call_id="call-1",
+        tool_arguments="{}",
+    )
+    result: str = await stop_agent.on_invoke_tool(
+        ctx, json.dumps({"target_agent_id": target_agent_id})
+    )
+    parsed: dict[str, Any] = json.loads(result)
+    return parsed
+
+
 @pytest.mark.asyncio
 async def test_reserve_stop_notifies_root_once(monkeypatch: pytest.MonkeyPatch) -> None:
     coordinator = AgentCoordinator()
@@ -548,6 +564,38 @@ async def test_agent_finish_report_suppresses_the_terminal_notice(tmp_path: Any)
     session.close()
 
 
+@pytest.mark.asyncio
+async def test_stop_agent_notifies_a_parent_that_is_not_the_stopper(tmp_path: Any) -> None:
+    coordinator = AgentCoordinator()
+    await coordinator.register("root", "strix", parent_id=None)
+    await coordinator.register("child", "recon", parent_id="root")
+    await coordinator.register("grandchild", "sqli", parent_id="child")
+    session = SQLiteSession("child", tmp_path / "agents.db")
+    await coordinator.attach_runtime("child", session=session)
+
+    await _call_stop_agent(coordinator, "root", "grandchild")
+
+    assert coordinator.statuses["grandchild"] == "stopped"
+    assert coordinator.pending_counts.get("child", 0) == 1
+    # The stopper already knows; only the waiting parent needs telling.
+    assert coordinator.pending_counts.get("root", 0) == 0
+    session.close()
+
+
+@pytest.mark.asyncio
+async def test_stop_agent_does_not_notify_the_stopping_parent(tmp_path: Any) -> None:
+    coordinator = AgentCoordinator()
+    await coordinator.register("root", "strix", parent_id=None)
+    await coordinator.register("child", "recon", parent_id="root")
+    session = SQLiteSession("root", tmp_path / "agents.db")
+    await coordinator.attach_runtime("root", session=session)
+
+    await _call_stop_agent(coordinator, "root", "child")
+
+    assert coordinator.pending_counts.get("root", 0) == 0
+    session.close()
+
+
 @pytest.mark.asyncio
 async def test_notify_parent_on_terminal_ignores_non_terminal_status(tmp_path: Any) -> None:
     coordinator = AgentCoordinator()