Anthropic AsyncMessageStream.get_final_message()/get_final_text() bypass TracedAsyncStream, so the span never finishes

[AronsonDan]

Title

Anthropic AsyncMessageStream.get_final_message()/get_final_text() bypass TracedAsyncStream, so the span never finishes

Description

When consuming an Anthropic streamed response via the SDK's own convenience
methods (stream.get_final_message() / stream.get_final_text()) instead
of manual async for iteration, the APM span created for the .stream()
call is never finished, and therefore never exported — to plain APM or
LLM Observability. No exception is raised and no error is logged anywhere;
the span is simply left open indefinitely.

Reproduction

import anthropic
import ddtrace  # or run under ddtrace-run / with DD_LLMOBS_ENABLED=true

client = anthropic.AsyncAnthropic()

async def generate():
    async with client.messages.stream(
        model="claude-opus-4-8",
        max_tokens=1024,
        messages=[{"role": "user", "content": "hello"}],
    ) as stream:
        message = await stream.get_final_message()   # <-- span never finishes
        return message

# No span shows up in APM traces or LLM Observability for this call,
# despite the request completing successfully with a 200 from the API.

Changing the body to manually iterate first works around it:

    async with client.messages.stream(...) as stream:
        async for _ in stream:   # <-- drives the span-finishing hook
            pass
        message = await stream.get_final_message()   # now just reads accumulated state
        return message

Root cause

ddtrace/contrib/internal/anthropic/patch.py wraps Messages.stream() /
AsyncMessages.stream() and, for streaming responses, hands the result to
handle_streamed_response() in ddtrace/contrib/internal/anthropic/_streaming.py,
which builds a TracedAsyncStream (from
ddtrace/llmobs/_integrations/base_stream_handler.py) wrapping the real
stream object via wrapt.ObjectProxy.

finalize_stream() — which calls ctx.dispatch_ended_event(), the only
thing that actually finishes the span — is called from exactly one place:

# TracedAsyncStream.__anext__
async def __anext__(self):
    self._ensure_started()
    try:
        chunk = await self._self_async_stream_iter.__anext__()
        await self._self_handler.process_chunk(chunk, self._self_async_stream_iter)
        return chunk
    except StopAsyncIteration:
        self._self_handler.finalize_stream()   # <-- only call site
        raise
    except Exception as e:
        self._self_handler.handle_exception(e)
        self._self_handler.finalize_stream(e)
        raise

Neither __aenter__ nor __aexit__ call it:

async def __aexit__(self, exc_type, exc_val, exc_tb):
    return await self.__wrapped__.__aexit__(exc_type, exc_val, exc_tb)

AsyncMessageStream.get_final_message() (Anthropic SDK,
anthropic/lib/streaming/_messages.py) is:

async def get_final_message(self) -> ...:
    await self.until_done()
    ...

get_final_message/until_done are methods defined on the original
AsyncMessageStream class, not on TracedAsyncStream. Because
TracedAsyncStream is a wrapt.ObjectProxy, stream.get_final_message()
resolves via __getattr__ to the unwrapped object's bound method — self
inside that method is the original stream, not the proxy — so
until_done()'s internal consumption never goes through
TracedAsyncStream.__anext__, and finalize_stream() is never called.

Why this matters

get_final_message()/get_final_text() are first-class, documented
Anthropic SDK methods for exactly this use case ("I don't need to process
individual events, just give me the final result") — this isn't an obscure
or unsupported usage pattern. Anyone using them with the streaming API loses
tracing entirely and has no indication anything is wrong.

Related issues (same class of bug, fixed individually per-integration)

• Span will not finish with langchain_aws.ChatBedrock with streaming=True #11295 — Bedrock via LangChain: langchain-aws calls close() on the
  generator, raising GeneratorExit, which isn't caught by
  except Exception. Fixed by moving cleanup into a finally block
  (Bedrock/botocore-specific file).
• fix(llmobs): eagerly finalize stream to prevent nested agent spans #18247 — Claude Agent SDK: async generators closed lazily left the first
  span open across sequential queries. Fixed by eagerly finalizing on a
  specific chunk type (Claude Agent SDK–specific file).
• dd-trace-js#7633 — Anthropic (Node): a related but distinct symptom
  (spans export, but with empty input/output) from the same underlying
  cause — BetaMessageStream consumes events through an internal queue
  before the Symbol.asyncIterator wrapper can intercept them.

All three are instances of the same architectural gap: the streaming
instrumentation only observes consumption through its own proxy's iterator
protocol, and any SDK-internal or convenience-method consumption path
bypasses it silently. Given how often this recurs, a generic fix in
base_stream_handler.py (e.g. having __aexit__/__exit__ also call
finalize_stream() if it hasn't fired yet) seems more robust than
continuing to patch each integration individually as it's discovered.

Environment

• ddtrace 4.11.1 (latest on PyPI as of this report)
• anthropic >=0.40.0
• Confirmed still present on the current dd-trace-py main branch
  (TracedAsyncStream.__aexit__/TracedStream.__exit__ unchanged)
• Python 3.11