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

@@ -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

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

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),

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"] == []

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}