feat(streaming): add EventStreamer for iterating stream events

[ajbozarth]

⚠️ Design POC — not finished, not for merge. This is a draft to evaluate the approach for consuming streaming events, following the discussion on #1543. Tests, typing checks, and docs are intentionally deferred (see below). Please review the design, not the completeness.

Issue

Follow-up to the design discussion on #1543 (now merged) — streaming event consumption ergonomics. No separate tracking issue yet — flagging for a maintainer to route.

Description

Adds an opt-in EventStreamer, returned by stream(..., as_events=True), that lets you iterate a stream's typed StreamEvent objects directly with async for — rather than the only route today, a process-global STREAMING_EVENT plugin hook demuxed on streaming_id.

It wraps a Streamer, drives it to completion on a background task, and delivers events through a queue. Because the queue's transport (put_nowait) is legal from teardown and during exception unwinding, the terminal CompletedEvent/ErrorEvent are delivered on every exit path — including an early break — which the alternative of yielding events from the driver generator cannot do (yield is illegal during GeneratorExit).

Additive and opt-in. The default stream() chunk-iterating contract is unchanged. Implementation: an optional per-stream event_queue is threaded through _emit_event; the STREAMING_EVENT hook still fires on all paths (telemetry unaffected). stream() is split into _stream() + a dispatching wrapper with typed @overloads on as_events.

Testing

• Tests added to the respective file if code was changed
• New code has 100% coverage if code was added
• Ensure existing tests and github automation passes (a maintainer will kick off the github automation when the rest of the PR is populated)

Deferred — this is a design-review draft. Not yet done: unit tests for EventStreamer, a test/typing/check_*.py for the overloads, a docs example, and two known edge cases (setup-failure event symmetry; external cancellation during setup). Verified locally: ruff format/check clean, mypy clean, existing test_streaming.py passes (37), and an offline harness exercises all four event-mode paths (natural completion, requirement-fail, error re-raise, early break with no task leak).

Attribution

• AI coding assistants used

---

• Component
• Requirement
• Sampling Strategy
• Tool

[ajbozarth]

@jakelorocco here's my initial draft POC for a events iterator for streaming. This is the better of the two POCs I wrote and I'll leave this here for your review while I'm out. Once I get back next week we can sync and I'll continue work on this

[jakelorocco]

This pretty much looks like exactly what I was looking for; thank you!

[jakelorocco]

Since we are awaiting the underlying _stream function, the stream still progresses even when not directly awaiting on this event streamer, right?

[jakelorocco]

Is it not an option to do the as_events boolean on the streamer itself and remove the need for discarding empty chunks and creating a new queue?