fix(832): the boundary carry belongs to the run owner

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

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

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):

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="pre<think>secret"),
+                    StreamChunk(reasoning_delta="R"),
+                    StreamChunk(content_delta="answer"),
+                    StreamChunk(finish_reason="stop"),
+                ],
+            ),
+            (
+                "split_close_tag_across_reasoning_boundary",
+                [
+                    StreamChunk(content_delta="pre<think>body</thi"),
+                    StreamChunk(reasoning_delta="R"),
+                    StreamChunk(content_delta="nk>post"),
+                    StreamChunk(finish_reason="stop"),
+                ],
+            ),
+            (
+                "unresolved_partial_tag_at_finish",
+                [
+                    StreamChunk(content_delta="Ans<thi"),
+                    StreamChunk(reasoning_delta="R"),
+                    StreamChunk(finish_reason="stop"),
+                ],
+            ),
+            (
+                "partial_tag_only_with_footer",
+                [
+                    StreamChunk(content_delta="<thi"),
+                    StreamChunk(reasoning_delta="R"),
+                    StreamChunk(finish_reason="stop"),
+                    StreamChunk(info_delta="Sources:\n- example.com"),
+                ],
+            ),
+            # Lax-gateway shape: content AFTER a post-finish footer —
+            # the fold must run once at stream end over the FULL answer
+            # (the drain's post-loop fold), never at footer arrival.
+            (
+                "late_content_after_footer",
+                [
+                    StreamChunk(content_delta="ans"),
+                    StreamChunk(finish_reason="stop"),
+                    StreamChunk(info_delta="Sources: s"),
+                    StreamChunk(content_delta="LATE"),
+                ],
+            ),
+            (
+                "footer_arrives_before_any_content",
+                [
+                    StreamChunk(finish_reason="stop"),
+                    StreamChunk(info_delta="Sources: s"),
+                    StreamChunk(content_delta="LATE"),
+                ],
+            ),
         ],
     )
     def test_mirror(self, name: str, chunks: list[StreamChunk]) -> None:

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 == []

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)

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

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<thi")
-        sp.close_run()
+        assert sp.close_run() == "<thi"
         assert events == [("Ans", False)]
-        assert sp.pending == "<thi"
+        assert sp.pending == ""
 
     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 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 sp.close_run() == ""
         assert events == []

turnstone/core/session.py

@@ -490,6 +490,12 @@ class _StreamTurnConsumer:
         self._path1_reasoning = False
         self._finish_seen = False
         self._saw_chunk = False
+        # Content-state text held across a reasoning block (the run-close
+        # carry) and post-finish footer parts held for the end-of-stream
+        # fold — both owned HERE, outside the splitter, so in_think flips
+        # cannot relabel them (mirrors the drain's separate carry).
+        self._boundary_carry = ""
+        self._trailing_info: list[str] = []
         # Per-attempt: a re-issued attempt's usage must never max-merge
         # onto the dead attempt's (the accumulator was attempt-local in
         # the pre-fold consumer too).
@@ -613,17 +619,19 @@ 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()
+            # Entering the native-reasoning phase closes the current run
+            # UNCONDITIONALLY, exactly as the drain does: decided text
+            # emits at its own state (content-state tail as content,
+            # in-think tail as reasoning — gating this on in_think left
+            # an open inline think block's carry to be relabeled as
+            # displayed ANSWER text by the later state flip, the CoT-leak
+            # half of the round-2 findings), and only a partial tag
+            # prefix carries across the block.  The carry lives on the
+            # CONSUMER, not in the splitter's pending — pending is read
+            # under whatever state later flushes hit, and the in_think
+            # flip below would relabel a content-state carry as
+            # reasoning (the other half).
+            self._boundary_carry += self._splitter.close_run()
             self._splitter.in_think = True
             self._path1_reasoning = True
             if s.show_reasoning:
@@ -635,55 +643,84 @@ class _StreamTurnConsumer:
             if self._path1_reasoning:
                 self._path1_reasoning = False
                 self._splitter.in_think = False
+            if self._boundary_carry:
+                # Content resumed after the reasoning block: the carry
+                # prefixes the new text so a tag the server split across
+                # the block reassembles (the drain prefixes its carried
+                # tail onto the next run the same way).
+                self._splitter.feed(self._boundary_carry)
+                self._boundary_carry = ""
             self._splitter.feed(chunk.content_delta)
 
         # Tool-call deltas: display-side this is only a run boundary —
         # accumulation lives in the drain.
         if chunk.tool_call_deltas:
             self._stop_spinner_once()
+            if self._boundary_carry:
+                # A partial tag never spans a TOOL boundary (the drain's
+                # rule): the carry is content-state text — flush it as
+                # content, whatever in_think says now.
+                self._flush_text(self._boundary_carry, False)
+                self._boundary_carry = ""
             self._splitter.flush_pending()
             self._splitter.in_think = False
 
         if chunk.info_delta:
             self._stop_spinner_once()
             if self._finish_seen:
-                # Trailing citations footer → CONTENT, per the drain's
-                # conditional fold — gate and separator are the SHARED
-                # module-level pair beside ``drain_stream``, so the two
-                # mirrors cannot drift.  The generation is over (finish
-                # seen), so the splitter's carry is final answer text,
-                # never a partial tag — flush it FIRST or the footer
-                # renders spliced into the middle of the answer's last
-                # characters.  Appended via the accumulator too, so the
-                # partial rule and the blankness test stay consistent
-                # with the committed content.
-                self._splitter.flush_pending()
-                if folds_trailing_info("".join(self._content_parts)):
-                    self._flush_text(TRAILING_INFO_SEPARATOR + chunk.info_delta, False)
+                # Post-finish info is the trailing citations footer: HELD
+                # and folded once at stream end after ALL content
+                # (:meth:`finish_stream`), exactly the drain's post-loop
+                # fold.  Folding at arrival diverges from the commit when
+                # a lax gateway emits content after finish — the fold
+                # gate must judge the FULL answer, and the footer must
+                # follow it.
+                self._trailing_info.append(chunk.info_delta)
             else:
                 s.ui.on_info(f"{GRAY}{chunk.info_delta}{RESET}")
 
     # -- partial preservation --------------------------------------------------
 
     def partial_content(self) -> 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 = {

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: