feat(agent): strip raw tool-scaffolding from streamed answers — deterministic filter with one bounded recovery

tests/unit/test_llm_client.py

@@ -27,6 +27,7 @@ from app.rag.llm import (
     ToolCallPiece,
     chat_stream_retried,
 )
+from app.rag.scaffolding import ScaffoldingFilter
 
 
 def _settings(**kwargs: Any) -> Settings:
@@ -337,13 +338,19 @@ class _FakeCompletions:
         self.completion = completion
         self.kwargs: dict | None = None
         self.chat_kwargs: dict | None = None
+        #: Every SDK-shaped stream handed out — teardown tests assert the
+        #: phase-48 ``close()`` on them (phase 71 task 02: with/without
+        #: a filter, the teardown path is the same object).
+        self.streams: list[_FakeChatStream] = []
 
     async def create(self, **kwargs) -> _FakeChatStream | _FakeCompletion:
         self.kwargs = kwargs
         if self.fail is not None:
             raise self.fail
         if kwargs.get("stream"):
-            return _FakeChatStream(self.chunks)
+            stream = _FakeChatStream(self.chunks)
+            self.streams.append(stream)
+            return stream
         self.chat_kwargs = kwargs
         assert self.completion is not None
         return self.completion
@@ -721,6 +728,146 @@ def test_chat_stream_non_object_tool_arguments_raise_llm_error() -> None:
         asyncio.run(drain())
 
 
+# ---------- scaffolding filter integration (phase 71, task 02) ----------
+
+#: The incident's raw span (2026-09-03) — the filter's reason to exist.
+_INCIDENT_SPAN = "<|tool_call_start|>[read(path='/homelab/backup-notes.md')]<|tool_call_end|>"
+
+
+def _collect_filtered(
+    llm: LLMClient,
+    messages: list[dict[str, str]],
+    scaffolding: ScaffoldingFilter | None,
+) -> list[StreamPiece]:
+    """Collect pieces from a tools-less filtered stream (phase 71): without
+    tools, no ToolCallPiece can appear (the phase-37 contract)."""
+    async def run() -> list[StreamPiece]:
+        pieces = [p async for p in llm.chat_stream(messages, scaffolding=scaffolding)]
+        assert all(isinstance(p, StreamPiece) for p in pieces)
+        return cast("list[StreamPiece]", pieces)
+
+    return asyncio.run(run())
+
+
+def test_chat_stream_with_filter_strips_span_mid_stream() -> None:
+    """A span mid-stream never reaches the pieces: the clean remainder
+    flows, ``stripped_chars`` is exact, and no piece carries scaffolding."""
+    f = ScaffoldingFilter()
+    llm, _ = _make_stream_client(
+        [_chunk("Hello "), _chunk(_INCIDENT_SPAN + " world"), _chunk("!")]
+    )
+    pieces = _collect_filtered(llm, _RETRY_MSGS, f)
+    assert [(p.kind, p.text) for p in pieces] == [
+        ("content", "Hello "),
+        ("content", " world"),
+        ("content", "!"),
+    ]
+    assert f.stripped_chars == len(_INCIDENT_SPAN)
+    assert all("<|" not in p.text for p in pieces if p.kind == "content")
+
+
+def test_chat_stream_with_filter_span_split_across_chunks() -> None:
+    """A span split across two chunks never emits a partial marker —
+    nothing leaks until the span completes, then the clean text on both
+    sides flows and the whole span counts as stripped."""
+    span = _INCIDENT_SPAN
+    cut = len("<|tool_call_start|>")  # split right after the start token
+    f = ScaffoldingFilter()
+    llm, _ = _make_stream_client([_chunk("A " + span[:cut]), _chunk(span[cut:] + " B")])
+    pieces = _collect_filtered(llm, _RETRY_MSGS, f)
+    assert [(p.kind, p.text) for p in pieces] == [
+        ("content", "A "),
+        ("content", " B"),
+    ]
+    assert f.stripped_chars == len(span)
+
+
+def test_chat_stream_with_filter_yields_nothing_for_pure_scaffolding() -> None:
+    """A content stream of pure scaffolding yields ZERO content pieces —
+    an empty clean result yields nothing (no empty ``delta`` frames)."""
+    f = ScaffoldingFilter()
+    llm, _ = _make_stream_client(
+        [_chunk(_INCIDENT_SPAN), _chunk("<|tool_calls|>")]
+    )
+    pieces = _collect_filtered(llm, _RETRY_MSGS, f)
+    assert pieces == []
+    assert f.stripped_chars == len(_INCIDENT_SPAN) + len("<|tool_calls|>")
+
+
+def test_chat_stream_with_filter_leaves_thinking_raw() -> None:
+    """Locked (phase 71): the scratchpad stays raw — a span inside
+    ``reasoning_content`` is yielded verbatim and never counts as
+    stripped."""
+    f = ScaffoldingFilter()
+    llm, _ = _make_stream_client(
+        [_chunk("", reasoning=_INCIDENT_SPAN), _chunk("ok")]
+    )
+    pieces = _collect_filtered(llm, _RETRY_MSGS, f)
+    assert [(p.kind, p.text) for p in pieces] == [
+        ("thinking", _INCIDENT_SPAN),
+        ("content", "ok"),
+    ]
+    assert f.stripped_chars == 0
+
+
+def test_chat_stream_without_filter_is_the_raw_path() -> None:
+    """``scaffolding=None`` (the default, and the explicit opt-out): a span
+    in content is yielded verbatim — byte-identical to the pre-phase-71
+    raw path for callers that do not pass a filter."""
+    llm, _ = _make_stream_client([_chunk(_INCIDENT_SPAN)])
+    explicit_none = _collect_filtered(llm, _RETRY_MSGS, None)
+    default = asyncio.run(_collect(llm, _RETRY_MSGS))
+    assert explicit_none == default == [StreamPiece("content", _INCIDENT_SPAN)]
+
+
+def test_chat_stream_flushed_tail_precedes_tool_call_pieces() -> None:
+    """Content-before-tools wire convention (phase 71): a stream that ends
+    with a held filter tail (a partial marker at EOF — flushed as-is) +
+    tool_calls deltas yields the flushed tail content piece BEFORE the
+    materialized ``ToolCallPiece``."""
+    f = ScaffoldingFilter()
+    llm, _ = _make_stream_client(
+        [
+            _chunk("done <|tool_call_st"),  # held: a live prefix of the start token
+            _chunk(
+                None,
+                tool_calls=[_tool_call(0, id="call_t", name="ls")],
+                finish_reason="tool_calls",
+            ),
+        ]
+    )
+
+    async def run() -> list[StreamPiece | ToolCallPiece]:
+        return [
+            p
+            async for p in llm.chat_stream(_RETRY_MSGS, tools=_AGENT_TOOLS, scaffolding=f)
+        ]
+
+    pieces = asyncio.run(run())
+    assert pieces == [
+        StreamPiece("content", "done "),
+        StreamPiece("content", "<|tool_call_st"),
+        ToolCallPiece(id="call_t", name="ls", arguments={}),
+    ]
+
+
+def test_chat_stream_abandon_with_filter_closes_stream() -> None:
+    """Phase-48 teardown is independent of the phase-71 filter: a consumer
+    abandon mid-filter still closes the endpoint stream exactly once."""
+    f = ScaffoldingFilter()
+    llm, completions = _make_stream_client([_chunk(f"piece {i} ") for i in range(1, 6)])
+
+    async def run() -> None:
+        gen = llm.chat_stream(_RETRY_MSGS, scaffolding=f)
+        first = await gen.__anext__()
+        assert isinstance(first, StreamPiece)
+        assert first.text == "piece 1 "
+        await gen.aclose()  # the consumer stops after the first piece
+
+    asyncio.run(run())
+    assert completions.streams[0].close_calls == 1
+
+
 # ---------- one-shot chat: LLMClient.chat (phase 30, task 01) ----------
 
 
@@ -826,11 +973,15 @@ class _ScriptedClient(LLMClient):
         ] = []
         #: Indices of attempts whose stream teardown has run.
         self.closed: list[int] = []
+        #: The caller-owned filter each attempt's ``chat_stream`` received
+        #: (phase 71 task 02 — the retry primitive forwards it).
+        self.scaffoldings: list[ScaffoldingFilter | None] = []
 
     def chat_stream(
         self,
         messages: list[dict[str, str]],
         tools: list[dict[str, Any]] | None = None,
+        scaffolding: ScaffoldingFilter | None = None,
     ) -> AsyncGenerator[StreamPiece | ToolCallPiece, None]:
         index = len(self.request_args)
         pieces, error = (
@@ -841,15 +992,30 @@ class _ScriptedClient(LLMClient):
         self.request_args.append(
             (list(messages), list(tools) if tools is not None else None)
         )
-        return self._attempt(index, pieces, error)
+        self.scaffoldings.append(scaffolding)
+        return self._attempt(index, pieces, error, scaffolding)
 
     async def _attempt(
-        self, index: int, pieces: list[StreamPiece | ToolCallPiece],
+        self,
+        index: int,
+        pieces: list[StreamPiece | ToolCallPiece],
         error: Exception | None,
+        scaffolding: ScaffoldingFilter | None,
     ) -> AsyncGenerator[StreamPiece | ToolCallPiece, None]:
         try:
             for piece in pieces:
-                yield piece
+                if (
+                    scaffolding is not None
+                    and isinstance(piece, StreamPiece)
+                    and piece.kind == "content"
+                ):
+                    # Emulate the real contract (phase 71): content feeds
+                    # the filter, an empty clean result yields nothing.
+                    cleaned = scaffolding.feed(piece.text)
+                    if cleaned:
+                        yield StreamPiece("content", cleaned)
+                else:
+                    yield piece
             if error is not None:
                 raise error
         finally:
@@ -875,12 +1041,18 @@ def _collect_retried(
     tools: list[dict[str, Any]] | None = None,
     retries: int,
     delay: float,
+    scaffolding: ScaffoldingFilter | None = None,
 ) -> list[StreamPiece | ToolCallPiece | RetryPiece]:
     async def run() -> list[StreamPiece | ToolCallPiece | RetryPiece]:
         return [
             p
             async for p in chat_stream_retried(
-                client, messages, tools=tools, retries=retries, delay=delay
+                client,
+                messages,
+                tools=tools,
+                retries=retries,
+                delay=delay,
+                scaffolding=scaffolding,
             )
         ]
 
@@ -1016,6 +1188,40 @@ def test_retried_zero_delay_still_notifies(monkeypatch: pytest.MonkeyPatch) -> N
     assert sleeps == [0]
 
 
+def test_retried_forwards_the_filter_to_every_attempt(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Phase 71: the caller-owned filter reaches EVERY attempt's
+    ``chat_stream`` — the same object on the dead attempt and the
+    surviving one."""
+    client = _ScriptedClient(
+        [([], LLMError("down")), ([StreamPiece("content", "ok")], None)]
+    )
+    f = ScaffoldingFilter()
+    _record_sleeps(monkeypatch)
+    pieces = _collect_retried(client, _RETRY_MSGS, retries=1, delay=0, scaffolding=f)
+    assert pieces == [RetryPiece(2, 2), StreamPiece("content", "ok")]
+    assert client.scaffoldings == [f, f]
+
+
+def test_retried_reuses_the_unfed_filter_after_a_dead_attempt(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Phase 71: reusing the SAME filter across attempts is safe by
+    construction — the dead attempt emitted no piece, so the filter was
+    never fed; the surviving attempt filters through it as if fresh."""
+    span = _INCIDENT_SPAN
+    client = _ScriptedClient(
+        [([], LLMError("down")), ([StreamPiece("content", span + " clean")], None)]
+    )
+    f = ScaffoldingFilter()
+    _record_sleeps(monkeypatch)
+    pieces = _collect_retried(client, _RETRY_MSGS, retries=1, delay=0, scaffolding=f)
+    assert pieces == [RetryPiece(2, 2), StreamPiece("content", " clean")]
+    assert f.stripped_chars == len(span)
+    assert client.scaffoldings == [f, f]
+
+
 def test_retried_abandon_mid_attempt_closes_the_attempt_stream() -> None:
     """Consumer abandon at a mid-attempt piece (the stop-generation path,
     phase 48): no exception leaks and the attempt's stream is torn down