Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
52 changes: 52 additions & 0 deletions agent_core/providers/openai_responses.py
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@

from openai import AsyncOpenAI

from agent_core.errors import LLMError
from agent_core.llm import LLMClient, LLMResponse, StreamDelta
from agent_core.messages import Message, ToolCall, text_of
from agent_core.providers._api_key import resolve_openai_api_key
Expand All @@ -45,6 +46,39 @@

logger = logging.getLogger(__name__)

# Equivalent HTTP statuses for the Responses SDK's ResponseError codes.
_RESPONSE_ERROR_STATUS_CODES = {
"rate_limit_exceeded": 429,
"server_error": 500,
"vector_store_timeout": 504,
"invalid_prompt": 400,
"data_residency_mismatch": 400,
"bio_policy": 400,
"invalid_image": 400,
"invalid_image_format": 400,
"invalid_base64_image": 400,
"invalid_image_url": 400,
"image_too_large": 400,
"image_too_small": 400,
"image_parse_error": 400,
"image_content_policy_violation": 400,
"invalid_image_mode": 400,
"image_file_too_large": 400,
"unsupported_image_media_type": 400,
"empty_image_file": 400,
"failed_to_download_image": 400,
"image_file_not_found": 400,
}


class _ResponsesError(LLMError):
"""A failed Responses result with signals for the shared retry classifier."""

def __init__(self, code: str, message: str) -> None:
super().__init__(f"{code}: {message}" if code else message)
self.code = code
self.status_code = _RESPONSE_ERROR_STATUS_CODES.get(code)


class OpenAIResponsesClient(LLMClient):
"""OpenAI Responses API adapter with encrypted-reasoning round-trip."""
Expand Down Expand Up @@ -162,6 +196,10 @@ async def stream(
"response.reasoning_text.delta",
):
yield StreamDelta(reasoning_content=getattr(event, "delta", "") or "")
elif etype == "response.failed":
raise _response_failure(
getattr(event, "response", None), fallback="Responses request failed",
)
elif etype in ("response.completed", "response.incomplete"):
resp = getattr(event, "response", None)
usage = _responses_usage_dict(getattr(resp, "usage", None))
Expand Down Expand Up @@ -323,7 +361,13 @@ def _parse_responses_output(raw: Any) -> LLMResponse:
``content`` is kept as a verbatim block list (reasoning items incl.
``encrypted_content`` + text blocks) so the ``content_block`` thinking
parser preserves them as ``raw_content_blocks`` for faithful replay.

A ``failed`` response raises instead of returning its partial output as an
answer: the caller would otherwise execute tool calls from a response the
server itself reported as failed.
"""
if str(_get(raw, "status", "") or "") == "failed":
raise _response_failure(raw, fallback="Responses request failed")
blocks_out: list[dict[str, Any]] = []
text_parts: list[str] = []
summary_parts: list[str] = []
Expand Down Expand Up @@ -403,6 +447,14 @@ def _parse_responses_output(raw: Any) -> LLMResponse:
)


def _response_failure(raw: Any, *, fallback: str) -> LLMError:
"""Preserve the provider's code and message for shared retry routing."""
error = _get(raw, "error", None)
return _ResponsesError(
str(_get(error, "code", "") or ""), _get(error, "message", "") or fallback,
)


def _responses_usage_dict(usage: Any) -> dict[str, int]:
"""Normalise a Responses ``usage`` object into the wire-shape token dict.

Expand Down
1 change: 1 addition & 0 deletions changes/responses-failed-error-routing.fix.md
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
Failed Responses results now raise instead of returning their partial output, so tool calls from a response the server reported as `failed` are never executed. The raised `LLMError` keeps the provider error code and message with an equivalent HTTP status: rate limits use 429, server errors use 500, vector-store timeouts use 504, and other known SDK codes use 400. Unknown or missing codes retain the existing generic retry behavior, while existing text-based fallback and transient-network rules still take precedence. Streaming `response.failed` events now raise the same error instead of being ignored.
162 changes: 162 additions & 0 deletions tests/test_responses_error_routing.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,162 @@
"""Failed Responses results retain error codes for the runtime's retry routing."""

from __future__ import annotations

import asyncio
from types import SimpleNamespace

import pytest

from agent_core.errors import LLMError
from agent_core.messages import user_msg
from agent_core.providers import openai_responses as rc
from agent_core.runtime.loop.llm_client import LLMCallExhausted, call_llm
from agent_core.runtime.retriable import is_retriable_with_fallback, is_transient_network

ERROR_CODES = [
("rate_limit_exceeded", 429),
("server_error", 500),
("vector_store_timeout", 504),
("invalid_prompt", 400),
("data_residency_mismatch", 400),
("bio_policy", 400),
("invalid_image", 400),
("invalid_image_format", 400),
("invalid_base64_image", 400),
("invalid_image_url", 400),
("image_too_large", 400),
("image_too_small", 400),
("image_parse_error", 400),
("image_content_policy_violation", 400),
("invalid_image_mode", 400),
("image_file_too_large", 400),
("unsupported_image_media_type", 400),
("empty_image_file", 400),
("failed_to_download_image", 400),
("image_file_not_found", 400),
("future_error", None),
("", None),
(None, None),
]


@pytest.mark.parametrize("code,status", ERROR_CODES)
@pytest.mark.parametrize("as_object", [False, True])
def test_failed_output_retains_error_code_and_status(code, status, as_object):
error = {"code": code, "message": "original failure message"}
raw = {"status": "failed", "error": error, "output": [{"type": "function_call"}]}
if as_object:
raw = SimpleNamespace(**{**raw, "error": SimpleNamespace(**error)})

with pytest.raises(LLMError) as caught:
rc._parse_responses_output(raw)

assert caught.value.code == (code or "")
assert caught.value.status_code == status
assert "original failure message" in str(caught.value)
if code:
assert code in str(caught.value)


@pytest.mark.parametrize("error", [None, {}, {"code": "invalid_prompt", "message": ""}])
def test_failed_output_without_message_uses_fallback(error):
with pytest.raises(LLMError, match="Responses request failed") as caught:
rc._parse_responses_output({"status": "failed", "error": error})
assert caught.value.code == (error or {}).get("code", "")
assert caught.value.status_code == (400 if error else None)


ROUTES = [
(
code, "request failed",
"chain_advance" if code == "image_content_policy_violation"
else "non_transient" if status == 400 else "exhausted",
30 if status == 429 else 2,
status in (500, 504),
)
for code, status in ERROR_CODES
] + [
("bio_policy", "Your request was blocked for safety reasons", "chain_advance", 2, False),
("invalid_prompt", "We've limited access to this content for safety reasons",
"chain_advance", 2, False),
("invalid_prompt", "upstream timeout", "exhausted", 2, True),
("server_error", "server overloaded", "chain_advance", 2, False),
]


@pytest.mark.asyncio
@pytest.mark.parametrize("code,message,reason,base,transient", ROUTES)
async def test_failed_output_routes_through_call_llm(
monkeypatch, code, message, reason, base, transient,
):
with pytest.raises(LLMError) as caught:
rc._parse_responses_output({
"status": "failed", "error": {"code": code, "message": message},
})
error = caught.value
sleeps = []
real_sleep = asyncio.sleep

async def record_sleep(duration):
sleeps.append(duration)
await real_sleep(0)

monkeypatch.setattr("asyncio.sleep", record_sleep)
calls = 0

async def chat(_messages, **_kwargs):
nonlocal calls
calls += 1
raise error

with pytest.raises(LLMCallExhausted) as exhausted:
await call_llm(
SimpleNamespace(chat=chat), [user_msg("hi")],
timeout=10, max_retries=3, turn=0, chain_fallback_active=lambda: True,
)

assert exhausted.value.reason == reason
assert exhausted.value.last_exc is error
backoffs = [duration for duration in sleeps if duration > 0]
if reason == "exhausted":
assert calls == 3
assert len(backoffs) == 2
for duration, expected in zip(backoffs, [base, base * 2], strict=True):
assert expected * 0.75 <= duration <= expected * 1.25
else:
assert calls == 1
assert backoffs == []
assert is_transient_network(error) is transient
assert is_retriable_with_fallback(error) is (reason == "chain_advance")


@pytest.mark.asyncio
@pytest.mark.parametrize("code,status", ERROR_CODES)
async def test_stream_failed_event_raises_same_error(code, status):
raw = SimpleNamespace(
status="failed", error=SimpleNamespace(code=code, message="stream failure"),
)

async def events():
yield SimpleNamespace(type="response.output_text.delta", delta="partial")
yield SimpleNamespace(type="response.failed", response=raw)
yield SimpleNamespace(type="response.completed", response=SimpleNamespace())

async def create(**kwargs):
assert kwargs["stream"] is True
return events()

client = rc.OpenAIResponsesClient("gpt-x", api_key="x")
client._client = SimpleNamespace(responses=SimpleNamespace(create=create))
deltas = []
with pytest.raises(LLMError) as streamed:
async for delta in client.stream([user_msg("hi")]):
deltas.append(delta)
with pytest.raises(LLMError) as parsed:
rc._parse_responses_output(raw)

assert len(deltas) == 1 and deltas[0].content == "partial"
assert type(streamed.value) is type(parsed.value)
assert streamed.value.code == (code or "")
assert streamed.value.status_code == status
assert str(streamed.value) == str(parsed.value)
Loading