util-genai: streamed invocations leak the span and the OTel context when the caller walks away

A stream the caller never drains and never closes leaks both the span and the OTel context.

Every stream wrapper derives from SyncStreamWrapper / AsyncStreamWrapper, so this is not package specific: openai, anthropic, portkey, google-genai, smolagents, qwen-agent.

Steps to Reproduce

import gc

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.util.genai.handler import TelemetryHandler
from opentelemetry.util.genai.stream import SyncStreamWrapper

exporter = InMemorySpanExporter()
provider = TracerProvider()
provider.add_span_processor(SimpleSpanProcessor(exporter))
handler = TelemetryHandler(tracer_provider=provider)
tracer = provider.get_tracer("test")


class Wrapper(SyncStreamWrapper):
    def __init__(self, stream, invocation):
        super().__init__(stream)
        self._self_inv = invocation

    def _process_chunk(self, chunk):
        pass

    def _on_stream_end(self):
        self._self_inv.stop()

    def _on_stream_error(self, error):
        self._self_inv.fail(error)


def chunks():
    yield "a"
    yield "b"


def abandon():
    inv = handler.invoke_local_agent(agent_name="A", request_model="m")
    for _ in Wrapper(chunks(), inv):
        break


abandon()
gc.collect()

with tracer.start_as_current_span("later"):
    pass

spans = {s.name: s for s in exporter.get_finished_spans()}
print("agent span ended:", "invoke_agent A" in spans)
print("later.parent:", spans["later"].parent)

Expected Result

agent span ended: True
later.parent: None

Actual Result

agent span ended: False
later.parent: SpanContext(trace_id=0x..., span_id=0x..., ...)

later.parent is the worse half: every span on that thread from then on reparents under an invocation that never ended.

Additional context

Needs investigation, the constraints pull against each other. Whatever lands should satisfy:

• An abandoned stream ends its span, with the stream's duration, not the collector's.
• An abandoned stream leaves nothing attached to the caller's context.
• Spans created while the stream is driven still nest under the invocation. Agent-run generators do the work on each next(), so a model call on step 3 lands long after the generator was handed over.
• Work the library moves off the calling thread keeps its parent. TransformersModel.generate_stream generates on a thread, CodeAgent's local executor runs code in a worker. Anything scoped to a caller-controlled frame misses these.
• Sync and async match. #391 documents the async path already losing parentage.

Only anthropic, openai-agents and smolagents have abandoned-stream tests. qwen-agent, portkey, google-genai and openai's plain streams have none.

Start with SyncStreamWrapper and AsyncStreamWrapper, then inspect TelemetryHandler and the _on_stream_end/_on_stream_error hooks shown in the reproduction. Compare the abandoned-stream tests for anthropic, openai-agents, and smolagents with the missing coverage for qwen-agent, portkey, google-genai, and plain OpenAI streams. Done means abandoned sync and async streams end spans with stream duration, detach caller context, and preserve parentage for work during iteration and on worker threads.