diff --git a/tests/_session_helpers.py b/tests/_session_helpers.py
new file mode 100644
index 00000000..d1b1cca2
--- /dev/null
+++ b/tests/_session_helpers.py
@@ -0,0 +1,45 @@
+"""Shared session-test helpers.
+
+Two reasoning-test modules (``test_session_replay_reasoning.py`` and
+``test_session_synth_reasoning_block.py``) need the same minimal
+``ChatSession`` factory + a ``SessionUIBase`` no-op subclass.  Hoisting
+keeps a future third caller from drifting on the defaults — the third
+existing ``_make_session`` (``test_model_registry.py``) deliberately
+takes a different signature (registry / model_alias / reasoning_effort
++ ``_FakeUI``) and is NOT a candidate for sharing this helper.
+
+Module is named with a leading underscore so pytest doesn't try to
+collect it as a test file — it's an importable utility, not a test.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+from unittest.mock import MagicMock
+
+from turnstone.core.session import ChatSession
+from turnstone.core.session_ui_base import SessionUIBase
+
+
+class NullUI(SessionUIBase):
+    """Bare-bones UI satisfying the SessionUIBase contract for tests
+    that don't care about UI side effects."""
+
+    def __init__(self) -> None:
+        super().__init__()
+
+
+def make_session(**kwargs: Any) -> ChatSession:
+    """Build a ChatSession with minimal defaults; tests override
+    individual fields via kwargs."""
+    defaults: dict[str, Any] = {
+        "client": MagicMock(),
+        "model": "test-model",
+        "ui": NullUI(),
+        "instructions": None,
+        "temperature": 0.5,
+        "max_tokens": 4096,
+        "tool_timeout": 30,
+    }
+    defaults.update(kwargs)
+    return ChatSession(**defaults)
diff --git a/tests/test_build_history_reminders.py b/tests/test_build_history_reminders.py
index 832633c0..511eaa30 100644
--- a/tests/test_build_history_reminders.py
+++ b/tests/test_build_history_reminders.py
@@ -147,8 +147,8 @@ class _StubRegistry:
     """Minimal model registry — only ``get_config`` is read by
     ``_build_history``."""
 
-    def __init__(self, persist_reasoning: bool = True) -> None:
-        self._cfg = SimpleNamespace(persist_reasoning=persist_reasoning)
+    def __init__(self, surface_persisted_reasoning: bool = True) -> None:
+        self._cfg = SimpleNamespace(surface_persisted_reasoning=surface_persisted_reasoning)
 
     def get_config(self, alias: str) -> Any:
         return self._cfg
@@ -156,12 +156,12 @@ class _StubRegistry:
 
 def _build_with_registry(
     messages: list[dict[str, Any]],
-    persist_reasoning: bool = True,
+    surface_persisted_reasoning: bool = True,
 ) -> list[dict[str, Any]]:
     session = SimpleNamespace(
         messages=messages,
         _ws_id="ws-test",
-        _registry=_StubRegistry(persist_reasoning=persist_reasoning),
+        _registry=_StubRegistry(surface_persisted_reasoning=surface_persisted_reasoning),
         _model_alias="claude-opus-4-7",
     )
     with patch(
@@ -187,7 +187,7 @@ class TestReasoningSurfacing:
                 {"type": "text", "text": "Final answer."},
             ],
         }
-        history = _build_with_registry([msg], persist_reasoning=True)
+        history = _build_with_registry([msg], surface_persisted_reasoning=True)
         assert len(history) == 1
         assert history[0]["reasoning"] == "let me think"
 
@@ -199,7 +199,7 @@ class TestReasoningSurfacing:
                 {"type": "thinking", "thinking": "hidden", "signature": "s"},
             ],
         }
-        history = _build_with_registry([msg], persist_reasoning=False)
+        history = _build_with_registry([msg], surface_persisted_reasoning=False)
         assert "reasoning" not in history[0]
 
     def test_provider_content_never_in_wire_entry(self) -> None:
@@ -212,12 +212,12 @@ class TestReasoningSurfacing:
                 {"type": "thinking", "thinking": "x", "signature": "s"},
             ],
         }
-        history = _build_with_registry([msg], persist_reasoning=True)
+        history = _build_with_registry([msg], surface_persisted_reasoning=True)
         assert "_provider_content" not in history[0]
 
     def test_no_reasoning_field_when_provider_content_missing(self) -> None:
         msg = {"role": "assistant", "content": "plain answer"}
-        history = _build_with_registry([msg], persist_reasoning=True)
+        history = _build_with_registry([msg], surface_persisted_reasoning=True)
         assert "reasoning" not in history[0]
 
     def test_no_reasoning_field_for_non_assistant_messages(self) -> None:
@@ -232,7 +232,7 @@ class TestReasoningSurfacing:
                 "_provider_content": [{"type": "thinking", "thinking": "leak", "signature": "s"}],
             },
         ]
-        history = _build_with_registry(msgs, persist_reasoning=True)
+        history = _build_with_registry(msgs, surface_persisted_reasoning=True)
         assert "reasoning" not in history[0]
         assert "reasoning" not in history[1]
 
diff --git a/tests/test_history_decoration.py b/tests/test_history_decoration.py
index f65039e6..116ab83b 100644
--- a/tests/test_history_decoration.py
+++ b/tests/test_history_decoration.py
@@ -480,21 +480,21 @@ class TestExtractReasoningForHistory:
         from turnstone.core.history_decoration import extract_reasoning_for_history
 
         messages = [self._anthropic_thinking_msg("let me think")]
-        extract_reasoning_for_history(messages, persist_reasoning_flag=True)
+        extract_reasoning_for_history(messages, surface_persisted_reasoning_flag=True)
         assert messages[0]["reasoning"] == "let me think"
 
     def test_strips_provider_content_after_extraction(self) -> None:
         from turnstone.core.history_decoration import extract_reasoning_for_history
 
         messages = [self._anthropic_thinking_msg("anything")]
-        extract_reasoning_for_history(messages, persist_reasoning_flag=True)
+        extract_reasoning_for_history(messages, surface_persisted_reasoning_flag=True)
         assert "_provider_content" not in messages[0]
 
     def test_strips_provider_content_when_flag_false(self) -> None:
         from turnstone.core.history_decoration import extract_reasoning_for_history
 
         messages = [self._anthropic_thinking_msg("anything")]
-        extract_reasoning_for_history(messages, persist_reasoning_flag=False)
+        extract_reasoning_for_history(messages, surface_persisted_reasoning_flag=False)
         # Strip is unconditional; reasoning is the conditional bit.
         assert "_provider_content" not in messages[0]
         assert "reasoning" not in messages[0]
@@ -515,7 +515,7 @@ class TestExtractReasoningForHistory:
                 ],
             }
         ]
-        extract_reasoning_for_history(messages, persist_reasoning_flag=True)
+        extract_reasoning_for_history(messages, surface_persisted_reasoning_flag=True)
         assert messages[0]["reasoning"] == "first"
 
     def test_first_block_reasoning_dispatches_to_openai_responses(self) -> None:
@@ -535,7 +535,7 @@ class TestExtractReasoningForHistory:
                 ],
             }
         ]
-        extract_reasoning_for_history(messages, persist_reasoning_flag=True)
+        extract_reasoning_for_history(messages, surface_persisted_reasoning_flag=True)
         assert messages[0]["reasoning"] == "s"
         assert "_provider_content" not in messages[0]
 
@@ -549,7 +549,7 @@ class TestExtractReasoningForHistory:
                 "_provider_content": [{"type": "text", "text": "no reasoning here"}],
             }
         ]
-        extract_reasoning_for_history(messages, persist_reasoning_flag=True)
+        extract_reasoning_for_history(messages, surface_persisted_reasoning_flag=True)
         assert "reasoning" not in messages[0]
         assert "_provider_content" not in messages[0]
 
@@ -557,7 +557,7 @@ class TestExtractReasoningForHistory:
         from turnstone.core.history_decoration import extract_reasoning_for_history
 
         messages = [{"role": "assistant", "content": "plain"}]
-        extract_reasoning_for_history(messages, persist_reasoning_flag=True)
+        extract_reasoning_for_history(messages, surface_persisted_reasoning_flag=True)
         assert "reasoning" not in messages[0]
         assert messages[0]["content"] == "plain"
 
@@ -569,7 +569,7 @@ class TestExtractReasoningForHistory:
             {"role": "tool", "tool_call_id": "c1", "content": "out"},
             self._anthropic_thinking_msg("only this one"),
         ]
-        extract_reasoning_for_history(messages, persist_reasoning_flag=True)
+        extract_reasoning_for_history(messages, surface_persisted_reasoning_flag=True)
         assert "reasoning" not in messages[0]
         assert "reasoning" not in messages[1]
         assert messages[2]["reasoning"] == "only this one"
@@ -578,7 +578,7 @@ class TestExtractReasoningForHistory:
         from turnstone.core.history_decoration import extract_reasoning_for_history
 
         messages = [{"role": "assistant", "content": "x", "_provider_content": []}]
-        extract_reasoning_for_history(messages, persist_reasoning_flag=True)
+        extract_reasoning_for_history(messages, surface_persisted_reasoning_flag=True)
         assert "reasoning" not in messages[0]
         # Empty-list provider_content is still stripped from the wire.
         assert "_provider_content" not in messages[0]
@@ -593,7 +593,7 @@ class TestExtractReasoningForHistory:
                 "_provider_content": ["bogus"],
             }
         ]
-        extract_reasoning_for_history(messages, persist_reasoning_flag=True)
+        extract_reasoning_for_history(messages, surface_persisted_reasoning_flag=True)
         assert "reasoning" not in messages[0]
         assert "_provider_content" not in messages[0]
 
@@ -613,6 +613,35 @@ class TestExtractReasoningForHistory:
                 ],
             }
         ]
-        extract_reasoning_for_history(messages, persist_reasoning_flag=True)
+        extract_reasoning_for_history(messages, surface_persisted_reasoning_flag=True)
         assert messages[0]["reasoning"] == "synth thought"
         assert "_provider_content" not in messages[0]
+
+    def test_first_block_redacted_thinking_dispatches_to_anthropic(self) -> None:
+        # Anthropic's extended-thinking API documents that
+        # ``redacted_thinking`` blocks (sealed by the safety system)
+        # can appear before, after, or interleaved with regular
+        # ``thinking`` blocks.  When the redacted block lands first,
+        # the dispatcher must still route to AnthropicProvider so the
+        # surrounding real thinking text surfaces — without this the
+        # reasoning bubble silently disappears on history rehydration.
+        # Pinned by registering "redacted_thinking" as a second key
+        # in _BLOCK_TYPE_PROVIDER_FACTORY pointing at the Anthropic
+        # factory; Anthropic's extractor's type=="thinking" filter
+        # already correctly skips the redacted block.
+        from turnstone.core.history_decoration import extract_reasoning_for_history
+
+        messages = [
+            {
+                "role": "assistant",
+                "content": "answer",
+                "_provider_content": [
+                    {"type": "redacted_thinking", "data": "sealed-blob"},
+                    {"type": "thinking", "thinking": "real thought", "signature": "s"},
+                    {"type": "text", "text": "answer"},
+                ],
+            }
+        ]
+        extract_reasoning_for_history(messages, surface_persisted_reasoning_flag=True)
+        assert messages[0]["reasoning"] == "real thought"
+        assert "_provider_content" not in messages[0]
diff --git a/tests/test_model_definition_storage.py b/tests/test_model_definition_storage.py
index ea3b2dd5..bc44776b 100644
--- a/tests/test_model_definition_storage.py
+++ b/tests/test_model_definition_storage.py
@@ -214,12 +214,12 @@ class TestModelDefinitionStorage:
         assert m["temperature"] is None
 
     def test_reasoning_flags_default(self, db: SQLiteBackend) -> None:
-        """persist_reasoning defaults True; replay_reasoning_to_model defaults False."""
+        """surface_persisted_reasoning defaults True; replay_reasoning_to_model defaults False."""
         did = _make_id()
         db.create_model_definition(definition_id=did, alias="reason-default", model="gpt-5")
         m = db.get_model_definition(did)
         assert m is not None
-        assert m["persist_reasoning"] is True
+        assert m["surface_persisted_reasoning"] is True
         assert m["replay_reasoning_to_model"] is False
 
     def test_create_with_explicit_reasoning_flags(self, db: SQLiteBackend) -> None:
@@ -228,27 +228,27 @@ class TestModelDefinitionStorage:
             definition_id=did,
             alias="reason-explicit",
             model="claude-opus-4-7",
-            persist_reasoning=False,
+            surface_persisted_reasoning=False,
             replay_reasoning_to_model=True,
         )
         m = db.get_model_definition(did)
         assert m is not None
-        assert m["persist_reasoning"] is False
+        assert m["surface_persisted_reasoning"] is False
         assert m["replay_reasoning_to_model"] is True
         # Same values must round-trip via the alias lookup too.
         m_alias = db.get_model_definition_by_alias("reason-explicit")
         assert m_alias is not None
-        assert m_alias["persist_reasoning"] is False
+        assert m_alias["surface_persisted_reasoning"] is False
         assert m_alias["replay_reasoning_to_model"] is True
 
-    def test_update_persist_reasoning(self, db: SQLiteBackend) -> None:
+    def test_update_surface_persisted_reasoning(self, db: SQLiteBackend) -> None:
         did = _make_id()
         db.create_model_definition(definition_id=did, alias="upd-persist", model="gpt-5")
-        ok = db.update_model_definition(did, persist_reasoning=False)
+        ok = db.update_model_definition(did, surface_persisted_reasoning=False)
         assert ok is True
         m = db.get_model_definition(did)
         assert m is not None
-        assert m["persist_reasoning"] is False
+        assert m["surface_persisted_reasoning"] is False
         assert m["replay_reasoning_to_model"] is False  # untouched
 
     def test_update_replay_reasoning_to_model(self, db: SQLiteBackend) -> None:
@@ -258,7 +258,7 @@ class TestModelDefinitionStorage:
         assert ok is True
         m = db.get_model_definition(did)
         assert m is not None
-        assert m["persist_reasoning"] is True  # untouched
+        assert m["surface_persisted_reasoning"] is True  # untouched
         assert m["replay_reasoning_to_model"] is True
 
     def test_list_returns_reasoning_flags(self, db: SQLiteBackend) -> None:
@@ -266,19 +266,19 @@ class TestModelDefinitionStorage:
             definition_id=_make_id(),
             alias="list-a",
             model="gpt-5",
-            persist_reasoning=True,
+            surface_persisted_reasoning=True,
             replay_reasoning_to_model=False,
         )
         db.create_model_definition(
             definition_id=_make_id(),
             alias="list-b",
             model="claude-opus-4-7",
-            persist_reasoning=False,
+            surface_persisted_reasoning=False,
             replay_reasoning_to_model=True,
         )
         models = db.list_model_definitions()
         by_alias = {m["alias"]: m for m in models}
-        assert by_alias["list-a"]["persist_reasoning"] is True
+        assert by_alias["list-a"]["surface_persisted_reasoning"] is True
         assert by_alias["list-a"]["replay_reasoning_to_model"] is False
-        assert by_alias["list-b"]["persist_reasoning"] is False
+        assert by_alias["list-b"]["surface_persisted_reasoning"] is False
         assert by_alias["list-b"]["replay_reasoning_to_model"] is True
diff --git a/tests/test_model_registry.py b/tests/test_model_registry.py
index 85d53e10..fba575ad 100644
--- a/tests/test_model_registry.py
+++ b/tests/test_model_registry.py
@@ -78,7 +78,7 @@ class TestModelConfig:
 
     def test_reasoning_flags_default(self) -> None:
         cfg = ModelConfig(alias="x", base_url="x", api_key="x", model="x")
-        assert cfg.persist_reasoning is True
+        assert cfg.surface_persisted_reasoning is True
         assert cfg.replay_reasoning_to_model is False
 
     def test_reasoning_flags_set(self) -> None:
@@ -87,10 +87,10 @@ class TestModelConfig:
             base_url="x",
             api_key="x",
             model="x",
-            persist_reasoning=False,
+            surface_persisted_reasoning=False,
             replay_reasoning_to_model=True,
         )
-        assert cfg.persist_reasoning is False
+        assert cfg.surface_persisted_reasoning is False
         assert cfg.replay_reasoning_to_model is True
 
 
@@ -716,7 +716,7 @@ class TestLoadModelRegistryWithDB:
                     "context_window": 200000,
                     "capabilities": "{}",
                     "enabled": True,
-                    "persist_reasoning": False,
+                    "surface_persisted_reasoning": False,
                     "replay_reasoning_to_model": True,
                 }
             ]
@@ -724,7 +724,7 @@ class TestLoadModelRegistryWithDB:
         with patch("turnstone.core.model_registry.load_config", return_value={}):
             reg = load_model_registry("http://x/v1", "x", "x", storage=storage)
         cfg = reg.get_config("anth-thinking")
-        assert cfg.persist_reasoning is False
+        assert cfg.surface_persisted_reasoning is False
         assert cfg.replay_reasoning_to_model is True
 
     def test_db_reasoning_flags_default_when_absent(self) -> None:
@@ -740,14 +740,14 @@ class TestLoadModelRegistryWithDB:
                     "context_window": 32768,
                     "capabilities": "{}",
                     "enabled": True,
-                    # persist_reasoning + replay_reasoning_to_model intentionally absent
+                    # surface_persisted_reasoning + replay_reasoning_to_model intentionally absent
                 }
             ]
         )
         with patch("turnstone.core.model_registry.load_config", return_value={}):
             reg = load_model_registry("http://x/v1", "x", "x", storage=storage)
         cfg = reg.get_config("legacy-row")
-        assert cfg.persist_reasoning is True
+        assert cfg.surface_persisted_reasoning is True
         assert cfg.replay_reasoning_to_model is False
 
     def test_db_default_alias_not_clobbered(self) -> None:
diff --git a/tests/test_provider_anthropic_reasoning.py b/tests/test_provider_anthropic_reasoning.py
index 21bec44a..389e3f41 100644
--- a/tests/test_provider_anthropic_reasoning.py
+++ b/tests/test_provider_anthropic_reasoning.py
@@ -21,7 +21,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 (
-    MAX_REASONING_DISPLAY_BYTES as _MAX_REASONING_DISPLAY_BYTES,
+    MAX_REASONING_DISPLAY_CHARS as _MAX_REASONING_DISPLAY_CHARS,
 )
 
 
@@ -77,13 +77,13 @@ class TestExtractReasoningText:
         assert anthropic.extract_reasoning_text(blocks) == ""
 
     def test_truncation_at_64kib_cap(self, anthropic: AnthropicProvider) -> None:
-        long_text = "x" * (_MAX_REASONING_DISPLAY_BYTES + 1024)
+        long_text = "x" * (_MAX_REASONING_DISPLAY_CHARS + 1024)
         blocks = [{"type": "thinking", "thinking": long_text, "signature": "s"}]
         result = anthropic.extract_reasoning_text(blocks)
-        assert len(result) == _MAX_REASONING_DISPLAY_BYTES
+        assert len(result) == _MAX_REASONING_DISPLAY_CHARS
 
     def test_just_under_cap_not_truncated(self, anthropic: AnthropicProvider) -> None:
-        text = "y" * (_MAX_REASONING_DISPLAY_BYTES - 1)
+        text = "y" * (_MAX_REASONING_DISPLAY_CHARS - 1)
         blocks = [{"type": "thinking", "thinking": text, "signature": "s"}]
         assert anthropic.extract_reasoning_text(blocks) == text
 
diff --git a/tests/test_provider_openai_responses_reasoning.py b/tests/test_provider_openai_responses_reasoning.py
index 1a712a9a..84f841be 100644
--- a/tests/test_provider_openai_responses_reasoning.py
+++ b/tests/test_provider_openai_responses_reasoning.py
@@ -22,7 +22,7 @@ from turnstone.core.providers._openai_responses import (
     _reasoning_item_for_input,
 )
 from turnstone.core.providers._protocol import (
-    MAX_REASONING_DISPLAY_BYTES as _MAX_REASONING_DISPLAY_BYTES,
+    MAX_REASONING_DISPLAY_CHARS as _MAX_REASONING_DISPLAY_CHARS,
 )
 from turnstone.core.providers._protocol import ModelCapabilities
 
@@ -98,7 +98,7 @@ class TestExtractReasoningText:
         assert "raw reasoning" in result
 
     def test_truncation_at_64kib_cap(self, provider: OpenAIResponsesProvider) -> None:
-        long_text = "x" * (_MAX_REASONING_DISPLAY_BYTES + 1024)
+        long_text = "x" * (_MAX_REASONING_DISPLAY_CHARS + 1024)
         blocks = [
             {
                 "type": "reasoning",
@@ -107,7 +107,7 @@ class TestExtractReasoningText:
             }
         ]
         result = provider.extract_reasoning_text(blocks)
-        assert len(result) == _MAX_REASONING_DISPLAY_BYTES
+        assert len(result) == _MAX_REASONING_DISPLAY_CHARS
 
     def test_malformed_summary_entry_skipped(self, provider: OpenAIResponsesProvider) -> None:
         blocks = [
diff --git a/tests/test_reasoning_audit_log_discipline.py b/tests/test_reasoning_audit_log_discipline.py
index 630682ee..f1e90017 100644
--- a/tests/test_reasoning_audit_log_discipline.py
+++ b/tests/test_reasoning_audit_log_discipline.py
@@ -28,11 +28,15 @@ from types import SimpleNamespace
 from typing import Any
 from unittest.mock import patch
 
+from tests._session_helpers import make_session
 from turnstone.core.history_decoration import (
     extract_reasoning_for_history,
     extract_reasoning_text_from_provider_content,
 )
 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.server import _build_history
 
 _MARKER = "SECRET_REASONING_MARKER_xyz123_unlikely_collision"
@@ -151,7 +155,7 @@ class TestReasoningAuditLogDiscipline:
             p.start()
         try:
             messages = [self._thinking_msg(_MARKER)]
-            extract_reasoning_for_history(messages, persist_reasoning_flag=True)
+            extract_reasoning_for_history(messages, surface_persisted_reasoning_flag=True)
             assert messages[0]["reasoning"] == _MARKER  # UI-bound is allowed
         finally:
             for p in patchers:
@@ -166,7 +170,9 @@ class TestReasoningAuditLogDiscipline:
         )
 
     def test_build_history_does_not_log_reasoning(self) -> None:
-        registry = SimpleNamespace(get_config=lambda alias: SimpleNamespace(persist_reasoning=True))
+        registry = SimpleNamespace(
+            get_config=lambda alias: SimpleNamespace(surface_persisted_reasoning=True)
+        )
         session = SimpleNamespace(
             messages=[self._thinking_msg(_MARKER)],
             _ws_id="ws-audit",
@@ -192,3 +198,136 @@ class TestReasoningAuditLogDiscipline:
             if _payload_contains_marker(args, kwargs)
         ]
         assert offending == [], f"_build_history leaked reasoning text into INFO+ logs: {offending}"
+
+    # ------------------------------------------------------------------
+    # Phase 2 + Phase 3 surfaces — added in response to a code-review
+    # finding that the original 4-test coverage missed every code path
+    # introduced after Phase 1.  Each new test mirrors the structure
+    # above: capture every Logger.info / warning / error call across
+    # the operation, assert the marker doesn't appear in any captured
+    # payload (UI-bound returns IS allowed; logging at INFO+ is NOT).
+    # ------------------------------------------------------------------
+
+    def test_openai_responses_extractor_does_not_log_reasoning(self) -> None:
+        captured, patchers = _capture_log_calls()
+        for p in patchers:
+            p.start()
+        try:
+            provider = OpenAIResponsesProvider()
+            blocks = [
+                {
+                    "type": "reasoning",
+                    "id": "r_1",
+                    "summary": [{"type": "summary_text", "text": _MARKER}],
+                }
+            ]
+            text = provider.extract_reasoning_text(blocks)
+            assert _MARKER in text  # UI-bound return is allowed
+        finally:
+            for p in patchers:
+                p.stop()
+        offending = [
+            (lvl, args, kwargs)
+            for lvl, args, kwargs in captured
+            if _payload_contains_marker(args, kwargs)
+        ]
+        assert offending == [], (
+            f"OpenAIResponsesProvider.extract_reasoning_text leaked reasoning "
+            f"text into INFO+ logs: {offending}"
+        )
+
+    def test_openai_chat_extractor_does_not_log_reasoning(self) -> None:
+        captured, patchers = _capture_log_calls()
+        for p in patchers:
+            p.start()
+        try:
+            provider = OpenAIChatCompletionsProvider()
+            blocks = [{"type": "reasoning_text", "text": _MARKER, "source": "vllm"}]
+            text = provider.extract_reasoning_text(blocks)
+            assert text == _MARKER
+        finally:
+            for p in patchers:
+                p.stop()
+        offending = [
+            (lvl, args, kwargs)
+            for lvl, args, kwargs in captured
+            if _payload_contains_marker(args, kwargs)
+        ]
+        assert offending == [], (
+            f"OpenAIChatCompletionsProvider.extract_reasoning_text leaked "
+            f"reasoning text into INFO+ logs: {offending}"
+        )
+
+    def test_synth_reasoning_block_via_stream_response_does_not_log_reasoning(
+        self,
+    ) -> None:
+        """Drives ChatSession._stream_response (which calls
+        _maybe_synth_reasoning_block at end-of-stream) 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),
+            ),
+        ]
+        captured, patchers = _capture_log_calls()
+        for p in patchers:
+            p.start()
+        try:
+            msg = session._stream_response(iter(chunks))
+            # Synth block stamped onto _provider_content with the marker.
+            assert msg["_provider_content"][0]["text"] == _MARKER
+        finally:
+            for p in patchers:
+                p.stop()
+        offending = [
+            (lvl, args, kwargs)
+            for lvl, args, kwargs in captured
+            if _payload_contains_marker(args, kwargs)
+        ]
+        assert offending == [], (
+            f"_stream_response + _maybe_synth_reasoning_block leaked reasoning "
+            f"text into INFO+ logs: {offending}"
+        )
+
+    def test_anthropic_convert_messages_strip_does_not_log_reasoning(self) -> None:
+        """Drives the Phase 2 strip predicate
+        (``replay_reasoning_to_model=False``) which walks thinking
+        blocks to filter them out before the wire payload is built;
+        asserts no log call carried the marker text."""
+        captured, patchers = _capture_log_calls()
+        for p in patchers:
+            p.start()
+        try:
+            provider = AnthropicProvider()
+            messages = [
+                {
+                    "role": "assistant",
+                    "content": "Final answer.",
+                    "_provider_content": [
+                        {"type": "thinking", "thinking": _MARKER, "signature": "s"},
+                        {"type": "text", "text": "Final answer."},
+                    ],
+                },
+            ]
+            _, converted = provider._convert_messages(messages, replay_reasoning_to_model=False)
+            # Strip fired — thinking block dropped from wire.
+            assistant = next(m for m in converted if m["role"] == "assistant")
+            block_types = [b.get("type") for b in assistant["content"]]
+            assert "thinking" not in block_types
+        finally:
+            for p in patchers:
+                p.stop()
+        offending = [
+            (lvl, args, kwargs)
+            for lvl, args, kwargs in captured
+            if _payload_contains_marker(args, kwargs)
+        ]
+        assert offending == [], (
+            f"AnthropicProvider._convert_messages strip predicate leaked "
+            f"reasoning text into INFO+ logs: {offending}"
+        )
diff --git a/tests/test_session_replay_reasoning.py b/tests/test_session_replay_reasoning.py
index e018ed20..3844c5fd 100644
--- a/tests/test_session_replay_reasoning.py
+++ b/tests/test_session_replay_reasoning.py
@@ -26,29 +26,7 @@ from types import SimpleNamespace
 from typing import Any
 from unittest.mock import MagicMock, patch
 
-from turnstone.core.session import ChatSession
-from turnstone.core.session_ui_base import SessionUIBase
-
-
-class _NullUI(SessionUIBase):
-    """Bare-bones UI satisfying the SessionUIBase contract for these tests."""
-
-    def __init__(self) -> None:
-        super().__init__()
-
-
-def _make_session(**kwargs: Any) -> ChatSession:
-    defaults: dict[str, Any] = {
-        "client": MagicMock(),
-        "model": "test-model",
-        "ui": _NullUI(),
-        "instructions": None,
-        "temperature": 0.5,
-        "max_tokens": 4096,
-        "tool_timeout": 30,
-    }
-    defaults.update(kwargs)
-    return ChatSession(**defaults)
+from tests._session_helpers import make_session as _make_session
 
 
 def _registry_with_flag(persist: bool = True, replay: bool = False) -> Any:
@@ -56,7 +34,7 @@ def _registry_with_flag(persist: bool = True, replay: bool = False) -> Any:
     flags under test."""
     return SimpleNamespace(
         get_config=lambda alias: SimpleNamespace(
-            persist_reasoning=persist,
+            surface_persisted_reasoning=persist,
             replay_reasoning_to_model=replay,
         )
     )
diff --git a/tests/test_session_synth_reasoning_block.py b/tests/test_session_synth_reasoning_block.py
index d8d539ff..7345dbac 100644
--- a/tests/test_session_synth_reasoning_block.py
+++ b/tests/test_session_synth_reasoning_block.py
@@ -26,34 +26,13 @@ from __future__ import annotations
 
 from types import SimpleNamespace
 from typing import Any
-from unittest.mock import MagicMock
 
+from tests._session_helpers import make_session as _make_session
 from turnstone.core.providers._anthropic import (
     ANTHROPIC_VALID_BLOCK_TYPES,
     AnthropicProvider,
 )
 from turnstone.core.providers._openai_chat import OpenAIChatCompletionsProvider
-from turnstone.core.session import ChatSession
-from turnstone.core.session_ui_base import SessionUIBase
-
-
-class _NullUI(SessionUIBase):
-    def __init__(self) -> None:
-        super().__init__()
-
-
-def _make_session(**kwargs: Any) -> ChatSession:
-    defaults: dict[str, Any] = {
-        "client": MagicMock(),
-        "model": "test-model",
-        "ui": _NullUI(),
-        "instructions": None,
-        "temperature": 0.5,
-        "max_tokens": 4096,
-        "tool_timeout": 30,
-    }
-    defaults.update(kwargs)
-    return ChatSession(**defaults)
 
 
 class TestMaybeSynthReasoningBlock:
diff --git a/tests/test_workstream_endpoints.py b/tests/test_workstream_endpoints.py
index d033c374..881fe1d1 100644
--- a/tests/test_workstream_endpoints.py
+++ b/tests/test_workstream_endpoints.py
@@ -1788,7 +1788,7 @@ class TestHistoryReasoningRehydration:
             ws_id, "assistant", "Final answer.", provider_data=provider_data
         )
         # No live session — exercises the storage-only path which
-        # falls back to default persist_reasoning=True.
+        # falls back to default surface_persisted_reasoning=True.
         mock_mgr = MagicMock()
         mock_mgr.get.return_value = None
         client = _build_history_app(mock_mgr, _inject_storage)
@@ -1814,7 +1814,7 @@ class TestHistoryReasoningRehydration:
             assert "_provider_content" not in m
 
     def test_history_handler_with_persist_flag_false_via_live_session(self, _inject_storage):
-        """Operator-flipped ``persist_reasoning=False`` on the active
+        """Operator-flipped ``surface_persisted_reasoning=False`` on the active
         model suppresses the reasoning field even when the data is
         stored. ``_provider_content`` is still stripped from the wire.
         """
@@ -1825,7 +1825,7 @@ class TestHistoryReasoningRehydration:
         live_session = SimpleNamespace(
             id=ws_id,
             _registry=SimpleNamespace(
-                get_config=lambda alias: SimpleNamespace(persist_reasoning=False)
+                get_config=lambda alias: SimpleNamespace(surface_persisted_reasoning=False)
             ),
             _model_alias="claude-opus-4-7",
         )
@@ -1844,7 +1844,7 @@ class TestHistoryReasoningRehydration:
         """Cold workstream (no live session) — the handler walks
         ``workstream_config.model_alias`` (persisted at first send by
         the SessionManager rehydrate path) and looks up the active
-        model's ``persist_reasoning`` flag through the global registry
+        model's ``surface_persisted_reasoning`` flag through the global registry
         on ``app.state``. Operator flag-flip is honored uniformly
         across live and cold workstreams.
         """
@@ -1884,7 +1884,7 @@ class TestHistoryReasoningRehydration:
         app.state.auth_storage = _inject_storage
         app.state.registry = SimpleNamespace(
             get_config=lambda alias: SimpleNamespace(
-                persist_reasoning=(alias != "claude-opus-4-7"),
+                surface_persisted_reasoning=(alias != "claude-opus-4-7"),
             )
         )
         client = TestClient(app)
diff --git a/turnstone/api/console_schemas.py b/turnstone/api/console_schemas.py
index 96266d33..9eb583f3 100644
--- a/turnstone/api/console_schemas.py
+++ b/turnstone/api/console_schemas.py
@@ -908,7 +908,7 @@ class ModelDefinitionInfo(BaseModel):
     temperature: float | None = None
     max_tokens: int | None = None
     reasoning_effort: str | None = None
-    persist_reasoning: bool = True
+    surface_persisted_reasoning: bool = True
     replay_reasoning_to_model: bool = False
     source: str = ""
     created_by: str = ""
@@ -928,7 +928,7 @@ class CreateModelDefinitionRequest(BaseModel):
     temperature: float | None = None
     max_tokens: int | None = None
     reasoning_effort: str | None = None
-    persist_reasoning: bool = True
+    surface_persisted_reasoning: bool = True
     replay_reasoning_to_model: bool = False
 
 
@@ -944,7 +944,7 @@ class UpdateModelDefinitionRequest(BaseModel):
     temperature: float | None = None
     max_tokens: int | None = None
     reasoning_effort: str | None = None
-    persist_reasoning: bool | None = None
+    surface_persisted_reasoning: bool | None = None
     replay_reasoning_to_model: bool | None = None
 
 
diff --git a/turnstone/console/server.py b/turnstone/console/server.py
index f2c61d6d..c50ffeed 100644
--- a/turnstone/console/server.py
+++ b/turnstone/console/server.py
@@ -9904,7 +9904,7 @@ async def admin_create_model_definition(request: Request) -> JSONResponse:
         if not reasoning_effort:
             reasoning_effort = None
 
-    persist_reasoning = bool(body.get("persist_reasoning", True))
+    surface_persisted_reasoning = bool(body.get("surface_persisted_reasoning", True))
     replay_reasoning_to_model = bool(body.get("replay_reasoning_to_model", False))
 
     storage.create_model_definition(
@@ -9921,7 +9921,7 @@ async def admin_create_model_definition(request: Request) -> JSONResponse:
         temperature=temperature,
         max_tokens=max_tokens,
         reasoning_effort=reasoning_effort,
-        persist_reasoning=persist_reasoning,
+        surface_persisted_reasoning=surface_persisted_reasoning,
         replay_reasoning_to_model=replay_reasoning_to_model,
     )
 
@@ -10084,8 +10084,8 @@ async def admin_update_model_definition(request: Request) -> JSONResponse:
                 )
             else:
                 updates["reasoning_effort"] = re_val
-    if "persist_reasoning" in body:
-        updates["persist_reasoning"] = bool(body["persist_reasoning"])
+    if "surface_persisted_reasoning" in body:
+        updates["surface_persisted_reasoning"] = bool(body["surface_persisted_reasoning"])
     if "replay_reasoning_to_model" in body:
         updates["replay_reasoning_to_model"] = bool(body["replay_reasoning_to_model"])
 
diff --git a/turnstone/console/static/admin.js b/turnstone/console/static/admin.js
index b88e69f1..708c422a 100644
--- a/turnstone/console/static/admin.js
+++ b/turnstone/console/static/admin.js
@@ -5011,7 +5011,7 @@ function _renderModels(items) {
     // Reasoning persistence flags surface only when non-default
     // (persist=False is the operator opt-out; replay=True is the
     // operator opt-in). Default values are silent.
-    if (m.persist_reasoning === false) overrides.push("persist=off");
+    if (m.surface_persisted_reasoning === false) overrides.push("surface=off");
     if (m.replay_reasoning_to_model === true) overrides.push("replay=on");
     if (overrides.length) {
       var ovrSpan = document.createElement("span");
@@ -5201,7 +5201,7 @@ function showCreateModelModal() {
     el.style.borderColor = "";
   });
   document.getElementById("model-enabled").checked = true;
-  document.getElementById("model-persist-reasoning").checked = true;
+  document.getElementById("model-surface-persisted-reasoning").checked = true;
   document.getElementById("model-replay-reasoning").checked = false;
   document.getElementById("model-detect-result").style.display = "none";
   document.getElementById("model-detect-btn").disabled = false;
@@ -5288,8 +5288,8 @@ function showEditModelModal(definitionId) {
       // Reasoning persistence flags — defaults match the dataclass
       // defaults (persist=true, replay=false) when the API returns
       // them as undefined (legacy / pre-052 row).
-      document.getElementById("model-persist-reasoning").checked =
-        m.persist_reasoning !== false;
+      document.getElementById("model-surface-persisted-reasoning").checked =
+        m.surface_persisted_reasoning !== false;
       document.getElementById("model-replay-reasoning").checked =
         m.replay_reasoning_to_model === true;
       _applyProviderDefaults();
@@ -5436,8 +5436,8 @@ function submitCreateModel() {
   // Reasoning persistence flags — always serialize so a flip from
   // default takes effect on PUT (the server's update path keys off
   // "field present in body").
-  form.persist_reasoning = document.getElementById(
-    "model-persist-reasoning",
+  form.surface_persisted_reasoning = document.getElementById(
+    "model-surface-persisted-reasoning",
   ).checked;
   form.replay_reasoning_to_model = document.getElementById(
     "model-replay-reasoning",
diff --git a/turnstone/console/static/coordinator/coordinator.js b/turnstone/console/static/coordinator/coordinator.js
index ca2d7941..641ce509 100644
--- a/turnstone/console/static/coordinator/coordinator.js
+++ b/turnstone/console/static/coordinator/coordinator.js
@@ -4202,7 +4202,7 @@
           // snapshot ":2021" call sites — same appendMsg("reasoning")
           // helper, just driven from history-render rather than the
           // SSE handler. Only present when the active model's
-          // persist_reasoning flag is true and the message round-tripped
+          // surface_persisted_reasoning flag is true and the message round-tripped
           // a thinking lane.
           if (typeof m.reasoning === "string" && m.reasoning.length) {
             const rEl = appendMsg("reasoning", "", { label: "reasoning" });
diff --git a/turnstone/console/static/index.html b/turnstone/console/static/index.html
index ea5791c2..6f6a256f 100644
--- a/turnstone/console/static/index.html
+++ b/turnstone/console/static/index.html
@@ -4127,13 +4127,13 @@
           >
           <label
             style="margin: 0; font-size: 12px; color: var(--fg-dim)"
-            title="Surface stored reasoning text on /history responses (UI bubble on page reload). Storage of reasoning is independent of this flag."
+            title="Surface stored reasoning text on /history responses (UI bubble on page reload). Storage of reasoning bytes is unaffected by this flag — they ride in provider_data regardless."
             ><input
               type="checkbox"
-              id="model-persist-reasoning"
+              id="model-surface-persisted-reasoning"
               checked
               style="margin-right: 5px"
-            />Persist reasoning</label
+            />Surface persisted reasoning</label
           >
           <label
             style="margin: 0; font-size: 12px; color: var(--fg-dim)"
diff --git a/turnstone/core/history_decoration.py b/turnstone/core/history_decoration.py
index 84ad6cc0..d2fbd531 100644
--- a/turnstone/core/history_decoration.py
+++ b/turnstone/core/history_decoration.py
@@ -277,31 +277,44 @@ if TYPE_CHECKING:
 def _make_provider_factory(module_path: str, class_name: str) -> Callable[[], LLMProvider]:
     """Build a thread-unsafe lazy-init factory for a provider singleton.
 
-    Single source of truth for the dispatcher's per-provider lazy-load
-    pattern — each block-type entry in ``_BLOCK_TYPE_PROVIDER_FACTORY``
-    closes over this with its own (module_path, class_name) pair.
-    Adding a fourth provider is a single tuple in the dict, not a
-    new 9-line getter.
+    Each block-type entry in ``_BLOCK_TYPE_PROVIDER_FACTORY`` closes
+    over its own (module_path, class_name) pair.  Adding a fourth
+    provider is a single tuple in the dict, not a new 9-line getter.
+
+    Uses ``nonlocal`` instead of ``functools.lru_cache`` so the cache
+    state stays inside this closure (lru_cache would attach state to
+    the inner function object, which is correct but adds a per-call
+    hash lookup on a bound key for what's effectively a single-slot
+    cache).
     """
-    cached: dict[str, LLMProvider] = {}
+    instance: LLMProvider | None = None
 
     def factory() -> LLMProvider:
-        if "instance" not in cached:
+        nonlocal instance
+        if instance is None:
             import importlib
 
             module = importlib.import_module(module_path)
-            cached["instance"] = getattr(module, class_name)()
-        return cached["instance"]
+            instance = getattr(module, class_name)()
+        return instance
 
     return factory
 
 
 # Block-type → provider factory.  Routing is structural — block shape
-# is non-overlapping across providers by API design.  Three block
-# types are recognised today:
+# is non-overlapping across providers by API design.  Recognised
+# block types today:
 #
 # * ``"thinking"`` — Anthropic native (Phase 1).  Walks the
 #   ``thinking`` field on each block.
+# * ``"redacted_thinking"`` — Anthropic native (Phase 1).  Anthropic's
+#   safety system rewrites a thinking block into a sealed
+#   ``redacted_thinking`` block; the Anthropic docs note these can
+#   appear before, after, or interleaved with regular ``thinking``
+#   blocks.  Same factory: AnthropicProvider's extractor walks the
+#   full block list and filters to ``type == "thinking"``, so the
+#   redacted blocks are correctly skipped while the surrounding
+#   real thinking text still surfaces.
 # * ``"reasoning"`` — OpenAI Responses native (Phase 3).  Walks
 #   ``summary[*].text`` (always present) and ``content[*].text``
 #   (present when ``include=["reasoning.encrypted_content"]`` is
@@ -310,8 +323,12 @@ def _make_provider_factory(module_path: str, class_name: str) -> Callable[[], LL
 #   ``ChatSession._maybe_synth_reasoning_block`` for Chat Completions
 #   paths (vLLM, llama.cpp, Gemini-compat) where reasoning surfaces
 #   only as ``reasoning_delta`` chunks with no native block shape.
+_anthropic_factory = _make_provider_factory(
+    "turnstone.core.providers._anthropic", "AnthropicProvider"
+)
 _BLOCK_TYPE_PROVIDER_FACTORY: dict[str, Callable[[], LLMProvider]] = {
-    "thinking": _make_provider_factory("turnstone.core.providers._anthropic", "AnthropicProvider"),
+    "thinking": _anthropic_factory,
+    "redacted_thinking": _anthropic_factory,
     "reasoning": _make_provider_factory(
         "turnstone.core.providers._openai_responses", "OpenAIResponsesProvider"
     ),
@@ -347,7 +364,7 @@ def extract_reasoning_text_from_provider_content(provider_content: Any) -> str:
 
 def extract_reasoning_for_history(
     messages: list[dict[str, Any]],
-    persist_reasoning_flag: bool,
+    surface_persisted_reasoning_flag: bool,
 ) -> None:
     """Surface stored reasoning text on each assistant message; strip the
     raw provider content from the wire payload.
@@ -357,7 +374,7 @@ def extract_reasoning_for_history(
     — both extraction source and stamp destination are the same dict.
     Walks *messages* in place: for every assistant message, dispatches
     via :func:`extract_reasoning_text_from_provider_content` and stamps
-    ``msg["reasoning"]`` when *persist_reasoning_flag* is True and the
+    ``msg["reasoning"]`` when *surface_persisted_reasoning_flag* is True and the
     dispatcher returned non-empty text.  Strips ``_provider_content``
     unconditionally — the field is internal and never read by either UI.
 
@@ -375,12 +392,12 @@ def extract_reasoning_for_history(
             continue
         provider_content = msg.get("_provider_content")
         # Always strip the internal lane before the wire payload leaves
-        # the helper, even when persist_reasoning_flag is False or the
+        # the helper, even when surface_persisted_reasoning_flag is False or the
         # field is empty/missing.  The strip is the contract; reasoning
         # surfacing is conditional on top of it.
         if "_provider_content" in msg:
             del msg["_provider_content"]
-        if not persist_reasoning_flag:
+        if not surface_persisted_reasoning_flag:
             continue
         text = extract_reasoning_text_from_provider_content(provider_content)
         if text:
diff --git a/turnstone/core/model_registry.py b/turnstone/core/model_registry.py
index 64402117..bf6f9308 100644
--- a/turnstone/core/model_registry.py
+++ b/turnstone/core/model_registry.py
@@ -40,10 +40,10 @@ class ModelConfig:
     max_tokens: int | None = None
     reasoning_effort: str | None = None
     # Per-model reasoning-persistence flags (db-backed, admin-toggleable).
-    # persist_reasoning controls UI rehydration of stored reasoning text in
+    # surface_persisted_reasoning controls UI rehydration of stored reasoning text in
     # /history responses; replay_reasoning_to_model controls whether
     # reasoning blocks ride the wire on subsequent provider calls.
-    persist_reasoning: bool = True
+    surface_persisted_reasoning: bool = True
     replay_reasoning_to_model: bool = False
     # Server compatibility settings for openai-compatible backends.
     # Populated from capabilities["server_compat"] during load.
@@ -413,7 +413,7 @@ def load_model_registry(
                 row_reasoning_effort = row.get("reasoning_effort")
                 # Per-model reasoning flags. Defaults match the dataclass so a
                 # pre-052 row missing these columns degrades gracefully.
-                row_persist_reasoning = bool(row.get("persist_reasoning", True))
+                row_surface_persisted_reasoning = bool(row.get("surface_persisted_reasoning", True))
                 row_replay_reasoning = bool(row.get("replay_reasoning_to_model", False))
                 configs[alias] = ModelConfig(
                     alias=alias,
@@ -429,7 +429,7 @@ def load_model_registry(
                     reasoning_effort=row_reasoning_effort
                     if row_reasoning_effort is not None
                     else None,
-                    persist_reasoning=row_persist_reasoning,
+                    surface_persisted_reasoning=row_surface_persisted_reasoning,
                     replay_reasoning_to_model=row_replay_reasoning,
                     server_compat=row_server_compat,
                 )
diff --git a/turnstone/core/providers/_anthropic.py b/turnstone/core/providers/_anthropic.py
index dcb62a5e..21cd8d86 100644
--- a/turnstone/core/providers/_anthropic.py
+++ b/turnstone/core/providers/_anthropic.py
@@ -12,12 +12,12 @@ import sys
 from typing import TYPE_CHECKING, Any
 
 from turnstone.core.providers._protocol import (
-    MAX_REASONING_DISPLAY_BYTES,
     CompletionResult,
     ModelCapabilities,
     StreamChunk,
     ToolCallDelta,
     UsageInfo,
+    _join_reasoning_with_cap,
     _lookup_capabilities,
 )
 
@@ -1045,12 +1045,7 @@ class AnthropicProvider:
             text = block.get("thinking")
             if isinstance(text, str) and text:
                 parts.append(text)
-        if not parts:
-            return ""
-        joined = "\n".join(parts)
-        if len(joined) > MAX_REASONING_DISPLAY_BYTES:
-            return joined[:MAX_REASONING_DISPLAY_BYTES]
-        return joined
+        return _join_reasoning_with_cap(parts)
 
 
 def _normalize_finish_reason(reason: str) -> str:
diff --git a/turnstone/core/providers/_openai_chat.py b/turnstone/core/providers/_openai_chat.py
index 9634ad3c..158d1301 100644
--- a/turnstone/core/providers/_openai_chat.py
+++ b/turnstone/core/providers/_openai_chat.py
@@ -24,11 +24,11 @@ from turnstone.core.providers._openai_common import (
     sanitize_messages,
 )
 from turnstone.core.providers._protocol import (
-    MAX_REASONING_DISPLAY_BYTES,
     CompletionResult,
     ModelCapabilities,
     StreamChunk,
     ToolCallDelta,
+    _join_reasoning_with_cap,
 )
 
 log = structlog.get_logger(__name__)
@@ -413,9 +413,4 @@ class OpenAIChatCompletionsProvider:
             text = block.get("text")
             if isinstance(text, str) and text:
                 parts.append(text)
-        if not parts:
-            return ""
-        joined = "\n".join(parts)
-        if len(joined) > MAX_REASONING_DISPLAY_BYTES:
-            return joined[:MAX_REASONING_DISPLAY_BYTES]
-        return joined
+        return _join_reasoning_with_cap(parts)
diff --git a/turnstone/core/providers/_openai_responses.py b/turnstone/core/providers/_openai_responses.py
index 778f705b..ba621ca8 100644
--- a/turnstone/core/providers/_openai_responses.py
+++ b/turnstone/core/providers/_openai_responses.py
@@ -28,11 +28,11 @@ from turnstone.core.providers._openai_common import (
     sanitize_messages,
 )
 from turnstone.core.providers._protocol import (
-    MAX_REASONING_DISPLAY_BYTES,
     CompletionResult,
     ModelCapabilities,
     StreamChunk,
     ToolCallDelta,
+    _join_reasoning_with_cap,
 )
 
 log = structlog.get_logger(__name__)
@@ -705,12 +705,7 @@ class OpenAIResponsesProvider:
                     text = c.get("text")
                     if isinstance(text, str) and text:
                         parts.append(text)
-        if not parts:
-            return ""
-        joined = "\n".join(parts)
-        if len(joined) > MAX_REASONING_DISPLAY_BYTES:
-            return joined[:MAX_REASONING_DISPLAY_BYTES]
-        return joined
+        return _join_reasoning_with_cap(parts)
 
 
 def _reasoning_item_for_input(stored: dict[str, Any]) -> dict[str, Any] | None:
diff --git a/turnstone/core/providers/_protocol.py b/turnstone/core/providers/_protocol.py
index f11c0588..b3ca4459 100644
--- a/turnstone/core/providers/_protocol.py
+++ b/turnstone/core/providers/_protocol.py
@@ -106,8 +106,34 @@ class ModelCapabilities:
 # tuning change propagates to every provider's display path uniformly.
 # Larger reasoning bodies are still stored verbatim in
 # ``provider_data``; only the rehydrated UI display payload is
-# truncated.  64 KiB matches the briefing's recommendation.
-MAX_REASONING_DISPLAY_BYTES = 64 * 1024
+# truncated.
+#
+# Named ``_CHARS`` (not ``_BYTES``) because the cap is enforced via
+# Python ``str`` slicing, which counts code points.  Reasoning text
+# that happens to contain 4-byte UTF-8 glyphs (CJK, emoji) will
+# serialise to a larger UTF-8 payload than the constant suggests —
+# fine for the UI display path (browsers handle the encoded length),
+# but worth knowing if this is ever wired to a byte-quota system.
+MAX_REASONING_DISPLAY_CHARS = 64 * 1024
+
+
+def _join_reasoning_with_cap(parts: list[str]) -> str:
+    """Join collected reasoning text parts with newline; truncate at the
+    operator-friendly UI cap.
+
+    Shared tail of every provider's ``extract_reasoning_text`` —
+    Anthropic walks ``thinking`` blocks, OpenAI Responses walks
+    ``reasoning`` items' ``summary`` + ``content``, OpenAI Chat walks
+    synthetic ``reasoning_text`` blocks.  All three converge on the
+    same emit pattern: collect strings, drop empties, join with
+    newline, cap at :data:`MAX_REASONING_DISPLAY_CHARS`.
+    """
+    if not parts:
+        return ""
+    joined = "\n".join(parts)
+    if len(joined) > MAX_REASONING_DISPLAY_CHARS:
+        return joined[:MAX_REASONING_DISPLAY_CHARS]
+    return joined
 
 
 def _lookup_capabilities(
@@ -168,6 +194,21 @@ class LLMProvider(Protocol):
         stream object (which has a ``.close()`` method) before yielding the
         first chunk.  The caller can then close it from another thread to
         abort a blocked HTTP read immediately.
+
+        ``replay_reasoning_to_model`` defaults to ``True`` here (and on
+        every concrete provider's ``create_streaming`` /
+        ``create_completion``) for back-compat with direct callers that
+        haven't been updated to thread the resolver — eval scripts,
+        ad-hoc tests, third-party harnesses.  This is INTENTIONALLY
+        the opposite of the operator-side default
+        (``ModelConfig.replay_reasoning_to_model = False``,
+        ``model_definitions`` server_default ``0``); the resolver in
+        ``ChatSession`` reads the operator value and passes it
+        explicitly, so production call sites never rely on the
+        kwarg-omitted path.  Provider-internal helpers (e.g.
+        ``OpenAIResponsesProvider._convert_messages``) default ``False``
+        because they're called BY the public entry points — once the
+        resolver-driven value lands, it's already explicit.
         """
         ...
 
diff --git a/turnstone/core/session_routes.py b/turnstone/core/session_routes.py
index 7b24cd96..bc803167 100644
--- a/turnstone/core/session_routes.py
+++ b/turnstone/core/session_routes.py
@@ -2378,7 +2378,7 @@ def make_history_handler(cfg: SessionEndpointConfig) -> Handler:
                 # shared mutable state beyond the per-call message
                 # list) so the off-loop hop is free.
                 await asyncio.to_thread(decorate_history_messages, messages, indexes[0], indexes[1])
-                # Active-model ``persist_reasoning`` flag.  Three-tier
+                # Active-model ``surface_persisted_reasoning`` flag.  Three-tier
                 # resolution so the operator's flag-flip takes effect
                 # uniformly — live session, storage-rehydratable cold
                 # workstream, or unknown workstream:
@@ -2393,15 +2393,22 @@ def make_history_handler(cfg: SessionEndpointConfig) -> Handler:
                 # 3. Neither available → conservative default ``True``,
                 #    matching the migration server_default and the
                 #    rehydration default in spec.
-                persist_reasoning = True
+                surface_persisted_reasoning = True
                 resolved_alias = ""
                 resolved_registry: Any = None
                 if live_session is not None:
                     resolved_registry = getattr(live_session, "_registry", None)
                     resolved_alias = getattr(live_session, "_model_alias", "") or ""
                 if not resolved_alias and storage is not None:
+                    # Off-loop the sync storage call (mirrors get_workstream
+                    # / load_messages / load_verdict_indexes / decorate /
+                    # extract_reasoning_for_history above).  Preserves the
+                    # try/except so a DB failure degrades to the
+                    # conservative-default branch instead of bubbling out.
                     try:
-                        ws_cfg = storage.load_workstream_config(ws_id) or {}
+                        ws_cfg = (
+                            await asyncio.to_thread(storage.load_workstream_config, ws_id) or {}
+                        )
                     except Exception:
                         ws_cfg = {}
                     resolved_alias = ws_cfg.get("model_alias") or ""
@@ -2415,12 +2422,14 @@ def make_history_handler(cfg: SessionEndpointConfig) -> Handler:
                     )
                 if resolved_registry is not None and resolved_alias:
                     try:
-                        persist_reasoning = bool(
-                            resolved_registry.get_config(resolved_alias).persist_reasoning
+                        surface_persisted_reasoning = bool(
+                            resolved_registry.get_config(resolved_alias).surface_persisted_reasoning
                         )
                     except Exception:
-                        persist_reasoning = True
-                await asyncio.to_thread(extract_reasoning_for_history, messages, persist_reasoning)
+                        surface_persisted_reasoning = True
+                await asyncio.to_thread(
+                    extract_reasoning_for_history, messages, surface_persisted_reasoning
+                )
             except Exception:
                 # Operationally interesting: a persistent decoration
                 # failure (missing migration, driver mismatch, schema
diff --git a/turnstone/core/storage/_postgresql.py b/turnstone/core/storage/_postgresql.py
index c2a2c806..8b2c15d5 100644
--- a/turnstone/core/storage/_postgresql.py
+++ b/turnstone/core/storage/_postgresql.py
@@ -4167,7 +4167,7 @@ class PostgreSQLBackend:
         temperature: float | None = None,
         max_tokens: int | None = None,
         reasoning_effort: str | None = None,
-        persist_reasoning: bool = True,
+        surface_persisted_reasoning: bool = True,
         replay_reasoning_to_model: bool = False,
     ) -> None:
         from sqlalchemy.dialects import postgresql
@@ -4189,7 +4189,7 @@ class PostgreSQLBackend:
                     temperature=temperature,
                     max_tokens=max_tokens,
                     reasoning_effort=reasoning_effort,
-                    persist_reasoning=1 if persist_reasoning else 0,
+                    surface_persisted_reasoning=1 if surface_persisted_reasoning else 0,
                     replay_reasoning_to_model=1 if replay_reasoning_to_model else 0,
                     created_by=created_by,
                     created=now,
@@ -4209,7 +4209,9 @@ class PostgreSQLBackend:
             ).fetchone()
             if row is None:
                 return None
-            return _row_to_dict(row, "enabled", "persist_reasoning", "replay_reasoning_to_model")
+            return _row_to_dict(
+                row, "enabled", "surface_persisted_reasoning", "replay_reasoning_to_model"
+            )
 
     def get_model_definition_by_alias(self, alias: str) -> dict[str, Any] | None:
 
@@ -4219,7 +4221,9 @@ class PostgreSQLBackend:
             ).fetchone()
             if row is None:
                 return None
-            return _row_to_dict(row, "enabled", "persist_reasoning", "replay_reasoning_to_model")
+            return _row_to_dict(
+                row, "enabled", "surface_persisted_reasoning", "replay_reasoning_to_model"
+            )
 
     def list_model_definitions(self, enabled_only: bool = False) -> list[dict[str, Any]]:
 
@@ -4229,7 +4233,9 @@ class PostgreSQLBackend:
                 q = q.where(model_definitions.c.enabled == 1)
             rows = conn.execute(q).fetchall()
             return [
-                _row_to_dict(r, "enabled", "persist_reasoning", "replay_reasoning_to_model")
+                _row_to_dict(
+                    r, "enabled", "surface_persisted_reasoning", "replay_reasoning_to_model"
+                )
                 for r in rows
             ]
 
@@ -4239,8 +4245,10 @@ class PostgreSQLBackend:
         fields["updated"] = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S")
         if "enabled" in fields:
             fields["enabled"] = 1 if fields["enabled"] else 0
-        if "persist_reasoning" in fields:
-            fields["persist_reasoning"] = 1 if fields["persist_reasoning"] else 0
+        if "surface_persisted_reasoning" in fields:
+            fields["surface_persisted_reasoning"] = (
+                1 if fields["surface_persisted_reasoning"] else 0
+            )
         if "replay_reasoning_to_model" in fields:
             fields["replay_reasoning_to_model"] = 1 if fields["replay_reasoning_to_model"] else 0
         with self._conn() as conn:
diff --git a/turnstone/core/storage/_protocol.py b/turnstone/core/storage/_protocol.py
index dcc222a4..f330b88c 100644
--- a/turnstone/core/storage/_protocol.py
+++ b/turnstone/core/storage/_protocol.py
@@ -1853,7 +1853,7 @@ class StorageBackend(Protocol):
         temperature: float | None = None,
         max_tokens: int | None = None,
         reasoning_effort: str | None = None,
-        persist_reasoning: bool = True,
+        surface_persisted_reasoning: bool = True,
         replay_reasoning_to_model: bool = False,
     ) -> None:
         """Create a model definition. No-op if definition_id already exists."""
diff --git a/turnstone/core/storage/_schema.py b/turnstone/core/storage/_schema.py
index 28ad2240..1045bf4c 100644
--- a/turnstone/core/storage/_schema.py
+++ b/turnstone/core/storage/_schema.py
@@ -660,7 +660,7 @@ model_definitions = sa.Table(
     sa.Column("temperature", sa.Float, nullable=True),
     sa.Column("max_tokens", sa.Integer, nullable=True),
     sa.Column("reasoning_effort", sa.Text, nullable=True),
-    sa.Column("persist_reasoning", sa.Integer, nullable=False, server_default="1"),
+    sa.Column("surface_persisted_reasoning", sa.Integer, nullable=False, server_default="1"),
     sa.Column("replay_reasoning_to_model", sa.Integer, nullable=False, server_default="0"),
     sa.Column("created_by", sa.Text, nullable=False, server_default=""),
     sa.Column("created", sa.Text, nullable=False),
diff --git a/turnstone/core/storage/_sqlite.py b/turnstone/core/storage/_sqlite.py
index 8080429f..3396fa16 100644
--- a/turnstone/core/storage/_sqlite.py
+++ b/turnstone/core/storage/_sqlite.py
@@ -4313,7 +4313,7 @@ class SQLiteBackend:
         temperature: float | None = None,
         max_tokens: int | None = None,
         reasoning_effort: str | None = None,
-        persist_reasoning: bool = True,
+        surface_persisted_reasoning: bool = True,
         replay_reasoning_to_model: bool = False,
     ) -> None:
 
@@ -4334,7 +4334,7 @@ class SQLiteBackend:
                     "temperature": temperature,
                     "max_tokens": max_tokens,
                     "reasoning_effort": reasoning_effort,
-                    "persist_reasoning": 1 if persist_reasoning else 0,
+                    "surface_persisted_reasoning": 1 if surface_persisted_reasoning else 0,
                     "replay_reasoning_to_model": (1 if replay_reasoning_to_model else 0),
                     "created_by": created_by,
                     "created": now,
@@ -4353,7 +4353,9 @@ class SQLiteBackend:
             ).fetchone()
             if row is None:
                 return None
-            return _row_to_dict(row, "enabled", "persist_reasoning", "replay_reasoning_to_model")
+            return _row_to_dict(
+                row, "enabled", "surface_persisted_reasoning", "replay_reasoning_to_model"
+            )
 
     def get_model_definition_by_alias(self, alias: str) -> dict[str, Any] | None:
 
@@ -4363,7 +4365,9 @@ class SQLiteBackend:
             ).fetchone()
             if row is None:
                 return None
-            return _row_to_dict(row, "enabled", "persist_reasoning", "replay_reasoning_to_model")
+            return _row_to_dict(
+                row, "enabled", "surface_persisted_reasoning", "replay_reasoning_to_model"
+            )
 
     def list_model_definitions(self, enabled_only: bool = False) -> list[dict[str, Any]]:
 
@@ -4373,7 +4377,9 @@ class SQLiteBackend:
                 q = q.where(model_definitions.c.enabled == 1)
             rows = conn.execute(q).fetchall()
             return [
-                _row_to_dict(r, "enabled", "persist_reasoning", "replay_reasoning_to_model")
+                _row_to_dict(
+                    r, "enabled", "surface_persisted_reasoning", "replay_reasoning_to_model"
+                )
                 for r in rows
             ]
 
@@ -4383,8 +4389,10 @@ class SQLiteBackend:
         fields["updated"] = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S")
         if "enabled" in fields:
             fields["enabled"] = 1 if fields["enabled"] else 0
-        if "persist_reasoning" in fields:
-            fields["persist_reasoning"] = 1 if fields["persist_reasoning"] else 0
+        if "surface_persisted_reasoning" in fields:
+            fields["surface_persisted_reasoning"] = (
+                1 if fields["surface_persisted_reasoning"] else 0
+            )
         if "replay_reasoning_to_model" in fields:
             fields["replay_reasoning_to_model"] = 1 if fields["replay_reasoning_to_model"] else 0
         with self._conn() as conn:
diff --git a/turnstone/core/storage/_utils.py b/turnstone/core/storage/_utils.py
index 0362477e..b9cf6da2 100644
--- a/turnstone/core/storage/_utils.py
+++ b/turnstone/core/storage/_utils.py
@@ -194,7 +194,7 @@ MODEL_DEFINITION_MUTABLE = frozenset(
         "temperature",
         "max_tokens",
         "reasoning_effort",
-        "persist_reasoning",
+        "surface_persisted_reasoning",
         "replay_reasoning_to_model",
     }
 )
diff --git a/turnstone/core/storage/migrations/versions/052_model_reasoning_persistence.py b/turnstone/core/storage/migrations/versions/052_model_reasoning_persistence.py
index 5faad800..a4bc68b7 100644
--- a/turnstone/core/storage/migrations/versions/052_model_reasoning_persistence.py
+++ b/turnstone/core/storage/migrations/versions/052_model_reasoning_persistence.py
@@ -2,12 +2,15 @@
 
 Adds two boolean (integer-coded) operator knobs:
 
-* ``persist_reasoning`` (default ``1``) — when true, the history-build
-  path extracts stored reasoning text from ``provider_data`` and surfaces
-  it on each assistant message dict so a page refresh re-renders the
-  reasoning bubble. Storage of the reasoning bytes happens regardless;
-  this flag only controls the extract-and-include step in
-  ``_build_history`` / ``decorate_history_messages``.
+* ``surface_persisted_reasoning`` (default ``1``) — when true, the
+  history-build path extracts stored reasoning text from
+  ``provider_data`` and surfaces it on each assistant message dict so a
+  page refresh re-renders the reasoning bubble. **Storage of the
+  reasoning bytes happens regardless of this flag** — it only controls
+  the extract-and-include step in ``_build_history`` /
+  ``decorate_history_messages``.  The earlier name ``surface_persisted_reasoning``
+  was renamed because it implied a storage-control switch; this flag
+  is purely about UI rehydration.
 * ``replay_reasoning_to_model`` (default ``0``) — when true, the
   wire-build path keeps reasoning blocks in the outgoing
   ``_provider_content`` lane on subsequent provider calls. False is the
@@ -41,7 +44,7 @@ def upgrade() -> None:
     with op.batch_alter_table("model_definitions") as batch:
         batch.add_column(
             sa.Column(
-                "persist_reasoning",
+                "surface_persisted_reasoning",
                 sa.Integer,
                 nullable=False,
                 server_default="1",
@@ -60,4 +63,4 @@ def upgrade() -> None:
 def downgrade() -> None:
     with op.batch_alter_table("model_definitions") as batch:
         batch.drop_column("replay_reasoning_to_model")
-        batch.drop_column("persist_reasoning")
+        batch.drop_column("surface_persisted_reasoning")
diff --git a/turnstone/sdk/events.py b/turnstone/sdk/events.py
index 657da6b3..82cbc4a2 100644
--- a/turnstone/sdk/events.py
+++ b/turnstone/sdk/events.py
@@ -65,7 +65,7 @@ class HistoryEvent(ServerEvent):
     * ``reasoning`` — concatenated reasoning text for assistant turns
       that round-tripped a thinking-block lane (Anthropic-with-thinking
       today; OpenAI Responses + Gemini in later phases). Present only
-      when the active model's ``persist_reasoning`` flag is true and
+      when the active model's ``surface_persisted_reasoning`` flag is true and
       the underlying ``provider_data`` carries reasoning blocks.
     """
 
diff --git a/turnstone/server.py b/turnstone/server.py
index 99406421..70e761a4 100644
--- a/turnstone/server.py
+++ b/turnstone/server.py
@@ -477,19 +477,21 @@ def _build_history(
     # Active-model reasoning-persistence flag — defaults True so that a
     # registry/alias lookup miss still surfaces reasoning bubbles.  The
     # default-True semantic mirrors the migration's server_default for
-    # ``model_definitions.persist_reasoning`` and matches the conservative
+    # ``model_definitions.surface_persisted_reasoning`` and matches the conservative
     # rehydration default (Phase 1 spec).
-    persist_reasoning = True
+    surface_persisted_reasoning = True
     registry = getattr(session, "_registry", None)
     model_alias = getattr(session, "_model_alias", "") or ""
     if registry is not None and model_alias:
         try:
-            persist_reasoning = bool(registry.get_config(model_alias).persist_reasoning)
+            surface_persisted_reasoning = bool(
+                registry.get_config(model_alias).surface_persisted_reasoning
+            )
         except Exception:
             # Unknown alias / partially-built registry / dataclass drift —
             # fall back to the conservative default rather than failing
             # the entire history build.
-            persist_reasoning = True
+            surface_persisted_reasoning = True
     history = []
     for msg in session.messages:
         content = msg.get("content")
@@ -580,7 +582,7 @@ def _build_history(
         # ``_provider_content`` lane on ``session.messages`` (set
         # post-commit at ``session.py:3768-3771``).  The lane itself is
         # never copied into ``entry`` — the wire payload stays tight.
-        if msg.get("role") == "assistant" and persist_reasoning:
+        if msg.get("role") == "assistant" and surface_persisted_reasoning:
             reasoning_text = _extract_reasoning_text(msg.get("_provider_content"))
             if reasoning_text:
                 entry["reasoning"] = reasoning_text
diff --git a/turnstone/ui/static/app.js b/turnstone/ui/static/app.js
index 7b7881a4..3cc963fa 100644
--- a/turnstone/ui/static/app.js
+++ b/turnstone/ui/static/app.js
@@ -1199,7 +1199,7 @@ Pane.prototype.replayHistory = function (messages) {
       // live SSE flow (reasoning_delta arrives before content_delta
       // for thinking-enabled models). Mirrors the live-stream
       // construction at the "case 'reasoning':" branch above. Only
-      // surfaces when the active model's persist_reasoning flag is
+      // surfaces when the active model's surface_persisted_reasoning flag is
       // true and the message round-tripped a thinking lane.
       if (msg.reasoning && msg.reasoning.length) {
         var reasonEl = document.createElement("div");