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/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_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):

tests/test_agent_runner_streamed.py

@@ -9,6 +9,7 @@
 from openai import APIConnectionError, BadRequestError
 from openai.types.responses import (
     ResponseCompletedEvent,
+    ResponseErrorEvent,
     ResponseFailedEvent,
     ResponseFunctionToolCall,
     ResponseIncompleteEvent,
@@ -42,7 +43,7 @@
 from agents.run import RunConfig
 from agents.run_internal import run_loop
 from agents.run_internal.run_loop import QueueCompleteSentinel
-from agents.stream_events import AgentUpdatedStreamEvent, StreamEvent
+from agents.stream_events import AgentUpdatedStreamEvent, RawResponsesStreamEvent, StreamEvent
 from agents.usage import Usage
 
 from .fake_model import FakeModel, get_response_obj
@@ -188,10 +189,71 @@ async def stream_response(
     agent = Agent(name="test", model=model)
 
     result = Runner.run_streamed(agent, input="test")
+    stream_events: list[StreamEvent] = []
     with pytest.raises(ModelBehaviorError, match=terminal_event_type):
-        async for _ in result.stream_events():
-            pass
+        async for event in result.stream_events():
+            stream_events.append(event)
+
+    assert len(stream_events) == 2
+    assert isinstance(stream_events[0], AgentUpdatedStreamEvent)
+    assert isinstance(stream_events[1], RawResponsesStreamEvent)
+    assert stream_events[1].data.type == terminal_event_type
+    assert result.final_output is None
+    assert result.raw_responses == []
+
+
+@pytest.mark.asyncio
+async def test_streamed_run_rejects_response_error_terminal_event() -> None:
+    class TerminalErrorFakeModel(FakeModel):
+        async def stream_response(
+            self,
+            system_instructions,
+            input,
+            model_settings,
+            tools,
+            output_schema,
+            handoffs,
+            tracing,
+            *,
+            previous_response_id=None,
+            conversation_id=None,
+            prompt=None,
+        ):
+            self.last_turn_args = {
+                "system_instructions": system_instructions,
+                "input": input,
+                "model_settings": model_settings,
+                "tools": tools,
+                "output_schema": output_schema,
+                "previous_response_id": previous_response_id,
+                "conversation_id": conversation_id,
+            }
+            if self.first_turn_args is None:
+                self.first_turn_args = self.last_turn_args.copy()
+
+            yield ResponseErrorEvent(
+                type="error",
+                code="invalid_request_error",
+                message="bad request",
+                param=None,
+                sequence_number=0,
+            )
+
+    model = TerminalErrorFakeModel()
+    agent = Agent(name="test", model=model)
 
+    result = Runner.run_streamed(agent, input="test")
+    stream_events: list[StreamEvent] = []
+    with pytest.raises(ModelBehaviorError, match="error"):
+        async for event in result.stream_events():
+            stream_events.append(event)
+
+    assert len(stream_events) == 2
+    assert isinstance(stream_events[0], AgentUpdatedStreamEvent)
+    assert isinstance(stream_events[1], RawResponsesStreamEvent)
+    assert stream_events[1].data.type == "error"
+    assert stream_events[1].data.code == "invalid_request_error"
+    assert stream_events[1].data.message == "bad request"
     assert result.final_output is None
     assert result.raw_responses == []
 
@@ -373,10 +435,15 @@ async def fake_open(
 
     agent = Agent(name="test", model=model)
     result = Runner.run_streamed(agent, input="test")
+    stream_events: list[StreamEvent] = []
     with pytest.raises(ModelBehaviorError, match=terminal_event_type):
-        async for _ in result.stream_events():
-            pass
+        async for event in result.stream_events():
+            stream_events.append(event)
 
+    assert len(stream_events) == 2
+    assert isinstance(stream_events[0], AgentUpdatedStreamEvent)
+    assert isinstance(stream_events[1], RawResponsesStreamEvent)
+    assert stream_events[1].data.type == terminal_event_type
     assert result.final_output is None
     assert result.raw_responses == []