fix(832): close the content run at a reasoning_delta boundary — display must mirror the drain

Live-caught on a deployed review exercise: the consumer's reasoning_delta
arm flipped the splitter's in_think with a buffered content tail still
pending, so a flush while in-think (stream finish, tool boundary)
relabeled that tail as reasoning. The drain closes each content run at
the same boundary, so the committed turn kept the tail as content —
display and commit diverged. Worst case: a short answer followed by
trailing reasoning displayed as NOTHING while the commit carried the
answer plus its citations footer (the display-side blankness gate saw
empty content and dropped the footer too).

Pre-fold, display and commit came from one continuous splitter and both
lost the tail; the fold's drain corrected the commit, leaving the display
behind. ThinkTagSplitter.close_run() now closes the run exactly as the
drain does — decided text emits at the current state, only a possible
partial-tag tail carries into the next run — and the consumer calls it
before entering the reasoning phase. This also heals the cancelled-
partial rule in the same window, and covers the content-reasoning-tool
sequence interleaved-thinking lanes emit.

Riding contract fix: partial_tag_tail required only startswith, so a
complete <reasoning>/<think> self-matched as a "partial" tail and the
drain carried a finished open tag across the run boundary, relabeling
the next run. A partial tag is now a PROPER prefix, per the function's
own documented contract.

Pins: TestDisplayCommitMirror (displayed content must equal committed
content across six reasoning-interleave scenarios — the combination the
replay-parity grid never scripted), TestPartialTagTail contract rows,
TestCloseRun unit pins, and three new interleave rows in the splitter
CASES table. Both fixes are mutation-probed: disabling close_run or
restoring the self-match fails the pins.

tests/test_832_parity.py

@@ -31,6 +31,9 @@ from tests._parity_832 import (
     run_scenario,
     write_fixture,
 )
+from tests._session_helpers import RecordingUI, make_session, scripted_provider
+from turnstone.core.providers._protocol import StreamChunk, UsageInfo
+from turnstone.core.trajectory import Turn
 
 
 def _apply_ruled_deltas(name: str, baseline: dict[str, Any]) -> dict[str, Any]:
@@ -107,3 +110,96 @@ def test_parity(name: str) -> None:
     )
     expected = _apply_ruled_deltas(name, load_fixture(name))
     assert record == expected
+
+
+class TestDisplayCommitMirror:
+    """The mirror LAW (no old-world baselines): with one chunk script,
+    the DISPLAYED content stream and the COMMITTED content must agree.
+
+    Post-fold the drain assembles the committed turn while the consumer
+    drives the display; these scenarios interleave provider-parsed
+    ``reasoning_delta`` with buffered content — the combination the
+    replay-parity grid never scripted, where a live review caught the
+    two lanes disagreeing (display dropped or relabeled the buffered
+    tail the commit kept; with a footer the display showed NOTHING
+    while the commit carried answer + sources).  The consumer's
+    ``close_run`` at the reasoning boundary and ``partial_tag_tail``'s
+    proper-prefix contract are what hold these together.
+    """
+
+    _USAGE = UsageInfo(prompt_tokens=1, completion_tokens=1, total_tokens=2)
+
+    def _mirror(self, chunks: list[StreamChunk]) -> tuple[str, str]:
+        ui = RecordingUI()
+        session = make_session(ui=ui)
+        session._provider = scripted_provider(chunks)
+        session.messages.append(Turn.user("hi"))
+        result = session._stream_response(0)
+        displayed = "".join(d for k, d in ui.events if k == "content")
+        return displayed, result.content
+
+    @pytest.mark.parametrize(
+        ("name", "chunks"),
+        [
+            (
+                "short_content_then_reasoning",
+                [
+                    StreamChunk(content_delta="Short"),
+                    StreamChunk(reasoning_delta="(r)"),
+                    StreamChunk(finish_reason="stop"),
+                ],
+            ),
+            (
+                "short_content_then_reasoning_with_footer",
+                [
+                    StreamChunk(content_delta="Short"),
+                    StreamChunk(reasoning_delta="(r)"),
+                    StreamChunk(finish_reason="stop"),
+                    StreamChunk(info_delta="Sources:\n- example.com"),
+                ],
+            ),
+            (
+                "long_content_then_reasoning",
+                [
+                    StreamChunk(content_delta="A much longer content run here"),
+                    StreamChunk(reasoning_delta="(r)"),
+                    StreamChunk(finish_reason="stop"),
+                ],
+            ),
+            (
+                "content_reasoning_content",
+                [
+                    StreamChunk(content_delta="Before "),
+                    StreamChunk(reasoning_delta="(r)"),
+                    StreamChunk(content_delta="after"),
+                    StreamChunk(finish_reason="stop"),
+                ],
+            ),
+            (
+                "partial_tag_spans_reasoning_boundary",
+                [
+                    StreamChunk(content_delta="Ans<thi"),
+                    StreamChunk(reasoning_delta="(r)"),
+                    StreamChunk(content_delta="nk>hidden</think>done"),
+                    StreamChunk(finish_reason="stop"),
+                ],
+            ),
+            (
+                "complete_tag_before_reasoning_boundary",
+                [
+                    StreamChunk(content_delta="Answer<reasoning>"),
+                    StreamChunk(reasoning_delta="(r)"),
+                    StreamChunk(content_delta=" resumed"),
+                    StreamChunk(finish_reason="stop"),
+                ],
+            ),
+        ],
+    )
+    def test_mirror(self, name: str, chunks: list[StreamChunk]) -> None:
+        stamped = [*chunks]
+        # Ride usage on the finish chunk so the strict gate passes.
+        for i, c in enumerate(stamped):
+            if c.finish_reason:
+                stamped[i] = StreamChunk(finish_reason=c.finish_reason, usage=self._USAGE)
+        displayed, committed = self._mirror(stamped)
+        assert displayed == committed

tests/test_think_tag_split.py

@@ -34,7 +34,11 @@ from tests._session_helpers import make_session, scripted_provider
 from turnstone.core.model_turn import ModelLane
 from turnstone.core.providers import StreamChunk, ToolCallDelta
 from turnstone.core.session import _CancelRef, _StreamTurnConsumer
-from turnstone.core.streaming_text import ThinkTagSplitter, split_inline_reasoning
+from turnstone.core.streaming_text import (
+    ThinkTagSplitter,
+    partial_tag_tail,
+    split_inline_reasoning,
+)
 from turnstone.core.trajectory import Turn
 
 
@@ -161,6 +165,41 @@ CASES = [
         [("reasoning", "y" * 8), ("reasoning", "y" * 12), ("content", "ok")],
         "ok",
     ),
+    # Reasoning-boundary run close (live-caught #832 divergence): a
+    # buffered content tail must emit as CONTENT when a reasoning_delta
+    # arrives, exactly as the drain closes its per-run split there.
+    (
+        "reasoning_boundary_closes_short_content_run",
+        [_c("Short"), StreamChunk(reasoning_delta="(r)"), _FINISH],
+        [("content", "Short"), ("reasoning", "(r)")],
+        "Short",
+    ),
+    (
+        "reasoning_boundary_closes_long_run_tail",
+        [_c("A much longer content run here"), StreamChunk(reasoning_delta="(r)"), _FINISH],
+        [
+            ("content", "A much longer cont"),
+            ("content", "ent run here"),
+            ("reasoning", "(r)"),
+        ],
+        "A much longer content run here",
+    ),
+    (
+        "partial_tag_carries_across_reasoning_boundary",
+        [
+            _c("Ans<thi"),
+            StreamChunk(reasoning_delta="(r)"),
+            _c("nk>hidden</think>done"),
+            _FINISH,
+        ],
+        [
+            ("content", "Ans"),
+            ("reasoning", "(r)"),
+            ("reasoning", "hidden"),
+            ("content", "done"),
+        ],
+        "Ansdone",
+    ),
 ]
 
 
@@ -313,3 +352,65 @@ def test_tool_calls_flush_pending_raw_at_current_state():
     assert result.tool_calls == [
         {"id": "tc1", "type": "function", "function": {"name": "bash", "arguments": "{}"}}
     ]
+
+
+class TestPartialTagTail:
+    """Boundary-contract rows for ``partial_tag_tail``: only a PROPER
+    prefix of a tag is a partial tag.  A complete tag self-matching via
+    ``startswith`` was the live-caught latent bug — the drain then
+    carried a finished ``<reasoning>`` across a run boundary as if it
+    might still grow, relabeling the next run."""
+
+    @pytest.mark.parametrize(
+        ("text", "tail"),
+        [
+            ("Answer<reasoning>", ""),  # complete tag is NOT partial
+            ("Answer<think>", ""),
+            ("orphan</think>", ""),
+            ("Answer</reasoning>", ""),
+            ("Answer<reasonin", "<reasonin"),
+            ("Ans<thi", "<thi"),
+            ("trailing<", "<"),
+            ("no tags here", ""),
+            ("", ""),
+        ],
+    )
+    def test_contract(self, text, tail):
+        assert partial_tag_tail(text) == tail
+
+
+class TestCloseRun:
+    """``close_run`` — the reasoning-boundary run close, mirroring the
+    drain's per-run rule: decided text emits at the current state, only
+    a partial tag prefix carries."""
+
+    def _splitter(self):
+        events = []
+        return ThinkTagSplitter(lambda t, r: events.append((t, r))), events
+
+    def test_plain_tail_emits_as_content(self):
+        sp, events = self._splitter()
+        sp.feed("Short")
+        sp.close_run()
+        assert events == [("Short", False)]
+        assert sp.pending == ""
+
+    def test_partial_tag_tail_is_held(self):
+        sp, events = self._splitter()
+        sp.feed("Ans<thi")
+        sp.close_run()
+        assert events == [("Ans", False)]
+        assert sp.pending == "<thi"
+
+    def test_reasoning_state_tail_emits_as_reasoning(self):
+        sp, events = self._splitter()
+        sp.in_think = True
+        sp.feed("held thought")
+        sp.close_run()
+        assert events == [("held thought", True)]
+        assert sp.pending == ""
+
+    def test_empty_pending_is_noop(self):
+        sp, events = self._splitter()
+        sp.close_run()
+        assert events == []

turnstone/core/session.py

@@ -613,6 +613,17 @@ class _StreamTurnConsumer:
         # Path 1: provider-normalized reasoning_delta.
         if chunk.reasoning_delta:
             self._stop_spinner_once()
+            if not self._splitter.in_think:
+                # Entering the native-reasoning phase closes the content
+                # run EXACTLY as the drain does: the non-tag tail is
+                # content, and only a partial tag prefix carries across
+                # the reasoning block.  Flipping in_think with the tail
+                # still pending relabels buffered content as reasoning at
+                # the next flush — the displayed stream then loses text
+                # the committed turn keeps (live-caught fold divergence;
+                # worst case a short answer displays as NOTHING while the
+                # commit carries it plus a citations footer).
+                self._splitter.close_run()
             self._splitter.in_think = True
             self._path1_reasoning = True
             if s.show_reasoning:

turnstone/core/streaming_text.py

@@ -72,6 +72,28 @@ class ThinkTagSplitter:
             self._emit(self.pending, self.in_think)
             self.pending = ""
 
+    def close_run(self) -> None:
+        """Close the current run at an out-of-band interleave signal.
+
+        For the boundary where a provider-parsed ``reasoning_delta``
+        arrives mid-stream: everything decided emits at the CURRENT
+        state, and only a possible partial-tag tail
+        (:func:`partial_tag_tail`) is held for the next run — the drain
+        closes its per-run split at the same boundary with the same
+        rule, which is what keeps the displayed and committed
+        interpretations of one stream identical there.  A consumer that
+        instead flipped :attr:`in_think` with the tail still pending
+        would relabel buffered content as reasoning at the next flush
+        (the live-caught #832 display/commit divergence).
+        """
+        if not self.pending:
+            return
+        tail = partial_tag_tail(self.pending)
+        closeable = self.pending[: len(self.pending) - len(tail)] if tail else self.pending
+        if closeable:
+            self._emit(closeable, self.in_think)
+        self.pending = tail
+
     def _drain(self) -> None:
         if not self._scan_tags:
             # Nothing to resolve, so nothing to hold: a tag-free contract
@@ -119,7 +141,12 @@ def partial_tag_tail(text: str) -> str:
     limit = min(len(text), ThinkTagSplitter.MAX_TAG_LEN - 1)
     for size in range(limit, 0, -1):
         suffix = text[-size:]
-        if any(tag.startswith(suffix) for tag in ThinkTagSplitter.ALL_TAGS):
+        # Proper prefix only: without the length check a COMPLETE tag
+        # shorter than the longest one self-matches via startswith and
+        # gets carried as a "partial", violating the contract above.
+        if any(
+            len(suffix) < len(tag) and tag.startswith(suffix) for tag in ThinkTagSplitter.ALL_TAGS
+        ):
             return suffix
     return ""