refactor(streaming)!: replace stream_with_chunking with single-task stream()

[ajbozarth]

Pull Request

Issue

Fixes #1440

Description

Replaces the two-task stream_with_chunking() + StreamChunkingResult with a single-task stream()/Streamer primitive consumed by a plain async for, and factors the inline chunk-boundary bookkeeping into a stateful Chunker. Builds on the POC in #1409.

stream() returns a Streamer driven by one async generator on the caller's task — there is no background orchestration task. It is consumed with async for, ideally inside async with, which provides a guaranteed cleanup contract (aclose()/__aexit__) that cancels an abandoned generation on an early break, an exception, or external cancellation (e.g. an asyncio.wait_for timeout) — the terminal events fire and the thunk is finalized on every path. Typed StreamEvents are emitted through the streaming_event plugin hook rather than a result.events() iterator, and acomplete() is removed.

Key changes:

• stream() / Streamer — single async generator on the caller's task; terminal state (failed_early, failure_reason, streaming_failures, full_text, final_validations, mot) lives on the thin Streamer handle.
• Chunker — new stateful driver that wraps a stateless ChunkingStrategy and holds only the pending fragment between deltas; delta-invariant (any slicing yields the same chunks as one split() over the whole text).
• Chunking strategy renames — SentenceChunker/WordChunker/ParagraphChunker → SentenceChunking/WordChunking/ParagraphChunking, freeing the Chunker name for the driver. String aliases (chunking="sentence", etc.) are unchanged.
• ModelOutputThunk iterator API — __aiter__/__anext__ wrap astream() in the async-iterator protocol with a single-consumer guard, plus aclose()/async with cleanup. astream() itself is unchanged.
• Telemetry — with the whole run on one task, the cross-task span-reattachment machinery and the STREAMING_ORCHESTRATION_START/_END hooks are removed; the span is renamed stream_with_chunking → stream; CompletedEvent moves to the streaming_event hook.
• Docs and examples — tutorials, how-to, tracing, and the streaming examples updated to the new API; added a multi-stream events example (multi_stream_events.py) and a raw MOT async-iterator example (async-iterator.py); added a migration guide at docs/dev/migrate-streaming-v0.8.md.

Breaking change, no deprecation shim: stream_with_chunking(), StreamChunkingResult, and the ...Chunker strategy names are removed. Migration guide: docs/dev/migrate-streaming-v0.8.md.

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)

Rewrote test/stdlib/test_streaming.py for the new API and added the Chunker delta-invariance suite to test/stdlib/test_chunking.py. Retargeted the telemetry span/metrics tests and the hook-call-site tests to the new topology (stream span roots the chat span; events via the streaming_event hook). Added the mocked integration twin for the streaming span-topology test so it is covered in the fast tier, not only behind the slow e2e path. Verified ruff, ruff format, and mypy clean; ran the tutorial example code against Ollama to confirm the documented sample output shapes.

Attribution

• AI coding assistants used

---

Adding a new component, requirement, sampling strategy, or tool?

• Component
• Requirement
• Sampling Strategy
• Tool

[AngeloDanducci]

Generally LGTM, a few review comments. I'll leave the larger design discussion to Jake since I think he's been more in the loop on the streaming refactor.

[planetf1]

Nice simplification overall — dropping the cross-task span-reattachment machinery in particular removes some genuinely hairy code. Four things below are worth fixing before merge; none call the core design into question, they're all narrow and fixable.

[planetf1]

Two more from a second pass, plus one I've folded into the existing thread on tracing_plugins.py:395 as a reply rather than a new comment since it's building on the same spot.

[jakelorocco]

Absent my other comments about the multi-chat streaming, I think it's unclear what the chunk actually contains here. Could you please add either an in docstring example or say that it returns string segments of the model output thunk?

[ajbozarth]

I added a clarifying sentence in c6a130a

[jakelorocco]

I did some tests with this multi stream events example, and I'm worried that it shows our current approach is actually difficult to utilize.

A few thoughts:

1. It is quite difficult to correlate a stream, the events, and the place those events need to go. For instance, I could get the stream id from the streamer and give it to some tui consumer. But I would have to have the plugin route to some queue / map that the tui consumer can grab events with that stream id from.
2. What is the preferred pattern if a user is creating and constantly streaming new mots? I feel like that pattern would either require multiple plugins that basically do the same thing (which might be a performance issue); and/or a complicated pattern to route the events to the correct consumer.
3. Is it possible to just have the streamer return events as well as mot chunks (or a different function of the streamer that can return the events) allowing the user to choose what to consume?

Additionally, I wonder if we could just register handlers directly to a streamer instead of having to go through the plugin system. Or maybe there's a way to utilize the plugin system to do something similar? After trying to utilize the plugin system to do some things beyond the example below, I'm just worried that it's quite convoluted to handle events from streamers.

[ajbozarth]

I actually went back and forth on how this should work a few items before settling on this implementation. I am open to further design discussions around it at or after Monday scrum.

If I'm understanding you correctly though this isn't an issue with the current implementation as it stands but the design we chose to implement. If that's the case we could choose to deal with this in a follow up PR before the next release. If you disagree and want to block merge of this PR on that design discussion we can do that too.

Even if we redesign, these hooks would need to stick around for the telemetry at least.

[jakelorocco]

Are we ensuring anywhere that the final mot == what was actually received in chunks from the model even if the chunker edits things during streaming?

I also think that if split drops whitespace, isn't it possible for .find to fail here, messing up _pending?

[ajbozarth]

I dug into this and part of it was already addressed in a fix suggested by @planetf1 above, I'll include Claudes explanation:

Good instinct — the mutation-via-find() case is now a hard error: Chunker.feed() raises ValueError if split() returns a chunk that isn't a verbatim substring of the buffered text, so a normalizing/rewriting strategy fails loudly instead of silently re-emitting. On the mot question: the final mot can't diverge from what the model produced, because mot.value is accumulated from the raw deltas directly and the chunker never writes back to it — chunking only shapes the consumer-facing async for stream. The only place chunk text feeds back into state is full_text on early exit (accumulated[:emitted_end], located by find()), and that's covered by the same substring guard.

[jakelorocco]

Should this get similar handling as mots to prevent multi-consumers?

[ajbozarth]

I actually asked this myself during my own self review, I'll let Claude explain:

Streamer doesn't need the MOT-style guard, because the failure mode it protects against can't occur here. MOT's __aiter__ returns self and re-arms, so a second async for would re-drive and split the stream — hence its explicit guard. Streamer's __aiter__ returns the same async generator object (self._gen), which is single-consumer by construction: once the first async for exhausts it, a second one just gets StopAsyncIteration immediately and yields nothing — no split, no re-run, full_text/mot intact (verified). So a second iteration is inert rather than dangerous. It's silent rather than loud (you get [], not an error); happy to add an explicit raise if you'd prefer the louder contract, but there's no correctness risk either way.

[ajbozarth]

@AngeloDanducci @planetf1 @jakelorocco I believe I had addressed all your review (except the open design point @jakelorocco raise) and left response on every comment. If you could re-review and mark them as resolved.

[jakelorocco]

lgtm; a few minor issues and discussed verbally about next steps

[jakelorocco]

Should we also enforce that the chunker always advances? or if not, that it doesn't produce empty chunks? This doesn't seem to impact our chunking strategies but maybe we should include this check to ensure third party chunkers don't fail in these modes?

[ajbozarth]

added another check for empty strings in 9d9f5ee which together with the existing check catch not advancing

[planetf1]

Good catch — added in 2663b15. feed() now raises ValueError before the find() loop when split() returns an empty string. Without it, find("", cursor) always returns 0, so _pending wouldn't advance and the empty string would be silently emitted as a chunk. The verbatim-substring guard catches mutation but not zero-length — they need separate checks.

The always-advances property follows from the two guards together: a non-empty verbatim-substring chunk strictly advances the cursor. Added test_chunker_rejects_empty_chunk alongside the existing test_chunker_rejects_mutating_strategy.

[jakelorocco]

I think this marker should be elsewhere / deleted, not on the class?

[ajbozarth]

removed in 9d9f5ee

[planetf1]

_FailOnSecondReq is a Requirement subclass used as a test fixture, not a test class. The marker had no effect and was misleading. Removed in 2663b15.

[jakelorocco]

Can you please add detail here that completed_normally doesn't prevent the final validation from raising an exception, etc...?

[ajbozarth]

addressed in 9d9f5ee

[ajbozarth]

@jakelorocco and @planetf1 I responded to or addressed your latest review. If you could check one more time.

In addition I am investigating the events API after our design discussion during scrum and intend to either open a follow up issue or PR to address it rather than include it in this PR. I'll open that Issue or PR by EOD once I have it properly scoped

[ajbozarth]

Per the design questions in #1543 (comment) and the discuss we had today in scrum I investigated better ways to consume the stream using events rather than output.

My initial attempt was to extend stream()/Streamer to take a flag that changes what it iterates on (events/output), but that ran into issues handling the error path. So I pivoted to an idea I had about wrapping the stream in another iterator that just drives the normal stream in the background. I had previously discarded this idea as convoluted, but decided to give it another try and ended up with the draft POC in ajbozarth#3 (opened in my fork so it diffs against this branch)

Feel free to review that draft and when I return from my trip next week I'll pick it back up and finish it based on feedback.

[ajbozarth]

@planetf1 @jakelorocco @AngeloDanducci unless I hear otherwise I intend to add this to the merge queue after from lunch (in ~2hr)