diff --git a/python/packages/openai/agent_framework_openai/_chat_client.py b/python/packages/openai/agent_framework_openai/_chat_client.py index a5c18ebda85..fdb8c3d8eae 100644 --- a/python/packages/openai/agent_framework_openai/_chat_client.py +++ b/python/packages/openai/agent_framework_openai/_chat_client.py @@ -87,6 +87,7 @@ from openai.types.responses import ( FunctionShellToolParam, ResponseCustomToolCall, + ResponseErrorEvent, ResponseFunctionShellToolCall, ResponseFunctionShellToolCallOutput, ResponseFunctionToolCallOutputItem, @@ -163,6 +164,25 @@ class _PromptCacheOptions(TypedDict, total=False): _TERMINAL_RESPONSE_STATUSES: Final[frozenset[str]] = frozenset({"completed", "failed", "incomplete", "cancelled"}) +class _OpenAIResponseStreamError(RuntimeError): + """Retain the provider metadata from a flat Responses API error event.""" + + def __init__(self, event: ResponseErrorEvent) -> None: + self.event = event + self.code = event.code + self.message = event.message + self.param = event.param + error_details = event.message + if event.code: + error_details = f"{event.code}: {error_details}" + if event.param: + error_details = f"{error_details} (parameter: {event.param})" + super().__init__(f"OpenAI streaming error: {error_details}") + + def __reduce__(self) -> tuple[type[_OpenAIResponseStreamError], tuple[ResponseErrorEvent], dict[str, Any]]: + return type(self), (self.event,), self.__dict__ + + def _is_refusal_text_content(content: Content) -> bool: return ( content.type == "text" @@ -3936,12 +3956,7 @@ def parse_annotations( # ResponseCustomToolCallInputDeltaEvent, # ResponseCustomToolCallInputDoneEvent, case "error": - error_details = event.message - if event.code: - error_details = f"{event.code}: {error_details}" - if event.param: - error_details = f"{error_details} (parameter: {event.param})" - self._handle_request_error(RuntimeError(f"OpenAI streaming error: {error_details}")) + self._handle_request_error(_OpenAIResponseStreamError(event)) case "response.content_part.added": event_part = event.part match event_part.type: diff --git a/python/packages/openai/tests/openai/test_openai_chat_client.py b/python/packages/openai/tests/openai/test_openai_chat_client.py index 15cadc9d9b9..ca9870e1ae3 100644 --- a/python/packages/openai/tests/openai/test_openai_chat_client.py +++ b/python/packages/openai/tests/openai/test_openai_chat_client.py @@ -2,10 +2,13 @@ import asyncio import base64 +import copy import inspect import json import os -from collections.abc import AsyncGenerator, AsyncIterable, Awaitable, Iterator, Mapping, Sequence +import pickle +import sys +from collections.abc import AsyncGenerator, AsyncIterable, Awaitable, Callable, Iterator, Mapping, Sequence from datetime import datetime, timezone from importlib import import_module from pathlib import Path @@ -9750,6 +9753,88 @@ def test_streaming_error_event_raises_chat_client_exception() -> None: assert "OpenAI streaming error: server_error: The response stream failed. (parameter: input)" in str(exc_info.value) +@pytest.mark.parametrize("request_path", ["create", "structured", "resume"]) +@pytest.mark.parametrize("code", ["server_error", "rate_limit_exceeded", None]) +@pytest.mark.parametrize("error_param", ["input", None]) +async def test_streaming_error_event_preserves_provider_details( + request_path: str, code: str | None, error_param: str | None +) -> None: + client = OpenAIChatClient(model="test-model", api_key="test-key") + error_event = ResponseErrorEvent( + code=code, + message="The response stream failed.", + param=error_param, + sequence_number=2, + type="error", + ) + events = _FakeAsyncEventStream([error_event]) + options: OpenAIChatOptions[Any] = {} + if request_path == "structured": + options["response_format"] = OutputStruct + request_patch = patch.object(client.client.responses, "stream", return_value=events) + else: + method = "retrieve" if request_path == "resume" else "create" + request_patch = patch.object(client.client.responses.with_raw_response, method, return_value=events) + if request_path == "resume": + options["continuation_token"] = {"response_id": "resp_stream_error"} + + with request_patch, pytest.raises(ChatClientException) as exc_info: + stream = client._inner_get_response( + messages=[Message(role="user", contents=["test"])], stream=True, options=options + ) + assert isinstance(stream, ResponseStream) + async for _ in stream: + pytest.fail("An error event must not yield a response update.") + + cause = exc_info.value.__cause__ + assert isinstance(cause, RuntimeError) + assert getattr(cause, "code", None) == code + assert getattr(cause, "message", None) == error_event.message + assert getattr(cause, "param", None) == error_param + assert getattr(cause, "event", None) is error_event + if request_path == "resume": + assert "continuation_token" not in options + + +@pytest.mark.parametrize( + "round_trip", + [ + param(copy.copy, id="copy"), + param(copy.deepcopy, id="deepcopy"), + param(lambda value: pickle.loads(pickle.dumps(value)), id="pickle"), + ], +) +@pytest.mark.parametrize("wrapped", [False, True], ids=["inner", "wrapper"]) +def test_streaming_error_exception_round_trip_preserves_state(round_trip: Callable[[Any], Any], wrapped: bool) -> None: + client = OpenAIChatClient(model="test-model", api_key="test-key") + error_event = ResponseErrorEvent( + code="server_error", message="boom", param="input", sequence_number=1, type="error" + ) + with pytest.raises(ChatClientException) as exc_info: + client._parse_chunk_from_openai(error_event, options={}, function_call_ids={}) + + cause = exc_info.value.__cause__ + assert isinstance(cause, RuntimeError) + cause.__dict__["trace_id"] = "trace-123" + if sys.version_info >= (3, 11): + cause.add_note("provider stream failed") + + error = exc_info.value if wrapped else cause + restored = round_trip(error) + assert type(restored) is type(error) + assert str(restored) == str(error) + restored_cause = restored.args[1] if wrapped else restored + assert type(restored_cause) is type(cause) + assert restored_cause.args == cause.args + assert getattr(restored_cause, "code", None) == error_event.code + assert getattr(restored_cause, "message", None) == error_event.message + assert getattr(restored_cause, "param", None) == error_event.param + assert getattr(restored_cause, "event", None) == error_event + assert restored_cause.__dict__["trace_id"] == "trace-123" + if sys.version_info >= (3, 11): + assert restored_cause.__notes__ == ["provider stream failed"] + + async def test_resumed_stream_error_clears_continuation_token() -> None: """A failed resumed stream should clear its caller-owned continuation token.""" client = OpenAIChatClient(model="test-model", api_key="test-key")