mirror of
https://github.com/usestrix/strix.git
synced 2026-08-16 01:16:40 +02:00
fix(core): wake the parent when a child ends without a completion report
This commit is contained in:
@@ -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:
|
||||
|
||||
+30
-5
@@ -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)
|
||||
|
||||
@@ -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(
|
||||
{
|
||||
|
||||
+67
-8
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user