From 28419c69e5cd35702e418222f8d3767f216bbb2a Mon Sep 17 00:00:00 2001
From: Xinmin Zeng <135568692+fancyboi999@users.noreply.github.com>
Date: Fri, 23 Oct 2026 21:29:53 -0700
Subject: [PATCH 1/7] fix(streaming): close the SSE event iterators a stream
 opens

AsyncStream stopped at [DONE] (or was closed early) without closing the
async generators it had opened: _iter_events, the decoder iterator and
the response.aiter_bytes() iterator, plus the __aiter__ wrapper after an
early exit. They stayed suspended until the garbage collector reached
them, where asyncio closes each one in its own aclose() task; a nested
chain closed concurrently fails with "aclose(): asynchronous generator
is already running" and skips its cleanup.

Each layer now closes what it opened: _iter_events closes the decoder
and byte iterators, __stream__ closes the event iterator before the
response, close() closes the stream iterator when it is not running,
and __aiter__ returns the stream iterator instead of wrapping it. The
sync Stream gets the symmetric change.
---
 src/openai/_streaming.py | 83 ++++++++++++++++++++++++++++++++--------
 tests/test_streaming.py  | 36 +++++++++++++++++
 2 files changed, 103 insertions(+), 16 deletions(-)

diff --git a/src/openai/_streaming.py b/src/openai/_streaming.py
index 17c4c55eb3..17a2754673 100644
--- a/src/openai/_streaming.py
+++ b/src/openai/_streaming.py
@@ -5,7 +5,18 @@
 import json
 import inspect
 from types import TracebackType
-from typing import TYPE_CHECKING, Any, Generic, TypeVar, Iterator, Optional, AsyncIterator, cast
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    Generic,
+    TypeVar,
+    Iterator,
+    Optional,
+    Generator,
+    AsyncIterator,
+    AsyncGenerator,
+    cast,
+)
 from typing_extensions import Self, Protocol, TypeGuard, override, get_origin, runtime_checkable
 
 import httpx2
@@ -49,18 +60,25 @@ def __next__(self) -> _T:
         return self._iterator.__next__()
 
     def __iter__(self) -> Iterator[_T]:
-        for item in self._iterator:
-            yield item
+        return self._iterator
 
-    def _iter_events(self) -> Iterator[ServerSentEvent]:
+    def _iter_events(self) -> Generator[ServerSentEvent, None, None]:
+        chunks = self.response.iter_bytes()
+        events = self._decoder.iter_bytes(chunks)
         try:
-            yield from self._decoder.iter_bytes(self.response.iter_bytes())
+            yield from events
         except timeout_exceptions() as err:
             raise APITimeoutError(request=self.response.request) from err
         except request_exceptions() as err:
             raise APIConnectionError(request=self.response.request) from err
+        finally:
+            # Close the iterators opened here; stopping early does not close them.
+            try:
+                _close(events)
+            finally:
+                _close(chunks)
 
-    def __stream__(self) -> Iterator[_T]:
+    def __stream__(self) -> Generator[_T, None, None]:
         cast_to = cast(Any, self._cast_to)
         response = self.response
         process_data = self._client._process_response_data
@@ -114,8 +132,12 @@ def __stream__(self) -> Iterator[_T]:
                         response=response,
                     )
         finally:
-            # Ensure the response is closed even if the consumer doesn't read all data
-            response.close()
+            # Close the event iterator explicitly; stopping at [DONE] leaves it suspended.
+            try:
+                iterator.close()
+            finally:
+                # Ensure the response is closed even if the consumer doesn't read all data
+                response.close()
 
     def __enter__(self) -> Self:
         return self
@@ -134,6 +156,8 @@ def close(self) -> None:
 
         Automatically called if the response body is read to completion.
         """
+        if not getattr(self._iterator, "gi_running", False):
+            self._iterator.close()
         self.response.close()
 
 
@@ -162,20 +186,27 @@ def __init__(
     async def __anext__(self) -> _T:
         return await self._iterator.__anext__()
 
-    async def __aiter__(self) -> AsyncIterator[_T]:
-        async for item in self._iterator:
-            yield item
+    def __aiter__(self) -> AsyncIterator[_T]:
+        return self._iterator
 
-    async def _iter_events(self) -> AsyncIterator[ServerSentEvent]:
+    async def _iter_events(self) -> AsyncGenerator[ServerSentEvent, None]:
+        chunks = self.response.aiter_bytes()
+        events = self._decoder.aiter_bytes(chunks)
         try:
-            async for sse in self._decoder.aiter_bytes(self.response.aiter_bytes()):
+            async for sse in events:
                 yield sse
         except timeout_exceptions() as err:
             raise APITimeoutError(request=self.response.request) from err
         except request_exceptions() as err:
             raise APIConnectionError(request=self.response.request) from err
+        finally:
+            # Close the iterators opened here; an interrupted `async for` does not.
+            try:
+                await _aclose(events)
+            finally:
+                await _aclose(chunks)
 
-    async def __stream__(self) -> AsyncIterator[_T]:
+    async def __stream__(self) -> AsyncGenerator[_T, None]:
         cast_to = cast(Any, self._cast_to)
         response = self.response
         process_data = self._client._process_response_data
@@ -234,8 +265,14 @@ async def __stream__(self) -> AsyncIterator[_T]:
                         response=response,
                     )
         finally:
-            # Ensure the response is closed even if the consumer doesn't read all data
-            await response.aclose()
+            # Close the event iterator explicitly. Stopping at [DONE] leaves it and the
+            # httpx2 byte iterators beneath it suspended; left to the garbage collector,
+            # asyncio closes those nested generators concurrently.
+            try:
+                await iterator.aclose()
+            finally:
+                # Ensure the response is closed even if the consumer doesn't read all data
+                await response.aclose()
 
     async def __aenter__(self) -> Self:
         return self
@@ -254,6 +291,8 @@ async def close(self) -> None:
 
         Automatically called if the response body is read to completion.
         """
+        if not getattr(self._iterator, "ag_running", False):
+            await self._iterator.aclose()
         await self.response.aclose()
 
     async def aclose(self) -> None:
@@ -435,6 +474,18 @@ def decode(self, line: str) -> ServerSentEvent | None:
         return None
 
 
+def _close(iterator: object) -> None:
+    close = getattr(iterator, "close", None)
+    if close is not None:
+        close()
+
+
+async def _aclose(iterator: object) -> None:
+    aclose = getattr(iterator, "aclose", None)
+    if aclose is not None:
+        await aclose()
+
+
 @runtime_checkable
 class SSEBytesDecoder(Protocol):
     def iter_bytes(self, iterator: Iterator[bytes]) -> Iterator[ServerSentEvent]:
diff --git a/tests/test_streaming.py b/tests/test_streaming.py
index 6378b1934d..c65c7d115f 100644
--- a/tests/test_streaming.py
+++ b/tests/test_streaming.py
@@ -1,7 +1,9 @@
 from __future__ import annotations
 
+import gc
 import os
 import ssl
+import sys
 import time
 import socket
 import asyncio
@@ -584,6 +586,40 @@ def body() -> Iterator[bytes]:
         assert response.is_closed
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("early_exit", [False, True], ids=["read-to-done", "early-exit"])
+async def test_async_stream_closes_its_iterators(early_exit: bool, async_client: AsyncOpenAI) -> None:
+    def body() -> Iterator[bytes]:
+        yield b'data: {"foo":true}\n\n'
+        yield b'data: {"bar":false}\n\n'
+        yield b"data: [DONE]\n\n"
+
+    stream = AsyncStream(cast_to=object, client=async_client, response=httpx2.Response(200, content=to_aiter(body())))
+    collected: list[str] = []
+    firstiter, finalizer = sys.get_asyncgen_hooks()
+
+    def record(agen: Any) -> None:
+        collected.append(agen.__qualname__)
+        if finalizer is not None:
+            finalizer(agen)
+
+    # Generators bind the finalizer when first iterated, so install it before iterating.
+    sys.set_asyncgen_hooks(firstiter=firstiter, finalizer=record)
+    try:
+        async for _ in stream:
+            if early_exit:
+                break
+        await stream.close()
+        del stream
+        gc.collect()
+    finally:
+        sys.set_asyncgen_hooks(firstiter=firstiter, finalizer=finalizer)
+
+    # Iterators the stream opened must be closed by the stream, not left to the
+    # garbage collector, where asyncio closes nested generators concurrently.
+    assert [name for name in collected if name.startswith(("AsyncStream.", "SSEDecoder."))] == []
+
+
 async def to_aiter(iter: Iterator[bytes]) -> AsyncIterator[bytes]:
     for chunk in iter:
         yield chunk

From df414c7b6fccccee729cc5fb6c46e3fbaeebbc19 Mon Sep 17 00:00:00 2001
From: Xinmin Zeng <135568692+fancyboi999@users.noreply.github.com>
Date: Fri, 23 Oct 2026 22:06:03 -0700
Subject: [PATCH 2/7] fix(streaming): address review of iterator cleanup

- Keep the Iterator/AsyncIterator contracts so existing __stream__
  overrides stay valid; close owned iterators through close_owned /
  aclose_owned, which skip objects without close()/aclose().
- Do not let a cleanup error replace the error or cancellation already
  propagating, and shield async cleanup from caller cancellation.
- Create the byte and decoder iterators inside the exception-translation
  block, and loop explicitly in the sync stream so the decoder iterator
  is closed once.
- close() closes the response even when iterator cleanup fails.
- Chat and Responses stream wrappers close their own iterator and the raw
  stream instead of only the response; agent streams close the parent
  stream iterator they consume.
- Tests cover the byte iterator, the chat wrapper, single close of custom
  decoder iterators, and error precedence.
---
 src/openai/_streaming.py                      | 122 +++++++-------
 src/openai/_streaming_cleanup.py              |  41 ++++-
 src/openai/lib/beta/agents/_stream.py         |  27 ++-
 src/openai/lib/streaming/chat/_completions.py |  29 +++-
 .../lib/streaming/responses/_responses.py     |  29 +++-
 tests/test_streaming.py                       | 159 ++++++++++++++++--
 6 files changed, 312 insertions(+), 95 deletions(-)

diff --git a/src/openai/_streaming.py b/src/openai/_streaming.py
index 17a2754673..9d2c17cbd4 100644
--- a/src/openai/_streaming.py
+++ b/src/openai/_streaming.py
@@ -5,18 +5,7 @@
 import json
 import inspect
 from types import TracebackType
-from typing import (
-    TYPE_CHECKING,
-    Any,
-    Generic,
-    TypeVar,
-    Iterator,
-    Optional,
-    Generator,
-    AsyncIterator,
-    AsyncGenerator,
-    cast,
-)
+from typing import TYPE_CHECKING, Any, Generic, TypeVar, Iterator, Optional, AsyncIterator, cast
 from typing_extensions import Self, Protocol, TypeGuard, override, get_origin, runtime_checkable
 
 import httpx2
@@ -24,7 +13,7 @@
 from ._utils import is_mapping, extract_type_var_from_base
 from ._httpx2 import request_exceptions, timeout_exceptions
 from ._exceptions import APIError, APITimeoutError, APIConnectionError
-from ._streaming_cleanup import drain_response_async
+from ._streaming_cleanup import close_owned, aclose_owned, drain_response_async
 
 if TYPE_CHECKING:
     from ._client import OpenAI, AsyncOpenAI
@@ -62,27 +51,35 @@ def __next__(self) -> _T:
     def __iter__(self) -> Iterator[_T]:
         return self._iterator
 
-    def _iter_events(self) -> Generator[ServerSentEvent, None, None]:
-        chunks = self.response.iter_bytes()
-        events = self._decoder.iter_bytes(chunks)
+    def _iter_events(self) -> Iterator[ServerSentEvent]:
+        chunks: Optional[Iterator[bytes]] = None
+        events: Optional[Iterator[ServerSentEvent]] = None
+        failure: Optional[BaseException] = None
         try:
-            yield from events
-        except timeout_exceptions() as err:
-            raise APITimeoutError(request=self.response.request) from err
-        except request_exceptions() as err:
-            raise APIConnectionError(request=self.response.request) from err
+            try:
+                chunks = self.response.iter_bytes()
+                events = self._decoder.iter_bytes(chunks)
+                # Not `yield from`: it would close `events` itself, and the
+                # cleanup below must stay the only close.
+                for sse in events:
+                    yield sse
+            except timeout_exceptions() as err:
+                raise APITimeoutError(request=self.response.request) from err
+            except request_exceptions() as err:
+                raise APIConnectionError(request=self.response.request) from err
+        except BaseException as err:
+            failure = err
+            raise
         finally:
             # Close the iterators opened here; stopping early does not close them.
-            try:
-                _close(events)
-            finally:
-                _close(chunks)
+            close_owned(events, chunks, primary=failure)
 
-    def __stream__(self) -> Generator[_T, None, None]:
+    def __stream__(self) -> Iterator[_T]:
         cast_to = cast(Any, self._cast_to)
         response = self.response
         process_data = self._client._process_response_data
         iterator = self._iter_events()
+        failure: Optional[BaseException] = None
 
         try:
             for sse in iterator:
@@ -131,10 +128,13 @@ def __stream__(self) -> Generator[_T, None, None]:
                         cast_to=cast_to,
                         response=response,
                     )
+        except BaseException as err:
+            failure = err
+            raise
         finally:
             # Close the event iterator explicitly; stopping at [DONE] leaves it suspended.
             try:
-                iterator.close()
+                close_owned(iterator, primary=failure)
             finally:
                 # Ensure the response is closed even if the consumer doesn't read all data
                 response.close()
@@ -156,9 +156,12 @@ def close(self) -> None:
 
         Automatically called if the response body is read to completion.
         """
-        if not getattr(self._iterator, "gi_running", False):
-            self._iterator.close()
-        self.response.close()
+        try:
+            # A running iterator belongs to its consumer; closing the response ends it.
+            if not getattr(self._iterator, "gi_running", False):
+                close_owned(self._iterator)
+        finally:
+            self.response.close()
 
 
 class AsyncStream(Generic[_T]):
@@ -189,28 +192,33 @@ async def __anext__(self) -> _T:
     def __aiter__(self) -> AsyncIterator[_T]:
         return self._iterator
 
-    async def _iter_events(self) -> AsyncGenerator[ServerSentEvent, None]:
-        chunks = self.response.aiter_bytes()
-        events = self._decoder.aiter_bytes(chunks)
+    async def _iter_events(self) -> AsyncIterator[ServerSentEvent]:
+        chunks: Optional[AsyncIterator[bytes]] = None
+        events: Optional[AsyncIterator[ServerSentEvent]] = None
+        failure: Optional[BaseException] = None
         try:
-            async for sse in events:
-                yield sse
-        except timeout_exceptions() as err:
-            raise APITimeoutError(request=self.response.request) from err
-        except request_exceptions() as err:
-            raise APIConnectionError(request=self.response.request) from err
+            try:
+                chunks = self.response.aiter_bytes()
+                events = self._decoder.aiter_bytes(chunks)
+                async for sse in events:
+                    yield sse
+            except timeout_exceptions() as err:
+                raise APITimeoutError(request=self.response.request) from err
+            except request_exceptions() as err:
+                raise APIConnectionError(request=self.response.request) from err
+        except BaseException as err:
+            failure = err
+            raise
         finally:
             # Close the iterators opened here; an interrupted `async for` does not.
-            try:
-                await _aclose(events)
-            finally:
-                await _aclose(chunks)
+            await aclose_owned(events, chunks, primary=failure)
 
-    async def __stream__(self) -> AsyncGenerator[_T, None]:
+    async def __stream__(self) -> AsyncIterator[_T]:
         cast_to = cast(Any, self._cast_to)
         response = self.response
         process_data = self._client._process_response_data
         iterator = self._iter_events()
+        failure: Optional[BaseException] = None
 
         try:
             async for sse in iterator:
@@ -264,12 +272,15 @@ async def __stream__(self) -> AsyncGenerator[_T, None]:
                         cast_to=cast_to,
                         response=response,
                     )
+        except BaseException as err:
+            failure = err
+            raise
         finally:
             # Close the event iterator explicitly. Stopping at [DONE] leaves it and the
             # httpx2 byte iterators beneath it suspended; left to the garbage collector,
             # asyncio closes those nested generators concurrently.
             try:
-                await iterator.aclose()
+                await aclose_owned(iterator, primary=failure)
             finally:
                 # Ensure the response is closed even if the consumer doesn't read all data
                 await response.aclose()
@@ -291,9 +302,12 @@ async def close(self) -> None:
 
         Automatically called if the response body is read to completion.
         """
-        if not getattr(self._iterator, "ag_running", False):
-            await self._iterator.aclose()
-        await self.response.aclose()
+        try:
+            # A running iterator belongs to its consumer; closing the response ends it.
+            if not getattr(self._iterator, "ag_running", False):
+                await aclose_owned(self._iterator)
+        finally:
+            await self.response.aclose()
 
     async def aclose(self) -> None:
         """Close the response and release the connection. Alias for `close()`."""
@@ -474,18 +488,6 @@ def decode(self, line: str) -> ServerSentEvent | None:
         return None
 
 
-def _close(iterator: object) -> None:
-    close = getattr(iterator, "close", None)
-    if close is not None:
-        close()
-
-
-async def _aclose(iterator: object) -> None:
-    aclose = getattr(iterator, "aclose", None)
-    if aclose is not None:
-        await aclose()
-
-
 @runtime_checkable
 class SSEBytesDecoder(Protocol):
     def iter_bytes(self, iterator: Iterator[bytes]) -> Iterator[ServerSentEvent]:
diff --git a/src/openai/_streaming_cleanup.py b/src/openai/_streaming_cleanup.py
index d05b57ff04..79467e65c8 100644
--- a/src/openai/_streaming_cleanup.py
+++ b/src/openai/_streaming_cleanup.py
@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from typing import AsyncIterator, cast
+from typing import Optional, AsyncIterator, cast
 
 import anyio
 import httpx2
@@ -27,3 +27,42 @@ async def drain_response_async(response: httpx2.Response) -> None:
         # cancellation. Cancellation still propagates after the response closes.
         with anyio.CancelScope(shield=True):
             await response.aclose()
+
+
+def close_owned(*iterators: object, primary: Optional[BaseException] = None) -> None:
+    """Close iterators this caller opened, in order.
+
+    Every iterator gets its close attempt. Objects without ``close()`` are skipped.
+    A cleanup error is raised only when no ``primary`` error is already propagating.
+    """
+    failure: Optional[BaseException] = None
+    for iterator in iterators:
+        close = getattr(iterator, "close", None)
+        if close is None:
+            continue
+        try:
+            close()
+        except Exception as err:
+            failure = failure or err
+    if failure is not None and primary is None:
+        raise failure
+
+
+async def aclose_owned(*iterators: object, primary: Optional[BaseException] = None) -> None:
+    """Async counterpart of ``close_owned``.
+
+    Cleanup is shielded from caller cancellation so an already-cancelled scope cannot
+    leave a generator closed with its cleanup half done; cancellation still propagates.
+    """
+    failure: Optional[BaseException] = None
+    with anyio.CancelScope(shield=True):
+        for iterator in iterators:
+            aclose = getattr(iterator, "aclose", None)
+            if aclose is None:
+                continue
+            try:
+                await aclose()
+            except Exception as err:
+                failure = failure or err
+    if failure is not None and primary is None:
+        raise failure
diff --git a/src/openai/lib/beta/agents/_stream.py b/src/openai/lib/beta/agents/_stream.py
index a2cf143905..cdaf6e94a3 100644
--- a/src/openai/lib/beta/agents/_stream.py
+++ b/src/openai/lib/beta/agents/_stream.py
@@ -6,6 +6,7 @@
 from ._result import OutputT, AgentTurnResult, AgentTurnResultError, AgentOutputParseError, AgentTurnResultCollection
 from ...._compat import cached_property
 from ...._streaming import Stream, AsyncStream
+from ...._streaming_cleanup import close_owned, aclose_owned
 from ...streaming.agents._dispatch import ToolDispatcher, AsyncToolDispatcher
 from ....types.beta.agent_session_event import AgentSessionEvent
 
@@ -26,8 +27,10 @@ def _collection(self) -> AgentTurnResultCollection[OutputT]:
 
     @override
     def __stream__(self) -> Iterator[AgentSessionEvent]:
+        events = super().__stream__()
+        failure: BaseException | None = None
         try:
-            for event in super().__stream__():
+            for event in events:
                 self._collection.accept(event)
                 pending = self._dispatcher.prepare(event) if self._dispatcher is not None else None
                 yield event
@@ -35,9 +38,16 @@ def __stream__(self) -> Iterator[AgentSessionEvent]:
                     self._dispatcher.dispatch(pending)
         except Exception as error:
             self._collection.record_error(error)
+            failure = error
+            raise
+        except BaseException as error:
+            failure = error
             raise
         finally:
-            self.close()
+            try:
+                close_owned(events, primary=failure)
+            finally:
+                self.close()
 
     def with_result_collection(self) -> Self:
         """Beta: retain final messages for a result; call before consuming events."""
@@ -81,8 +91,10 @@ def _collection(self) -> AgentTurnResultCollection[OutputT]:
 
     @override
     async def __stream__(self) -> AsyncIterator[AgentSessionEvent]:
+        events = super().__stream__()
+        failure: BaseException | None = None
         try:
-            async for event in super().__stream__():
+            async for event in events:
                 self._collection.accept(event)
                 pending = self._dispatcher.prepare(event) if self._dispatcher is not None else None
                 yield event
@@ -90,9 +102,16 @@ async def __stream__(self) -> AsyncIterator[AgentSessionEvent]:
                     await self._dispatcher.dispatch(pending)
         except Exception as error:
             self._collection.record_error(error)
+            failure = error
+            raise
+        except BaseException as error:
+            failure = error
             raise
         finally:
-            await self.close()
+            try:
+                await aclose_owned(events, primary=failure)
+            finally:
+                await self.close()
 
     def with_result_collection(self) -> Self:
         """Beta: retain final messages for a result; call before consuming events."""
diff --git a/src/openai/lib/streaming/chat/_completions.py b/src/openai/lib/streaming/chat/_completions.py
index 3476314f7f..d27d7cb247 100644
--- a/src/openai/lib/streaming/chat/_completions.py
+++ b/src/openai/lib/streaming/chat/_completions.py
@@ -38,6 +38,7 @@
 from ...._streaming import Stream, AsyncStream
 from ....types.chat import ChatCompletionChunk, ParsedChatCompletion, ChatCompletionToolUnionParam
 from ...._exceptions import LengthFinishReasonError, ContentFilterFinishReasonError
+from ...._streaming_cleanup import close_owned, aclose_owned
 from ....types.chat.chat_completion import Moderation, ChoiceLogprobs
 from ....types.chat.chat_completion_chunk import Choice as ChoiceChunk
 from ....types.chat.completion_create_params import ResponseFormat as ResponseFormatParam
@@ -68,8 +69,7 @@ def __next__(self) -> ChatCompletionStreamEvent[ResponseFormatT]:
         return self._iterator.__next__()
 
     def __iter__(self) -> Iterator[ChatCompletionStreamEvent[ResponseFormatT]]:
-        for item in self._iterator:
-            yield item
+        return self._iterator
 
     def __enter__(self) -> Self:
         return self
@@ -88,7 +88,15 @@ def close(self) -> None:
 
         Automatically called if the response body is read to completion.
         """
-        self._response.close()
+        try:
+            # A running iterator belongs to its consumer; closing the response ends it.
+            if not getattr(self._iterator, "gi_running", False):
+                close_owned(self._iterator)
+        finally:
+            try:
+                close_owned(self._raw_stream)
+            finally:
+                self._response.close()
 
     def get_final_completion(self) -> ParsedChatCompletion[ResponseFormatT]:
         """Waits until the stream has been read to completion and returns
@@ -190,9 +198,8 @@ def __init__(
     async def __anext__(self) -> ChatCompletionStreamEvent[ResponseFormatT]:
         return await self._iterator.__anext__()
 
-    async def __aiter__(self) -> AsyncIterator[ChatCompletionStreamEvent[ResponseFormatT]]:
-        async for item in self._iterator:
-            yield item
+    def __aiter__(self) -> AsyncIterator[ChatCompletionStreamEvent[ResponseFormatT]]:
+        return self._iterator
 
     async def __aenter__(self) -> Self:
         return self
@@ -211,7 +218,15 @@ async def close(self) -> None:
 
         Automatically called if the response body is read to completion.
         """
-        await self._response.aclose()
+        try:
+            # A running iterator belongs to its consumer; closing the response ends it.
+            if not getattr(self._iterator, "ag_running", False):
+                await aclose_owned(self._iterator)
+        finally:
+            try:
+                await aclose_owned(self._raw_stream)
+            finally:
+                await self._response.aclose()
 
     async def get_final_completion(self) -> ParsedChatCompletion[ResponseFormatT]:
         """Waits until the stream has been read to completion and returns
diff --git a/src/openai/lib/streaming/responses/_responses.py b/src/openai/lib/streaming/responses/_responses.py
index 6d379f8e6c..8012043b87 100644
--- a/src/openai/lib/streaming/responses/_responses.py
+++ b/src/openai/lib/streaming/responses/_responses.py
@@ -24,6 +24,7 @@
     ResponseOutputItemDoneEvent,
     ResponseOutputItemAddedEvent,
 )
+from ...._streaming_cleanup import close_owned, aclose_owned
 from ..._parsing._responses import TextFormatT, parse_text, parse_response
 from ....types.responses.tool_param import ToolParam
 from ....types.responses.parsed_response import (
@@ -54,8 +55,7 @@ def __next__(self) -> ResponseStreamEvent[TextFormatT]:
         return self._iterator.__next__()
 
     def __iter__(self) -> Iterator[ResponseStreamEvent[TextFormatT]]:
-        for item in self._iterator:
-            yield item
+        return self._iterator
 
     def __enter__(self) -> Self:
         return self
@@ -81,7 +81,15 @@ def close(self) -> None:
 
         Automatically called if the response body is read to completion.
         """
-        self._response.close()
+        try:
+            # A running iterator belongs to its consumer; closing the response ends it.
+            if not getattr(self._iterator, "gi_running", False):
+                close_owned(self._iterator)
+        finally:
+            try:
+                close_owned(self._raw_stream)
+            finally:
+                self._response.close()
 
     def get_final_response(self) -> ParsedResponse[TextFormatT]:
         """Waits until the stream has been read to completion and returns
@@ -155,9 +163,8 @@ def __init__(
     async def __anext__(self) -> ResponseStreamEvent[TextFormatT]:
         return await self._iterator.__anext__()
 
-    async def __aiter__(self) -> AsyncIterator[ResponseStreamEvent[TextFormatT]]:
-        async for item in self._iterator:
-            yield item
+    def __aiter__(self) -> AsyncIterator[ResponseStreamEvent[TextFormatT]]:
+        return self._iterator
 
     async def __stream__(self) -> AsyncIterator[ResponseStreamEvent[TextFormatT]]:
         async for sse_event in self._raw_stream:
@@ -183,7 +190,15 @@ async def close(self) -> None:
 
         Automatically called if the response body is read to completion.
         """
-        await self._response.aclose()
+        try:
+            # A running iterator belongs to its consumer; closing the response ends it.
+            if not getattr(self._iterator, "ag_running", False):
+                await aclose_owned(self._iterator)
+        finally:
+            try:
+                await aclose_owned(self._raw_stream)
+            finally:
+                await self._response.aclose()
 
     async def get_final_response(self) -> ParsedResponse[TextFormatT]:
         """Waits until the stream has been read to completion and returns
diff --git a/tests/test_streaming.py b/tests/test_streaming.py
index c65c7d115f..9efef655ce 100644
--- a/tests/test_streaming.py
+++ b/tests/test_streaming.py
@@ -9,8 +9,8 @@
 import asyncio
 import importlib
 import threading
-from typing import Any, Iterator, AsyncIterator
-from contextlib import aclosing, nullcontext
+from typing import Any, Iterator, Generator, AsyncIterator
+from contextlib import aclosing, nullcontext, contextmanager
 from http.server import ThreadingHTTPServer, BaseHTTPRequestHandler
 from typing_extensions import override
 
@@ -20,6 +20,7 @@
 
 from openai import OpenAI, AsyncOpenAI, APITimeoutError, APIConnectionError, DefaultAioHttpClient
 from openai._streaming import Stream, AsyncStream, ServerSentEvent
+from openai.types.chat import ChatCompletionChunk
 
 
 @pytest.fixture(
@@ -586,15 +587,12 @@ def body() -> Iterator[bytes]:
         assert response.is_closed
 
 
-@pytest.mark.asyncio
-@pytest.mark.parametrize("early_exit", [False, True], ids=["read-to-done", "early-exit"])
-async def test_async_stream_closes_its_iterators(early_exit: bool, async_client: AsyncOpenAI) -> None:
-    def body() -> Iterator[bytes]:
-        yield b'data: {"foo":true}\n\n'
-        yield b'data: {"bar":false}\n\n'
-        yield b"data: [DONE]\n\n"
+@contextmanager
+def gc_finalized_async_generators() -> Generator[list[str], None, None]:
+    """Record async generators that reach the GC finalizer instead of being closed.
 
-    stream = AsyncStream(cast_to=object, client=async_client, response=httpx2.Response(200, content=to_aiter(body())))
+    Generators bind the finalizer on first iteration, so enter this before iterating.
+    """
     collected: list[str] = []
     firstiter, finalizer = sys.get_asyncgen_hooks()
 
@@ -603,21 +601,150 @@ def record(agen: Any) -> None:
         if finalizer is not None:
             finalizer(agen)
 
-    # Generators bind the finalizer when first iterated, so install it before iterating.
     sys.set_asyncgen_hooks(firstiter=firstiter, finalizer=record)
     try:
+        yield collected
+    finally:
+        sys.set_asyncgen_hooks(firstiter=firstiter, finalizer=finalizer)
+
+
+# Generators the SDK opens; httpx2-internal ones are owned by httpx2.
+SDK_OWNED = ("AsyncStream.", "SSEDecoder.", "Response.aiter_bytes", "AsyncChatCompletionStream.")
+
+
+def sse_body() -> Iterator[bytes]:
+    yield b'data: {"foo":true}\n\n'
+    yield b'data: {"bar":false}\n\n'
+    yield b"data: [DONE]\n\n"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("early_exit", [False, True], ids=["read-to-done", "early-exit"])
+async def test_async_stream_closes_its_iterators(early_exit: bool, async_client: AsyncOpenAI) -> None:
+    stream = AsyncStream(
+        cast_to=object, client=async_client, response=httpx2.Response(200, content=to_aiter(sse_body()))
+    )
+    with gc_finalized_async_generators() as collected:
         async for _ in stream:
             if early_exit:
                 break
         await stream.close()
         del stream
         gc.collect()
-    finally:
-        sys.set_asyncgen_hooks(firstiter=firstiter, finalizer=finalizer)
 
-    # Iterators the stream opened must be closed by the stream, not left to the
-    # garbage collector, where asyncio closes nested generators concurrently.
-    assert [name for name in collected if name.startswith(("AsyncStream.", "SSEDecoder."))] == []
+    assert [name for name in collected if name.startswith(SDK_OWNED)] == []
+
+
+@pytest.mark.asyncio
+async def test_async_chat_completion_stream_closes_raw_stream_iterators(async_client: AsyncOpenAI) -> None:
+    from openai._types import omit
+    from openai.lib.streaming.chat import AsyncChatCompletionStream
+
+    chunk = b'data: {"id":"c","object":"chat.completion.chunk","created":0,"model":"m","choices":[{"index":0,"delta":{"content":"hi"},"finish_reason":null}]}\n\n'
+    raw = AsyncStream(
+        cast_to=ChatCompletionChunk,
+        client=async_client,
+        response=httpx2.Response(200, content=to_aiter(iter([chunk, chunk, b"data: [DONE]\n\n"]))),
+    )
+    stream = AsyncChatCompletionStream(raw_stream=raw, response_format=omit, input_tools=omit)
+    with gc_finalized_async_generators() as collected:
+        async for _ in stream:
+            break
+        await stream.close()
+        del stream, raw
+        gc.collect()
+
+    assert [name for name in collected if name.startswith(SDK_OWNED)] == []
+
+
+class _TrackedEvents:
+    """A decoder iterator that is not a generator and must be closed exactly once."""
+
+    def __init__(self, fail_after: int | None = None, close_error: bool = False) -> None:
+        self.closes = 0
+        self._sent = 0
+        self._fail_after = fail_after
+        self._close_error = close_error
+
+    def _next(self) -> ServerSentEvent:
+        if self._fail_after is not None and self._sent >= self._fail_after:
+            raise ValueError("decoder failed")
+        self._sent += 1
+        return ServerSentEvent(data='{"foo":true}')
+
+    def __iter__(self) -> "_TrackedEvents":
+        return self
+
+    def __next__(self) -> ServerSentEvent:
+        return self._next()
+
+    def close(self) -> None:
+        self.closes += 1
+        if self._close_error:
+            raise RuntimeError("close failed")
+
+    def __aiter__(self) -> "_TrackedEvents":
+        return self
+
+    async def __anext__(self) -> ServerSentEvent:
+        return self._next()
+
+    async def aclose(self) -> None:
+        self.close()
+
+
+class _TrackedDecoder:
+    def __init__(self, events: _TrackedEvents) -> None:
+        self.events = events
+
+    def iter_bytes(self, iterator: Iterator[bytes]) -> Iterator[ServerSentEvent]:  # noqa: ARG002
+        return self.events
+
+    def aiter_bytes(self, iterator: AsyncIterator[bytes]) -> AsyncIterator[ServerSentEvent]:  # noqa: ARG002
+        return self.events
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"])
+async def test_custom_decoder_iterator_is_closed_once(sync: bool, client: OpenAI, async_client: AsyncOpenAI) -> None:
+    events = _TrackedEvents()
+    if sync:
+        stream = Stream(cast_to=object, client=client, response=httpx2.Response(200, content=sse_body()))
+        stream._decoder = _TrackedDecoder(events)
+        next(iter(stream))
+        stream.close()
+    else:
+        async_stream = AsyncStream(
+            cast_to=object, client=async_client, response=httpx2.Response(200, content=to_aiter(sse_body()))
+        )
+        async_stream._decoder = _TrackedDecoder(events)
+        await async_stream.__anext__()
+        await async_stream.close()
+
+    assert events.closes == 1
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"])
+async def test_cleanup_error_does_not_replace_stream_error(
+    sync: bool, client: OpenAI, async_client: AsyncOpenAI
+) -> None:
+    events = _TrackedEvents(fail_after=1, close_error=True)
+    with pytest.raises(ValueError, match="decoder failed"):
+        if sync:
+            stream = Stream(cast_to=object, client=client, response=httpx2.Response(200, content=sse_body()))
+            stream._decoder = _TrackedDecoder(events)
+            for _ in stream:
+                pass
+        else:
+            async_stream = AsyncStream(
+                cast_to=object, client=async_client, response=httpx2.Response(200, content=to_aiter(sse_body()))
+            )
+            async_stream._decoder = _TrackedDecoder(events)
+            async for _ in async_stream:
+                pass
+
+    assert events.closes == 1
 
 
 async def to_aiter(iter: Iterator[bytes]) -> AsyncIterator[bytes]:

From 30f391ee90a78ac3ce40a247528d12d405d8e845 Mon Sep 17 00:00:00 2001
From: Xinmin Zeng <135568692+fancyboi999@users.noreply.github.com>
Date: Fri, 23 Oct 2026 22:22:32 -0700
Subject: [PATCH 3/7] fix(streaming): keep cleanup errors visible and close
 every resource

- GeneratorExit from close() or [DONE] is normal closure; it no longer
  counts as a primary failure that silences cleanup errors.
- close_owned/aclose_owned look up and call each closer inside the same
  handler, attempt every resource even after a BaseException, and never
  drop a cancellation raised during cleanup in favor of an ordinary
  error.
- The response joins the same cleanup call as the iterators, so it is
  shielded and follows the same error precedence.
- Comments no longer claim that closing the response ends a concurrent
  consumer.
- Tests check [DONE] cleanup without an explicit close(), cleanup errors
  surfacing on normal closure, and aclose_owned continuing after
  cancellation.
---
 src/openai/_streaming.py                      | 42 ++++++------
 src/openai/_streaming_cleanup.py              | 64 +++++++++++-------
 src/openai/lib/beta/agents/_stream.py         | 14 ++--
 src/openai/lib/streaming/chat/_completions.py | 26 +++-----
 .../lib/streaming/responses/_responses.py     | 26 +++-----
 tests/test_streaming.py                       | 65 ++++++++++++++++++-
 6 files changed, 146 insertions(+), 91 deletions(-)

diff --git a/src/openai/_streaming.py b/src/openai/_streaming.py
index 9d2c17cbd4..9265922c05 100644
--- a/src/openai/_streaming.py
+++ b/src/openai/_streaming.py
@@ -67,6 +67,8 @@ def _iter_events(self) -> Iterator[ServerSentEvent]:
                 raise APITimeoutError(request=self.response.request) from err
             except request_exceptions() as err:
                 raise APIConnectionError(request=self.response.request) from err
+        except GeneratorExit:
+            raise  # Normal closure, not a failure that should outrank cleanup errors.
         except BaseException as err:
             failure = err
             raise
@@ -128,16 +130,15 @@ def __stream__(self) -> Iterator[_T]:
                         cast_to=cast_to,
                         response=response,
                     )
+        except GeneratorExit:
+            raise  # Normal closure, not a failure that should outrank cleanup errors.
         except BaseException as err:
             failure = err
             raise
         finally:
             # Close the event iterator explicitly; stopping at [DONE] leaves it suspended.
-            try:
-                close_owned(iterator, primary=failure)
-            finally:
-                # Ensure the response is closed even if the consumer doesn't read all data
-                response.close()
+            # Ensure the response is closed even if the consumer doesn't read all data.
+            close_owned(iterator, response, primary=failure)
 
     def __enter__(self) -> Self:
         return self
@@ -156,12 +157,10 @@ def close(self) -> None:
 
         Automatically called if the response body is read to completion.
         """
-        try:
-            # A running iterator belongs to its consumer; closing the response ends it.
-            if not getattr(self._iterator, "gi_running", False):
-                close_owned(self._iterator)
-        finally:
-            self.response.close()
+        # Concurrent close does not cancel or join an active consumer; that consumer
+        # closes the iterator when it unwinds.
+        running = getattr(self._iterator, "gi_running", False)
+        close_owned(None if running else self._iterator, self.response)
 
 
 class AsyncStream(Generic[_T]):
@@ -206,6 +205,8 @@ async def _iter_events(self) -> AsyncIterator[ServerSentEvent]:
                 raise APITimeoutError(request=self.response.request) from err
             except request_exceptions() as err:
                 raise APIConnectionError(request=self.response.request) from err
+        except GeneratorExit:
+            raise  # Normal closure, not a failure that should outrank cleanup errors.
         except BaseException as err:
             failure = err
             raise
@@ -272,6 +273,8 @@ async def __stream__(self) -> AsyncIterator[_T]:
                         cast_to=cast_to,
                         response=response,
                     )
+        except GeneratorExit:
+            raise  # Normal closure, not a failure that should outrank cleanup errors.
         except BaseException as err:
             failure = err
             raise
@@ -279,11 +282,8 @@ async def __stream__(self) -> AsyncIterator[_T]:
             # Close the event iterator explicitly. Stopping at [DONE] leaves it and the
             # httpx2 byte iterators beneath it suspended; left to the garbage collector,
             # asyncio closes those nested generators concurrently.
-            try:
-                await aclose_owned(iterator, primary=failure)
-            finally:
-                # Ensure the response is closed even if the consumer doesn't read all data
-                await response.aclose()
+            # Ensure the response is closed even if the consumer doesn't read all data.
+            await aclose_owned(iterator, response, primary=failure)
 
     async def __aenter__(self) -> Self:
         return self
@@ -302,12 +302,10 @@ async def close(self) -> None:
 
         Automatically called if the response body is read to completion.
         """
-        try:
-            # A running iterator belongs to its consumer; closing the response ends it.
-            if not getattr(self._iterator, "ag_running", False):
-                await aclose_owned(self._iterator)
-        finally:
-            await self.response.aclose()
+        # Concurrent close does not cancel or join an active consumer; that consumer
+        # closes the iterator when it unwinds.
+        running = getattr(self._iterator, "ag_running", False)
+        await aclose_owned(None if running else self._iterator, self.response)
 
     async def aclose(self) -> None:
         """Close the response and release the connection. Alias for `close()`."""
diff --git a/src/openai/_streaming_cleanup.py b/src/openai/_streaming_cleanup.py
index 79467e65c8..6b57eac920 100644
--- a/src/openai/_streaming_cleanup.py
+++ b/src/openai/_streaming_cleanup.py
@@ -29,40 +29,58 @@ async def drain_response_async(response: httpx2.Response) -> None:
             await response.aclose()
 
 
-def close_owned(*iterators: object, primary: Optional[BaseException] = None) -> None:
-    """Close iterators this caller opened, in order.
+def _cleanup_error(primary: Optional[BaseException], failure: Optional[BaseException]) -> Optional[BaseException]:
+    """Pick the cleanup failure to raise, if any.
 
-    Every iterator gets its close attempt. Objects without ``close()`` are skipped.
-    A cleanup error is raised only when no ``primary`` error is already propagating.
+    A failure already propagating wins, except that a control-flow exception raised
+    during cleanup (for example cancellation) is never dropped in favor of an ordinary
+    error.
+    """
+    if failure is None:
+        return None
+    if primary is None:
+        return failure
+    if not isinstance(failure, Exception) and isinstance(primary, Exception):
+        return failure
+    return None
+
+
+def close_owned(*resources: object, primary: Optional[BaseException] = None) -> None:
+    """Close resources this caller owns, in order.
+
+    Every resource gets its close attempt; objects without ``close()`` are skipped.
+    The first cleanup failure is raised after all attempts unless ``primary``, the
+    failure already propagating, takes precedence (see ``_cleanup_error``).
     """
     failure: Optional[BaseException] = None
-    for iterator in iterators:
-        close = getattr(iterator, "close", None)
-        if close is None:
-            continue
+    for resource in resources:
         try:
-            close()
-        except Exception as err:
+            close = getattr(resource, "close", None)
+            if close is not None:
+                close()
+        except BaseException as err:  # re-raised below, after the remaining closes
             failure = failure or err
-    if failure is not None and primary is None:
-        raise failure
+    error = _cleanup_error(primary, failure)
+    if error is not None:
+        raise error
 
 
-async def aclose_owned(*iterators: object, primary: Optional[BaseException] = None) -> None:
+async def aclose_owned(*resources: object, primary: Optional[BaseException] = None) -> None:
     """Async counterpart of ``close_owned``.
 
-    Cleanup is shielded from caller cancellation so an already-cancelled scope cannot
-    leave a generator closed with its cleanup half done; cancellation still propagates.
+    The closes are shielded from AnyIO cancellation so an already-cancelled scope cannot
+    leave a resource half closed; cancellation still propagates afterwards. The shield
+    has no deadline, so a closer that never returns blocks the caller.
     """
     failure: Optional[BaseException] = None
     with anyio.CancelScope(shield=True):
-        for iterator in iterators:
-            aclose = getattr(iterator, "aclose", None)
-            if aclose is None:
-                continue
+        for resource in resources:
             try:
-                await aclose()
-            except Exception as err:
+                aclose = getattr(resource, "aclose", None)
+                if aclose is not None:
+                    await aclose()
+            except BaseException as err:  # re-raised below, after the remaining closes
                 failure = failure or err
-    if failure is not None and primary is None:
-        raise failure
+    error = _cleanup_error(primary, failure)
+    if error is not None:
+        raise error
diff --git a/src/openai/lib/beta/agents/_stream.py b/src/openai/lib/beta/agents/_stream.py
index cdaf6e94a3..a67eec39cb 100644
--- a/src/openai/lib/beta/agents/_stream.py
+++ b/src/openai/lib/beta/agents/_stream.py
@@ -40,14 +40,13 @@ def __stream__(self) -> Iterator[AgentSessionEvent]:
             self._collection.record_error(error)
             failure = error
             raise
+        except GeneratorExit:
+            raise  # Normal closure, not a failure that should outrank cleanup errors.
         except BaseException as error:
             failure = error
             raise
         finally:
-            try:
-                close_owned(events, primary=failure)
-            finally:
-                self.close()
+            close_owned(events, self, primary=failure)
 
     def with_result_collection(self) -> Self:
         """Beta: retain final messages for a result; call before consuming events."""
@@ -104,14 +103,13 @@ async def __stream__(self) -> AsyncIterator[AgentSessionEvent]:
             self._collection.record_error(error)
             failure = error
             raise
+        except GeneratorExit:
+            raise  # Normal closure, not a failure that should outrank cleanup errors.
         except BaseException as error:
             failure = error
             raise
         finally:
-            try:
-                await aclose_owned(events, primary=failure)
-            finally:
-                await self.close()
+            await aclose_owned(events, self, primary=failure)
 
     def with_result_collection(self) -> Self:
         """Beta: retain final messages for a result; call before consuming events."""
diff --git a/src/openai/lib/streaming/chat/_completions.py b/src/openai/lib/streaming/chat/_completions.py
index d27d7cb247..b434240af0 100644
--- a/src/openai/lib/streaming/chat/_completions.py
+++ b/src/openai/lib/streaming/chat/_completions.py
@@ -88,15 +88,10 @@ def close(self) -> None:
 
         Automatically called if the response body is read to completion.
         """
-        try:
-            # A running iterator belongs to its consumer; closing the response ends it.
-            if not getattr(self._iterator, "gi_running", False):
-                close_owned(self._iterator)
-        finally:
-            try:
-                close_owned(self._raw_stream)
-            finally:
-                self._response.close()
+        # Concurrent close does not cancel or join an active consumer; that consumer
+        # closes the iterator when it unwinds.
+        running = getattr(self._iterator, "gi_running", False)
+        close_owned(None if running else self._iterator, self._raw_stream, self._response)
 
     def get_final_completion(self) -> ParsedChatCompletion[ResponseFormatT]:
         """Waits until the stream has been read to completion and returns
@@ -218,15 +213,10 @@ async def close(self) -> None:
 
         Automatically called if the response body is read to completion.
         """
-        try:
-            # A running iterator belongs to its consumer; closing the response ends it.
-            if not getattr(self._iterator, "ag_running", False):
-                await aclose_owned(self._iterator)
-        finally:
-            try:
-                await aclose_owned(self._raw_stream)
-            finally:
-                await self._response.aclose()
+        # Concurrent close does not cancel or join an active consumer; that consumer
+        # closes the iterator when it unwinds.
+        running = getattr(self._iterator, "ag_running", False)
+        await aclose_owned(None if running else self._iterator, self._raw_stream, self._response)
 
     async def get_final_completion(self) -> ParsedChatCompletion[ResponseFormatT]:
         """Waits until the stream has been read to completion and returns
diff --git a/src/openai/lib/streaming/responses/_responses.py b/src/openai/lib/streaming/responses/_responses.py
index 8012043b87..dd079532ef 100644
--- a/src/openai/lib/streaming/responses/_responses.py
+++ b/src/openai/lib/streaming/responses/_responses.py
@@ -81,15 +81,10 @@ def close(self) -> None:
 
         Automatically called if the response body is read to completion.
         """
-        try:
-            # A running iterator belongs to its consumer; closing the response ends it.
-            if not getattr(self._iterator, "gi_running", False):
-                close_owned(self._iterator)
-        finally:
-            try:
-                close_owned(self._raw_stream)
-            finally:
-                self._response.close()
+        # Concurrent close does not cancel or join an active consumer; that consumer
+        # closes the iterator when it unwinds.
+        running = getattr(self._iterator, "gi_running", False)
+        close_owned(None if running else self._iterator, self._raw_stream, self._response)
 
     def get_final_response(self) -> ParsedResponse[TextFormatT]:
         """Waits until the stream has been read to completion and returns
@@ -190,15 +185,10 @@ async def close(self) -> None:
 
         Automatically called if the response body is read to completion.
         """
-        try:
-            # A running iterator belongs to its consumer; closing the response ends it.
-            if not getattr(self._iterator, "ag_running", False):
-                await aclose_owned(self._iterator)
-        finally:
-            try:
-                await aclose_owned(self._raw_stream)
-            finally:
-                await self._response.aclose()
+        # Concurrent close does not cancel or join an active consumer; that consumer
+        # closes the iterator when it unwinds.
+        running = getattr(self._iterator, "ag_running", False)
+        await aclose_owned(None if running else self._iterator, self._raw_stream, self._response)
 
     async def get_final_response(self) -> ParsedResponse[TextFormatT]:
         """Waits until the stream has been read to completion and returns
diff --git a/tests/test_streaming.py b/tests/test_streaming.py
index 9efef655ce..747ea3396d 100644
--- a/tests/test_streaming.py
+++ b/tests/test_streaming.py
@@ -628,7 +628,9 @@ async def test_async_stream_closes_its_iterators(early_exit: bool, async_client:
         async for _ in stream:
             if early_exit:
                 break
-        await stream.close()
+        if early_exit:
+            await stream.close()
+        # Reading to [DONE] must clean up without an explicit close().
         del stream
         gc.collect()
 
@@ -660,16 +662,19 @@ async def test_async_chat_completion_stream_closes_raw_stream_iterators(async_cl
 class _TrackedEvents:
     """A decoder iterator that is not a generator and must be closed exactly once."""
 
-    def __init__(self, fail_after: int | None = None, close_error: bool = False) -> None:
+    def __init__(self, fail_after: int | None = None, done_after: int | None = None, close_error: bool = False) -> None:
         self.closes = 0
         self._sent = 0
         self._fail_after = fail_after
+        self._done_after = done_after
         self._close_error = close_error
 
     def _next(self) -> ServerSentEvent:
         if self._fail_after is not None and self._sent >= self._fail_after:
             raise ValueError("decoder failed")
         self._sent += 1
+        if self._done_after is not None and self._sent > self._done_after:
+            return ServerSentEvent(data="[DONE]")
         return ServerSentEvent(data='{"foo":true}')
 
     def __iter__(self) -> "_TrackedEvents":
@@ -747,6 +752,62 @@ async def test_cleanup_error_does_not_replace_stream_error(
     assert events.closes == 1
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"])
+@pytest.mark.parametrize("ending", ["done", "close"])
+async def test_cleanup_error_surfaces_without_other_error(
+    sync: bool, ending: str, client: OpenAI, async_client: AsyncOpenAI
+) -> None:
+    # Normal closure (generator close() or [DONE]) must not hide a cleanup failure.
+    events = _TrackedEvents(done_after=1, close_error=True)
+    with pytest.raises(RuntimeError, match="close failed"):
+        if sync:
+            stream = Stream(cast_to=object, client=client, response=httpx2.Response(200, content=sse_body()))
+            stream._decoder = _TrackedDecoder(events)
+            if ending == "done":
+                for _ in stream:
+                    pass
+            else:
+                next(iter(stream))
+                stream.close()
+        else:
+            async_stream = AsyncStream(
+                cast_to=object, client=async_client, response=httpx2.Response(200, content=to_aiter(sse_body()))
+            )
+            async_stream._decoder = _TrackedDecoder(events)
+            if ending == "done":
+                async for _ in async_stream:
+                    pass
+            else:
+                await async_stream.__anext__()
+                await async_stream.close()
+
+    assert events.closes == 1
+
+
+@pytest.mark.asyncio
+async def test_aclose_owned_closes_every_resource_and_keeps_cancellation() -> None:
+    from openai._streaming_cleanup import aclose_owned
+
+    closed: list[str] = []
+
+    class Closer:
+        def __init__(self, name: str, error: BaseException | None = None) -> None:
+            self.name = name
+            self.error = error
+
+        async def aclose(self) -> None:
+            closed.append(self.name)
+            if self.error is not None:
+                raise self.error
+
+    cancelled = asyncio.CancelledError()
+    with pytest.raises(asyncio.CancelledError):
+        await aclose_owned(Closer("first", cancelled), object(), Closer("second"), primary=ValueError("primary"))
+
+    assert closed == ["first", "second"]
+
+
 async def to_aiter(iter: Iterator[bytes]) -> AsyncIterator[bytes]:
     for chunk in iter:
         yield chunk

From 26c2d12449a5f14daf9f891b5d3d2f11365d179c Mon Sep 17 00:00:00 2001
From: Xinmin Zeng <135568692+fancyboi999@users.noreply.github.com>
Date: Fri, 23 Oct 2026 22:41:42 -0700
Subject: [PATCH 4/7] fix(streaming): keep cancellation and agent result errors
 over cleanup errors

- Cleanup helpers let a later control-flow exception (for example
  cancellation) replace an earlier ordinary cleanup failure.
- Agent streams pass the propagating error into cleanup in
  get_final_result(), the event iterator and failed entry, so a decoder
  close failure no longer replaces AgentTurnResultError.
- run_cleanup() covers owners whose async close method is not aclose().
- Tests cover cancellation after an ordinary cleanup error and a failed
  agent turn whose decoder close also fails.
---
 src/openai/_streaming_cleanup.py            | 29 ++++++-
 src/openai/lib/beta/agents/_stream.py       | 60 ++++++++------
 src/openai/lib/streaming/agents/_streams.py | 87 ++++++++++++++-------
 tests/lib/streaming/agents/test_results.py  | 45 +++++++++++
 tests/test_streaming.py                     | 35 +++++++++
 5 files changed, 199 insertions(+), 57 deletions(-)

diff --git a/src/openai/_streaming_cleanup.py b/src/openai/_streaming_cleanup.py
index 6b57eac920..8ec298bc8c 100644
--- a/src/openai/_streaming_cleanup.py
+++ b/src/openai/_streaming_cleanup.py
@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from typing import Optional, AsyncIterator, cast
+from typing import Any, Callable, Optional, Awaitable, AsyncIterator, cast
 
 import anyio
 import httpx2
@@ -45,6 +45,13 @@ def _cleanup_error(primary: Optional[BaseException], failure: Optional[BaseExcep
     return None
 
 
+def _keep(failure: Optional[BaseException], err: BaseException) -> BaseException:
+    """Keep the first cleanup failure, but let a later control-flow exception replace an ordinary one."""
+    if failure is None or (isinstance(failure, Exception) and not isinstance(err, Exception)):
+        return err
+    return failure
+
+
 def close_owned(*resources: object, primary: Optional[BaseException] = None) -> None:
     """Close resources this caller owns, in order.
 
@@ -59,7 +66,7 @@ def close_owned(*resources: object, primary: Optional[BaseException] = None) ->
             if close is not None:
                 close()
         except BaseException as err:  # re-raised below, after the remaining closes
-            failure = failure or err
+            failure = _keep(failure, err)
     error = _cleanup_error(primary, failure)
     if error is not None:
         raise error
@@ -80,7 +87,23 @@ async def aclose_owned(*resources: object, primary: Optional[BaseException] = No
                 if aclose is not None:
                     await aclose()
             except BaseException as err:  # re-raised below, after the remaining closes
-                failure = failure or err
+                failure = _keep(failure, err)
+    error = _cleanup_error(primary, failure)
+    if error is not None:
+        raise error
+
+
+async def run_cleanup(cleanup: Callable[[], Awaitable[Any]], *, primary: Optional[BaseException] = None) -> None:
+    """Run one async cleanup callable with the shield and error precedence of ``aclose_owned``.
+
+    For owners whose async close method is not named ``aclose``.
+    """
+    failure: Optional[BaseException] = None
+    with anyio.CancelScope(shield=True):
+        try:
+            await cleanup()
+        except BaseException as err:  # re-raised below according to precedence
+            failure = err
     error = _cleanup_error(primary, failure)
     if error is not None:
         raise error
diff --git a/src/openai/lib/beta/agents/_stream.py b/src/openai/lib/beta/agents/_stream.py
index a67eec39cb..c849e36559 100644
--- a/src/openai/lib/beta/agents/_stream.py
+++ b/src/openai/lib/beta/agents/_stream.py
@@ -62,21 +62,27 @@ def get_final_result(self) -> AgentTurnResult[OutputT]:
         the cached result without consuming more events.
         """
         collector = self._collection.enable()
+        failure: BaseException | None = None
         try:
-            collector.check_outcome(self._dispatcher.handlers if self._dispatcher is not None else ())
-            if not collector.is_done():
-                for _ in self:
-                    collector.check_outcome(self._dispatcher.handlers if self._dispatcher is not None else ())
-                    if collector.is_done():
-                        break
-            return self._collection.result()
-        except (AgentTurnResultError, AgentOutputParseError):
+            try:
+                collector.check_outcome(self._dispatcher.handlers if self._dispatcher is not None else ())
+                if not collector.is_done():
+                    for _ in self:
+                        collector.check_outcome(self._dispatcher.handlers if self._dispatcher is not None else ())
+                        if collector.is_done():
+                            break
+                return self._collection.result()
+            except (AgentTurnResultError, AgentOutputParseError):
+                raise
+            except Exception as error:
+                collector.cause = error
+                raise collector.error("observation_failed") from error
+        except BaseException as error:
+            failure = error
             raise
-        except Exception as error:
-            collector.cause = error
-            raise collector.error("observation_failed") from error
         finally:
-            self.close()
+            # A cleanup error must not replace the result error being raised.
+            close_owned(self, primary=failure)
 
 
 class AsyncAgentSessionEventStream(AsyncStream[AgentSessionEvent], Generic[OutputT]):
@@ -123,18 +129,24 @@ async def get_final_result(self) -> AgentTurnResult[OutputT]:
         with_result_collection() before consuming events.
         """
         collector = self._collection.enable()
+        failure: BaseException | None = None
         try:
-            collector.check_outcome(self._dispatcher.handlers if self._dispatcher is not None else ())
-            if not collector.is_done():
-                async for _ in self:
-                    collector.check_outcome(self._dispatcher.handlers if self._dispatcher is not None else ())
-                    if collector.is_done():
-                        break
-            return self._collection.result()
-        except (AgentTurnResultError, AgentOutputParseError):
+            try:
+                collector.check_outcome(self._dispatcher.handlers if self._dispatcher is not None else ())
+                if not collector.is_done():
+                    async for _ in self:
+                        collector.check_outcome(self._dispatcher.handlers if self._dispatcher is not None else ())
+                        if collector.is_done():
+                            break
+                return self._collection.result()
+            except (AgentTurnResultError, AgentOutputParseError):
+                raise
+            except Exception as error:
+                collector.cause = error
+                raise collector.error("observation_failed") from error
+        except BaseException as error:
+            failure = error
             raise
-        except Exception as error:
-            collector.cause = error
-            raise collector.error("observation_failed") from error
         finally:
-            await self.close()
+            # A cleanup error must not replace the result error being raised.
+            await aclose_owned(self, primary=failure)
diff --git a/src/openai/lib/streaming/agents/_streams.py b/src/openai/lib/streaming/agents/_streams.py
index b222574295..bc46c307b6 100644
--- a/src/openai/lib/streaming/agents/_streams.py
+++ b/src/openai/lib/streaming/agents/_streams.py
@@ -12,6 +12,7 @@
 from ...._types import Omit, Headers, NotGiven, omit, not_given
 from ._dispatch import ToolDispatcher, AsyncToolDispatcher, _request_options
 from ...._streaming import Stream, AsyncStream
+from ...._streaming_cleanup import close_owned, run_cleanup
 from ...beta.agents._result import (
     OutputT,
     AgentTurnResult,
@@ -146,8 +147,8 @@ def __enter__(self) -> Self:
             self._sessions.events.create(
                 self._session_id, events=[self._input], idempotency_key=self._idempotency_key, **self._options
             )
-        except BaseException:
-            self.close()
+        except BaseException as error:
+            close_owned(self, primary=error)
             raise
         self._iterator = self._iterate()
         return self
@@ -186,21 +187,27 @@ def get_final_result(self) -> AgentTurnResult[OutputT]:
         To iterate first, call with_result_collection() before consuming events.
         """
         collector = self._collection.enable()
+        failure: BaseException | None = None
         try:
-            collector.check_outcome(self._handlers)
-            if not collector.is_done():
-                for _ in self:
-                    collector.check_outcome(self._handlers)
-                    if collector.is_done():
-                        break
-            return self._collection.result()
-        except (AgentTurnResultError, AgentOutputParseError):
+            try:
+                collector.check_outcome(self._handlers)
+                if not collector.is_done():
+                    for _ in self:
+                        collector.check_outcome(self._handlers)
+                        if collector.is_done():
+                            break
+                return self._collection.result()
+            except (AgentTurnResultError, AgentOutputParseError):
+                raise
+            except Exception as error:
+                self._collection.record_error(error)
+                raise collector.error("observation_failed") from error
+        except BaseException as error:
+            failure = error
             raise
-        except Exception as error:
-            self._collection.record_error(error)
-            raise collector.error("observation_failed") from error
         finally:
-            self.close()
+            # A cleanup error must not replace the result error being raised.
+            close_owned(self, primary=failure)
 
     def close(self) -> None:
         """Close the event connection without cancelling the backend turn."""
@@ -210,6 +217,7 @@ def close(self) -> None:
 
     def _iterate(self) -> Iterator[AgentSessionEvent]:
         assert self._stream is not None
+        failure: BaseException | None = None
         try:
             for event in self._stream:
                 if not self._state.accept(event):
@@ -227,9 +235,15 @@ def _iterate(self) -> Iterator[AgentSessionEvent]:
             raise RuntimeError("Session event stream ended before the turn reached idle or failed")
         except Exception as error:
             self._collection.record_error(error)
+            failure = error
+            raise
+        except GeneratorExit:
+            raise  # Normal closure, not a failure that should outrank cleanup errors.
+        except BaseException as error:
+            failure = error
             raise
         finally:
-            self.close()
+            close_owned(self, primary=failure)
 
 
 class AsyncAgentSessionStream(Generic[OutputT]):
@@ -284,8 +298,8 @@ async def __aenter__(self) -> Self:
             await self._sessions.events.create(
                 self._session_id, events=[self._input], idempotency_key=self._idempotency_key, **self._options
             )
-        except BaseException:
-            await self.close()
+        except BaseException as error:
+            await run_cleanup(self.close, primary=error)
             raise
         self._iterator = self._iterate()
         return self
@@ -324,21 +338,27 @@ async def get_final_result(self) -> AgentTurnResult[OutputT]:
         To iterate first, call with_result_collection() before consuming events.
         """
         collector = self._collection.enable()
+        failure: BaseException | None = None
         try:
-            collector.check_outcome(self._handlers)
-            if not collector.is_done():
-                async for _ in self:
-                    collector.check_outcome(self._handlers)
-                    if collector.is_done():
-                        break
-            return self._collection.result()
-        except (AgentTurnResultError, AgentOutputParseError):
+            try:
+                collector.check_outcome(self._handlers)
+                if not collector.is_done():
+                    async for _ in self:
+                        collector.check_outcome(self._handlers)
+                        if collector.is_done():
+                            break
+                return self._collection.result()
+            except (AgentTurnResultError, AgentOutputParseError):
+                raise
+            except Exception as error:
+                self._collection.record_error(error)
+                raise collector.error("observation_failed") from error
+        except BaseException as error:
+            failure = error
             raise
-        except Exception as error:
-            self._collection.record_error(error)
-            raise collector.error("observation_failed") from error
         finally:
-            await self.close()
+            # A cleanup error must not replace the result error being raised.
+            await run_cleanup(self.close, primary=failure)
 
     async def close(self) -> None:
         """Close the event connection without cancelling the backend turn."""
@@ -348,6 +368,7 @@ async def close(self) -> None:
 
     async def _iterate(self) -> AsyncIterator[AgentSessionEvent]:
         assert self._stream is not None
+        failure: BaseException | None = None
         try:
             async for event in self._stream:
                 if not self._state.accept(event):
@@ -365,6 +386,12 @@ async def _iterate(self) -> AsyncIterator[AgentSessionEvent]:
             raise RuntimeError("Session event stream ended before the turn reached idle or failed")
         except Exception as error:
             self._collection.record_error(error)
+            failure = error
+            raise
+        except GeneratorExit:
+            raise  # Normal closure, not a failure that should outrank cleanup errors.
+        except BaseException as error:
+            failure = error
             raise
         finally:
-            await self.close()
+            await run_cleanup(self.close, primary=failure)
diff --git a/tests/lib/streaming/agents/test_results.py b/tests/lib/streaming/agents/test_results.py
index abca51d3b3..9d63355f35 100644
--- a/tests/lib/streaming/agents/test_results.py
+++ b/tests/lib/streaming/agents/test_results.py
@@ -8,6 +8,7 @@
 import pytest
 
 from openai import OpenAI, AsyncOpenAI
+from openai._streaming import SSEDecoder
 from openai.lib.beta.agents import AgentTurnResult, AgentTurnResultError
 from openai.lib.streaming.agents import AgentSessionStream, AsyncAgentSessionStream
 from tests.lib.streaming.agents.test_streams import Server, EventBody, sdk as sdk, call, idle, session, turn_event
@@ -195,6 +196,50 @@ async def test_unsuccessful_turn(sdk: OpenAI | AsyncOpenAI, server: ResultServer
     assert server.body.closed
 
 
+class _CloseFailingDecoder(SSEDecoder):
+    """Decodes normally; closing its event iterator fails."""
+
+    class _Events:
+        def __init__(self, inner: Any) -> None:
+            self._inner = inner
+
+        def __iter__(self) -> Any:
+            return self
+
+        def __next__(self) -> Any:
+            return next(self._inner)
+
+        def __aiter__(self) -> Any:
+            return self
+
+        async def __anext__(self) -> Any:
+            return await self._inner.__anext__()
+
+        def close(self) -> None:
+            raise RuntimeError("decoder close failed")
+
+        async def aclose(self) -> None:
+            raise RuntimeError("decoder close failed")
+
+    @override
+    def iter_bytes(self, iterator: Iterator[bytes]) -> Any:
+        return self._Events(super().iter_bytes(iterator))
+
+    @override
+    def aiter_bytes(self, iterator: Any) -> Any:
+        return self._Events(super().aiter_bytes(iterator))
+
+
+async def test_cleanup_error_does_not_replace_result_error(
+    sdk: OpenAI | AsyncOpenAI, server: ResultServer, creation: bool, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(sdk, "_make_sse_decoder", lambda: _CloseFailingDecoder())
+    server.body = EventBody([turn_event("created"), message(), turn_event("failed"), idle()])
+    with pytest.raises(AgentTurnResultError) as exc:
+        await collect(sdk, creation)
+    assert exc.value.reason == "failed"
+
+
 @pytest.mark.parametrize("tail", [[], [turn_event("completed")]])
 async def test_incomplete_answer(
     sdk: OpenAI | AsyncOpenAI, server: ResultServer, creation: bool, tail: list[dict[str, Any]]
diff --git a/tests/test_streaming.py b/tests/test_streaming.py
index 747ea3396d..6b9d8bdbab 100644
--- a/tests/test_streaming.py
+++ b/tests/test_streaming.py
@@ -808,6 +808,41 @@ async def aclose(self) -> None:
     assert closed == ["first", "second"]
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"])
+@pytest.mark.parametrize("primary", [None, ValueError("primary")], ids=["no-primary", "ordinary-primary"])
+async def test_cleanup_keeps_later_cancellation_over_ordinary_error(sync: bool, primary: BaseException | None) -> None:
+    from openai._streaming_cleanup import close_owned, aclose_owned
+
+    closed: list[str] = []
+
+    class Closer:
+        def __init__(self, name: str, error: BaseException | None = None) -> None:
+            self.name = name
+            self.error = error
+
+        def close(self) -> None:
+            closed.append(self.name)
+            if self.error is not None:
+                raise self.error
+
+        async def aclose(self) -> None:
+            self.close()
+
+    resources = (
+        Closer("ordinary", RuntimeError("close failed")),
+        Closer("cancelled", asyncio.CancelledError()),
+        Closer("last"),
+    )
+    with pytest.raises(asyncio.CancelledError):
+        if sync:
+            close_owned(*resources, primary=primary)
+        else:
+            await aclose_owned(*resources, primary=primary)
+
+    assert closed == ["ordinary", "cancelled", "last"]
+
+
 async def to_aiter(iter: Iterator[bytes]) -> AsyncIterator[bytes]:
     for chunk in iter:
         yield chunk

From 348f1c281f2c4d6beaca80b2ea27fcea5cd5f57a Mon Sep 17 00:00:00 2001
From: Xinmin Zeng <135568692+fancyboi999@users.noreply.github.com>
Date: Fri, 23 Oct 2026 22:58:13 -0700
Subject: [PATCH 5/7] fix(agents): report terminal cleanup failures after the
 turn outcome

AgentSessionStream closes its connection when it sees the terminal event,
before yielding it. A cleanup failure there was raised before the caller
evaluated the outcome, turning a failed turn or an output parse error
into observation_failed. The failure is now kept and raised only after
the outcome is established (or on the next read when iterating by hand);
a result error still wins.

close() also closes the stream's own iterator when it is not running.
---
 src/openai/lib/streaming/agents/_streams.py | 52 +++++++++++++++++----
 tests/lib/streaming/agents/test_results.py  | 44 ++++++++++++++++-
 2 files changed, 86 insertions(+), 10 deletions(-)

diff --git a/src/openai/lib/streaming/agents/_streams.py b/src/openai/lib/streaming/agents/_streams.py
index bc46c307b6..0e40aaab55 100644
--- a/src/openai/lib/streaming/agents/_streams.py
+++ b/src/openai/lib/streaming/agents/_streams.py
@@ -12,7 +12,7 @@
 from ...._types import Omit, Headers, NotGiven, omit, not_given
 from ._dispatch import ToolDispatcher, AsyncToolDispatcher, _request_options
 from ...._streaming import Stream, AsyncStream
-from ...._streaming_cleanup import close_owned, run_cleanup
+from ...._streaming_cleanup import close_owned, run_cleanup, aclose_owned
 from ...beta.agents._result import (
     OutputT,
     AgentTurnResult,
@@ -132,6 +132,7 @@ def __init__(
         self._iterator: Iterator[AgentSessionEvent] | None = None
         self._entered = False
         self._closed = False
+        self._deferred_cleanup_error: BaseException | None = None
 
     def __enter__(self) -> Self:
         if self._entered or self._closed:
@@ -196,12 +197,15 @@ def get_final_result(self) -> AgentTurnResult[OutputT]:
                         collector.check_outcome(self._handlers)
                         if collector.is_done():
                             break
-                return self._collection.result()
+                result = self._collection.result()
             except (AgentTurnResultError, AgentOutputParseError):
                 raise
             except Exception as error:
                 self._collection.record_error(error)
                 raise collector.error("observation_failed") from error
+            # The outcome is established; a terminal cleanup failure may surface now.
+            self._raise_deferred_cleanup_error()
+            return result
         except BaseException as error:
             failure = error
             raise
@@ -212,8 +216,15 @@ def get_final_result(self) -> AgentTurnResult[OutputT]:
     def close(self) -> None:
         """Close the event connection without cancelling the backend turn."""
         self._closed = True
-        if self._stream is not None:
-            self._stream.close()
+        # Concurrent close does not cancel or join an active consumer; that consumer
+        # closes the iterator when it unwinds.
+        running = getattr(self._iterator, "gi_running", False)
+        close_owned(None if running else self._iterator, self._stream)
+
+    def _raise_deferred_cleanup_error(self) -> None:
+        error, self._deferred_cleanup_error = self._deferred_cleanup_error, None
+        if error is not None:
+            raise error
 
     def _iterate(self) -> Iterator[AgentSessionEvent]:
         assert self._stream is not None
@@ -225,10 +236,16 @@ def _iterate(self) -> Iterator[AgentSessionEvent]:
                 self._collection.accept(event)
                 terminal = self._state.terminal(event)
                 if terminal:
-                    self.close()
+                    # Close now, but report a cleanup failure only after this terminal
+                    # event's outcome has been evaluated.
+                    try:
+                        self.close()
+                    except Exception as error:
+                        self._deferred_cleanup_error = error
                 pending = self._dispatcher.prepare(event)
                 yield event
                 if terminal or self._closed:
+                    self._raise_deferred_cleanup_error()
                     return
                 if pending is not None:
                     self._dispatcher.dispatch(pending)
@@ -283,6 +300,7 @@ def __init__(
         self._iterator: AsyncIterator[AgentSessionEvent] | None = None
         self._entered = False
         self._closed = False
+        self._deferred_cleanup_error: BaseException | None = None
 
     async def __aenter__(self) -> Self:
         if self._entered or self._closed:
@@ -347,12 +365,15 @@ async def get_final_result(self) -> AgentTurnResult[OutputT]:
                         collector.check_outcome(self._handlers)
                         if collector.is_done():
                             break
-                return self._collection.result()
+                result = self._collection.result()
             except (AgentTurnResultError, AgentOutputParseError):
                 raise
             except Exception as error:
                 self._collection.record_error(error)
                 raise collector.error("observation_failed") from error
+            # The outcome is established; a terminal cleanup failure may surface now.
+            self._raise_deferred_cleanup_error()
+            return result
         except BaseException as error:
             failure = error
             raise
@@ -363,8 +384,15 @@ async def get_final_result(self) -> AgentTurnResult[OutputT]:
     async def close(self) -> None:
         """Close the event connection without cancelling the backend turn."""
         self._closed = True
-        if self._stream is not None:
-            await self._stream.close()
+        # Concurrent close does not cancel or join an active consumer; that consumer
+        # closes the iterator when it unwinds.
+        running = getattr(self._iterator, "ag_running", False)
+        await aclose_owned(None if running else self._iterator, self._stream)
+
+    def _raise_deferred_cleanup_error(self) -> None:
+        error, self._deferred_cleanup_error = self._deferred_cleanup_error, None
+        if error is not None:
+            raise error
 
     async def _iterate(self) -> AsyncIterator[AgentSessionEvent]:
         assert self._stream is not None
@@ -376,10 +404,16 @@ async def _iterate(self) -> AsyncIterator[AgentSessionEvent]:
                 self._collection.accept(event)
                 terminal = self._state.terminal(event)
                 if terminal:
-                    await self.close()
+                    # Close now, but report a cleanup failure only after this terminal
+                    # event's outcome has been evaluated.
+                    try:
+                        await self.close()
+                    except Exception as error:
+                        self._deferred_cleanup_error = error
                 pending = self._dispatcher.prepare(event)
                 yield event
                 if terminal or self._closed:
+                    self._raise_deferred_cleanup_error()
                     return
                 if pending is not None:
                     await self._dispatcher.dispatch(pending)
diff --git a/tests/lib/streaming/agents/test_results.py b/tests/lib/streaming/agents/test_results.py
index 9d63355f35..dac7efb516 100644
--- a/tests/lib/streaming/agents/test_results.py
+++ b/tests/lib/streaming/agents/test_results.py
@@ -6,10 +6,11 @@
 
 import httpx2
 import pytest
+from pydantic import BaseModel
 
 from openai import OpenAI, AsyncOpenAI
 from openai._streaming import SSEDecoder
-from openai.lib.beta.agents import AgentTurnResult, AgentTurnResultError
+from openai.lib.beta.agents import AgentTurnResult, AgentTurnResultError, AgentOutputParseError
 from openai.lib.streaming.agents import AgentSessionStream, AsyncAgentSessionStream
 from tests.lib.streaming.agents.test_streams import Server, EventBody, sdk as sdk, call, idle, session, turn_event
 
@@ -240,6 +241,47 @@ async def test_cleanup_error_does_not_replace_result_error(
     assert exc.value.reason == "failed"
 
 
+async def _final_result_through_stream(sdk: OpenAI | AsyncOpenAI, **kwargs: Any) -> Any:
+    if isinstance(sdk, AsyncOpenAI):
+        async with sdk.beta.agents.sessions.stream("session_test", input="Question", **kwargs) as async_stream:
+            return await async_stream.get_final_result()
+    with sdk.beta.agents.sessions.stream("session_test", input="Question", **kwargs) as stream:
+        return stream.get_final_result()
+
+
+async def test_terminal_cleanup_error_does_not_replace_failed_session(
+    sdk: OpenAI | AsyncOpenAI, server: ResultServer, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(sdk, "_make_sse_decoder", lambda: _CloseFailingDecoder())
+    server.body = EventBody(
+        [turn_event("created"), message(), {"type": "agent.session.failed", "session": session("failed")}]
+    )
+    with pytest.raises(AgentTurnResultError) as exc:
+        await _final_result_through_stream(sdk)
+    assert exc.value.reason == "failed"
+
+
+async def test_terminal_cleanup_error_does_not_replace_parse_error(
+    sdk: OpenAI | AsyncOpenAI, server: ResultServer, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    class Summary(BaseModel):
+        summary: str
+
+    monkeypatch.setattr(sdk, "_make_sse_decoder", lambda: _CloseFailingDecoder())
+    server.body = EventBody([turn_event("created"), message("not json"), turn_event("completed"), idle()])
+    with pytest.raises(AgentOutputParseError):
+        await _final_result_through_stream(sdk, output_type=Summary)
+
+
+async def test_terminal_cleanup_error_surfaces_after_successful_result(
+    sdk: OpenAI | AsyncOpenAI, server: ResultServer, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(sdk, "_make_sse_decoder", lambda: _CloseFailingDecoder())
+    server.body = EventBody([turn_event("created"), message(), turn_event("completed"), idle()])
+    with pytest.raises(RuntimeError, match="decoder close failed"):
+        await _final_result_through_stream(sdk)
+
+
 @pytest.mark.parametrize("tail", [[], [turn_event("completed")]])
 async def test_incomplete_answer(
     sdk: OpenAI | AsyncOpenAI, server: ResultServer, creation: bool, tail: list[dict[str, Any]]

From ceb58fc7471478221fde57bfa6cc50c4fd279b16 Mon Sep 17 00:00:00 2001
From: Xinmin Zeng <135568692+fancyboi999@users.noreply.github.com>
Date: Fri, 23 Oct 2026 23:13:33 -0700
Subject: [PATCH 6/7] fix(agents): deliver a pending terminal cleanup failure
 exactly once

The terminal close marks the stream closed, so the next read stopped
before reaching the deferred cleanup failure and ordinary iteration,
until_done() and close-after-terminal all hid it. Reads on a closed
stream and close() now raise the pending failure once. A result error
from get_final_result() or an exception leaving the with block still
takes precedence and discards it.
---
 src/openai/lib/streaming/agents/_streams.py | 12 ++++
 tests/lib/streaming/agents/test_results.py  | 62 +++++++++++++++++++++
 2 files changed, 74 insertions(+)

diff --git a/src/openai/lib/streaming/agents/_streams.py b/src/openai/lib/streaming/agents/_streams.py
index 0e40aaab55..81530ffb13 100644
--- a/src/openai/lib/streaming/agents/_streams.py
+++ b/src/openai/lib/streaming/agents/_streams.py
@@ -157,6 +157,8 @@ def __enter__(self) -> Self:
     def __exit__(
         self, exc_type: type[BaseException] | None, exc: BaseException | None, exc_tb: TracebackType | None
     ) -> None:
+        if exc is not None:
+            self._deferred_cleanup_error = None  # The propagating exception wins.
         self.close()
 
     def __iter__(self) -> Self:
@@ -164,6 +166,7 @@ def __iter__(self) -> Self:
 
     def __next__(self) -> AgentSessionEvent:
         if self._closed:
+            self._raise_deferred_cleanup_error()
             raise StopIteration
         if self._iterator is None:
             raise RuntimeError("Use sessions.stream in a with block")
@@ -211,6 +214,8 @@ def get_final_result(self) -> AgentTurnResult[OutputT]:
             raise
         finally:
             # A cleanup error must not replace the result error being raised.
+            if failure is not None:
+                self._deferred_cleanup_error = None
             close_owned(self, primary=failure)
 
     def close(self) -> None:
@@ -220,6 +225,7 @@ def close(self) -> None:
         # closes the iterator when it unwinds.
         running = getattr(self._iterator, "gi_running", False)
         close_owned(None if running else self._iterator, self._stream)
+        self._raise_deferred_cleanup_error()
 
     def _raise_deferred_cleanup_error(self) -> None:
         error, self._deferred_cleanup_error = self._deferred_cleanup_error, None
@@ -325,6 +331,8 @@ async def __aenter__(self) -> Self:
     async def __aexit__(
         self, exc_type: type[BaseException] | None, exc: BaseException | None, exc_tb: TracebackType | None
     ) -> None:
+        if exc is not None:
+            self._deferred_cleanup_error = None  # The propagating exception wins.
         await self.close()
 
     def __aiter__(self) -> Self:
@@ -332,6 +340,7 @@ def __aiter__(self) -> Self:
 
     async def __anext__(self) -> AgentSessionEvent:
         if self._closed:
+            self._raise_deferred_cleanup_error()
             raise StopAsyncIteration
         if self._iterator is None:
             raise RuntimeError("Use sessions.stream in an async with block")
@@ -379,6 +388,8 @@ async def get_final_result(self) -> AgentTurnResult[OutputT]:
             raise
         finally:
             # A cleanup error must not replace the result error being raised.
+            if failure is not None:
+                self._deferred_cleanup_error = None
             await run_cleanup(self.close, primary=failure)
 
     async def close(self) -> None:
@@ -388,6 +399,7 @@ async def close(self) -> None:
         # closes the iterator when it unwinds.
         running = getattr(self._iterator, "ag_running", False)
         await aclose_owned(None if running else self._iterator, self._stream)
+        self._raise_deferred_cleanup_error()
 
     def _raise_deferred_cleanup_error(self) -> None:
         error, self._deferred_cleanup_error = self._deferred_cleanup_error, None
diff --git a/tests/lib/streaming/agents/test_results.py b/tests/lib/streaming/agents/test_results.py
index dac7efb516..4945b03f98 100644
--- a/tests/lib/streaming/agents/test_results.py
+++ b/tests/lib/streaming/agents/test_results.py
@@ -282,6 +282,68 @@ async def test_terminal_cleanup_error_surfaces_after_successful_result(
         await _final_result_through_stream(sdk)
 
 
+@pytest.mark.parametrize("consume", ["iterate", "until_done", "close_after_terminal"])
+async def test_terminal_cleanup_error_surfaces_once_without_result(
+    sdk: OpenAI | AsyncOpenAI, server: ResultServer, monkeypatch: pytest.MonkeyPatch, consume: str
+) -> None:
+    monkeypatch.setattr(sdk, "_make_sse_decoder", lambda: _CloseFailingDecoder())
+    server.body = EventBody([turn_event("created"), message(), turn_event("completed"), idle()])
+    raised = 0
+    if isinstance(sdk, AsyncOpenAI):
+        async_stream = sdk.beta.agents.sessions.stream("session_test", input="Question")
+        try:
+            async with async_stream:
+                if consume == "iterate":
+                    async for _ in async_stream:
+                        pass
+                elif consume == "until_done":
+                    await async_stream.until_done()
+                else:
+                    async for event in async_stream:
+                        if event.type == "agent.session.idle":
+                            break
+        except RuntimeError as error:
+            assert str(error) == "decoder close failed"
+            raised += 1
+        await async_stream.close()  # Delivered once; a later close does not raise again.
+    else:
+        stream = sdk.beta.agents.sessions.stream("session_test", input="Question")
+        try:
+            with stream:
+                if consume == "iterate":
+                    for _ in stream:
+                        pass
+                elif consume == "until_done":
+                    stream.until_done()
+                else:
+                    for event in stream:
+                        if event.type == "agent.session.idle":
+                            break
+        except RuntimeError as error:
+            assert str(error) == "decoder close failed"
+            raised += 1
+        stream.close()
+    assert raised == 1
+
+
+async def test_context_exception_wins_over_terminal_cleanup_error(
+    sdk: OpenAI | AsyncOpenAI, server: ResultServer, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(sdk, "_make_sse_decoder", lambda: _CloseFailingDecoder())
+    server.body = EventBody([turn_event("created"), message(), turn_event("completed"), idle()])
+    with pytest.raises(KeyError, match="caller"):
+        if isinstance(sdk, AsyncOpenAI):
+            async with sdk.beta.agents.sessions.stream("session_test", input="Question") as async_stream:
+                async for event in async_stream:
+                    if event.type == "agent.session.idle":
+                        raise KeyError("caller")
+        else:
+            with sdk.beta.agents.sessions.stream("session_test", input="Question") as stream:
+                for event in stream:
+                    if event.type == "agent.session.idle":
+                        raise KeyError("caller")
+
+
 @pytest.mark.parametrize("tail", [[], [turn_event("completed")]])
 async def test_incomplete_answer(
     sdk: OpenAI | AsyncOpenAI, server: ResultServer, creation: bool, tail: list[dict[str, Any]]

From 1aa7953e06b27d0d6b21aef80be7e6083ebe1f57 Mon Sep 17 00:00:00 2001
From: Xinmin Zeng <135568692+fancyboi999@users.noreply.github.com>
Date: Fri, 23 Oct 2026 23:46:24 -0700
Subject: [PATCH 7/7] fix(streaming): keep the exception leaving a with block
 over cleanup errors

Context-manager exits closed streams without passing the exception that
was leaving the block, so a cleanup failure could replace the caller's
exception or a task cancellation. Stream, AsyncStream, the Chat and
Responses streams and their managers, and the agent session streams now
pass it as the primary failure.
---
 src/openai/_streaming.py                      |  8 ++--
 src/openai/lib/streaming/agents/_streams.py   |  4 +-
 src/openai/lib/streaming/chat/_completions.py | 14 ++++--
 .../lib/streaming/responses/_responses.py     | 14 ++++--
 tests/lib/streaming/agents/test_results.py    | 30 ++++++++++++
 tests/test_streaming.py                       | 48 +++++++++++++++++++
 6 files changed, 103 insertions(+), 15 deletions(-)

diff --git a/src/openai/_streaming.py b/src/openai/_streaming.py
index 9265922c05..5660f39a06 100644
--- a/src/openai/_streaming.py
+++ b/src/openai/_streaming.py
@@ -13,7 +13,7 @@
 from ._utils import is_mapping, extract_type_var_from_base
 from ._httpx2 import request_exceptions, timeout_exceptions
 from ._exceptions import APIError, APITimeoutError, APIConnectionError
-from ._streaming_cleanup import close_owned, aclose_owned, drain_response_async
+from ._streaming_cleanup import close_owned, run_cleanup, aclose_owned, drain_response_async
 
 if TYPE_CHECKING:
     from ._client import OpenAI, AsyncOpenAI
@@ -149,7 +149,8 @@ def __exit__(
         exc: BaseException | None,
         exc_tb: TracebackType | None,
     ) -> None:
-        self.close()
+        # A cleanup error must not replace the exception leaving the block.
+        close_owned(self, primary=exc)
 
     def close(self) -> None:
         """
@@ -294,7 +295,8 @@ async def __aexit__(
         exc: BaseException | None,
         exc_tb: TracebackType | None,
     ) -> None:
-        await self.close()
+        # A cleanup error must not replace the exception leaving the block.
+        await run_cleanup(self.close, primary=exc)
 
     async def close(self) -> None:
         """
diff --git a/src/openai/lib/streaming/agents/_streams.py b/src/openai/lib/streaming/agents/_streams.py
index 81530ffb13..d7fb89b99c 100644
--- a/src/openai/lib/streaming/agents/_streams.py
+++ b/src/openai/lib/streaming/agents/_streams.py
@@ -159,7 +159,7 @@ def __exit__(
     ) -> None:
         if exc is not None:
             self._deferred_cleanup_error = None  # The propagating exception wins.
-        self.close()
+        close_owned(self, primary=exc)
 
     def __iter__(self) -> Self:
         return self
@@ -333,7 +333,7 @@ async def __aexit__(
     ) -> None:
         if exc is not None:
             self._deferred_cleanup_error = None  # The propagating exception wins.
-        await self.close()
+        await run_cleanup(self.close, primary=exc)
 
     def __aiter__(self) -> Self:
         return self
diff --git a/src/openai/lib/streaming/chat/_completions.py b/src/openai/lib/streaming/chat/_completions.py
index b434240af0..374616f8bb 100644
--- a/src/openai/lib/streaming/chat/_completions.py
+++ b/src/openai/lib/streaming/chat/_completions.py
@@ -38,7 +38,7 @@
 from ...._streaming import Stream, AsyncStream
 from ....types.chat import ChatCompletionChunk, ParsedChatCompletion, ChatCompletionToolUnionParam
 from ...._exceptions import LengthFinishReasonError, ContentFilterFinishReasonError
-from ...._streaming_cleanup import close_owned, aclose_owned
+from ...._streaming_cleanup import close_owned, run_cleanup, aclose_owned
 from ....types.chat.chat_completion import Moderation, ChoiceLogprobs
 from ....types.chat.chat_completion_chunk import Choice as ChoiceChunk
 from ....types.chat.completion_create_params import ResponseFormat as ResponseFormatParam
@@ -80,7 +80,8 @@ def __exit__(
         exc: BaseException | None,
         exc_tb: TracebackType | None,
     ) -> None:
-        self.close()
+        # A cleanup error must not replace the exception leaving the block.
+        close_owned(self, primary=exc)
 
     def close(self) -> None:
         """
@@ -166,7 +167,8 @@ def __exit__(
         exc_tb: TracebackType | None,
     ) -> None:
         if self.__stream is not None:
-            self.__stream.close()
+            # A cleanup error must not replace the exception leaving the block.
+            close_owned(self.__stream, primary=exc)
 
 
 class AsyncChatCompletionStream(Generic[ResponseFormatT]):
@@ -205,7 +207,8 @@ async def __aexit__(
         exc: BaseException | None,
         exc_tb: TracebackType | None,
     ) -> None:
-        await self.close()
+        # A cleanup error must not replace the exception leaving the block.
+        await run_cleanup(self.close, primary=exc)
 
     async def close(self) -> None:
         """
@@ -291,7 +294,8 @@ async def __aexit__(
         exc_tb: TracebackType | None,
     ) -> None:
         if self.__stream is not None:
-            await self.__stream.close()
+            # A cleanup error must not replace the exception leaving the block.
+            await run_cleanup(self.__stream.close, primary=exc)
 
 
 class ChatCompletionStreamState(Generic[ResponseFormatT]):
diff --git a/src/openai/lib/streaming/responses/_responses.py b/src/openai/lib/streaming/responses/_responses.py
index dd079532ef..b05c3cc7f8 100644
--- a/src/openai/lib/streaming/responses/_responses.py
+++ b/src/openai/lib/streaming/responses/_responses.py
@@ -24,7 +24,7 @@
     ResponseOutputItemDoneEvent,
     ResponseOutputItemAddedEvent,
 )
-from ...._streaming_cleanup import close_owned, aclose_owned
+from ...._streaming_cleanup import close_owned, run_cleanup, aclose_owned
 from ..._parsing._responses import TextFormatT, parse_text, parse_response
 from ....types.responses.tool_param import ToolParam
 from ....types.responses.parsed_response import (
@@ -73,7 +73,8 @@ def __exit__(
         exc: BaseException | None,
         exc_tb: TracebackType | None,
     ) -> None:
-        self.close()
+        # A cleanup error must not replace the exception leaving the block.
+        close_owned(self, primary=exc)
 
     def close(self) -> None:
         """
@@ -137,7 +138,8 @@ def __exit__(
         exc_tb: TracebackType | None,
     ) -> None:
         if self.__stream is not None:
-            self.__stream.close()
+            # A cleanup error must not replace the exception leaving the block.
+            close_owned(self.__stream, primary=exc)
 
 
 class AsyncResponseStream(Generic[TextFormatT]):
@@ -177,7 +179,8 @@ async def __aexit__(
         exc: BaseException | None,
         exc_tb: TracebackType | None,
     ) -> None:
-        await self.close()
+        # A cleanup error must not replace the exception leaving the block.
+        await run_cleanup(self.close, primary=exc)
 
     async def close(self) -> None:
         """
@@ -241,7 +244,8 @@ async def __aexit__(
         exc_tb: TracebackType | None,
     ) -> None:
         if self.__stream is not None:
-            await self.__stream.close()
+            # A cleanup error must not replace the exception leaving the block.
+            await run_cleanup(self.__stream.close, primary=exc)
 
 
 def _output_identity(
diff --git a/tests/lib/streaming/agents/test_results.py b/tests/lib/streaming/agents/test_results.py
index 4945b03f98..21bb779871 100644
--- a/tests/lib/streaming/agents/test_results.py
+++ b/tests/lib/streaming/agents/test_results.py
@@ -344,6 +344,36 @@ async def test_context_exception_wins_over_terminal_cleanup_error(
                         raise KeyError("caller")
 
 
+async def test_pre_terminal_context_exception_wins_over_cleanup_error(
+    sdk: OpenAI | AsyncOpenAI, server: ResultServer, creation: bool, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(sdk, "_make_sse_decoder", lambda: _CloseFailingDecoder())
+    server.body = EventBody([turn_event("created"), message(), turn_event("completed"), idle()])
+    with pytest.raises(KeyError, match="caller"):
+        if isinstance(sdk, AsyncOpenAI):
+            async_stream = (
+                await sdk.beta.agents.sessions.create(
+                    agent={"model": "test-model"}, environment={"type": "none"}, input="Question", stream=True
+                )
+                if creation
+                else sdk.beta.agents.sessions.stream("session_test", input="Question")
+            )
+            async with async_stream:
+                async for _ in async_stream:
+                    raise KeyError("caller")
+        else:
+            stream = (
+                sdk.beta.agents.sessions.create(
+                    agent={"model": "test-model"}, environment={"type": "none"}, input="Question", stream=True
+                )
+                if creation
+                else sdk.beta.agents.sessions.stream("session_test", input="Question")
+            )
+            with stream:
+                for _ in stream:
+                    raise KeyError("caller")
+
+
 @pytest.mark.parametrize("tail", [[], [turn_event("completed")]])
 async def test_incomplete_answer(
     sdk: OpenAI | AsyncOpenAI, server: ResultServer, creation: bool, tail: list[dict[str, Any]]
diff --git a/tests/test_streaming.py b/tests/test_streaming.py
index 6b9d8bdbab..06ae20d281 100644
--- a/tests/test_streaming.py
+++ b/tests/test_streaming.py
@@ -843,6 +843,54 @@ async def aclose(self) -> None:
     assert closed == ["ordinary", "cancelled", "last"]
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"])
+async def test_context_exit_keeps_caller_exception_over_cleanup_error(
+    sync: bool, client: OpenAI, async_client: AsyncOpenAI
+) -> None:
+    events = _TrackedEvents(close_error=True)
+    with pytest.raises(KeyError, match="caller"):
+        if sync:
+            stream = Stream(cast_to=object, client=client, response=httpx2.Response(200, content=sse_body()))
+            stream._decoder = _TrackedDecoder(events)
+            with stream:
+                next(iter(stream))
+                raise KeyError("caller")
+        else:
+            async_stream = AsyncStream(
+                cast_to=object, client=async_client, response=httpx2.Response(200, content=to_aiter(sse_body()))
+            )
+            async_stream._decoder = _TrackedDecoder(events)
+            async with async_stream:
+                await async_stream.__anext__()
+                raise KeyError("caller")
+    assert events.closes == 1
+
+
+@pytest.mark.asyncio
+async def test_context_exit_keeps_cancellation_over_cleanup_error(async_client: AsyncOpenAI) -> None:
+    events = _TrackedEvents(close_error=True)
+    started = asyncio.Event()
+
+    async def consume() -> None:
+        async_stream = AsyncStream(
+            cast_to=object, client=async_client, response=httpx2.Response(200, content=to_aiter(sse_body()))
+        )
+        async_stream._decoder = _TrackedDecoder(events)
+        async with async_stream:
+            await async_stream.__anext__()
+            started.set()
+            await asyncio.Event().wait()
+
+    task = asyncio.create_task(consume())
+    await started.wait()
+    task.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await task
+    assert task.cancelled()
+    assert events.closes == 1
+
+
 async def to_aiter(iter: Iterator[bytes]) -> AsyncIterator[bytes]:
     for chunk in iter:
         yield chunk