From 1edafd3e802f9a866eb26aa03fa35e7e3b2d65db Mon Sep 17 00:00:00 2001 From: Ahmed Allam Date: Wed, 2 Sep 2026 18:34:21 +0000 Subject: [PATCH] fix(agents): stop parents waiting on finished non-interactive children A non-interactive agent's loop returns after its terminal state, yet send_message_to_agent kept reporting messages to it as delivered and the parent then waited out wait_for_agents on a reply that could never come. - AgentRuntime.resumable records whether the loop parks for wake-ups after a terminal state; run_agent_loop / _start_child_runner set it from interactive. - AgentCoordinator.send returns False (nothing queued) for a terminal agent that is not resumable; send_message_to_agent surfaces target_status and delivery_status=not_delivered with a pointer to list_reports / get_report. - wait_for_agents returns wait_outcome=no_active_agents at once when no other agent is running or waiting in a non-interactive run. - agent_finish reads the reports the finishing agent filed from the report state and puts their ids in the completion report, the parent message (filed_report_ids) and its own return payload, so parents no longer have to infer what was filed from prose. --- strix/core/agents.py | 37 +++- strix/core/execution.py | 3 +- strix/tools/agents_graph/tools.py | 96 ++++++++- tests/test_agent_graph_coordination.py | 258 +++++++++++++++++++++++++ tests/test_wait_dedupe.py | 2 + 5 files changed, 391 insertions(+), 5 deletions(-) create mode 100644 tests/test_agent_graph_coordination.py diff --git a/strix/core/agents.py b/strix/core/agents.py index edd7863b..4d3d65cc 100644 --- a/strix/core/agents.py +++ b/strix/core/agents.py @@ -24,6 +24,8 @@ logger = logging.getLogger(__name__) Status = Literal["running", "waiting", "completed", "stopped", "crashed", "failed", "budget_paused"] +TERMINAL_STATUSES: frozenset[str] = frozenset({"completed", "stopped", "crashed", "failed"}) + # Why an agent parked. The user can message any agent, so this - not the agent's # position in the tree - decides whether waiting is bounded: only an agent waiting # on other agents is re-checked on a timer. @@ -36,6 +38,10 @@ class AgentRuntime: task: asyncio.Task[Any] | None = None stream: Any | None = None interrupt_on_message: bool = False + # Whether the agent's loop parks after a terminal state and can be woken by a + # later message. A non-interactive loop returns instead, so once such an + # agent is terminal nothing will ever read its mailbox again. + resumable: bool = True wake: asyncio.Event = field(default_factory=asyncio.Event) mailbox: list[dict[str, Any]] = field(default_factory=list) user_wake_required: bool = False @@ -175,6 +181,7 @@ class AgentCoordinator: session: Session | None = None, task: asyncio.Task[Any] | None = None, interrupt_on_message: bool | None = None, + resumable: bool | None = None, ) -> None: async with self._lock: runtime = self.runtimes.setdefault(agent_id, AgentRuntime()) @@ -184,6 +191,8 @@ class AgentCoordinator: runtime.task = task if interrupt_on_message is not None: runtime.interrupt_on_message = interrupt_on_message + if resumable is not None: + runtime.resumable = resumable async def mark_running(self, agent_id: str) -> None: async with self._lock: @@ -275,10 +284,29 @@ class AgentCoordinator: self._parent_notified.add(agent_id) return True + def _unreachable_locked(self, agent_id: str) -> bool: + """True when the agent is terminal and no loop will ever read its mailbox.""" + if self.statuses.get(agent_id) not in TERMINAL_STATUSES: + return False + runtime = self.runtimes.get(agent_id) + return runtime is not None and not runtime.resumable + + async def reachability(self, agent_id: str) -> tuple[bool, Status | None]: + """Whether a message to ``agent_id`` can still be acted on, plus its status.""" + async with self._lock: + status = self.statuses.get(agent_id) + if status is None: + return False, None + return not self._unreachable_locked(agent_id), status + async def send( self, target_agent_id: str, message: dict[str, Any], *, interrupt: bool = True ) -> bool: - """Queue a user/peer message in the target's mailbox and wake it.""" + """Queue a user/peer message in the target's mailbox and wake it. + + Returns False when nothing will ever read the message: the target is + unknown, or it is terminal and its loop does not park for wake-ups. + """ from_user = message.get("from") == "user" if from_user and self._budget_paused: await self.resume_from_budget_pause(exclude=target_agent_id) @@ -286,6 +314,13 @@ class AgentCoordinator: if target_agent_id not in self.statuses: logger.debug("agent.send dropped unknown target=%s", target_agent_id) return False + if self._unreachable_locked(target_agent_id): + logger.info( + "agent.send dropped: target=%s is %s and cannot be woken", + target_agent_id, + self.statuses[target_agent_id], + ) + return False runtime = self.runtimes.setdefault(target_agent_id, AgentRuntime()) runtime.mailbox.append(dict(message)) self.pending_counts[target_agent_id] = self.pending_counts.get(target_agent_id, 0) + 1 diff --git a/strix/core/execution.py b/strix/core/execution.py index 9f2674d2..dfcd39fa 100644 --- a/strix/core/execution.py +++ b/strix/core/execution.py @@ -202,6 +202,7 @@ async def run_agent_loop( agent_id, session=session, interrupt_on_message=interactive, + resumable=interactive, ) result: RunResultBase | None = None @@ -1006,7 +1007,7 @@ async def _start_child_runner( ) -> None: session = open_agent_session(child_id, agents_db_path) sessions_to_close.append(session) - await coordinator.attach_runtime(child_id, session=session) + await coordinator.attach_runtime(child_id, session=session, resumable=interactive) child_ctx: dict[str, Any] = dict(parent_ctx) child_ctx["agent_id"] = child_id diff --git a/strix/tools/agents_graph/tools.py b/strix/tools/agents_graph/tools.py index ad16fe03..da05bb1e 100644 --- a/strix/tools/agents_graph/tools.py +++ b/strix/tools/agents_graph/tools.py @@ -15,6 +15,7 @@ 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.core.hooks import LLM_TURN_KEY +from strix.report.state import get_global_report_state from strix.skills import validate_requested_skills @@ -28,6 +29,40 @@ def _ctx(ctx: RunContextWrapper) -> dict[str, Any]: return ctx.context if isinstance(ctx.context, dict) else {} +def _filed_reports_by(agent_id: str) -> list[dict[str, Any]]: + """Vulnerability reports the agent actually filed, from report state. + + The narrative ``findings`` an agent hands to ``agent_finish`` is prose; a + parent that wants to act on a child's work needs the report ids. Read them + from the report state rather than trusting the child's description. + """ + state = get_global_report_state() + if state is None: + return [] + filed: list[dict[str, Any]] = [] + seen: set[str] = set() + for report in state.get_existing_vulnerabilities(): + if report.get("agent_id") != agent_id: + continue + report_id = str(report.get("id") or "") + if not report_id or report_id in seen: + continue + seen.add(report_id) + filed.append(report) + return filed + + +def _render_filed_report(report: dict[str, Any]) -> str: + line = f"- {report.get('id')}" + severity = report.get("severity") + if severity: + line += f" [{str(severity).upper()}]" + title = report.get("title") + if title: + line += f" {title}" + return line + + def _render_completion_report( *, agent_name: str, @@ -38,6 +73,7 @@ def _render_completion_report( findings: list[str], recommendations: list[str], open_items: list[str], + filed_reports: list[dict[str, Any]] | None = None, ) -> str: """Render a child's completion report as plain structured text. @@ -63,6 +99,12 @@ def _render_completion_report( lines.append("Findings:") lines.extend(f"- {f}" for f in findings) lines.append("") + lines.append("Vulnerability reports filed by this agent (authoritative; use these ids):") + if filed_reports: + lines.extend(_render_filed_report(r) for r in filed_reports) + else: + lines.append("- (none)") + lines.append("") lines.append("Open items (unresolved, need follow-up):") if open_items: lines.extend(f"- {o}" for o in open_items) @@ -149,8 +191,11 @@ async def send_message_to_agent( **Don't** use for routine "hello/status" pings, for context the target already has (children inherit parent history), or when parent/child completion via ``agent_finish`` already covers the - flow. Messages to any registered agent wake it, regardless of + flow. In interactive runs a message wakes the target regardless of status, so a follow-up can restart a completed/stopped/failed agent. + In non-interactive runs a finished agent is gone for good: the call + fails with the target's status, and you should read its filed + reports (``list_reports``) or spawn a new agent instead of waiting. Args: target_agent_id: Recipient's 8-char id. @@ -195,10 +240,23 @@ async def send_message_to_agent( }, ) if not delivered: + _, status = await coordinator.reachability(target_agent_id) + if status is None: + error = f"Target agent '{target_agent_id}' not found" + else: + error = ( + f"Target agent '{target_agent_id}' is '{status}' and cannot be woken in " + "this run; it will never read this message. Its filed reports are in " + "list_reports / get_report. Do not wait_for_agents on it - spawn a new " + "agent if more work is needed." + ) return json.dumps( { "success": False, - "error": f"Target agent '{target_agent_id}' not found or message delivery failed", + "error": error, + "target_agent_id": target_agent_id, + "target_status": status, + "delivery_status": "not_delivered", }, ensure_ascii=False, default=str, @@ -364,6 +422,31 @@ async def wait_for_agents( # noqa: PLR0911 default=str, ) + # Non-interactive agents cannot be woken once terminal, so with nobody + # running or waiting there is no message left to wait for. + if not await coordinator.active_agents_except(me): + _, statuses, names, _ = await coordinator.graph_snapshot() + return json.dumps( + { + "success": True, + "wait_outcome": "no_active_agents", + "reason": reason, + "agents": [ + {"agent_id": aid, "name": names.get(aid, aid), "status": status} + for aid, status in statuses.items() + if aid != me + ], + "note": ( + "No other agent is running or waiting, so no message can arrive. " + "Finished agents' results are in list_reports / get_report and their " + "completion reports are already in your history. Continue your own " + "work, spawn a new agent, or finish." + ), + }, + ensure_ascii=False, + default=str, + ) + await coordinator.park_waiting(me, wait_kind="agents") try: await asyncio.wait_for(coordinator.wait_for_message(me), timeout_seconds) @@ -610,6 +693,9 @@ async def agent_finish( default=str, ) + filed_reports = _filed_reports_by(me) + filed_report_ids = [str(r.get("id")) for r in filed_reports] + parent_notified = False if report_to_parent and await coordinator.claim_parent_notice(me): async with coordinator._lock: @@ -623,6 +709,7 @@ async def agent_finish( findings=list(findings or []), recommendations=list(final_recommendations or []), open_items=list(open_items or []), + filed_reports=filed_reports, ) await coordinator.send( parent_id, @@ -632,6 +719,7 @@ async def agent_finish( "content": report, "type": "completion", "priority": "high", + "filed_report_ids": filed_report_ids, }, ) parent_notified = True @@ -642,10 +730,11 @@ async def agent_finish( await notify_parent_on_terminal(coordinator, me, "completed") logger.info( - "agent_finish: %s success=%s findings=%d parent_notified=%s", + "agent_finish: %s success=%s findings=%d filed_reports=%d parent_notified=%s", me, success, len(findings or []), + len(filed_report_ids), parent_notified, ) @@ -656,6 +745,7 @@ async def agent_finish( "parent_notified": parent_notified, "agent_id": me, "summary": result_summary, + "filed_report_ids": filed_report_ids, "findings_count": len(findings or []), "open_items_count": len(open_items or []), "has_recommendations": bool(final_recommendations), diff --git a/tests/test_agent_graph_coordination.py b/tests/test_agent_graph_coordination.py new file mode 100644 index 00000000..9694656a --- /dev/null +++ b/tests/test_agent_graph_coordination.py @@ -0,0 +1,258 @@ +"""Tests for parent/child coordination once a non-interactive child has finished. + +A non-interactive agent's loop returns after its terminal state, so nothing will +ever read a message sent to it afterwards. Messaging it must say so instead of +reporting delivery, waiting on it must return at once, and its completion report +must carry the ids of the reports it actually filed so the parent does not have +to go asking. +""" + +from __future__ import annotations + +import json +from typing import TYPE_CHECKING, Any, cast + +import pytest +from agents.tool_context import ToolContext + +from strix.core.agents import AgentCoordinator +from strix.report.state import ReportState, set_global_report_state +from strix.tools.agents_graph.tools import agent_finish, send_message_to_agent, wait_for_agents + + +if TYPE_CHECKING: + from collections.abc import Iterator + from pathlib import Path + + +@pytest.fixture +def report_state(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Iterator[ReportState]: + monkeypatch.chdir(tmp_path) + state = ReportState(run_name="test-run") + set_global_report_state(state) + yield state + set_global_report_state(None) + + +async def _graph(*, interactive: bool) -> AgentCoordinator: + coordinator = AgentCoordinator() + await coordinator.register("root", "strix", parent_id=None) + await coordinator.register("child", "Validator", parent_id="root") + await coordinator.attach_runtime("root", resumable=interactive) + await coordinator.attach_runtime("child", resumable=interactive) + return coordinator + + +async def _call( + tool: Any, coordinator: AgentCoordinator, agent_id: str, args: dict[str, Any], **extra: Any +) -> dict[str, Any]: + ctx = ToolContext( + context={"coordinator": coordinator, "agent_id": agent_id, **extra}, + tool_name=tool.name, + tool_call_id="call-1", + tool_arguments="{}", + ) + raw: str = await tool.on_invoke_tool(ctx, json.dumps(args)) + return cast("dict[str, Any]", json.loads(raw)) + + +# --- send_message_to_agent ------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_message_to_finished_non_interactive_child_is_not_delivered() -> None: + coordinator = await _graph(interactive=False) + await coordinator.set_status("child", "completed") + + result = await _call( + send_message_to_agent, + coordinator, + "root", + {"target_agent_id": "child", "message": "did you file it?", "message_type": "query"}, + ) + + assert result["success"] is False + assert result["delivery_status"] == "not_delivered" + assert result["target_status"] == "completed" + assert "list_reports" in result["error"] + assert coordinator.pending_counts.get("child", 0) == 0 + assert coordinator.runtimes["child"].mailbox == [] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("status", ["stopped", "failed", "crashed"]) +async def test_every_terminal_non_interactive_status_is_unreachable(status: str) -> None: + coordinator = await _graph(interactive=False) + await coordinator.set_status("child", status) + + assert await coordinator.send("child", {"from": "root", "content": "hi"}) is False + assert await coordinator.reachability("child") == (False, status) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("status", ["running", "waiting"]) +async def test_message_to_live_child_is_delivered(status: str) -> None: + coordinator = await _graph(interactive=False) + await coordinator.set_status("child", status) + + result = await _call( + send_message_to_agent, + coordinator, + "root", + {"target_agent_id": "child", "message": "wrap up"}, + ) + + assert result["success"] is True + assert result["delivery_status"] == "delivered" + assert coordinator.pending_counts["child"] == 1 + + +@pytest.mark.asyncio +async def test_message_to_finished_interactive_child_still_wakes_it() -> None: + # An interactive loop parks after finishing and resumes on a message. + coordinator = await _graph(interactive=True) + await coordinator.set_status("child", "completed") + + result = await _call( + send_message_to_agent, + coordinator, + "root", + {"target_agent_id": "child", "message": "one more thing"}, + ) + + assert result["success"] is True + assert coordinator.pending_counts["child"] == 1 + + +@pytest.mark.asyncio +async def test_unknown_target_is_reported_as_not_found() -> None: + coordinator = await _graph(interactive=False) + + result = await _call( + send_message_to_agent, + coordinator, + "root", + {"target_agent_id": "ghost", "message": "hello"}, + ) + + assert result["success"] is False + assert result["target_status"] is None + assert "not found" in result["error"] + + +# --- wait_for_agents ------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_wait_returns_at_once_when_no_child_can_answer() -> None: + coordinator = await _graph(interactive=False) + await coordinator.set_status("child", "completed") + # The completion report was already consumed in an earlier turn. + + result = await _call( + wait_for_agents, + coordinator, + "root", + {"reason": "waiting for validator", "timeout_seconds": 240}, + ) + + assert result["wait_outcome"] == "no_active_agents" + assert result["agents"] == [{"agent_id": "child", "name": "Validator", "status": "completed"}] + assert coordinator.statuses["root"] == "running" + + +@pytest.mark.asyncio +async def test_wait_delivers_a_pending_report_before_checking_liveness() -> None: + coordinator = await _graph(interactive=False) + await coordinator.send("root", {"from": "child", "type": "completion", "content": "done"}) + await coordinator.set_status("child", "completed") + + result = await _call(wait_for_agents, coordinator, "root", {"timeout_seconds": 5}) + + assert result["wait_outcome"] == "message_arrived" + assert result["pending_messages"] == 1 + + +@pytest.mark.asyncio +async def test_wait_still_parks_while_a_child_is_running() -> None: + coordinator = await _graph(interactive=False) + + result = await _call(wait_for_agents, coordinator, "root", {"timeout_seconds": 1}) + + assert result["wait_outcome"] == "timeout" + + +@pytest.mark.asyncio +async def test_interactive_wait_parks_even_without_active_children() -> None: + # In an interactive run a finished child can be woken later, so parking is + # legitimate; the run loop's own auto-resume bounds the wait. + coordinator = await _graph(interactive=True) + await coordinator.set_status("child", "completed") + + result = await _call( + wait_for_agents, coordinator, "root", {"timeout_seconds": 5}, interactive=True + ) + + assert result["wait_outcome"] == "waiting" + + +# --- agent_finish ---------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_agent_finish_lists_the_reports_the_child_filed(report_state: ReportState) -> None: + coordinator = await _graph(interactive=False) + mine = report_state.add_vulnerability_report( + title="IDOR on /api/audits", severity="high", agent_id="child", agent_name="Validator" + ) + report_state.add_vulnerability_report(title="Root's own", severity="low", agent_id="root") + + result = await _call( + agent_finish, + coordinator, + "child", + {"result_summary": "confirmed", "findings": ["IDOR confirmed"]}, + parent_id="root", + ) + + assert result["filed_report_ids"] == [mine] + delivered = coordinator.runtimes["root"].mailbox + assert len(delivered) == 1 + assert delivered[0]["filed_report_ids"] == [mine] + body = delivered[0]["content"] + assert f"- {mine} [HIGH] IDOR on /api/audits" in body + assert "Root's own" not in body + + +@pytest.mark.asyncio +async def test_agent_finish_states_explicitly_when_nothing_was_filed( + report_state: ReportState, +) -> None: + coordinator = await _graph(interactive=False) + report_state.add_vulnerability_report(title="Someone else's", severity="low", agent_id="root") + + result = await _call( + agent_finish, + coordinator, + "child", + {"result_summary": "nothing exploitable", "findings": ["ruled out X"]}, + parent_id="root", + ) + + assert result["filed_report_ids"] == [] + body = coordinator.runtimes["root"].mailbox[0]["content"] + assert "Vulnerability reports filed by this agent" in body + assert body.index("filed by this agent") < body.index("- (none)") + + +@pytest.mark.asyncio +async def test_agent_finish_without_report_state_still_completes() -> None: + set_global_report_state(None) + coordinator = await _graph(interactive=False) + + result = await _call( + agent_finish, coordinator, "child", {"result_summary": "done"}, parent_id="root" + ) + + assert result["success"] is True + assert result["filed_report_ids"] == [] diff --git a/tests/test_wait_dedupe.py b/tests/test_wait_dedupe.py index db97fb4d..9cbe6392 100644 --- a/tests/test_wait_dedupe.py +++ b/tests/test_wait_dedupe.py @@ -41,6 +41,8 @@ def _fast_wait(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]: async def _context() -> dict[str, Any]: coordinator = AgentCoordinator() await coordinator.register("root", "strix", parent_id=None) + # A live child keeps the wait genuine: with nobody to hear from it returns at once. + await coordinator.register("child", "recon", parent_id="root") return {"agent_id": "root", "coordinator": coordinator}