From 36d3776ff54c0fde32ad77059170f9f08074f51a Mon Sep 17 00:00:00 2001 From: Justin Beckwith Date: Sun, 27 Sep 2026 09:45:46 -0700 Subject: [PATCH 1/3] fix(responses): preserve terminal errors through stream cleanup --- src/agents/models/openai_responses.py | 5 + tests/models/test_openai_responses.py | 140 +++++++++++++++++++++++--- 2 files changed, 131 insertions(+), 14 deletions(-) diff --git a/src/agents/models/openai_responses.py b/src/agents/models/openai_responses.py index 47d4fa7a93..87e4b5bec3 100644 --- a/src/agents/models/openai_responses.py +++ b/src/agents/models/openai_responses.py @@ -262,6 +262,7 @@ class _ResponseStreamWithRequestId: "response.completed", "response.failed", "response.incomplete", + "error", "response.error", } @@ -778,6 +779,10 @@ async def stream_response( trace_include_sensitive_data=tracing.include_data(), ) yield chunk + if terminal_failure_error is not None: + # Close explicitly rather than advancing a failed stream: transport + # teardown during iteration must not replace the provider's error. + break except asyncio.CancelledError: close_stream_in_background = True self._schedule_async_iterator_close(stream) diff --git a/tests/models/test_openai_responses.py b/tests/models/test_openai_responses.py index 66f69573cd..8f17583d58 100644 --- a/tests/models/test_openai_responses.py +++ b/tests/models/test_openai_responses.py @@ -4,6 +4,7 @@ import copy import json import logging +from contextlib import nullcontext from dataclasses import asdict, fields, replace from types import SimpleNamespace from typing import Any, cast @@ -908,7 +909,10 @@ def __init__(self): @pytest.mark.allow_call_model_methods @pytest.mark.asyncio -async def test_stream_response_ignores_streaming_context_exit_failure_after_terminal_event(): +@pytest.mark.parametrize("event_type", ["response.completed", "error"]) +async def test_stream_response_ignores_streaming_context_exit_failure_after_terminal_event( + event_type, +): class DummyHTTPStream: def __init__(self): self._yielded = False @@ -920,6 +924,14 @@ async def __anext__(self): if self._yielded: raise StopAsyncIteration self._yielded = True + if event_type == "error": + return ResponseErrorEvent( + type="error", + code="server_error", + message="synthetic provider failure", + param=None, + sequence_number=0, + ) return ResponseCompletedEvent( type="response.completed", response=get_response_obj([], response_id="resp-stream-request-id"), @@ -958,23 +970,123 @@ def __init__(self): model = OpenAIResponsesModel(model="gpt-4", openai_client=DummyResponsesClient()) # type: ignore[arg-type] - events: list[ResponseCompletedEvent] = [] - async for event in model.stream_response( - system_instructions=None, - input="hi", - model_settings=ModelSettings(), - tools=[], - output_schema=None, - handoffs=[], - tracing=ModelTracing.DISABLED, - ): - assert isinstance(event, ResponseCompletedEvent) - events.append(event) + events = [] + expected_error = ( + pytest.raises(ModelBehaviorError, match="synthetic provider failure") + if event_type == "error" + else nullcontext() + ) + with expected_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 [event.type for event in events] == [event_type] assert aexit_calls == [(None, None, None)] +@pytest.mark.asyncio +async def test_response_stream_preserves_error_event_when_cleanup_fails() -> None: + event = ResponseErrorEvent( + type="error", + code="server_error", + message="synthetic provider failure", + param=None, + sequence_number=0, + ) + cleanup_calls = 0 + + async def events(): + yield event + + async def cleanup(): + nonlocal cleanup_calls + cleanup_calls += 1 + raise RuntimeError("transport close failed") + + stream = _ResponseStreamWithRequestId(events(), request_id=None, cleanup=cleanup) + assert [item async for item in stream] == [event] + assert cleanup_calls == 1 + + +@pytest.mark.allow_call_model_methods +@pytest.mark.asyncio +@pytest.mark.parametrize("event_type", ["error", "response.failed", "response.incomplete", None]) +async def test_stream_response_preserves_terminal_failure_when_http_close_fails( + event_type: str | None, +) -> None: + # Exercise real SSE parsing and transport teardown, which a model fake would bypass. + close_calls = 0 + if event_type == "error": + payload = ResponseErrorEvent( + type="error", + code="server_error", + message="synthetic provider failure", + param=None, + sequence_number=0, + ).model_dump() + expected_message = "code=server_error; message=synthetic provider failure" + elif event_type is not None: + status = event_type.removeprefix("response.") + payload = { + "type": event_type, + "sequence_number": 0, + "response": _response_with_terminal_status(status).model_dump(), + } + expected_message = f"status={status}" + else: + payload = None + expected_message = "transport close failed" + + class FailingCloseStream(httpx2.AsyncByteStream): + async def __aiter__(self): + if payload is not None: + yield f"event: {event_type}\ndata: {json.dumps(payload)}\n\n".encode() + yield b"data: [DONE]\n\n" + + async def aclose(self): + nonlocal close_calls + close_calls += 1 + raise RuntimeError("transport close failed") + + async def handler(request: httpx2.Request) -> httpx2.Response: + return httpx2.Response( + 200, + headers={"content-type": "text/event-stream"}, + stream=FailingCloseStream(), + request=request, + ) + + events = [] + async with httpx2.AsyncClient(transport=httpx2.MockTransport(handler)) as http_client: + model = OpenAIResponsesModel( + model="gpt-4", + openai_client=AsyncOpenAI(api_key="test-key", http_client=http_client), + ) + expected_exception = ModelBehaviorError if event_type is not None else RuntimeError + with pytest.raises(expected_exception, match=expected_message): + 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 [event.type for event in events] == ([event_type] if event_type is not None else []) + assert close_calls == 1 + + @pytest.mark.allow_call_model_methods @pytest.mark.asyncio async def test_stream_response_close_closes_inner_http_stream_with_async_close(monkeypatch): From a843fa06d522e87eb2ad1627ff7efd4bbfecbcaf Mon Sep 17 00:00:00 2001 From: Justin Beckwith Date: Mon, 28 Sep 2026 10:25:59 -0700 Subject: [PATCH 2/3] fix(sessions): accept namespaced compaction model names --- .../openai_responses_compaction_session.py | 1 + ...est_openai_responses_compaction_session.py | 20 +++++++++++++++---- 2 files changed, 17 insertions(+), 4 deletions(-) diff --git a/src/agents/memory/openai_responses_compaction_session.py b/src/agents/memory/openai_responses_compaction_session.py index d118f110cc..d728125f19 100644 --- a/src/agents/memory/openai_responses_compaction_session.py +++ b/src/agents/memory/openai_responses_compaction_session.py @@ -77,6 +77,7 @@ def is_openai_model_name(model: str) -> bool: # Handle fine-tuned models: ft:gpt-4.1:org:proj:suffix without_ft_prefix = trimmed[3:] if trimmed.startswith("ft:") else trimmed root = without_ft_prefix.split(":", 1)[0] + root = root.rsplit("/", 1)[-1] # Allow gpt-* and o* models if root.startswith("gpt-"): diff --git a/tests/memory/test_openai_responses_compaction_session.py b/tests/memory/test_openai_responses_compaction_session.py index be06ab8361..09337dd1f3 100644 --- a/tests/memory/test_openai_responses_compaction_session.py +++ b/tests/memory/test_openai_responses_compaction_session.py @@ -289,14 +289,15 @@ def create_mock_session(self) -> MagicMock: mock.clear_session = AsyncMock() return mock - def test_init_validates_model(self) -> None: + @pytest.mark.parametrize("model", ["claude-3", "anthropic/claude-3"]) + def test_init_validates_model(self, model: str) -> None: mock_session = self.create_mock_session() with pytest.raises(ValueError, match="Unsupported model"): OpenAIResponsesCompactionSession( session_id="test", underlying_session=mock_session, - model="claude-3", + model=model, ) def test_init_accepts_valid_model(self) -> None: @@ -672,7 +673,17 @@ def __call__(self, context: dict[str, Any]) -> bool: mock_client.responses.compact.assert_not_awaited() @pytest.mark.asyncio - async def test_run_compaction_input_mode_without_response_id(self) -> None: + @pytest.mark.parametrize( + "model", + [ + "gpt-4.1", + "ft:gpt-4.1:my-org::id", + "openai/gpt-4.1", + "openai/openai/gpt-5.6-terra", + "openai/o3", + ], + ) + async def test_run_compaction_input_mode_without_response_id(self, model: str) -> None: mock_session = self.create_mock_session() items: list[TResponseInputItem] = [ cast(TResponseInputItem, {"type": "message", "role": "user", "content": "hello"}), @@ -699,6 +710,7 @@ async def test_run_compaction_input_mode_without_response_id(self) -> None: session_id="test", underlying_session=mock_session, client=mock_client, + model=model, compaction_mode="input", ) @@ -706,7 +718,7 @@ async def test_run_compaction_input_mode_without_response_id(self) -> None: mock_client.responses.compact.assert_called_once() call_kwargs = mock_client.responses.compact.call_args.kwargs - assert call_kwargs.get("model") == "gpt-4.1" + assert call_kwargs.get("model") == model assert "previous_response_id" not in call_kwargs assert call_kwargs.get("input") == items From 88c0e672dbc3182dd247fb1de8f78346a6b03fb1 Mon Sep 17 00:00:00 2001 From: Justin Beckwith Date: Mon, 28 Sep 2026 10:35:19 -0700 Subject: [PATCH 3/3] fix(sessions): recognize namespaced fine-tuned compaction models --- src/agents/memory/openai_responses_compaction_session.py | 3 +-- tests/memory/test_openai_responses_compaction_session.py | 1 + 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/src/agents/memory/openai_responses_compaction_session.py b/src/agents/memory/openai_responses_compaction_session.py index 6cce52dd28..a3792f68b0 100644 --- a/src/agents/memory/openai_responses_compaction_session.py +++ b/src/agents/memory/openai_responses_compaction_session.py @@ -71,14 +71,13 @@ def default_should_trigger_compaction(context: dict[str, Any]) -> bool: def is_openai_model_name(model: str) -> bool: """Validate model name follows OpenAI conventions.""" - trimmed = model.strip() + trimmed = model.strip().rsplit("/", 1)[-1] if not trimmed: return False # Handle fine-tuned models: ft:gpt-4.1:org:proj:suffix without_ft_prefix = trimmed[3:] if trimmed.startswith("ft:") else trimmed root = without_ft_prefix.split(":", 1)[0] - root = root.rsplit("/", 1)[-1] # Allow gpt-* and o* models if root.startswith("gpt-"): diff --git a/tests/memory/test_openai_responses_compaction_session.py b/tests/memory/test_openai_responses_compaction_session.py index 5c2fc4fcb2..6d8c47cbb1 100644 --- a/tests/memory/test_openai_responses_compaction_session.py +++ b/tests/memory/test_openai_responses_compaction_session.py @@ -768,6 +768,7 @@ def __call__(self, context: dict[str, Any]) -> bool: "openai/gpt-4.1", "openai/openai/gpt-5.6-terra", "openai/o3", + "openai/ft:gpt-4.1:org:proj:id", ], ) async def test_run_compaction_input_mode_without_response_id(self, model: str) -> None: