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

docs/dev/migrate-streaming-v0.8.md

@@ -0,0 +1,181 @@
+# Migrating streaming from v0.7 to the single-task `stream()` API
+
+v0.7 exposed streaming validation through `stream_with_chunking()`, which
+returned a `StreamChunkingResult` driven by a background orchestration task. You
+observed it through `result.events()` (typed events) or `result.astream()` (raw
+chunks), and called `result.acomplete()` to wait for the background task to
+finish.
+
+That two-task model is replaced by a single-task primitive: `stream()` returns a
+`Streamer` you consume directly with `async for`, on your own task. There is no
+background orchestrator and no `acomplete()`. Typed events now come from the
+`streaming_event` plugin hook rather than an iterator on the result. This is a
+**breaking change with no deprecation shim** — call sites must be updated.
+
+## API mapping
+
+| v0.7 | v0.8 | Notes |
+| --- | --- | --- |
+| `stream_with_chunking(...)` | `stream(...)` | Same arguments, except `chunking` now defaults to `None` (raw deltas) instead of `"sentence"`; pass `chunking="sentence"` to preserve v0.7 chunk boundaries |
+| returns `StreamChunkingResult` | returns `Streamer` | Consume with `async for`, ideally inside `async with` |
+| `async for chunk in result.astream()` | `async for chunk in streamer` | Iterate the `Streamer` directly |
+| `async for event in result.events()` | `@hook("streaming_event")` plugin | Events move to the hook (see below) |
+| `await result.acomplete()` | *(removed)* | Consuming the stream drives it to completion |
+| `STREAMING_ORCHESTRATION_START`/`_END` hooks | *(removed)* | No replacement — the whole run is on one task now, so there is nothing to reattach a span across |
+| `result.completed` | `not streamer.failed_early` | |
+| *(new)* | `streamer.completed_normally` | `True` only on natural completion; unlike `not failed_early`, it is `False` after an early `break` |
+| `result.full_text` | `streamer.full_text` | Same |
+| `result.streaming_failures` | `streamer.streaming_failures` | Same |
+| `result.final_validations` | `streamer.final_validations` | Same |
+| `result.as_thunk` | `streamer.mot` | Set on natural completion |
+| `SentenceChunker` | `SentenceChunking` | Strategy classes renamed |
+| `WordChunker` | `WordChunking` | |
+| `ParagraphChunker` | `ParagraphChunking` | |
+
+Wrap consumption in `async with` so the generation is cancelled on every exit
+path — an early `break` or an exception — instead of leaking an abandoned
+background stream.
+
+## Chunking strategy renames
+
+The three built-in strategy classes were renamed from `...Chunker` to
+`...Chunking`, freeing the `Chunker` name for the new stateful driver:
+
+```python
+# v0.7
+from mellea.stdlib.chunking import SentenceChunker, WordChunker, ParagraphChunker
+
+# v0.8
+from mellea.stdlib.chunking import SentenceChunking, WordChunking, ParagraphChunking
+```
+
+This only affects code that imports a strategy class by name (for example, to
+subclass it or pass an instance). Passing a string alias — `chunking="sentence"`,
+`"word"`, or `"paragraph"` — is unchanged.
+
+## Before and after: observing events
+
+If you consumed typed events with `result.events()`, the events move to a
+`streaming_event` plugin. Both snippets below produce the same output — they are
+the `main()` from `docs/examples/streaming/validated_streaming.py`, v0.7 then
+v0.8, using the same requirement, prompt, and chunking.
+
+### v0.7
+
+```python
+result = await stream_with_chunking(
+    action, backend, ctx, requirements=[req], chunking="sentence"
+)
+
+print("Streaming events as they arrive:")
+async for event in result.events():
+    match event:
+        case ChunkEvent():
+            print(f"  CHUNK[{event.chunk_index}]: {event.text!r}")
+        case QuickCheckEvent(passed=False):
+            print(
+                f"  QUICK_CHECK[{event.chunk_index}]: FAIL — "
+                f"{event.results[0].reason if event.results else 'unknown reason'}"
+            )
+        case QuickCheckEvent():
+            print(f"  QUICK_CHECK[{event.chunk_index}]: pass")
+        case StreamingDoneEvent():
+            print(f"  STREAMING_DONE: {len(event.full_text)} chars accumulated")
+        case FullValidationEvent():
+            print(f"  FULL_VALIDATION: {'PASS' if event.passed else 'FAIL'}")
+        case CompletedEvent():
+            print(f"  COMPLETED: success={event.success}")
+        case _:
+            pass
+
+await result.acomplete()
+
+print(f"\nCompleted normally: {result.completed}")
+print(f"Full text: {result.full_text!r}")
+```
+
+### v0.8
+
+```python
+@hook("streaming_event")
+async def print_events(payload, ctx) -> None:
+    event = payload.event
+    match event:
+        case ChunkEvent():
+            print(f"  CHUNK[{event.chunk_index}]: {event.text!r}")
+        case QuickCheckEvent(passed=False):
+            print(
+                f"  QUICK_CHECK[{event.chunk_index}]: FAIL — "
+                f"{event.results[0].reason if event.results else 'unknown reason'}"
+            )
+        case QuickCheckEvent():
+            print(f"  QUICK_CHECK[{event.chunk_index}]: pass")
+        case StreamingDoneEvent():
+            print(f"  STREAMING_DONE: {len(event.full_text)} chars accumulated")
+        case FullValidationEvent():
+            print(f"  FULL_VALIDATION: {'PASS' if event.passed else 'FAIL'}")
+        case CompletedEvent():
+            print(f"  COMPLETED: success={event.success}")
+        case _:
+            pass
+
+
+register(print_events)
+
+print("Stream events as they arrive:")
+async with await stream(
+    action, backend, ctx, requirements=[req], chunking="sentence"
+) as streamer:
+    # Draining the stream fires the events; the hook does the printing.
+    async for _chunk in streamer:
+        pass
+
+print(f"\nCompleted normally: {streamer.completed_normally}")
+print(f"Full text: {streamer.full_text!r}")
+```
+
+The three transformations to make:
+
+1. **`stream_with_chunking(...)` → `async with await stream(...) as streamer:`**,
+   and iterate `streamer` directly. The `async with` replaces `acomplete()` for
+   cleanup.
+2. **The `result.events()` match loop → a `@hook("streaming_event")` plugin.**
+   The event vocabulary and payloads are unchanged; register the plugin (with
+   `register()` or a `plugin_scope`) before consuming. Draining the stream is
+   what fires the events.
+3. **`result.<attr>` → `streamer.<attr>`**. `result.completed` maps directly to
+   `not streamer.failed_early`; prefer the new `streamer.completed_normally` if
+   you need a signal that also excludes an early `break` (used above).
+
+## Before and after: consuming raw chunks
+
+If you never used `events()` and only consumed the validated chunk text with
+`result.astream()`, the migration is smaller — iterate the `Streamer` directly
+and drop `acomplete()`:
+
+### v0.7
+
+```python
+result = await stream_with_chunking(
+    action, backend, ctx, requirements=[req], chunking="sentence"
+)
+async for chunk in result.astream():
+    print(chunk)
+await result.acomplete()
+```
+
+### v0.8
+
+```python
+async with await stream(
+    action, backend, ctx, requirements=[req], chunking="sentence"
+) as streamer:
+    async for chunk in streamer:
+        print(chunk)
+```
+
+## See also
+
+- [Streaming validation tutorial](../docs/tutorials/06-streaming-validation.md)
+- [How-to: Async and streaming](../docs/how-to/use-async-and-streaming.md)
+- [`docs/examples/streaming/`](../examples/streaming/)

docs/docs/concepts/requirements-system.md

@@ -342,8 +342,8 @@ field:
 - `"fail"` — the stream is cancelled immediately; no further chunks reach the
   consumer; `validate()` is skipped for this requirement.
 
-State isolation is per-clone: `stream_with_chunking()` copies each requirement
-with `copy()` before starting the orchestrator, so the original objects are never
+State isolation is per-clone: `stream()` copies each requirement
+with `copy()` before generation starts, so the original objects are never
 mutated. Requirements that accumulate state across chunks (e.g. a running word
 count) should reassign mutable containers rather than mutate in place, since
 clones share the original's `__dict__` values at copy time.

docs/docs/examples/index.md

@@ -32,7 +32,7 @@ to run.
 | `context/` | Context inspection, sampling with context trees, parallel context branches |
 | `sessions/` | Custom session types and backend selection |
 | `async/` | How to utilize basic async capabilities |
-| `streaming/` | `stream_with_chunking()` with per-chunk validation, typed event vocabulary, early-exit on fail |
+| `streaming/` | `stream()` with per-chunk validation, typed event vocabulary, early-exit on fail |
 
 ### Data and documents
 

docs/docs/how-to/use-async-and-streaming.md

@@ -142,6 +142,40 @@ How `astream()` behaves:
 > **Warning:** Do not call `astream()` from multiple coroutines simultaneously on
 > the same thunk. Each thunk should have a single reader.
 
+### Iterating a thunk with `async for`
+
+`astream()` is the low-level primitive. To consume a thunk as an async iterator,
+use `async for` — ideally inside `async with`, which cancels the generation if
+you leave the loop early (an exception or `break`) so an abandoned stream does
+not keep running:
+
+```python
+# Requires: mellea
+# Returns: None
+import asyncio
+import mellea
+from mellea.backends import ModelOption
+
+async def main():
+    m = mellea.start_session()
+    mot = await m.ainstruct(
+        "Write a short story about a robot learning to cook.",
+        model_options={ModelOption.STREAM: True},
+    )
+
+    async with mot:
+        async for delta in mot:
+            print(delta, end="", flush=True)
+    print()  # newline after streaming completes
+
+asyncio.run(main())
+# Output will vary — LLM responses depend on model and temperature.
+```
+
+Each iteration yields the same delta `astream()` would return; iteration ends
+when the thunk is computed. Like `astream()`, a thunk has a single reader — a
+second `async for` over the same thunk raises rather than splitting the stream.
+
 ### Streaming timeout
 
 Mellea waits up to 120 seconds for each chunk by default, including the first
@@ -207,14 +241,15 @@ For parallel generation, use `SimpleContext`.
 
 ## Streaming with per-chunk validation
 
-`stream_with_chunking()` adds per-chunk validation to a streaming generation.
-It splits the accumulated text into semantic units (sentences, words, or
-paragraphs), calls `stream_validate()` on each chunk in parallel, and can
-exit early if any requirement returns `"fail"` — preventing the consumer from
-seeing invalid content mid-stream.
+`stream()` adds per-chunk validation to a streaming generation. It splits the
+accumulated text into semantic units (sentences, words, or paragraphs), calls
+`stream_validate()` on each chunk in parallel, and can exit early if any
+requirement returns `"fail"` — preventing the consumer from seeing invalid
+content mid-stream.
 
-The primary way to observe a `stream_with_chunking()` run is via typed
-`StreamEvent` objects from `result.events()`:
+`stream()` returns a `Streamer` you consume with `async for`, ideally inside
+`async with` so the generation is released on every exit path (including an
+early `break` or exception):
 
 ```python
 # Requires: mellea
@@ -225,14 +260,7 @@ from mellea.core.backend import Backend
 from mellea.core.base import Context
 from mellea.core.requirement import PartialValidationResult, Requirement, ValidationResult
 from mellea.stdlib.components import Instruction
-from mellea.stdlib.streaming import (
-    ChunkEvent,
-    CompletedEvent,
-    FullValidationEvent,
-    QuickCheckEvent,
-    StreamingDoneEvent,
-    stream_with_chunking,
-)
+from mellea.stdlib.streaming import stream
 
 
 class MaxSentencesReq(Requirement):
@@ -267,48 +295,24 @@ async def main() -> None:
     action = Instruction("Write a two-sentence summary of the water cycle.")
     req = MaxSentencesReq(limit=3)
 
-    result = await stream_with_chunking(
+    async with await stream(
         action, m.backend, m.ctx, requirements=[req], chunking="sentence"
-    )
+    ) as streamer:
+        async for chunk in streamer:
+            print(chunk)
 
-    async for event in result.events():
-        match event:
-            case ChunkEvent():
-                print(f"  chunk[{event.chunk_index}]: {event.text!r}")
-            case QuickCheckEvent(passed=False):
-                print(f"  FAIL at chunk {event.chunk_index}: {event.results}")
-            case StreamingDoneEvent():
-                print(f"  stream done — {len(event.full_text)} chars")
-            case FullValidationEvent():
-                print(f"  final: {'pass' if event.passed else 'fail'}")
-            case CompletedEvent():
-                print(f"  completed — success={event.success}")
-            case _:
-                pass  # ErrorEvent and other future types
-
-    await result.acomplete()
-    print(f"completed={result.completed}, failures={len(result.streaming_failures)}")
+    # Terminal state on the Streamer, after the loop.
+    print(f"Completed normally: {streamer.completed_normally}")
+    for _req, result in streamer.streaming_failures:
+        print(f"Streaming failure: {result.reason}")
 
 
 asyncio.run(main())
 ```
 
-If you only need the raw validated text without event metadata, use
-`result.astream()` instead:
-
-```python
-result = await stream_with_chunking(
-    action, m.backend, m.ctx, requirements=[req], chunking="sentence"
-)
-async for chunk in result.astream():
-    print(chunk)
-await result.acomplete()
-```
-
-Both `astream()` (raw chunks) and `events()` are available on the same result
-object. They use independent queues, so you can run them concurrently with
-`asyncio.gather`. Both are **single-consumer** — a second iteration on either
-will block indefinitely.
+To observe the run through typed `StreamEvent` objects instead, register a
+plugin on the `streaming_event` hook — see the
+[streaming validation tutorial](../tutorials/06-streaming-validation.md).
 
 ### The `stream_validate` tri-state
 

docs/docs/observability/tracing.md

@@ -193,16 +193,16 @@ One per requirement-validation batch, wherever requirements are checked.
 | `mellea.validation.failed_count` | Number of requirements that failed |
 | `mellea.validation.failure_reasons` | List of failing requirements' reasons; recorded only when `MELLEA_TRACES_CONTENT=true` |
 
-#### `stream_with_chunking` span
+#### `stream` span
 
-One per `stream_with_chunking()` run, wrapping the backend generation and any
-per-chunk validation.
+One per `stream()` run, wrapping the backend generation and any per-chunk
+validation.
 
 | Attribute | Description |
 | --------- | ----------- |
 | `mellea.streaming.has_requirements` | Whether requirements were supplied |
 | `mellea.streaming.requirement_count` | Number of requirements supplied |
-| `mellea.streaming.chunking_strategy` | `ChunkingStrategy` class name (e.g., `SentenceChunker`) |
+| `mellea.streaming.chunking_strategy` | `ChunkingStrategy` class name (e.g., `SentenceChunking`) |
 | `mellea.streaming.full_text_length` | Length of the accumulated text at completion |
 | `gen_ai.request.model` | Model ID, when known |
 | `gen_ai.provider.name` | Provider name, when known |
@@ -283,21 +283,20 @@ Tool execution happens after the generating call completes, so `execute_tool`
 spans are not nested inside the `action` that requested them. Outside a session
 they are root spans.
 
-In a `stream_with_chunking` run, the backend generation and each per-chunk
-validation call nest under the `stream_with_chunking` span as sibling `chat`
-spans:
+In a `stream` run, the backend generation and each per-chunk validation call
+nest under the `stream` span as sibling `chat` spans:
 
 ```text
-stream_with_chunking      (mellea.application)
-│                         [mellea.streaming.chunking_strategy=SentenceChunker]
+stream                    (mellea.application)
+│                         [mellea.streaming.chunking_strategy=SentenceChunking]
 ├── chat                  (mellea.backend)    ← streaming generation
 │                         [gen_ai.request.model=granite4.1:3b]
 └── chat                  (mellea.backend)    ← per-chunk validation
                           [gen_ai.request.model=granite4.1:3b]
 ```
 
-The `stream_with_chunking` span itself parents under whatever span is active
-when the run starts, or is a root span when none is.
+The `stream` span itself parents under whatever span is active when the run
+starts, or is a root span when none is.
 
 > **Note:** Full span nesting requires Python 3.12+. On Python 3.11 some spans
 > may appear flattened rather than nested; all spans and attributes are still