From fea1952e1e5ee18279157949f731fa4e5b746e04 Mon Sep 17 00:00:00 2001
From: Zening Chen <zening.chen@snowflake.com>
Date: Fri, 4 Sep 2026 01:07:03 +0000
Subject: [PATCH 1/8] [`opentelemetry-instrumentation-genai-openai`] Record
 response telemetry for async `with_streaming_response`

`RawResponseStreamProxy.parse()` assumed the wrapped response returned a
stream. That holds for `LegacyAPIResponse` and `APIResponse`, but the async
client's `with_streaming_response` yields an `AsyncAPIResponse`, whose
`parse()` is a coroutine. The isinstance check rejected it, so the caller
awaited an uninstrumented stream and the span closed carrying only
request-side attributes: no output messages, usage, finish reasons, response
id or response model.

This is the path the OpenAI Agents SDK takes for streamed runs, which resolves
`responses.with_streaming_response.create` and then awaits `parse()`, so every
streamed agent run produced an empty chat span.

Await the coroutine and wrap what it resolves to. `parse()` stays awaitable, so
callers are unaffected. Since `_raw_response` is shared, this covers both chat
completions and responses.
---
 .../.changelog/589.fixed                      |  1 +
 .../genai/openai/_raw_response.py             | 24 ++++-
 .../tests/test_async_chat_completions.py      | 51 ++++++++++
 .../tests/test_async_responses.py             | 46 +++++++++
 .../tests/test_raw_response_proxy.py          | 93 +++++++++++++++++--
 .../tests/test_responses.py                   | 35 +++++++
 6 files changed, 239 insertions(+), 11 deletions(-)
 create mode 100644 instrumentation/opentelemetry-instrumentation-genai-openai/.changelog/589.fixed

diff --git a/instrumentation/opentelemetry-instrumentation-genai-openai/.changelog/589.fixed b/instrumentation/opentelemetry-instrumentation-genai-openai/.changelog/589.fixed
new file mode 100644
index 000000000..c5ffcf3c5
--- /dev/null
+++ b/instrumentation/opentelemetry-instrumentation-genai-openai/.changelog/589.fixed
@@ -0,0 +1 @@
+Record response telemetry for ``with_streaming_response`` calls on the async client, whose ``parse()`` returns a coroutine.
diff --git a/instrumentation/opentelemetry-instrumentation-genai-openai/src/opentelemetry/instrumentation/genai/openai/_raw_response.py b/instrumentation/opentelemetry-instrumentation-genai-openai/src/opentelemetry/instrumentation/genai/openai/_raw_response.py
index b1d39a549..6f337e9bc 100644
--- a/instrumentation/opentelemetry-instrumentation-genai-openai/src/opentelemetry/instrumentation/genai/openai/_raw_response.py
+++ b/instrumentation/opentelemetry-instrumentation-genai-openai/src/opentelemetry/instrumentation/genai/openai/_raw_response.py
@@ -6,6 +6,7 @@
 from __future__ import annotations
 
 import functools
+import inspect
 import logging
 from collections.abc import Awaitable, Callable
 from typing import (
@@ -70,9 +71,9 @@ 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.
+    (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.
 
     The span is finalized independently of ``parse()``. Whether the caller
     parses and drains the wrapper, drains the body directly via the raw
@@ -149,7 +150,22 @@ def parse(self, *args: Any, **kwargs: Any) -> object:
         # would consume it twice anyway.
         if self._self_parsed is not None:
             return self._self_parsed
-        stream = self.__wrapped__.parse(*args, **kwargs)
+        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 _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)):
             self._self_parsed = self._self_wrap_stream(stream)
             return self._self_parsed
diff --git a/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_async_chat_completions.py b/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_async_chat_completions.py
index 394effd64..e78d9cc0f 100644
--- a/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_async_chat_completions.py
+++ b/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_async_chat_completions.py
@@ -492,6 +492,57 @@ 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),
+        )
+
+
 class _CustomChatCompletion(ChatCompletion):
     """Caller-defined response type passed to non-streaming parse(to=...)."""
 
diff --git a/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_async_responses.py b/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_async_responses.py
index 53baeb704..b44123207 100644
--- a/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_async_responses.py
+++ b/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_async_responses.py
@@ -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."""
 
diff --git a/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_raw_response_proxy.py b/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_raw_response_proxy.py
index 0192b87c3..c9c922898 100644
--- a/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_raw_response_proxy.py
+++ b/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_raw_response_proxy.py
@@ -4,14 +4,16 @@
 """Unit tests for the streaming ``with_raw_response`` proxy.
 
 These exercise ``RawResponseStreamProxy.parse()`` — it wraps only SDK streams
-it can drive and hands anything else back untouched — and the close fallback
-that finalizes the span when the caller never parses, so the span never leaks.
+it can drive, awaits the coroutine the async client's ``parse()`` returns, and
+hands anything else back untouched — and the close fallback that finalizes the
+span when the caller never parses, so the span never leaks.
 """
 
+import inspect
 import logging
 
 import pytest
-from openai import Stream
+from openai import AsyncStream, Stream
 
 from opentelemetry.instrumentation.genai.openai._raw_response import (
     RawResponseStreamProxy,
@@ -38,6 +40,12 @@ def __init__(self):
         pass
 
 
+class _FakeAsyncStream(AsyncStream):
+    # See _FakeStream.
+    def __init__(self):
+        pass
+
+
 class _RawResponse:
     headers = {"openai-version": "2020-10-01"}
     request_id = "req_123"
@@ -50,6 +58,13 @@ def parse(self, *, to=None):
         return self._parse_result
 
 
+class _AsyncRawResponse(_RawResponse):
+    """Shaped like ``AsyncAPIResponse``, whose ``parse()`` is a coroutine."""
+
+    async def parse(self, *, to=None):
+        return self._parse_result
+
+
 def _noop() -> None:
     pass
 
@@ -80,10 +95,10 @@ def test_parse_wraps_stream_and_memoizes():
 
 
 def test_parse_non_stream_returned_untouched(caplog):
-    # parse() may return something we can't drive — e.g. the coroutine the SDK
-    # documents parse() "will become in the next major version", or a custom
-    # non-stream target. The proxy hands it back untouched, logs, and does NOT
-    # finalize; the close fallback stays armed to finalize on drain/close.
+    # parse() may return something we can't drive — e.g. a custom non-stream
+    # target passed as ``to=``. The proxy hands it back untouched, logs, and
+    # does NOT finalize; the close fallback stays armed to finalize on
+    # drain/close.
     raw = _RawResponse("not-a-stream")
     wrapped_calls = []
     finalize_calls = []
@@ -108,6 +123,70 @@ def test_parse_non_stream_returned_untouched(caplog):
     assert finalize_calls == [True]
 
 
+@pytest.mark.asyncio()
+async def test_parse_awaits_coroutine_and_wraps():
+    # The async client's with_streaming_response hands back an AsyncAPIResponse
+    # whose parse() is a coroutine. parse() must stay awaitable for the caller
+    # and resolve to the instrumented wrapper, not the bare SDK stream.
+    stream = _FakeAsyncStream()
+    raw = _AsyncRawResponse(stream)
+    proxy = RawResponseStreamProxy(
+        raw,
+        wrap_stream=lambda s: ("wrapped", s),
+        finalize=_noop,
+    )
+
+    pending = proxy.parse()
+    assert inspect.isawaitable(pending)  # caller still writes `await parse()`
+    assert await pending == ("wrapped", stream)
+
+    # Memoized, so a second parse() shares the one wrapper / span. The proxy
+    # already holds the wrapper, so this call returns it without awaiting.
+    assert proxy.parse() == ("wrapped", stream)
+
+
+@pytest.mark.asyncio()
+async def test_parse_awaited_non_stream_returned_untouched(caplog):
+    # A coroutine parse() resolving to something we can't drive is handed back
+    # untouched, exactly like the synchronous case.
+    raw = _AsyncRawResponse("not-a-stream")
+    finalize_calls = []
+    proxy = RawResponseStreamProxy(
+        raw,
+        wrap_stream=lambda s: ("wrapped", s),
+        finalize=lambda: finalize_calls.append(True),
+    )
+
+    with caplog.at_level(
+        logging.DEBUG,
+        logger="opentelemetry.instrumentation.genai.openai._raw_response",
+    ):
+        parsed = await proxy.parse()
+
+    assert parsed == "not-a-stream"
+    assert finalize_calls == []  # parse must not finalize; the close hook does
+    assert "skipping stream instrumentation" in caplog.text
+
+
+@pytest.mark.asyncio()
+async def test_close_before_awaiting_parse_finalizes_once():
+    # parse() returning an un-awaited coroutine leaves the proxy without a
+    # wrapper, so the close fallback is still what finalizes the span.
+    raw = _AsyncRawResponse(_FakeAsyncStream())
+    finalize_calls = []
+    proxy = RawResponseStreamProxy(
+        raw,
+        wrap_stream=lambda s: ("wrapped", s),
+        finalize=lambda: finalize_calls.append(True),
+    )
+
+    pending = proxy.parse()
+    raw.http_response.close()
+    assert finalize_calls == [True]  # span closed, did not leak
+
+    await pending  # tidy up the coroutine so it is not left un-awaited
+
+
 def test_close_without_parse_finalizes_once():
     # A caller can drain the body off the http response without ever calling
     # parse(). Closing it must finalize the span exactly once (so it does not
diff --git a/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_responses.py b/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_responses.py
index 9753d71e0..05662ff5e 100644
--- a/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_responses.py
+++ b/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_responses.py
@@ -779,6 +779,41 @@ def test_responses_with_raw_response_streaming(
     )
 
 
+def test_responses_with_streaming_response_parse(
+    span_exporter, openai_client, instrument_with_content, vcr
+):
+    """``with_streaming_response`` + ``parse()`` traces like a plain stream."""
+    _skip_if_not_latest()
+
+    with vcr.use_cassette(
+        "test_responses_create_streaming[content_mode0].yaml"
+    ):
+        with 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 = _collect_completed_response(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."""
 

From 474cedbf8613e3fce1e5c8441efdb1a67180f161 Mon Sep 17 00:00:00 2001
From: Zening Chen <zening.chen@snowflake.com>
Date: Fri, 4 Sep 2026 01:09:58 +0000
Subject: [PATCH 2/8] Rename changelog fragment to the PR number

---
 .../.changelog/{589.fixed => 610.fixed}                           | 0
 1 file changed, 0 insertions(+), 0 deletions(-)
 rename instrumentation/opentelemetry-instrumentation-genai-openai/.changelog/{589.fixed => 610.fixed} (100%)

diff --git a/instrumentation/opentelemetry-instrumentation-genai-openai/.changelog/589.fixed b/instrumentation/opentelemetry-instrumentation-genai-openai/.changelog/610.fixed
similarity index 100%
rename from instrumentation/opentelemetry-instrumentation-genai-openai/.changelog/589.fixed
rename to instrumentation/opentelemetry-instrumentation-genai-openai/.changelog/610.fixed

From dfda6518dcfdc2ac183f9c13d25ef350d9e2bb8d Mon Sep 17 00:00:00 2001
From: Zening Chen <zening.chen@snowflake.com>
Date: Fri, 4 Sep 2026 01:22:10 +0000
Subject: [PATCH 3/8] Keep parse() awaitable when the memo is hit on the async
 path

The memo check at the top of parse() returned the wrapper directly, so a
second call on an async response handed back a non-awaitable value and the
caller's await raised TypeError.

The wrapper type cannot tell us whether parse() is async: the async client
pairs a synchronous LegacyAPIResponse.parse() with an AsyncStream. So record
whether the wrapped parse() actually returned an awaitable, and keep memo hits
awaitable when it did.
---
 .../genai/openai/_raw_response.py             |  8 +++++++
 .../tests/test_raw_response_proxy.py          | 22 ++++++++++++++++---
 2 files changed, 27 insertions(+), 3 deletions(-)

diff --git a/instrumentation/opentelemetry-instrumentation-genai-openai/src/opentelemetry/instrumentation/genai/openai/_raw_response.py b/instrumentation/opentelemetry-instrumentation-genai-openai/src/opentelemetry/instrumentation/genai/openai/_raw_response.py
index 6f337e9bc..9b905ea84 100644
--- a/instrumentation/opentelemetry-instrumentation-genai-openai/src/opentelemetry/instrumentation/genai/openai/_raw_response.py
+++ b/instrumentation/opentelemetry-instrumentation-genai-openai/src/opentelemetry/instrumentation/genai/openai/_raw_response.py
@@ -93,6 +93,7 @@ def __init__(
         self._self_wrap_stream = wrap_stream
         self._self_finalize: Callable[[], None] | None = finalize
         self._self_parsed: object | None = None
+        self._self_parse_returns_awaitable = False
         self._install_close_fallback(raw_response)
 
     def _install_close_fallback(self, raw_response: RawResponseLike) -> None:
@@ -149,14 +150,21 @@ 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 self._self_parse_returns_awaitable:
+                return self._parsed_as_awaitable()
             return self._self_parsed
         parsed = self.__wrapped__.parse(*args, **kwargs)
         if inspect.isawaitable(parsed):
             # Only ``AsyncAPIResponse`` (the async ``with_streaming_response``)
             # has a coroutine ``parse()``.
+            self._self_parse_returns_awaitable = True
             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,
diff --git a/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_raw_response_proxy.py b/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_raw_response_proxy.py
index c9c922898..52f0f5f28 100644
--- a/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_raw_response_proxy.py
+++ b/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_raw_response_proxy.py
@@ -140,9 +140,25 @@ async def test_parse_awaits_coroutine_and_wraps():
     assert inspect.isawaitable(pending)  # caller still writes `await parse()`
     assert await pending == ("wrapped", stream)
 
-    # Memoized, so a second parse() shares the one wrapper / span. The proxy
-    # already holds the wrapper, so this call returns it without awaiting.
-    assert proxy.parse() == ("wrapped", stream)
+
+@pytest.mark.asyncio()
+async def test_parse_stays_awaitable_once_memoized():
+    # Memoizing must not change the shape parse() returns: the caller writes
+    # `await parse()` every time, so a memoized hit has to stay awaitable
+    # instead of handing back the bare wrapper.
+    stream = _FakeAsyncStream()
+    raw = _AsyncRawResponse(stream)
+    proxy = RawResponseStreamProxy(
+        raw,
+        wrap_stream=lambda s: ("wrapped", s),
+        finalize=_noop,
+    )
+
+    first = await proxy.parse()
+
+    second = proxy.parse()
+    assert inspect.isawaitable(second)
+    assert await second is first  # one wrapper / span, shared
 
 
 @pytest.mark.asyncio()

From 5c6b162e1dfefb0fde1fe86d08b2e151c960eff3 Mon Sep 17 00:00:00 2001
From: Zening Chen <zening.chen@snowflake.com>
Date: Fri, 4 Sep 2026 20:49:46 +0000
Subject: [PATCH 4/8] Do not wrap a stream once the close fallback has ended
 the span

When parse() is called but not awaited before the httpx response closes, the
close fallback sees no wrapper and finalizes the span. If the coroutine then
resolved to a stream, _wrap_parsed still built a wrapper: it would collect
attributes and call stop() on an invocation that had already finished, so the
response data was silently dropped.

Hand the stream back uninstrumented in that case. _self_finalize is cleared
only by the close fallback, so it already records whether the span is gone.
---
 .../genai/openai/_raw_response.py             |  5 +++
 .../tests/test_raw_response_proxy.py          | 33 +++++++++++++++++++
 2 files changed, 38 insertions(+)

diff --git a/instrumentation/opentelemetry-instrumentation-genai-openai/src/opentelemetry/instrumentation/genai/openai/_raw_response.py b/instrumentation/opentelemetry-instrumentation-genai-openai/src/opentelemetry/instrumentation/genai/openai/_raw_response.py
index 9b905ea84..9099b30fb 100644
--- a/instrumentation/opentelemetry-instrumentation-genai-openai/src/opentelemetry/instrumentation/genai/openai/_raw_response.py
+++ b/instrumentation/opentelemetry-instrumentation-genai-openai/src/opentelemetry/instrumentation/genai/openai/_raw_response.py
@@ -175,6 +175,11 @@ async def _parse_awaited(self, pending: Awaitable[Any]) -> object:
 
     def _wrap_parsed(self, stream: Any) -> object:
         if isinstance(stream, (Stream, AsyncStream)):
+            if self._self_finalize is None:
+                # The close fallback ended the span while parse() was still
+                # pending. A wrapper would collect attributes with nowhere left
+                # to record them, so hand the stream back uninstrumented.
+                return stream
             self._self_parsed = self._self_wrap_stream(stream)
             return self._self_parsed
         # Not a stream we can drive; hand it back untouched. The close fallback
diff --git a/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_raw_response_proxy.py b/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_raw_response_proxy.py
index 52f0f5f28..d039539b1 100644
--- a/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_raw_response_proxy.py
+++ b/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_raw_response_proxy.py
@@ -9,6 +9,7 @@
 span when the caller never parses, so the span never leaks.
 """
 
+import asyncio
 import inspect
 import logging
 
@@ -203,6 +204,38 @@ async def test_close_before_awaiting_parse_finalizes_once():
     await pending  # tidy up the coroutine so it is not left un-awaited
 
 
+@pytest.mark.asyncio()
+async def test_close_while_parse_pending_does_not_wrap():
+    # Same race as above, but the coroutine resolves to a stream we could wrap.
+    # The close fallback has already ended the span by then, so a wrapper would
+    # accumulate attributes that silently go nowhere -- worse than no wrapper.
+    resume = asyncio.Event()
+    stream = _FakeAsyncStream()
+
+    class _SlowAsyncRawResponse(_AsyncRawResponse):
+        async def parse(self, *, to=None):
+            await resume.wait()
+            return self._parse_result
+
+    raw = _SlowAsyncRawResponse(stream)
+    finalize_calls, wrap_calls = [], []
+    proxy = RawResponseStreamProxy(
+        raw,
+        wrap_stream=lambda s: wrap_calls.append(s) or ("wrapped", s),
+        finalize=lambda: finalize_calls.append(True),
+    )
+
+    pending = proxy.parse()
+    raw.http_response.close()
+    assert finalize_calls == [True]  # close fallback already finalized
+
+    resume.set()
+    result = await pending
+
+    assert wrap_calls == []  # no wrapper built against a finished span
+    assert result is stream  # caller still gets a usable stream
+
+
 def test_close_without_parse_finalizes_once():
     # A caller can drain the body off the http response without ever calling
     # parse(). Closing it must finalize the span exactly once (so it does not

From 43cb3b5bb13683f540349b2e4a249b2b2cffe96c Mon Sep 17 00:00:00 2001
From: Zening Chen <zening.chen@snowflake.com>
Date: Mon, 7 Sep 2026 06:47:46 +0000
Subject: [PATCH 5/8] End the span when the caller abandons a parsed stream

Once parse() built a wrapper, the close fallback stood aside on the grounds
that the wrapper owns finalization. But a wrapper only finalizes when it is
drained, closed, or used as a context manager, so a caller that breaks out of
the loop or raises inside the with block left nobody to end the span. On the
async path this was a regression from wrapping the coroutine parse(): before,
no wrapper was built and the fallback still ended the span, empty.

The hooks now drive the wrapper instead of standing aside, so it finalizes
with what it saw. Closing it closes the SDK stream, which closes this same
httpx response and re-enters the hook. Rather than guard that separately, swap
the existing take-once finalize callback for the wrapper close: taking it
clears it, so finalization stays idempotent and the re-entrant call finds
nothing to run.
---
 .../.changelog/610.fixed                      |   2 +-
 .../genai/openai/_raw_response.py             |  77 +++++++++----
 .../tests/test_async_chat_completions.py      |  30 +++++
 .../tests/test_chat_completions.py            |  42 +++++++
 .../tests/test_raw_response_proxy.py          | 109 +++++++++++++++---
 5 files changed, 221 insertions(+), 39 deletions(-)

diff --git a/instrumentation/opentelemetry-instrumentation-genai-openai/.changelog/610.fixed b/instrumentation/opentelemetry-instrumentation-genai-openai/.changelog/610.fixed
index c5ffcf3c5..149bc1087 100644
--- a/instrumentation/opentelemetry-instrumentation-genai-openai/.changelog/610.fixed
+++ b/instrumentation/opentelemetry-instrumentation-genai-openai/.changelog/610.fixed
@@ -1 +1 @@
-Record response telemetry for ``with_streaming_response`` calls on the async client, whose ``parse()`` returns a coroutine.
+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.
diff --git a/instrumentation/opentelemetry-instrumentation-genai-openai/src/opentelemetry/instrumentation/genai/openai/_raw_response.py b/instrumentation/opentelemetry-instrumentation-genai-openai/src/opentelemetry/instrumentation/genai/openai/_raw_response.py
index 9099b30fb..52f6ad1a0 100644
--- a/instrumentation/opentelemetry-instrumentation-genai-openai/src/opentelemetry/instrumentation/genai/openai/_raw_response.py
+++ b/instrumentation/opentelemetry-instrumentation-genai-openai/src/opentelemetry/instrumentation/genai/openai/_raw_response.py
@@ -75,12 +75,12 @@ class RawResponseStreamProxy(ObjectProxy):
     we just log and step aside. On the async client ``parse()`` is a coroutine,
     so we await it and wrap what it resolves to.
 
-    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.
+    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__(
@@ -91,9 +91,10 @@ 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._self_parse_returns_awaitable = False
         self._install_close_fallback(raw_response)
 
     def _install_close_fallback(self, raw_response: RawResponseLike) -> None:
@@ -127,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
@@ -151,14 +182,13 @@ def parse(self, *args: Any, **kwargs: Any) -> object:
         # would consume it twice anyway.
         if self._self_parsed is not None:
             # An awaitable parse() is awaited on every call, memo hits included.
-            if self._self_parse_returns_awaitable:
+            if inspect.iscoroutinefunction(self.__wrapped__.parse):
                 return self._parsed_as_awaitable()
             return self._self_parsed
         parsed = self.__wrapped__.parse(*args, **kwargs)
         if inspect.isawaitable(parsed):
             # Only ``AsyncAPIResponse`` (the async ``with_streaming_response``)
             # has a coroutine ``parse()``.
-            self._self_parse_returns_awaitable = True
             return self._parse_awaited(parsed)
         return self._wrap_parsed(parsed)
 
@@ -176,12 +206,13 @@ async def _parse_awaited(self, pending: Awaitable[Any]) -> object:
     def _wrap_parsed(self, stream: Any) -> object:
         if isinstance(stream, (Stream, AsyncStream)):
             if self._self_finalize is None:
-                # The close fallback ended the span while parse() was still
-                # pending. A wrapper would collect attributes with nowhere left
-                # to record them, so hand the stream back uninstrumented.
+                # Already ended while parse() was pending, so a wrapper would
+                # collect attributes with nowhere left to record them.
                 return stream
-            self._self_parsed = self._self_wrap_stream(stream)
-            return self._self_parsed
+            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(
diff --git a/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_async_chat_completions.py b/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_async_chat_completions.py
index 9c5b7a7ae..c66002468 100644
--- a/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_async_chat_completions.py
+++ b/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_async_chat_completions.py
@@ -543,6 +543,36 @@ async def test_async_chat_completion_with_streaming_response_parse(
         )
 
 
+@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=...)."""
 
diff --git a/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_chat_completions.py b/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_chat_completions.py
index 7a963b1df..f3b9c29b4 100644
--- a/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_chat_completions.py
+++ b/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_chat_completions.py
@@ -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
 ):
diff --git a/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_raw_response_proxy.py b/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_raw_response_proxy.py
index d039539b1..514335657 100644
--- a/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_raw_response_proxy.py
+++ b/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_raw_response_proxy.py
@@ -33,6 +33,20 @@ def __init__(self):
     def close(self):
         self.close_calls += 1
 
+    async def aclose(self):
+        self.close_calls += 1
+
+
+class _SyncWrapper:
+    """Stands in for a sync stream wrapper, which finalizes on close()."""
+
+    def __init__(self, stream=None):
+        self.stream = stream
+        self.close_calls = 0
+
+    def close(self):
+        self.close_calls += 1
+
 
 class _FakeStream(Stream):
     # Bypass Stream.__init__ (needs an httpx response + client); the proxy only
@@ -81,7 +95,7 @@ def test_parse_wraps_stream_and_memoizes():
     raw = _RawResponse(stream)
     proxy = RawResponseStreamProxy(
         raw,
-        wrap_stream=lambda s: ("wrapped", s),
+        wrap_stream=_SyncWrapper,
         finalize=_noop,
     )
 
@@ -90,7 +104,7 @@ def test_parse_wraps_stream_and_memoizes():
     assert proxy.request_id == "req_123"
 
     parsed = proxy.parse()
-    assert parsed == ("wrapped", stream)
+    assert parsed.stream is stream
     # Memoized so repeated calls share one wrapper / span.
     assert proxy.parse() is parsed
 
@@ -105,7 +119,7 @@ def test_parse_non_stream_returned_untouched(caplog):
     finalize_calls = []
     proxy = RawResponseStreamProxy(
         raw,
-        wrap_stream=lambda s: wrapped_calls.append(s) or ("wrapped", s),
+        wrap_stream=lambda s: wrapped_calls.append(s) or _SyncWrapper(s),
         finalize=lambda: finalize_calls.append(True),
     )
 
@@ -133,13 +147,13 @@ async def test_parse_awaits_coroutine_and_wraps():
     raw = _AsyncRawResponse(stream)
     proxy = RawResponseStreamProxy(
         raw,
-        wrap_stream=lambda s: ("wrapped", s),
+        wrap_stream=_SyncWrapper,
         finalize=_noop,
     )
 
     pending = proxy.parse()
     assert inspect.isawaitable(pending)  # caller still writes `await parse()`
-    assert await pending == ("wrapped", stream)
+    assert (await pending).stream is stream
 
 
 @pytest.mark.asyncio()
@@ -151,7 +165,7 @@ async def test_parse_stays_awaitable_once_memoized():
     raw = _AsyncRawResponse(stream)
     proxy = RawResponseStreamProxy(
         raw,
-        wrap_stream=lambda s: ("wrapped", s),
+        wrap_stream=_SyncWrapper,
         finalize=_noop,
     )
 
@@ -170,7 +184,7 @@ async def test_parse_awaited_non_stream_returned_untouched(caplog):
     finalize_calls = []
     proxy = RawResponseStreamProxy(
         raw,
-        wrap_stream=lambda s: ("wrapped", s),
+        wrap_stream=_SyncWrapper,
         finalize=lambda: finalize_calls.append(True),
     )
 
@@ -193,7 +207,7 @@ async def test_close_before_awaiting_parse_finalizes_once():
     finalize_calls = []
     proxy = RawResponseStreamProxy(
         raw,
-        wrap_stream=lambda s: ("wrapped", s),
+        wrap_stream=_SyncWrapper,
         finalize=lambda: finalize_calls.append(True),
     )
 
@@ -221,7 +235,7 @@ async def parse(self, *, to=None):
     finalize_calls, wrap_calls = [], []
     proxy = RawResponseStreamProxy(
         raw,
-        wrap_stream=lambda s: wrap_calls.append(s) or ("wrapped", s),
+        wrap_stream=lambda s: wrap_calls.append(s) or _SyncWrapper(s),
         finalize=lambda: finalize_calls.append(True),
     )
 
@@ -246,7 +260,7 @@ def test_close_without_parse_finalizes_once():
     # http_response keeps it alive, so we drive the response directly.
     RawResponseStreamProxy(
         raw,
-        wrap_stream=lambda s: ("wrapped", s),
+        wrap_stream=_SyncWrapper,
         finalize=lambda: finalize_calls.append(True),
     )
 
@@ -258,21 +272,86 @@ def test_close_without_parse_finalizes_once():
     assert finalize_calls == [True]
 
 
-def test_close_after_parse_does_not_double_finalize():
-    # When parse() built a stream wrapper, that wrapper owns finalization; the
-    # close fallback must stay out of the way and not finalize the span itself.
+class _AsyncWrapper:
+    """Stands in for an async stream wrapper, whose close() is a coroutine."""
+
+    def __init__(self, stream=None):
+        self.stream = stream
+        self.close_calls = 0
+
+    async def close(self):
+        self.close_calls += 1
+
+
+def test_close_after_parse_drives_the_wrapper():
+    # A wrapper owns finalization, but only once something drives it. A caller
+    # that parses and then walks away (an early break, an exception inside the
+    # with block) never does, so the close fallback closes the wrapper rather
+    # than standing aside; the wrapper finalizes with what it saw.
+    wrapper = _SyncWrapper()
     raw = _RawResponse(_FakeStream())
     finalize_calls = []
     proxy = RawResponseStreamProxy(
         raw,
-        wrap_stream=lambda s: ("wrapped", s),
+        wrap_stream=lambda s: wrapper,
         finalize=lambda: finalize_calls.append(True),
     )
 
     proxy.parse()
     raw.http_response.close()
     assert raw.http_response.close_calls == 1  # real close still ran
-    assert finalize_calls == []  # fallback did not fire
+    assert wrapper.close_calls == 1  # wrapper was driven
+    assert finalize_calls == []  # the wrapper ends the span, not the fallback
+
+    raw.http_response.close()  # a second close must not drive it again
+    assert wrapper.close_calls == 1
+    assert finalize_calls == []
+
+
+def test_closing_the_wrapper_reentrantly_does_not_end_the_span():
+    # Closing the wrapper closes the SDK stream, which closes this same httpx
+    # response and re-enters the hook. The re-entrant call must not take the
+    # "nobody parsed" branch and end the span before the wrapper finalizes it.
+    raw = _RawResponse(_FakeStream())
+    finalize_calls = []
+
+    class _ReentrantWrapper:
+        def close(self):
+            raw.http_response.close()
+
+    proxy = RawResponseStreamProxy(
+        raw,
+        wrap_stream=lambda s: _ReentrantWrapper(),
+        finalize=lambda: finalize_calls.append(True),
+    )
+
+    proxy.parse()
+    raw.http_response.close()
+    assert finalize_calls == []  # the re-entrant close did not finalize
+
+
+@pytest.mark.asyncio()
+async def test_async_wrapper_is_closed_by_aclose_not_close():
+    # httpx rejects a sync close of an async response and nothing in the sync
+    # hook can await, so an async wrapper has to wait for the aclose hook.
+    wrapper = _AsyncWrapper()
+    raw = _AsyncRawResponse(_FakeAsyncStream())
+    finalize_calls = []
+    proxy = RawResponseStreamProxy(
+        raw,
+        wrap_stream=lambda s: wrapper,
+        finalize=lambda: finalize_calls.append(True),
+    )
+
+    await proxy.parse()
+
+    raw.http_response.close()  # sync hook leaves the wrapper alone
+    assert wrapper.close_calls == 0
+    assert finalize_calls == []
+
+    await raw.http_response.aclose()
+    assert wrapper.close_calls == 1  # aclose drove it
+    assert finalize_calls == []
 
 
 def test_wrap_result_non_stream_finalizes_on_close(

From 2bf302af7758de44aeac7230e8ac0c5aecc61c7e Mon Sep 17 00:00:00 2001
From: Zening Chen <zening.chen@snowflake.com>
Date: Mon, 14 Sep 2026 01:44:24 +0000
Subject: [PATCH 6/8] Replay the shape the first parse returned on memo hits

The memo hit derived the shape from iscoroutinefunction on the wrapped
parse, while the first parse decided on isawaitable of the value it got
back. Those agree for the SDK response classes, but not for a plain method
that hands back an awaitable -- a sync decorator around an async parse, or
one of the duck-typed responses the structural protocol deliberately
accepts. Record what the first parse actually returned and replay that.

Also widen the finalize parameter annotation to match the field it lands
in: it is swapped for an async wrapper close, which returns a coroutine.
---
 .../genai/openai/_raw_response.py             | 10 +++---
 .../tests/test_raw_response_proxy.py          | 35 +++++++++++++++++++
 2 files changed, 40 insertions(+), 5 deletions(-)

diff --git a/instrumentation/opentelemetry-instrumentation-genai-openai/src/opentelemetry/instrumentation/genai/openai/_raw_response.py b/instrumentation/opentelemetry-instrumentation-genai-openai/src/opentelemetry/instrumentation/genai/openai/_raw_response.py
index 52f6ad1a0..284499bc3 100644
--- a/instrumentation/opentelemetry-instrumentation-genai-openai/src/opentelemetry/instrumentation/genai/openai/_raw_response.py
+++ b/instrumentation/opentelemetry-instrumentation-genai-openai/src/opentelemetry/instrumentation/genai/openai/_raw_response.py
@@ -87,14 +87,14 @@ def __init__(
         self,
         raw_response: RawResponseLike,
         wrap_stream: Callable[[AnyStream], object],
-        finalize: Callable[[], None],
+        finalize: Callable[[], Any],
     ) -> None:
         super().__init__(raw_response)
         self._self_wrap_stream = wrap_stream
-        # Swapped for the wrapper's ``close`` once there is one, so an
-        # abandoned stream finalizes with what it saw.
+        # Swapped for the wrapper's ``close`` once there is one.
         self._self_finalize: Callable[[], Any] | None = finalize
         self._self_parsed: object | None = None
+        self._self_parse_returns_awaitable = False
         self._install_close_fallback(raw_response)
 
     def _install_close_fallback(self, raw_response: RawResponseLike) -> None:
@@ -181,14 +181,14 @@ 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):
+            if self._self_parse_returns_awaitable:
                 return self._parsed_as_awaitable()
             return self._self_parsed
         parsed = self.__wrapped__.parse(*args, **kwargs)
         if inspect.isawaitable(parsed):
             # Only ``AsyncAPIResponse`` (the async ``with_streaming_response``)
             # has a coroutine ``parse()``.
+            self._self_parse_returns_awaitable = True
             return self._parse_awaited(parsed)
         return self._wrap_parsed(parsed)
 
diff --git a/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_raw_response_proxy.py b/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_raw_response_proxy.py
index 514335657..66d375783 100644
--- a/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_raw_response_proxy.py
+++ b/instrumentation/opentelemetry-instrumentation-genai-openai/tests/test_raw_response_proxy.py
@@ -80,6 +80,20 @@ async def parse(self, *, to=None):
         return self._parse_result
 
 
+class _AwaitableParseRawResponse(_RawResponse):
+    """A plain ``parse()`` that nonetheless hands back an awaitable.
+
+    ``iscoroutinefunction`` says no while ``isawaitable`` says yes, so the two
+    parse paths have to agree on the value rather than on the signature.
+    """
+
+    def parse(self, *, to=None):
+        async def _resolve():
+            return self._parse_result
+
+        return _resolve()
+
+
 def _noop() -> None:
     pass
 
@@ -176,6 +190,27 @@ async def test_parse_stays_awaitable_once_memoized():
     assert await second is first  # one wrapper / span, shared
 
 
+@pytest.mark.asyncio()
+async def test_memo_replays_the_shape_the_first_parse_returned():
+    # The memo hit has no value left to inspect, so it has to replay what the
+    # first parse actually returned. Deriving it from the wrapped method's
+    # signature instead would miss a plain parse() that returns an awaitable.
+    raw = _AwaitableParseRawResponse(_FakeAsyncStream())
+    assert not inspect.iscoroutinefunction(type(raw).parse)  # signature lies
+
+    proxy = RawResponseStreamProxy(
+        raw,
+        wrap_stream=_AsyncWrapper,
+        finalize=_noop,
+    )
+
+    first = await proxy.parse()
+
+    second = proxy.parse()
+    assert inspect.isawaitable(second)
+    assert await second is first
+
+
 @pytest.mark.asyncio()
 async def test_parse_awaited_non_stream_returned_untouched(caplog):
     # A coroutine parse() resolving to something we can't drive is handed back

From cd60c9b516ac36f83c86ba44cb7c81da49172db9 Mon Sep 17 00:00:00 2001
From: Zening Chen <zening.chen@snowflake.com>
Date: Sun, 13 Sep 2026 19:00:25 -0700
Subject: [PATCH 7/8] Potential fix for pull request finding

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
---
 .../opentelemetry/instrumentation/genai/openai/_raw_response.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/instrumentation/opentelemetry-instrumentation-genai-openai/src/opentelemetry/instrumentation/genai/openai/_raw_response.py b/instrumentation/opentelemetry-instrumentation-genai-openai/src/opentelemetry/instrumentation/genai/openai/_raw_response.py
index 284499bc3..1acc24c83 100644
--- a/instrumentation/opentelemetry-instrumentation-genai-openai/src/opentelemetry/instrumentation/genai/openai/_raw_response.py
+++ b/instrumentation/opentelemetry-instrumentation-genai-openai/src/opentelemetry/instrumentation/genai/openai/_raw_response.py
@@ -152,7 +152,7 @@ def _finalize_once(self) -> None:
             return
         try:
             finalize()
-        except Exception:  # pylint: disable=broad-exception-caught
+        except BaseException:  # 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

From 647898fb3bd0eb466762250552a8e5bc81aab352 Mon Sep 17 00:00:00 2001
From: Zening Chen <zening.chen@snowflake.com>
Date: Sun, 13 Sep 2026 20:26:15 -0700
Subject: [PATCH 8/8] Potential fix for pull request finding

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
---
 .../opentelemetry/instrumentation/genai/openai/_raw_response.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/instrumentation/opentelemetry-instrumentation-genai-openai/src/opentelemetry/instrumentation/genai/openai/_raw_response.py b/instrumentation/opentelemetry-instrumentation-genai-openai/src/opentelemetry/instrumentation/genai/openai/_raw_response.py
index 1acc24c83..31a279782 100644
--- a/instrumentation/opentelemetry-instrumentation-genai-openai/src/opentelemetry/instrumentation/genai/openai/_raw_response.py
+++ b/instrumentation/opentelemetry-instrumentation-genai-openai/src/opentelemetry/instrumentation/genai/openai/_raw_response.py
@@ -168,7 +168,7 @@ async def _afinalize_once(self) -> None:
             result = finalize()
             if inspect.isawaitable(result):
                 await result
-        except Exception:  # pylint: disable=broad-exception-caught
+        except BaseException:  # pylint: disable=broad-exception-caught
             # See _finalize_once.
             _logger.debug(
                 "error finalizing an abandoned raw response", exc_info=True