diff --git a/instrumentation/opentelemetry-instrumentation-genai-openai/.changelog/610.fixed b/instrumentation/opentelemetry-instrumentation-genai-openai/.changelog/610.fixed
new file mode 100644
index 000000000..149bc1087
--- /dev/null
+++ b/instrumentation/opentelemetry-instrumentation-genai-openai/.changelog/610.fixed
@@ -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.
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..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
@@ -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(
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 641b2f9d2..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
@@ -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=...)."""
 
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_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 0192b87c3..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
@@ -4,14 +4,17 @@
 """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 asyncio
+import inspect
 import logging
 
 import pytest
-from openai import Stream
+from openai import AsyncStream, Stream
 
 from opentelemetry.instrumentation.genai.openai._raw_response import (
     RawResponseStreamProxy,
@@ -30,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
@@ -38,6 +55,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 +73,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
 
@@ -65,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,
     )
 
@@ -74,22 +104,22 @@ 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
 
 
 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 = []
     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),
     )
 
@@ -108,6 +138,118 @@ 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=_SyncWrapper,
+        finalize=_noop,
+    )
+
+    pending = proxy.parse()
+    assert inspect.isawaitable(pending)  # caller still writes `await parse()`
+    assert (await pending).stream is 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=_SyncWrapper,
+        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()
+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=_SyncWrapper,
+        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=_SyncWrapper,
+        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
+
+
+@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 _SyncWrapper(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
@@ -118,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),
     )
 
@@ -130,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(
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."""