feat(stdlib): flush trailing chunk fragment at end of stream

ChunkingStrategy.split() withholds the trailing fragment by design
(#899). Previously the orchestrator discarded it — it appeared in
full_text and the final validate() saw it, but it was never yielded to
astream() consumers and never seen by stream_validate. For a response
that did not end in a chunk terminator (e.g. "Sentence one. Sentence
two." with no trailing whitespace under SentenceChunker), the last
sentence silently bypassed streaming validation.

Adds ChunkingStrategy.flush(accumulated_text) -> list[str]:
- Default in the ABC returns [] (backward-compatible — external
  chunkers retain the old discard behaviour until they opt in).
- SentenceChunker, WordChunker, ParagraphChunker each override to
  return the withheld trailing fragment as a single-element list.

_orchestrate_streaming calls chunking.flush(accumulated) after the main
loop (only when the stream ended naturally, not on early exit — a
cancelled stream's trailing fragment is by definition incomplete).
Each flushed chunk goes through the same stream_validate / emit path
as regular chunks, so the "no unvalidated content reaches the consumer"
invariant extends to the trailing fragment, and a fail on the fragment
still records a streaming failure and skips final validate().

Tests:
- 13 new chunker tests covering the default-discard behaviour and each
  built-in's flush logic (empty input, fragment-present, already-
  terminated cases).
- test_trailing_fragment_is_flushed_to_consumer: stream_validate sees
  the fragment and astream yields it.
- test_early_exit_on_trailing_fragment: fail on the flushed fragment
  propagates to streaming_failures and skips final validation.

Assisted-by: Claude Code

Author: planetf1

mellea/stdlib/chunking.py

@@ -35,6 +35,27 @@ def split(self, accumulated_text: str) -> list[str]:
         """
         ...
 
+    def flush(self, accumulated_text: str) -> list[str]:
+        """Return any trailing fragment that ``split`` withheld.
+
+        Called once by the orchestrator after the stream has ended naturally
+        (not on early-exit cancellation).  Gives the chunker a chance to
+        release the final fragment that did not reach a terminator.
+
+        The default implementation returns an empty list — the trailing
+        fragment is discarded.  Built-in chunkers override this to return
+        the withheld fragment as a single-element list when non-empty.
+
+        Args:
+            accumulated_text: The full accumulated text at stream end.
+
+        Returns:
+            The trailing fragment as ``[fragment]`` if it should be treated
+            as a final chunk, or an empty list to discard it.
+        """
+        _ = accumulated_text
+        return []
+
 
 # Sentence boundary: sentence-ending punctuation, optionally followed by a closing
 # quote or paren, then whitespace.
@@ -94,6 +115,19 @@ def split(self, accumulated_text: str) -> list[str]:
 
         return chunks
 
+    def flush(self, accumulated_text: str) -> list[str]:
+        """Return the trailing sentence fragment (if any) as a final chunk."""
+        if not accumulated_text:
+            return []
+        remaining = accumulated_text
+        while True:
+            match = _SENTENCE_BOUNDARY.search(remaining)
+            if match is None:
+                break
+            remaining = remaining[match.end() :].lstrip()
+        trailing = remaining.strip()
+        return [trailing] if trailing else []
+
 
 class WordChunker(ChunkingStrategy):
     """Splits accumulated text on whitespace boundaries.
@@ -134,6 +168,18 @@ def split(self, accumulated_text: str) -> list[str]:
 
         return parts
 
+    def flush(self, accumulated_text: str) -> list[str]:
+        """Return the trailing word fragment (if any) as a final chunk."""
+        if not accumulated_text:
+            return []
+        if accumulated_text[-1].isspace():
+            return []
+        parts = _WHITESPACE.split(accumulated_text)
+        for part in reversed(parts):
+            if part:
+                return [part]
+        return []
+
 
 class ParagraphChunker(ChunkingStrategy):
     r"""Splits accumulated text on double-newline paragraph boundaries.
@@ -168,3 +214,13 @@ def split(self, accumulated_text: str) -> list[str]:
 
         # _PARA_BOUNDARY.split on leading \n\n produces an empty first element.
         return [p for p in parts if p]
+
+    def flush(self, accumulated_text: str) -> list[str]:
+        """Return the trailing paragraph fragment (if any) as a final chunk."""
+        if not accumulated_text:
+            return []
+        if _PARA_BOUNDARY_END.search(accumulated_text):
+            return []
+        parts = _PARA_BOUNDARY.split(accumulated_text)
+        trailing = parts[-1] if parts else ""
+        return [trailing] if trailing else []

mellea/stdlib/streaming.py

@@ -144,6 +144,35 @@ async def _orchestrate_streaming(
     failed_indices: set[int] = set()
     early_exit = False
 
+    async def _validate_and_emit(c: str) -> bool:
+        """Run stream_validate on chunk c across active requirements.
+
+        Returns True if a failure was recorded (caller should early-exit),
+        False otherwise (chunk was emitted to the consumer queue).
+        """
+        active = [
+            (i, req) for i, req in enumerate(cloned_reqs) if i not in failed_indices
+        ]
+        if active:
+            pvrs: list[PartialValidationResult] = list(
+                await asyncio.gather(
+                    *[
+                        req.stream_validate(c, backend=val_backend, ctx=ctx)
+                        for _, req in active
+                    ]
+                )
+            )
+            for (idx, req), pvr in zip(active, pvrs):
+                if pvr.success == "fail":
+                    failed_indices.add(idx)
+                    result.streaming_failures.append((req, pvr))
+
+        if failed_indices:
+            return True
+
+        await result._chunk_queue.put(c)
+        return False
+
     try:
         while not mot.is_computed():
             try:
@@ -157,36 +186,27 @@ async def _orchestrate_streaming(
             prev_chunk_count = len(chunks)
 
             for c in new_chunks:
-                active = [
-                    (i, req)
-                    for i, req in enumerate(cloned_reqs)
-                    if i not in failed_indices
-                ]
-                if active:
-                    pvrs: list[PartialValidationResult] = list(
-                        await asyncio.gather(
-                            *[
-                                req.stream_validate(c, backend=val_backend, ctx=ctx)
-                                for _, req in active
-                            ]
-                        )
-                    )
-                    for (idx, req), pvr in zip(active, pvrs):
-                        if pvr.success == "fail":
-                            failed_indices.add(idx)
-                            result.streaming_failures.append((req, pvr))
-
-                if failed_indices:
+                failed = await _validate_and_emit(c)
+                if failed:
                     early_exit = True
                     result.completed = False
                     await mot.cancel_generation()
                     break
 
-                await result._chunk_queue.put(c)
-
             if early_exit:
                 break
 
+        # Stream ended naturally: flush any withheld trailing fragment and
+        # run stream_validate on it. Skipped on early exit — the generation
+        # was cancelled, the trailing fragment is incomplete.
+        if not early_exit:
+            for c in chunking.flush(accumulated):
+                failed = await _validate_and_emit(c)
+                if failed:
+                    early_exit = True
+                    result.completed = False
+                    break
+
         result.full_text = accumulated
 
         non_failed = [
@@ -238,6 +258,12 @@ async def stream_with_chunking(
     failing chunk is not emitted to the consumer; use
     :attr:`StreamChunkingResult.streaming_failures` to inspect what failed.
 
+    When the stream ends naturally, any trailing fragment withheld by the
+    chunking strategy (see :meth:`~mellea.stdlib.chunking.ChunkingStrategy.flush`)
+    is released as a final chunk and run through ``stream_validate`` on the
+    same terms as the regular chunks.  On early exit, the trailing fragment
+    is discarded because the generation was cancelled mid-token.
+
     After the stream ends (naturally or via early exit), ``validate()`` is
     called on all requirements that did not return ``"fail"``.  Requirements
     are cloned (``copy(req)``) before use so originals are never mutated.

test/stdlib/test_chunking.py

@@ -242,3 +242,79 @@ def test_paragraph_chunker_incremental_simulation():
         "First paragraph.",
         "Second paragraph.",
     ]
+
+
+# ---------------------------------------------------------------------------
+# flush() — trailing-fragment release at end of stream
+# ---------------------------------------------------------------------------
+
+
+def test_default_flush_returns_empty_list():
+    """The ABC default discards the trailing fragment."""
+
+    class Minimal(ChunkingStrategy):
+        def split(self, accumulated_text: str) -> list[str]:
+            _ = accumulated_text
+            return []
+
+    assert Minimal().flush("anything at all") == []
+    assert Minimal().flush("") == []
+
+
+def test_sentence_chunker_flush_empty():
+    assert SentenceChunker().flush("") == []
+
+
+def test_sentence_chunker_flush_only_complete():
+    """All text ends in a complete sentence with trailing whitespace → no fragment."""
+    assert SentenceChunker().flush("One. Two. ") == []
+
+
+def test_sentence_chunker_flush_trailing_fragment():
+    """Final sentence without trailing whitespace is released by flush."""
+    assert SentenceChunker().flush("One. Two without period") == ["Two without period"]
+
+
+def test_sentence_chunker_flush_terminated_no_trailing_space():
+    """Final sentence with terminator but no trailing whitespace is a fragment
+    under split() semantics and gets released by flush()."""
+    assert SentenceChunker().flush("One. Two.") == ["Two."]
+
+
+def test_sentence_chunker_flush_single_sentence_no_terminator():
+    assert SentenceChunker().flush("Incomplete sentence") == ["Incomplete sentence"]
+
+
+def test_word_chunker_flush_empty():
+    assert WordChunker().flush("") == []
+
+
+def test_word_chunker_flush_trailing_whitespace():
+    """Trailing whitespace means all words are complete → no fragment."""
+    assert WordChunker().flush("one two three ") == []
+
+
+def test_word_chunker_flush_trailing_fragment():
+    assert WordChunker().flush("one two three") == ["three"]
+
+
+def test_word_chunker_flush_single_word():
+    assert WordChunker().flush("solo") == ["solo"]
+
+
+def test_paragraph_chunker_flush_empty():
+    assert ParagraphChunker().flush("") == []
+
+
+def test_paragraph_chunker_flush_only_complete():
+    assert ParagraphChunker().flush("Para one.\n\nPara two.\n\n") == []
+
+
+def test_paragraph_chunker_flush_trailing_fragment():
+    assert ParagraphChunker().flush("Para one.\n\nPara two (no sep)") == [
+        "Para two (no sep)"
+    ]
+
+
+def test_paragraph_chunker_flush_single_paragraph_no_separator():
+    assert ParagraphChunker().flush("Only paragraph") == ["Only paragraph"]

test/stdlib/test_streaming.py

@@ -443,6 +443,124 @@ def _capturing_copy(self: ChunkRecordingReq) -> ChunkRecordingReq:
     assert not all(len(seen[i]) < len(seen[i + 1]) for i in range(len(seen) - 1))
 
 
+@pytest.mark.asyncio
+async def test_trailing_fragment_is_flushed_to_consumer() -> None:
+    """Response without trailing whitespace: final sentence reaches astream() and stream_validate."""
+
+    class ChunkRecordingReq(Requirement):
+        def __init__(self) -> None:
+            self.seen_chunks: list[str] = []
+
+        def __copy__(self) -> "ChunkRecordingReq":
+            clone = ChunkRecordingReq()
+            clone.seen_chunks = []
+            return clone
+
+        def format_for_llm(self) -> str:
+            return "chunk recorder"
+
+        async def stream_validate(
+            self, chunk: str, *, backend: Any, ctx: Any
+        ) -> PartialValidationResult:
+            self.seen_chunks.append(chunk)
+            return PartialValidationResult("unknown")
+
+        async def validate(
+            self,
+            backend: Any,
+            ctx: Any,
+            *,
+            format: Any = None,
+            model_options: Any = None,
+        ) -> ValidationResult:
+            return ValidationResult(result=True)
+
+    # No trailing whitespace after the final sentence — SentenceChunker withholds it.
+    response = "First sentence. Second sentence."
+    backend = StreamingMockBackend(response, token_size=4)
+    req = ChunkRecordingReq()
+
+    captured: list[ChunkRecordingReq] = []
+    original_copy = ChunkRecordingReq.__copy__
+
+    def _capturing_copy(self: ChunkRecordingReq) -> ChunkRecordingReq:
+        clone = original_copy(self)
+        captured.append(clone)
+        return clone
+
+    ChunkRecordingReq.__copy__ = _capturing_copy  # type: ignore[method-assign]
+    try:
+        result = await stream_with_chunking(
+            _action(),
+            backend,
+            _ctx(),
+            quick_check_requirements=[req],
+            chunking="sentence",
+        )
+        yielded: list[str] = []
+        async for chunk in result.astream():
+            yielded.append(chunk)
+        await result.acomplete()
+    finally:
+        ChunkRecordingReq.__copy__ = original_copy  # type: ignore[method-assign]
+
+    # Both sentences reach the consumer, including the terminating one without trailing whitespace.
+    assert yielded == ["First sentence.", "Second sentence."]
+    # stream_validate was called on both — the flush path is not a shortcut.
+    assert captured[0].seen_chunks == ["First sentence.", "Second sentence."]
+    assert result.completed is True
+
+
+@pytest.mark.asyncio
+async def test_early_exit_on_trailing_fragment() -> None:
+    """A fail on the flushed fragment records a streaming failure and skips final validate()."""
+
+    class FailOnSecondSentence(Requirement):
+        def __init__(self) -> None:
+            self._count = 0
+
+        def format_for_llm(self) -> str:
+            return "fail on second sentence"
+
+        async def stream_validate(
+            self, chunk: str, *, backend: Any, ctx: Any
+        ) -> PartialValidationResult:
+            _ = chunk, backend, ctx
+            self._count += 1
+            if self._count >= 2:
+                return PartialValidationResult("fail", reason="second sentence hit")
+            return PartialValidationResult("unknown")
+
+        async def validate(
+            self,
+            backend: Any,
+            ctx: Any,
+            *,
+            format: Any = None,
+            model_options: Any = None,
+        ) -> ValidationResult:
+            return ValidationResult(result=True)
+
+    response = "First sentence. Second sentence."
+    backend = StreamingMockBackend(response, token_size=4)
+    req = FailOnSecondSentence()
+
+    result = await stream_with_chunking(
+        _action(), backend, _ctx(), quick_check_requirements=[req], chunking="sentence"
+    )
+    yielded: list[str] = []
+    async for chunk in result.astream():
+        yielded.append(chunk)
+    await result.acomplete()
+
+    assert result.completed is False
+    assert len(result.streaming_failures) == 1
+    # First sentence was emitted; second (the flushed fragment) failed and wasn't emitted.
+    assert yielded == ["First sentence."]
+    # Early exit on fail skips final validate().
+    assert result.final_validations == []
+
+
 @pytest.mark.asyncio
 async def test_no_requirements_streams_without_validation() -> None:
     """quick_check_requirements=None → chunks produced, no validate() called."""