From 082d4ae62c256a05046d109ddfe1d6900f803079 Mon Sep 17 00:00:00 2001 From: Ahmed Allam Date: Mon, 27 Jul 2026 11:16:58 +0000 Subject: [PATCH] fix(runtime): wake parent when child hits a terminal state (MaxTurnsExceeded) --- strix/core/execution.py | 34 +++++++++++++++++++++++++--------- tests/test_execution.py | 40 +++++++++++++++++++++++++++++++++++++++- 2 files changed, 64 insertions(+), 10 deletions(-) diff --git a/strix/core/execution.py b/strix/core/execution.py index 4efd5f9d..7c251777 100644 --- a/strix/core/execution.py +++ b/strix/core/execution.py @@ -417,7 +417,7 @@ 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") + await _notify_parent_on_terminal(coordinator, agent_id, "crashed") raise MaxTurnsExceeded( "Agent exhausted non-interactive recovery attempts without calling " "finish_scan or agent_finish." @@ -582,7 +582,7 @@ 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) + await _notify_parent_on_terminal(coordinator, agent_id, status) return None else: await _settle_run_result(coordinator, agent_id, interactive) @@ -646,12 +646,31 @@ async def _append_noninteractive_tool_required_message( return [] -async def _notify_parent_on_crash( +_TERMINAL_NOTICE = { + "crashed": ( + "[Agent crash] {name} ({agent_id}) terminated unexpectedly. " + "Stop waiting on this child unless you want to message it again." + ), + "failed": ( + "[Agent failed] {name} ({agent_id}) stopped with an error and will not " + "send a completion report. Stop waiting on this child unless you want to " + "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." + ), +} + + +async def _notify_parent_on_terminal( coordinator: AgentCoordinator, agent_id: str, status: str, ) -> None: - if status != "crashed": + template = _TERMINAL_NOTICE.get(status) + if template is None: return async with coordinator._lock: parent = coordinator.parent_of.get(agent_id) @@ -662,12 +681,9 @@ async def _notify_parent_on_crash( parent, { "from": agent_id, - "type": "crash", + "type": status, "priority": "high", - "content": ( - f"[Agent crash] {name} ({agent_id}) terminated unexpectedly. " - "Stop waiting on this child unless you want to message it again." - ), + "content": template.format(name=name, agent_id=agent_id), }, ) diff --git a/tests/test_execution.py b/tests/test_execution.py index 0350e470..985541c0 100644 --- a/tests/test_execution.py +++ b/tests/test_execution.py @@ -12,7 +12,7 @@ from agents.memory import SQLiteSession from agents.tool_context import ToolContext from strix.core.agents import AgentCoordinator -from strix.core.execution import _notify_root_on_budget_reserve +from strix.core.execution import _notify_parent_on_terminal, _notify_root_on_budget_reserve from strix.tools.finish.tool import finish_scan @@ -427,3 +427,41 @@ async def test_snapshot_round_trip_preserves_budget_pause() -> None: await restored.restore(snap) assert restored.budget_paused is True assert restored.statuses["root"] == "budget_paused" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("status", ["stopped", "failed", "crashed"]) +async def test_terminal_child_wakes_parked_parent(tmp_path: Any, status: str) -> None: + # Regression for #870: a child reaching a terminal state (e.g. MaxTurnsExceeded + # -> "stopped") must wake the parent parked in wait_for_message, so the root can + # finalize the scan instead of hanging for a completion report that never arrives. + coordinator = AgentCoordinator() + await coordinator.register("root", "strix", parent_id=None) + await coordinator.register("child", "SQL Injection", parent_id="root") + session = SQLiteSession("root", tmp_path / "agents.db") + await coordinator.attach_runtime("root", session=session) + + root_waiter = asyncio.create_task(coordinator.wait_for_message("root")) + await asyncio.sleep(0) + assert not root_waiter.done() + + await coordinator.set_status("child", status, error="Max turns (500) exceeded") + await _notify_parent_on_terminal(coordinator, "child", status) + + await asyncio.wait_for(root_waiter, timeout=1.0) + 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() + 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 _notify_parent_on_terminal(coordinator, "child", "waiting") + + assert coordinator.pending_counts.get("root", 0) == 0 + session.close()