From 6212783e23f997b082fcde7f8bf91e7dbd6c03de Mon Sep 17 00:00:00 2001 From: Patrick Buckley Date: Wed, 5 Aug 2026 18:32:41 -0700 Subject: [PATCH] =?UTF-8?q?fix(832):=20close=20the=20content=20run=20at=20?= =?UTF-8?q?a=20reasoning=5Fdelta=20boundary=20=E2=80=94=20display=20must?= =?UTF-8?q?=20mirror=20the=20drain?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 / 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 | 96 ++++++++++++++++++++++++++++ tests/test_think_tag_split.py | 103 ++++++++++++++++++++++++++++++- turnstone/core/session.py | 11 ++++ turnstone/core/streaming_text.py | 29 ++++++++- 4 files changed, 237 insertions(+), 2 deletions(-) diff --git a/tests/test_832_parity.py b/tests/test_832_parity.py index 0c2b8a59..431e6fb9 100644 --- a/tests/test_832_parity.py +++ b/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 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 diff --git a/tests/test_think_tag_split.py b/tests/test_think_tag_split.py index 71757918..281a2ace 100644 --- a/tests/test_think_tag_split.py +++ b/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("Anshiddendone"), + _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 ```` across a run boundary as if it + might still grow, relabeling the next run.""" + + @pytest.mark.parametrize( + ("text", "tail"), + [ + ("Answer", ""), # complete tag is NOT partial + ("Answer", ""), + ("orphan", ""), + ("Answer", ""), + ("Answer 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 ""