From 06ec1a8629262767365dde3e3b0fb588efc3c719 Mon Sep 17 00:00:00 2001 From: Patrick Buckley Date: Wed, 5 Aug 2026 19:47:22 -0700 Subject: [PATCH] fix(832): the boundary carry belongs to the run owner MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The mandated cross-lane interleave angle found the two residual holes in the reasoning-boundary close: the close was gated on not-in_think, so an open inline think block at the boundary never closed and the later state flip relabeled held chain-of-thought as displayed ANSWER text; and the carry parked in the splitter's own pending was re-read under whatever state later flushes hit, relabeling a content-state tail as reasoning. close_run() now closes unconditionally (as the drain does) and RETURNS the partial-tag tail; the consumer owns the carry in a state-immune slot mirroring the drain's separate variable — re-fed when content resumes so a split tag still reassembles, flushed as content at tool, finish, and cancel boundaries, and included in the partial-content rule. The trailing citations footer is now HELD and folded once at stream end over the full answer — structurally the drain's post-loop fold — instead of folding at arrival, which diverged from the commit whenever a lax gateway emitted content after finish. Two non-mirror fixes: the fallback-failure UI line carries the exception class only (its text can embed a credential-bearing base_url; detail goes to the server log, same rule as the re-issue log arm), and a never-armed Stop (creation window, no prior death, zero tokens) writes NO assistant row again — restoring pre-fold semantics; a marker-only row would replay to the model as context on every later turn. Armed zero-token Stops still record their marker. Hygiene riding along: the parity runner zeroes the ladder backoff (the exhaust scenario was sleeping 3.2s of real backoff per suite run, with the retry-notice transform strings updated in step); test_session's porting docstring points at the helper's real module; test_cancel and test_session wrap the shared session factory instead of re-implementing its defaults; arm_session's armed handle is an ArmedHandle with real closed state instead of a MagicMock that satisfies any assertion; and send() derives the tool-call list once for both the persisted mirror and the executed set. All fixes are mutation-probed: re-gating the close, discarding the carry, dropping the promote gate, unredacting the fallback line, and restoring the arrival-time fold each fail their pins. --- tests/_parity_832.py | 5 ++ tests/_session_helpers.py | 2 +- tests/test_832_parity.py | 62 ++++++++++++- tests/test_cancel.py | 46 +++++++--- tests/test_midstream_retry.py | 24 +++++ tests/test_session.py | 22 ++--- tests/test_think_tag_split.py | 16 ++-- turnstone/core/session.py | 148 ++++++++++++++++++++++--------- turnstone/core/streaming_text.py | 25 ++++-- 9 files changed, 264 insertions(+), 86 deletions(-) diff --git a/tests/_parity_832.py b/tests/_parity_832.py index 7e60a063..330718c0 100644 --- a/tests/_parity_832.py +++ b/tests/_parity_832.py @@ -172,6 +172,11 @@ def run_scenario(name: str) -> dict[str, Any]: """ ui = RecordingUI() session = make_session(ui=ui) + # Zero the ladder backoff: a scenario that reaches the mid-stream + # re-issue ladder (no_finish_clean_exhaust) must not sleep real + # exponential delays in a unit run. The retry-notice transform in + # test_832_parity hardcodes the matching "0s" wording. + session._RETRY_BASE_DELAY = 0 session._provider = scripted_provider(SCENARIOS[name]) pre_fold = "msgs" in inspect.signature(type(session)._stream_response).parameters diff --git a/tests/_session_helpers.py b/tests/_session_helpers.py index 8a3bc28d..73c03ef7 100644 --- a/tests/_session_helpers.py +++ b/tests/_session_helpers.py @@ -549,7 +549,7 @@ def arm_session( provider.provider_name = name provider.get_capabilities.return_value = ModelCapabilities() provider.retryable_error_names = retryable - provider._armed_handle = MagicMock() + provider._armed_handle = ArmedHandle() remaining = list(streams) def _create(**kwargs: Any): diff --git a/tests/test_832_parity.py b/tests/test_832_parity.py index 431e6fb9..9ec72548 100644 --- a/tests/test_832_parity.py +++ b/tests/test_832_parity.py @@ -77,7 +77,7 @@ def _apply_ruled_deltas(name: str, baseline: dict[str, Any]) -> dict[str, Any]: [ "info", f"[stream died mid-response (IncompleteStreamError) — retrying in " - f"{2 ** (attempt - 1)}s ({attempt}/2)]", + f"0s ({attempt}/2)]", ], ["stream_discarded", ""], ["thinking_start", ""], @@ -132,6 +132,7 @@ class TestDisplayCommitMirror: def _mirror(self, chunks: list[StreamChunk]) -> tuple[str, str]: ui = RecordingUI() session = make_session(ui=ui) + session._RETRY_BASE_DELAY = 0 session._provider = scripted_provider(chunks) session.messages.append(Turn.user("hi")) result = session._stream_response(0) @@ -193,6 +194,65 @@ class TestDisplayCommitMirror: StreamChunk(finish_reason="stop"), ], ), + # Round-2 classes: the boundary close must run while an + # INLINE think block is open (the CoT-leak half), and the + # cross-boundary carry must survive state flips (the + # relabel half). + ( + "open_inline_think_at_reasoning_boundary", + [ + StreamChunk(content_delta="presecret"), + StreamChunk(reasoning_delta="R"), + StreamChunk(content_delta="answer"), + StreamChunk(finish_reason="stop"), + ], + ), + ( + "split_close_tag_across_reasoning_boundary", + [ + StreamChunk(content_delta="prebody None: diff --git a/tests/test_cancel.py b/tests/test_cancel.py index a45eea6b..82e68250 100644 --- a/tests/test_cancel.py +++ b/tests/test_cancel.py @@ -9,9 +9,8 @@ from unittest.mock import MagicMock, patch import pytest -from tests._session_helpers import arm_session +from tests._session_helpers import arm_session, make_session from turnstone.core.session import ( - ChatSession, GenerationCancelled, _CancelRef, _tool_turn_meta, @@ -100,18 +99,11 @@ class NullUI: def _make_session(ui=None, **kwargs): - """Helper to construct a ChatSession with minimal setup.""" - defaults = dict( - client=MagicMock(), - model="test-model", - ui=ui or NullUI(), - instructions=None, - temperature=0.5, - max_tokens=4096, - tool_timeout=30, - ) - defaults.update(kwargs) - return ChatSession(**defaults) + """Wrap the shared session factory; this suite defaults to its + recording NullUI. The defaults live in + tests/_session_helpers.make_session — duplicating them here is + exactly the drift its docstring warns about.""" + return make_session(ui=ui or NullUI(), **kwargs) class TestCancelEvent: @@ -1272,3 +1264,29 @@ class TestEffectStatusPersistence: turns = reconstruct_turns([sys_row], "ws1") assert turns[0].meta.extra.get("source_meta") == {"watch_name": "x"} assert turns[0].effect_status is None + + +class TestNeverArmedStopLeavesNoRow: + def test_stop_during_creation_persists_nothing(self, tmp_db): + """A Stop landing while creation is still connecting — nothing + armed, zero tokens streamed — must not write an assistant row: + pre-fold no row existed for a turn that never streamed, and a + marker-only row would replay to the model as context on every + later turn. (An ARMED zero-token Stop still records its marker + via record_cancelled_partial — TestCancelDuringStreaming pins + that side.)""" + ui = NullUI() + session = _make_session(ui=ui) + provider = arm_session(session) # provider shell; create scripted below + + def create_cancel_then_fail(**kwargs): + session._cancel_event.set() + raise ConnectionError("connect blew up mid-dial") + + provider.create_streaming = MagicMock(side_effect=create_cancel_then_fail) + session.send("test") + + assert ui.states[-1] == "idle" + assert any("cancelled" in i.lower() for i in ui.infos) + assistant = [m for m in dicts_from_turns(session.messages) if m["role"] == "assistant"] + assert assistant == [] diff --git a/tests/test_midstream_retry.py b/tests/test_midstream_retry.py index 5824d32f..30205855 100644 --- a/tests/test_midstream_retry.py +++ b/tests/test_midstream_retry.py @@ -874,3 +874,27 @@ class TestDebugDumpLatch: session._stream_response(0) session._stream_response(0) assert dump.call_count == 2 + + +class TestFallbackFailureRedaction: + def test_fallback_ui_line_carries_class_name_only(self, tmp_db): + """The fallback-failure info line lands in the browser transcript + and persisted event stream — it carries the exception CLASS, never + its text (a ConnectError's str can embed a credential-bearing + base_url; same rule as the re-issue log arm).""" + ui = RecordingUI() + session = _make_session(ui) + registry = MagicMock() + registry.resolve_binding.side_effect = httpx.ConnectError( + "dial http://user:SECRETKEY@gw.example/v1 failed" + ) + session._registry = registry + from turnstone.core.session import _StreamTurnConsumer + + consumer = _StreamTurnConsumer(session, 0) + result = session._try_fallback_lane("fb", consumer, lambda w: w, 0) + + assert result is None + infos = ui.of("info") + assert any("Fallback fb also failed: ConnectError" in i for i in infos) + assert not any("SECRETKEY" in i for i in infos) diff --git a/tests/test_session.py b/tests/test_session.py index 213847e8..ccf5ce4e 100644 --- a/tests/test_session.py +++ b/tests/test_session.py @@ -16,6 +16,7 @@ from tests._session_helpers import ( FakeAnthropicBlock, as_stream, make_result, + make_session, mock_completion_result, scripted_anthropic_client, scripted_chat_client, @@ -105,19 +106,14 @@ def _make_session( instructions=None, **kwargs, ): - """Helper to construct a ChatSession with minimal setup.""" - client = mock_openai_client or MagicMock() - defaults = dict( - client=client, - model="test-model", - ui=NullUI(), - instructions=instructions, - temperature=0.5, - max_tokens=4096, - tool_timeout=30, + """Wrap the shared session factory with this suite's conveniences + (positional mock client; local recording NullUI default). The + defaults live in tests/_session_helpers.make_session — duplicating + them here is exactly the drift its docstring warns about.""" + kwargs.setdefault("ui", NullUI()) + return make_session( + client=mock_openai_client or MagicMock(), instructions=instructions, **kwargs ) - defaults.update(kwargs) - return ChatSession(**defaults) @contextlib.contextmanager @@ -137,7 +133,7 @@ def _send_with_mocks(session, responses, mock_execute, **extra_patches): session, value is the ``side_effect`` to inject. ``responses`` are ``ModelTurnResult``s (build them with - ``tests._parity_832.make_result``) — the streaming seam's return + ``tests._session_helpers.make_result``) — the streaming seam's return type since #832 folded creation and drain into ``model_turn``. None of these tests care HOW the turn was produced, only that one happened, so they patch the whole ``_stream_response`` seam rather diff --git a/tests/test_think_tag_split.py b/tests/test_think_tag_split.py index 281a2ace..719d866f 100644 --- a/tests/test_think_tag_split.py +++ b/tests/test_think_tag_split.py @@ -391,26 +391,30 @@ class TestCloseRun: def test_plain_tail_emits_as_content(self): sp, events = self._splitter() sp.feed("Short") - sp.close_run() + assert sp.close_run() == "" assert events == [("Short", False)] assert sp.pending == "" - def test_partial_tag_tail_is_held(self): + def test_partial_tag_tail_is_returned_not_held(self): + # The carry's lifetime belongs to the RUN OWNER: the splitter's + # own pending would be re-read under a flipped in_think (the + # round-2 relabel defect), so close_run hands the tail back and + # clears its buffer. sp, events = self._splitter() sp.feed("Ans str: - """THE partial-content rule: flushed content plus the splitter's - carry tail when it is content-state (an in-think tail is reasoning - and stays out) — one closure serving the cancel arms and the - re-issue ladder's dead-partial promotion alike.""" - return "".join(self._content_parts) + ( - self._splitter.pending if not self._splitter.in_think else "" + """THE partial-content rule: flushed content, plus the boundary + carry (content-state by construction), plus the splitter's carry + tail when it is content-state (an in-think tail is reasoning and + stays out) — one closure serving the cancel arms and the re-issue + ladder's dead-partial promotion alike.""" + return ( + "".join(self._content_parts) + + self._boundary_carry + + (self._splitter.pending if not self._splitter.in_think else "") ) + def _flush_terminal_carries(self) -> None: + """Terminal display flush shared by the finish and cancel arms: + the boundary carry emits as CONTENT (its state was fixed when the + run closed — the drain appends its dangling carry to content the + same way), then the splitter's own pending at the current state.""" + if self._boundary_carry: + self._flush_text(self._boundary_carry, False) + self._boundary_carry = "" + self._splitter.flush_pending() + def finish_stream(self) -> None: - """End-of-stream display flush: emit the splitter's held carry. + """End-of-stream display flush: the held carries, then the + trailing citations footer. The drain assembled the canonical content already; without this the DISPLAYED stream is missing its last ≤MAX_TAG_LEN characters - (the partial-tag carry). Success-path only, after the trailing - Stop re-check — the cancel arms flush via - :meth:`record_cancelled_partial`, and a dead attempt deliberately - does not flush (the partial rule reads the carry directly).""" - self._splitter.flush_pending() + (the carries), and the footer fold must run AFTER them — once, + over the full answer, with the shared gate and separator — or + the displayed fold diverges from the drain's post-loop fold. + Success-path only, after the trailing Stop re-check — the cancel + arms flush via :meth:`record_cancelled_partial` and drop the + footer (a cancelled turn commits no drained content to fold + onto).""" + self._flush_terminal_carries() + if self._trailing_info and folds_trailing_info("".join(self._content_parts)): + for info in self._trailing_info: + self._flush_text(TRAILING_INFO_SEPARATOR + info, False) + self._trailing_info = [] def record_cancelled_partial(self) -> None: """Flush, finalize the stream in the UI, and stash the partial for @@ -695,7 +732,7 @@ class _StreamTurnConsumer: if self._session._generation != self._my_generation: return content = self.partial_content() - self._splitter.flush_pending() + self._flush_terminal_carries() self._session.ui.on_stream_end() self._session._cancelled_partial_msg = {"role": "assistant", "content": content} @@ -6339,7 +6376,18 @@ class ChatSession: raise if fb_tracker: fb_tracker.record_failure() - self.ui.on_info(f"[Fallback {alias} also failed: {fb_err}]") + # Class name only in the UI line — the same rule as the + # re-issue log arm: a ConnectError's text can carry a + # credential-bearing base_url verbatim, and this string lands + # in the browser transcript and persisted event stream. The + # full detail goes to the server log. + log.warning( + "fallback.failed", + alias=alias, + error_type=type(fb_err).__name__, + ) + log.debug("fallback failure detail", exc_info=True) + self.ui.on_info(f"[Fallback {alias} also failed: {type(fb_err).__name__}]") return None def _stop_retrying( @@ -7402,13 +7450,16 @@ class ChatSession: ) ) - # Log assistant message to conversation history + # Log assistant message to conversation history. ONE + # binding for the call list: the persisted mirror and the + # executed set below must be the same value by + # construction, not by coincidence. content = result.content - tc = result.tool_calls or None + tool_calls = result.tool_calls or None native = result.turn.native provider_data = json.dumps(list(native.blocks)) if native else None - tool_calls_json: str | None = json.dumps(tc) if tc else None + tool_calls_json: str | None = json.dumps(tool_calls) if tool_calls else None # Save assistant message atomically (content + tool_calls in one row) if content or provider_data is not None or tool_calls_json: @@ -7422,7 +7473,6 @@ class ChatSession: producer=result.producer or None, ) - tool_calls = result.tool_calls or None if not tool_calls: # Did the model stop because we asked it to wind down for a # compaction (cooperative), or because the task is actually @@ -8275,6 +8325,20 @@ class ChatSession: """ if self._generation != my_generation: return + if ( + self._cancelled_partial_msg is None + and last_stream_death is None + and not dead_partial + ): + # Nothing ever streamed this send: a Stop in the + # creation/walk window with no prior armed death. + # Pre-fold, the first creation ran OUTSIDE the promote + # arm and no assistant row was written for a turn that + # never streamed — keep that: a marker-only row would + # replay to the model as context on every later turn + # (round-2 finding; an ARMED zero-token Stop still + # records its marker via record_cancelled_partial). + return cur = self._cancelled_partial_msg if cur is None or (not cur.get("content") and dead_partial): self._cancelled_partial_msg = { diff --git a/turnstone/core/streaming_text.py b/turnstone/core/streaming_text.py index 4131e6f4..333d6688 100644 --- a/turnstone/core/streaming_text.py +++ b/turnstone/core/streaming_text.py @@ -72,27 +72,34 @@ class ThinkTagSplitter: self._emit(self.pending, self.in_think) self.pending = "" - def close_run(self) -> None: + def close_run(self) -> str: """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 + state, and a possible partial-tag tail + (:func:`partial_tag_tail`) is RETURNED to the caller — 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). + interpretations of one stream identical there. The carry's + lifetime belongs to the run owner, NOT to :attr:`pending`: the + tail was cut in the closing run's state, while ``pending`` is + read under whatever state later flushes hit (``in_think`` flips + across the reasoning block), which would relabel a content-state + carry as reasoning — both halves of the live-caught #832 + display/commit divergence. The caller re-feeds the carry when + content resumes (reassembling a tag the server split across the + block) or flushes it at its original state at a terminal + boundary, mirroring the drain's separate carry variable. """ if not self.pending: - return + 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 + self.pending = "" + return tail def _drain(self) -> None: if not self._scan_tags: