fix(observe): close abandoned generators when their span already ended

[DavidTraina]

What does this PR do?

Once a wrapped generator's span has ended, aclose()/close() return without closing the generator. But a span can end while the generator is still suspended mid-stream — for example, a client disconnect that cancels iteration before the generator resumes. The generator is then never closed: the GC finalizes it later, running its finally block under whatever request context happens to be active at that moment.

In our production streaming service this fired many times a day: "Setting attribute on ended span" warnings from finally-block span updates, running under other requests' contexts.

Fix: always close the generator, even if the span already ended. Span behavior is unchanged. Also narrows a pre-existing except TypeError that could mistake an error raised by generator cleanup for the Python 3.10 create_task signature fallback and silently swallow it.

Alternative: #1792 builds on this and also preserves the generator's final span updates in the cancellation case. Pick whichever you prefer; happy to close the other.

Minimal repro (fails before, passes after)

import asyncio, contextvars, gc
from langfuse import observe

request_ctx = contextvars.ContextVar("request", default="none")

@observe(name="stream")
async def stream():
    try:
        yield "a"
        yield "b"
    finally:
        print(f"finally ran under {request_ctx.get()!r}")

async def main():
    request_ctx.set("request-A")
    gen = stream()

    async def consume():
        async for _ in gen:
            await asyncio.sleep(0)

    task = asyncio.create_task(consume())
    for _ in range(4):
        await asyncio.sleep(0)
    asyncio.get_running_loop().call_soon(task.cancel)  # lands before the inner task's first step
    try:
        await task
    except asyncio.CancelledError:
        pass

    await gen.aclose()            # was a no-op: span already ended
    request_ctx.set("request-B")
    del gen; gc.collect(); await asyncio.sleep(0.05)
    # before this fix: prints "finally ran under 'request-B'"

asyncio.run(main())

Type of change

• Bug fix
• New feature
• Breaking change
• Refactor
• Documentation update
• Tooling, CI, or repo maintenance

Verification

uv run --frozen pytest -n auto --dist worksteal tests/unit   # also run on 3.10 and 3.11
uv run --frozen ruff check .
uv run --frozen ruff format --check .
uv run --frozen mypy langfuse --no-error-summary

Checklist

• I self-reviewed the diff using code_review.md.
• I added or updated tests for behavior changes.
• I updated docs, examples, or .env.template if needed.
• I did not hand-edit generated files; if generated files changed, I used the upstream regeneration path.
• I did not commit secrets or credentials.

[claude]

Claude Code Review

This pull request is from a fork — automated review is disabled. A repository maintainer can comment @claude review to run a one-time review.