diff --git a/strix/core/execution.py b/strix/core/execution.py index a0280937..dd9692d7 100644 --- a/strix/core/execution.py +++ b/strix/core/execution.py @@ -13,7 +13,13 @@ from agents import RunConfig, Runner from agents.exceptions import AgentsException, MaxTurnsExceeded, UserError from agents.sandbox.errors import ExecTransportError from docker import errors as docker_errors # type: ignore[import-untyped, unused-ignore] -from openai import APIError +from openai import ( + APIConnectionError, + APIError, + APIStatusError, + APITimeoutError, + RateLimitError, +) from strix.core.hooks import BudgetExceededError from strix.core.inputs import child_initial_input @@ -78,6 +84,52 @@ async def _compact_session( ) +# Retryable HTTP statuses for a model/provider call: request timeout + 5xx server errors. +# 429 is intentionally excluded here; a persistent rate limit is handled as a graceful, +# resumable scan stop in ``runner.run_strix_scan``. +_TRANSIENT_MODEL_STATUS_CODES = frozenset({408, 500, 502, 503, 504}) +_MAX_TRANSIENT_MODEL_RETRIES = 4 +_TRANSIENT_MODEL_RETRY_BASE_DELAY_S = 2.0 +_TRANSIENT_MODEL_RETRY_MAX_DELAY_S = 30.0 + + +def _model_error_status_code(exc: BaseException) -> int | None: + code = getattr(exc, "status_code", None) + return code if isinstance(code, int) else None + + +def _is_transient_model_error(exc: BaseException) -> bool: + """Return whether a model/provider error is a transient upstream failure worth replaying. + + Three families are treated as transient: + + * network-level faults reaching the provider (connection reset, read timeout); + * retryable HTTP statuses (request timeout + 5xx) surfaced as ``APIStatusError``; + * mid-stream provider errors, where the server injects an ``{"error": ...}`` frame into + an already-200 SSE stream and the OpenAI SDK raises a bare ``APIError`` with no status + code. The agents SDK cannot replay these once tokens have streamed, so an untreated + one propagates and crashes the whole scan. + + A persistent rate limit is deliberately excluded (handled as a graceful scan stop at the + runner level), as are permanent 4xx client errors (they carry a status code and are + surfaced as ``APIStatusError``). + """ + if isinstance(exc, RateLimitError): + return False + if isinstance(exc, APITimeoutError | APIConnectionError): + 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 + + +def _transient_model_retry_delay(attempt: int) -> float: + delay = _TRANSIENT_MODEL_RETRY_BASE_DELAY_S * float(2 ** (attempt - 1)) + return min(delay, _TRANSIENT_MODEL_RETRY_MAX_DELAY_S) + + async def run_agent_loop( *, agent: Any, @@ -387,6 +439,7 @@ async def _run_cycle( # noqa: PLR0912, PLR0915 ) -> RunResultBase | None: image_strips = 0 compactions = 0 + model_retries = 0 while True: try: await coordinator.mark_running(agent_id) @@ -488,6 +541,24 @@ async def _run_cycle( # noqa: PLR0912, PLR0915 ) input_data = [] continue + if model_retries < _MAX_TRANSIENT_MODEL_RETRIES and _is_transient_model_error(exc): + model_retries += 1 + delay = _transient_model_retry_delay(model_retries) + logger.warning( + "transient model/provider error for %s; replaying turn " + "(attempt %d/%d, backoff %.1fs): %r", + agent_id, + model_retries, + _MAX_TRANSIENT_MODEL_RETRIES, + delay, + exc, + ) + await asyncio.sleep(delay) + # The turn's input was already persisted to the session before the model + # call, so replay from session state to avoid duplicating it. + if session is not None: + input_data = [] + continue if not interactive: raise if isinstance(exc, MaxTurnsExceeded): diff --git a/tests/test_execution_transient_retry.py b/tests/test_execution_transient_retry.py new file mode 100644 index 00000000..790454fb --- /dev/null +++ b/tests/test_execution_transient_retry.py @@ -0,0 +1,171 @@ +"""Tests for transient model/provider error replay in the agent run cycle. + +A transient upstream error (e.g. a mid-stream ``openai.APIError`` injected into an +already-200 SSE stream) must be retried at the turn level instead of crashing the +whole scan, while permanent errors and rate limits keep their existing behavior. +""" + +from __future__ import annotations + +from typing import Any, cast + +import httpx +import pytest +from agents import RunConfig, Runner +from openai import ( + APIConnectionError, + APIError, + APIStatusError, + APITimeoutError, + BadRequestError, + InternalServerError, + RateLimitError, +) + +from strix.core import execution +from strix.core.agents import AgentCoordinator + + +def _request() -> httpx.Request: + return httpx.Request("POST", "https://api.openai.com/v1/responses") + + +def _midstream_api_error() -> APIError: + """A bare APIError with no status code — the mid-stream generic provider error.""" + return APIError("An error occurred while processing the request.", _request(), body=None) + + +def _status_error(status: int) -> APIStatusError: + return APIStatusError( + f"status {status}", + response=httpx.Response(status_code=status, request=_request()), + body=None, + ) + + +def test_midstream_api_error_is_transient() -> None: + assert execution._is_transient_model_error(_midstream_api_error()) is True + + +def test_network_errors_are_transient() -> None: + assert execution._is_transient_model_error(APITimeoutError(_request())) is True + assert execution._is_transient_model_error(APIConnectionError(request=_request())) is True + + +def test_server_errors_are_transient() -> None: + assert ( + execution._is_transient_model_error( + InternalServerError("boom", response=httpx.Response(500, request=_request()), body=None) + ) + is True + ) + for status in (502, 503, 504, 408): + assert execution._is_transient_model_error(_status_error(status)) is True + + +def test_rate_limit_is_not_retried_here() -> None: + # A persistent rate limit is handled as a graceful scan stop at the runner level. + rate_limited = RateLimitError( + "slow down", response=httpx.Response(429, request=_request()), body=None + ) + assert execution._is_transient_model_error(rate_limited) is False + + +def test_client_errors_are_not_transient() -> None: + bad_request = BadRequestError( + "bad", response=httpx.Response(400, request=_request()), body=None + ) + assert execution._is_transient_model_error(bad_request) is False + assert execution._is_transient_model_error(_status_error(404)) is False + assert execution._is_transient_model_error(ValueError("nope")) is False + + +class _FakeStream: + """Minimal stand-in for a streamed run result.""" + + def __init__(self, exc: BaseException | None = None) -> None: + self._exc = exc + self._events: list[Any] = [] + self.run_loop_exception: BaseException | None = None + + async def stream_events(self) -> Any: + if self._exc is not None: + raise self._exc + for event in self._events: + yield event + + +def _patch_fast_backoff(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(execution, "_TRANSIENT_MODEL_RETRY_BASE_DELAY_S", 0.0) + monkeypatch.setattr(execution, "_TRANSIENT_MODEL_RETRY_MAX_DELAY_S", 0.0) + + +async def _run_once( + monkeypatch: pytest.MonkeyPatch, + streams: list[_FakeStream], +) -> Any: + _patch_fast_backoff(monkeypatch) + calls = {"n": 0} + + def _fake_run_streamed(*_args: Any, **_kwargs: Any) -> _FakeStream: + stream = streams[calls["n"]] + calls["n"] += 1 + return stream + + monkeypatch.setattr(Runner, "run_streamed", _fake_run_streamed) + + coordinator = AgentCoordinator() + await coordinator.register("root", "strix", parent_id=None) + + result = await execution._run_cycle( + object(), + coordinator, + "root", + input_data="task", + run_config=cast("RunConfig", object()), + context={}, + max_turns=5, + session=None, + interactive=False, + event_sink=None, + hooks=None, + ) + return result, calls["n"], coordinator + + +@pytest.mark.asyncio +async def test_run_cycle_retries_transient_midstream_error( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A transient mid-stream error is replayed and the turn then succeeds.""" + streams = [_FakeStream(exc=_midstream_api_error()), _FakeStream()] + result, attempts, _coordinator = await _run_once(monkeypatch, streams) + + assert result is streams[1] + assert attempts == 2 + + +@pytest.mark.asyncio +async def test_run_cycle_gives_up_after_max_retries( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Persistent transient errors exhaust the bound, then propagate (non-interactive).""" + streams = [ + _FakeStream(exc=_midstream_api_error()) + for _ in range(execution._MAX_TRANSIENT_MODEL_RETRIES + 1) + ] + with pytest.raises(APIError): + await _run_once(monkeypatch, streams) + + +@pytest.mark.asyncio +async def test_run_cycle_does_not_retry_permanent_error( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A permanent client error is not replayed; it propagates on the first attempt.""" + bad_request = BadRequestError( + "bad", response=httpx.Response(400, request=_request()), body=None + ) + streams = [_FakeStream(exc=bad_request), _FakeStream()] + with pytest.raises(BadRequestError): + await _run_once(monkeypatch, streams)