diff --git a/raven/providers/openai_codex_provider.py b/raven/providers/openai_codex_provider.py index b38e1770..b6e93921 100644 --- a/raven/providers/openai_codex_provider.py +++ b/raven/providers/openai_codex_provider.py @@ -14,6 +14,51 @@ DEFAULT_CODEX_URL = "https://chatgpt.com/backend-api/codex/responses" DEFAULT_ORIGINATOR = "raven" +_SSE_ERROR_TYPE_LIMIT = 128 +_SSE_ERROR_CODE_LIMIT = 128 +_SSE_ERROR_MESSAGE_LIMIT = 512 + + +def _bounded_error_field(value: Any, limit: int) -> str | None: + if not isinstance(value, str): + return None + value = " ".join("".join(char if char.isprintable() else " " for char in value).split()) + if not value: + return None + if len(value) > limit: + return value[: limit - 3] + "..." + return value + + +class _CodexSSEError(RuntimeError): + def __init__(self, payload: Any): + payload = payload if isinstance(payload, dict) else {} + self.error_type = _bounded_error_field(payload.get("type"), _SSE_ERROR_TYPE_LIMIT) + self.code = _bounded_error_field(payload.get("code"), _SSE_ERROR_CODE_LIMIT) + self.message = _bounded_error_field(payload.get("message"), _SSE_ERROR_MESSAGE_LIMIT) + + details = [ + f"{name}={value}" + for name, value in (("type", self.error_type), ("code", self.code), ("message", self.message)) + if value is not None + ] + summary = "Codex response failed" + if details: + summary += ": " + ", ".join(details) + super().__init__(summary) + + +class ServiceUnavailableError(_CodexSSEError): + """Codex capacity error recognized by the shared provider classifier.""" + + +def _codex_sse_error(payload: Any) -> _CodexSSEError: + if isinstance(payload, dict): + error_type = payload.get("type") + code = payload.get("code") + if error_type == "service_unavailable_error" or code == "server_is_overloaded": + return ServiceUnavailableError(payload) + return _CodexSSEError(payload) class OpenAICodexProvider(LLMProvider): @@ -372,7 +417,12 @@ async def _consume_sse(response: httpx.Response, timeout: float) -> tuple[str, l status = (event.get("response") or {}).get("status") finish_reason = _map_finish_reason(status) elif event_type in {"error", "response.failed"}: - raise RuntimeError("Codex response failed") + if event_type == "error": + error = event.get("error") + else: + response_payload = event.get("response") + error = response_payload.get("error") if isinstance(response_payload, dict) else None + raise _codex_sse_error(error) return content, tool_calls, finish_reason diff --git a/tests/test_openai_codex_provider.py b/tests/test_openai_codex_provider.py index d75b4175..a21c8484 100644 --- a/tests/test_openai_codex_provider.py +++ b/tests/test_openai_codex_provider.py @@ -8,13 +8,19 @@ from __future__ import annotations import asyncio +import json +from types import SimpleNamespace +import oauth_cli_kit import pytest +from raven.providers import openai_codex_provider as codex_module +from raven.providers.base import LLMProvider from raven.providers.openai_codex_provider import ( DEFAULT_CODEX_URL, OpenAICodexProvider, _build_headers, + _consume_sse, _convert_messages, _convert_tool_output, _iter_sse, @@ -53,6 +59,177 @@ async def aiter_lines(self): await asyncio.sleep(10) +def _sse_response(event: dict) -> _FakeStreamResponse: + return _FakeStreamResponse([f"data: {json.dumps(event)}", ""]) + + +_OVERLOADED_ERROR = { + "type": "service_unavailable_error", + "code": "server_is_overloaded", + "message": "Our servers are currently overloaded. Please try again later.", +} + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "event", + [ + {"type": "error", "error": _OVERLOADED_ERROR}, + {"type": "response.failed", "response": {"error": _OVERLOADED_ERROR}}, + ], +) +async def test_consume_sse_preserves_structured_error_details(event: dict): + with pytest.raises(RuntimeError) as raised: + await _consume_sse(_sse_response(event), timeout=0.05) + + message = str(raised.value) + assert "service_unavailable_error" in message + assert "server_is_overloaded" in message + assert _OVERLOADED_ERROR["message"] in message + + classification = LLMProvider.classify_error(raised.value) + assert classification.category == "server" + assert classification.retryable is True + assert classification.should_fallback is True + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "error", + [ + {"type": "service_unavailable_error"}, + {"code": "server_is_overloaded"}, + ], +) +async def test_consume_sse_classifies_sparse_capacity_errors_as_retryable(error: dict): + with pytest.raises(RuntimeError) as raised: + await _consume_sse(_sse_response({"type": "error", "error": error}), timeout=0.05) + + classification = LLMProvider.classify_error(raised.value) + assert classification.category == "server" + assert classification.retryable is True + assert classification.should_fallback is True + + +@pytest.mark.asyncio +async def test_consume_sse_bounds_error_fields_and_ignores_nested_values(): + event = { + "type": "error", + "error": { + "type": {"unexpected": "nested"}, + "code": "x" * 1_000, + "message": "line one\n\x1b[31m" + "m" * 2_000, + "request": {"authorization": "must not be serialized"}, + }, + } + + with pytest.raises(RuntimeError) as raised: + await _consume_sse(_sse_response(event), timeout=0.05) + + message = str(raised.value) + assert "unexpected" not in message + assert "authorization" not in message + assert "\n" not in message + assert "\x1b" not in message + assert "x" * 100 in message + assert "line one" in message + assert len(message) < 1_000 + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "event", + [ + {"type": "error", "error": "unstructured failure"}, + {"type": "response.failed", "response": None}, + ], +) +async def test_consume_sse_uses_generic_message_for_malformed_error(event: dict): + with pytest.raises(RuntimeError, match=r"^Codex response failed$"): + await _consume_sse(_sse_response(event), timeout=0.05) + + +@pytest.mark.asyncio +async def test_consume_sse_keeps_unrelated_error_non_retryable(): + event = { + "type": "error", + "error": { + "type": "invalid_request_error", + "code": "invalid_prompt", + "message": "Invalid request: malformed input", + }, + } + + with pytest.raises(RuntimeError) as raised: + await _consume_sse(_sse_response(event), timeout=0.05) + + classification = LLMProvider.classify_error(raised.value) + assert classification.category == "invalid_request" + assert classification.retryable is False + assert classification.should_fallback is False + + +def _patch_codex_token(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr( + oauth_cli_kit, + "get_token", + lambda: SimpleNamespace(account_id="acct-test", access="token-test"), + ) + + +@pytest.mark.asyncio +async def test_codex_sse_overload_retries_then_succeeds(monkeypatch: pytest.MonkeyPatch): + _patch_codex_token(monkeypatch) + calls = 0 + + async def request(*args, **kwargs): + nonlocal calls + calls += 1 + if calls == 1: + return await _consume_sse( + _sse_response({"type": "response.failed", "response": {"error": _OVERLOADED_ERROR}}), + timeout=0.05, + ) + return "recovered", [], "stop" + + monkeypatch.setattr(codex_module, "_request_codex", request) + provider = OpenAICodexProvider() + provider._CHAT_RETRY_DELAYS = (0,) + + response = await provider.chat_with_retry(messages=[], model=provider.default_model) + + assert response.content == "recovered" + assert response.finish_reason == "stop" + assert calls == 2 + + +@pytest.mark.asyncio +async def test_codex_sse_overload_exhausts_retry_ladder(monkeypatch: pytest.MonkeyPatch): + _patch_codex_token(monkeypatch) + calls = 0 + + async def request(*args, **kwargs): + nonlocal calls + calls += 1 + return await _consume_sse( + _sse_response({"type": "error", "error": _OVERLOADED_ERROR}), + timeout=0.05, + ) + + monkeypatch.setattr(codex_module, "_request_codex", request) + provider = OpenAICodexProvider() + provider._CHAT_RETRY_DELAYS = (0, 0, 0) + + response = await provider.chat_with_retry(messages=[], model=provider.default_model) + + assert response.finish_reason == "error" + assert response.error_classification is not None + assert response.error_classification.category == "server" + assert response.error_classification.retryable is True + assert "server_is_overloaded" in response.content + assert calls == 4 + + @pytest.mark.asyncio async def test_iter_sse_per_event_idle_timeout_raises(): """A stream that stalls after a complete event trips the per-event idle cap