diff --git a/strix/core/agents.py b/strix/core/agents.py index 97d94548..3f20629d 100644 --- a/strix/core/agents.py +++ b/strix/core/agents.py @@ -32,6 +32,8 @@ class AgentRuntime: stream: Any | None = None interrupt_on_message: bool = False wake: asyncio.Event = field(default_factory=asyncio.Event) + mailbox: list[dict[str, Any]] = field(default_factory=list) + user_wake_required: bool = False class AgentCoordinator: @@ -179,6 +181,7 @@ class AgentCoordinator: if agent_id in self.statuses: self.statuses[agent_id] = "running" self.errors.pop(agent_id, None) + self.runtimes.setdefault(agent_id, AgentRuntime()).user_wake_required = False await self._maybe_snapshot() async def park_waiting(self, agent_id: str) -> None: @@ -196,6 +199,7 @@ class AgentCoordinator: elif status == "running": self.errors.pop(agent_id, None) 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() @@ -203,49 +207,47 @@ class AgentCoordinator: 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: + """Queue a user/peer message in the target's mailbox and wake it.""" + from_user = message.get("from") == "user" + if from_user and self._budget_paused: await self.resume_from_budget_pause(exclude=target_agent_id) async with self._lock: if target_agent_id not in self.statuses: logger.debug("agent.send dropped unknown target=%s", target_agent_id) return False runtime = self.runtimes.setdefault(target_agent_id, AgentRuntime()) - session = runtime.session + runtime.mailbox.append(dict(message)) + self.pending_counts[target_agent_id] = self.pending_counts.get(target_agent_id, 0) + 1 + if from_user: + runtime.user_wake_required = False + runtime.wake.set() stream = runtime.stream 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", - target_agent_id, - ) - return False - try: - async with session_write_lock(session): - await session.add_items([self._message_to_session_item(message)]) - except Exception: - logger.exception( - "agent.send failed to append to SDK session target=%s", - target_agent_id, - ) - return False - 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 and interrupt_on_message: stream.cancel(mode="immediate") await self._maybe_snapshot() return True - async def wait_for_message(self, agent_id: str) -> None: + async def wait_for_message(self, agent_id: str, *, timeout: float | None = None) -> bool: + """Wait until a message is ready for ``agent_id``; False on ``timeout``.""" while True: async with self._lock: + runtime = self.runtimes.setdefault(agent_id, AgentRuntime()) reserve_exit = self._reserve_stopped and self.parent_of.get(agent_id) is not None - if self._budget_stopped or reserve_exit or self.pending_counts.get(agent_id, 0) > 0: - return - wake = self.runtimes.setdefault(agent_id, AgentRuntime()).wake + pending_ready = ( + self.pending_counts.get(agent_id, 0) > 0 and not runtime.user_wake_required + ) + if self._budget_stopped or reserve_exit or pending_ready: + return True + wake = runtime.wake wake.clear() - await wake.wait() + if timeout is None: + await wake.wait() + else: + try: + await asyncio.wait_for(wake.wait(), timeout) + except TimeoutError: + return False async def consume_pending( self, @@ -253,17 +255,38 @@ class AgentCoordinator: *, include_items: bool = False, ) -> tuple[int, list[Any]]: + """Drain the agent's mailbox into its own SDK session.""" async with self._lock: - count = self.pending_counts.get(agent_id, 0) + runtime = self.runtimes.setdefault(agent_id, AgentRuntime()) + queued = list(runtime.mailbox) + runtime.mailbox.clear() + count = max(self.pending_counts.get(agent_id, 0), len(queued)) self.pending_counts[agent_id] = 0 - session = self.runtimes.get(agent_id, AgentRuntime()).session + session = runtime.session if count <= 0: return 0, [] + items = [self._message_to_session_item(m) for m in queued] + if items: + if session is None: + logger.warning( + "agent %s has no SDK session attached; %d queued messages were not persisted", + agent_id, + len(items), + ) + else: + try: + async with session_write_lock(session): + await session.add_items(items) + except Exception: + logger.exception( + "failed to append %d queued messages to the session of %s", + len(items), + agent_id, + ) await self._maybe_snapshot() - if not include_items or session is None: + if not include_items: return count, [] - items = await session.get_items() - return count, list(items[-count:]) + return count, items async def request_stop(self, agent_id: str) -> None: async with self._lock: @@ -374,6 +397,11 @@ class AgentCoordinator: "names": dict(self.names), "metadata": {aid: dict(md) for aid, md in self.metadata.items()}, "pending_counts": dict(self.pending_counts), + "mailboxes": { + aid: [dict(m) for m in runtime.mailbox] + for aid, runtime in self.runtimes.items() + if runtime.mailbox + }, "errors": dict(self.errors), "budget_stopped": self._budget_stopped, "reserve_stopped": self._reserve_stopped, @@ -388,6 +416,12 @@ class AgentCoordinator: self.metadata = {aid: dict(md) for aid, md in snap.get("metadata", {}).items()} self.pending_counts = dict(snap.get("pending_counts", {})) self.errors = dict(snap.get("errors", {})) + mailboxes = snap.get("mailboxes", {}) + if isinstance(mailboxes, dict): + for aid, msgs in mailboxes.items(): + if isinstance(msgs, list): + runtime = self.runtimes.setdefault(aid, AgentRuntime()) + runtime.mailbox = [dict(m) for m in msgs if isinstance(m, dict)] self._budget_stopped = bool(snap.get("budget_stopped", False)) self._reserve_stopped = bool(snap.get("reserve_stopped", False)) self._budget_paused = bool(snap.get("budget_paused", False)) diff --git a/strix/core/execution.py b/strix/core/execution.py index 7272a8fe..762fcd3f 100644 --- a/strix/core/execution.py +++ b/strix/core/execution.py @@ -9,6 +9,7 @@ import uuid from collections.abc import Callable from typing import TYPE_CHECKING, Any, cast +import litellm from agents import RunConfig, Runner from agents.exceptions import AgentsException, MaxTurnsExceeded, UserError from agents.sandbox.errors import ExecTransportError @@ -16,9 +17,7 @@ from docker import errors as docker_errors # type: ignore[import-untyped, unuse from openai import ( APIConnectionError, APIError, - APIStatusError, APITimeoutError, - RateLimitError, ) from strix.config import codex @@ -31,6 +30,8 @@ from strix.core.inputs import child_initial_input from strix.core.sessions import ( enforce_image_budget, open_agent_session, + replace_session_items, + seed_initial_input, strip_all_images_from_session, ) from strix.llm.compaction import is_context_overflow, maybe_compact @@ -106,15 +107,9 @@ 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 +_MAX_TRANSIENT_MODEL_RETRIES = 5 _TRANSIENT_MODEL_RETRY_BASE_DELAY_S = 2.0 -_TRANSIENT_MODEL_RETRY_MAX_DELAY_S = 30.0 +_TRANSIENT_MODEL_RETRY_MAX_DELAY_S = 90.0 def _model_error_status_code(exc: BaseException) -> int | None: @@ -123,15 +118,16 @@ def _model_error_status_code(exc: BaseException) -> int | None: def _is_transient_model_error(exc: BaseException) -> bool: - if isinstance(exc, RateLimitError): + if codex.is_content_guardrail_error(exc): return False - if isinstance(exc, APITimeoutError | APIConnectionError): + if isinstance( + exc, APITimeoutError | APIConnectionError | TimeoutError | ConnectionError | OSError + ): return True - if isinstance(exc, APIStatusError): - return exc.status_code in _TRANSIENT_MODEL_STATUS_CODES - if isinstance(exc, APIError): - return _model_error_status_code(exc) is None - return False + code = _model_error_status_code(exc) + if code is not None: + return bool(litellm._should_retry(code)) + return isinstance(exc, APIError) def _transient_model_retry_delay(attempt: int) -> float: @@ -139,6 +135,40 @@ def _transient_model_retry_delay(attempt: int) -> float: return min(delay, _TRANSIENT_MODEL_RETRY_MAX_DELAY_S) +async def _salvage_stream_to_session( + session: Session, + pre_run_items: list[Any], + stream: Any, + agent_id: str, +) -> None: + """Persist a crashed run's full history so a revived agent loses no context.""" + if stream is None: + return + try: + replay = list(stream.to_input_list()) + except Exception: + logger.exception("could not build salvage history for %s", agent_id) + return + desired = list(pre_run_items) + replay + if len(desired) <= len(pre_run_items): + return + try: + await replace_session_items(session, desired) + except Exception: + logger.exception("salvaging crashed run history failed for %s", agent_id) + + +async def _seed_and_prepare_first_input( + session: Session | None, initial_input: Any, *, start_parked: bool +) -> Any: + """Persist the opening input up front so it survives a first-turn crash.""" + if initial_input and session is not None and not start_parked: + with contextlib.suppress(Exception): + if await seed_initial_input(session, initial_input): + return [] + return initial_input + + async def run_agent_loop( *, agent: Any, @@ -161,6 +191,10 @@ async def run_agent_loop( ) result: RunResultBase | None = None + first_cycle_input = await _seed_and_prepare_first_input( + session, initial_input, start_parked=start_parked + ) + budget_stopped = coordinator.budget_stopped reserve_stopped = coordinator.reserve_stopped if budget_stopped: @@ -176,16 +210,15 @@ async def run_agent_loop( if not (start_parked and interactive): if interactive: with contextlib.suppress(BudgetPausedError): - result = await _run_cycle( + result = await _run_cycle_parked( agent, coordinator, agent_id, - input_data=initial_input, + input_data=first_cycle_input, run_config=run_config, context=context, max_turns=max_turns, session=session, - interactive=interactive, event_sink=event_sink, hooks=hooks, ) @@ -194,7 +227,7 @@ async def run_agent_loop( agent, coordinator, agent_id, - initial_input=initial_input, + initial_input=first_cycle_input, run_config=run_config, context=context, max_turns=max_turns, @@ -207,8 +240,9 @@ async def run_agent_loop( return result while True: + timeout = await _plain_waiting_timeout(coordinator, agent_id, context) try: - await coordinator.wait_for_message(agent_id) + woke = await coordinator.wait_for_message(agent_id, timeout=timeout) except asyncio.CancelledError: return result @@ -220,9 +254,21 @@ async def run_agent_loop( await coordinator.set_status(agent_id, "stopped") raise SubagentBudgetReservedError("scan reached the sub-agent budget reserve") + if not woke: + logger.info("agent %s reached its waiting timeout; auto-resuming", agent_id) + await coordinator.send( + agent_id, + { + "from": "system", + "type": "auto_resume", + "content": "Waiting timeout reached. Resuming execution.", + }, + interrupt=False, + ) + await coordinator.consume_pending(agent_id) with contextlib.suppress(BudgetPausedError): - result = await _run_cycle( + result = await _run_cycle_parked( agent, coordinator, agent_id, @@ -231,7 +277,6 @@ async def run_agent_loop( context=context, max_turns=max_turns, session=session, - interactive=interactive, event_sink=event_sink, hooks=hooks, ) @@ -327,7 +372,6 @@ 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, @@ -340,8 +384,7 @@ async def respawn_subagents( for child_id, name, parent_id, md in candidates: try: restored_status = str(md.get("_restored_status") or "running") - recoverable_park = restored_status == "waiting" and bool(md.get("_restored_error")) - start_parked = interactive and restored_status != "running" and not recoverable_park + start_parked = interactive and restored_status != "running" if start_parked: logger.warning( @@ -456,6 +499,64 @@ async def _run_noninteractive_until_lifecycle( ) +_WAITING_AUTO_RESUME_TIMEOUT_S = 600.0 + + +async def _plain_waiting_timeout( + coordinator: AgentCoordinator, + agent_id: str, + context: dict[str, Any], +) -> float | None: + """Auto-resume timeout for a plainly-waiting subagent; None waits forever.""" + if context.get("parent_id") is None: + return None + async with coordinator._lock: + status = coordinator.statuses.get(agent_id) + has_error = agent_id in coordinator.errors + runtime = coordinator.runtimes.get(agent_id) + gated = runtime.user_wake_required if runtime is not None else False + if status == "waiting" and not has_error and not gated: + return _WAITING_AUTO_RESUME_TIMEOUT_S + return None + + +async def _run_cycle_parked( + agent: Any, + coordinator: AgentCoordinator, + agent_id: str, + *, + input_data: Any, + run_config: RunConfig, + context: dict[str, Any], + max_turns: int, + session: Session | None, + event_sink: StreamEventSink | None, + hooks: RunHooks[dict[str, Any]] | None, +) -> RunResultBase | None: + """Interactive run cycle that parks on any error instead of killing the runner.""" + try: + return await _run_cycle( + agent, + coordinator, + agent_id, + input_data=input_data, + run_config=run_config, + context=context, + max_turns=max_turns, + session=session, + interactive=True, + event_sink=event_sink, + hooks=hooks, + ) + except (BudgetExceededError, BudgetPausedError, SubagentBudgetReservedError): + raise + 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") + return None + + async def _run_cycle( # noqa: PLR0912, PLR0915 agent: Any, coordinator: AgentCoordinator, @@ -474,6 +575,8 @@ async def _run_cycle( # noqa: PLR0912, PLR0915 compactions = 0 model_retries = 0 while True: + stream: Any = None + pre_run_items: list[Any] = [] try: await coordinator.mark_running(agent_id) if session is not None: @@ -487,6 +590,8 @@ async def _run_cycle( # noqa: PLR0912, PLR0915 await _compact_session(agent, session, run_config, force=False) except Exception: logger.exception("proactive compaction failed for %s", agent_id) + with contextlib.suppress(Exception): + pre_run_items = list(await session.get_items()) stream = Runner.run_streamed( agent, input=input_data, @@ -599,10 +704,8 @@ 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 session is not None: + await _salvage_stream_to_session(session, pre_run_items, stream, agent_id) 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)) @@ -622,23 +725,7 @@ async def _run_cycle( # noqa: PLR0912, PLR0915 return None else: await _settle_run_result(coordinator, agent_id, interactive) - 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 + return cast("RunResultBase | None", stream) async def _settle_run_result( diff --git a/strix/core/runner.py b/strix/core/runner.py index 01725cab..ec439866 100644 --- a/strix/core/runner.py +++ b/strix/core/runner.py @@ -377,12 +377,6 @@ 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, @@ -394,7 +388,7 @@ async def run_strix_scan( agent_id=root_id, interactive=interactive, session=root_session, - start_parked=root_start_parked, + start_parked=bool(interactive and is_resume and root_status != "running"), event_sink=event_sink, hooks=hooks, ) diff --git a/strix/core/sessions.py b/strix/core/sessions.py index 2bb83982..8879d359 100644 --- a/strix/core/sessions.py +++ b/strix/core/sessions.py @@ -7,6 +7,7 @@ import logging from typing import TYPE_CHECKING, Any, cast from weakref import WeakKeyDictionary +from agents.items import ItemHelpers from agents.memory import SQLiteSession @@ -26,6 +27,18 @@ def open_agent_session(agent_id: str, path: Path) -> SQLiteSession: return SQLiteSession(session_id=agent_id, db_path=path) +async def seed_initial_input(session: Session, initial_input: Any) -> bool: + """Commit an agent's opening identity/task input before its first run cycle.""" + items = ItemHelpers.input_to_new_input_list(initial_input) + if not items: + return False + async with session_write_lock(session): + if await session.get_items(): + return False + await session.add_items(items) + return True + + _IMAGE_REJECTED_TEXT = "[image rejected by the model]" _IMAGE_ELIDED_TEXT = "[older screenshot elided to bound context memory]" _INHERITED_IMAGE_TEXT = "[screenshot omitted from inherited context]" diff --git a/strix/interface/tui/app.py b/strix/interface/tui/app.py index 14bc6cb1..21715d0c 100644 --- a/strix/interface/tui/app.py +++ b/strix/interface/tui/app.py @@ -1041,9 +1041,9 @@ class StrixTUIApp(App): # type: ignore[misc] name=names.get(agent_id, agent_id), parent_id=parent_of.get(agent_id), status=status, - error_message=error, + error_message=error or "", ) - if status in {"failed", "crashed"} and error: + if error: if agent_id not in self._error_noted_agents: self._error_noted_agents.add(agent_id) self.live_view.record_agent_error(agent_id, error) @@ -1293,6 +1293,10 @@ class StrixTUIApp(App): # type: ignore[misc] text.append("Send a message to continue", style="dim") keymap = keymap_styled([("ctrl-q", "quit")]) else: + error_msg = agent_data.get("error_message") or "" + if error_msg: + text.append(error_msg, style="red") + text.append(" \u00b7 ", style="dim") text.append("Send message to resume", style="dim") return (text, keymap, False) diff --git a/strix/interface/tui/live_view.py b/strix/interface/tui/live_view.py index 4401fb30..11a4b033 100644 --- a/strix/interface/tui/live_view.py +++ b/strix/interface/tui/live_view.py @@ -82,7 +82,7 @@ class TuiLiveView: current["parent_id"] = parent_id if status is not None: current["status"] = status - if error_message: + if error_message is not None: current["error_message"] = error_message current["updated_at"] = now diff --git a/tests/test_execution.py b/tests/test_execution.py index 8fa043f1..c1a13a72 100644 --- a/tests/test_execution.py +++ b/tests/test_execution.py @@ -5,7 +5,7 @@ from __future__ import annotations import asyncio import contextlib import json -from typing import Any +from typing import Any, cast from unittest.mock import MagicMock import pytest @@ -14,18 +14,19 @@ from agents.memory import SQLiteSession from agents.tool_context import ToolContext from openai.types.responses import ResponseOutputMessage, ResponseOutputRefusal -from strix.config import codex from strix.core import execution from strix.core.agents import AgentCoordinator from strix.core.execution import ( - _handle_content_guardrail, _notify_parent_on_terminal, _notify_root_on_budget_reserve, - respawn_subagents, ) +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 @@ -43,8 +44,8 @@ class _StructuredRefusalStream: ] async def stream_events(self) -> Any: - if False: - yield None + for event in _NO_STREAM_EVENTS: + yield event def cancel(self, mode: str = "immediate") -> None: # noqa: ARG002 return @@ -529,44 +530,152 @@ async def test_terminal_notice_does_not_cancel_parent_stream(tmp_path: Any) -> N @pytest.mark.asyncio -async def test_guardrail_interactive_parks_agent_wakeable(tmp_path: Any) -> None: +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) - 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 await coordinator.send("root", {"from": "user", "content": "hello"}) is True + assert coordinator.pending_counts["root"] == 1 - assert result is None - assert coordinator.statuses["child"] == "waiting" - assert "STRIX_LLM" in coordinator.errors["child"] + session = SQLiteSession("root", tmp_path / "agents.db") + await coordinator.attach_runtime("root", session=session) - 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) + 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_guardrail_noninteractive_fails_only_blocked_agent(tmp_path: Any) -> None: +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("root", tmp_path / "agents.db") - await coordinator.attach_runtime("root", session=session) - exc = codex.CodexContentGuardrailError("chatgpt/gpt-5.6-sol") + session = SQLiteSession("child", tmp_path / "agents.db") + await coordinator.attach_runtime("child", session=session) + await coordinator.set_status("child", "crashed", error="boom") - result = await _handle_content_guardrail(coordinator, "child", exc, interactive=False) + 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["child"] == "failed" - assert "STRIX_LLM" in coordinator.errors["child"] - assert coordinator.statuses["root"] == "running" - assert coordinator.pending_counts.get("root", 0) > 0 + 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() @@ -576,7 +685,9 @@ async def test_structured_provider_refusal_fails_interactive_agent( ) -> None: refusal = "This request was blocked under the provider's usage policy." stream = _StructuredRefusalStream(refusal) - monkeypatch.setattr(execution.Runner, "run_streamed", lambda *_args, **_kwargs: stream) + monkeypatch.setattr( + "strix.core.execution.Runner.run_streamed", lambda *_args, **_kwargs: stream + ) coordinator = AgentCoordinator() await coordinator.register("root", "strix", parent_id=None) @@ -606,7 +717,9 @@ async def test_structured_provider_refusal_fails_noninteractive_child( ) -> None: refusal = "This request was blocked under the provider's usage policy." stream = _StructuredRefusalStream(refusal) - monkeypatch.setattr(execution.Runner, "run_streamed", lambda *_args, **_kwargs: stream) + 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") @@ -635,34 +748,40 @@ async def test_structured_provider_refusal_fails_noninteractive_child( @pytest.mark.asyncio -async def test_resume_revives_guardrail_parked_child_but_not_plain_waiting( +async def test_run_agent_loop_seeds_identity_before_first_cycle( 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") + await coordinator.register("child", "recon", parent_id="root") + session = SQLiteSession("child", tmp_path / "agents.db") - parked: dict[str, bool] = {} + captured: dict[str, Any] = {} - async def _fake_start_child_runner(**kwargs: Any) -> None: - parked[kwargs["child_id"]] = bool(kwargs["start_parked"]) + 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, "_start_child_runner", _fake_start_child_runner) + monkeypatch.setattr(execution, "_run_cycle", _crash_first_turn) - 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", - ) + 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, + ) - assert parked["blocked"] is False - assert parked["peer_waiter"] is True + # 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() diff --git a/tests/test_execution_transient_retry.py b/tests/test_execution_transient_retry.py index 889eb96f..d81e4e70 100644 --- a/tests/test_execution_transient_retry.py +++ b/tests/test_execution_transient_retry.py @@ -15,6 +15,7 @@ from openai import ( RateLimitError, ) +from strix.config import codex from strix.core import execution from strix.core.agents import AgentCoordinator @@ -55,11 +56,27 @@ def test_server_errors_are_transient() -> None: assert execution._is_transient_model_error(_status_error(status)) is True -def test_rate_limit_is_not_retried_here() -> None: +def test_rate_limit_is_retried() -> None: rate_limited = RateLimitError( "slow down", response=httpx.Response(429, request=_request()), body=None ) - assert execution._is_transient_model_error(rate_limited) is False + assert execution._is_transient_model_error(rate_limited) is True + + +def test_dns_and_connection_errors_are_transient() -> None: + assert execution._is_transient_model_error(OSError("nodename nor servname provided")) is True + assert execution._is_transient_model_error(ConnectionError("reset")) is True + assert execution._is_transient_model_error(TimeoutError("timed out")) is True + + +def test_content_guardrail_is_not_retried() -> None: + guardrail = APIError( + "This content was flagged for possible cybersecurity risk", + _request(), + body=None, + ) + assert codex.is_content_guardrail_error(guardrail) is True + assert execution._is_transient_model_error(guardrail) is False def test_client_errors_are_not_transient() -> None: