fix(anthropic): preserve caller errors in streaming responses (#397)

* fix(anthropic): preserve caller errors in streaming responses

* chore(anthropic): add changelog fragment for streaming errors

* fix(anthropic): preserve streaming exception semantics

* fix(anthropic): preserve response context manager types

Forward caller failures through the SDK response context managers while retaining the native SDK types and idempotent stream finalization. Add sync and async regression coverage.

Assisted-by: ChatGPT 5.2

* test(anthropic): cover instrumentation round trips

Add sync and async regression coverage for preserving the SDK streaming response context manager across instrument and uninstrument cycles.\n\nAssisted-by: ChatGPT 5.2

* test(anthropic): cover raw response caller failures

Assisted-by: ChatGPT 5.2

* docs(anthropic): clarify async response wrapper

Assisted-by: ChatGPT 5.2

* docs(anthropic): fix async_messages_create docstring

---------

Co-authored-by: Surya <suryateja9108@gmail.com>
Co-authored-by: Liudmila Molkova <neskazu@gmail.com>

Author: 3 people

instrumentation/opentelemetry-instrumentation-genai-anthropic/.changelog/397.fixed

@@ -0,0 +1 @@
+preserve caller exceptions when closing Anthropic streaming responses

instrumentation/opentelemetry-instrumentation-genai-anthropic/src/opentelemetry/instrumentation/genai/anthropic/__init__.py

@@ -52,8 +52,10 @@
 from .patch import (
     async_messages_create,
     async_messages_stream,
+    async_response_context_manager_exit,
     messages_create,
     messages_stream,
+    response_context_manager_exit,
 )
 
 
@@ -134,6 +136,16 @@ def _instrument(self, **kwargs: Any) -> None:
             "AsyncMessages.stream",
             async_messages_stream(handler),
         )
+        wrap_function_wrapper(
+            "anthropic._response",
+            "ResponseContextManager.__exit__",
+            response_context_manager_exit,
+        )
+        wrap_function_wrapper(
+            "anthropic._response",
+            "AsyncResponseContextManager.__aexit__",
+            async_response_context_manager_exit,
+        )
 
         # parse() wraps create() internally in the Anthropic SDK and returns a
         # parsed message whose telemetry-relevant fields match Message, so the
@@ -174,6 +186,13 @@ def _uninstrument(self, **kwargs: Any) -> None:
             anthropic.resources.messages.AsyncMessages,  # pyright: ignore[reportAttributeAccessIssue,reportUnknownMemberType,reportUnknownArgumentType]
             "stream",
         )
+        from anthropic._response import (  # pylint: disable=import-outside-toplevel
+            AsyncResponseContextManager,
+            ResponseContextManager,
+        )
+
+        unwrap(ResponseContextManager, "__exit__")
+        unwrap(AsyncResponseContextManager, "__aexit__")
         if self._parse_supported:
             unwrap(
                 anthropic.resources.messages.Messages,  # pyright: ignore[reportAttributeAccessIssue,reportUnknownMemberType,reportUnknownArgumentType]

instrumentation/opentelemetry-instrumentation-genai-anthropic/src/opentelemetry/instrumentation/genai/anthropic/_raw_response.py

@@ -108,6 +108,15 @@ def __init__(
         self._self_stream_wrapper: Any = None
         self._install_hooks(raw_response)
 
+    def _fail(self, error: BaseException) -> None:
+        """Finalize this response with a caller-side failure, once."""
+        if self._self_stream_wrapper is not None:
+            self._self_stream_wrapper._finalize_failure(error)
+            return
+        if self._self_span_open:
+            self._self_span_open = False
+            self._self_invocation.fail(error)
+
     def _install_hooks(self, raw_response: Any) -> None:
         http_response = getattr(raw_response, "http_response", None)
         if http_response is None:

instrumentation/opentelemetry-instrumentation-genai-anthropic/src/opentelemetry/instrumentation/genai/anthropic/patch.py

@@ -49,6 +49,52 @@
 ANTHROPIC = "anthropic"
 
 
+def response_context_manager_exit(
+    wrapped: Callable[..., Any],
+    instance: Any,
+    args: tuple[Any, ...],
+    kwargs: dict[str, Any],
+) -> Any:
+    """Forward caller failures before Anthropic closes a raw response."""
+    error = args[1] if len(args) > 1 else kwargs.get("exc")
+    if isinstance(error, BaseException):
+        _fail_context_manager_response(instance, error)
+    return wrapped(*args, **kwargs)
+
+
+async def async_response_context_manager_exit(
+    wrapped: Callable[..., Any],
+    instance: Any,
+    args: tuple[Any, ...],
+    kwargs: dict[str, Any],
+) -> Any:
+    """Async counterpart to :func:`response_context_manager_exit`."""
+    error = args[1] if len(args) > 1 else kwargs.get("exc")
+    if isinstance(error, BaseException):
+        _fail_context_manager_response(instance, error)
+    return await wrapped(*args, **kwargs)
+
+
+def _fail_context_manager_response(
+    instance: Any, error: BaseException
+) -> None:
+    response = getattr(instance, "_ResponseContextManager__response", None)
+    if response is None:
+        response = getattr(
+            instance, "_AsyncResponseContextManager__response", None
+        )
+    fail = getattr(response, "_fail", None)
+    if not callable(fail):
+        return
+    try:
+        fail(error)
+    except Exception:  # pylint: disable=broad-exception-caught
+        _logger.debug(
+            "Failed to record an exception from a streaming response",
+            exc_info=True,
+        )
+
+
 def _is_raw_response(result: object) -> bool:
     """Whether ``result`` is a raw-response object to route through the proxy.
 

instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_async_messages.py

@@ -13,7 +13,7 @@
 except ImportError:
     import httpx as _http_lib
 from anthropic import APIConnectionError, AsyncAnthropic, NotFoundError
-from anthropic._response import AsyncAPIResponse
+from anthropic._response import AsyncAPIResponse, AsyncResponseContextManager
 
 try:
     from anthropic._legacy_response import LegacyAPIResponse
@@ -23,7 +23,10 @@
 from anthropic.resources.messages import AsyncMessages as _AsyncMessages
 from anthropic.types import Message
 
-from opentelemetry.instrumentation.genai.anthropic import _raw_response
+from opentelemetry.instrumentation.genai.anthropic import (
+    AnthropicInstrumentor,
+    _raw_response,
+)
 from opentelemetry.instrumentation.genai.anthropic._raw_response import (
     RawResponseProxy,
 )
@@ -1542,14 +1545,6 @@ async def test_async_messages_raw_response_parse_after_exit(
     assert message.model == model
 
 
-@pytest.mark.skip(
-    reason="Known gap, tracked in #389: the SDK's "
-    "AsyncResponseContextManager.__aexit__ discards the caller's exception "
-    "before closing the response, so the proxy never sees it and the span is "
-    "finalized as a success. Fixing it means instrumenting "
-    "AsyncMessagesWithStreamingResponse.create and wrapping the context "
-    "manager itself, which is a new patch target and out of scope here."
-)
 @pytest.mark.cassette("test_async_messages_create_streaming_with_raw_response")
 @pytest.mark.asyncio
 @pytest.mark.vcr()
@@ -1585,6 +1580,108 @@ async def test_async_messages_with_streaming_response_user_exception(
     assert span.attributes[ErrorAttributes.ERROR_TYPE] == "ValueError"
 
 
+@pytest.mark.cassette("test_async_messages_create_streaming_with_raw_response")
+@pytest.mark.asyncio
+@pytest.mark.vcr()
+async def test_async_messages_with_streaming_response_user_exception_before_parse(
+    span_exporter, async_anthropic_client, instrument_no_content
+):
+    """A caller error before parsing still fails the raw-response span once."""
+    with pytest.raises(ValueError, match="User raised exception"):
+        async with (
+            async_anthropic_client.messages.with_streaming_response.create(
+                model="claude-sonnet-4-20250514",
+                max_tokens=100,
+                messages=[
+                    {"role": "user", "content": "Say hello in one word."}
+                ],
+                stream=True,
+            )
+        ):
+            raise ValueError("User raised exception")
+
+    spans = span_exporter.get_finished_spans()
+    assert len(spans) == 1
+    assert spans[0].attributes[ErrorAttributes.ERROR_TYPE] == "ValueError"
+
+
+@pytest.mark.cassette("test_async_messages_create_streaming_with_raw_response")
+@pytest.mark.asyncio
+@pytest.mark.vcr()
+async def test_async_messages_with_streaming_response_user_exception_after_drain(
+    span_exporter, async_anthropic_client, instrument_no_content
+):
+    """A caller error after draining does not finalize the span twice."""
+    with pytest.raises(ValueError, match="User raised exception"):
+        async with (
+            async_anthropic_client.messages.with_streaming_response.create(
+                model="claude-sonnet-4-20250514",
+                max_tokens=100,
+                messages=[
+                    {"role": "user", "content": "Say hello in one word."}
+                ],
+                stream=True,
+            )
+        ) as raw_response:
+            async for _ in await raw_response.parse():
+                pass
+            raise ValueError("User raised exception")
+
+    spans = span_exporter.get_finished_spans()
+    assert len(spans) == 1
+    assert ErrorAttributes.ERROR_TYPE not in spans[0].attributes
+
+
+@pytest.mark.cassette("test_async_messages_create_with_raw_response")
+@pytest.mark.asyncio
+@pytest.mark.vcr()
+async def test_async_messages_with_streaming_response_nonstreaming_user_exception(
+    span_exporter, async_anthropic_client, instrument_no_content
+):
+    """A caller error in a non-streaming async response context fails the span."""
+    model = "claude-sonnet-4-20250514"
+
+    with pytest.raises(ValueError, match="User raised exception"):
+        async with (
+            async_anthropic_client.messages.with_streaming_response.create(
+                model=model,
+                max_tokens=100,
+                messages=[
+                    {"role": "user", "content": "Say hello in one word."}
+                ],
+            )
+        ):
+            raise ValueError("User raised exception")
+
+    spans = span_exporter.get_finished_spans()
+    assert len(spans) == 1
+    assert spans[0].attributes[ErrorAttributes.ERROR_TYPE] == "ValueError"
+
+
+@pytest.mark.cassette("test_async_messages_create_with_raw_response")
+@pytest.mark.asyncio
+@pytest.mark.vcr()
+async def test_async_messages_with_raw_response_caller_exception(
+    span_exporter, async_anthropic_client, instrument_no_content
+):
+    """A caller error while handling a raw response is separate from the call."""
+    raw_response = (
+        await async_anthropic_client.messages.with_raw_response.create(
+            model="claude-sonnet-4-20250514",
+            max_tokens=100,
+            messages=[{"role": "user", "content": "Say hello in one word."}],
+        )
+    )
+
+    with pytest.raises(ValueError, match="User raised exception"):
+        raise ValueError("User raised exception")
+
+    spans = span_exporter.get_finished_spans()
+    assert len(spans) == 1
+    assert ErrorAttributes.ERROR_TYPE not in spans[0].attributes
+    assert raw_response.headers is not None
+
+
 @pytest.mark.cassette("test_async_messages_create_api_error")
 @pytest.mark.asyncio
 @pytest.mark.vcr()
@@ -1608,6 +1705,52 @@ async def test_async_messages_with_raw_response_api_error(
     assert "NotFoundError" in span.attributes[ErrorAttributes.ERROR_TYPE]
 
 
+@pytest.mark.asyncio
+async def test_async_streaming_response_type_survives_instrumentation_round_trip(
+    tracer_provider, logger_provider, meter_provider
+):
+    """Async instrumentation round-trips keep the SDK manager type intact."""
+    instrumentor = AnthropicInstrumentor()
+    client = AsyncAnthropic()
+
+    def create_context_manager():
+        context_manager = client.messages.with_streaming_response.create(
+            model="claude-sonnet-4-20250514",
+            max_tokens=100,
+            messages=[{"role": "user", "content": "Hello"}],
+            stream=True,
+        )
+        request = getattr(context_manager, "_api_request", None)
+        if inspect.iscoroutine(request):
+            request.close()
+        return context_manager
+
+    instrumentor.instrument(
+        tracer_provider=tracer_provider,
+        logger_provider=logger_provider,
+        meter_provider=meter_provider,
+    )
+    try:
+        assert (
+            create_context_manager().__class__ is AsyncResponseContextManager
+        )
+        instrumentor.uninstrument()
+        assert (
+            create_context_manager().__class__ is AsyncResponseContextManager
+        )
+        instrumentor.instrument(
+            tracer_provider=tracer_provider,
+            logger_provider=logger_provider,
+            meter_provider=meter_provider,
+        )
+        assert (
+            create_context_manager().__class__ is AsyncResponseContextManager
+        )
+    finally:
+        instrumentor.uninstrument()
+        await client.close()
+
+
 @pytest.mark.cassette("test_async_messages_create_with_raw_response")
 @pytest.mark.asyncio
 @pytest.mark.vcr()
@@ -1634,12 +1777,17 @@ async def test_async_messages_streaming_response_is_transparent(
     span_exporter, async_anthropic_client, instrument_no_content
 ):
     """The streaming proxy must also keep the SDK's type and its parsed stream."""
-    async with async_anthropic_client.messages.with_streaming_response.create(
-        model="claude-sonnet-4-20250514",
-        max_tokens=100,
-        messages=[{"role": "user", "content": "Say hello in one word."}],
-        stream=True,
-    ) as raw_response:
+    context_manager = (
+        async_anthropic_client.messages.with_streaming_response.create(
+            model="claude-sonnet-4-20250514",
+            max_tokens=100,
+            messages=[{"role": "user", "content": "Say hello in one word."}],
+            stream=True,
+        )
+    )
+    assert isinstance(context_manager, AsyncResponseContextManager)
+    assert context_manager.__class__ is AsyncResponseContextManager
+    async with context_manager as raw_response:
         assert isinstance(raw_response, AsyncAPIResponse)
         assert raw_response.__class__ is AsyncAPIResponse
         stream = await raw_response.parse()