fix: raise RuntimeError on second ctx.stream() consumption

[Copilot]

Calling ctx.stream() more than once silently deadlocks — the [DONE] sentinel is consumed on the first iteration, leaving the queue permanently empty and the second consumer blocked forever.

Changes

• StreamContext.__init__ — adds _streamed: bool = False flag
• StreamContext.stream() — raises RuntimeError immediately on re-entry before touching the queue; sets _streamed = True on first call

# Before: hangs forever
async for _ in ctx.stream(): pass  # drains OK
async for _ in ctx.stream(): pass  # blocks indefinitely

# After: fails fast
async for _ in ctx.stream(): pass  # drains OK
async for _ in ctx.stream(): pass  # RuntimeError: ctx.stream() can only be consumed once

• TestEdgeCases — two new tests: one asserting RuntimeError is raised on the second call, one confirming the error surfaces on __anext__() before any queue wait

[vercel]

The latest updates on your projects. Learn more about Vercel for GitHub.

Project	Deployment	Actions	Updated (UTC)
ai-sdk-stream-python	Ready	Preview, Comment, Open in v0	May 30, 2026 7:47pm

[shloimy-wiesel]

@claude review this pr

[claude]

Claude finished @shloimy-wiesel's task in 1m 58s —— View job

---

PR Review: fix: raise RuntimeError on second ctx.stream() consumption

• Gather context and understand the request
• Read changed files (context.py, tests/test_context.py)
• Run diff and analyze changes
• Post review feedback

---

Summary

This PR fixes a real bug: double-consuming ctx.stream() previously deadlocked silently because the None sentinel was consumed on the first pass, leaving the second consumer blocked on an empty queue forever. The fix is correct and well-targeted.

What's Good

Implementation (context.py:772–777): The _streamed flag + early RuntimeError is the right approach. Since asyncio is single-threaded, there's no TOCTOU race between the check and the flag set — no await separates them, so no other coroutine can interleave.

Tests (test_context.py:611–639): Two tests, each covering a distinct behavior:

1. test_stream_raises_on_second_consumption — verifies the error is raised when iterating a second time
2. test_stream_raises_before_iteration_on_second_call — verifies the error surfaces on __anext__(), not when the queue would block

This distinction matters: because stream() is an async def with yield, calling ctx.stream() returns the generator object without executing any code. The check only runs on the first __anext__(). The second test explicitly documents this Python async generator behavior, which is the right call.

Version bump: pyproject.toml → 0.4.0, which is reflected in uv.lock. Appropriate for a behavioral change.

---

One thing to flag

The collect_stream helper (used in both new tests) breaks on [DONE] without calling aclose() on the generator:

# tests/test_context.py:29–38
async def collect_stream(ctx: StreamContext) -> list[dict]:
    async for chunk in ctx.stream():
        if chunk.strip() == "data: [DONE]":
            break   # generator abandoned here, aclose() deferred to GC
        ...

This is pre-existing behavior, not introduced by this PR, and it doesn't cause any correctness issues here — _streamed = True is already set on the first __anext__() so the second call correctly raises. But if stream() ever grows a try/finally cleanup block, this pattern would defer that cleanup non-deterministically.

Not a blocker for this PR, but worth a follow-up async with contextlib.aclosing(ctx.stream()) as gen: refactor in the test helper.

---

Verdict

Approve. The fix is minimal, correct, and well-tested. No issues blocking merge.