fix(providers): preserve Codex SSE error details

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
 

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