From 32d59249a327e04efd32cb55435cd72c7055b6ae Mon Sep 17 00:00:00 2001 From: Enes Deniz <142517728+3nesdeniz@users.noreply.github.com> Date: Sat, 25 Jul 2026 03:22:36 +0300 Subject: [PATCH 1/2] fix(core): wake parent on terminal child status --- strix/core/agents.py | 39 +++++- strix/core/execution.py | 28 ---- strix/tools/agents_graph/tools.py | 9 +- tests/test_agent_terminal_notifications.py | 155 +++++++++++++++++++++ 4 files changed, 201 insertions(+), 30 deletions(-) create mode 100644 tests/test_agent_terminal_notifications.py diff --git a/strix/core/agents.py b/strix/core/agents.py index c08b24d60..b77f57407 100644 --- a/strix/core/agents.py +++ b/strix/core/agents.py @@ -21,6 +21,7 @@ logger = logging.getLogger(__name__) Status = Literal["running", "waiting", "completed", "stopped", "crashed", "failed"] +_TERMINAL_STATUSES: frozenset[str] = frozenset({"completed", "stopped", "crashed", "failed"}) @dataclass(slots=True) @@ -115,11 +116,18 @@ async def park_waiting(self, agent_id: str) -> None: await self.set_status(agent_id, "waiting") async def set_status( - self, agent_id: str, status: Status | str, *, error: str | None = None + self, + agent_id: str, + status: Status | str, + *, + error: str | None = None, + notify_parent: bool = True, ) -> None: + """Set an agent status and notify its parent once on a terminal transition.""" async with self._lock: if agent_id not in self.statuses: return + previous_status = self.statuses[agent_id] self.statuses[agent_id] = status # type: ignore[assignment] if error is not None: self.errors[agent_id] = error @@ -127,7 +135,36 @@ async def set_status( self.errors.pop(agent_id, None) runtime = self.runtimes.setdefault(agent_id, AgentRuntime()) runtime.wake.set() + parent_id = self.parent_of.get(agent_id) + should_notify_parent = ( + notify_parent + and parent_id is not None + and previous_status not in _TERMINAL_STATUSES + and status in _TERMINAL_STATUSES + ) + agent_name = self.names.get(agent_id, agent_id) logger.info("agent.status %s=%s", agent_id, status) + if should_notify_parent and parent_id is not None: + detail = f" Reason: {error}" if error else "" + delivered = await self.send( + parent_id, + { + "from": agent_id, + "type": "terminal_status", + "priority": "high", + "content": ( + f"[Agent status] {agent_name} ({agent_id}) entered terminal " + f"status '{status}'.{detail} Do not keep waiting for a " + "completion message from this child." + ), + }, + ) + if not delivered: + logger.warning( + "agent.status could not notify parent=%s of terminal child=%s", + parent_id, + agent_id, + ) await self._maybe_snapshot() async def send(self, target_agent_id: str, message: dict[str, Any]) -> bool: diff --git a/strix/core/execution.py b/strix/core/execution.py index f7cf3759d..874680c80 100644 --- a/strix/core/execution.py +++ b/strix/core/execution.py @@ -321,7 +321,6 @@ async def _run_noninteractive_until_lifecycle( if invalid_final_outputs >= invalid_final_output_limit: await coordinator.set_status(agent_id, "crashed") - await _notify_parent_on_crash(coordinator, agent_id, "crashed") raise MaxTurnsExceeded( "Agent exhausted non-interactive recovery attempts without calling " "finish_scan or agent_finish." @@ -438,7 +437,6 @@ async def _run_cycle( # noqa: PLR0912, PLR0915 status = "crashed" logger.exception("agent run failed for %s; parking as %s", agent_id, status) await coordinator.set_status(agent_id, status, error=str(exc) or type(exc).__name__) - await _notify_parent_on_crash(coordinator, agent_id, status) return None else: await _settle_run_result(coordinator, agent_id, interactive) @@ -502,32 +500,6 @@ async def _append_noninteractive_tool_required_message( return [] -async def _notify_parent_on_crash( - coordinator: AgentCoordinator, - agent_id: str, - status: str, -) -> None: - if status != "crashed": - return - async with coordinator._lock: - parent = coordinator.parent_of.get(agent_id) - name = coordinator.names.get(agent_id, agent_id) - if parent is None: - return - await coordinator.send( - parent, - { - "from": agent_id, - "type": "crash", - "priority": "high", - "content": ( - f"[Agent crash] {name} ({agent_id}) terminated unexpectedly. " - "Stop waiting on this child unless you want to message it again." - ), - }, - ) - - async def _start_child_runner( *, parent_ctx: dict[str, Any], diff --git a/strix/tools/agents_graph/tools.py b/strix/tools/agents_graph/tools.py index 7f9374000..51a9bb2e1 100644 --- a/strix/tools/agents_graph/tools.py +++ b/strix/tools/agents_graph/tools.py @@ -572,7 +572,14 @@ async def agent_finish( len(findings or []), parent_notified, ) - await coordinator.set_status(me, "completed") + # The completion report above is normally the parent notification. If that + # delivery failed, let the coordinator make one generic fallback attempt; + # an explicit report_to_parent=False still suppresses all parent messaging. + await coordinator.set_status( + me, + "completed", + notify_parent=report_to_parent and not parent_notified, + ) return json.dumps( { diff --git a/tests/test_agent_terminal_notifications.py b/tests/test_agent_terminal_notifications.py new file mode 100644 index 000000000..5c9a55f86 --- /dev/null +++ b/tests/test_agent_terminal_notifications.py @@ -0,0 +1,155 @@ +"""Regression tests for terminal child-agent notifications.""" + +from __future__ import annotations + +import asyncio +import json +from typing import TYPE_CHECKING, Any, cast + +import pytest +from agents.exceptions import MaxTurnsExceeded +from agents.tool_context import ToolContext + +from strix.core.agents import AgentCoordinator, Status +from strix.core.execution import _run_cycle # pyright: ignore[reportPrivateUsage] +from strix.tools.agents_graph.tools import agent_finish + + +if TYPE_CHECKING: + from collections.abc import AsyncIterator + + from agents import RunConfig + from agents.memory import Session + + +class _RecordingSession: + def __init__(self) -> None: + self.items: list[Any] = [] + + async def add_items(self, items: list[Any]) -> None: + self.items.extend(items) + + async def get_items(self, limit: int | None = None) -> list[Any]: + return self.items[-limit:] if limit is not None else list(self.items) + + +class _MaxTurnsStream: + run_loop_exception = MaxTurnsExceeded("Max turns (500) exceeded") + + def __init__(self) -> None: + self.events: list[Any] = [] + + async def stream_events(self) -> AsyncIterator[Any]: + for event in self.events: + yield event + + +async def _coordinator_with_child() -> tuple[AgentCoordinator, _RecordingSession]: + coordinator = AgentCoordinator() + parent_session = _RecordingSession() + await coordinator.register("parent", "coordinator", parent_id=None) + await coordinator.attach_runtime( + "parent", + session=cast("Session", parent_session), + ) + await coordinator.register("child", "sql-injection", parent_id="parent") + return coordinator, parent_session + + +@pytest.mark.asyncio +@pytest.mark.parametrize("status", ["completed", "stopped", "failed", "crashed"]) +async def test_terminal_child_status_notifies_waiting_parent(status: Status) -> None: + coordinator, parent_session = await _coordinator_with_child() + waiter = asyncio.create_task(coordinator.wait_for_message("parent")) + await asyncio.sleep(0) + + await coordinator.set_status("child", status, error="child terminated") + await coordinator.set_status("child", status, error="duplicate transition") + + await asyncio.wait_for(waiter, timeout=1.0) + pending, items = await coordinator.consume_pending("parent", include_items=True) + assert pending == 1 + assert items == parent_session.items + assert f"entered terminal status '{status}'" in str(items[0]) + assert "child terminated" in str(items[0]) + + +@pytest.mark.asyncio +async def test_terminal_child_notification_can_be_suppressed() -> None: + coordinator, parent_session = await _coordinator_with_child() + + await coordinator.set_status("child", "completed", notify_parent=False) + + pending, items = await coordinator.consume_pending("parent", include_items=True) + assert pending == 0 + assert items == [] + assert parent_session.items == [] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("report_to_parent, expected_messages", [(True, 1), (False, 0)]) +async def test_agent_finish_avoids_duplicate_terminal_notification( + report_to_parent: bool, + expected_messages: int, +) -> None: + coordinator, parent_session = await _coordinator_with_child() + tool_arguments = json.dumps( + { + "result_summary": "Testing finished", + "report_to_parent": report_to_parent, + } + ) + ctx = ToolContext( + context={ + "coordinator": coordinator, + "agent_id": "child", + "parent_id": "parent", + "task": "Test SQL injection", + }, + tool_name="agent_finish", + tool_call_id="call-agent-finish", + tool_arguments=tool_arguments, + ) + + raw_result = await agent_finish.on_invoke_tool(ctx, tool_arguments) + + assert json.loads(raw_result)["agent_completed"] is True + pending, _ = await coordinator.consume_pending("parent", include_items=True) + assert pending == expected_messages + assert len(parent_session.items) == expected_messages + assert all("entered terminal status" not in str(item) for item in parent_session.items) + + +@pytest.mark.asyncio +async def test_max_turns_exceeded_child_wakes_parent( + monkeypatch: pytest.MonkeyPatch, +) -> None: + coordinator, _ = await _coordinator_with_child() + waiter = asyncio.create_task(coordinator.wait_for_message("parent")) + await asyncio.sleep(0) + + def run_streamed(*_args: Any, **_kwargs: Any) -> _MaxTurnsStream: + return _MaxTurnsStream() + + monkeypatch.setattr("strix.core.execution.Runner.run_streamed", run_streamed) + + result = await _run_cycle( + agent=object(), + coordinator=coordinator, + agent_id="child", + input_data=[], + run_config=cast("RunConfig", object()), + context={}, + max_turns=500, + session=None, + interactive=True, + event_sink=None, + hooks=None, + ) + + assert result is None + assert coordinator.statuses["child"] == "stopped" + await asyncio.wait_for(waiter, timeout=1.0) + pending, items = await coordinator.consume_pending("parent", include_items=True) + assert pending == 1 + assert "Max turns (500) exceeded" in str(items[0]) From e6709de370553e92457508b5cc45eb16f3782868 Mon Sep 17 00:00:00 2001 From: Enes Deniz <142517728+3nesdeniz@users.noreply.github.com> Date: Sat, 25 Jul 2026 03:43:16 +0300 Subject: [PATCH 2/2] fix(core): preserve terminal notification semantics --- strix/core/agents.py | 20 ++- strix/tools/agents_graph/tools.py | 16 ++- tests/test_agent_terminal_notifications.py | 159 ++++++++++++++++++++- 3 files changed, 183 insertions(+), 12 deletions(-) diff --git a/strix/core/agents.py b/strix/core/agents.py index b77f57407..04f6122e9 100644 --- a/strix/core/agents.py +++ b/strix/core/agents.py @@ -227,17 +227,15 @@ async def consume_pending( items = await session.get_items() return count, list(items[-count:]) - async def request_stop(self, agent_id: str) -> None: + async def request_stop(self, agent_id: str, *, notify_parent: bool = True) -> None: async with self._lock: if agent_id not in self.statuses: return - self.statuses[agent_id] = "stopped" runtime = self.runtimes.setdefault(agent_id, AgentRuntime()) - runtime.wake.set() stream = runtime.stream + await self.set_status(agent_id, "stopped", notify_parent=notify_parent) if stream is not None: stream.cancel(mode="after_turn") - await self._maybe_snapshot() async def cancel_descendants(self, agent_id: str) -> None: tasks = [] @@ -251,11 +249,21 @@ async def cancel_descendants(self, agent_id: str) -> None: 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, + *, + notify_parent: bool = True, + ) -> None: async with self._lock: order = self._subtree_order_locked(agent_id) for aid in reversed(order): - await self.request_stop(aid) + # Descendant parents are part of the same stop cascade, so only the + # top-level target's parent needs a terminal notification. + await self.request_stop( + aid, + notify_parent=notify_parent and aid == agent_id, + ) await self._maybe_snapshot() async def attach_stream( diff --git a/strix/tools/agents_graph/tools.py b/strix/tools/agents_graph/tools.py index 51a9bb2e1..635acbdf7 100644 --- a/strix/tools/agents_graph/tools.py +++ b/strix/tools/agents_graph/tools.py @@ -553,7 +553,7 @@ async def agent_finish( findings=list(findings or []), recommendations=list(final_recommendations or []), ) - await coordinator.send( + parent_notified = await coordinator.send( parent_id, { "id": f"report_{uuid.uuid4().hex[:8]}", @@ -563,7 +563,6 @@ async def agent_finish( "priority": "high", }, ) - parent_notified = True logger.info( "agent_finish: %s success=%s findings=%d parent_notified=%s", @@ -642,7 +641,7 @@ async def stop_agent( ensure_ascii=False, default=str, ) - _, statuses, _, _ = await coordinator.graph_snapshot() + parent_of, statuses, _, _ = await coordinator.graph_snapshot() if target_agent_id not in statuses: return json.dumps( {"success": False, "error": f"Unknown agent_id: {target_agent_id}"}, @@ -669,10 +668,17 @@ async def stop_agent( default=str, ) + notify_target_parent = parent_of.get(target_agent_id) != me if cascade: - await coordinator.cancel_descendants_graceful(target_agent_id) + await coordinator.cancel_descendants_graceful( + target_agent_id, + notify_parent=notify_target_parent, + ) else: - await coordinator.request_stop(target_agent_id) + await coordinator.request_stop( + target_agent_id, + notify_parent=notify_target_parent, + ) logger.info( "stop_agent: target=%s cascade=%s reason=%r", diff --git a/tests/test_agent_terminal_notifications.py b/tests/test_agent_terminal_notifications.py index 5c9a55f86..bdcde7f07 100644 --- a/tests/test_agent_terminal_notifications.py +++ b/tests/test_agent_terminal_notifications.py @@ -5,6 +5,7 @@ import asyncio import json from typing import TYPE_CHECKING, Any, cast +from unittest.mock import AsyncMock import pytest from agents.exceptions import MaxTurnsExceeded @@ -12,7 +13,7 @@ from strix.core.agents import AgentCoordinator, Status from strix.core.execution import _run_cycle # pyright: ignore[reportPrivateUsage] -from strix.tools.agents_graph.tools import agent_finish +from strix.tools.agents_graph.tools import agent_finish, stop_agent if TYPE_CHECKING: @@ -86,6 +87,127 @@ async def test_terminal_child_notification_can_be_suppressed() -> None: assert parent_session.items == [] +@pytest.mark.asyncio +async def test_request_stop_notifies_waiting_parent() -> None: + coordinator, _ = await _coordinator_with_child() + waiter = asyncio.create_task(coordinator.wait_for_message("parent")) + await asyncio.sleep(0) + + await coordinator.request_stop("child") + + await asyncio.wait_for(waiter, timeout=1.0) + pending, items = await coordinator.consume_pending("parent", include_items=True) + assert pending == 1 + assert "entered terminal status 'stopped'" in str(items[0]) + + +@pytest.mark.asyncio +async def test_request_stop_can_suppress_parent_notification() -> None: + coordinator, parent_session = await _coordinator_with_child() + + await coordinator.request_stop("child", notify_parent=False) + + pending, items = await coordinator.consume_pending("parent", include_items=True) + assert pending == 0 + assert items == [] + assert parent_session.items == [] + + +@pytest.mark.asyncio +async def test_stop_agent_suppresses_notification_to_direct_parent() -> None: + coordinator, parent_session = await _coordinator_with_child() + tool_arguments = json.dumps( + { + "target_agent_id": "child", + "cascade": False, + } + ) + ctx = ToolContext( + context={ + "coordinator": coordinator, + "agent_id": "parent", + "parent_id": None, + }, + tool_name="stop_agent", + tool_call_id="call-stop-agent", + tool_arguments=tool_arguments, + ) + + raw_result = await stop_agent.on_invoke_tool(ctx, tool_arguments) + + assert json.loads(raw_result)["success"] is True + assert coordinator.statuses["child"] == "stopped" + pending, items = await coordinator.consume_pending("parent", include_items=True) + assert pending == 0 + assert items == [] + assert parent_session.items == [] + + +@pytest.mark.asyncio +async def test_stop_agent_notifies_target_actual_parent() -> None: + coordinator, _ = await _coordinator_with_child() + child_session = _RecordingSession() + await coordinator.attach_runtime( + "child", + session=cast("Session", child_session), + ) + await coordinator.register("grandchild", "xss", parent_id="child") + tool_arguments = json.dumps( + { + "target_agent_id": "grandchild", + "cascade": False, + } + ) + ctx = ToolContext( + context={ + "coordinator": coordinator, + "agent_id": "parent", + "parent_id": None, + }, + tool_name="stop_agent", + tool_call_id="call-stop-grandchild", + tool_arguments=tool_arguments, + ) + + raw_result = await stop_agent.on_invoke_tool(ctx, tool_arguments) + + assert json.loads(raw_result)["success"] is True + assert coordinator.statuses["grandchild"] == "stopped" + pending, items = await coordinator.consume_pending("child", include_items=True) + assert pending == 1 + assert "entered terminal status 'stopped'" in str(items[0]) + + +@pytest.mark.asyncio +async def test_cascade_stop_only_notifies_top_target_parent() -> None: + coordinator, parent_session = await _coordinator_with_child() + child_session = _RecordingSession() + await coordinator.attach_runtime( + "child", + session=cast("Session", child_session), + ) + await coordinator.register("grandchild", "xss", parent_id="child") + + await coordinator.cancel_descendants_graceful("child") + + assert coordinator.statuses["child"] == "stopped" + assert coordinator.statuses["grandchild"] == "stopped" + parent_pending, parent_items = await coordinator.consume_pending( + "parent", + include_items=True, + ) + child_pending, child_items = await coordinator.consume_pending( + "child", + include_items=True, + ) + assert parent_pending == 1 + assert "child" in str(parent_items[0]) + assert child_pending == 0 + assert child_items == [] + assert child_session.items == [] + assert len(parent_session.items) == 1 + + @pytest.mark.asyncio @pytest.mark.parametrize("report_to_parent, expected_messages", [(True, 1), (False, 0)]) async def test_agent_finish_avoids_duplicate_terminal_notification( @@ -120,6 +242,41 @@ async def test_agent_finish_avoids_duplicate_terminal_notification( assert all("entered terminal status" not in str(item) for item in parent_session.items) +@pytest.mark.asyncio +async def test_agent_finish_falls_back_when_completion_delivery_fails( + monkeypatch: pytest.MonkeyPatch, +) -> None: + coordinator, _ = await _coordinator_with_child() + send = AsyncMock(side_effect=[False, True]) + monkeypatch.setattr(coordinator, "send", send) + tool_arguments = json.dumps( + { + "result_summary": "Testing finished", + "report_to_parent": True, + } + ) + ctx = ToolContext( + context={ + "coordinator": coordinator, + "agent_id": "child", + "parent_id": "parent", + "task": "Test SQL injection", + }, + tool_name="agent_finish", + tool_call_id="call-agent-finish", + tool_arguments=tool_arguments, + ) + + raw_result = await agent_finish.on_invoke_tool(ctx, tool_arguments) + + assert json.loads(raw_result)["parent_notified"] is False + assert send.await_count == 2 + first_message = send.await_args_list[0].args[1] + fallback_message = send.await_args_list[1].args[1] + assert first_message["type"] == "completion" + assert fallback_message["type"] == "terminal_status" + + @pytest.mark.asyncio async def test_max_turns_exceeded_child_wakes_parent( monkeypatch: pytest.MonkeyPatch,