open-telemetry · sfc-gh-zeningchen · Sep 4, 2026 · Sep 4, 2026 · Sep 4, 2026 · Sep 4, 2026
@@ -0,0 +1 @@
+Record response telemetry for ``with_streaming_response`` calls on the async client, whose ``parse()`` returns a coroutine, and end the span for a parsed stream the caller abandons.
@@ -6,6 +6,7 @@
 from __future__ import annotations
 
 import functools
+import inspect
 import logging
 from collections.abc import Awaitable, Callable
 from typing import (
@@ -70,16 +71,16 @@ class RawResponseStreamProxy(ObjectProxy):
 
     ``parse()`` wraps the result only when it is an SDK ``Stream`` /
     ``AsyncStream`` we know how to drive. Anything else is handed back untouched
-    (for example the coroutine ``parse()`` the SDK documents it "will become in
-    the next major version" for the async client, or a custom non-stream parse
-    target): we can't instrument it, so we just log and step aside.
-
-    The span is finalized independently of ``parse()``. Whether the caller
-    parses and drains the wrapper, drains the body directly via the raw
-    response's ``read()``/``iter_bytes()``, or never parses at all, every path
-    ends by closing the underlying httpx response — so a fallback on its
-    ``close``/``aclose`` finalizes the span when ``parse()`` never built a
-    wrapper that would finalize it instead.
+    (for example a custom non-stream parse target): we can't instrument it, so
+    we just log and step aside. On the async client ``parse()`` is a coroutine,
+    so we await it and wrap what it resolves to.
+
+    Every path — draining the wrapper, reading the body directly, never parsing
+    at all — ends by closing the underlying httpx response, so hooks on its
+    ``close``/``aclose`` are what guarantee the span ends exactly once. A caller
+    that parses and then walks away (an early ``break``, an exception inside the
+    ``with`` block) leaves a wrapper nobody drove, so the hook closes it and the
+    wrapper finalizes with what it saw.
     """
 
     def __init__(
@@ -90,7 +91,9 @@ def __init__(
     ) -> None:
         super().__init__(raw_response)
         self._self_wrap_stream = wrap_stream
-        self._self_finalize: Callable[[], None] | None = finalize
+        # Swapped for the wrapper's ``close`` once there is one, so an
+        # abandoned stream finalizes with what it saw.
+        self._self_finalize: Callable[[], Any] | None = finalize
         self._self_parsed: object | None = None
         self._install_close_fallback(raw_response)
 
@@ -125,21 +128,51 @@ async def _aclose(*args: object, **kwargs: object) -> object:
             try:
                 return await original(*args, **kwargs)
             finally:
-                self._finalize_close_fallback()
+                await self._afinalize_close_fallback()
 
         return _aclose
 
     def _finalize_close_fallback(self) -> None:
-        # When ``parse()`` built a stream wrapper, that wrapper owns
-        # finalization; only finalize here for callers that never parsed.
-        if self._self_parsed is None:
-            self._finalize_once()
+        if inspect.iscoroutinefunction(self._self_finalize):
+            return  # needs an await; ``aclose`` will run it
+        self._finalize_once()
+
+    async def _afinalize_close_fallback(self) -> None:
+        await self._afinalize_once()
+
+    def _take_finalize(self) -> Callable[[], Any] | None:
+        # Clearing on take also stops the recursion: closing the wrapper closes
+        # the SDK stream, which closes this response and re-enters the hook.
+        finalize, self._self_finalize = self._self_finalize, None
+        return finalize
 
     def _finalize_once(self) -> None:
-        # ``stop()`` is not idempotent, so finalize at most once.
-        if self._self_finalize is not None:
-            finalize, self._self_finalize = self._self_finalize, None
+        finalize = self._take_finalize()
+        if finalize is None:
+            return
+        try:
             finalize()
+        except Exception:  # pylint: disable=broad-exception-caught
+            # The caller already walked away; nobody is left to raise at.
+            _logger.debug(
+                "error finalizing an abandoned raw response", exc_info=True
+            )
+
+    async def _afinalize_once(self) -> None:
+        finalize = self._take_finalize()
+        if finalize is None:
+            return
+        try:
+            # ``invocation.stop`` returns None; an async wrapper's ``close``
+            # returns a coroutine.
+            result = finalize()
+            if inspect.isawaitable(result):
+                await result
+        except Exception:  # pylint: disable=broad-exception-caught
+            # See _finalize_once.
+            _logger.debug(
+                "error finalizing an abandoned raw response", exc_info=True
+            )
 
     def parse(self, *args: Any, **kwargs: Any) -> object:
         # We memoize the first parse regardless of the arguments it was called
@@ -148,11 +181,38 @@ def parse(self, *args: Any, **kwargs: Any) -> object:
         # deliberate: one raw response backs one span, and re-parsing a stream
         # would consume it twice anyway.
         if self._self_parsed is not None:
+            # An awaitable parse() is awaited on every call, memo hits included.
+            if inspect.iscoroutinefunction(self.__wrapped__.parse):
+                return self._parsed_as_awaitable()
             return self._self_parsed
-        stream = self.__wrapped__.parse(*args, **kwargs)
-        if isinstance(stream, (Stream, AsyncStream)):
-            self._self_parsed = self._self_wrap_stream(stream)
+        parsed = self.__wrapped__.parse(*args, **kwargs)
+        if inspect.isawaitable(parsed):
+            # Only ``AsyncAPIResponse`` (the async ``with_streaming_response``)
+            # has a coroutine ``parse()``.
+            return self._parse_awaited(parsed)
+        return self._wrap_parsed(parsed)
+
+    async def _parsed_as_awaitable(self) -> object:
+        return self._self_parsed
+
+    async def _parse_awaited(self, pending: Awaitable[Any]) -> object:
+        stream = await pending
+        # parse() can hand out several coroutines before any of them resolves,
+        # so re-check the memo to keep one wrapper per response.
+        if self._self_parsed is not None:
             return self._self_parsed
+        return self._wrap_parsed(stream)
+
+    def _wrap_parsed(self, stream: Any) -> object:
+        if isinstance(stream, (Stream, AsyncStream)):
+            if self._self_finalize is None:
+                # Already ended while parse() was pending, so a wrapper would
+                # collect attributes with nowhere left to record them.
+                return stream
+            wrapper = self._self_wrap_stream(stream)
+            self._self_parsed = wrapper
+            self._self_finalize = wrapper.close
+            return wrapper
         # Not a stream we can drive; hand it back untouched. The close fallback
         # finalizes the span once the caller drains/closes the response body.
         _logger.debug(

@@ -492,6 +492,87 @@ async def test_chat_completion_with_raw_response_streaming(
         )
 
 
+@pytest.mark.asyncio()
+async def test_async_chat_completion_with_streaming_response_parse(
+    span_exporter, async_openai_client, instrument_with_content, vcr
+):
+    """``AsyncAPIResponse.parse()`` is a coroutine, and must still be traced.
+
+    The async client's ``with_streaming_response`` is the only raw-response
+    entry point whose ``parse()`` is ``async``, so it returns a coroutine
+    rather than a stream.
+    """
+    with vcr.use_cassette(
+        "test_chat_completion_with_raw_response_streaming.yaml"
+    ):
+        async with (
+            async_openai_client.chat.completions.with_streaming_response.create(
+                messages=USER_ONLY_PROMPT,
+                model=DEFAULT_MODEL,
+                stream=True,
+                stream_options={"include_usage": True},
+            ) as raw_response
+        ):
+            assert "openai-version" in raw_response.headers
+
+            message_content = ""
+            usage = model = response_id = None
+            async for chunk in await raw_response.parse():
+                if chunk.choices:
+                    message_content += chunk.choices[0].delta.content or ""
+                if getattr(chunk, "usage", None):
+                    usage = chunk.usage
+                    model = chunk.model
+                    response_id = chunk.id
+
+    (span,) = span_exporter.get_finished_spans()
+    assert_all_attributes(
+        span,
+        DEFAULT_MODEL,
+        is_experimental_mode(),
+        response_id,
+        model,
+        usage.prompt_tokens,
+        usage.completion_tokens,
+        response_service_tier="default",
+    )
+    if is_experimental_mode():
+        assert_messages_attribute(
+            span.attributes["gen_ai.output.messages"],
+            format_simple_expected_output_message(message_content),
+        )
+
+
+@pytest.mark.asyncio()
+async def test_abandoned_async_streaming_response_still_emits_span(
+    span_exporter, async_openai_client, instrument_with_content, vcr
+):
+    """Walking away from a parsed stream must still end its span.
+
+    An early ``break`` leaves a wrapper nobody drained. Exiting the
+    ``with_streaming_response`` block closes the http response, and the close
+    fallback closes the wrapper so it finalizes with what it saw.
+    """
+    with vcr.use_cassette(
+        "test_chat_completion_with_raw_response_streaming.yaml"
+    ):
+        async with (
+            async_openai_client.chat.completions.with_streaming_response.create(
+                messages=USER_ONLY_PROMPT,
+                model=DEFAULT_MODEL,
+                stream=True,
+                stream_options={"include_usage": True},
+            ) as raw_response
+        ):
+            async for _chunk in await raw_response.parse():
+                break
+
+    (span,) = span_exporter.get_finished_spans()
+    assert span.end_time is not None
+    # The one chunk that was read is on the span; the rest never arrived.
+    assert span.attributes["gen_ai.response.id"]
+
+
 class _CustomChatCompletion(ChatCompletion):
     """Caller-defined response type passed to non-streaming parse(to=...)."""
 

@@ -516,6 +516,52 @@ async def test_async_responses_with_raw_response_streaming(
     )
 
 
+@pytest.mark.asyncio()
+async def test_async_responses_with_streaming_response_parse(
+    span_exporter, async_openai_client, instrument_with_content, vcr
+):
+    """``AsyncAPIResponse.parse()`` is a coroutine, and must still be traced.
+
+    Unlike every other raw-response entry point, the async client's
+    ``with_streaming_response`` returns a response whose ``parse()`` is
+    ``async``, so it hands back a coroutine rather than a stream. This is the
+    path the OpenAI Agents SDK takes for streamed runs.
+    """
+    _skip_if_not_latest()
+
+    with vcr.use_cassette(
+        "test_responses_create_streaming[content_mode0].yaml"
+    ):
+        async with (
+            async_openai_client.responses.with_streaming_response.create(
+                model=DEFAULT_MODEL,
+                instructions=SYSTEM_INSTRUCTIONS,
+                input=USER_ONLY_PROMPT[0]["content"],
+                service_tier="default",
+                stream=True,
+            )
+        ) as raw_response:
+            # Metadata resolves natively off the wrapper.
+            assert "openai-version" in raw_response.headers
+
+            response = await _collect_completed_response(
+                await raw_response.parse()
+            )
+
+    (span,) = span_exporter.get_finished_spans()
+    assert_all_attributes(
+        span,
+        DEFAULT_MODEL,
+        True,
+        response.id,
+        response.model,
+        response.usage.input_tokens,
+        response.usage.output_tokens,
+        request_service_tier="default",
+        response_service_tier=getattr(response, "service_tier", None),
+    )
+
+
 class _UnrelatedEvent(BaseModel):
     """An event type unrelated to the Responses stream events."""
 

@@ -767,6 +767,48 @@ def test_chat_completion_with_raw_response_streaming_read_without_parse(
     assert spans[0].end_time is not None
 
 
+def test_abandoned_streaming_response_still_emits_span(
+    span_exporter, openai_client, instrument_with_content, vcr
+):
+    """Sync counterpart of the async abandoned-stream case."""
+    with vcr.use_cassette(
+        "test_chat_completion_with_raw_response_streaming.yaml"
+    ):
+        with openai_client.chat.completions.with_streaming_response.create(
+            messages=USER_ONLY_PROMPT,
+            model=DEFAULT_MODEL,
+            stream=True,
+            stream_options={"include_usage": True},
+        ) as raw_response:
+            for _chunk in raw_response.parse():
+                break
+
+    (span,) = span_exporter.get_finished_spans()
+    assert span.end_time is not None
+    assert span.attributes["gen_ai.response.id"]
+
+
+def test_streaming_response_exception_in_block_still_emits_span(
+    span_exporter, openai_client, instrument_with_content, vcr
+):
+    """A caller exception inside the block also abandons the stream."""
+    with vcr.use_cassette(
+        "test_chat_completion_with_raw_response_streaming.yaml"
+    ):
+        with pytest.raises(ValueError):
+            with openai_client.chat.completions.with_streaming_response.create(
+                messages=USER_ONLY_PROMPT,
+                model=DEFAULT_MODEL,
+                stream=True,
+                stream_options={"include_usage": True},
+            ) as raw_response:
+                for _chunk in raw_response.parse():
+                    raise ValueError("caller blew up")
+
+    (span,) = span_exporter.get_finished_spans()
+    assert span.end_time is not None
+
+
 def test_chat_completion_tool_calls_with_content(
     span_exporter, log_exporter, openai_client, instrument_with_content, vcr
 ):
Original file line number	Diff line number	Diff line change
		@@ -0,0 +1 @@
		Record response telemetry for ``with_streaming_response`` calls on the async client, whose ``parse()`` returns a coroutine, and end the span for a parsed stream the caller abandons.