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 ""