@@ -13,6 +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
 
 if TYPE_CHECKING:
     from ._client import OpenAI, AsyncOpenAI
@@ -183,6 +184,11 @@ async def __stream__(self) -> AsyncIterator[_T]:
         try:
             async for sse in iterator:
                 if sse.data.startswith("[DONE]"):
+                    # HTTP/1.1 needs the remaining body consumed for connection reuse.
+                    # HTTP/2 can close this stream and keep the connection, so draining
+                    # it would add unnecessary latency after [DONE].
+                    if response.http_version == "HTTP/1.1":
+                        await drain_response_async(response)
                     break
 
                 # we have to special case the Assistants `thread.` events since we won't have an "event" key in the data
 
@@ -0,0 +1,29 @@
+from __future__ import annotations
+
+from typing import AsyncIterator, cast
+
+import anyio
+import httpx2
+from anyio.lowlevel import checkpoint
+
+from ._httpx2 import request_exceptions, stream_consumed_exceptions
+
+_DRAIN_TIMEOUT = 0.1
+
+
+async def drain_response_async(response: httpx2.Response) -> None:
+    if response.is_closed:
+        return
+    try:
+        with anyio.move_on_after(_DRAIN_TIMEOUT):
+            # Discard raw transport bytes without decoding trailing SSE or content.
+            async for _ in cast(AsyncIterator[bytes], response.stream):
+                # Also enforce the deadline on transports yielding buffered data.
+                await checkpoint()
+    except request_exceptions() + stream_consumed_exceptions():
+        pass  # Cleanup must not turn completed output into a request error.
+    finally:
+        # Close outside our timeout scope and shield cleanup from caller
+        # cancellation. Cancellation still propagates after the response closes.
+        with anyio.CancelScope(shield=True):
+            await response.aclose()
@@ -2,15 +2,21 @@
 
 import os
 import ssl
+import time
+import socket
+import asyncio
 import importlib
+import threading
 from typing import Any, Iterator, AsyncIterator
 from contextlib import aclosing, nullcontext
+from http.server import ThreadingHTTPServer, BaseHTTPRequestHandler
+from typing_extensions import override
 
 import anyio
 import httpx2
 import pytest
 
-from openai import OpenAI, AsyncOpenAI, APITimeoutError, APIConnectionError
+from openai import OpenAI, AsyncOpenAI, APITimeoutError, APIConnectionError, DefaultAioHttpClient
 from openai._streaming import Stream, AsyncStream, ServerSentEvent
 
 
@@ -29,6 +35,167 @@ def http_module(request: pytest.FixtureRequest) -> Any:
     return importlib.import_module(request.param)
 
 
+@pytest.mark.parametrize("mode", ["async", "aiohttp"])
+@pytest.mark.parametrize("ending", ["complete", "truncate", "stall", "trickle"])
+async def test_done_http11_connection_reuse(mode: str, ending: str, http_module: Any) -> None:
+    if mode == "aiohttp" and http_module.__name__ == "httpx":
+        pytest.skip("aiohttp uses HTTPX2")
+    connections = 0
+    requests = 0
+    release = threading.Event()
+
+    class Handler(BaseHTTPRequestHandler):
+        protocol_version = "HTTP/1.1"
+
+        @override
+        def setup(self) -> None:
+            nonlocal connections
+            super().setup()
+            connections += 1
+            self.connection.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
+
+        def do_POST(self) -> None:
+            nonlocal requests
+            self.rfile.read(int(self.headers["Content-Length"]))
+            requests += 1
+            self.send_response(200)
+            self.send_header("Content-Type", "text/event-stream")
+            self.send_header("Transfer-Encoding", "chunked")
+            self.end_headers()
+            payload = (
+                b'data: {"id":"synthetic","object":"chat.completion.chunk","created":0,"model":"synthetic",'
+                b'"choices":[{"index":0,"delta":{"content":"hello"}}]}\n\ndata: [DONE]\n\n'
+            )
+            try:
+                self.wfile.write(f"{len(payload):x}\r\n".encode() + payload + b"\r\n")
+                if requests == 1:
+                    if ending == "truncate":
+                        self.close_connection = True
+                        return
+                    if ending == "stall":
+                        release.wait(timeout=0.6)
+                    if ending == "trickle":
+                        # Trailer bytes do not yield body chunks, but must still
+                        # obey the total cleanup deadline, not a per-read timeout.
+                        self.wfile.write(b"0\r\nX-Synthetic: ")
+                        for _ in range(15):
+                            if release.wait(timeout=0.04):
+                                return
+                            self.wfile.write(b"a")
+                        self.wfile.write(b"\r\n\r\n")
+                        return
+                self.wfile.write(b"0\r\n\r\n")
+            except ConnectionError:
+                self.close_connection = True
+
+        @override
+        def log_message(self, format: str, *args: object) -> None:  # noqa: A002
+            pass
+
+    server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+    thread = threading.Thread(target=server.serve_forever, kwargs={"poll_interval": 0.01})
+    thread.start()
+    options: dict[str, Any] = {
+        "api_key": "synthetic",
+        "base_url": f"http://127.0.0.1:{server.server_port}/v1",
+        "max_retries": 0,
+        "timeout": http_module.Timeout(5, read=None),
+    }
+    transport_options: dict[str, Any] = {"trust_env": False, "limits": http_module.Limits(max_connections=1)}
+    try:
+        http_client = (
+            DefaultAioHttpClient(**transport_options)
+            if mode == "aiohttp"
+            else http_module.AsyncClient(**transport_options)
+        )
+        async with AsyncOpenAI(**options, http_client=http_client) as async_client:
+            for _ in range(2):
+                async with await async_client.chat.completions.create(
+                    model="synthetic", messages=[], stream=True
+                ) as async_stream:
+                    assert (await async_stream.__anext__()).choices[0].delta.content == "hello"
+                    start = time.monotonic()
+                    assert [chunk async for chunk in async_stream] == []
+                    assert time.monotonic() - start < 0.4
+    finally:
+        release.set()
+        server.shutdown()
+        thread.join(timeout=5)
+        server.server_close()
+
+    assert not thread.is_alive()
+    assert requests == 2
+    # A failed drain must discard the connection and release the only pool slot.
+    assert connections == (1 if ending == "complete" else 2)
+
+
+@pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"])
+@pytest.mark.parametrize("http_version", [b"HTTP/1.1", b"HTTP/2"])
+async def test_done_discards_bytes_only_for_async_http11(
+    sync: bool, http_version: bytes, http_module: Any, client: OpenAI, async_client: AsyncOpenAI
+) -> None:
+    read_tail = False
+
+    def body() -> Iterator[bytes]:
+        nonlocal read_tail
+        yield b"data: [DONE]\n\n"
+        read_tail = True
+        yield b"\xff\n\n"  # Trailing data must not be decoded as SSE.
+
+    iterator = body()
+    async_iterator = to_aiter(iterator)
+
+    class Body(http_module.SyncByteStream, http_module.AsyncByteStream):
+        def __iter__(self) -> Iterator[bytes]:
+            return iterator
+
+        def __aiter__(self) -> AsyncIterator[bytes]:
+            return async_iterator
+
+    response = http_module.Response(200, stream=Body(), extensions={"http_version": http_version})
+    if sync:
+        assert list(Stream(cast_to=object, response=response, client=client)) == []
+    else:
+        assert [event async for event in AsyncStream(cast_to=object, response=response, client=async_client)] == []
+    assert response.is_closed
+    assert read_tail == (not sync and http_version == b"HTTP/1.1")
+
+
+@pytest.mark.parametrize("cancel", [True, False], ids=["caller-cancel", "deadline"])
+async def test_cancellation_during_done_drain_closes_response(
+    cancel: bool, http_module: Any, async_client: AsyncOpenAI
+) -> None:
+    draining = asyncio.Event()
+
+    class Body(http_module.AsyncByteStream):
+        closed = False
+
+        async def __aiter__(self) -> AsyncIterator[bytes]:
+            yield b"data: [DONE]\n\n"
+            draining.set()
+            await asyncio.Event().wait()
+
+        async def aclose(self) -> None:
+            await anyio.sleep(0)
+            self.closed = True
+
+    body = Body()
+    response = http_module.Response(200, stream=body)
+    stream = AsyncStream(cast_to=object, response=response, client=async_client)
+    task = asyncio.create_task(stream.__anext__())
+    try:
+        await asyncio.wait_for(draining.wait(), timeout=5)
+        if cancel:
+            task.cancel()
+        with pytest.raises(asyncio.CancelledError if cancel else StopAsyncIteration):
+            await task
+        assert response.is_closed
+        assert body.closed
+    finally:
+        task.cancel()
+        await asyncio.gather(task, return_exceptions=True)
+
+
 @pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"])
 @pytest.mark.parametrize("delivered", [False, True], ids=["before-first-event", "after-first-event"])
 @pytest.mark.parametrize(