diff --git a/strix/core/agents.py b/strix/core/agents.py index 93aa4975..97d94548 100644 --- a/strix/core/agents.py +++ b/strix/core/agents.py @@ -200,7 +200,9 @@ class AgentCoordinator: logger.info("agent.status %s=%s", agent_id, status) await self._maybe_snapshot() - async def send(self, target_agent_id: str, message: dict[str, Any]) -> bool: + async def send( + self, target_agent_id: str, message: dict[str, Any], *, interrupt: bool = True + ) -> bool: """Deliver a user/peer message by appending it to the target SDK session.""" if message.get("from") == "user" and self._budget_paused: await self.resume_from_budget_pause(exclude=target_agent_id) @@ -211,7 +213,7 @@ class AgentCoordinator: runtime = self.runtimes.setdefault(target_agent_id, AgentRuntime()) session = runtime.session stream = runtime.stream - interrupt = runtime.interrupt_on_message + interrupt_on_message = runtime.interrupt_on_message if session is None: logger.warning( "agent.send dropped target=%s because its SDK session is not attached", @@ -230,7 +232,7 @@ class AgentCoordinator: async with self._lock: self.pending_counts[target_agent_id] = self.pending_counts.get(target_agent_id, 0) + 1 self.runtimes.setdefault(target_agent_id, AgentRuntime()).wake.set() - if stream is not None and interrupt: + if stream is not None and interrupt and interrupt_on_message: stream.cancel(mode="immediate") await self._maybe_snapshot() return True diff --git a/strix/core/execution.py b/strix/core/execution.py index 7c251777..ae726ed3 100644 --- a/strix/core/execution.py +++ b/strix/core/execution.py @@ -21,6 +21,7 @@ from openai import ( RateLimitError, ) +from strix.config import codex from strix.core.hooks import ( BudgetExceededError, BudgetPausedError, @@ -88,6 +89,11 @@ async def _compact_session( ) +_GUARDRAIL_PARK_ERROR = ( + "Blocked by the model's content guardrail (flagged as a possible cybersecurity risk). " + "Set STRIX_LLM to a model that isn't blocked and resume the scan to continue." +) + _TRANSIENT_MODEL_STATUS_CODES = frozenset({408, 500, 502, 503, 504}) _MAX_TRANSIENT_MODEL_RETRIES = 4 _TRANSIENT_MODEL_RETRY_BASE_DELAY_S = 2.0 @@ -304,6 +310,7 @@ async def respawn_subagents( if coordinator.parent_of.get(aid) is None or aid == root_id: continue md["_restored_status"] = status + md["_restored_error"] = coordinator.errors.get(aid) candidates.append( ( aid, @@ -316,7 +323,8 @@ async def respawn_subagents( for child_id, name, parent_id, md in candidates: try: restored_status = str(md.get("_restored_status") or "running") - start_parked = interactive and restored_status != "running" + recoverable_park = restored_status == "waiting" and bool(md.get("_restored_error")) + start_parked = interactive and restored_status != "running" and not recoverable_park if start_parked: logger.warning( @@ -572,6 +580,10 @@ async def _run_cycle( # noqa: PLR0912, PLR0915 if session is not None: input_data = [] continue + if codex.is_content_guardrail_error(exc): + return await _handle_content_guardrail( + coordinator, agent_id, exc, interactive=interactive + ) if not interactive: raise if isinstance(exc, MaxTurnsExceeded): @@ -589,6 +601,22 @@ async def _run_cycle( # noqa: PLR0912, PLR0915 return stream +async def _handle_content_guardrail( + coordinator: AgentCoordinator, + agent_id: str, + exc: BaseException, + *, + interactive: bool, +) -> RunResultBase | None: + logger.warning("agent %s blocked by the model's content guardrail: %s", agent_id, exc) + if interactive: + await coordinator.set_status(agent_id, "waiting", error=_GUARDRAIL_PARK_ERROR) + return None + await coordinator.set_status(agent_id, "failed", error=_GUARDRAIL_PARK_ERROR) + await _notify_parent_on_terminal(coordinator, agent_id, "failed") + return None + + async def _settle_run_result( coordinator: AgentCoordinator, agent_id: str, @@ -685,6 +713,7 @@ async def _notify_parent_on_terminal( "priority": "high", "content": template.format(name=name, agent_id=agent_id), }, + interrupt=False, ) diff --git a/strix/core/runner.py b/strix/core/runner.py index fdb432d0..c5f51b15 100644 --- a/strix/core/runner.py +++ b/strix/core/runner.py @@ -376,6 +376,12 @@ async def run_strix_scan( async with coordinator._lock: root_status = coordinator.statuses.get(root_id) + root_error = coordinator.errors.get(root_id) + + root_recoverable_park = root_status == "waiting" and bool(root_error) + root_start_parked = bool( + interactive and is_resume and root_status != "running" and not root_recoverable_park + ) result = await run_agent_loop( agent=root_agent, @@ -387,7 +393,7 @@ async def run_strix_scan( agent_id=root_id, interactive=interactive, session=root_session, - start_parked=bool(interactive and is_resume and root_status != "running"), + start_parked=root_start_parked, event_sink=event_sink, hooks=hooks, ) diff --git a/tests/test_execution.py b/tests/test_execution.py index 985541c0..2ecd9a84 100644 --- a/tests/test_execution.py +++ b/tests/test_execution.py @@ -6,13 +6,21 @@ import asyncio import contextlib import json from typing import Any +from unittest.mock import MagicMock import pytest from agents.memory import SQLiteSession from agents.tool_context import ToolContext +from strix.config import codex +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.execution import ( + _handle_content_guardrail, + _notify_parent_on_terminal, + _notify_root_on_budget_reserve, + respawn_subagents, +) from strix.tools.finish.tool import finish_scan @@ -465,3 +473,106 @@ async def test_notify_parent_on_terminal_ignores_non_terminal_status(tmp_path: A 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_guardrail_interactive_parks_agent_wakeable(tmp_path: Any) -> None: + coordinator = AgentCoordinator() + await coordinator.register("root", "strix", parent_id=None) + await coordinator.register("child", "recon", parent_id="root") + exc = codex.CodexContentGuardrailError("chatgpt/gpt-5.6-sol") + + result = await _handle_content_guardrail(coordinator, "child", exc, interactive=True) + + assert result is None + assert coordinator.statuses["child"] == "waiting" + assert "STRIX_LLM" in coordinator.errors["child"] + + waiter = asyncio.create_task(coordinator.wait_for_message("child")) + await asyncio.sleep(0) + assert not waiter.done() + session = SQLiteSession("child", tmp_path / "agents.db") + await coordinator.attach_runtime("child", session=session) + await coordinator.send("child", {"from": "user", "content": "switched model, resume"}) + await asyncio.wait_for(waiter, timeout=1.0) + session.close() + + +@pytest.mark.asyncio +async def test_guardrail_noninteractive_fails_only_blocked_agent(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) + exc = codex.CodexContentGuardrailError("chatgpt/gpt-5.6-sol") + + result = await _handle_content_guardrail(coordinator, "child", exc, interactive=False) + + assert result is None + assert coordinator.statuses["child"] == "failed" + assert "STRIX_LLM" in coordinator.errors["child"] + assert coordinator.statuses["root"] == "running" + assert coordinator.pending_counts.get("root", 0) > 0 + session.close() + + +@pytest.mark.asyncio +async def test_resume_revives_guardrail_parked_child_but_not_plain_waiting( + tmp_path: Any, monkeypatch: pytest.MonkeyPatch +) -> None: + coordinator = AgentCoordinator() + await coordinator.register("root", "strix", parent_id=None) + await coordinator.register("blocked", "recon", parent_id="root") + await coordinator.register("peer_waiter", "recon", parent_id="root") + await coordinator.set_status("blocked", "waiting", error="STRIX_LLM guardrail") + await coordinator.set_status("peer_waiter", "waiting") + + parked: dict[str, bool] = {} + + async def _fake_start_child_runner(**kwargs: Any) -> None: + parked[kwargs["child_id"]] = bool(kwargs["start_parked"]) + + monkeypatch.setattr(execution, "_start_child_runner", _fake_start_child_runner) + + await respawn_subagents( + coordinator=coordinator, + factory=lambda **_kwargs: object(), + agents_db_path=tmp_path / "agents.db", + sessions_to_close=[], + run_config=MagicMock(), + max_turns=10, + interactive=True, + parent_ctx={"agent_id": "root", "parent_id": None}, + root_id="root", + ) + + assert parked["blocked"] is False + assert parked["peer_waiter"] is True