Repository navigation
Description
Replace the two-task stream_with_chunking() + StreamChunkingResult with a single-task stream()/Streamer primitive consumed by a plain async for, and factor the inline chunk-boundary bookkeeping into a Chunker object. Chunking stays shared across validators in this issue; per-validator chunking and chunk-state ownership are a follow-up (#1441).
Builds on the POC in PR #1409.
Scope
Already prototyped (POC in PR #1409)
stream()/Streamer— single generator on the caller's task, no background orchestration task. Durable state is the thinStreamerhandle (failed_early,failure_reason,streaming_failures,full_text,final_validations, terminalmot).ModelOutputThunk.__aiter__/__anext__— wrapsastream()in the async-iterator protocol so plain and validated streaming share one loop.- Event emission via
STREAMING_EVENThook — full Phase 1 vocabulary (QuickCheckEvent,ChunkEvent,StreamingDoneEvent,FullValidationEvent,ErrorEvent,CompletedEvent) fired uniformly on the single task.
New in this issue
Chunker— see design below. Factors the chunk-boundary state currently inline in the orchestrator into a reusable object.- Rename the Phase 1 chunking strategies —
SentenceChunker→SentenceChunking,WordChunker→WordChunking,ParagraphChunker→ParagraphChunking(theChunkingStrategyABC keeps its name). This freesChunkerfor the new stateful object above. Rationale: feat(stdlib): add ChunkingStrategy ABC and built-in chunkers #899 specified these as...Chunking; they shipped as...Chunkerin feat(stdlib): add ChunkingStrategy ABC and built-in chunkers #923 with no recorded rationale (silent implementation choice)....Chunker(an agent noun) reads as "the thing that chunks," which fits the new stateful driver better than a stateless strategy — so the strategies are restored to their...Chunkingspec andChunkernames the driver. Note: these names are exported (__all__) and used acrossdocs/examples/streaming/, tests, and docs — this is a breaking rename of public names, with all call sites updated (consistent with this issue's hard break, below). __aiter__single-consumer guard —__aiter__/__anext__make MOT look like a normal async iterator, but a second consumer splits the same stream rather than starting an independent one. Add a real guard: mark the thunk consumed on first iteration, raise if a second consumer starts. (Full parallel-consumer support — buffering, per-consumer cursors, late-joiner semantics — is out of scope unless a concrete need appears; the guard is enough for now.)- Events example — an example demonstrating an events plugin that consumes
STREAMING_EVENTand surfaces streaming events for multiple parallel streams. The plugin may live in the example or insrcdepending on implementation. This is the primary user-facing deliverable for event consumption (hooks are the interface; the example is how users learn to consume them). - Hard break, no deprecation shim —
stream()replacesstream_with_chunking()outright. A shim would keep the old orchestrator alive alongside the new one, reintroducing the complexity this issue removes; the API is new enough to break directly.
Constraints
- Plain strings only.
stream()yieldsstrchunks. The typed-chunk / multi-modal envelope belongs tostream_parsed_repr(feat(core): stream_parsed_repr — streaming parsed representation on the MOT #1442); mixing it in here means changing the same surface twice.
This issue's chunking scope
Exactly one Chunker on the Streamer, fed by the single chunking= param; all validators validate the same chunks (as today). Per-validator chunking — each Requirement carrying its own strategy, and the resulting chunk-state ownership question — is #1441. The Chunker carries its own small state, so that follow-up relocates/multiplies it without an internal reshape.
Chunker class
Factors the chunk-boundary bookkeeping that currently lives inline in _orchestrate_streaming / _drive (accumulated scanning, prev_chunk_count, withheld-fragment handling) into a dedicated object. It is the stateful counterpart to a ChunkingStrategy: the strategy is stateless "how to split," the Chunker holds "how far through this stream we are."
Shape
__init__(strategy: ChunkingStrategy)— holds the (stateless) strategy.feed(delta: str) -> list[str]— takes one stream delta, returns any new complete chunks, holding the trailing fragment between calls.- A flush/finalize method for the trailing fragment at stream end.
State division
| State | Owner | Notes |
|---|---|---|
accumulated (full raw text) |
Streamer |
one raw stream even when N validators later; needed for full_text regardless |
| pending fragment (text since last boundary) | Chunker |
small (≤ one chunk); all the Chunker needs |
full_text (validated-emitted output) |
Streamer |
validation concern, not chunking |
The Chunker holds only the pending fragment, not the full accumulated text. The Streamer keeps accumulated for its own full_text needs and feeds deltas to the Chunker.
Implementation approach — reuse split(), don't reimplement boundaries
Chunker.feed() should call the existing stateless ChunkingStrategy.split() on the pending fragment, not reimplement boundary detection:
feed(delta):
self._pending += delta
chunks = self._strategy.split(self._pending)
self._pending = <tail after last returned chunk>
return chunks
split() already withholds the trailing fragment correctly, so feeding it the pending fragment reuses the exact boundary logic that already works. The only new logic is computing the tail carried forward.
Behavioral-change risk: chunking on the delta, not full accumulated
Today split() is called on the full accumulated every delta (O(n) re-scan). Moving to incremental delta input is a real behavioral change, not a pure refactor. Risk is confined to boundaries that span the delta seam:
- Sentence — boundary is punctuation + whitespace; the
.and the following space can arrive in different deltas. - Word — trailing partial word split across deltas.
- Paragraph —
\n{2,}is greedy; the newlines can split across deltas, and a prefix (\n\n) must not be emitted as a complete boundary before a possible third\narrives.
Reusing split() on the accumulated pending fragment (rather than delta-native detection) sidesteps most of this, because split()'s withhold-the-tail contract already handles incomplete trailing boundaries.
Required test: delta-invariance — for any way a text is sliced into deltas, feeding the slices to Chunker yields the same chunks as one split(full_text). Fixtures: 1 delta, per-character deltas, random splits, plus the three seam cases above. The test/core/test_astream_mock.py mock gives deterministic delta control.
Activity
ajbozarth commentedon Jul 31, 2026
stream()lazy start — behavior change, needs sign-offWiring the #1440 refactor in surfaced a trace-nesting bug: the generation span becomes the parent of the streaming span, so the parent closes before its child and the OTel context unwinds out of order — a broken trace. The clean fix requires changing when a stream starts, which is outside the original design scope, so flagging before proceeding.
Today:
await stream(...)immediately starts generation, firesgeneration_pre_call, and raises any setup error — all before the firstasync for.Proposed:
stream()returns a ready handle but starts nothing. The firstasync forstarts generation, fires the hook, and surfaces setup errors.await stream(...)async forgeneration_pre_callfiresawait stream(...)async forawait stream(...)async forWhy it's arguably better: "iterating starts the work" matches the single-task design; an unconsumed stream costs nothing; and it removes the lifecycle bug rather than masking it.
Trade-off: callers can no longer start a stream and do other work while the model warms up — generation now waits for the loop. For the usual
s = await stream(...); async for ... in sthis is negligible.Ask: OK to make
stream()lazy? Clean fix for the trace bug, better contract, but a behavior change beyond #1440's original scope.ajbozarth commentedon Jul 31, 2026
@jakelorocco @planetf1 are you ok with the above change? Given it was not part of my design pitch during the call I wanted to confirm before making the change. I'll also raise this during Monday scrum for a decision.
jakelorocco commentedon Aug 3, 2026
I do like the current behavior of immediately starting generation while allowing other work to be done elsewhere (which is how regular generation works now).
I remember we had talked about reorganizing the calls / spans so that the streaming span could totally encompass the generation span. Is that not option?
ajbozarth commentedon Aug 3, 2026
@jakelorocco The other option would be to mirror what I did with out current code, which is to move the start hook out of the iterator into the start of the function that creates the iterator. The issue with that is it would require a user to always
async for ... in sotherwise the already open Otel span will never close, also if they decided to do anything between thes = await stream(...)andasync for ... in sthe span would already be open. IMHO this behavior though similar to how it is currently is not great.planetf1 commentedon Aug 4, 2026
Thinking about it more — I don't think it has to be either/or. What both suggestions were missing was a close that actually happens, whether or not iteration finishes. If someone just breaks out of the loop early, a plain async generator won't reliably clean itself up - the leak concern. But if the stream is something you explicitly open and close (
async with) rather than just loop over, we get both: it stays responsive, and the span closes properly regardless of how the loop ends. That also makes it the outer span wrapping generation - which Jake mentioned.Claude says (implementation notes, checked against current code):
stream()needs to become sync, with generation moved into__aenter__— so streaming starts unambiguously at the point you enter theasync with, not buried inside the generator.async with await stream(...)would otherwise reproduce the same ordering bug.mainalready has (STREAMING_STARTfires before generation starts) — feat: validated streaming refactor POC + Phase 2 epic breakdown #1409's POC regressed it by moving the start inside the generator.except BaseExceptionthe current code uses, since cleanup awaits are themselves cancellable.Follow-up issue needed to fix other incomplete spans — e.g.
STREAMING_ENDtoday only fires fromacomplete(), not from the orchestrator'sfinallyinstreaming.py, so an abandoned stream already leaks that span onmain, independent of this refactor.refactor(streaming)!: replace stream_with_chunking with single-task s…
refactor(streaming)!: replace stream_with_chunking with single-task s…
refactor(streaming)!: replace stream_with_chunking with single-task s…