mirror of
https://github.com/usestrix/strix.git
synced 2026-08-16 09:26:39 +02:00
fix(runtime): wake parent when child hits a terminal state (MaxTurnsExceeded)
This commit is contained in:
+25
-9
@@ -417,7 +417,7 @@ async def _run_noninteractive_until_lifecycle(
|
|||||||
|
|
||||||
if invalid_final_outputs >= invalid_final_output_limit:
|
if invalid_final_outputs >= invalid_final_output_limit:
|
||||||
await coordinator.set_status(agent_id, "crashed")
|
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(
|
raise MaxTurnsExceeded(
|
||||||
"Agent exhausted non-interactive recovery attempts without calling "
|
"Agent exhausted non-interactive recovery attempts without calling "
|
||||||
"finish_scan or agent_finish."
|
"finish_scan or agent_finish."
|
||||||
@@ -582,7 +582,7 @@ async def _run_cycle( # noqa: PLR0912, PLR0915
|
|||||||
status = "crashed"
|
status = "crashed"
|
||||||
logger.exception("agent run failed for %s; parking as %s", agent_id, status)
|
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 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
|
return None
|
||||||
else:
|
else:
|
||||||
await _settle_run_result(coordinator, agent_id, interactive)
|
await _settle_run_result(coordinator, agent_id, interactive)
|
||||||
@@ -646,12 +646,31 @@ async def _append_noninteractive_tool_required_message(
|
|||||||
return []
|
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,
|
coordinator: AgentCoordinator,
|
||||||
agent_id: str,
|
agent_id: str,
|
||||||
status: str,
|
status: str,
|
||||||
) -> None:
|
) -> None:
|
||||||
if status != "crashed":
|
template = _TERMINAL_NOTICE.get(status)
|
||||||
|
if template is None:
|
||||||
return
|
return
|
||||||
async with coordinator._lock:
|
async with coordinator._lock:
|
||||||
parent = coordinator.parent_of.get(agent_id)
|
parent = coordinator.parent_of.get(agent_id)
|
||||||
@@ -662,12 +681,9 @@ async def _notify_parent_on_crash(
|
|||||||
parent,
|
parent,
|
||||||
{
|
{
|
||||||
"from": agent_id,
|
"from": agent_id,
|
||||||
"type": "crash",
|
"type": status,
|
||||||
"priority": "high",
|
"priority": "high",
|
||||||
"content": (
|
"content": template.format(name=name, agent_id=agent_id),
|
||||||
f"[Agent crash] {name} ({agent_id}) terminated unexpectedly. "
|
|
||||||
"Stop waiting on this child unless you want to message it again."
|
|
||||||
),
|
|
||||||
},
|
},
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|||||||
+39
-1
@@ -12,7 +12,7 @@ from agents.memory import SQLiteSession
|
|||||||
from agents.tool_context import ToolContext
|
from agents.tool_context import ToolContext
|
||||||
|
|
||||||
from strix.core.agents import AgentCoordinator
|
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
|
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)
|
await restored.restore(snap)
|
||||||
assert restored.budget_paused is True
|
assert restored.budget_paused is True
|
||||||
assert restored.statuses["root"] == "budget_paused"
|
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()
|
||||||
|
|||||||
Reference in New Issue
Block a user