fix(anthropic): preserve caller errors in 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_messages_with_streaming_response_init,
     messages_create,
     messages_stream,
+    messages_with_streaming_response_init,
 )
 
 
@@ -134,6 +136,16 @@ def _instrument(self, **kwargs: Any) -> None:
             "AsyncMessages.stream",
             async_messages_stream(handler),
         )
+        wrap_function_wrapper(
+            "anthropic.resources.messages",
+            "MessagesWithStreamingResponse.__init__",
+            messages_with_streaming_response_init,
+        )
+        wrap_function_wrapper(
+            "anthropic.resources.messages",
+            "AsyncMessagesWithStreamingResponse.__init__",
+            async_messages_with_streaming_response_init,
+        )
 
         # parse() wraps create() internally in the Anthropic SDK and returns a
         # parsed message whose telemetry-relevant fields match Message, so the
@@ -174,6 +186,14 @@ def _uninstrument(self, **kwargs: Any) -> None:
             anthropic.resources.messages.AsyncMessages,  # pyright: ignore[reportAttributeAccessIssue,reportUnknownMemberType,reportUnknownArgumentType]
             "stream",
         )
+        unwrap(
+            anthropic.resources.messages.MessagesWithStreamingResponse,  # pyright: ignore[reportAttributeAccessIssue,reportUnknownMemberType,reportUnknownArgumentType]
+            "__init__",
+        )
+        unwrap(
+            anthropic.resources.messages.AsyncMessagesWithStreamingResponse,  # pyright: ignore[reportAttributeAccessIssue,reportUnknownMemberType,reportUnknownArgumentType]
+            "__init__",
+        )
         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

@@ -105,6 +105,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._fail(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

@@ -5,6 +5,7 @@
 
 from __future__ import annotations
 
+import functools
 import logging
 from collections.abc import Awaitable, Callable
 from typing import TYPE_CHECKING, Any, cast
@@ -27,8 +28,10 @@
 )
 from .utils import is_anthropic_async_stream, is_anthropic_stream
 from .wrappers import (
+    AsyncMessagesStreamingResponseContextManagerWrapper,
     AsyncMessagesStreamManagerWrapper,
     AsyncMessagesStreamWrapper,
+    MessagesStreamingResponseContextManagerWrapper,
     MessagesStreamManagerWrapper,
     MessagesStreamWrapper,
     MessageWrapper,
@@ -49,6 +52,58 @@
 ANTHROPIC = "anthropic"
 
 
+def _wrap_streaming_response_create(
+    create: Callable[..., Any],
+) -> Callable[..., Any]:
+    """Wrap an SDK ``with_streaming_response.create`` bound method."""
+
+    @functools.wraps(create)
+    def traced_create(*args: Any, **kwargs: Any) -> Any:
+        return MessagesStreamingResponseContextManagerWrapper(
+            create(*args, **kwargs)
+        )
+
+    return traced_create
+
+
+def _wrap_async_streaming_response_create(
+    create: Callable[..., Any],
+) -> Callable[..., Any]:
+    """Wrap an async SDK ``with_streaming_response.create`` bound method."""
+
+    @functools.wraps(create)
+    def traced_create(*args: Any, **kwargs: Any) -> Any:
+        return AsyncMessagesStreamingResponseContextManagerWrapper(
+            create(*args, **kwargs)
+        )
+
+    return traced_create
+
+
+def messages_with_streaming_response_init(
+    wrapped: Callable[..., Any],
+    instance: Any,
+    args: tuple[Any, ...],
+    kwargs: dict[str, Any],
+) -> Any:
+    """Instrument the dynamically-created sync streaming ``create`` method."""
+    result = wrapped(*args, **kwargs)
+    instance.create = _wrap_streaming_response_create(instance.create)
+    return result
+
+
+def async_messages_with_streaming_response_init(
+    wrapped: Callable[..., Any],
+    instance: Any,
+    args: tuple[Any, ...],
+    kwargs: dict[str, Any],
+) -> Any:
+    """Instrument the dynamically-created async streaming ``create`` method."""
+    result = wrapped(*args, **kwargs)
+    instance.create = _wrap_async_streaming_response_create(instance.create)
+    return result
+
+
 def _is_raw_response(result: object) -> bool:
     """Whether ``result`` is a raw-response object to route through the proxy.
 

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

@@ -49,6 +49,7 @@
 
 
 ResponseT = TypeVar("ResponseT")
+ResponseT_co = TypeVar("ResponseT_co", covariant=True)
 ResponseFormatT = TypeVar("ResponseFormatT")
 accumulate_event = cast("Callable[..., Message] | None", _sdk_accumulate_event)
 
@@ -58,6 +59,28 @@ class _StreamWrapperWithStream(Protocol):
     def stream(self) -> object: ...
 
 
+class _SyncResponseContextManager(Protocol[ResponseT_co]):
+    def __enter__(self) -> ResponseT_co: ...
+
+    def __exit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc_val: BaseException | None,
+        exc_tb: TracebackType | None,
+    ) -> Any: ...
+
+
+class _AsyncResponseContextManager(Protocol[ResponseT_co]):
+    async def __aenter__(self) -> ResponseT_co: ...
+
+    async def __aexit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc_val: BaseException | None,
+        exc_tb: TracebackType | None,
+    ) -> Any: ...
+
+
 def _set_response_attributes(
     invocation: InferenceInvocation,
     result: Message | None,
@@ -96,6 +119,68 @@ def __getattr__(self, name: str):
         return getattr(self._response, name)
 
 
+class MessagesStreamingResponseContextManagerWrapper(Generic[ResponseT]):
+    """Preserve caller exceptions across Anthropic's response context manager.
+
+    ``ResponseContextManager.__exit__`` closes the response without forwarding
+    the exception raised inside the ``with`` block.  The raw-response proxy
+    therefore only sees a normal close and records a successful invocation.
+    Give the proxy the exception before delegating to the SDK manager so its
+    normal close hook remains responsible for cleanup without changing the
+    exception visible to the caller.
+    """
+
+    def __init__(self, manager: _SyncResponseContextManager[ResponseT]):
+        self._manager = manager
+        self._response: ResponseT | None = None
+
+    def __enter__(self) -> ResponseT:
+        self._response = self._manager.__enter__()
+        return self._response
+
+    def __exit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc_val: BaseException | None,
+        exc_tb: TracebackType | None,
+    ) -> Any:
+        if exc_val is not None:
+            fail = getattr(self._response, "_fail", None)
+            if fail is not None:
+                fail(exc_val)
+        return self._manager.__exit__(exc_type, exc_val, exc_tb)
+
+    def __getattr__(self, name: str) -> Any:
+        return getattr(self._manager, name)
+
+
+class AsyncMessagesStreamingResponseContextManagerWrapper(Generic[ResponseT]):
+    """Async counterpart of ``MessagesStreamingResponseContextManagerWrapper``."""
+
+    def __init__(self, manager: _AsyncResponseContextManager[ResponseT]):
+        self._manager = manager
+        self._response: ResponseT | None = None
+
+    async def __aenter__(self) -> ResponseT:
+        self._response = await self._manager.__aenter__()
+        return self._response
+
+    async def __aexit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc_val: BaseException | None,
+        exc_tb: TracebackType | None,
+    ) -> Any:
+        if exc_val is not None:
+            fail = getattr(self._response, "_fail", None)
+            if fail is not None:
+                fail(exc_val)
+        return await self._manager.__aexit__(exc_type, exc_val, exc_tb)
+
+    def __getattr__(self, name: str) -> Any:
+        return getattr(self._manager, name)
+
+
 class MessageWrapper:
     """Wrapper for non-streaming Message response that handles telemetry."""
 

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

@@ -1523,14 +1523,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()

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

@@ -1773,14 +1773,6 @@ def boom(*args, **kwargs):
     assert len(span_exporter.get_finished_spans()) == 1
 
 
-@pytest.mark.skip(
-    reason="Known gap, tracked in #389: the SDK's "
-    "ResponseContextManager.__exit__ 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 "
-    "MessagesWithStreamingResponse.create and wrapping the context manager "
-    "itself, which is a new patch target and out of scope here."
-)
 @pytest.mark.vcr()
 @pytest.mark.cassette("test_sync_messages_create_streaming_with_raw_response")
 def test_sync_messages_with_streaming_response_user_exception(