mirror of
https://github.com/usestrix/strix.git
synced 2026-08-16 09:26:39 +02:00
Ahmed Allam
69a60f3b7a
Interactive turns ended by plain text left the agent parked in 'waiting' forever. Require an explicit lifecycle tool in both modes and nudge a text-only turn back into a tool call, bounded by a recovery limit.
934 lines
32 KiB
Python
934 lines
32 KiB
Python
"""Tests for the scan-wide budget-stop signal on the agent coordinator."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import contextlib
|
|
import json
|
|
from typing import Any, cast
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
from agents.exceptions import MaxTurnsExceeded
|
|
from agents.items import MessageOutputItem
|
|
from agents.memory import SQLiteSession
|
|
from agents.tool_context import ToolContext
|
|
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,
|
|
)
|
|
from strix.core.sessions import seed_initial_input
|
|
from strix.tools.finish.tool import finish_scan
|
|
|
|
|
|
_NO_STREAM_EVENTS: list[Any] = []
|
|
|
|
|
|
class _StructuredRefusalStream:
|
|
def __init__(self, refusal: str) -> None:
|
|
self.run_loop_exception: BaseException | None = None
|
|
self.new_items = [
|
|
MessageOutputItem(
|
|
agent=MagicMock(),
|
|
raw_item=ResponseOutputMessage(
|
|
id="msg-refusal",
|
|
content=[ResponseOutputRefusal(type="refusal", refusal=refusal)],
|
|
role="assistant",
|
|
status="completed",
|
|
type="message",
|
|
),
|
|
)
|
|
]
|
|
|
|
async def stream_events(self) -> Any:
|
|
for event in _NO_STREAM_EVENTS:
|
|
yield event
|
|
|
|
def cancel(self, mode: str = "immediate") -> None: # noqa: ARG002
|
|
return
|
|
|
|
|
|
async def _call_finish_scan(
|
|
coordinator: AgentCoordinator, agent_id: str, parent_id: str | None
|
|
) -> dict[str, Any]:
|
|
ctx = ToolContext(
|
|
context={"coordinator": coordinator, "agent_id": agent_id, "parent_id": parent_id},
|
|
tool_name="finish_scan",
|
|
tool_call_id="call-1",
|
|
tool_arguments="{}",
|
|
)
|
|
fields = ("executive_summary", "methodology", "technical_analysis", "recommendations")
|
|
result: str = await finish_scan.on_invoke_tool(ctx, json.dumps(dict.fromkeys(fields, "x")))
|
|
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()
|
|
await coordinator.register("root", "strix", parent_id=None)
|
|
await coordinator.register("child-a", "recon", parent_id="root")
|
|
await coordinator.register("child-b", "recon", parent_id="root")
|
|
|
|
sent: list[tuple[str, dict[str, Any]]] = []
|
|
|
|
async def _record(target_agent_id: str, message: dict[str, Any]) -> bool:
|
|
sent.append((target_agent_id, message))
|
|
return True
|
|
|
|
monkeypatch.setattr(coordinator, "send", _record)
|
|
|
|
await _notify_root_on_budget_reserve(coordinator)
|
|
await _notify_root_on_budget_reserve(coordinator)
|
|
|
|
assert len(sent) == 1
|
|
target, message = sent[0]
|
|
assert target == "root"
|
|
assert message["type"] == "budget_reserve_stop"
|
|
assert "finish_scan" in str(message["content"])
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_concurrent_reserve_claims_yield_single_root() -> None:
|
|
coordinator = AgentCoordinator()
|
|
await coordinator.register("root", "strix", parent_id=None)
|
|
for i in range(12):
|
|
await coordinator.register(f"child-{i}", "recon", parent_id="root")
|
|
|
|
results = await asyncio.gather(*(coordinator.claim_reserve_notification() for _ in range(12)))
|
|
|
|
assert results.count("root") == 1
|
|
assert all(r is None for r in results if r != "root")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_claim_reserve_sets_flag_and_wakes_parked_agents() -> None:
|
|
coordinator = AgentCoordinator()
|
|
await coordinator.register("root", "strix", parent_id=None)
|
|
await coordinator.register("child", "recon", parent_id="root")
|
|
|
|
flag_before = coordinator.reserve_stopped
|
|
assert flag_before is False
|
|
waiter = asyncio.create_task(coordinator.wait_for_message("child"))
|
|
await asyncio.sleep(0)
|
|
assert not waiter.done()
|
|
|
|
await coordinator.claim_reserve_notification()
|
|
|
|
flag_after = coordinator.reserve_stopped
|
|
assert flag_after is True
|
|
await asyncio.wait_for(waiter, timeout=1.0)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_finish_scan_bypasses_active_agent_guard_after_reserve() -> None:
|
|
coordinator = AgentCoordinator()
|
|
await coordinator.register("root", "strix", parent_id=None)
|
|
await coordinator.register("child", "recon", parent_id="root")
|
|
await coordinator.set_status("child", "running")
|
|
|
|
blocked = await _call_finish_scan(coordinator, "root", None)
|
|
assert blocked["scan_completed"] is False
|
|
assert blocked["active_agents"]
|
|
|
|
await coordinator.claim_reserve_notification()
|
|
|
|
finished = await _call_finish_scan(coordinator, "root", None)
|
|
assert finished["scan_completed"] is True
|
|
assert coordinator.statuses["root"] == "completed"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_finish_scan_gate_ignores_sub_agent_caller() -> None:
|
|
coordinator = AgentCoordinator()
|
|
await coordinator.register("root", "strix", parent_id=None)
|
|
await coordinator.register("child", "recon", parent_id="root")
|
|
await coordinator.set_status("child", "running")
|
|
|
|
result = await _call_finish_scan(coordinator, "child", "root")
|
|
assert "active_agents" not in result
|
|
assert result["success"] is False
|
|
assert "root" in result["error"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reserve_stop_notify_noop_without_root(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
coordinator = AgentCoordinator()
|
|
await coordinator.register("child", "recon", parent_id="missing")
|
|
|
|
sent: list[tuple[str, dict[str, Any]]] = []
|
|
|
|
async def _record(target_agent_id: str, message: dict[str, Any]) -> bool:
|
|
sent.append((target_agent_id, message))
|
|
return True
|
|
|
|
monkeypatch.setattr(coordinator, "send", _record)
|
|
await _notify_root_on_budget_reserve(coordinator)
|
|
|
|
assert sent == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_snapshot_round_trip_preserves_stop_flags() -> None:
|
|
coordinator = AgentCoordinator()
|
|
await coordinator.register("root", "strix", parent_id=None)
|
|
await coordinator.trigger_budget_stop()
|
|
await coordinator.claim_reserve_notification()
|
|
|
|
snap = await coordinator.snapshot()
|
|
assert snap["budget_stopped"] is True
|
|
assert snap["reserve_stopped"] is True
|
|
|
|
restored = AgentCoordinator()
|
|
await restored.restore(snap)
|
|
assert restored.budget_stopped is True
|
|
assert restored.reserve_stopped is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_legacy_snapshot_without_stop_flags_defaults_to_false() -> None:
|
|
coordinator = AgentCoordinator()
|
|
await coordinator.register("root", "strix", parent_id=None)
|
|
snap = await coordinator.snapshot()
|
|
del snap["budget_stopped"]
|
|
del snap["reserve_stopped"]
|
|
|
|
restored = AgentCoordinator()
|
|
await restored.restore(snap)
|
|
assert restored.budget_stopped is False
|
|
assert restored.reserve_stopped is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_randomized_reserve_claim_race_many_interleavings() -> None:
|
|
for seed in range(25):
|
|
coordinator = AgentCoordinator()
|
|
await coordinator.register("root", "strix", parent_id=None)
|
|
child_ids = [f"child-{i}" for i in range(8)]
|
|
for child_id in child_ids:
|
|
await coordinator.register(child_id, "recon", parent_id="root")
|
|
|
|
waiters = [asyncio.create_task(coordinator.wait_for_message(cid)) for cid in child_ids]
|
|
await asyncio.sleep(0)
|
|
|
|
async def _claim(delay: float, coord: AgentCoordinator = coordinator) -> str | None:
|
|
await asyncio.sleep(delay)
|
|
return await coord.claim_reserve_notification()
|
|
|
|
delays = [((seed * 31 + i * 17) % 50) / 10_000 for i in range(len(child_ids))]
|
|
results = await asyncio.gather(*(_claim(delay) for delay in delays))
|
|
|
|
assert results.count("root") == 1, f"seed {seed}: expected exactly one winner"
|
|
await asyncio.wait_for(asyncio.gather(*waiters), timeout=1.0)
|
|
assert coordinator.reserve_stopped is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reserve_claim_never_loses_root_wake() -> None:
|
|
for _ in range(10):
|
|
coordinator = AgentCoordinator()
|
|
await coordinator.register("root", "strix", parent_id=None)
|
|
await coordinator.register("child", "recon", parent_id="root")
|
|
|
|
root_waiter = asyncio.create_task(coordinator.wait_for_message("root"))
|
|
await asyncio.sleep(0)
|
|
assert not root_waiter.done()
|
|
|
|
await coordinator.claim_reserve_notification()
|
|
await asyncio.sleep(0)
|
|
|
|
async with coordinator._lock:
|
|
coordinator.pending_counts["root"] = 1
|
|
coordinator.runtimes["root"].wake.set()
|
|
|
|
await asyncio.wait_for(root_waiter, timeout=1.0)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_budget_stop_takes_precedence_over_reserve_for_all_roles() -> None:
|
|
coordinator = AgentCoordinator()
|
|
await coordinator.register("root", "strix", parent_id=None)
|
|
await coordinator.register("child", "recon", parent_id="root")
|
|
await coordinator.claim_reserve_notification()
|
|
await coordinator.trigger_budget_stop()
|
|
|
|
await asyncio.wait_for(coordinator.wait_for_message("root"), timeout=1.0)
|
|
await asyncio.wait_for(coordinator.wait_for_message("child"), timeout=1.0)
|
|
assert coordinator.budget_stopped is True
|
|
assert coordinator.reserve_stopped is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_root_not_released_by_reserve_alone() -> None:
|
|
coordinator = AgentCoordinator()
|
|
await coordinator.register("root", "strix", parent_id=None)
|
|
await coordinator.register("child", "recon", parent_id="root")
|
|
|
|
await coordinator.claim_reserve_notification()
|
|
|
|
root_waiter = asyncio.create_task(coordinator.wait_for_message("root"))
|
|
await asyncio.sleep(0.02)
|
|
assert not root_waiter.done()
|
|
|
|
root_waiter.cancel()
|
|
with contextlib.suppress(asyncio.CancelledError):
|
|
await root_waiter
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_snapshot_during_concurrent_claims_is_consistent() -> None:
|
|
coordinator = AgentCoordinator()
|
|
await coordinator.register("root", "strix", parent_id=None)
|
|
for i in range(6):
|
|
await coordinator.register(f"child-{i}", "recon", parent_id="root")
|
|
|
|
claims = [asyncio.create_task(coordinator.claim_reserve_notification()) for _ in range(6)]
|
|
snap = await coordinator.snapshot()
|
|
await asyncio.gather(*claims)
|
|
|
|
assert isinstance(snap["reserve_stopped"], bool)
|
|
final_snap = await coordinator.snapshot()
|
|
assert final_snap["reserve_stopped"] is True
|
|
|
|
restored = AgentCoordinator()
|
|
await restored.restore(final_snap)
|
|
assert restored.reserve_stopped is True
|
|
assert await restored.claim_reserve_notification() is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_budget_stop_sets_flag() -> None:
|
|
coordinator = AgentCoordinator()
|
|
await coordinator.register("root", "strix", parent_id=None)
|
|
|
|
assert coordinator.budget_stopped is False
|
|
await coordinator.trigger_budget_stop()
|
|
assert coordinator.budget_stopped is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_budget_stop_unblocks_parked_agent() -> None:
|
|
# A parent parked in wait_for_message (awaiting a child) must be released so
|
|
# it can exit, no matter where in the tree the budget limit was hit.
|
|
coordinator = AgentCoordinator()
|
|
await coordinator.register("parent", "strix", parent_id=None)
|
|
|
|
waiter = asyncio.create_task(coordinator.wait_for_message("parent"))
|
|
await asyncio.sleep(0) # let the waiter park
|
|
assert not waiter.done()
|
|
|
|
await coordinator.trigger_budget_stop()
|
|
await asyncio.wait_for(waiter, timeout=1.0)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_wait_for_message_returns_immediately_after_budget_stop() -> None:
|
|
coordinator = AgentCoordinator()
|
|
await coordinator.register("agent", "recon", parent_id="parent")
|
|
await coordinator.trigger_budget_stop()
|
|
|
|
# No pending messages, but the stop flag short-circuits the wait.
|
|
await asyncio.wait_for(coordinator.wait_for_message("agent"), timeout=1.0)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_pause_for_budget_sets_flag_and_status() -> None:
|
|
coordinator = AgentCoordinator()
|
|
await coordinator.register("root", "strix", parent_id=None)
|
|
|
|
await coordinator.pause_for_budget("root")
|
|
assert coordinator.budget_paused is True
|
|
assert coordinator.statuses["root"] == "budget_paused"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resume_from_budget_pause_extends_and_nudges(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
coordinator = AgentCoordinator()
|
|
await coordinator.register("root", "strix", parent_id=None)
|
|
await coordinator.register("child-a", "recon", parent_id="root")
|
|
await coordinator.register("child-b", "recon", parent_id="root")
|
|
await coordinator.pause_for_budget("root")
|
|
await coordinator.pause_for_budget("child-a")
|
|
await coordinator.pause_for_budget("child-b")
|
|
|
|
extensions: list[int] = []
|
|
coordinator.set_budget_extender(lambda: extensions.append(1))
|
|
|
|
sent: list[tuple[str, dict[str, Any]]] = []
|
|
|
|
async def _record(target_agent_id: str, message: dict[str, Any]) -> bool:
|
|
sent.append((target_agent_id, message))
|
|
return True
|
|
|
|
monkeypatch.setattr(coordinator, "send", _record)
|
|
|
|
await coordinator.resume_from_budget_pause(exclude="root")
|
|
|
|
assert coordinator.budget_paused is False
|
|
assert len(extensions) == 1
|
|
assert all(coordinator.statuses[aid] == "waiting" for aid in ("root", "child-a", "child-b"))
|
|
assert sorted(target for target, _ in sent) == ["child-a", "child-b"]
|
|
assert all(message["type"] == "budget_extended" for _, message in sent)
|
|
|
|
await coordinator.resume_from_budget_pause(exclude="root")
|
|
assert len(extensions) == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_user_send_resumes_budget_pause(tmp_path: Any) -> None:
|
|
coordinator = AgentCoordinator()
|
|
await coordinator.register("root", "strix", parent_id=None)
|
|
session = SQLiteSession("root", tmp_path / "agents.db")
|
|
await coordinator.attach_runtime("root", session=session)
|
|
await coordinator.pause_for_budget("root")
|
|
|
|
extensions: list[int] = []
|
|
coordinator.set_budget_extender(lambda: extensions.append(1))
|
|
|
|
delivered = await coordinator.send("root", {"from": "user", "content": "keep going"})
|
|
|
|
assert delivered is True
|
|
assert coordinator.budget_paused is False
|
|
assert len(extensions) == 1
|
|
assert coordinator.statuses["root"] == "waiting"
|
|
assert coordinator.pending_counts["root"] == 1
|
|
session.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_non_user_send_does_not_resume_budget_pause(tmp_path: Any) -> None:
|
|
coordinator = AgentCoordinator()
|
|
await coordinator.register("root", "strix", parent_id=None)
|
|
session = SQLiteSession("root", tmp_path / "agents.db")
|
|
await coordinator.attach_runtime("root", session=session)
|
|
await coordinator.pause_for_budget("root")
|
|
|
|
extensions: list[int] = []
|
|
coordinator.set_budget_extender(lambda: extensions.append(1))
|
|
|
|
await coordinator.send("root", {"from": "system", "content": "status"})
|
|
|
|
assert coordinator.budget_paused is True
|
|
assert extensions == []
|
|
assert coordinator.statuses["root"] == "budget_paused"
|
|
session.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reset_budget_stops_clears_pause_and_normalizes_statuses() -> None:
|
|
coordinator = AgentCoordinator()
|
|
await coordinator.register("root", "strix", parent_id=None)
|
|
await coordinator.trigger_budget_stop()
|
|
await coordinator.claim_reserve_notification()
|
|
await coordinator.pause_for_budget("root")
|
|
|
|
await coordinator.reset_budget_stops(budget_stopped=False, reserve_stopped=False)
|
|
|
|
assert coordinator.budget_stopped is False
|
|
assert coordinator.reserve_stopped is False
|
|
assert coordinator.budget_paused is False
|
|
assert coordinator.statuses["root"] == "waiting"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reset_budget_stops_can_preserve_pause() -> None:
|
|
coordinator = AgentCoordinator()
|
|
await coordinator.register("root", "strix", parent_id=None)
|
|
await coordinator.pause_for_budget("root")
|
|
|
|
await coordinator.reset_budget_stops(
|
|
budget_stopped=False, reserve_stopped=False, budget_paused=True
|
|
)
|
|
|
|
assert coordinator.budget_paused is True
|
|
assert coordinator.statuses["root"] == "budget_paused"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_snapshot_round_trip_preserves_budget_pause() -> None:
|
|
coordinator = AgentCoordinator()
|
|
await coordinator.register("root", "strix", parent_id=None)
|
|
await coordinator.pause_for_budget("root")
|
|
|
|
snap = await coordinator.snapshot()
|
|
assert snap["budget_paused"] is True
|
|
|
|
restored = AgentCoordinator()
|
|
await restored.restore(snap)
|
|
assert restored.budget_paused is True
|
|
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()
|
|
|
|
|
|
class _RecordingStream:
|
|
def __init__(self) -> None:
|
|
self.cancelled = False
|
|
self.cancel_mode: str | None = None
|
|
|
|
def cancel(self, mode: str = "immediate") -> None:
|
|
self.cancelled = True
|
|
self.cancel_mode = mode
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_terminal_notice_does_not_cancel_parent_stream(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")
|
|
stream = _RecordingStream()
|
|
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")
|
|
|
|
assert stream.cancelled is False
|
|
assert coordinator.pending_counts.get("root", 0) > 0
|
|
session.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_queues_without_session_and_drains_on_consume(tmp_path: Any) -> None:
|
|
coordinator = AgentCoordinator()
|
|
await coordinator.register("root", "strix", parent_id=None)
|
|
|
|
assert await coordinator.send("root", {"from": "user", "content": "hello"}) is True
|
|
assert coordinator.pending_counts["root"] == 1
|
|
|
|
session = SQLiteSession("root", tmp_path / "agents.db")
|
|
await coordinator.attach_runtime("root", session=session)
|
|
|
|
count, items = await coordinator.consume_pending("root", include_items=True)
|
|
assert count == 1
|
|
assert items[0]["content"] == "hello"
|
|
stored = await session.get_items()
|
|
last = cast("dict[str, Any]", stored[-1])
|
|
assert last["content"] == "hello"
|
|
session.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_error_parked_agent_only_released_by_user_message(tmp_path: Any) -> None:
|
|
coordinator = AgentCoordinator()
|
|
await coordinator.register("root", "strix", parent_id=None)
|
|
await coordinator.register("child", "recon", parent_id="root")
|
|
session = SQLiteSession("child", tmp_path / "agents.db")
|
|
await coordinator.attach_runtime("child", session=session)
|
|
await coordinator.set_status("child", "crashed", error="boom")
|
|
|
|
await coordinator.send("child", {"from": "root", "content": "peer nudge"})
|
|
waiter = asyncio.create_task(coordinator.wait_for_message("child"))
|
|
await asyncio.sleep(0.05)
|
|
assert not waiter.done()
|
|
|
|
await coordinator.send("child", {"from": "user", "content": "wake up"})
|
|
assert await asyncio.wait_for(waiter, timeout=1.0) is True
|
|
|
|
count, items = await coordinator.consume_pending("child", include_items=True)
|
|
assert count == 2
|
|
assert items[0]["content"].endswith("peer nudge")
|
|
assert items[1]["content"] == "wake up"
|
|
session.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_wait_for_message_timeout_returns_false() -> None:
|
|
coordinator = AgentCoordinator()
|
|
await coordinator.register("child", "recon", parent_id="root")
|
|
|
|
assert await coordinator.wait_for_message("child", timeout=0.05) is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_snapshot_round_trip_preserves_mailboxes() -> None:
|
|
coordinator = AgentCoordinator()
|
|
await coordinator.register("root", "strix", parent_id=None)
|
|
await coordinator.send("root", {"from": "user", "content": "queued"})
|
|
|
|
snap = await coordinator.snapshot()
|
|
restored = AgentCoordinator()
|
|
await restored.restore(snap)
|
|
|
|
assert restored.pending_counts["root"] == 1
|
|
assert restored.runtimes["root"].mailbox == [{"from": "user", "content": "queued"}]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_cycle_parked_parks_instead_of_raising(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
async def _boom(*_args: Any, **_kwargs: Any) -> Any:
|
|
raise RuntimeError("unexpected explosion")
|
|
|
|
monkeypatch.setattr(execution, "_run_cycle", _boom)
|
|
|
|
coordinator = AgentCoordinator()
|
|
await coordinator.register("root", "strix", parent_id=None)
|
|
|
|
result = await execution._run_cycle_parked(
|
|
object(),
|
|
coordinator,
|
|
"root",
|
|
input_data=[],
|
|
run_config=None, # type: ignore[arg-type]
|
|
context={},
|
|
max_turns=5,
|
|
session=None,
|
|
event_sink=None,
|
|
hooks=None,
|
|
)
|
|
|
|
assert result is None
|
|
assert coordinator.statuses["root"] == "failed"
|
|
assert coordinator.errors["root"] == "unexpected explosion"
|
|
|
|
|
|
class _SalvageStream:
|
|
def __init__(self, replay: list[dict[str, Any]]) -> None:
|
|
self._replay = replay
|
|
|
|
def to_input_list(self) -> list[dict[str, Any]]:
|
|
return self._replay
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_salvage_stream_to_session_preserves_full_history(tmp_path: Any) -> None:
|
|
session = SQLiteSession("child", tmp_path / "agents.db")
|
|
await session.add_items([{"role": "user", "content": "identity + task"}])
|
|
pre_run = list(await session.get_items())
|
|
|
|
# A crash mid-run: the stream produced two turns the SDK never committed.
|
|
stream = _SalvageStream(
|
|
[
|
|
{"role": "assistant", "content": "recon turn 1"},
|
|
{"role": "assistant", "content": "recon turn 2"},
|
|
]
|
|
)
|
|
await execution._salvage_stream_to_session(session, pre_run, stream, "child")
|
|
|
|
stored = [cast("dict[str, Any]", i) for i in await session.get_items()]
|
|
assert [i["content"] for i in stored] == [
|
|
"identity + task",
|
|
"recon turn 1",
|
|
"recon turn 2",
|
|
]
|
|
|
|
# A crash with nothing new to salvage leaves the session untouched.
|
|
await execution._salvage_stream_to_session(
|
|
session, list(await session.get_items()), _SalvageStream([]), "child"
|
|
)
|
|
assert len(await session.get_items()) == 3
|
|
session.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_seed_initial_input_persists_and_is_idempotent(tmp_path: Any) -> None:
|
|
session = SQLiteSession("child", tmp_path / "agents.db")
|
|
identity = [{"role": "user", "content": "You are agent recon (abc); do X."}]
|
|
|
|
assert await seed_initial_input(session, identity) is True
|
|
assert len(await session.get_items()) == 1
|
|
|
|
# A populated session is left untouched (no duplicate identity message).
|
|
assert await seed_initial_input(session, identity) is False
|
|
assert len(await session.get_items()) == 1
|
|
|
|
assert await seed_initial_input(session, []) is False
|
|
session.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_structured_provider_refusal_fails_interactive_agent(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
refusal = "This request was blocked under the provider's usage policy."
|
|
stream = _StructuredRefusalStream(refusal)
|
|
monkeypatch.setattr(
|
|
"strix.core.execution.Runner.run_streamed", lambda *_args, **_kwargs: stream
|
|
)
|
|
coordinator = AgentCoordinator()
|
|
await coordinator.register("root", "strix", parent_id=None)
|
|
|
|
result = await execution._run_cycle(
|
|
MagicMock(),
|
|
coordinator,
|
|
"root",
|
|
input_data="task",
|
|
run_config=MagicMock(),
|
|
context={},
|
|
max_turns=5,
|
|
session=None,
|
|
interactive=True,
|
|
event_sink=None,
|
|
hooks=None,
|
|
)
|
|
|
|
assert result is None
|
|
assert coordinator.statuses["root"] == "failed"
|
|
assert coordinator.errors["root"] == refusal
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_structured_provider_refusal_fails_noninteractive_child(
|
|
tmp_path: Any,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
refusal = "This request was blocked under the provider's usage policy."
|
|
stream = _StructuredRefusalStream(refusal)
|
|
monkeypatch.setattr(
|
|
"strix.core.execution.Runner.run_streamed", lambda *_args, **_kwargs: stream
|
|
)
|
|
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)
|
|
|
|
result = await execution._run_cycle(
|
|
MagicMock(),
|
|
coordinator,
|
|
"child",
|
|
input_data="task",
|
|
run_config=MagicMock(),
|
|
context={"parent_id": "root"},
|
|
max_turns=5,
|
|
session=None,
|
|
interactive=False,
|
|
event_sink=None,
|
|
hooks=None,
|
|
)
|
|
|
|
assert result is None
|
|
assert coordinator.statuses["child"] == "failed"
|
|
assert coordinator.errors["child"] == refusal
|
|
assert coordinator.pending_counts.get("root", 0) > 0
|
|
session.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_agent_loop_seeds_identity_before_first_cycle(
|
|
tmp_path: Any, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
coordinator = AgentCoordinator()
|
|
await coordinator.register("child", "recon", parent_id="root")
|
|
session = SQLiteSession("child", tmp_path / "agents.db")
|
|
|
|
captured: dict[str, Any] = {}
|
|
|
|
async def _crash_first_turn(*_args: Any, **kwargs: Any) -> Any:
|
|
captured["input_data"] = kwargs.get("input_data")
|
|
captured["items_at_start"] = await session.get_items()
|
|
raise RuntimeError("first-turn crash")
|
|
|
|
monkeypatch.setattr(execution, "_run_cycle", _crash_first_turn)
|
|
|
|
identity = [{"role": "user", "content": "You are agent recon (abc); maintain your identity."}]
|
|
with pytest.raises(RuntimeError, match="first-turn crash"):
|
|
await execution.run_agent_loop(
|
|
agent=object(),
|
|
initial_input=identity,
|
|
run_config=None, # type: ignore[arg-type]
|
|
context={"agent_id": "child", "parent_id": "root"},
|
|
max_turns=5,
|
|
coordinator=coordinator,
|
|
agent_id="child",
|
|
interactive=False,
|
|
session=session,
|
|
)
|
|
|
|
# The first cycle ran with an empty input against the pre-seeded session.
|
|
assert captured["input_data"] == []
|
|
assert captured["items_at_start"]
|
|
# The identity/task survives the first-turn crash, so a revival can resume it.
|
|
stored = await session.get_items()
|
|
assert any("recon" in str(cast("dict[str, Any]", i).get("content", "")) for i in stored)
|
|
session.close()
|
|
|
|
|
|
def _scripted_cycle(
|
|
coordinator: AgentCoordinator,
|
|
agent_id: str,
|
|
statuses: list[str],
|
|
calls: list[Any],
|
|
) -> Any:
|
|
"""Fake run cycle that leaves ``agent_id`` in a scripted status per call."""
|
|
|
|
async def _cycle(*_args: Any, **kwargs: Any) -> Any:
|
|
calls.append(kwargs.get("input_data"))
|
|
status = statuses[min(len(calls) - 1, len(statuses) - 1)]
|
|
await coordinator.set_status(agent_id, status)
|
|
return MagicMock(final_output="plain text, no tool call")
|
|
|
|
return _cycle
|
|
|
|
|
|
async def _drive(
|
|
coordinator: AgentCoordinator,
|
|
agent_id: str,
|
|
*,
|
|
interactive: bool,
|
|
max_turns: int = 5,
|
|
) -> Any:
|
|
return await execution._run_until_lifecycle(
|
|
MagicMock(),
|
|
coordinator,
|
|
agent_id,
|
|
initial_input=[],
|
|
run_config=MagicMock(),
|
|
context={"agent_id": agent_id, "parent_id": None},
|
|
max_turns=max_turns,
|
|
session=None,
|
|
interactive=interactive,
|
|
event_sink=None,
|
|
hooks=None,
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_interactive_text_only_turn_is_nudged_instead_of_parking(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""A no-tool-call turn must not silently hand control back to the user."""
|
|
coordinator = AgentCoordinator()
|
|
await coordinator.register("root", "strix", parent_id=None)
|
|
calls: list[Any] = []
|
|
monkeypatch.setattr(
|
|
execution,
|
|
"_run_cycle_parked",
|
|
_scripted_cycle(coordinator, "root", ["running", "completed"], calls),
|
|
)
|
|
|
|
await _drive(coordinator, "root", interactive=True)
|
|
|
|
assert len(calls) == 2
|
|
# The retry carries an explicit "call a tool" nudge rather than empty input.
|
|
nudge = calls[1][0]["content"]
|
|
assert "without a tool call" in nudge
|
|
assert "wait_for_message" in nudge
|
|
assert coordinator.statuses["root"] == "completed"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_interactive_wait_for_message_parks_without_a_nudge(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""``waiting`` is now only reachable by an explicit wait_for_message call."""
|
|
coordinator = AgentCoordinator()
|
|
await coordinator.register("root", "strix", parent_id=None)
|
|
calls: list[Any] = []
|
|
monkeypatch.setattr(
|
|
execution,
|
|
"_run_cycle_parked",
|
|
_scripted_cycle(coordinator, "root", ["waiting"], calls),
|
|
)
|
|
|
|
await _drive(coordinator, "root", interactive=True)
|
|
|
|
assert len(calls) == 1
|
|
assert coordinator.statuses["root"] == "waiting"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_interactive_recovery_exhaustion_parks_instead_of_crashing(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""A human can resume an interactive scan, so exhaustion parks rather than dies."""
|
|
coordinator = AgentCoordinator()
|
|
await coordinator.register("root", "strix", parent_id=None)
|
|
calls: list[Any] = []
|
|
monkeypatch.setattr(
|
|
execution,
|
|
"_run_cycle_parked",
|
|
_scripted_cycle(coordinator, "root", ["running"], calls),
|
|
)
|
|
|
|
await _drive(coordinator, "root", interactive=True)
|
|
|
|
assert len(calls) == execution._INTERACTIVE_TOOL_RECOVERY_LIMIT
|
|
assert coordinator.statuses["root"] == "waiting"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_noninteractive_recovery_exhaustion_crashes(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""No user is present to resume an autonomous run, so it still fails loudly."""
|
|
coordinator = AgentCoordinator()
|
|
await coordinator.register("root", "strix", parent_id=None)
|
|
calls: list[Any] = []
|
|
monkeypatch.setattr(
|
|
execution,
|
|
"_run_cycle",
|
|
_scripted_cycle(coordinator, "root", ["running"], calls),
|
|
)
|
|
|
|
with pytest.raises(MaxTurnsExceeded):
|
|
await _drive(coordinator, "root", interactive=False, max_turns=2)
|
|
|
|
assert len(calls) == 2
|
|
assert coordinator.statuses["root"] == "crashed"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_tool_required_message_is_persisted_to_the_session(tmp_path: Any) -> None:
|
|
session = SQLiteSession("root", tmp_path / "agents.db")
|
|
|
|
assert (
|
|
await execution._append_tool_required_message(
|
|
session=session,
|
|
context={"parent_id": None},
|
|
attempt=1,
|
|
limit=3,
|
|
interactive=True,
|
|
)
|
|
== []
|
|
)
|
|
|
|
stored = [cast("dict[str, Any]", i) for i in await session.get_items()]
|
|
assert "finish_scan" in stored[0]["content"]
|
|
assert "wait_for_message" in stored[0]["content"]
|
|
session.close()
|