diff --git a/tests/_parity_832.py b/tests/_parity_832.py
index 41d7e629..18d5ae50 100644
--- a/tests/_parity_832.py
+++ b/tests/_parity_832.py
@@ -95,6 +95,51 @@ class ArmedHandle:
         self.closed = True
 
 
+def arm_session(
+    session: Any,
+    *streams: Any,
+    retryable: frozenset[str] = frozenset({"IncompleteStreamError"}),
+    name: str = "openai-compatible",
+) -> MagicMock:
+    """Install a sequential multi-turn armed provider fake on *session*.
+
+    Each ``create_streaming`` call serves the next element of *streams*:
+    an iterable/generator is armed (a closeable sentinel appended to
+    ``cancel_ref`` — the eager append every real adapter performs, which
+    the fold's creation-vs-midstream classifier keys on) and returned to
+    be consumed once; an EXCEPTION instance is raised at create time
+    WITHOUT arming — a creation-phase failure the per-lane ladder owns.
+    Calls beyond the script fail loudly (the pre-fold lax consumer used
+    to absorb an exhausted iterator as a silent empty turn; the strict
+    finish gate rejects that now, so an under-scripted test must say so).
+
+    Title generation is latched off — with a provider-LEVEL fake the
+    best-effort title lane would otherwise consume the first script
+    before the main loop ran.
+    """
+    session._title_generated = True
+    provider = MagicMock()
+    provider.provider_name = name
+    provider.get_capabilities.return_value = ModelCapabilities()
+    provider.retryable_error_names = retryable
+    provider._armed_handle = MagicMock()
+    remaining = list(streams)
+
+    def _create(**kwargs: Any):
+        assert remaining, "arm_session: script exhausted — send looped for more turns than scripted"
+        nxt = remaining.pop(0)
+        if isinstance(nxt, BaseException):
+            raise nxt
+        ref = kwargs.get("cancel_ref")
+        if ref is not None:
+            ref.append(provider._armed_handle)
+        return iter(nxt) if not hasattr(nxt, "__next__") else nxt
+
+    provider.create_streaming = MagicMock(side_effect=_create)
+    session._provider = provider
+    return provider
+
+
 def scripted_provider(chunks: list[StreamChunk]) -> MagicMock:
     """Provider fake replaying *chunks*, arming ``cancel_ref`` eagerly.
 
diff --git a/tests/test_cooperative_compaction.py b/tests/test_cooperative_compaction.py
index 8afc2d26..001e9157 100644
--- a/tests/test_cooperative_compaction.py
+++ b/tests/test_cooperative_compaction.py
@@ -20,6 +20,7 @@ from unittest.mock import MagicMock, patch
 
 import pytest
 
+from tests._parity_832 import make_result
 from tests._session_helpers import make_session
 from turnstone.core.session import (
     COMPACTION_SOURCE,
@@ -204,10 +205,7 @@ class TestCompactionLatch:
         session._compaction_advised = True  # stale latch from a prior turn
 
         with (
-            patch.object(session, "_create_stream_with_retry", return_value=iter([])),
-            patch.object(
-                session, "_stream_response", return_value={"role": "assistant", "content": "done"}
-            ),
+            patch.object(session, "_stream_response", return_value=make_result("done")),
             patch.object(session, "_full_messages", return_value=[]),
             patch.object(session, "_update_token_table"),
             patch.object(session, "_print_status_line"),
@@ -249,11 +247,10 @@ class TestEndOfTurnAutoResume:
             calls["n"] += 1
             if calls["n"] == 1:
                 session._compaction_advised = True  # advisory fired this turn
-                return {"role": "assistant", "content": "paused; plan recorded"}
-            return {"role": "assistant", "content": "done"}
+                return make_result("paused; plan recorded")
+            return make_result("done")
 
         with (
-            patch.object(session, "_create_stream_with_retry", return_value=iter([])),
             patch.object(session, "_stream_response", side_effect=stream),
             patch.object(session, "_full_messages", return_value=[]),
             patch.object(session, "_update_token_table"),
@@ -282,10 +279,7 @@ class TestEndOfTurnAutoResume:
         session._compaction_advised = False
 
         with (
-            patch.object(session, "_create_stream_with_retry", return_value=iter([])),
-            patch.object(
-                session, "_stream_response", return_value={"role": "assistant", "content": "done"}
-            ),
+            patch.object(session, "_stream_response", return_value=make_result("done")),
             patch.object(session, "_full_messages", return_value=[]),
             patch.object(session, "_update_token_table"),
             patch.object(session, "_print_status_line"),
@@ -321,11 +315,10 @@ class TestEndOfTurnAutoResume:
             calls["n"] += 1
             if calls["n"] == 1:
                 session._compaction_advised = True  # advised stop
-                return {"role": "assistant", "content": "paused"}
-            return {"role": "assistant", "content": "done"}
+                return make_result("paused")
+            return make_result("done")
 
         with (
-            patch.object(session, "_create_stream_with_retry", return_value=iter([])),
             patch.object(session, "_stream_response", side_effect=stream),
             patch.object(session, "_full_messages", return_value=[]),
             patch.object(session, "_update_token_table"),
@@ -365,14 +358,13 @@ class TestEndOfTurnAutoResume:
             n["i"] += 1
             if n["i"] == 1:
                 session._compaction_advised = True  # advisory fired this turn
-                return {"role": "assistant", "content": "pausing to compact"}
-            return {"role": "assistant", "content": "all done"}
+                return make_result("pausing to compact")
+            return make_result("all done")
 
         def est(*_a, **_k):
             return 9_999 if n["i"] <= 1 else 10  # over threshold only on the stop turn
 
         with (
-            patch.object(session, "_create_stream_with_retry", return_value=iter([])),
             patch.object(session, "_stream_response", side_effect=stream),
             patch.object(session, "_full_messages", return_value=[]),
             patch.object(session, "_update_token_table"),
@@ -454,11 +446,10 @@ class TestCompactBeforeTruncate:
         def stream(*_a, **_k):
             n["i"] += 1
             if n["i"] == 1:
-                return {"role": "assistant", "content": "", "tool_calls": [tc]}
-            return {"role": "assistant", "content": "done"}
+                return make_result("", tool_calls=[tc])
+            return make_result("done")
 
         with (
-            patch.object(session, "_create_stream_with_retry", return_value=iter([])),
             patch.object(session, "_stream_response", side_effect=stream),
             patch.object(session, "_execute_tools", return_value=([("call_1", "out")], "")),
             patch.object(session, "_full_messages", return_value=[]),
@@ -493,11 +484,10 @@ class TestCompactBeforeTruncate:
         def stream(*_a, **_k):
             n["i"] += 1
             if n["i"] == 1:
-                return {"role": "assistant", "content": "", "tool_calls": [tc]}
-            return {"role": "assistant", "content": "done"}
+                return make_result("", tool_calls=[tc])
+            return make_result("done")
 
         with (
-            patch.object(session, "_create_stream_with_retry", return_value=iter([])),
             patch.object(session, "_stream_response", side_effect=stream),
             patch.object(session, "_execute_tools", return_value=([("call_1", "out")], "")),
             patch.object(session, "_full_messages", return_value=[]),
@@ -534,11 +524,10 @@ class TestCompactBeforeTruncate:
         def stream(*_a, **_k):
             n["i"] += 1
             if n["i"] == 1:
-                return {"role": "assistant", "content": "", "tool_calls": [tc]}
-            return {"role": "assistant", "content": "done"}
+                return make_result("", tool_calls=[tc])
+            return make_result("done")
 
         with (
-            patch.object(session, "_create_stream_with_retry", return_value=iter([])),
             patch.object(session, "_stream_response", side_effect=stream),
             patch.object(session, "_execute_tools", return_value=([("call_1", "out")], "")),
             patch.object(session, "_full_messages", return_value=[]),
@@ -1095,7 +1084,7 @@ class TestProactivePreSend:
 
         def fake_stream_response(*_args, **_kwargs):
             order.append("stream")
-            return {"role": "assistant", "content": "done"}
+            return make_result("done")
 
         with (
             # 9999 > hard (9000) → compaction is owed at send time.
@@ -1475,7 +1464,7 @@ class TestChunkerOverflowSplit:
         with (
             patch.object(session, "_estimated_prompt_tokens", return_value=10),  # under hard
             patch.object(session, "_check_metacognitive_nudge", return_value=None),
-            patch.object(session, "_create_stream_with_retry", side_effect=cancel_midstream),
+            patch.object(session, "_stream_response", side_effect=cancel_midstream),
             patch.object(session, "_full_messages", return_value=[]),
             patch.object(session, "_update_token_table"),
             patch.object(session, "_print_status_line"),
@@ -2330,7 +2319,12 @@ class TestOrphanedCompactionRetirement:
     def test_summary_call_registers_abortable_stream(self, session):
         """Each summary attempt passes a fresh _CancelRef so cancel() can
         close the in-flight summary HTTP stream — Stop during compaction
-        aborts the blocked read instead of waiting out a model call."""
+        aborts the blocked read instead of waiting out a model call.
+
+        Freshness is pinned per CALL rather than against a session-wide
+        register: #832 retired the shared ``_cancel_ref`` slot outright, so
+        "not the shared ref" is no longer a statement anything can violate —
+        "every call mints its own, and no session-wide slot exists" is."""
         from turnstone.core.session import _CancelRef
 
         seen: list[object] = []
@@ -2344,9 +2338,11 @@ class TestOrphanedCompactionRetirement:
 
         with patch.object(session, "_utility_completion", side_effect=fake_uc):
             assert session._summarize_once("sys", "body") == "dense"
-        assert len(seen) == 1
-        assert isinstance(seen[0], _CancelRef)
-        assert seen[0] is not session._cancel_ref  # scoped, never the shared ref
+            assert session._summarize_once("sys", "body") == "dense"
+        assert len(seen) == 2
+        assert all(isinstance(ref, _CancelRef) for ref in seen)
+        assert seen[0] is not seen[1]  # scoped to its call, never reused
+        assert not hasattr(session, "_cancel_ref")  # and no shared register
 
     def test_cancel_ref_aborted_property_tracks_event(self, session):
         """model_turn consults cancel_ref.aborted to suppress drain retries
diff --git a/tests/test_idle_nudge_wake_integration.py b/tests/test_idle_nudge_wake_integration.py
index 11e445df..f022ba6c 100644
--- a/tests/test_idle_nudge_wake_integration.py
+++ b/tests/test_idle_nudge_wake_integration.py
@@ -2,8 +2,8 @@
 
 Drives a *real* :class:`SessionManager` + a *real* :class:`ChatSession`
 + a *real* :class:`IdleNudgeWatcher` end-to-end.  The only stub is the
-LLM provider (patched ``_create_stream_with_retry``); every other layer
-is production code:
+model turn (patched ``_stream_response``, returning a canned
+``ModelTurnResult``); every other layer is production code:
 
 * ``SessionManager.set_state`` snapshotting + iterating subscribers
 * ``IdleNudgeWatcher._on_state`` peeking the queue
@@ -29,6 +29,7 @@ from unittest.mock import MagicMock, patch
 import pytest
 
 from tests._helpers import wait_until as _wait_until
+from tests._parity_832 import make_result
 from tests.test_session_manager import FakeStorage
 from turnstone.core import session_worker
 from turnstone.core.idle_nudge_watcher import IdleNudgeWatcher, wake_workstream_if_pending
@@ -234,12 +235,7 @@ def test_idle_event_through_real_session_manager_drives_wake_send(real_mgr, tmp_
         # without any real provider.  We patch on the just-built ChatSession;
         # the patches are reverted by the `with` block.
         with (
-            patch.object(ws.session, "_create_stream_with_retry", return_value=iter([])),
-            patch.object(
-                ws.session,
-                "_stream_response",
-                return_value={"role": "assistant", "content": "ok"},
-            ),
+            patch.object(ws.session, "_stream_response", return_value=make_result(content="ok")),
             patch.object(ws.session, "_update_token_table"),
             patch.object(ws.session, "_print_status_line"),
             patch.object(ws.session, "_visible_memory_count", return_value=0),
@@ -341,12 +337,7 @@ def test_watch_fire_on_already_idle_session_drives_wake_send(real_mgr, tmp_db):
     )
 
     with (
-        patch.object(ws.session, "_create_stream_with_retry", return_value=iter([])),
-        patch.object(
-            ws.session,
-            "_stream_response",
-            return_value={"role": "assistant", "content": "ok"},
-        ),
+        patch.object(ws.session, "_stream_response", return_value=make_result(content="ok")),
         patch.object(ws.session, "_update_token_table"),
         patch.object(ws.session, "_print_status_line"),
         patch.object(ws.session, "_visible_memory_count", return_value=0),
@@ -450,11 +441,8 @@ def test_coord_idle_with_active_children_emits_envelope_via_real_managers(coord_
         coord.session.messages.append(turn_from_dict({"role": "assistant", "content": "ok"}))
 
         with (
-            patch.object(coord.session, "_create_stream_with_retry", return_value=iter([])),
             patch.object(
-                coord.session,
-                "_stream_response",
-                return_value={"role": "assistant", "content": "ack"},
+                coord.session, "_stream_response", return_value=make_result(content="ack")
             ),
             patch.object(coord.session, "_full_messages", return_value=[]),
             patch.object(coord.session, "_update_token_table"),
@@ -542,11 +530,8 @@ def test_coord_idle_with_children_and_open_tasks_delivers_both(coord_mgr, tmp_db
         coord.session.messages.append(turn_from_dict({"role": "assistant", "content": "ok"}))
 
         with (
-            patch.object(coord.session, "_create_stream_with_retry", return_value=iter([])),
             patch.object(
-                coord.session,
-                "_stream_response",
-                return_value={"role": "assistant", "content": "ack"},
+                coord.session, "_stream_response", return_value=make_result(content="ack")
             ),
             patch.object(coord.session, "_full_messages", return_value=[]),
             patch.object(coord.session, "_update_token_table"),
@@ -647,11 +632,8 @@ def test_coord_idle_with_open_tasks_and_no_children_omits_children_content(coord
         coord.session.messages.append(turn_from_dict({"role": "assistant", "content": "ok"}))
 
         with (
-            patch.object(coord.session, "_create_stream_with_retry", return_value=iter([])),
             patch.object(
-                coord.session,
-                "_stream_response",
-                return_value={"role": "assistant", "content": "ack"},
+                coord.session, "_stream_response", return_value=make_result(content="ack")
             ),
             patch.object(coord.session, "_full_messages", return_value=[]),
             patch.object(coord.session, "_update_token_table"),
@@ -757,11 +739,8 @@ def test_stop_latch_survives_the_liveness_wake(coord_mgr, tmp_db):
         assert coord.session._metacog_state.get("idle_tasks") is None
 
         with (
-            patch.object(coord.session, "_create_stream_with_retry", return_value=iter([])),
             patch.object(
-                coord.session,
-                "_stream_response",
-                return_value={"role": "assistant", "content": "ack"},
+                coord.session, "_stream_response", return_value=make_result(content="ack")
             ),
             patch.object(coord.session, "_full_messages", return_value=[]),
             patch.object(coord.session, "_update_token_table"),
@@ -860,11 +839,8 @@ def test_coord_idle_emitted_from_worker_thread_still_wakes(coord_mgr, tmp_db):
         coord.session.messages.append(turn_from_dict({"role": "assistant", "content": "ok"}))
 
         with (
-            patch.object(coord.session, "_create_stream_with_retry", return_value=iter([])),
             patch.object(
-                coord.session,
-                "_stream_response",
-                return_value={"role": "assistant", "content": "ack"},
+                coord.session, "_stream_response", return_value=make_result(content="ack")
             ),
             patch.object(coord.session, "_full_messages", return_value=[]),
             patch.object(coord.session, "_update_token_table"),
@@ -943,10 +919,7 @@ def test_wake_delivery_contains_generation_cancelled(tmp_db):
 def _patch_llm_surface(session: Any) -> tuple[Any, ...]:
     """The file's standard LLM-stub patch set, for make_chat_session tests."""
     return (
-        patch.object(session, "_create_stream_with_retry", return_value=iter([])),
-        patch.object(
-            session, "_stream_response", return_value={"role": "assistant", "content": "ok"}
-        ),
+        patch.object(session, "_stream_response", return_value=make_result(content="ok")),
         patch.object(session, "_update_token_table"),
         patch.object(session, "_print_status_line"),
         patch.object(session, "_visible_memory_count", return_value=0),
@@ -967,7 +940,7 @@ def test_wake_channel_survives_real_seam_drains_and_delivers_via_wake(tmp_db):
     session._nudge_queue.enqueue("idle_children", "kids waiting", "wake")
 
     p = _patch_llm_surface(session)
-    with p[0], p[1], p[2], p[3], p[4], p[5]:
+    with p[0], p[1], p[2], p[3], p[4]:
         # Real user-seam drain: appends any drained entry as a system
         # turn — a wake-channel entry must neither drain nor render.
         session._emit_pending_user_nudges()
@@ -1053,7 +1026,7 @@ def test_quiet_ride_along_still_delivers_when_wake_proceeds(tmp_db):
     session._nudge_queue.enqueue("idle_children", "kids waiting", "wake")
 
     p = _patch_llm_surface(session)
-    with p[0], p[1], p[2], p[3], p[4], p[5]:
+    with p[0], p[1], p[2], p[3], p[4]:
         session.deliver_wake_nudge_from_queue()
 
     msgs = dicts_from_turns(session.messages)
@@ -1089,7 +1062,7 @@ def test_interjection_handoff_delivers_externals_and_drops_only_idle_nudges(tmp_
     session.queue_message("pivot: focus on the flaky login test")
 
     p = _patch_llm_surface(session)
-    with p[0], p[1], p[2], p[3], p[4], p[5]:
+    with p[0], p[1], p[2], p[3], p[4]:
         session.deliver_wake_nudge_from_queue()
 
     msgs = dicts_from_turns(session.messages)
@@ -1160,11 +1133,8 @@ def test_queued_interjection_owns_the_idle_seam(coord_mgr, tmp_db):
         coord.session.messages.append(turn_from_dict({"role": "assistant", "content": "ok"}))
 
         with (
-            patch.object(coord.session, "_create_stream_with_retry", return_value=iter([])),
             patch.object(
-                coord.session,
-                "_stream_response",
-                return_value={"role": "assistant", "content": "ack"},
+                coord.session, "_stream_response", return_value=make_result(content="ack")
             ),
             patch.object(coord.session, "_full_messages", return_value=[]),
             patch.object(coord.session, "_update_token_table"),
diff --git a/tests/test_mcp_client.py b/tests/test_mcp_client.py
index 65357a17..c3d49ced 100644
--- a/tests/test_mcp_client.py
+++ b/tests/test_mcp_client.py
@@ -771,27 +771,27 @@ class TestSessionIntegration:
         assert "Malformed tool call" in session.ui.on_error.call_args[0][0]
 
     def test_ensure_tool_call_ids_dict(self, tmp_db):
-        """_ensure_tool_call_ids fills empty IDs on streaming-style dict."""
-        from turnstone.core.session import ChatSession
+        """ensure_tool_call_ids fills empty IDs on streaming-style dict."""
+        from turnstone.core.model_turn import ensure_tool_call_ids
 
         tool_calls_acc = {
             0: {"id": "", "function": {"name": "bash", "arguments": "{}"}},
             1: {"id": "", "function": {"name": "read_file", "arguments": "{}"}},
         }
-        ChatSession._ensure_tool_call_ids(tool_calls_acc)
+        ensure_tool_call_ids(tool_calls_acc)
         ids = [tc["id"] for tc in tool_calls_acc.values()]
         assert all(id_.startswith("call_") for id_ in ids)
         assert len(set(ids)) == 2  # unique
 
     def test_ensure_tool_call_ids_list(self, tmp_db):
-        """_ensure_tool_call_ids fills empty IDs on list (agent path)."""
-        from turnstone.core.session import ChatSession
+        """ensure_tool_call_ids fills empty IDs on list (agent path)."""
+        from turnstone.core.model_turn import ensure_tool_call_ids
 
         tool_calls = [
             {"id": None, "function": {"name": "bash", "arguments": "{}"}},
             {"id": "call_existing", "function": {"name": "bash", "arguments": "{}"}},
         ]
-        ChatSession._ensure_tool_call_ids(tool_calls)
+        ensure_tool_call_ids(tool_calls)
         assert tool_calls[0]["id"].startswith("call_")
         assert tool_calls[1]["id"] == "call_existing"  # preserved
 
diff --git a/tests/test_midstream_retry.py b/tests/test_midstream_retry.py
index ec8d45b0..f129083c 100644
--- a/tests/test_midstream_retry.py
+++ b/tests/test_midstream_retry.py
@@ -2,7 +2,7 @@
 
 A wire death DURING body iteration surfaces after the request already
 returned its stream handle, so neither the SDK's ``max_retries`` nor the
-creation-time ``_try_stream`` ladder ever sees it.  These tests drive
+creation-time per-lane ladder (``_model_turn_with_retry``) ever sees it.  These tests drive
 ``ChatSession.send()`` with scripted provider streams and pin the retry
 loop's contract: bounded re-issue on the normalized retryable shape, the
 dead attempt finalized client-side before the retry notice
@@ -18,12 +18,12 @@ retry pays real exponential backoff).
 """
 
 import logging
-from types import SimpleNamespace
 from unittest.mock import MagicMock, patch
 
 import httpx
 import pytest
 
+from tests._parity_832 import arm_session
 from tests._session_helpers import NullUI, RecordingUI, make_session
 from turnstone.core.memory import load_last_error
 from turnstone.core.providers import IncompleteStreamError, StreamChunk, UsageInfo
@@ -83,8 +83,8 @@ class TestMidStreamRetry:
             _dying_stream(first_text, exc=httpx.ReadError("[SSL] record layer failure")),
             _good_stream("Hello world"),
         ]
+        create = arm_session(session, *streams).create_streaming
         with (
-            patch.object(session, "_create_stream_with_retry", side_effect=streams) as create,
             patch.object(session, "_full_messages", return_value=[]),
             caplog.at_level(logging.WARNING, logger="turnstone.core.session"),
         ):
@@ -135,8 +135,8 @@ class TestMidStreamRetry:
         streams = [
             _dying_stream("a", exc=httpx.ReadError("[SSL] record layer failure")) for _ in range(3)
         ]
+        create = arm_session(session, *streams).create_streaming
         with (
-            patch.object(session, "_create_stream_with_retry", side_effect=streams) as create,
             patch.object(session, "_full_messages", return_value=[]),
             caplog.at_level(logging.INFO, logger="turnstone.core.session"),
             pytest.raises(IncompleteStreamError, match="ReadError"),
@@ -174,12 +174,10 @@ class TestMidStreamRetry:
             session.cancel()
             real_backoff(delay, my_generation)
 
+        create = arm_session(
+            session, _dying_stream("Hel", exc=httpx.ReadError("wire died"))
+        ).create_streaming
         with (
-            patch.object(
-                session,
-                "_create_stream_with_retry",
-                side_effect=[_dying_stream("Hel", exc=httpx.ReadError("wire died"))],
-            ) as create,
             patch.object(session, "_backoff_or_cancelled", side_effect=cancel_then_backoff),
             patch.object(session, "_full_messages", return_value=[]),
         ):
@@ -210,8 +208,8 @@ class TestMidStreamRetry:
             # object (the identity signal the wrapper keys on).
             session.client = MagicMock()
 
+        create = arm_session(session, *streams).create_streaming
         with (
-            patch.object(session, "_create_stream_with_retry", side_effect=streams) as create,
             patch.object(session, "_refresh_model_from_registry", side_effect=swap_binding),
             patch.object(session, "_full_messages", return_value=[]),
             patch.object(
@@ -234,8 +232,8 @@ class TestMidStreamRetry:
             _dying_stream("y", exc=httpx.ReadError("wire died again")),
             _good_stream("ok"),
         ]
+        create = arm_session(session, *streams).create_streaming
         with (
-            patch.object(session, "_create_stream_with_retry", side_effect=streams) as create,
             patch.object(session, "_full_messages", return_value=[]),
             patch.object(
                 session,
@@ -259,8 +257,8 @@ class TestMidStreamRetry:
             _dying_stream(exc=httpx.ReadError("pre-token wire death")),
             _good_stream("ok"),
         ]
+        arm_session(session, *streams)
         with (
-            patch.object(session, "_create_stream_with_retry", side_effect=streams),
             patch.object(session, "_full_messages", return_value=[]),
         ):
             session.send("test")
@@ -284,12 +282,8 @@ class TestMidStreamRetry:
             session.cancel()
             real_backoff(delay, my_generation)
 
+        arm_session(session, _dying_stream(exc=httpx.ReadError("died before first token")))
         with (
-            patch.object(
-                session,
-                "_create_stream_with_retry",
-                side_effect=[_dying_stream(exc=httpx.ReadError("died before first token"))],
-            ),
             patch.object(session, "_backoff_or_cancelled", side_effect=cancel_then_backoff),
             patch.object(session, "_full_messages", return_value=[]),
         ):
@@ -318,8 +312,8 @@ class TestMidStreamRetry:
             _dying_stream("Hel", exc=httpx.ReadError("wire died")),
             second_stream(),
         ]
+        arm_session(session, *streams)
         with (
-            patch.object(session, "_create_stream_with_retry", side_effect=streams),
             patch.object(session, "_full_messages", return_value=[]),
         ):
             session.send("test")
@@ -342,8 +336,8 @@ class TestMidStreamRetry:
             # ends cleanly under transport_guarded's post-finish tolerance.
             session.cancel()
 
+        arm_session(session, gen())
         with (
-            patch.object(session, "_create_stream_with_retry", side_effect=[gen()]),
             patch.object(session, "_full_messages", return_value=[]),
         ):
             session.send("test")
@@ -360,12 +354,8 @@ class TestMidStreamRetry:
     def test_keyboard_interrupt_finalizes_dead_attempt(self, tmp_db, caplog):
         ui = RecordingUI()
         session = _make_session(ui)
+        arm_session(session, _dying_stream("Hel", exc=KeyboardInterrupt()))
         with (
-            patch.object(
-                session,
-                "_create_stream_with_retry",
-                side_effect=[_dying_stream("Hel", exc=KeyboardInterrupt())],
-            ),
             patch.object(session, "_full_messages", return_value=[]),
             caplog.at_level(logging.INFO, logger="turnstone.core.session"),
             pytest.raises(KeyboardInterrupt),
@@ -378,53 +368,43 @@ class TestMidStreamRetry:
         assert ui.kinds().count("stream_end") == 1
         assert ("state", "error") in ui.events
 
-    def test_gate_consults_live_stream_provider_not_primary(self, tmp_db):
+    def test_gate_consults_serving_lane_provider(self, tmp_db):
         class FlakyError(Exception):
             pass
 
         ui = RecordingUI()
         session = _make_session(ui)
-        live = SimpleNamespace(retryable_error_names=frozenset({"FlakyError"}))
-        calls = {"n": 0}
+        # Post-fold the retry gate reads the SERVING lane's provider off
+        # the frame's consumer (the creation-time handoff register is
+        # gone).  FlakyError is retryable only per THIS provider's set —
+        # the re-issue proves the gate consulted the lane that armed the
+        # stream, not some global default.  (The distinct-fallback-lane
+        # variant of this contract is exercised through the real fallback
+        # walk in test_model_registry's TestSessionFallback.)
+        provider = arm_session(
+            session,
+            _dying_stream("x", exc=FlakyError("in-band transient")),
+            _good_stream("ok"),
+            retryable=frozenset({"FlakyError"}),
+        )
+        session.send("test")
 
-        def create(msgs):
-            calls["n"] += 1
-            if calls["n"] == 1:
-                # What _try_stream records at creation: the provider that
-                # owns the live stream (a fallback here — its retryable
-                # set differs from the primary's).
-                session._active_stream_provider = live
-                return _dying_stream("x", exc=FlakyError("in-band transient"))
-            return _good_stream("ok")
-
-        with (
-            patch.object(session, "_create_stream_with_retry", side_effect=create),
-            patch.object(session, "_full_messages", return_value=[]),
-        ):
-            session.send("test")
-
-        # FlakyError is NOT in the primary provider's retryable set — the
-        # re-issue proves the gate consulted the live stream's provider.
-        assert calls["n"] == 2
+        assert provider.create_streaming.call_count == 2
         assert _assistant_msgs(session)[-1]["content"] == "ok"
 
     def test_recreate_overflow_falls_through_to_compact_retry(self, tmp_db):
         ui = RecordingUI()
         session = _make_session(ui)
-        calls = {"n": 0}
-
-        def create(msgs):
-            calls["n"] += 1
-            if calls["n"] == 1:
-                return _dying_stream("x", exc=httpx.ReadError("wire died"))
-            if calls["n"] == 2:
-                # A mid-retry rebind landed on a smaller-window model: the
-                # re-create overflows deterministically.
-                raise RuntimeError("maximum context length exceeded")
-            return _good_stream("recovered")
-
+        provider = arm_session(
+            session,
+            _dying_stream("x", exc=httpx.ReadError("wire died")),
+            # A mid-retry rebind landed on a smaller-window model: the
+            # re-create overflows deterministically (a creation-phase
+            # raise — unarmed).
+            RuntimeError("maximum context length exceeded"),
+            _good_stream("recovered"),
+        )
         with (
-            patch.object(session, "_create_stream_with_retry", side_effect=create),
             patch.object(session, "_compact_messages") as compact,
             patch.object(session, "_full_messages", return_value=[]),
         ):
@@ -432,23 +412,19 @@ class TestMidStreamRetry:
 
         # The overflow surfaced as ITSELF (not masked behind the stream
         # death), so send()'s compact-and-retry arm recovered the turn.
-        assert calls["n"] == 3
+        assert provider.create_streaming.call_count == 3
         compact.assert_called_once()
         assert _assistant_msgs(session)[-1]["content"] == "recovered"
 
     def test_postcompaction_failure_surfaces_as_itself(self, tmp_db):
         ui = RecordingUI()
         session = _make_session(ui)
-        calls = {"n": 0}
-
-        def create(msgs):
-            calls["n"] += 1
-            if calls["n"] == 1:
-                raise RuntimeError("maximum context length exceeded")
-            raise ValueError("backend exploded")
-
+        arm_session(
+            session,
+            RuntimeError("maximum context length exceeded"),
+            ValueError("backend exploded"),
+        )
         with (
-            patch.object(session, "_create_stream_with_retry", side_effect=create),
             patch.object(session, "_compact_messages"),
             patch.object(session, "_full_messages", return_value=[]),
             pytest.raises(ValueError, match="backend exploded"),
@@ -480,8 +456,8 @@ class TestMidStreamRetry:
             if refreshes["n"] == 2:
                 session.model = "swapped-model"
 
+        arm_session(session, *streams)
         with (
-            patch.object(session, "_create_stream_with_retry", side_effect=streams),
             patch.object(session, "_refresh_model_from_registry", side_effect=swap_model),
             patch.object(session, "_full_messages", return_value=[]),
             patch.object(
@@ -501,8 +477,9 @@ class TestMidStreamRetry:
             yield StreamChunk(content_delta="orphan text")
             yield StreamChunk(content_delta="more")
 
+        arm_session(session, chunks())
         with pytest.raises(GenerationCancelled):
-            session._stream_attempt(iter(chunks()), my_generation=3)
+            session._stream_response(3)
 
         # The superseded thread's cancel arm must touch neither the UI
         # (its stream_end would reset the successor's inflight buffers)
@@ -518,12 +495,8 @@ class TestMidStreamRetry:
             session._generation += 1  # force-cancel claimed a successor
             raise GenerationCancelled()
 
+        arm_session(session, _dying_stream("Hel", exc=httpx.ReadError("wire died")))
         with (
-            patch.object(
-                session,
-                "_create_stream_with_retry",
-                side_effect=[_dying_stream("Hel", exc=httpx.ReadError("wire died"))],
-            ),
             patch.object(session, "_backoff_or_cancelled", side_effect=supersede_then_cancel),
             patch.object(session, "_full_messages", return_value=[]),
         ):
@@ -552,8 +525,8 @@ class TestMidStreamRetry:
             ),
             _good_stream("final answer"),
         ]
+        arm_session(session, *streams)
         with (
-            patch.object(session, "_create_stream_with_retry", side_effect=streams),
             patch.object(session, "_full_messages", return_value=[]),
         ):
             session.send("test")
@@ -582,12 +555,8 @@ class TestMidStreamRetry:
             real_backoff(delay, my_generation)
 
         dead_text = "a dead attempt long enough to flush past the carry window"
+        arm_session(session, _dying_stream(dead_text, exc=httpx.ReadError("wire died")))
         with (
-            patch.object(
-                session,
-                "_create_stream_with_retry",
-                side_effect=[_dying_stream(dead_text, exc=httpx.ReadError("wire died"))],
-            ),
             patch.object(session, "_backoff_or_cancelled", side_effect=cancel_then_backoff),
             patch.object(session, "_full_messages", return_value=[]),
         ):
@@ -612,8 +581,8 @@ class TestMidStreamRetry:
             _dying_stream("x", exc=httpx.ReadError("wire died")),
             _good_stream("ok"),
         ]
+        arm_session(session, *streams)
         with (
-            patch.object(session, "_create_stream_with_retry", side_effect=streams),
             patch.object(session, "_full_messages", return_value=[]),
         ):
             session.send("test")
@@ -632,25 +601,21 @@ class TestMidStreamRetry:
 
         ui = _BufferUI()
         session = _make_session(ui)
-        calls = {"n": 0}
-
-        def create(msgs):
-            calls["n"] += 1
-            if calls["n"] == 1:
-                return _dying_stream(
-                    "dead overflow text",
-                    exc=RuntimeError("maximum context length exceeded"),
-                )
-            return _good_stream("recovered")
-
+        provider = arm_session(
+            session,
+            _dying_stream(
+                "dead overflow text",
+                exc=RuntimeError("maximum context length exceeded"),
+            ),
+            _good_stream("recovered"),
+        )
         with (
-            patch.object(session, "_create_stream_with_retry", side_effect=create),
             patch.object(session, "_compact_messages"),
             patch.object(session, "_full_messages", return_value=[]),
         ):
             session.send("test")
 
-        assert calls["n"] == 2
+        assert provider.create_streaming.call_count == 2
         assert "".join(ui._ws_turn_content) == "recovered"
 
     def test_orphan_death_records_no_fatal_over_successor(self, tmp_db):
@@ -665,11 +630,12 @@ class TestMidStreamRetry:
             session._generation += 1
             raise ValueError("orphan death")
 
-        with (
-            patch.object(session, "_create_stream_with_retry", side_effect=[dying_superseded()]),
-            patch.object(session, "_full_messages", return_value=[]),
-            pytest.raises(ValueError, match="orphan death"),
-        ):
+        arm_session(session, dying_superseded())
+        with patch.object(session, "_full_messages", return_value=[]):
+            # Post-fold the orphan's death converts at the ladder's
+            # generation check and send() ends SILENTLY as cancelled — no
+            # arbitrary exception class escapes into the thread runner
+            # (named orphan-exit delta, design D12).
             session.send("test")
 
         # The orphan must not flash an error banner over the live
@@ -677,16 +643,15 @@ class TestMidStreamRetry:
         assert ("state", "error") not in ui.events
         assert not ui.of("error")
         assert not load_last_error(session._ws_id)
+        assert not _assistant_msgs(session)
 
     def test_non_retryable_error_is_immediately_fatal(self, tmp_db):
         ui = RecordingUI()
         session = _make_session(ui)
+        create = arm_session(
+            session, _dying_stream("Hel", exc=ValueError("model exploded"))
+        ).create_streaming
         with (
-            patch.object(
-                session,
-                "_create_stream_with_retry",
-                side_effect=[_dying_stream("Hel", exc=ValueError("model exploded"))],
-            ) as create,
             patch.object(session, "_full_messages", return_value=[]),
             pytest.raises(ValueError, match="model exploded"),
         ):
@@ -705,15 +670,12 @@ class TestMidStreamRetry:
         replace the operator-actionable stream-death error with its own."""
         ui = RecordingUI()
         session = _make_session(ui)
+        create = arm_session(
+            session,
+            _dying_stream("He", exc=httpx.ReadError("[SSL] record layer failure")),
+            RuntimeError("Cannot send a request, as the client has been closed."),
+        ).create_streaming
         with (
-            patch.object(
-                session,
-                "_create_stream_with_retry",
-                side_effect=[
-                    _dying_stream("He", exc=httpx.ReadError("[SSL] record layer failure")),
-                    RuntimeError("Cannot send a request, as the client has been closed."),
-                ],
-            ) as create,
             patch.object(session, "_full_messages", return_value=[]),
             caplog.at_level(logging.WARNING, logger="turnstone.core.session"),
             pytest.raises(IncompleteStreamError, match="ReadError"),
@@ -739,8 +701,8 @@ class TestMidStreamRetry:
             yield StreamChunk(finish_reason="stop")
             raise httpx.ReadError("late blip")  # the usage chunk is lost
 
+        arm_session(session, blipping())
         with (
-            patch.object(session, "_create_stream_with_retry", return_value=blipping()),
             patch.object(session, "_full_messages", return_value=[]),
         ):
             session.send("test")
diff --git a/tests/test_model_provider_obo.py b/tests/test_model_provider_obo.py
index af7be292..5ea1030b 100644
--- a/tests/test_model_provider_obo.py
+++ b/tests/test_model_provider_obo.py
@@ -1518,66 +1518,61 @@ class TestModelOboToken:
         )
         session._mcp_mint_client.mint_app_token_sync.assert_not_called()
 
-    def test_primary_stream_binds_backend_token_once_before_retry_loop(self) -> None:
-        """The main streaming path mirrors model_turn's SDK binding."""
+    def test_main_lane_carries_backend_auth_resolver(self) -> None:
+        """The main loop's lane wires the session's mint resolver — the
+        credential then resolves and binds INSIDE model_turn per attempt,
+        after its entry abort read (the #972/#832 ordering: a pre-set Stop
+        mints nothing, pinned in test_cancel; the with_options SDK binding
+        itself is model_turn's own pinned behavior).  This replaces the
+        retired pin on _try_stream's hoisted once-per-ladder resolve."""
         sess = MagicMock()
-        sess._MAX_RETRIES = 2
-        sess._provider = MagicMock()
-        sess._provider.create_streaming.return_value = iter(())
-        sess._get_capabilities.return_value = SimpleNamespace(default_reasoning_effort=None)
-        sess._maybe_attach_vllm_chat_reasoning.side_effect = lambda messages, _provider, _alias: (
-            messages
-        )
-        sess._model_backend_auth_token.return_value = "minted-jwt"
-        sess._cancel_ref = []
-        sess._get_active_tools.return_value = []
-        sess.max_tokens = 1024
-        sess.temperature = None
+        sess._registry = None
+        sess._config_store = None
+        sess.temperature = 0.5
         sess.reasoning_effort = None
-        sess._provider_extra_params.return_value = None
-        sess._get_deferred_names.return_value = frozenset()
-        sess._resolve_replay_reasoning_to_model.return_value = False
-        base_client = MagicMock()
-        base_client.base_url = "https://gateway.example.com"
-        bound_client = object()
-        base_client.with_options.return_value = bound_client
 
-        stream = ChatSession._try_stream(
+        lane = ChatSession._build_main_lane(
             sess,
-            base_client,
-            "vmg/opus",
-            [{"role": "user", "content": "hi"}],
-            model_alias="tf",
+            provider=MagicMock(provider_name="openai-compatible"),
+            client=MagicMock(),
+            model="vmg/opus",
+            alias="tf",
+            capabilities=SimpleNamespace(),
         )
 
-        assert list(stream) == []
-        sess._model_backend_auth_token.assert_called_once_with("tf")
-        base_client.with_options.assert_called_once_with(api_key="minted-jwt")
-        assert sess._provider.create_streaming.call_args.kwargs["client"] is bound_client
+        assert lane.backend_auth_resolver is sess._model_backend_auth_token
+        assert lane.alias == "tf"
+        # The session's own sampling knobs override the lane's operator
+        # rungs (wire parity with the pre-fold loop).
+        assert lane.temperature == 0.5
+        assert lane.reasoning_effort is None
 
-    def test_primary_stream_forwards_alias_for_obo(self) -> None:
-        # Regression: the primary _create_stream_with_retry call must pass
-        # model_alias, or the backend-auth resolver can't resolve the OBO token and an
-        # entra_obo main turn goes out on the static client key. The fallback
-        # path and utility (title) completions always passed the alias; the
-        # primary path silently didn't.
+    def test_primary_lane_built_with_session_alias_for_obo(self) -> None:
+        # Regression: the primary lane must carry the session alias, or the
+        # backend-auth resolver can't resolve the OBO token and an
+        # entra_obo main turn goes out on the static client key. (The
+        # pre-fold bug lived in _try_stream's missing model_alias kwarg;
+        # the lane build is the one place the alias enters now.)
         sess = MagicMock()
         sess._model_alias = "oboagent"
-        ChatSession._create_stream_with_retry(sess, [{"role": "user", "content": "hi"}])
-        sess._try_stream.assert_called_once()
-        assert sess._try_stream.call_args.kwargs.get("model_alias") == "oboagent"
+        ChatSession._model_turn_with_fallback(sess, MagicMock(), lambda wire: wire)
+        sess._build_main_lane.assert_called_once()
+        assert sess._build_main_lane.call_args.kwargs["alias"] == "oboagent"
 
     def test_fail_closed_refusal_never_enters_model_fallback_chain(self) -> None:
         sess = MagicMock()
         sess._model_alias = "oboagent"
-        sess._try_stream.side_effect = BackendAuthUnavailableError("mint failed")
+        sess._model_turn_with_retry.side_effect = BackendAuthUnavailableError("mint failed")
         sess._registry.fallback = ["static-backup"]
-        sess._get_health_tracker.return_value = None
+        tracker = MagicMock()
+        sess._get_health_tracker.return_value = tracker
 
         with pytest.raises(BackendAuthUnavailableError):
-            ChatSession._create_stream_with_retry(sess, [{"role": "user", "content": "hi"}])
+            ChatSession._model_turn_with_fallback(sess, MagicMock(), lambda wire: wire)
 
-        sess._try_fallback.assert_not_called()
+        sess._try_fallback_lane.assert_not_called()
+        # An auth refusal is never reinterpreted as backend health.
+        tracker.record_failure.assert_not_called()
 
     def test_static_alias_returns_none_and_never_mints(self) -> None:
         static_cfg = ModelConfig(
diff --git a/tests/test_model_registry.py b/tests/test_model_registry.py
index 814ac2db..00be0aa9 100644
--- a/tests/test_model_registry.py
+++ b/tests/test_model_registry.py
@@ -2185,41 +2185,44 @@ class TestSessionConstructionFailureLatch:
 
 class TestSessionFallback:
     def test_fallback_on_primary_failure(self) -> None:
+        # provider="openai-compatible" pins the Chat Completions surface, the
+        # one the patched ``chat.completions.create`` stubs below speak (see
+        # TestSessionRemovedAliasDegradedTurns._registry for the precedent).
         reg = ModelRegistry(
             models={
-                "primary": ModelConfig("primary", "http://p/v1", "k", "p-model"),
-                "fallback": ModelConfig("fallback", "http://f/v1", "k", "f-model"),
+                "primary": ModelConfig(
+                    "primary", "http://p/v1", "k", "p-model", provider="openai-compatible"
+                ),
+                "fallback": ModelConfig(
+                    "fallback", "http://f/v1", "k", "f-model", provider="openai-compatible"
+                ),
             },
             default="primary",
             fallback=["fallback"],
         )
         session = _make_session(registry=reg, model_alias="primary")
+        # Primary: an unarmed creation failure (raises before any chunk, so
+        # cancel_ref is never appended) — a non-retryable class, so the
+        # per-lane ladder gives up after one attempt and the fallback walk
+        # takes over.
+        session.client.chat.completions.create = MagicMock(
+            side_effect=ConnectionError("Primary down")
+        )
+        # Fallback: resolved through the REAL registry binding, so the fake
+        # goes on the registry's own client for that alias, not the session's.
+        fb_client = reg.get_client("fallback")
+        fb_client.chat.completions.create = scripted_chat_client({"content": "fallback_response"})
 
-        # _try_stream: first call (primary) raises, second call (fallback) succeeds
-        call_count = 0
+        session.send("hi")
 
-        def fake_try_stream(client: Any, model: str, msgs: Any, **kwargs: Any) -> str:
-            nonlocal call_count
-            call_count += 1
-            if call_count == 1:
-                raise ConnectionError("Primary down")
-            return "fallback_response"
-
-        session._try_stream = fake_try_stream  # type: ignore[assignment]
-        result = session._create_stream_with_retry([{"role": "user", "content": "hi"}])
-        assert result == "fallback_response"
-        assert call_count == 2
+        assert session.messages[-1].text == "fallback_response"
         assert any("falling back" in i for i in session.ui.infos)
 
     def test_no_fallback_without_registry(self) -> None:
         session = _make_session()
-
-        def fake_try_stream(client: Any, model: str, msgs: Any, **kwargs: Any) -> str:
-            raise ConnectionError("Down")
-
-        session._try_stream = fake_try_stream  # type: ignore[assignment]
+        session.client.chat.completions.create = MagicMock(side_effect=ConnectionError("Down"))
         with pytest.raises(ConnectionError):
-            session._create_stream_with_retry([{"role": "user", "content": "hi"}])
+            session.send("hi")
 
 
 class TestSessionAgentModel:
diff --git a/tests/test_persona_guards.py b/tests/test_persona_guards.py
index 5c30f4af..2966d0d6 100644
--- a/tests/test_persona_guards.py
+++ b/tests/test_persona_guards.py
@@ -9,11 +9,12 @@ stamped-at-create isolation from later persona edits.
 from __future__ import annotations
 
 import json
-from typing import Any
+from typing import TYPE_CHECKING, Any
 from unittest.mock import MagicMock, patch
 
 import pytest
 
+from tests._parity_832 import make_result
 from turnstone.core.personas import PersonaSnapshot, snapshot_from_persona
 from turnstone.core.session import ChatSession
 from turnstone.core.storage import get_storage
@@ -21,6 +22,9 @@ from turnstone.core.storage._utils import PERSONA_MUTABLE
 from turnstone.core.tools import TASK_AGENT_TOOLS
 from turnstone.core.workstream import WorkstreamKind
 
+if TYPE_CHECKING:
+    from turnstone.core.model_turn import ModelTurnResult
+
 
 def _snap(
     *,
@@ -333,18 +337,17 @@ class TestMemoryOff:
         summary = SimpleNamespace(content="## Open tasks\nfinish it", finish_reason="stop")
         n = {"i": 0}
 
-        def stream(*_a: Any, **_k: Any) -> dict[str, str]:
+        def stream(*_a: Any, **_k: Any) -> ModelTurnResult:
             n["i"] += 1
             if n["i"] == 1:
                 session._compaction_advised = True  # advisory fired this turn
-                return {"role": "assistant", "content": "pausing to compact"}
-            return {"role": "assistant", "content": "all done"}
+                return make_result(content="pausing to compact")
+            return make_result(content="all done")
 
         def est(*_a: Any, **_k: Any) -> int:
             return 9_999 if n["i"] <= 1 else 10  # over threshold only on the stop turn
 
         with (
-            patch.object(session, "_create_stream_with_retry", return_value=iter([])),
             patch.object(session, "_stream_response", side_effect=stream),
             patch.object(session, "_full_messages", return_value=[]),
             patch.object(session, "_update_token_table"),
diff --git a/tests/test_reasoning_audit_log_discipline.py b/tests/test_reasoning_audit_log_discipline.py
index e089b61d..d7aa9308 100644
--- a/tests/test_reasoning_audit_log_discipline.py
+++ b/tests/test_reasoning_audit_log_discipline.py
@@ -27,6 +27,7 @@ from types import SimpleNamespace
 from typing import Any
 from unittest.mock import patch
 
+from tests._parity_832 import scripted_provider
 from tests._session_helpers import make_session
 from turnstone.core.history_decoration import (
     extract_reasoning_for_history,
@@ -36,6 +37,7 @@ from turnstone.core.providers._anthropic import AnthropicProvider
 from turnstone.core.providers._openai_chat import OpenAIChatCompletionsProvider
 from turnstone.core.providers._openai_responses import OpenAIResponsesProvider
 from turnstone.core.providers._protocol import StreamChunk, UsageInfo
+from turnstone.core.trajectory import Turn
 
 _MARKER = "SECRET_REASONING_MARKER_xyz123_unlikely_collision"
 
@@ -226,30 +228,34 @@ class TestReasoningAuditLogDiscipline:
             f"reasoning text into INFO+ logs: {offending}"
         )
 
-    def test_synth_reasoning_block_via_stream_attempt_does_not_log_reasoning(
+    def test_synth_reasoning_block_via_stream_response_does_not_log_reasoning(
         self,
     ) -> None:
-        """Drives ChatSession._stream_attempt (which invokes
-        model_turn.synth_reasoning_block at end-of-stream via
-        _finalize_provider_blocks) with a fake
-        ``reasoning_delta=_MARKER`` chunk; asserts no log call carried
-        the marker text."""
+        """Drives session._stream_response (the real drain seam —
+        _stream_attempt no longer exists post-#832; invokes
+        model_turn.synth_reasoning_block at end-of-turn via
+        finalize_provider_blocks) with a fake ``reasoning_delta=_MARKER``
+        chunk; asserts no log call carried the marker text."""
         session = make_session()
-        chunks = [
-            StreamChunk(reasoning_delta=_MARKER, is_first=True),
-            StreamChunk(content_delta="answer"),
-            StreamChunk(
-                finish_reason="stop",
-                usage=UsageInfo(prompt_tokens=10, completion_tokens=20, total_tokens=30),
-            ),
-        ]
+        session._provider = scripted_provider(
+            [
+                StreamChunk(reasoning_delta=_MARKER, is_first=True),
+                StreamChunk(content_delta="answer"),
+                StreamChunk(
+                    finish_reason="stop",
+                    usage=UsageInfo(prompt_tokens=10, completion_tokens=20, total_tokens=30),
+                ),
+            ]
+        )
+        session.messages.append(Turn.user("hi"))
         captured, patchers = _capture_log_calls()
         for p in patchers:
             p.start()
         try:
-            msg = session._stream_attempt(iter(chunks))
-            # Synth block stamped onto _provider_content with the marker.
-            assert msg["_provider_content"][0]["text"] == _MARKER
+            result = session._stream_response(0)
+            # Synth block stamped onto the native lane with the marker.
+            assert result.turn.native is not None
+            assert result.turn.native.blocks[0]["text"] == _MARKER
         finally:
             for p in patchers:
                 p.stop()
@@ -259,7 +265,7 @@ class TestReasoningAuditLogDiscipline:
             if _payload_contains_marker(args, kwargs)
         ]
         assert offending == [], (
-            f"_stream_attempt + synth_reasoning_block leaked reasoning "
+            f"_stream_response + synth_reasoning_block leaked reasoning "
             f"text into INFO+ logs: {offending}"
         )
 
@@ -332,31 +338,34 @@ class TestReasoningAuditLogDiscipline:
         )
 
     def test_maybe_attach_vllm_chat_reasoning_does_not_log_reasoning(self) -> None:
-        """Phase 5 gate method on ChatSession — the session-level
-        composite gate calls ``attach_vllm_chat_reasoning_field`` when
-        all 3 conditions pass.  Pin that the gate path itself doesn't
-        log reasoning text (the registry / capability lookups happen
-        adjacent to the reasoning bytes; a defensive ``log.warning``
-        showing the message dict on an error path would silently
-        violate the contract)."""
+        """Phase 5 gate — ``model_turn.maybe_attach_vllm_chat_reasoning``.
+
+        Post-#832 this composite gate is a plain module function (no
+        ``ChatSession`` delegate survives; ``model_turn.model_turn`` calls
+        it directly with the lane's own registry/alias). Pin that the
+        gate path itself doesn't log reasoning text (the registry /
+        capability lookups happen adjacent to the reasoning bytes; a
+        defensive ``log.warning`` showing the message dict on an error
+        path would silently violate the contract)."""
+        from turnstone.core.model_turn import maybe_attach_vllm_chat_reasoning
         from turnstone.core.providers._openai_chat import OpenAIChatCompletionsProvider
 
-        session = make_session()
-        session._registry = SimpleNamespace(
+        registry = SimpleNamespace(
             get_config=lambda _alias: SimpleNamespace(
                 replay_reasoning_to_model=True,
                 capabilities={},
                 server_compat={"server_type": "vllm"},
             )
         )
-        session._model_alias = "qwen3"
         provider = OpenAIChatCompletionsProvider()
 
         captured, patchers = _capture_log_calls()
         for p in patchers:
             p.start()
         try:
-            out = session._maybe_attach_vllm_chat_reasoning([self._thinking_msg(_MARKER)], provider)
+            out = maybe_attach_vllm_chat_reasoning(
+                [self._thinking_msg(_MARKER)], provider, registry, "qwen3"
+            )
             assert out[0]["reasoning"] == _MARKER
         finally:
             for p in patchers:
@@ -367,6 +376,6 @@ class TestReasoningAuditLogDiscipline:
             if _payload_contains_marker(args, kwargs)
         ]
         assert offending == [], (
-            f"ChatSession._maybe_attach_vllm_chat_reasoning leaked reasoning "
+            f"model_turn.maybe_attach_vllm_chat_reasoning leaked reasoning "
             f"text into INFO+ logs: {offending}"
         )
diff --git a/tests/test_sdk_stream_boundary.py b/tests/test_sdk_stream_boundary.py
index ecf74362..b690d3bb 100644
--- a/tests/test_sdk_stream_boundary.py
+++ b/tests/test_sdk_stream_boundary.py
@@ -9,9 +9,9 @@ transports (no network, no live backend):
 2. The Anthropic ``messages.stream()`` helper propagates the same shape.
 3. Closing an httpx-backed SDK client from another thread while a read is
    blocked (the ``ModelRegistry.reload()`` shape) surfaces as an
-   ``httpx.TransportError`` on the blocked ``next()`` — which is why the
-   resilient ``_stream_response`` re-resolves the registry binding before
-   re-creating.
+   ``httpx.TransportError`` on the blocked ``next()`` — which is why
+   ``_stream_response``'s mid-stream re-issue ladder re-resolves the
+   registry binding before re-creating.
 
 If an SDK/httpx upgrade changes any of these, the ``transport_guarded``
 conversion (and the retry gate consuming it) must be re-verified — these
diff --git a/tests/test_session.py b/tests/test_session.py
index 2085a9a1..2817a276 100644
--- a/tests/test_session.py
+++ b/tests/test_session.py
@@ -12,6 +12,7 @@ from unittest.mock import MagicMock, patch
 import pytest
 
 from tests._oidc_test_helpers import keyed_app_state
+from tests._parity_832 import make_result
 from tests._session_helpers import (
     FakeAnthropicBlock,
     as_stream,
@@ -20,6 +21,7 @@ from tests._session_helpers import (
     scripted_chat_client,
     seam_provider,
 )
+from turnstone.core.model_turn import ModelTurnResult, provider_extra_params
 from turnstone.core.session import _IMAGE_EXTENSIONS, _IMAGE_SIZE_CAP, ChatSession
 from turnstone.core.trajectory import (
     Turn,
@@ -133,10 +135,17 @@ def _send_with_mocks(session, responses, mock_execute, **extra_patches):
     Extra per-test patches (e.g. wrapping ``_collect_advisories``) ride
     via ``**extra_patches`` — keyword name maps to attribute on the
     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
+    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
+    than script a provider.
     """
     from unittest.mock import patch as _patch
 
-    def mock_response(_msgs, _gen):
+    def mock_response(_gen):
         return responses.pop(0)
 
     with contextlib.ExitStack() as stack:
@@ -2050,18 +2059,17 @@ class TestTitleRetry:
         # The assistant's opening turn is ALL tool calls — under the old
         # trigger no title would generate until a later text-only turn.
         responses = [
-            {
-                "role": "assistant",
-                "content": "working",
-                "tool_calls": [
+            make_result(
+                "working",
+                tool_calls=[
                     {
                         "id": "c1",
                         "type": "function",
                         "function": {"name": "echo", "arguments": "{}"},
                     }
                 ],
-            },
-            {"role": "assistant", "content": "done"},
+            ),
+            make_result("done"),
         ]
         capture_cls, started = _capturing_thread_cls()
 
@@ -2091,7 +2099,7 @@ class TestTitleRetry:
         for user_input, kwargs in (("   ", {}), ("a real message", {"from_wake": True})):
             session = _make_session()
             with (
-                _send_with_mocks(session, [{"role": "assistant", "content": "ok"}], mock_execute),
+                _send_with_mocks(session, [make_result("ok")], mock_execute),
                 patch("turnstone.core.session.threading.Thread", capture_cls),
             ):
                 session.send(user_input, **kwargs)
@@ -2923,7 +2931,6 @@ class TestAgentChildRegistration:
         with (
             patch.object(session, "_prepare_tool", side_effect=fake_prepare),
             patch.object(session, "_resolve_capabilities", return_value=OPENAI_COMPAT_DEFAULT),
-            patch.object(session, "_provider_extra_params", return_value={}),
         ):
             session._run_agent(
                 turns,
@@ -2983,7 +2990,6 @@ class TestAgentChildRegistration:
         with (
             patch.object(session, "_prepare_tool", side_effect=fake_prepare),
             patch.object(session, "_resolve_capabilities", return_value=OPENAI_COMPAT_DEFAULT),
-            patch.object(session, "_provider_extra_params", return_value={}),
         ):
             session._run_agent(
                 turns,
@@ -3929,7 +3935,16 @@ class TestTruncateBeforeJudge:
 
 
 class TestProviderExtraParams:
-    """Tests for _provider_extra_params — server_compat passthrough only."""
+    """Tests for the session lane's extra_params resolution — server_compat
+    passthrough only.
+
+    #832 deleted ``ChatSession._provider_extra_params``: it was a thin
+    delegate whose last caller was the retired stream-creation ladder, and
+    the resolution now happens inside ``resolve_lane``.  These pin the
+    module function every lane goes through,
+    :func:`turnstone.core.model_turn.provider_extra_params`, with the
+    session's own binding supplied explicitly.
+    """
 
     def _session_with_provider(self, provider_name: str, tmp_db) -> ChatSession:
         from turnstone.core.providers import create_provider
@@ -3938,19 +3953,29 @@ class TestProviderExtraParams:
         session._provider = create_provider(provider_name)
         return session
 
+    @staticmethod
+    def _extra(session: ChatSession, alias: str | None = None):
+        """The session binding's extra_params, as ``resolve_lane`` resolves
+        them (*alias* overrides the primary — the fallback-lane case)."""
+        return provider_extra_params(
+            session._provider,
+            session._registry,
+            alias if alias is not None else (session._model_alias or ""),
+        )
+
     def test_openai_compatible_no_compat_returns_none(self, tmp_db):
         """No server_compat → no extra_body needed (no auto-injection)."""
         session = self._session_with_provider("openai-compatible", tmp_db)
-        assert session._provider_extra_params() is None
+        assert self._extra(session) is None
 
     def test_openai_commercial_no_compat_returns_none(self, tmp_db):
         """Cloud OpenAI without server_compat → None."""
         session = self._session_with_provider("openai", tmp_db)
-        assert session._provider_extra_params() is None
+        assert self._extra(session) is None
 
     def test_anthropic_returns_none(self, tmp_db):
         session = self._session_with_provider("anthropic", tmp_db)
-        assert session._provider_extra_params() is None
+        assert self._extra(session) is None
 
     def test_no_reasoning_effort_kwarg(self, tmp_db):
         """reasoning_effort is not part of the surface; passing it should TypeError.
@@ -3964,7 +3989,7 @@ class TestProviderExtraParams:
         bad_kwargs = {"reasoning_effort": "high"}
         session = self._session_with_provider("openai-compatible", tmp_db)
         with pytest.raises(TypeError):
-            session._provider_extra_params(**bad_kwargs)
+            provider_extra_params(session._provider, session._registry, "", **bad_kwargs)
 
     def test_server_compat_extra_body_passes_through(self, tmp_db):
         """server_compat.extra_body workarounds forward as extra_params."""
@@ -3980,8 +4005,7 @@ class TestProviderExtraParams:
         )
         session._registry = ModelRegistry(models={"test": cfg}, default="test")
         session._model_alias = "test"
-        result = session._provider_extra_params()
-        assert result == {"skip_special_tokens": False}
+        assert self._extra(session) == {"skip_special_tokens": False}
 
     def test_operator_chat_template_kwargs_pass_through(self, tmp_db):
         """Operator-set chat_template_kwargs (e.g. for gpt-oss) forwards verbatim."""
@@ -3997,11 +4021,11 @@ class TestProviderExtraParams:
         )
         session._registry = ModelRegistry(models={"test": cfg}, default="test")
         session._model_alias = "test"
-        result = session._provider_extra_params()
-        assert result == {"chat_template_kwargs": {"reasoning_effort": "high"}}
+        assert self._extra(session) == {"chat_template_kwargs": {"reasoning_effort": "high"}}
 
     def test_model_alias_resolves_target_compat(self, tmp_db):
-        """model_alias parameter selects compat from the target, not the primary."""
+        """The alias argument selects compat from the target, not the primary
+        — the fallback lane's own extra_params, resolved per lane swap."""
         from turnstone.core.model_registry import ModelConfig, ModelRegistry
 
         session = self._session_with_provider("openai-compatible", tmp_db)
@@ -4027,9 +4051,9 @@ class TestProviderExtraParams:
         session._model_alias = "primary"
 
         # Primary alias → gets Gemma workaround
-        assert session._provider_extra_params() == {"skip_special_tokens": False}
+        assert self._extra(session) == {"skip_special_tokens": False}
         # Fallback alias → no compat at all
-        assert session._provider_extra_params(model_alias="fallback") is None
+        assert self._extra(session, "fallback") is None
 
 
 class TestSafePrepareTool:
@@ -5007,7 +5031,7 @@ class TestMemoryCompositionDeferral:
             seen_queries.append(extract_recent_context(dicts_from_turns(session.messages)))
             real_init()
 
-        responses = [{"role": "assistant", "content": "ok"}]
+        responses = [make_result("ok")]
         with _send_with_mocks(
             session, responses, lambda _tc: ([], None), _init_system_messages=spy_init
         ):
@@ -5030,7 +5054,7 @@ class TestMemoryCompositionDeferral:
             nonlocal init_calls
             init_calls += 1
 
-        responses = [{"role": "assistant", "content": "ok"}]
+        responses = [make_result("ok")]
         with _send_with_mocks(
             session, responses, lambda _tc: ([], None), _init_system_messages=spy_init
         ):
@@ -5601,18 +5625,17 @@ class TestMetacognitiveBuffers:
         full ``send`` loop, not just ``_collect_advisories`` in isolation."""
         session = _make_session()
         responses = [
-            {
-                "role": "assistant",
-                "content": "calling",
-                "tool_calls": [
+            make_result(
+                "calling",
+                tool_calls=[
                     {
                         "id": "call_x",
                         "type": "function",
                         "function": {"name": "echo", "arguments": "{}"},
                     }
                 ],
-            },
-            {"role": "assistant", "content": "ack"},
+            ),
+            make_result("ack"),
         ]
 
         def mock_execute(_tool_calls):
@@ -5703,18 +5726,17 @@ class TestMetacognitiveBuffers:
         """
         session = _make_session()
         responses = [
-            {
-                "role": "assistant",
-                "content": "calling",
-                "tool_calls": [
+            make_result(
+                "calling",
+                tool_calls=[
                     {
                         "id": "call_x",
                         "type": "function",
                         "function": {"name": "echo", "arguments": "{}"},
                     }
                 ],
-            },
-            {"role": "assistant", "content": "ack"},
+            ),
+            make_result("ack"),
         ]
 
         def mock_execute(_tool_calls):
@@ -5764,18 +5786,17 @@ class TestMetacognitiveBuffers:
         a prefix-only call."""
         session = _make_session()
         responses = [
-            {
-                "role": "assistant",
-                "content": "calling",
-                "tool_calls": [
+            make_result(
+                "calling",
+                tool_calls=[
                     {
                         "id": "call_x",
                         "type": "function",
                         "function": {"name": "echo", "arguments": "{}"},
                     }
                 ],
-            },
-            {"role": "assistant", "content": "ack"},
+            ),
+            make_result("ack"),
         ]
 
         def mock_execute(_tool_calls):
@@ -5819,18 +5840,17 @@ class TestMetacognitiveBuffers:
         breaks this test."""
         session = _make_session()
         responses = [
-            {
-                "role": "assistant",
-                "content": "calling",
-                "tool_calls": [
+            make_result(
+                "calling",
+                tool_calls=[
                     {
                         "id": "call_x",
                         "type": "function",
                         "function": {"name": "echo", "arguments": "{}"},
                     }
                 ],
-            },
-            {"role": "assistant", "content": "ack"},
+            ),
+            make_result("ack"),
         ]
 
         def mock_execute(_tool_calls):
@@ -5913,18 +5933,17 @@ class TestMetacognitiveBuffers:
         ``system`` DB row appended after the tool row."""
         session = _make_session()
         responses = [
-            {
-                "role": "assistant",
-                "content": "calling",
-                "tool_calls": [
+            make_result(
+                "calling",
+                tool_calls=[
                     {
                         "id": "call_x",
                         "type": "function",
                         "function": {"name": "echo", "arguments": "{}"},
                     }
                 ],
-            },
-            {"role": "assistant", "content": "ack"},
+            ),
+            make_result("ack"),
         ]
 
         def mock_execute(_tool_calls):
@@ -5953,18 +5972,17 @@ class TestMetacognitiveBuffers:
         is never spliced into tool content anymore."""
         session = _make_session()
         responses = [
-            {
-                "role": "assistant",
-                "content": "calling",
-                "tool_calls": [
+            make_result(
+                "calling",
+                tool_calls=[
                     {
                         "id": "call_x",
                         "type": "function",
                         "function": {"name": "echo", "arguments": "{}"},
                     }
                 ],
-            },
-            {"role": "assistant", "content": "ack"},
+            ),
+            make_result("ack"),
         ]
 
         def mock_execute(_tool_calls):
@@ -5990,18 +6008,17 @@ class TestMetacognitiveBuffers:
         interjection rides its own ``system`` row."""
         session = _make_session()
         responses = [
-            {
-                "role": "assistant",
-                "content": "calling",
-                "tool_calls": [
+            make_result(
+                "calling",
+                tool_calls=[
                     {
                         "id": "call_x",
                         "type": "function",
                         "function": {"name": "view_image", "arguments": "{}"},
                     }
                 ],
-            },
-            {"role": "assistant", "content": "ack"},
+            ),
+            make_result("ack"),
         ]
 
         def mock_execute(_tool_calls):
@@ -6039,18 +6056,17 @@ class TestMetacognitiveBuffers:
         carries the interjection — both replay from their own rows."""
         session = _make_session()
         responses = [
-            {
-                "role": "assistant",
-                "content": "calling",
-                "tool_calls": [
+            make_result(
+                "calling",
+                tool_calls=[
                     {
                         "id": "call_x",
                         "type": "function",
                         "function": {"name": "view_image", "arguments": "{}"},
                     }
                 ],
-            },
-            {"role": "assistant", "content": "ack"},
+            ),
+            make_result("ack"),
         ]
 
         def mock_execute(_tool_calls):
@@ -6103,11 +6119,7 @@ class TestMetacognitiveBuffers:
         # gate passes — content of the memories doesn't matter here.
         with (
             patch.object(session, "_visible_memory_count", return_value=3),
-            patch.object(
-                session,
-                "_create_stream_with_retry",
-                side_effect=GenerationCancelled(),
-            ),
+            patch.object(session, "_stream_response", side_effect=GenerationCancelled()),
         ):
             session.send("first user message")
 
@@ -6191,11 +6203,7 @@ class TestMetacognitiveBuffers:
         session._queue_tool_advisory("tool_error", "leftover")
         with (
             patch.object(session, "_visible_memory_count", return_value=0),
-            patch.object(
-                session,
-                "_create_stream_with_retry",
-                side_effect=GenerationCancelled(),
-            ),
+            patch.object(session, "_stream_response", side_effect=GenerationCancelled()),
         ):
             session.send("user input")
 
@@ -6426,7 +6434,12 @@ class TestUpdateTokenTableMsgsParam:
     """``_update_token_table(msgs=...)`` reuses the wire-bound message
     list already built for the stream call instead of re-folding the
     system turns (perf-2), so the calibration char count matches the
-    bytes the provider counted."""
+    bytes the provider counted.
+
+    Post-#832 the main loop feeds it ``ModelTurnResult.wire_msgs``, and
+    the on-the-fly re-fold fallback survives for callers (fake results,
+    direct calls) that have no wire list.  The old leading
+    ``assistant_msg`` argument is gone — the body never read it."""
 
     def test_uses_provided_msgs_skips_re_application(self, tmp_db):
         session = _make_session()
@@ -6440,13 +6453,15 @@ class TestUpdateTokenTableMsgsParam:
         ) as m_prep:
             pre_built = session._prepare_wire_messages(session._full_messages())
             calls_after_prebuild = m_prep.call_count
-            session._update_token_table({"role": "assistant", "content": "ok"}, msgs=pre_built)
+            session._update_token_table(msgs=pre_built)
             # Calibration must not have re-folded.
             assert m_prep.call_count == calls_after_prebuild
 
     def test_falls_back_to_apply_when_msgs_missing(self, tmp_db):
         """The optional kwarg has a fallback so callers that don't (or
-        can't) pre-build the wire copy still get a sane calibration."""
+        can't) pre-build the wire copy still get a sane calibration —
+        a ``ModelTurnResult`` with ``wire_msgs=None`` (the fake-result
+        shape send() passes straight through) takes this path."""
         session = _make_session()
         session._last_usage = {"prompt_tokens": 100, "completion_tokens": 50}
         session.messages.append(turn_from_dict({"role": "user", "content": "hi"}))
@@ -6455,7 +6470,7 @@ class TestUpdateTokenTableMsgsParam:
             "_prepare_wire_messages",
             wraps=session._prepare_wire_messages,
         ) as m_prep:
-            session._update_token_table({"role": "assistant", "content": "ok"})
+            session._update_token_table(msgs=make_result("ok").wire_msgs)
             # Fallback path folds on the fly.
             assert m_prep.call_count == 1
 
@@ -6475,11 +6490,7 @@ class TestUserAdvisoryCancelClear:
         session._queue_user_advisory("denial", "leftover")
         with (
             patch.object(session, "_visible_memory_count", return_value=0),
-            patch.object(
-                session,
-                "_create_stream_with_retry",
-                side_effect=GenerationCancelled(),
-            ),
+            patch.object(session, "_stream_response", side_effect=GenerationCancelled()),
         ):
             session.send("user input")
         assert _user_pending(session) == []
@@ -6489,11 +6500,7 @@ class TestUserAdvisoryCancelClear:
         session._queue_user_advisory("correction", "leftover")
         with (
             patch.object(session, "_visible_memory_count", return_value=0),
-            patch.object(
-                session,
-                "_create_stream_with_retry",
-                side_effect=KeyboardInterrupt(),
-            ),
+            patch.object(session, "_stream_response", side_effect=KeyboardInterrupt()),
             contextlib.suppress(KeyboardInterrupt),
         ):
             session.send("user input")
@@ -6504,11 +6511,7 @@ class TestUserAdvisoryCancelClear:
         session._queue_user_advisory("resume", "leftover")
         with (
             patch.object(session, "_visible_memory_count", return_value=0),
-            patch.object(
-                session,
-                "_create_stream_with_retry",
-                side_effect=RuntimeError("boom"),
-            ),
+            patch.object(session, "_stream_response", side_effect=RuntimeError("boom")),
             contextlib.suppress(RuntimeError),
         ):
             session.send("user input")
@@ -6535,7 +6538,7 @@ class TestUserAdvisoryCancelClear:
         session._title_generated = True
         stream_calls = 0
 
-        def mock_stream_response(msgs, my_generation=0):
+        def mock_stream_response(my_generation=0):
             nonlocal stream_calls
             stream_calls += 1
             if stream_calls == 1:
@@ -6543,7 +6546,7 @@ class TestUserAdvisoryCancelClear:
                 # time the no-tool branch runs ``_flush_queued_messages``,
                 # this item is in the queue waiting to be drained.
                 session.queue_message("late arrival", queue_msg_id="q-late")
-            return {"role": "assistant", "content": "ok"}
+            return make_result("ok")
 
         with (
             patch.object(session, "_stream_response", side_effect=mock_stream_response),
@@ -6591,11 +6594,11 @@ class TestDeliverWakeNudge:
         # won't match.  Bail before synthesizing an empty user turn.
         session._queue_tool_advisory("tool_error", "stale")
         before_len = len(session.messages)
-        with patch.object(session, "_create_stream_with_retry") as stream:
+        with patch.object(session, "_stream_response") as turn:
             session.deliver_wake_nudge_from_queue()
-        # No send → no message appended → stream untouched.
+        # No send → no message appended → the streaming seam untouched.
         assert len(session.messages) == before_len
-        assert stream.call_count == 0
+        assert turn.call_count == 0
         # Tool entry still queued (would orphan in production today; the
         # bail just protects against the empty-envelope failure mode).
         assert _tool_pending(session) == [("tool_error", "stale")]
@@ -6605,10 +6608,10 @@ class TestDeliverWakeNudge:
     def test_no_op_when_queue_is_empty(self, tmp_db):
         session = _make_session()
         before_len = len(session.messages)
-        with patch.object(session, "_create_stream_with_retry") as stream:
+        with patch.object(session, "_stream_response") as turn:
             session.deliver_wake_nudge_from_queue()
         assert len(session.messages) == before_len
-        assert stream.call_count == 0
+        assert turn.call_count == 0
         assert session._wake_source_tag == ""
 
     def test_drains_any_channel_onto_synthetic_empty_user_turn(self, tmp_db):
@@ -6620,12 +6623,7 @@ class TestDeliverWakeNudge:
         session._title_generated = True  # suppress auto-title thread
         session._nudge_queue.enqueue("idle_children", "your kids", "any")
         with (
-            patch.object(session, "_create_stream_with_retry", return_value=iter([])),
-            patch.object(
-                session,
-                "_stream_response",
-                return_value={"role": "assistant", "content": "ok"},
-            ),
+            patch.object(session, "_stream_response", return_value=make_result("ok")),
             patch.object(session, "_full_messages", return_value=[]),
             patch.object(session, "_update_token_table"),
             patch.object(session, "_print_status_line"),
@@ -6652,12 +6650,7 @@ class TestDeliverWakeNudge:
         session._title_generated = True
         session._queue_user_advisory("denial", "leftover")
         with (
-            patch.object(session, "_create_stream_with_retry", return_value=iter([])),
-            patch.object(
-                session,
-                "_stream_response",
-                return_value={"role": "assistant", "content": "ok"},
-            ),
+            patch.object(session, "_stream_response", return_value=make_result("ok")),
             patch.object(session, "_full_messages", return_value=[]),
             patch.object(session, "_update_token_table"),
             patch.object(session, "_print_status_line"),
@@ -6675,12 +6668,7 @@ class TestDeliverWakeNudge:
         session._title_generated = True
         session._queue_user_advisory("denial", "x")
         with (
-            patch.object(session, "_create_stream_with_retry", return_value=iter([])),
-            patch.object(
-                session,
-                "_stream_response",
-                return_value={"role": "assistant", "content": "ok"},
-            ),
+            patch.object(session, "_stream_response", return_value=make_result("ok")),
             patch.object(session, "_full_messages", return_value=[]),
             patch.object(session, "_update_token_table"),
             patch.object(session, "_print_status_line"),
@@ -6711,12 +6699,7 @@ class TestDeliverWakeNudge:
         # otherwise fire a fresh correction nudge.
         session.messages.append(turn_from_dict({"role": "user", "content": "earlier"}))
         with (
-            patch.object(session, "_create_stream_with_retry", return_value=iter([])),
-            patch.object(
-                session,
-                "_stream_response",
-                return_value={"role": "assistant", "content": "ok"},
-            ),
+            patch.object(session, "_stream_response", return_value=make_result("ok")),
             patch.object(session, "_full_messages", return_value=[]),
             patch.object(session, "_update_token_table"),
             patch.object(session, "_print_status_line"),
@@ -6758,22 +6741,17 @@ class TestDeliverWakeNudge:
         session._title_generated = True
         session._nudge_queue.enqueue("idle_children", "kids", "any")
 
-        def _queue_then_reply(*_a: Any, **_k: Any) -> dict[str, Any]:
+        def _queue_then_reply(*_a: Any, **_k: Any) -> ModelTurnResult:
             # First stream call: a real user message lands mid-wake-turn.
             # Subsequent calls: plain replies until the flush seam empties.
             if not session._queued_messages and not any(
                 "real user input" in str(m.content) for m in session.messages
             ):
                 session.queue_message("real user input", queue_msg_id="q-1")
-            return {"role": "assistant", "content": "ok"}
+            return make_result("ok")
 
         with (
-            patch.object(session, "_create_stream_with_retry", return_value=iter([])),
-            patch.object(
-                session,
-                "_stream_response",
-                side_effect=_queue_then_reply,
-            ),
+            patch.object(session, "_stream_response", side_effect=_queue_then_reply),
             patch.object(session, "_full_messages", return_value=[]),
             patch.object(session, "_update_token_table"),
             patch.object(session, "_print_status_line"),
@@ -6803,11 +6781,7 @@ class TestDeliverWakeNudge:
         session._queue_user_advisory("denial", "leftover")
         with (
             patch.object(session, "_visible_memory_count", return_value=0),
-            patch.object(
-                session,
-                "_create_stream_with_retry",
-                side_effect=RuntimeError("boom"),
-            ),
+            patch.object(session, "_stream_response", side_effect=RuntimeError("boom")),
             contextlib.suppress(RuntimeError),
         ):
             session.deliver_wake_nudge_from_queue()
@@ -6832,12 +6806,7 @@ class TestDeliverWakeNudge:
         session._title_generated = True
         session._queue_user_advisory("denial", "leftover")
         with (
-            patch.object(session, "_create_stream_with_retry", return_value=iter([])),
-            patch.object(
-                session,
-                "_stream_response",
-                return_value={"role": "assistant", "content": "ok"},
-            ),
+            patch.object(session, "_stream_response", return_value=make_result("ok")),
             patch.object(session, "_full_messages", return_value=[]),
             patch.object(session, "_update_token_table"),
             patch.object(session, "_print_status_line"),
@@ -6863,12 +6832,7 @@ class TestDeliverWakeNudge:
         session._title_generated = True
         session._queue_user_advisory("denial", "do not do that")
         with (
-            patch.object(session, "_create_stream_with_retry", return_value=iter([])),
-            patch.object(
-                session,
-                "_stream_response",
-                return_value={"role": "assistant", "content": "ok"},
-            ),
+            patch.object(session, "_stream_response", return_value=make_result("ok")),
             patch.object(session, "_full_messages", return_value=[]),
             patch.object(session, "_update_token_table"),
             patch.object(session, "_print_status_line"),
diff --git a/tests/test_session_chat_reasoning_replay.py b/tests/test_session_chat_reasoning_replay.py
index 1ce5dbaf..9e049902 100644
--- a/tests/test_session_chat_reasoning_replay.py
+++ b/tests/test_session_chat_reasoning_replay.py
@@ -3,17 +3,23 @@
 
 Phase 5 is the only reasoning-replay path that does NOT use the static
 ``supports_reasoning_replay`` capability gate.  It's a parallel path to
-Paths 1+2, gated entirely at the session level on three conditions:
+Paths 1+2, gated on three conditions and nothing else:
 
 1. Provider is ``OpenAIChatCompletionsProvider``.
 2. ``server_compat.server_type == "vllm"``.
 3. Operator-set ``ModelConfig.replay_reasoning_to_model`` is True.
 
-These tests drive through ``ChatSession._maybe_attach_vllm_chat_reasoning``
+These tests drive through ``model_turn.maybe_attach_vllm_chat_reasoning``
 to pin each gate independently, then one round-trip test through the real
 OpenAI Python SDK + httpx MockTransport confirms the ``reasoning`` field
 actually reaches the wire bytes (the SDK-boundary guarantee that the
-session-level attach approach hinges on).
+attach approach hinges on).
+
+The gate ran behind a ``ChatSession`` wrapper until #832 folded the main
+loop onto ``model_turn``; the attach is now one of the seam's own lowering
+passes, reading the lane's registry + alias.  Same three gates, one
+indirection down — and both session funnels (the streaming turn and
+``_utility_completion``) reach it through that one seam.
 """
 
 from __future__ import annotations
@@ -21,13 +27,15 @@ from __future__ import annotations
 import json
 from types import SimpleNamespace
 from typing import Any
-from unittest.mock import MagicMock, patch
+from unittest.mock import patch
 
 import httpx
 import pytest
 
-from tests._session_helpers import as_stream, mock_completion_result
+from tests._parity_832 import ArmedHandle
+from tests._session_helpers import as_stream, mock_completion_result, think_tag_stream
 from tests._session_helpers import make_session as _make_session
+from turnstone.core.model_turn import maybe_attach_vllm_chat_reasoning
 from turnstone.core.providers._anthropic import AnthropicProvider
 from turnstone.core.providers._openai_chat import OpenAIChatCompletionsProvider
 from turnstone.core.providers._openai_responses import OpenAIResponsesProvider
@@ -90,34 +98,33 @@ def _assistant_msg_with_thinking(text: str = "let me think") -> dict[str, Any]:
 
 
 # ---------------------------------------------------------------------------
-# Gate tests via ``_maybe_attach_vllm_chat_reasoning`` directly
+# Gate tests via ``maybe_attach_vllm_chat_reasoning`` directly
 # ---------------------------------------------------------------------------
 
 
 class TestMaybeAttachVllmChatReasoningGates:
-    """The session-level method that combines all three Phase 5 gates."""
+    """The seam pass that combines all three Phase 5 gates.
+
+    Reads the registry + alias the LANE carries — what the session
+    wrapper used to hand it, resolved per call inside ``model_turn`` so a
+    mid-session admin toggle keeps applying.
+    """
 
     def test_all_gates_pass_attaches_reasoning(self) -> None:
-        session = _make_session()
-        session._registry = _vllm_registry(replay=True)
-        session._model_alias = "qwen3"
         provider = OpenAIChatCompletionsProvider()
 
         msgs = [{"role": "user", "content": "q"}, _assistant_msg_with_thinking("CoT")]
-        out = session._maybe_attach_vllm_chat_reasoning(msgs, provider)
+        out = maybe_attach_vllm_chat_reasoning(msgs, provider, _vllm_registry(replay=True), "qwen3")
         assert out[1]["reasoning"] == "CoT"
 
     def test_non_chat_completions_provider_is_no_op(self) -> None:
         # Provider isinstance gate: Anthropic / Responses / Google all
         # have their own reasoning-replay paths (Paths 1 / 2) — Phase 5
         # must not double-attach.
-        session = _make_session()
-        session._registry = _vllm_registry(replay=True)
-        session._model_alias = "qwen3"
         provider = AnthropicProvider()
 
         msgs = [_assistant_msg_with_thinking()]
-        out = session._maybe_attach_vllm_chat_reasoning(msgs, provider)
+        out = maybe_attach_vllm_chat_reasoning(msgs, provider, _vllm_registry(replay=True), "qwen3")
         assert "reasoning" not in out[0]
         # Same reference — no copy made.
         assert out[0] is msgs[0]
@@ -127,13 +134,10 @@ class TestMaybeAttachVllmChatReasoningGates:
         # OpenAIChatCompletionsProvider) — the isinstance gate rejects
         # it cleanly.  This is the load-bearing distinction; an
         # accidental inheritance refactor would break the gate.
-        session = _make_session()
-        session._registry = _vllm_registry(replay=True)
-        session._model_alias = "qwen3"
         provider = OpenAIResponsesProvider()
 
         msgs = [_assistant_msg_with_thinking()]
-        out = session._maybe_attach_vllm_chat_reasoning(msgs, provider)
+        out = maybe_attach_vllm_chat_reasoning(msgs, provider, _vllm_registry(replay=True), "qwen3")
         assert "reasoning" not in out[0]
 
     @pytest.mark.parametrize("server_type", ["", "llama.cpp", "sglang", "openai", "unknown"])
@@ -141,43 +145,34 @@ class TestMaybeAttachVllmChatReasoningGates:
         # Server-type pin bounds blast radius — canonical OpenAI Chat
         # Completions, llama.cpp, sglang, and any unrecognised server
         # never receive the non-standard ``reasoning`` field.
-        session = _make_session()
-        session._registry = _registry_with_server_type(server_type, replay=True)
-        session._model_alias = "some-model"
+        registry = _registry_with_server_type(server_type, replay=True)
         provider = OpenAIChatCompletionsProvider()
 
         msgs = [_assistant_msg_with_thinking()]
-        out = session._maybe_attach_vllm_chat_reasoning(msgs, provider)
+        out = maybe_attach_vllm_chat_reasoning(msgs, provider, registry, "some-model")
         assert "reasoning" not in out[0]
 
     def test_operator_flag_off_is_no_op(self) -> None:
-        session = _make_session()
-        session._registry = _vllm_registry(replay=False)  # operator flag OFF
-        session._model_alias = "qwen3"
+        registry = _vllm_registry(replay=False)  # operator flag OFF
         provider = OpenAIChatCompletionsProvider()
 
         msgs = [_assistant_msg_with_thinking()]
-        out = session._maybe_attach_vllm_chat_reasoning(msgs, provider)
+        out = maybe_attach_vllm_chat_reasoning(msgs, provider, registry, "qwen3")
         assert "reasoning" not in out[0]
 
     def test_missing_registry_is_no_op(self) -> None:
-        session = _make_session()
-        session._registry = None
-        session._model_alias = "qwen3"
         provider = OpenAIChatCompletionsProvider()
 
         msgs = [_assistant_msg_with_thinking()]
-        out = session._maybe_attach_vllm_chat_reasoning(msgs, provider)
+        out = maybe_attach_vllm_chat_reasoning(msgs, provider, None, "qwen3")
         assert "reasoning" not in out[0]
 
     def test_missing_alias_is_no_op(self) -> None:
-        session = _make_session()
-        session._registry = _vllm_registry(replay=True)
-        session._model_alias = ""
+        # A lane outside the registry carries ``alias=""``.
         provider = OpenAIChatCompletionsProvider()
 
         msgs = [_assistant_msg_with_thinking()]
-        out = session._maybe_attach_vllm_chat_reasoning(msgs, provider)
+        out = maybe_attach_vllm_chat_reasoning(msgs, provider, _vllm_registry(replay=True), "")
         assert "reasoning" not in out[0]
 
     def test_registry_exception_is_no_op(self) -> None:
@@ -187,20 +182,17 @@ class TestMaybeAttachVllmChatReasoningGates:
         def boom(_alias: str) -> Any:
             raise KeyError("missing")
 
-        session = _make_session()
-        session._registry = SimpleNamespace(get_config=boom)
-        session._model_alias = "qwen3"
+        registry = SimpleNamespace(get_config=boom)
         provider = OpenAIChatCompletionsProvider()
 
         msgs = [_assistant_msg_with_thinking()]
-        out = session._maybe_attach_vllm_chat_reasoning(msgs, provider)
+        out = maybe_attach_vllm_chat_reasoning(msgs, provider, registry, "qwen3")
         assert "reasoning" not in out[0]
 
-    def test_explicit_alias_arg_overrides_session_default(self) -> None:
-        # When _try_stream forwards an explicit ``model_alias`` (different
-        # from the session's primary), the helper must read THAT alias'
-        # config — not the session's primary.  Mirrors the per-alias
-        # behaviour pinned for _resolve_replay_reasoning_to_model.
+    def test_alias_selects_its_own_config(self) -> None:
+        # The gate reads the config of the alias the LANE resolved — a
+        # fallback lane's alias, not the session's primary.  Mirrors the
+        # per-alias behaviour pinned for resolve_replay_reasoning_to_model.
         def per_alias(alias: str) -> Any:
             return SimpleNamespace(
                 replay_reasoning_to_model=(alias == "wants-replay"),
@@ -208,18 +200,16 @@ class TestMaybeAttachVllmChatReasoningGates:
                 server_compat={"server_type": "vllm"},
             )
 
-        session = _make_session()
-        session._registry = SimpleNamespace(get_config=per_alias)
-        session._model_alias = "primary"
+        registry = SimpleNamespace(get_config=per_alias)
         provider = OpenAIChatCompletionsProvider()
 
         msgs = [_assistant_msg_with_thinking()]
-        # Default alias → flag off → no attach.
-        out_default = session._maybe_attach_vllm_chat_reasoning(msgs, provider)
-        assert "reasoning" not in out_default[0]
-        # Explicit alias arg → flag on → attached.
-        out_explicit = session._maybe_attach_vllm_chat_reasoning(msgs, provider, "wants-replay")
-        assert out_explicit[0]["reasoning"] == "let me think"
+        # Flag off for this alias → no attach.
+        out_primary = maybe_attach_vllm_chat_reasoning(msgs, provider, registry, "primary")
+        assert "reasoning" not in out_primary[0]
+        # Flag on for this one → attached.
+        out_replay = maybe_attach_vllm_chat_reasoning(msgs, provider, registry, "wants-replay")
+        assert out_replay[0]["reasoning"] == "let me think"
 
 
 # ---------------------------------------------------------------------------
@@ -278,7 +268,7 @@ class TestReasoningFieldReachesWireBytes:
         provider = OpenAIChatCompletionsProvider()
 
         # Mimic the post-attach message shape that
-        # ``_maybe_attach_vllm_chat_reasoning`` produces, then sanitize.
+        # ``maybe_attach_vllm_chat_reasoning`` produces, then sanitize.
         # ``sanitize_messages`` runs inside provider._prepare_messages
         # and must preserve the non-``_``-prefixed ``reasoning`` field.
         messages = [
@@ -348,58 +338,55 @@ class TestReasoningFieldReachesWireBytes:
 
 
 # ---------------------------------------------------------------------------
-# Call-site integration: confirm _try_stream and _utility_completion both
-# invoke the helper.  Pins that the 2 hoist points stay in sync; a missed
-# call site is exactly the kind of regression this catches.  The agent
-# _run_agent path is deliberately NOT a Phase 5 hoist — see the NOTE
-# comment inside _run_agent's nested _api_call closure (grep session.py
-# for "Phase 5 vLLM ``reasoning`` field replay is intentionally NOT
-# wired here"): agent assistant messages don't carry
-# ``_provider_content`` so the helper would no-op every turn anyway.
+# Call-site integration: confirm the streaming turn and _utility_completion
+# both reach the attach.  Post-#832 both funnel through ``model_turn``,
+# which runs the pass itself — so these pin that each funnel still goes
+# through that seam (a call site that grew a private wire path would skip
+# it).  The sub-agent loop rides the same seam via ``_run_agent``'s
+# ``_api_call``; its assistant turns carry no ``_provider_content`` in
+# practice, so the pass no-ops there rather than being wired around.
 # ---------------------------------------------------------------------------
 
 
 class TestCallSitesInvokeMaybeAttach:
-    """The helper does nothing unless one of the 2 call sites calls it.
-    Verify the wiring at each — without this, a refactor that drops a
-    call site would silently regress Phase 5 on that path."""
+    """The pass does nothing for a call site that doesn't reach it.
+    Verify the wiring at each — without this, a refactor that gives one
+    funnel its own wire build would silently regress Phase 5 there."""
 
-    def test_try_stream_call_site_attaches(self) -> None:
+    def test_streaming_call_site_attaches(self) -> None:
         session = _make_session()
         session._registry = _vllm_registry(replay=True)
         session._model_alias = "qwen3"
+        session.model = "qwen3"
 
         captured: dict[str, Any] = {}
 
         def capture_streaming(**kwargs: Any) -> Any:
             captured.update(kwargs)
-            return iter([])
+            # The armed/creation classifier reads the cancel_ref, so the
+            # fake arms it eagerly like every real adapter.
+            ref = kwargs.get("cancel_ref")
+            if ref is not None:
+                ref.append(ArmedHandle())
+            return think_tag_stream("ok")
 
         provider = OpenAIChatCompletionsProvider()
         # Patch only the network-facing method so we don't actually call
         # an LLM, but keep the real provider instance (so the isinstance
         # gate sees the right type).
         provider.create_streaming = capture_streaming  # type: ignore[method-assign]
+        session._provider = provider
+        session.messages = turns_from_dicts([_assistant_msg_with_thinking("from the main loop")])
 
-        with (
-            patch.object(session, "_get_active_tools", return_value=None),
-            patch.object(session, "_provider_extra_params", return_value=None),
-            patch.object(session, "_get_deferred_names", return_value=frozenset()),
-            patch.object(session, "_check_cancelled"),
-        ):
-            session._try_stream(
-                client=MagicMock(),
-                model="qwen3",
-                msgs=[_assistant_msg_with_thinking("from try_stream")],
-                provider=provider,
-                model_alias="qwen3",
-            )
+        session._stream_response(0)
 
         # The messages handed to the provider include the attached
-        # reasoning field — proves _try_stream invoked
-        # _maybe_attach_vllm_chat_reasoning before the call.
+        # reasoning field — proves the streaming turn reached the attach.
+        # The wire list carries the session's system messages now, so the
+        # assistant turn is found by role, not by index.
         msgs_sent = captured["messages"]
-        assert msgs_sent[0]["reasoning"] == "from try_stream"
+        assistant = next(m for m in msgs_sent if m["role"] == "assistant")
+        assert assistant["reasoning"] == "from the main loop"
 
     def test_utility_completion_call_site_attaches(self) -> None:
         session = _make_session()
@@ -416,11 +403,11 @@ class TestCallSitesInvokeMaybeAttach:
         provider.create_streaming = capture_streaming  # type: ignore[method-assign]
         session._provider = provider
 
-        with (
-            patch.object(session, "_provider_extra_params", return_value=None),
-            patch.object(
-                session, "_get_capabilities", return_value=provider.get_capabilities("qwen3")
-            ),
+        # No extra_params patch: _utility_completion resolves them inside
+        # resolve_lane (a module seam reading the registry config), which
+        # a session-attribute patch cannot intercept.
+        with patch.object(
+            session, "_get_capabilities", return_value=provider.get_capabilities("qwen3")
         ):
             session._utility_completion(
                 turns_from_dicts([_assistant_msg_with_thinking("from utility")]),
diff --git a/tests/test_session_replay_reasoning.py b/tests/test_session_replay_reasoning.py
index 13c78670..4c359cd1 100644
--- a/tests/test_session_replay_reasoning.py
+++ b/tests/test_session_replay_reasoning.py
@@ -5,19 +5,24 @@ Phase 2 of optional reasoning persistence reads the per-model
 site and threads it through ``provider.create_streaming`` (the one
 transport post-#831).  These tests pin:
 
-1. The resolver helper (``ChatSession._resolve_replay_reasoning_to_model``)
-   walks the registry correctly and falls back to ``False`` (the
-   conservative default matching the migration server_default) when
-   the lookup fails.
-2. The streaming wire-build call site at ``session.py:_try_stream``
-   actually passes the resolved flag down — without this, the Phase
-   2 work is dead code (the strip-when-False predicate never fires).
-3. The non-streaming wire-build call site at
-   ``session.py:_utility_completion`` does the same.
+1. The resolver (``model_turn.resolve_replay_reasoning_to_model``) walks
+   the registry correctly and falls back to ``False`` (the conservative
+   default matching the migration server_default) when the lookup fails.
+2. The streaming wire-build call site actually passes the resolved flag
+   down — post-#832 that call site is ``model_turn``, reached through
+   ``session._stream_response`` and its lane-swap fallback walk.  Without
+   this the Phase 2 work is dead code (the strip-when-False predicate
+   never fires).
+3. The non-streaming call site at ``session.py:_utility_completion``
+   does the same, through the same ``model_turn`` seam.
 
-Drives through the real ``ChatSession._resolve_replay_reasoning_to_model``
-with a stub registry, then captures the kwarg passed to a mock provider
-to verify the flow end-to-end.
+Drives the real resolver against a stub registry, then reads the kwarg a
+fake provider captured — the same assertion surface as before the fold,
+one caller down.  The capability half of the resolver's AND-gate now
+reaches it as the LANE's capabilities (provider static table + the
+registry's operator overrides) instead of a ``capabilities=`` argument at
+the call site, so tests that supplied their own ``ModelCapabilities``
+state it through the stub registry's ``capabilities`` dict.
 """
 
 from __future__ import annotations
@@ -26,179 +31,165 @@ from types import SimpleNamespace
 from typing import Any
 from unittest.mock import MagicMock, patch
 
-from tests._session_helpers import as_stream, mock_completion_result
+from tests._parity_832 import SCENARIOS, scripted_provider
+from tests._session_helpers import (
+    FakeAnthropicBlock,
+    as_stream,
+    fake_anthropic_stream,
+    mock_completion_result,
+)
 from tests._session_helpers import make_session as _make_session
-from turnstone.core.trajectory import Turn
+from turnstone.core.model_turn import resolve_replay_reasoning_to_model
+from turnstone.core.providers._protocol import ModelCapabilities
+from turnstone.core.trajectory import Turn, turns_from_dicts
 
 
-def _registry_with_flag(persist: bool = True, replay: bool = False) -> Any:
+def _registry_with_flag(
+    persist: bool = True,
+    replay: bool = False,
+    caps_overrides: dict[str, Any] | None = None,
+) -> Any:
     """Stub registry returning a ModelConfig-shaped object with the
-    flags under test."""
+    flags under test.  *caps_overrides* rides the config's
+    ``capabilities`` dict, which is how an operator states a model
+    capability the lane must resolve."""
     return SimpleNamespace(
         get_config=lambda alias: SimpleNamespace(
             surface_persisted_reasoning=persist,
             replay_reasoning_to_model=replay,
+            capabilities=dict(caps_overrides or {}),
         )
     )
 
 
+def _flag_capture_provider(*, supports_replay: bool = True) -> MagicMock:
+    """Provider fake recording the kwargs ``model_turn`` built, then
+    replaying a finished stream.
+
+    The eager ``cancel_ref`` arming is mandatory at this seam (the
+    creation-vs-midstream classifier reads it), so the shape comes from
+    ``tests._parity_832.scripted_provider`` verbatim; only the advertised
+    reasoning-replay capability is per-test.
+    """
+    provider = scripted_provider(SCENARIOS["content_only"])
+    provider.get_capabilities.return_value = ModelCapabilities(
+        supports_reasoning_replay=supports_replay
+    )
+    return provider
+
+
+def _drive_stream(session: Any, provider: MagicMock) -> dict[str, Any]:
+    """Run ONE real streaming turn against *provider* and return the
+    kwargs that reached ``create_streaming``."""
+    session._provider = provider
+    session.messages.append(Turn.user("hi"))
+    session._stream_response(0)
+    kwargs: dict[str, Any] = provider.create_streaming.call_args.kwargs
+    return kwargs
+
+
 class TestResolveReplayReasoningToModel:
-    """Direct unit tests for the resolver."""
+    """Direct unit tests for the resolver.
+
+    The session wrapper this class used to call was deleted with the
+    ``_try_stream`` seam (#832): the lane carries the registry and the
+    alias, and ``model_turn`` reads the flag off them per call.  Same
+    resolver, one indirection down — so the case table is unchanged.
+    """
 
     def test_returns_false_when_no_registry(self) -> None:
-        session = _make_session()
-        session._registry = None
-        session._model_alias = "anything"
-        assert session._resolve_replay_reasoning_to_model() is False
+        assert resolve_replay_reasoning_to_model(None, "anything") is False
 
     def test_returns_false_when_no_alias(self) -> None:
-        session = _make_session()
-        session._registry = _registry_with_flag(replay=True)
-        session._model_alias = ""
-        assert session._resolve_replay_reasoning_to_model() is False
+        # A lane outside the registry carries ``alias=""``.
+        assert resolve_replay_reasoning_to_model(_registry_with_flag(replay=True), "") is False
 
     def test_returns_false_default(self) -> None:
-        session = _make_session()
-        session._registry = _registry_with_flag(replay=False)
-        session._model_alias = "claude-opus-4-7"
-        assert session._resolve_replay_reasoning_to_model() is False
+        registry = _registry_with_flag(replay=False)
+        assert resolve_replay_reasoning_to_model(registry, "claude-opus-4-7") is False
 
     def test_returns_true_when_flag_set(self) -> None:
-        session = _make_session()
-        session._registry = _registry_with_flag(replay=True)
-        session._model_alias = "claude-opus-4-7"
-        assert session._resolve_replay_reasoning_to_model() is True
-
-    def test_explicit_alias_arg_overrides_default(self) -> None:
-        session = _make_session()
+        registry = _registry_with_flag(replay=True)
+        assert resolve_replay_reasoning_to_model(registry, "claude-opus-4-7") is True
 
+    def test_alias_selects_its_own_config(self) -> None:
+        # The flag tracks the alias the LANE resolved, not any session
+        # default — the property the fallback walk depends on (pinned
+        # end-to-end by test_fallback_alias_uses_its_own_flag below).
         def per_alias(alias: str) -> Any:
             return SimpleNamespace(
                 replay_reasoning_to_model=(alias == "needs-replay"),
             )
 
-        session._registry = SimpleNamespace(get_config=per_alias)
-        session._model_alias = "primary"
-        # Default reads session._model_alias → False.
-        assert session._resolve_replay_reasoning_to_model() is False
-        # Explicit alias arg → True for "needs-replay".
-        assert session._resolve_replay_reasoning_to_model("needs-replay") is True
+        registry = SimpleNamespace(get_config=per_alias)
+        assert resolve_replay_reasoning_to_model(registry, "primary") is False
+        assert resolve_replay_reasoning_to_model(registry, "needs-replay") is True
 
     def test_returns_false_on_registry_exception(self) -> None:
-        session = _make_session()
-
         def boom(alias: str) -> Any:
             raise KeyError(alias)
 
-        session._registry = SimpleNamespace(get_config=boom)
-        session._model_alias = "missing"
+        registry = SimpleNamespace(get_config=boom)
         # Conservative fallback — losing the strip is a UX nuisance,
         # but accepting wire-side reasoning replay against an unknown
         # operator preference is a worse default.
-        assert session._resolve_replay_reasoning_to_model() is False
+        assert resolve_replay_reasoning_to_model(registry, "missing") is False
 
     def test_caps_none_preserves_back_compat(self) -> None:
         # When ``caps`` is omitted, the resolver returns the operator
         # flag unchanged — matching pre-PR behaviour for any caller
         # that hasn't been updated to thread caps yet.
-        session = _make_session()
-        session._registry = _registry_with_flag(replay=True)
-        session._model_alias = "claude-opus-4-7"
-        assert session._resolve_replay_reasoning_to_model() is True
-        assert session._resolve_replay_reasoning_to_model(caps=None) is True
+        registry = _registry_with_flag(replay=True)
+        assert resolve_replay_reasoning_to_model(registry, "claude-opus-4-7") is True
+        assert resolve_replay_reasoning_to_model(registry, "claude-opus-4-7", caps=None) is True
 
     def test_caps_supports_replay_true_passes_through(self) -> None:
-        from turnstone.core.providers._protocol import ModelCapabilities
-
-        session = _make_session()
-        session._registry = _registry_with_flag(replay=True)
-        session._model_alias = "claude-opus-4-7"
+        registry = _registry_with_flag(replay=True)
         caps = ModelCapabilities(supports_reasoning_replay=True)
-        assert session._resolve_replay_reasoning_to_model(caps=caps) is True
+        assert resolve_replay_reasoning_to_model(registry, "claude-opus-4-7", caps=caps) is True
 
     def test_caps_supports_replay_false_blocks_replay(self) -> None:
-        from turnstone.core.providers._protocol import ModelCapabilities
-
         # Operator flipped replay=True but the model's capability
         # advertises supports_reasoning_replay=False — AND-gate blocks
         # replay so the strip predicate runs at the wire build.
-        session = _make_session()
-        session._registry = _registry_with_flag(replay=True)
-        session._model_alias = "hypothetical-no-replay-claude"
+        registry = _registry_with_flag(replay=True)
         caps = ModelCapabilities(supports_reasoning_replay=False)
-        assert session._resolve_replay_reasoning_to_model(caps=caps) is False
+        alias = "hypothetical-no-replay-claude"
+        assert resolve_replay_reasoning_to_model(registry, alias, caps=caps) is False
 
     def test_caps_supports_replay_true_does_not_force_replay(self) -> None:
-        from turnstone.core.providers._protocol import ModelCapabilities
-
         # Capability True but operator flag False — result must be
         # False (the AND has to be False on either side).
-        session = _make_session()
-        session._registry = _registry_with_flag(replay=False)
-        session._model_alias = "claude-opus-4-7"
+        registry = _registry_with_flag(replay=False)
         caps = ModelCapabilities(supports_reasoning_replay=True)
-        assert session._resolve_replay_reasoning_to_model(caps=caps) is False
+        assert resolve_replay_reasoning_to_model(registry, "claude-opus-4-7", caps=caps) is False
 
 
 class TestStreamingCallSitePassesFlag:
-    """Pin that ``_try_stream`` actually passes the resolved flag to
+    """Pin that the streaming turn actually passes the resolved flag to
     ``provider.create_streaming`` — without this the Phase 2 work is
-    dead code at the call site."""
+    dead code at the call site.
+
+    The call site is ``model_turn`` now, driven through the real
+    ``_stream_response`` wrapper (creation walk → drain → finalize), so
+    the flag rides the lane the walk actually served the turn on.
+    """
 
     def test_replay_true_propagates_to_provider(self) -> None:
         session = _make_session()
         session._registry = _registry_with_flag(replay=True)
         session._model_alias = "claude-opus-4-7"
-        # Stub provider: capture the kwargs passed to create_streaming.
-        captured: dict[str, Any] = {}
-
-        def capture_streaming(**kwargs: Any) -> Any:
-            captured.update(kwargs)
-            return iter([])
-
-        mock_provider = MagicMock()
-        mock_provider.create_streaming = capture_streaming
-        with (
-            patch.object(session, "_get_active_tools", return_value=None),
-            patch.object(session, "_provider_extra_params", return_value=None),
-            patch.object(session, "_get_deferred_names", return_value=frozenset()),
-            patch.object(session, "_check_cancelled"),
-        ):
-            session._try_stream(
-                client=MagicMock(),
-                model="claude-opus-4-7",
-                msgs=[{"role": "user", "content": "hi"}],
-                provider=mock_provider,
-                model_alias="claude-opus-4-7",
-            )
-        assert captured["replay_reasoning_to_model"] is True
+        kwargs = _drive_stream(session, _flag_capture_provider(supports_replay=True))
+        assert kwargs["replay_reasoning_to_model"] is True
 
     def test_replay_false_propagates_to_provider(self) -> None:
         session = _make_session()
         session._registry = _registry_with_flag(replay=False)
         session._model_alias = "claude-opus-4-7"
-        captured: dict[str, Any] = {}
-
-        def capture_streaming(**kwargs: Any) -> Any:
-            captured.update(kwargs)
-            return iter([])
-
-        mock_provider = MagicMock()
-        mock_provider.create_streaming = capture_streaming
-        with (
-            patch.object(session, "_get_active_tools", return_value=None),
-            patch.object(session, "_provider_extra_params", return_value=None),
-            patch.object(session, "_get_deferred_names", return_value=frozenset()),
-            patch.object(session, "_check_cancelled"),
-        ):
-            session._try_stream(
-                client=MagicMock(),
-                model="claude-opus-4-7",
-                msgs=[{"role": "user", "content": "hi"}],
-                provider=mock_provider,
-                model_alias="claude-opus-4-7",
-            )
-        assert captured["replay_reasoning_to_model"] is False
+        # Capability advertises replay support: the False comes from the
+        # operator flag alone, not from the AND-gate's other half.
+        kwargs = _drive_stream(session, _flag_capture_provider(supports_replay=True))
+        assert kwargs["replay_reasoning_to_model"] is False
 
     def test_fallback_alias_uses_its_own_flag(self) -> None:
         # When the primary fails and we fall back to an alias with a
@@ -209,52 +200,45 @@ class TestStreamingCallSitePassesFlag:
         def per_alias(alias: str) -> Any:
             return SimpleNamespace(
                 replay_reasoning_to_model=(alias == "fallback-with-replay"),
+                capabilities={},
             )
 
-        session._registry = SimpleNamespace(get_config=per_alias)
+        fb_provider = _flag_capture_provider(supports_replay=True)
+        session._registry = SimpleNamespace(
+            get_config=per_alias,
+            fallback=["fallback-with-replay"],
+            resolve_binding=lambda alias: (MagicMock(), "fallback-model", None, fb_provider, None),
+        )
         session._model_alias = "primary"  # primary has replay=False
-        captured: dict[str, Any] = {}
-
-        def capture_streaming(**kwargs: Any) -> Any:
-            captured.update(kwargs)
-            return iter([])
-
-        mock_provider = MagicMock()
-        mock_provider.create_streaming = capture_streaming
-        with (
-            patch.object(session, "_get_active_tools", return_value=None),
-            patch.object(session, "_provider_extra_params", return_value=None),
-            patch.object(session, "_get_deferred_names", return_value=frozenset()),
-            patch.object(session, "_check_cancelled"),
-        ):
-            session._try_stream(
-                client=MagicMock(),
-                model="fallback-model",
-                msgs=[{"role": "user", "content": "hi"}],
-                provider=mock_provider,
-                model_alias="fallback-with-replay",
-            )
+        # The primary lane dies at CREATION (raises without arming its
+        # cancel_ref), which is what sends the walk to the next alias; a
+        # non-retryable class keeps the ladder from burning backoff.
+        primary = _flag_capture_provider(supports_replay=True)
+        primary.create_streaming = MagicMock(side_effect=RuntimeError("primary is down"))
+        _drive_stream(session, primary)
         # Resolved against the FALLBACK alias, not the session's primary.
-        assert captured["replay_reasoning_to_model"] is True
+        assert fb_provider.create_streaming.call_args.kwargs["replay_reasoning_to_model"] is True
+        # ...and the primary's own attempt resolved its own alias' flag.
+        assert primary.create_streaming.call_args.kwargs["replay_reasoning_to_model"] is False
 
 
 class TestSessionToWireBoundaryIntegration:
-    """End-to-end integration: session._try_stream -> real
-    AnthropicProvider.create_streaming -> captured Anthropic SDK
+    """End-to-end integration: session._stream_response -> model_turn ->
+    real AnthropicProvider.create_streaming -> captured Anthropic SDK
     boundary call.  Verifies the strip-when-False predicate actually
     fires at the wire payload, not just at the captured kwarg.
 
-    The bare-function-stub tests above (TestStreamingCallSitePassesFlag)
-    pin that ``_try_stream`` PASSES the flag; this test pins that the
-    real provider USES it.  Together they catch:
-      - kwarg renamed at provider boundary -> stub-tests still pass,
-        this one fails on its real-provider assertion.
-      - _convert_messages stops reading the kwarg -> stub-tests still
-        pass, this one fails because the wire payload still carries
+    The provider-fake tests above (TestStreamingCallSitePassesFlag) pin
+    that the streaming turn PASSES the flag; this test pins that the real
+    provider USES it.  Together they catch:
+      - kwarg renamed at provider boundary -> fake-provider tests still
+        pass, this one fails on its real-provider assertion.
+      - _convert_messages stops reading the kwarg -> fake-provider tests
+        still pass, this one fails because the wire payload still carries
         the thinking block.
-      - _try_stream stops calling create_streaming -> stub-tests fail
-        on the captured kwarg, this one fails because the SDK boundary
-        was never reached.
+      - the streaming turn stops calling create_streaming -> fake-provider
+        tests fail on the captured kwarg, this one fails because the SDK
+        boundary was never reached.
 
     Drives through the real ``AnthropicProvider`` with a mock client
     whose ``client.messages.stream`` is captured — the smallest possible
@@ -273,19 +257,16 @@ class TestSessionToWireBoundaryIntegration:
     def _stub_anthropic_client(self) -> tuple[MagicMock, dict[str, object]]:
         """Build a mock Anthropic client + captured-kwargs dict.
 
-        ``client.messages.stream(**kwargs)`` returns a context manager
-        whose ``__enter__`` yields an iterable of zero events — enough
-        to satisfy the ``_iter_with_cleanup`` shape without exercising
-        actual streaming protocol.
+        ``client.messages.stream(**kwargs)`` returns the real event
+        grammar for a one-block reply, so the fused create+drain reaches
+        a finish reason instead of exhausting finish-less (which the
+        post-#832 seam re-issues as an ``IncompleteStreamError``).
         """
         captured: dict[str, object] = {}
 
         def stream(**kwargs: object) -> object:
             captured.update(kwargs)
-            cm = MagicMock()
-            cm.__enter__ = MagicMock(return_value=iter([]))
-            cm.__exit__ = MagicMock(return_value=False)
-            return cm
+            return fake_anthropic_stream([FakeAnthropicBlock(type="text", text="ok")])
 
         client = MagicMock()
         client.messages.stream = stream
@@ -295,38 +276,31 @@ class TestSessionToWireBoundaryIntegration:
         self,
         replay_flag: bool,
         msgs: list[dict[str, object]],
+        caps_overrides: dict[str, Any] | None = None,
     ) -> dict[str, object]:
-        """Run session._try_stream against a real AnthropicProvider with
-        the resolver pre-set to *replay_flag*.  Returns the kwargs
+        """Run one real streaming turn against a real AnthropicProvider
+        with the resolver pre-set to *replay_flag*.  Returns the kwargs
         dict that reached the (mocked) Anthropic SDK boundary.
+
+        *msgs* seeds the session trajectory (the wire list is built from
+        it inside ``model_turn`` now — lowering plus the session's own
+        ``prepare_wire`` passes — instead of being handed to the seam).
         """
         from turnstone.core.providers._anthropic import AnthropicProvider
 
         session = _make_session()
-        session._registry = _registry_with_flag(replay=replay_flag)
+        session._registry = _registry_with_flag(replay=replay_flag, caps_overrides=caps_overrides)
         session._model_alias = "claude-opus-4-7"
+        session.model = "claude-opus-4-7"
         client, captured = self._stub_anthropic_client()
-        real_provider = AnthropicProvider()
-        with (
-            patch.object(session, "_get_active_tools", return_value=None),
-            patch.object(session, "_provider_extra_params", return_value=None),
-            patch.object(session, "_get_deferred_names", return_value=frozenset()),
-            patch.object(session, "_check_cancelled"),
-        ):
-            stream = session._try_stream(
-                client=client,
-                model="claude-opus-4-7",
-                msgs=msgs,
-                provider=real_provider,
-                model_alias="claude-opus-4-7",
-            )
-            # Iterate the stream to drain the (empty) generator and ensure
-            # convert / build_kwargs all ran.
-            list(stream)
+        session.client = client
+        session._provider = AnthropicProvider()
+        session.messages = turns_from_dicts(msgs)
+        session._stream_response(0)
         return captured
 
     def test_replay_false_strips_thinking_at_wire(self) -> None:
-        msgs: list[dict[str, object]] = [
+        msgs: list[dict[str, Any]] = [
             {"role": "user", "content": "hello"},
             {
                 "role": "assistant",
@@ -357,7 +331,7 @@ class TestSessionToWireBoundaryIntegration:
         assert "secret reasoning" not in flat, "Reasoning text leaked into the SDK boundary payload"
 
     def test_replay_true_preserves_thinking_at_wire(self) -> None:
-        msgs: list[dict[str, object]] = [
+        msgs: list[dict[str, Any]] = [
             {"role": "user", "content": "hello"},
             {
                 "role": "assistant",
@@ -384,11 +358,11 @@ class TestSessionToWireBoundaryIntegration:
         # replay=True but the model's capability advertises
         # supports_reasoning_replay=False.  AND-gate at the resolver
         # blocks replay, so the strip predicate fires at the wire and
-        # the thinking block does NOT reach the SDK boundary.
-        from turnstone.core.providers._anthropic import AnthropicProvider
-        from turnstone.core.providers._protocol import ModelCapabilities
-
-        msgs: list[dict[str, object]] = [
+        # the thinking block does NOT reach the SDK boundary.  The
+        # capability reaches the resolver as the LANE's — an operator
+        # override on the alias, since the lane resolves its own caps
+        # rather than taking them from the caller.
+        msgs: list[dict[str, Any]] = [
             {"role": "user", "content": "hello"},
             {
                 "role": "assistant",
@@ -401,27 +375,11 @@ class TestSessionToWireBoundaryIntegration:
             {"role": "user", "content": "ack"},
         ]
 
-        session = _make_session()
-        session._registry = _registry_with_flag(replay=True)  # operator opted in
-        session._model_alias = "hypothetical-no-replay-claude"
-        caps = ModelCapabilities(supports_reasoning_replay=False)
-        client, captured = self._stub_anthropic_client()
-        real_provider = AnthropicProvider()
-        with (
-            patch.object(session, "_get_active_tools", return_value=None),
-            patch.object(session, "_provider_extra_params", return_value=None),
-            patch.object(session, "_get_deferred_names", return_value=frozenset()),
-            patch.object(session, "_check_cancelled"),
-        ):
-            stream = session._try_stream(
-                client=client,
-                model="hypothetical-no-replay-claude",
-                msgs=msgs,
-                provider=real_provider,
-                capabilities=caps,
-                model_alias="hypothetical-no-replay-claude",
-            )
-            list(stream)
+        captured = self._drive_session_through_anthropic(
+            True,  # operator opted in
+            msgs,
+            caps_overrides={"supports_reasoning_replay": False},
+        )
 
         wire_msgs = captured.get("messages")
         assert isinstance(wire_msgs, list), (
@@ -440,8 +398,8 @@ class TestSessionToWireBoundaryIntegration:
 
 
 class TestSessionToOpenAIResponsesBoundaryIntegration:
-    """End-to-end integration: session._try_stream -> real
-    OpenAIResponsesProvider.create_streaming -> captured Responses
+    """End-to-end integration: session._stream_response -> model_turn ->
+    real OpenAIResponsesProvider.create_streaming -> captured Responses
     SDK boundary call.  Mirrors the AnthropicProvider test above
     but for the path-2 (Responses API) replay flow.
 
@@ -452,12 +410,14 @@ class TestSessionToOpenAIResponsesBoundaryIntegration:
 
     def _stub_responses_client(self) -> tuple[MagicMock, dict[str, object]]:
         """Mock OpenAI Responses client.  ``client.responses.create``
-        captures kwargs and returns an empty stream iterator."""
+        captures kwargs and returns a stream carrying only the terminal
+        event — the fused create+drain needs a finish reason (a
+        finish-less exhaust is an ``IncompleteStreamError`` post-#832)."""
         captured: dict[str, object] = {}
 
         def create(**kwargs: object) -> object:
             captured.update(kwargs)
-            return iter([])
+            return iter([SimpleNamespace(type="response.completed", response=None)])
 
         client = MagicMock()
         client.responses.create = create
@@ -466,114 +426,71 @@ class TestSessionToOpenAIResponsesBoundaryIntegration:
     def _registry_with_reasoning_capability(
         self, replay: bool = True, supports_replay: bool = True
     ) -> Any:
-        from turnstone.core.providers._protocol import ModelCapabilities
-
+        """Stub registry stating both halves of the gate: the operator
+        flag and — as an alias capability override, the operator's way to
+        state one — the model's reasoning-replay support.  The lane
+        resolves its own capabilities now, so this is where a test says
+        what the model can do."""
         return SimpleNamespace(
             get_config=lambda alias: SimpleNamespace(
                 replay_reasoning_to_model=replay,
-                capabilities={},  # no overrides
-            ),
-            _caps=ModelCapabilities(
-                context_window=400000,
-                supports_temperature=False,
-                reasoning_effort_values=("low", "medium", "high"),
-                default_reasoning_effort="medium",
-                supports_reasoning_replay=supports_replay,
+                capabilities={"supports_reasoning_replay": supports_replay},
             ),
         )
 
-    def test_replay_true_adds_include_to_responses_request(self) -> None:
+    def _drive(self, session: Any, msgs: list[dict[str, Any]]) -> dict[str, object]:
+        """One real streaming turn through the real Responses provider."""
         from turnstone.core.providers._openai_responses import OpenAIResponsesProvider
 
-        registry = self._registry_with_reasoning_capability(replay=True, supports_replay=True)
-        session = _make_session()
-        session._registry = registry
-        session._model_alias = "gpt-5"
         client, captured = self._stub_responses_client()
-        real_provider = OpenAIResponsesProvider()
-        with (
-            patch.object(session, "_get_active_tools", return_value=None),
-            patch.object(session, "_provider_extra_params", return_value=None),
-            patch.object(session, "_get_deferred_names", return_value=frozenset()),
-            patch.object(session, "_check_cancelled"),
-        ):
-            stream = session._try_stream(
-                client=client,
-                model="gpt-5",
-                msgs=[{"role": "user", "content": "hi"}],
-                provider=real_provider,
-                capabilities=registry._caps,
-                model_alias="gpt-5",
-            )
-            list(stream)
+        session.client = client
+        session._provider = OpenAIResponsesProvider()
+        session.messages = turns_from_dicts(msgs)
+        session._stream_response(0)
+        return captured
+
+    def test_replay_true_adds_include_to_responses_request(self) -> None:
+        session = _make_session()
+        session._registry = self._registry_with_reasoning_capability(
+            replay=True, supports_replay=True
+        )
+        session._model_alias = "gpt-5"
+        session.model = "gpt-5"
+        captured = self._drive(session, [{"role": "user", "content": "hi"}])
         assert captured.get("include") == ["reasoning.encrypted_content"]
 
     def test_replay_false_omits_include(self) -> None:
-        from turnstone.core.providers._openai_responses import OpenAIResponsesProvider
-
-        registry = self._registry_with_reasoning_capability(replay=False, supports_replay=True)
         session = _make_session()
-        session._registry = registry
+        session._registry = self._registry_with_reasoning_capability(
+            replay=False, supports_replay=True
+        )
         session._model_alias = "gpt-5"
-        client, captured = self._stub_responses_client()
-        real_provider = OpenAIResponsesProvider()
-        with (
-            patch.object(session, "_get_active_tools", return_value=None),
-            patch.object(session, "_provider_extra_params", return_value=None),
-            patch.object(session, "_get_deferred_names", return_value=frozenset()),
-            patch.object(session, "_check_cancelled"),
-        ):
-            stream = session._try_stream(
-                client=client,
-                model="gpt-5",
-                msgs=[{"role": "user", "content": "hi"}],
-                provider=real_provider,
-                capabilities=registry._caps,
-                model_alias="gpt-5",
-            )
-            list(stream)
+        session.model = "gpt-5"
+        captured = self._drive(session, [{"role": "user", "content": "hi"}])
         assert "include" not in captured
 
     def test_capability_false_omits_include_even_when_flag_true(self) -> None:
-        from turnstone.core.providers._openai_responses import OpenAIResponsesProvider
-
         # Operator flips replay=True but the model has
         # supports_reasoning_replay=False (e.g. gpt-4o via Responses).
         # Capability gate prevents the include= from being sent.
-        registry = self._registry_with_reasoning_capability(replay=True, supports_replay=False)
         session = _make_session()
-        session._registry = registry
+        session._registry = self._registry_with_reasoning_capability(
+            replay=True, supports_replay=False
+        )
         session._model_alias = "gpt-4o"
-        client, captured = self._stub_responses_client()
-        real_provider = OpenAIResponsesProvider()
-        with (
-            patch.object(session, "_get_active_tools", return_value=None),
-            patch.object(session, "_provider_extra_params", return_value=None),
-            patch.object(session, "_get_deferred_names", return_value=frozenset()),
-            patch.object(session, "_check_cancelled"),
-        ):
-            stream = session._try_stream(
-                client=client,
-                model="gpt-4o",
-                msgs=[{"role": "user", "content": "hi"}],
-                provider=real_provider,
-                capabilities=registry._caps,
-                model_alias="gpt-4o",
-            )
-            list(stream)
+        session.model = "gpt-4o"
+        captured = self._drive(session, [{"role": "user", "content": "hi"}])
         assert "include" not in captured
 
     def test_replay_true_emits_reasoning_input_item(self) -> None:
-        from turnstone.core.providers._openai_responses import OpenAIResponsesProvider
-
-        registry = self._registry_with_reasoning_capability(replay=True, supports_replay=True)
         session = _make_session()
-        session._registry = registry
+        session._registry = self._registry_with_reasoning_capability(
+            replay=True, supports_replay=True
+        )
         session._model_alias = "gpt-5"
-        client, captured = self._stub_responses_client()
-        real_provider = OpenAIResponsesProvider()
+        session.model = "gpt-5"
         # Multi-turn conversation with stored reasoning on assistant turn.
-        msgs: list[dict[str, object]] = [
+        msgs: list[dict[str, Any]] = [
             {"role": "user", "content": "explain"},
             {
                 "role": "assistant",
@@ -589,21 +506,7 @@ class TestSessionToOpenAIResponsesBoundaryIntegration:
             },
             {"role": "user", "content": "follow-up"},
         ]
-        with (
-            patch.object(session, "_get_active_tools", return_value=None),
-            patch.object(session, "_provider_extra_params", return_value=None),
-            patch.object(session, "_get_deferred_names", return_value=frozenset()),
-            patch.object(session, "_check_cancelled"),
-        ):
-            stream = session._try_stream(
-                client=client,
-                model="gpt-5",
-                msgs=msgs,
-                provider=real_provider,
-                capabilities=registry._caps,
-                model_alias="gpt-5",
-            )
-            list(stream)
+        captured = self._drive(session, msgs)
         # Walk the wire input items — one of them must be the reasoning
         # round-trip (id matches what we stored).
         wire_input = captured.get("input")
@@ -619,8 +522,6 @@ class TestUtilityCompletionPassesFlag:
     same plumbing requirement as streaming."""
 
     def test_utility_completion_passes_resolved_flag(self) -> None:
-        from turnstone.core.providers._protocol import ModelCapabilities
-
         session = _make_session()
         session._registry = _registry_with_flag(replay=True)
         session._model_alias = "claude-opus-4-7"
@@ -634,9 +535,9 @@ class TestUtilityCompletionPassesFlag:
         mock_provider.create_streaming = capture_streaming
         session._provider = mock_provider
         caps = ModelCapabilities(max_output_tokens=0, supports_reasoning_replay=True)
-        # No _provider_extra_params patch: _utility_completion resolves
-        # extra_params inside resolve_lane (module seam), which a
-        # session-attribute patch cannot intercept.
+        # No extra_params patch: _utility_completion resolves them inside
+        # resolve_lane (a module seam reading the registry config), which
+        # a session-attribute patch cannot intercept.
         with patch.object(session, "_get_capabilities", return_value=caps):
             session._utility_completion(
                 [Turn.user("summarize")],
diff --git a/tests/test_session_synth_reasoning_block.py b/tests/test_session_synth_reasoning_block.py
index 53d94c6e..87a6ec62 100644
--- a/tests/test_session_synth_reasoning_block.py
+++ b/tests/test_session_synth_reasoning_block.py
@@ -8,9 +8,8 @@ provider_blocks shape on the wire, so the captured reasoning text is
 stamped onto ``_provider_content`` as a synthetic
 ``{type: "reasoning_text"}`` block at the end of the turn by
 ``model_turn.synth_reasoning_block`` — the one synthesizer every lane
-runs (the main loop reaches it through ``ChatSession._stream_attempt``
-→ ``_finalize_provider_blocks``; agents and judges through
-``model_turn``).
+runs (every lane — main loop, agents, judges — reaches it through
+``model_turn.model_turn``'s ``finalize_provider_blocks`` call).
 
 These tests pin:
 1. The synthesizer fires only when no native blocks were emitted AND
@@ -29,6 +28,7 @@ from __future__ import annotations
 from types import SimpleNamespace
 from typing import Any
 
+from tests._parity_832 import scripted_provider
 from tests._session_helpers import make_session as _make_session
 from turnstone.core.model_turn import _server_type_of, synth_reasoning_block
 from turnstone.core.providers._anthropic import (
@@ -36,6 +36,8 @@ from turnstone.core.providers._anthropic import (
     AnthropicProvider,
 )
 from turnstone.core.providers._openai_chat import OpenAIChatCompletionsProvider
+from turnstone.core.providers._protocol import StreamChunk, UsageInfo
+from turnstone.core.trajectory import Turn
 
 
 class TestSynthReasoningBlock:
@@ -210,25 +212,24 @@ class TestOpenAIChatExtractReasoningText:
         assert provider.extract_reasoning_text(blocks) == ""
 
 
-class TestStreamAttemptSynthBlockIntegration:
+class TestStreamResponseSynthBlockIntegration:
     """Integration test: drives a fake reasoning-emitting stream
-    through ``ChatSession._stream_attempt`` and asserts the
-    synthesizer wires up correctly.  Pins the call site at
-    ``session.py`` (where ``model_turn.synth_reasoning_block`` is
-    invoked — via ``_finalize_provider_blocks`` — on the assembled
-    provider_blocks before stamping ``_provider_content``) — without
+    through ``session._stream_response`` (the real drain seam —
+    ``_stream_attempt`` no longer exists post-#832) and asserts the
+    synthesizer wires up correctly.  Pins the call site inside
+    ``model_turn.model_turn`` (where ``model_turn.synth_reasoning_block``
+    is invoked — via ``finalize_provider_blocks`` — on the assembled
+    provider_blocks before the result's native lane is built) — without
     this, a future refactor that drops the synthesizer call would
     silently break path-3 capture (vLLM/llama.cpp/Gemini-compat
     reasoning would be visible live but invisible on history reload).
     """
 
-    def _make_stream(self, content: str, reasoning: str) -> Any:
-        """Build an iterator of StreamChunks that mimic a path-3
-        capture (reasoning_delta chunks, content chunks, no
-        provider_blocks emitted).
+    def _make_chunks(self, content: str, reasoning: str) -> list[StreamChunk]:
+        """Build a chunk script that mimics a path-3 capture
+        (reasoning_delta chunks, content chunks, no provider_blocks
+        emitted).
         """
-        from turnstone.core.providers._protocol import StreamChunk, UsageInfo
-
         chunks = []
         # Reasoning first (matches live SSE order).
         if reasoning:
@@ -248,39 +249,44 @@ class TestStreamAttemptSynthBlockIntegration:
                 usage=UsageInfo(prompt_tokens=10, completion_tokens=20, total_tokens=30),
             )
         )
-        return iter(chunks)
+        return chunks
 
-    def test_stream_attempt_stamps_synth_block_when_path3_reasoning_captured(
+    def test_stream_response_stamps_synth_block_when_path3_reasoning_captured(
         self,
     ) -> None:
         """Drive a fake stream emitting reasoning_delta chunks (no
-        native provider_blocks) through ``_stream_attempt``; assert
-        the resulting assistant_msg carries a synthetic reasoning_text
-        block stamped onto ``_provider_content``."""
+        native provider_blocks) through ``_stream_response``; assert
+        the resulting turn carries a synthetic reasoning_text block on
+        its native lane."""
         session = _make_session()
         # No registry → source field omitted from synth block.
-        stream = self._make_stream(content="Final answer.", reasoning="path-3 reasoning")
-        msg = session._stream_attempt(stream)
-        assert msg["role"] == "assistant"
-        assert msg["content"] == "Final answer."
-        # Synthetic block should be stamped onto _provider_content.
-        provider_content = msg.get("_provider_content")
-        assert isinstance(provider_content, list)
-        assert len(provider_content) == 1
-        assert provider_content[0]["type"] == "reasoning_text"
-        assert provider_content[0]["text"] == "path-3 reasoning"
+        session._provider = scripted_provider(
+            self._make_chunks(content="Final answer.", reasoning="path-3 reasoning")
+        )
+        session.messages.append(Turn.user("hi"))
+        result = session._stream_response(0)
+        assert result.content == "Final answer."
+        # Synthetic block should be stamped onto the native lane.
+        assert result.turn.native is not None
+        blocks = list(result.turn.native.blocks)
+        assert len(blocks) == 1
+        assert blocks[0]["type"] == "reasoning_text"
+        assert blocks[0]["text"] == "path-3 reasoning"
 
-    def test_stream_attempt_no_synth_when_no_reasoning_captured(self) -> None:
+    def test_stream_response_no_synth_when_no_reasoning_captured(self) -> None:
         """Stream emits only content (no reasoning_delta).  No synth
-        block stamped — _provider_content key absent on assistant_msg."""
+        block stamped — the result's native lane is absent."""
         session = _make_session()
-        stream = self._make_stream(content="just content", reasoning="")
-        msg = session._stream_attempt(stream)
-        assert msg["content"] == "just content"
-        # No synth block (and no native blocks either) → key absent.
-        assert "_provider_content" not in msg
+        session._provider = scripted_provider(
+            self._make_chunks(content="just content", reasoning="")
+        )
+        session.messages.append(Turn.user("hi"))
+        result = session._stream_response(0)
+        assert result.content == "just content"
+        # No synth block (and no native blocks either) → native absent.
+        assert result.turn.native is None
 
-    def test_stream_attempt_synth_block_carries_source_when_server_type_resolvable(
+    def test_stream_response_synth_block_carries_source_when_server_type_resolvable(
         self,
     ) -> None:
         """When the active model has server_compat.server_type set,
@@ -293,11 +299,14 @@ class TestStreamAttemptSynthBlockIntegration:
             )
         )
         session._model_alias = "qwen3-32b"
-        stream = self._make_stream(content="answer", reasoning="reasoning text")
-        msg = session._stream_attempt(stream)
-        provider_content = msg.get("_provider_content")
-        assert isinstance(provider_content, list)
-        assert provider_content[0]["source"] == "vllm"
+        session._provider = scripted_provider(
+            self._make_chunks(content="answer", reasoning="reasoning text")
+        )
+        session.messages.append(Turn.user("hi"))
+        result = session._stream_response(0)
+        assert result.turn.native is not None
+        blocks = list(result.turn.native.blocks)
+        assert blocks[0]["source"] == "vllm"
 
 
 class TestServerTypeOf:
diff --git a/tests/test_think_tag_split.py b/tests/test_think_tag_split.py
index a5bf53ca..1021a9c2 100644
--- a/tests/test_think_tag_split.py
+++ b/tests/test_think_tag_split.py
@@ -1,10 +1,12 @@
 """Behavior pins for the interactive think-tag splitting layer.
 
-``ChatSession._stream_attempt`` splits streamed content into content vs
-reasoning around ``<think>``/``<reasoning>`` tags, buffering potential
-partial tags across chunk boundaries.  These tables pin the CURRENT
-emission behavior — exact UI token sequence and final message content —
-so the logic can move into a standalone ``ThinkTagSplitter`` class with
+``turnstone.core.session._StreamTurnConsumer`` (the main loop's
+chunk→UI translation, ``model_turn``'s ``on_chunk`` body post-#832)
+splits streamed content into content vs reasoning around
+``<think>``/``<reasoning>`` tags, buffering potential partial tags
+across chunk boundaries.  These tables pin the CURRENT emission
+behavior — exact UI token sequence and final displayed content — so
+the logic can move into a standalone ``ThinkTagSplitter`` class with
 byte-identical output.  Every case drives the real chunk consumer end to
 end; none reaches into the implementation, so the same rows must stay
 green across the extraction.
@@ -23,18 +25,23 @@ Pinned rules:
 """
 
 import random
+from unittest.mock import MagicMock
 
 import pytest
 
+from tests._parity_832 import scripted_provider
 from tests._reasoning_dialect import CASES as DIALECT_CASES
 from tests._session_helpers import make_session
+from turnstone.core.model_turn import ModelLane
 from turnstone.core.providers import StreamChunk, ToolCallDelta
+from turnstone.core.session import _StreamTurnConsumer
 from turnstone.core.streaming_text import ThinkTagSplitter, split_inline_reasoning
+from turnstone.core.trajectory import Turn
 
 
 class _TokenRecorderUI:
     """Records dispatched token events; stubs the rest of the surface
-    ``_stream_attempt`` touches."""
+    ``_StreamTurnConsumer`` touches."""
 
     def __init__(self):
         self.tokens = []
@@ -58,13 +65,24 @@ class _TokenRecorderUI:
         pass
 
 
-def _drive(chunks, *, show_reasoning=True):
+def _drive(chunks, *, show_reasoning=True, capabilities=None):
+    """Drive *chunks* through a bare ``_StreamTurnConsumer`` — the
+    display-grid seam post-#832 (``_stream_attempt`` is gone; tool_calls
+    assembly is the drain's job now, so this helper is for display-only
+    pins — content emission order plus the accumulated displayed text).
+    """
     session = make_session()
     session.show_reasoning = show_reasoning
     ui = _TokenRecorderUI()
     session.ui = ui
-    msg = session._stream_attempt(iter(chunks))
-    return msg, ui.tokens
+    lane = ModelLane(
+        provider=MagicMock(), client=MagicMock(), model="test-model", capabilities=capabilities
+    )
+    consumer = _StreamTurnConsumer(session, lane, 0)
+    for chunk in chunks:
+        consumer(chunk)
+    consumer.finish_stream()
+    return "".join(consumer._content_parts), ui.tokens
 
 
 def _c(text):
@@ -150,15 +168,15 @@ CASES = [
     ids=[c[0] for c in CASES],
 )
 def test_tag_splitting_emissions(chunks, expected_events, expected_content):
-    msg, tokens = _drive(chunks)
+    content, tokens = _drive(chunks)
     assert tokens == expected_events
-    assert msg["content"] == expected_content
+    assert content == expected_content
 
 
 def test_show_reasoning_off_suppresses_reasoning_dispatch_only():
-    msg, tokens = _drive([_c("<think>deep</think>answer"), _FINISH], show_reasoning=False)
+    content, tokens = _drive([_c("<think>deep</think>answer"), _FINISH], show_reasoning=False)
     assert tokens == [("content", "answer")]
-    assert msg["content"] == "answer"
+    assert content == "answer"
 
 
 def test_splitter_standalone_contract():
@@ -213,19 +231,19 @@ def test_scan_tags_off_returns_every_utterance_byte_identical(case):
 
 def test_session_consumer_scan_follows_server_parses_reasoning():
     """The interactive consumer wires ``scan_tags`` from the SAME capability
-    the drain seam reads (``server_parses_reasoning``), so the two lanes
-    cannot disagree.  With the flag declared, streamed tag text reaches the
-    UI verbatim as content — it is prose on such a backend, not a
-    boundary."""
+    the drain seam reads (``server_parses_reasoning``), taken off the
+    ACTIVE lane post-#832 (no more session-level ``_cached_capabilities``
+    read at the consumer).  With the flag declared, streamed tag text
+    reaches the UI verbatim as content — it is prose on such a backend,
+    not a boundary."""
     from turnstone.core.providers._protocol import ModelCapabilities
 
-    session = make_session()
-    session._cached_capabilities = ModelCapabilities(server_parses_reasoning=True)
-    ui = _TokenRecorderUI()
-    session.ui = ui
-    msg = session._stream_attempt(iter([_c("<think>quoted</think>answer"), _FINISH]))
-    assert msg["content"] == "<think>quoted</think>answer"
-    assert all(kind == "content" for kind, _ in ui.tokens)
+    content, tokens = _drive(
+        [_c("<think>quoted</think>answer"), _FINISH],
+        capabilities=ModelCapabilities(server_parses_reasoning=True),
+    )
+    assert content == "<think>quoted</think>answer"
+    assert all(kind == "content" for kind, _ in tokens)
 
 
 def test_scan_tags_off_holds_no_carry_and_honors_out_of_band_state():
@@ -268,7 +286,12 @@ def test_one_shot_equivalent_to_streaming_over_random_chunkings(case):
 
 def test_tool_calls_flush_pending_raw_at_current_state():
     # Once tool calls begin, buffered text cannot be a partial tag: it
-    # flushes RAW (no tag scan) at the current in_think state.
+    # flushes RAW (no tag scan) at the current in_think state.  Tool_calls
+    # assembly is the drain's job post-#832 (the display consumer only
+    # flushes the splitter at the boundary), so this one needs the real
+    # seam — session._stream_response over scripted_provider — to pin the
+    # display order and the assembled call together, as the fused
+    # pre-fold consumer did.
     chunks = [
         _c("part<thi"),
         StreamChunk(tool_call_deltas=[ToolCallDelta(index=0, id="tc1", name="bash")]),
@@ -277,9 +300,14 @@ def test_tool_calls_flush_pending_raw_at_current_state():
             finish_reason="tool_calls",
         ),
     ]
-    msg, tokens = _drive(chunks)
-    assert tokens == [("content", "part<thi")]
-    assert msg["content"] == "part<thi"
-    assert msg["tool_calls"] == [
+    session = make_session()
+    ui = _TokenRecorderUI()
+    session.ui = ui
+    session._provider = scripted_provider(chunks)
+    session.messages.append(Turn.user("hi"))
+    result = session._stream_response(0)
+    assert ui.tokens == [("content", "part<thi")]
+    assert result.content == "part<thi"
+    assert result.tool_calls == [
         {"id": "tc1", "type": "function", "function": {"name": "bash", "arguments": "{}"}}
     ]
diff --git a/tests/test_tool_truncation.py b/tests/test_tool_truncation.py
index b2122d7e..3007608a 100644
--- a/tests/test_tool_truncation.py
+++ b/tests/test_tool_truncation.py
@@ -7,6 +7,7 @@ from unittest.mock import MagicMock, patch
 
 import pytest
 
+from tests._parity_832 import make_result
 from turnstone.core.session import ChatSession
 from turnstone.core.trajectory import Role, turns_from_dicts
 
@@ -205,12 +206,12 @@ class TestContextOverflowRecovery:
 
         call_count = 0
 
-        def mock_stream_response(msgs, my_generation=0):
+        def mock_stream_response(my_generation=0):
             nonlocal call_count
             call_count += 1
             if call_count == 1:
                 raise Exception("maximum context length exceeded")
-            return {"role": "assistant", "content": "ok"}
+            return make_result(content="ok")
 
         compact_mock = MagicMock()
         with (
@@ -235,12 +236,12 @@ class TestContextOverflowRecovery:
 
         call_count = 0
 
-        def mock_stream_response(msgs, my_generation=0):
+        def mock_stream_response(my_generation=0):
             nonlocal call_count
             call_count += 1
             if call_count == 1:
                 raise Exception("prompt is too long: 250000 tokens > 200000 maximum")
-            return {"role": "assistant", "content": "ok"}
+            return make_result(content="ok")
 
         compact_mock = MagicMock()
         with (
@@ -263,7 +264,7 @@ class TestContextOverflowRecovery:
         with (
             patch.object(
                 session,
-                "_create_stream_with_retry",
+                "_stream_response",
                 side_effect=Exception("authentication failed"),
             ),
             patch.object(session, "_full_messages", return_value=[]),
@@ -280,7 +281,7 @@ class TestContextOverflowRecovery:
         with (
             patch.object(
                 session,
-                "_create_stream_with_retry",
+                "_stream_response",
                 side_effect=Exception("maximum context length exceeded"),
             ),
             patch.object(session, "_compact_messages", side_effect=RuntimeError("compact failed")),
@@ -302,13 +303,13 @@ def _send_with_tool_batches(session, batches, **extra_patches):
     """Drive one ``send()`` through the tool-execution drain with canned results.
 
     *batches* is a list of ``(tool_calls, results)`` pairs, one send-loop
-    iteration each: ``_stream_response`` returns an assistant turn carrying
-    each batch's *tool_calls* in order, then a plain reply ends the loop.
-    Each *results* is what ``_execute_tools`` hands the drain — the
-    truncation/floor/compact path under test runs REAL code between the
-    mocked boundaries.  Mirrors ``tests/test_session.py::_send_with_mocks``;
-    kept local because these tests patch the budget/compaction seam
-    differently per scenario.
+    iteration each: ``_stream_response`` returns a ``ModelTurnResult`` whose
+    ``.tool_calls`` carries each batch's *tool_calls* in order, then a plain
+    reply ends the loop.  Each *results* is what ``_execute_tools`` hands the
+    drain — the truncation/floor/compact path under test runs REAL code
+    between the mocked boundaries.  Mirrors
+    ``tests/test_session.py::_send_with_mocks``; kept local because these
+    tests patch the budget/compaction seam differently per scenario.
 
     ``_estimated_prompt_tokens`` is pinned LOW so the end-of-turn/owed
     compaction paths stay quiet — every compaction observed by these tests
@@ -317,12 +318,12 @@ def _send_with_tool_batches(session, batches, **extra_patches):
     no background utility-completion thread churns against the mock client.
     """
     session._title_generated = True
-    responses = [
-        {"role": "assistant", "content": "", "tool_calls": tool_calls} for tool_calls, _ in batches
-    ] + [{"role": "assistant", "content": "done"}]
+    responses = [make_result(content="", tool_calls=tool_calls) for tool_calls, _ in batches] + [
+        make_result(content="done")
+    ]
     exec_results = [(results, []) for _, results in batches]
 
-    def mock_response(_msgs, _gen):
+    def mock_response(_gen):
         return responses.pop(0)
 
     with contextlib.ExitStack() as stack:
diff --git a/tests/test_watch_integration.py b/tests/test_watch_integration.py
index 482e3332..f842544c 100644
--- a/tests/test_watch_integration.py
+++ b/tests/test_watch_integration.py
@@ -3,8 +3,9 @@
 Drives a real :class:`ChatSession` + a real :class:`WatchRunner` (with
 its daemon thread skipped — we call ``_dispatch_result`` directly to
 avoid the timer dependency) end-to-end through the chat-loop drain
-seam.  The only stub is the LLM provider (patched
-``_create_stream_with_retry``); every other layer is production code:
+seam.  The only stub is the model turn (patched ``_stream_response``,
+returning a canned ``ModelTurnResult``); every other layer is
+production code:
 
 * ``WatchRunner._dispatch_result`` releasing the dispatch lock before
   fan-out
@@ -29,6 +30,7 @@ from unittest.mock import MagicMock, patch
 import pytest
 
 from tests._helpers import patch_session_storage
+from tests._parity_832 import make_result
 from turnstone.core.session import ChatSession
 from turnstone.core.storage import get_storage
 from turnstone.core.trajectory import dicts_from_turns
@@ -117,16 +119,11 @@ def test_watch_fires_then_user_send_drains_envelope(tmp_db, monkeypatch):
     pending = session._nudge_queue.pending(channel="any")
     assert pending == [("watch_triggered", "watch payload body")]
 
-    # 3. Run the chat loop with the LLM patched.  We don't care about
-    # the assistant turn's content; only the wire payload sent to the
-    # provider matters.
+    # 3. Run the chat loop with the model turn patched.  We don't care
+    # about the assistant turn's content; only what the drain seam put
+    # into history alongside it matters.
     with (
-        patch.object(session, "_create_stream_with_retry", return_value=iter([])),
-        patch.object(
-            session,
-            "_stream_response",
-            return_value={"role": "assistant", "content": "ok"},
-        ),
+        patch.object(session, "_stream_response", return_value=make_result(content="ok")),
         patch.object(session, "_update_token_table"),
         patch.object(session, "_print_status_line"),
         patch.object(session, "_visible_memory_count", return_value=0),
@@ -182,12 +179,7 @@ def test_three_back_to_back_watch_fires_drain_into_one_turn(tmp_db, monkeypatch)
     assert len(session._nudge_queue) == 3
 
     with (
-        patch.object(session, "_create_stream_with_retry", return_value=iter([])),
-        patch.object(
-            session,
-            "_stream_response",
-            return_value={"role": "assistant", "content": "got it"},
-        ),
+        patch.object(session, "_stream_response", return_value=make_result(content="got it")),
         patch.object(session, "_update_token_table"),
         patch.object(session, "_print_status_line"),
         patch.object(session, "_visible_memory_count", return_value=0),