🐛 Bug Report: Anthropic instrumentation: streaming spans are never ended on early exit (with-block, messages.stream() helper, close) #4536

import asyncio, gc, json, anthropic
try:
    import httpx2 as httpx         
except ImportError:
    import httpx
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
from opentelemetry.instrumentation.anthropic import AnthropicInstrumentor

def ev(name, data): return f"event: {name}\ndata: {json.dumps(data)}\n\n"
SSE = (ev("message_start", {"type": "message_start", "message": {"id": "msg_1", "type": "message",
        "role": "assistant", "model": "m", "content": [], "stop_reason": None, "stop_sequence": None,
        "usage": {"input_tokens": 3, "output_tokens": 0}}})
     + ev("content_block_start", {"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}})
     + "".join(ev("content_block_delta", {"type": "content_block_delta", "index": 0,
                  "delta": {"type": "text_delta", "text": p}}) for p in ["Hel", "lo", "!"])
     + ev("content_block_stop", {"type": "content_block_stop", "index": 0})
     + ev("message_delta", {"type": "message_delta", "delta": {"stop_reason": "end_turn", "stop_sequence": None},
                            "usage": {"output_tokens": 3}})
     + ev("message_stop", {"type": "message_stop"})).encode()
handler = lambda r: httpx.Response(200, headers={"content-type": "text/event-stream"}, content=SSE)

exporter = InMemorySpanExporter()
provider = TracerProvider(); provider.add_span_processor(SimpleSpanProcessor(exporter))
AnthropicInstrumentor().instrument(tracer_provider=provider)
KW = dict(model="m", max_tokens=10, messages=[{"role": "user", "content": "hi"}])
client = anthropic.Anthropic(api_key="x", base_url="http://mock",
                             http_client=httpx.Client(transport=httpx.MockTransport(handler)))

def spans(): gc.collect(); return len(exporter.get_finished_spans())

# 1. Recommended helper, stop after the first text chunk
exporter.clear()
with client.messages.stream(**KW) as s:
    for text in s.text_stream:
        break
del s
print("messages.stream() early exit:", spans())   # expected 1, actual 0

# 2. Plain stream in a with-block, stop after the first event
exporter.clear()
with client.messages.create(stream=True, **KW) as s:
    next(iter(s))
del s
print("with-block early exit:      ", spans())   # expected 1, actual 0

# 3. Async with-block
async def main():
    aclient = anthropic.AsyncAnthropic(api_key="x", base_url="http://mock",
                                       http_client=httpx.AsyncClient(transport=httpx.MockTransport(handler)))
    exporter.clear()
    async with await aclient.messages.create(stream=True, **KW) as s:
        await s.__aiter__().__anext__()
    del s
    print("async with-block early exit:", spans())  # expected 1, actual 0
asyncio.run(main())