fix: preserve responses terminal stream events

Author: Aphroq

src/agents/exceptions.py

@@ -16,6 +16,16 @@
 
 from .util._pretty_print import pretty_print_run_error_details
 
+_DRAIN_STREAM_EVENTS_ATTR = "_agents_drain_queued_stream_events"
+
+
+def _mark_error_to_drain_stream_events(error: Exception) -> None:
+    setattr(error, _DRAIN_STREAM_EVENTS_ATTR, True)
+
+
+def _should_drain_stream_events_before_raising(error: Exception) -> bool:
+    return bool(getattr(error, _DRAIN_STREAM_EVENTS_ATTR, False))
+
 
 @dataclass
 class RunErrorDetails:

src/agents/extensions/models/any_llm_model.py

@@ -29,6 +29,10 @@
 from ...logger import logger
 from ...model_settings import ModelSettings
 from ...models._openai_retry import get_openai_retry_advice
+from ...models._response_terminal import (
+    response_error_event_failure_error,
+    response_terminal_failure_error,
+)
 from ...models._retry_runtime import should_disable_provider_managed_retries
 from ...models.chatcmpl_converter import Converter
 from ...models.chatcmpl_helpers import HEADERS, HEADERS_OVERRIDE, ChatCmplHelpers
@@ -421,18 +425,30 @@ async def _stream_response_via_responses(
             )
 
             final_response: Response | None = None
+            terminal_failure_error: ModelBehaviorError | None = None
             try:
                 async for chunk in stream:
+                    chunk_type = getattr(chunk, "type", None)
                     if isinstance(chunk, ResponseCompletedEvent):
                         final_response = chunk.response
-                    elif getattr(chunk, "type", None) in {"response.failed", "response.incomplete"}:
+                    elif chunk_type in {"response.failed", "response.incomplete"}:
                         terminal_response = getattr(chunk, "response", None)
-                        if isinstance(terminal_response, Response):
-                            final_response = terminal_response
+                        terminal_failure_error = response_terminal_failure_error(
+                            cast(str, chunk_type),
+                            terminal_response if isinstance(terminal_response, Response) else None,
+                        )
+                    elif chunk_type in {"error", "response.error"}:
+                        terminal_failure_error = response_error_event_failure_error(
+                            cast(str, chunk_type),
+                            chunk,
+                        )
                     yield chunk
             finally:
                 await self._maybe_aclose(stream)
 
+            if terminal_failure_error is not None:
+                raise terminal_failure_error
+
             if tracing.include_data() and final_response:
                 span_response.span_data.response = final_response
                 span_response.span_data.input = input

src/agents/models/_response_terminal.py

@@ -1,7 +1,11 @@
 from __future__ import annotations
 
+from typing import Any
+
 from openai.types.responses import Response
 
+from ..exceptions import ModelBehaviorError, _mark_error_to_drain_stream_events
+
 
 def format_response_terminal_failure(
     event_type: str,
@@ -25,3 +29,36 @@ def format_response_terminal_failure(
     if details:
         message = f"{message} {'; '.join(details)}."
     return message
+
+
+def format_response_error_event(event_type: str, event: Any) -> str:
+    message = f"Responses stream ended with terminal event `{event_type}`."
+    details: list[str] = []
+    code = getattr(event, "code", None)
+    if code:
+        details.append(f"code={code}")
+    error_message = getattr(event, "message", None)
+    if error_message:
+        details.append(f"message={error_message}")
+    param = getattr(event, "param", None)
+    if param:
+        details.append(f"param={param}")
+
+    if details:
+        message = f"{message} {'; '.join(details)}."
+    return message
+
+
+def response_terminal_failure_error(
+    event_type: str,
+    response: Response | None,
+) -> ModelBehaviorError:
+    error = ModelBehaviorError(format_response_terminal_failure(event_type, response))
+    _mark_error_to_drain_stream_events(error)
+    return error
+
+
+def response_error_event_failure_error(event_type: str, event: Any) -> ModelBehaviorError:
+    error = ModelBehaviorError(format_response_error_event(event_type, event))
+    _mark_error_to_drain_stream_events(error)
+    return error

src/agents/models/openai_responses.py

@@ -68,7 +68,7 @@
 from ..util._json import _to_dump_compatible
 from ..version import __version__
 from ._openai_retry import get_openai_retry_advice
-from ._response_terminal import format_response_terminal_failure
+from ._response_terminal import response_error_event_failure_error, response_terminal_failure_error
 from ._retry_runtime import (
     should_disable_provider_managed_retries,
     should_disable_websocket_pre_event_retries,
@@ -569,18 +569,22 @@ async def stream_response(
                             "response.incomplete",
                         }:
                             terminal_response = getattr(chunk, "response", None)
-                            terminal_failure_error = ModelBehaviorError(
-                                format_response_terminal_failure(
-                                    cast(str, chunk_type),
-                                    terminal_response
-                                    if isinstance(terminal_response, Response)
-                                    else None,
-                                )
+                            terminal_failure_error = response_terminal_failure_error(
+                                cast(str, chunk_type),
+                                terminal_response
+                                if isinstance(terminal_response, Response)
+                                else None,
+                            )
+                        elif chunk_type in {"error", "response.error"}:
+                            terminal_failure_error = response_error_event_failure_error(
+                                cast(str, chunk_type),
+                                chunk,
                             )
                         if chunk_type in {
                             "response.completed",
                             "response.failed",
                             "response.incomplete",
+                            "error",
                             "response.error",
                         }:
                             yielded_terminal_event = True
@@ -1099,11 +1103,9 @@ async def _fetch_response(
             elif event_type in {"response.incomplete", "response.failed"}:
                 terminal_event_type = cast(str, event_type)
                 terminal_response = getattr(event, "response", None)
-                raise ModelBehaviorError(
-                    format_response_terminal_failure(
-                        terminal_event_type,
-                        terminal_response if isinstance(terminal_response, Response) else None,
-                    )
+                raise response_terminal_failure_error(
+                    terminal_event_type,
+                    terminal_response if isinstance(terminal_response, Response) else None,
                 )
 
         if final_response is None:

src/agents/result.py

@@ -18,6 +18,7 @@
     InputGuardrailTripwireTriggered,
     MaxTurnsExceeded,
     RunErrorDetails,
+    _should_drain_stream_events_before_raising,
 )
 from .guardrail import InputGuardrailResult, OutputGuardrailResult
 from .items import (
@@ -705,7 +706,12 @@ async def stream_events(self) -> AsyncIterator[StreamEvent]:
         try:
             while True:
                 self._check_errors()
-                should_drain_queued_events = isinstance(self._stored_exception, MaxTurnsExceeded)
+                should_drain_queued_events = isinstance(
+                    self._stored_exception, MaxTurnsExceeded
+                ) or (
+                    self._stored_exception is not None
+                    and _should_drain_stream_events_before_raising(self._stored_exception)
+                )
                 if self._stored_exception and (
                     not should_drain_queued_events or self._event_queue.empty()
                 ):

src/agents/run_internal/run_loop.py

@@ -58,7 +58,10 @@
 from ..lifecycle import RunHooks
 from ..logger import logger
 from ..memory import Session
-from ..models._response_terminal import format_response_terminal_failure
+from ..models._response_terminal import (
+    response_error_event_failure_error,
+    response_terminal_failure_error,
+)
 from ..result import RunResultStreaming
 from ..run_config import ReasoningItemIdPolicy, RunConfig
 from ..run_context import AgentHookContext, RunContextWrapper, TContext
@@ -1487,12 +1490,12 @@ async def rewind_model_request() -> None:
         elif getattr(event, "type", None) in {"response.incomplete", "response.failed"}:
             event_type = cast(str, event.type)
             maybe_response = getattr(event, "response", None)
-            raise ModelBehaviorError(
-                format_response_terminal_failure(
-                    event_type,
-                    maybe_response if isinstance(maybe_response, Response) else None,
-                )
+            raise response_terminal_failure_error(
+                event_type,
+                maybe_response if isinstance(maybe_response, Response) else None,
             )
+        elif getattr(event, "type", None) in {"error", "response.error"}:
+            raise response_error_event_failure_error(cast(str, event.type), event)
 
         if terminal_response is not None:
             if is_completed_event and not terminal_response.output and streamed_response_output:

tests/models/test_any_llm_model.py

@@ -17,6 +17,9 @@
 from openai.types.chat.chat_completion_chunk import ChoiceDelta
 from openai.types.completion_usage import CompletionUsage, PromptTokensDetails
 from openai.types.responses import Response, ResponseCompletedEvent, ResponseOutputMessage
+from openai.types.responses.response_error_event import ResponseErrorEvent
+from openai.types.responses.response_failed_event import ResponseFailedEvent
+from openai.types.responses.response_incomplete_event import ResponseIncompleteEvent
 from openai.types.responses.response_output_text import ResponseOutputText
 from openai.types.responses.response_usage import (
     InputTokensDetails,
@@ -28,6 +31,7 @@
 from agents import (
     Agent,
     Handoff,
+    ModelBehaviorError,
     ModelSettings,
     ModelTracing,
     Tool,
@@ -534,6 +538,91 @@ async def response_stream() -> AsyncIterator[ResponseCompletedEvent]:
     assert provider.private_responses_calls[0]["params"].conversation == "conv_123"
 
 
+@pytest.mark.allow_call_model_methods
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("terminal_event_type", "terminal_event_cls"),
+    [
+        ("response.incomplete", ResponseIncompleteEvent),
+        ("response.failed", ResponseFailedEvent),
+    ],
+)
+async def test_any_llm_responses_stream_rejects_failed_terminal_events(
+    monkeypatch,
+    terminal_event_type: str,
+    terminal_event_cls: type[Any],
+) -> None:
+    async def response_stream() -> AsyncIterator[Any]:
+        yield terminal_event_cls(
+            type=terminal_event_type,
+            response=_response("partial", response_id="resp-terminal"),
+            sequence_number=1,
+        )
+
+    provider = FakeAnyLLMProvider(supports_responses=True, responses_response=response_stream())
+    module, _create_calls = _import_any_llm_module(monkeypatch, provider)
+    AnyLLMModel = module.AnyLLMModel
+
+    model = AnyLLMModel(model="openai/gpt-5.4-mini")
+    events = []
+    with pytest.raises(ModelBehaviorError, match=terminal_event_type):
+        async for event in model.stream_response(
+            system_instructions=None,
+            input="hi",
+            model_settings=ModelSettings(),
+            tools=[],
+            output_schema=None,
+            handoffs=[],
+            tracing=ModelTracing.DISABLED,
+            previous_response_id=None,
+            conversation_id=None,
+            prompt=None,
+        ):
+            events.append(event)
+
+    assert len(events) == 1
+    assert events[0].type == terminal_event_type
+    assert events[0].response.id == "resp-terminal"
+
+
+@pytest.mark.allow_call_model_methods
+@pytest.mark.asyncio
+async def test_any_llm_responses_stream_rejects_error_event(monkeypatch) -> None:
+    async def response_stream() -> AsyncIterator[ResponseErrorEvent]:
+        yield ResponseErrorEvent(
+            type="error",
+            code="invalid_request_error",
+            message="bad request",
+            param=None,
+            sequence_number=1,
+        )
+
+    provider = FakeAnyLLMProvider(supports_responses=True, responses_response=response_stream())
+    module, _create_calls = _import_any_llm_module(monkeypatch, provider)
+    AnyLLMModel = module.AnyLLMModel
+
+    model = AnyLLMModel(model="openai/gpt-5.4-mini")
+    events = []
+    with pytest.raises(ModelBehaviorError, match="invalid_request_error"):
+        async for event in model.stream_response(
+            system_instructions=None,
+            input="hi",
+            model_settings=ModelSettings(),
+            tools=[],
+            output_schema=None,
+            handoffs=[],
+            tracing=ModelTracing.DISABLED,
+            previous_response_id=None,
+            conversation_id=None,
+            prompt=None,
+        ):
+            events.append(event)
+
+    assert len(events) == 1
+    assert events[0].type == "error"
+    assert events[0].code == "invalid_request_error"
+
+
 @pytest.mark.allow_call_model_methods
 @pytest.mark.asyncio
 async def test_any_llm_responses_path_passes_transport_kwargs_via_private_provider_api(

tests/models/test_openai_responses.py

@@ -8,7 +8,7 @@
 import httpx
 import pytest
 from openai import NOT_GIVEN, APIConnectionError, RateLimitError, omit
-from openai.types.responses import ResponseCompletedEvent
+from openai.types.responses import ResponseCompletedEvent, ResponseErrorEvent
 from openai.types.shared.reasoning import Reasoning
 
 from agents import (
@@ -1770,6 +1770,56 @@ async def fake_open(
     assert cast(Any, events[0]).response.id == "resp-terminal"
 
 
+@pytest.mark.allow_call_model_methods
+@pytest.mark.asyncio
+async def test_stream_response_rejects_response_error_terminal_event(monkeypatch):
+    model = OpenAIResponsesModel(model="gpt-4", openai_client=object())  # type: ignore[arg-type]
+
+    async def dummy_fetch_response(
+        system_instructions,
+        input,
+        model_settings,
+        tools,
+        output_schema,
+        handoffs,
+        previous_response_id,
+        conversation_id,
+        stream,
+        prompt,
+    ):
+        class DummyStream:
+            async def __aiter__(self):
+                yield ResponseErrorEvent(
+                    type="error",
+                    code="invalid_request_error",
+                    message="bad request",
+                    param=None,
+                    sequence_number=0,
+                )
+
+        return DummyStream()
+
+    monkeypatch.setattr(model, "_fetch_response", dummy_fetch_response)
+
+    events = []
+    with pytest.raises(ModelBehaviorError, match="invalid_request_error"):
+        async for event in model.stream_response(
+            system_instructions=None,
+            input="hi",
+            model_settings=ModelSettings(),
+            tools=[],
+            output_schema=None,
+            handoffs=[],
+            tracing=ModelTracing.DISABLED,
+        ):
+            events.append(event)
+
+    assert len(events) == 1
+    assert events[0].type == "error"
+    assert events[0].code == "invalid_request_error"
+    assert events[0].message == "bad request"
+
+
 @pytest.mark.allow_call_model_methods
 @pytest.mark.asyncio
 async def test_websocket_model_get_response_surfaces_response_error_event(monkeypatch):