diff --git a/strix/core/agents.py b/strix/core/agents.py index e4b26989..7438f09d 100644 --- a/strix/core/agents.py +++ b/strix/core/agents.py @@ -55,6 +55,7 @@ class AgentCoordinator: self.idle_resume_counts: dict[str, int] = {} self.wait_kinds: dict[str, WaitKind] = {} self.runtimes: dict[str, AgentRuntime] = {} + self._parent_notified: set[str] = set() self._lock = asyncio.Lock() self._snapshot_path: Path | None = None self.is_shutting_down = False @@ -191,6 +192,7 @@ class AgentCoordinator: self.errors.pop(agent_id, None) self.wait_kinds.pop(agent_id, None) self.runtimes.setdefault(agent_id, AgentRuntime()).user_wake_required = False + self._parent_notified.discard(agent_id) await self._maybe_snapshot() async def park_waiting(self, agent_id: str, *, wait_kind: WaitKind) -> None: @@ -252,12 +254,27 @@ class AgentCoordinator: self.errors[agent_id] = error elif status == "running": self.errors.pop(agent_id, None) + if status == "running": + # Running again means a fresh stint that owes its parent its own notice. + self._parent_notified.discard(agent_id) runtime = self.runtimes.setdefault(agent_id, AgentRuntime()) runtime.user_wake_required = status in {"failed", "crashed"} runtime.wake.set() logger.info("agent.status %s=%s", agent_id, status) await self._maybe_snapshot() + async def claim_parent_notice(self, agent_id: str) -> bool: + """Reserve the one notice a child owes its parent when it stops running. + + A completion report and a terminal notice carry the same information, so + whichever comes first claims the slot and the other is skipped. + """ + async with self._lock: + if agent_id in self._parent_notified: + return False + self._parent_notified.add(agent_id) + return True + async def send( self, target_agent_id: str, message: dict[str, Any], *, interrupt: bool = True ) -> bool: diff --git a/strix/core/execution.py b/strix/core/execution.py index 9822ec82..e833d88e 100644 --- a/strix/core/execution.py +++ b/strix/core/execution.py @@ -539,7 +539,7 @@ async def _exhausted_recovery( """ if not interactive: await coordinator.set_status(agent_id, "crashed") - await _notify_parent_on_terminal(coordinator, agent_id, "crashed") + await notify_parent_on_terminal(coordinator, agent_id, "crashed") raise MaxTurnsExceeded( "Agent exhausted recovery attempts without calling finish_scan or agent_finish." ) @@ -622,7 +622,7 @@ async def _run_cycle_parked( except Exception as exc: logger.exception("error escaped the run cycle for %s; parking as failed", agent_id) await coordinator.set_status(agent_id, "failed", error=str(exc) or type(exc).__name__) - await _notify_parent_on_terminal(coordinator, agent_id, "failed") + await notify_parent_on_terminal(coordinator, agent_id, "failed") return None @@ -778,7 +778,7 @@ async def _run_cycle( # noqa: PLR0912, PLR0915 if isinstance(exc, ProviderRefusalError): logger.warning("agent %s refused by the model provider: %s", agent_id, exc) await coordinator.set_status(agent_id, "failed", error=str(exc)) - await _notify_parent_on_terminal(coordinator, agent_id, "failed") + await notify_parent_on_terminal(coordinator, agent_id, "failed") return None if not interactive: raise @@ -790,7 +790,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_terminal(coordinator, agent_id, status) + await notify_parent_on_terminal(coordinator, agent_id, status) return None else: return cast("RunResultBase | None", stream) @@ -851,6 +851,11 @@ async def _append_tool_required_message( _TERMINAL_NOTICE = { + "completed": ( + "[Agent completed] {name} ({agent_id}) finished and is no longer running, but it " + "sent no completion report. Stop waiting on this child; ask it directly if you " + "need its results." + ), "crashed": ( "[Agent crash] {name} ({agent_id}) terminated unexpectedly. " "Stop waiting on this child unless you want to message it again." @@ -898,7 +903,7 @@ async def _notify_parent_on_stall( ) -async def _notify_parent_on_terminal( +async def notify_parent_on_terminal( coordinator: AgentCoordinator, agent_id: str, status: str, @@ -911,6 +916,8 @@ async def _notify_parent_on_terminal( name = coordinator.names.get(agent_id, agent_id) if parent is None: return + if not await coordinator.claim_parent_notice(agent_id): + return await coordinator.send( parent, { @@ -945,6 +952,21 @@ async def _notify_root_on_budget_reserve(coordinator: AgentCoordinator) -> None: await coordinator.send(root, _reserve_notice()) +async def _notify_parent_on_exit( + coordinator: AgentCoordinator, + agent_id: str, +) -> None: + """Backstop for a child whose loop ended without telling its parent. + + Every terminal state counts, including ``completed``: a child that skips its + completion report leaves the parent waiting on a message nobody will send. + """ + status = await _agent_status(coordinator, agent_id) + if status is None: + return + await notify_parent_on_terminal(coordinator, agent_id, status) + + async def _start_child_runner( *, parent_ctx: dict[str, Any], @@ -998,6 +1020,9 @@ async def _start_child_runner( logger.info("child %s stopped after reaching the scan budget limit", child_id) except SubagentBudgetReservedError: logger.info("child %s stopped at the sub-agent budget reserve", child_id) + finally: + if not coordinator.is_shutting_down: + await _notify_parent_on_exit(coordinator, child_id) task_handle = asyncio.create_task(_child_loop(), name=f"agent-{name}-{child_id}") await coordinator.attach_runtime(child_id, task=task_handle) diff --git a/strix/tools/agents_graph/tools.py b/strix/tools/agents_graph/tools.py index 28bb9e07..33033cf5 100644 --- a/strix/tools/agents_graph/tools.py +++ b/strix/tools/agents_graph/tools.py @@ -13,6 +13,7 @@ from typing import Any, Literal, get_args from agents import RunContextWrapper, function_tool from strix.core.agents import Status, coordinator_from_context +from strix.core.execution import notify_parent_on_terminal from strix.skills import validate_requested_skills @@ -558,7 +559,7 @@ async def agent_finish( ) parent_notified = False - if report_to_parent: + if report_to_parent and await coordinator.claim_parent_notice(me): async with coordinator._lock: agent_name = coordinator.names.get(me, me) report = _render_completion_report( @@ -582,6 +583,11 @@ async def agent_finish( ) parent_notified = True + await coordinator.set_status(me, "completed") + if not parent_notified: + # Silence here would leave a parent waiting on a report that is never coming. + await notify_parent_on_terminal(coordinator, me, "completed") + logger.info( "agent_finish: %s success=%s findings=%d parent_notified=%s", me, @@ -589,7 +595,6 @@ async def agent_finish( len(findings or []), parent_notified, ) - await coordinator.set_status(me, "completed") return json.dumps( { diff --git a/tests/test_execution.py b/tests/test_execution.py index 3d435c0c..8e5b57f6 100644 --- a/tests/test_execution.py +++ b/tests/test_execution.py @@ -18,10 +18,11 @@ from openai.types.responses import ResponseOutputMessage, ResponseOutputRefusal from strix.core import execution from strix.core.agents import AgentCoordinator from strix.core.execution import ( - _notify_parent_on_terminal, _notify_root_on_budget_reserve, + notify_parent_on_terminal, ) from strix.core.sessions import seed_initial_input +from strix.tools.agents_graph.tools import agent_finish from strix.tools.finish.tool import finish_scan @@ -67,6 +68,27 @@ async def _call_finish_scan( return parsed +async def _call_agent_finish( + coordinator: AgentCoordinator, + agent_id: str, + parent_id: str | None, + *, + report_to_parent: bool, +) -> dict[str, Any]: + ctx = ToolContext( + context={"coordinator": coordinator, "agent_id": agent_id, "parent_id": parent_id}, + tool_name="agent_finish", + tool_call_id="call-1", + tool_arguments="{}", + ) + result: str = await agent_finish.on_invoke_tool( + ctx, + json.dumps({"result_summary": "done", "report_to_parent": report_to_parent}), + ) + 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() @@ -466,11 +488,11 @@ async def test_snapshot_round_trip_preserves_budget_pause() -> None: @pytest.mark.asyncio -@pytest.mark.parametrize("status", ["stopped", "failed", "crashed"]) +@pytest.mark.parametrize("status", ["completed", "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. + # Regression for #870 and #947: a child reaching any terminal state - including a + # plain "completed" - must wake the parent parked in wait_for_agents, so the root + # can finalize the scan instead of hanging for a report that never arrives. coordinator = AgentCoordinator() await coordinator.register("root", "strix", parent_id=None) await coordinator.register("child", "SQL Injection", parent_id="root") @@ -482,13 +504,50 @@ async def test_terminal_child_wakes_parked_parent(tmp_path: Any, status: str) -> 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 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_agent_finish_without_report_still_wakes_parent(tmp_path: Any) -> None: + # Regression for #947: a child that completes with report_to_parent=False owes its + # parent a terminal notice, otherwise the parent waits out its full timeout. + 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) + + root_waiter = asyncio.create_task(coordinator.wait_for_message("root")) + await asyncio.sleep(0) + + await _call_agent_finish(coordinator, "child", "root", report_to_parent=False) + + await asyncio.wait_for(root_waiter, timeout=1.0) + assert coordinator.statuses["child"] == "completed" + assert coordinator.pending_counts.get("root", 0) == 1 + session.close() + + +@pytest.mark.asyncio +async def test_agent_finish_report_suppresses_the_terminal_notice(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_agent_finish(coordinator, "child", "root", report_to_parent=True) + # The exit backstop must not duplicate the report the child already delivered. + await execution._notify_parent_on_exit(coordinator, "child") + + assert coordinator.pending_counts.get("root", 0) == 1 + session.close() + + @pytest.mark.asyncio async def test_notify_parent_on_terminal_ignores_non_terminal_status(tmp_path: Any) -> None: coordinator = AgentCoordinator() @@ -497,7 +556,7 @@ async def test_notify_parent_on_terminal_ignores_non_terminal_status(tmp_path: A session = SQLiteSession("root", tmp_path / "agents.db") await coordinator.attach_runtime("root", session=session) - await _notify_parent_on_terminal(coordinator, "child", "waiting") + await notify_parent_on_terminal(coordinator, "child", "waiting") assert coordinator.pending_counts.get("root", 0) == 0 session.close() @@ -523,7 +582,7 @@ async def test_terminal_notice_does_not_cancel_parent_stream(tmp_path: Any) -> N await coordinator.attach_runtime("root", session=session, interrupt_on_message=True) await coordinator.attach_stream("root", stream) - await _notify_parent_on_terminal(coordinator, "child", "crashed") + await notify_parent_on_terminal(coordinator, "child", "crashed") assert stream.cancelled is False assert coordinator.pending_counts.get("root", 0) > 0