diff --git a/sdk/typescript/src/events.ts b/sdk/typescript/src/events.ts
index 5e44a601..b8332435 100644
--- a/sdk/typescript/src/events.ts
+++ b/sdk/typescript/src/events.ts
@@ -13,6 +13,18 @@ export interface ConnectedEvent {
 
 export interface HistoryEvent {
   type: "history";
+  /**
+   * Per-message dicts the frontend consumes directly. Common optional keys:
+   * - `role`: "user" | "assistant" | "tool"
+   * - `content`: string or list (image/document parts)
+   * - `tool_calls`: assistant turns — list of `{id, name, arguments, verdict?, output_assessment?}`
+   * - `tool_call_id`: tool turns — id of the originating call
+   * - `reminders`: metacognitive nudge bubbles (user/tool channels)
+   * - `advisories`: extracted `UserInterjection` payloads on tool turns
+   * - `reasoning`: concatenated reasoning text for assistant turns whose
+   *   `provider_data` carried thinking blocks (Anthropic today). Present
+   *   only when the active model's `persist_reasoning` flag is true.
+   */
   messages: Array<Record<string, unknown>>;
 }
 
diff --git a/tests/test_build_history_reminders.py b/tests/test_build_history_reminders.py
index 50150cee..832633c0 100644
--- a/tests/test_build_history_reminders.py
+++ b/tests/test_build_history_reminders.py
@@ -141,3 +141,126 @@ class TestRemindersWidening:
         }
         history = _build([msg])
         assert history[0]["reminders"] == [{"type": "denial", "text": "ok"}]
+
+
+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 get_config(self, alias: str) -> Any:
+        return self._cfg
+
+
+def _build_with_registry(
+    messages: list[dict[str, Any]],
+    persist_reasoning: bool = True,
+) -> list[dict[str, Any]]:
+    session = SimpleNamespace(
+        messages=messages,
+        _ws_id="ws-test",
+        _registry=_StubRegistry(persist_reasoning=persist_reasoning),
+        _model_alias="claude-opus-4-7",
+    )
+    with patch(
+        "turnstone.server._load_verdict_indexes",
+        return_value=({}, {}),
+    ):
+        return _build_history(session)
+
+
+class TestReasoningSurfacing:
+    """Phase 1 — surface stored Anthropic thinking blocks on the
+    history payload so refresh-the-page rehydrates the reasoning bubble.
+    Drives through the real ``AnthropicProvider`` extractor (no mock-of-
+    extractor) — only the model registry is stubbed.
+    """
+
+    def test_reasoning_surfaces_for_anthropic_thinking_msg(self) -> None:
+        msg = {
+            "role": "assistant",
+            "content": "Final answer.",
+            "_provider_content": [
+                {"type": "thinking", "thinking": "let me think", "signature": "s"},
+                {"type": "text", "text": "Final answer."},
+            ],
+        }
+        history = _build_with_registry([msg], persist_reasoning=True)
+        assert len(history) == 1
+        assert history[0]["reasoning"] == "let me think"
+
+    def test_reasoning_empty_when_persist_flag_false(self) -> None:
+        msg = {
+            "role": "assistant",
+            "content": "Final answer.",
+            "_provider_content": [
+                {"type": "thinking", "thinking": "hidden", "signature": "s"},
+            ],
+        }
+        history = _build_with_registry([msg], persist_reasoning=False)
+        assert "reasoning" not in history[0]
+
+    def test_provider_content_never_in_wire_entry(self) -> None:
+        # The build path does not copy ``_provider_content`` into the
+        # entry dict regardless of flag — wire payload stays tight.
+        msg = {
+            "role": "assistant",
+            "content": "Final answer.",
+            "_provider_content": [
+                {"type": "thinking", "thinking": "x", "signature": "s"},
+            ],
+        }
+        history = _build_with_registry([msg], persist_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)
+        assert "reasoning" not in history[0]
+
+    def test_no_reasoning_field_for_non_assistant_messages(self) -> None:
+        # Defensive — user/tool messages with a stray _provider_content
+        # do not get the reasoning field stamped.
+        msgs: list[dict[str, Any]] = [
+            {"role": "user", "content": "hi"},
+            {
+                "role": "tool",
+                "tool_call_id": "c1",
+                "content": "out",
+                "_provider_content": [{"type": "thinking", "thinking": "leak", "signature": "s"}],
+            },
+        ]
+        history = _build_with_registry(msgs, persist_reasoning=True)
+        assert "reasoning" not in history[0]
+        assert "reasoning" not in history[1]
+
+    def test_default_true_when_registry_lookup_raises(self) -> None:
+        # Conservative default — Phase 1 spec mandates rehydration on
+        # refresh.  A registry/alias mismatch must not silently kill the
+        # bubble.
+        class BrokenRegistry:
+            def get_config(self, alias: str) -> Any:
+                raise KeyError(alias)
+
+        session = SimpleNamespace(
+            messages=[
+                {
+                    "role": "assistant",
+                    "content": "x",
+                    "_provider_content": [
+                        {"type": "thinking", "thinking": "still works", "signature": "s"}
+                    ],
+                }
+            ],
+            _ws_id="ws-test",
+            _registry=BrokenRegistry(),
+            _model_alias="missing-alias",
+        )
+        with patch(
+            "turnstone.server._load_verdict_indexes",
+            return_value=({}, {}),
+        ):
+            history = _build_history(session)
+        assert history[0]["reasoning"] == "still works"
diff --git a/tests/test_history_decoration.py b/tests/test_history_decoration.py
index f737377a..31fe72c8 100644
--- a/tests/test_history_decoration.py
+++ b/tests/test_history_decoration.py
@@ -453,3 +453,146 @@ class TestDecorateAdvisoryExtraction:
         assert tool_msg["advisories"] == [
             {"type": "user_interjection", "text": "check the logs", "priority": "notice"}
         ]
+
+
+class TestExtractReasoningForHistory:
+    """``extract_reasoning_for_history`` — Phase 1 surfaces stored
+    Anthropic thinking blocks on assistant messages and strips
+    ``_provider_content`` from the wire payload.
+
+    Drives through the real ``AnthropicProvider.extract_reasoning_text``
+    (no mock-of-extractor) — the helper test and the provider unit
+    test (``tests/test_provider_anthropic_reasoning.py``) together
+    catch a regression at either layer distinctly.
+    """
+
+    def _anthropic_thinking_msg(self, text: str = "let me think") -> dict[str, object]:
+        return {
+            "role": "assistant",
+            "content": "Final answer.",
+            "_provider_content": [
+                {"type": "thinking", "thinking": text, "signature": "sig"},
+                {"type": "text", "text": "Final answer."},
+            ],
+        }
+
+    def test_extract_thinking_surfaces_reasoning_field(self) -> None:
+        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)
+        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)
+        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)
+        # Strip is unconditional; reasoning is the conditional bit.
+        assert "_provider_content" not in messages[0]
+        assert "reasoning" not in messages[0]
+
+    def test_first_block_thinking_dispatches_to_anthropic(self) -> None:
+        # Even when text and tool_use blocks follow, the first-block-type
+        # discriminator routes thinking-prefixed payloads correctly.
+        from turnstone.core.history_decoration import extract_reasoning_for_history
+
+        messages = [
+            {
+                "role": "assistant",
+                "content": "x",
+                "_provider_content": [
+                    {"type": "thinking", "thinking": "first", "signature": "s"},
+                    {"type": "text", "text": "spoken"},
+                    {"type": "tool_use", "id": "t1", "name": "f", "input": {}},
+                ],
+            }
+        ]
+        extract_reasoning_for_history(messages, persist_reasoning_flag=True)
+        assert messages[0]["reasoning"] == "first"
+
+    def test_first_block_reasoning_dispatches_to_openai_phase3_stub(self) -> None:
+        # OpenAI Responses extractor returns "" until Phase 3 wires
+        # ``include=["reasoning.encrypted_content"]``.  The dispatcher
+        # must still route to it (not silently fall through to "").
+        # We assert the dispatcher routed by checking the strip happens
+        # AND no reasoning field is added (because the stub returns "").
+        from turnstone.core.history_decoration import extract_reasoning_for_history
+
+        messages = [
+            {
+                "role": "assistant",
+                "content": "x",
+                "_provider_content": [
+                    {"type": "reasoning", "summary": [{"type": "summary_text", "text": "s"}]}
+                ],
+            }
+        ]
+        extract_reasoning_for_history(messages, persist_reasoning_flag=True)
+        assert "reasoning" not in messages[0]
+        assert "_provider_content" not in messages[0]
+
+    def test_unknown_first_block_type_no_op(self) -> None:
+        from turnstone.core.history_decoration import extract_reasoning_for_history
+
+        messages = [
+            {
+                "role": "assistant",
+                "content": "x",
+                "_provider_content": [{"type": "text", "text": "no reasoning here"}],
+            }
+        ]
+        extract_reasoning_for_history(messages, persist_reasoning_flag=True)
+        assert "reasoning" not in messages[0]
+        assert "_provider_content" not in messages[0]
+
+    def test_skips_messages_without_provider_content(self) -> None:
+        from turnstone.core.history_decoration import extract_reasoning_for_history
+
+        messages = [{"role": "assistant", "content": "plain"}]
+        extract_reasoning_for_history(messages, persist_reasoning_flag=True)
+        assert "reasoning" not in messages[0]
+        assert messages[0]["content"] == "plain"
+
+    def test_user_and_tool_messages_untouched(self) -> None:
+        from turnstone.core.history_decoration import extract_reasoning_for_history
+
+        messages: list[dict[str, object]] = [
+            {"role": "user", "content": "hi"},
+            {"role": "tool", "tool_call_id": "c1", "content": "out"},
+            self._anthropic_thinking_msg("only this one"),
+        ]
+        extract_reasoning_for_history(messages, persist_reasoning_flag=True)
+        assert "reasoning" not in messages[0]
+        assert "reasoning" not in messages[1]
+        assert messages[2]["reasoning"] == "only this one"
+
+    def test_empty_provider_content_no_extraction(self) -> None:
+        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)
+        assert "reasoning" not in messages[0]
+        # Empty-list provider_content is still stripped from the wire.
+        assert "_provider_content" not in messages[0]
+
+    def test_first_block_not_a_dict_skipped(self) -> None:
+        from turnstone.core.history_decoration import extract_reasoning_for_history
+
+        messages: list[dict[str, object]] = [
+            {
+                "role": "assistant",
+                "content": "x",
+                "_provider_content": ["bogus"],
+            }
+        ]
+        extract_reasoning_for_history(messages, persist_reasoning_flag=True)
+        assert "reasoning" not in messages[0]
+        assert "_provider_content" not in messages[0]
diff --git a/tests/test_model_definition_storage.py b/tests/test_model_definition_storage.py
index e7591f13..ea3b2dd5 100644
--- a/tests/test_model_definition_storage.py
+++ b/tests/test_model_definition_storage.py
@@ -212,3 +212,73 @@ class TestModelDefinitionStorage:
         m = db.get_model_definition(did)
         assert m is not None
         assert m["temperature"] is None
+
+    def test_reasoning_flags_default(self, db: SQLiteBackend) -> None:
+        """persist_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["replay_reasoning_to_model"] is False
+
+    def test_create_with_explicit_reasoning_flags(self, db: SQLiteBackend) -> None:
+        did = _make_id()
+        db.create_model_definition(
+            definition_id=did,
+            alias="reason-explicit",
+            model="claude-opus-4-7",
+            persist_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["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["replay_reasoning_to_model"] is True
+
+    def test_update_persist_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)
+        assert ok is True
+        m = db.get_model_definition(did)
+        assert m is not None
+        assert m["persist_reasoning"] is False
+        assert m["replay_reasoning_to_model"] is False  # untouched
+
+    def test_update_replay_reasoning_to_model(self, db: SQLiteBackend) -> None:
+        did = _make_id()
+        db.create_model_definition(definition_id=did, alias="upd-replay", model="gpt-5")
+        ok = db.update_model_definition(did, replay_reasoning_to_model=True)
+        assert ok is True
+        m = db.get_model_definition(did)
+        assert m is not None
+        assert m["persist_reasoning"] is True  # untouched
+        assert m["replay_reasoning_to_model"] is True
+
+    def test_list_returns_reasoning_flags(self, db: SQLiteBackend) -> None:
+        db.create_model_definition(
+            definition_id=_make_id(),
+            alias="list-a",
+            model="gpt-5",
+            persist_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,
+            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"]["replay_reasoning_to_model"] is False
+        assert by_alias["list-b"]["persist_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 683eab40..85d53e10 100644
--- a/tests/test_model_registry.py
+++ b/tests/test_model_registry.py
@@ -76,6 +76,23 @@ class TestModelConfig:
         assert cfg.temperature == 0.0
         assert cfg.temperature is not None
 
+    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.replay_reasoning_to_model is False
+
+    def test_reasoning_flags_set(self) -> None:
+        cfg = ModelConfig(
+            alias="x",
+            base_url="x",
+            api_key="x",
+            model="x",
+            persist_reasoning=False,
+            replay_reasoning_to_model=True,
+        )
+        assert cfg.persist_reasoning is False
+        assert cfg.replay_reasoning_to_model is True
+
 
 # ---------------------------------------------------------------------------
 # ModelRegistry
@@ -686,6 +703,53 @@ class TestLoadModelRegistryWithDB:
         assert cfg.max_tokens is None
         assert cfg.reasoning_effort is None
 
+    def test_db_reasoning_flags_loaded(self) -> None:
+        """Per-model reasoning flags from DB are carried in ModelConfig."""
+        storage = _MockStorage(
+            [
+                {
+                    "alias": "anth-thinking",
+                    "model": "claude-opus-4-7",
+                    "provider": "anthropic",
+                    "base_url": "",
+                    "api_key": "sk-anth",
+                    "context_window": 200000,
+                    "capabilities": "{}",
+                    "enabled": True,
+                    "persist_reasoning": False,
+                    "replay_reasoning_to_model": True,
+                }
+            ]
+        )
+        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.replay_reasoning_to_model is True
+
+    def test_db_reasoning_flags_default_when_absent(self) -> None:
+        """Pre-052 rows without the columns degrade to dataclass defaults."""
+        storage = _MockStorage(
+            [
+                {
+                    "alias": "legacy-row",
+                    "model": "gpt-5",
+                    "provider": "openai",
+                    "base_url": "",
+                    "api_key": "",
+                    "context_window": 32768,
+                    "capabilities": "{}",
+                    "enabled": True,
+                    # persist_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.replay_reasoning_to_model is False
+
     def test_db_default_alias_not_clobbered(self) -> None:
         """DB model with alias='default' is not overwritten by CLI args."""
         storage = _MockStorage(
diff --git a/tests/test_provider_anthropic_reasoning.py b/tests/test_provider_anthropic_reasoning.py
new file mode 100644
index 00000000..af5972fc
--- /dev/null
+++ b/tests/test_provider_anthropic_reasoning.py
@@ -0,0 +1,123 @@
+"""Tests for ``AnthropicProvider.extract_reasoning_text``.
+
+Phase 1 of the optional-reasoning-persistence feature: provider-side
+extractor that walks stored ``provider_blocks`` and returns the
+concatenated thinking text, capped at the operator-friendly UI display
+size.
+
+These tests drive through the real ``AnthropicProvider`` instance — no
+mocks of the extractor itself — using fixture-shaped blocks that match
+what ``_iter_anthropic_stream`` actually accumulates at
+``_anthropic.py:713-724`` (``thinking_delta`` + ``signature_delta``
+combined into ``{"type": "thinking", "thinking": <text>, "signature":
+<sig>}``).
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from turnstone.core.providers._anthropic import (
+    _MAX_REASONING_DISPLAY_BYTES,
+    AnthropicProvider,
+)
+from turnstone.core.providers._openai_chat import OpenAIChatCompletionsProvider
+from turnstone.core.providers._openai_responses import OpenAIResponsesProvider
+
+
+@pytest.fixture
+def anthropic() -> AnthropicProvider:
+    return AnthropicProvider()
+
+
+class TestExtractReasoningText:
+    def test_none_input_returns_empty_string(self, anthropic: AnthropicProvider) -> None:
+        assert anthropic.extract_reasoning_text(None) == ""
+
+    def test_empty_list_returns_empty_string(self, anthropic: AnthropicProvider) -> None:
+        assert anthropic.extract_reasoning_text([]) == ""
+
+    def test_no_thinking_blocks_returns_empty(self, anthropic: AnthropicProvider) -> None:
+        blocks: list[dict[str, object]] = [
+            {"type": "text", "text": "hello"},
+            {"type": "tool_use", "id": "t1", "name": "x", "input": {}},
+        ]
+        assert anthropic.extract_reasoning_text(blocks) == ""
+
+    def test_single_thinking_block_returns_text(self, anthropic: AnthropicProvider) -> None:
+        blocks = [{"type": "thinking", "thinking": "Let me think about this.", "signature": "abc"}]
+        assert anthropic.extract_reasoning_text(blocks) == "Let me think about this."
+
+    def test_multiple_thinking_blocks_joined_with_newline(
+        self, anthropic: AnthropicProvider
+    ) -> None:
+        blocks = [
+            {"type": "thinking", "thinking": "first thought", "signature": "s1"},
+            {"type": "thinking", "thinking": "second thought", "signature": "s2"},
+        ]
+        assert anthropic.extract_reasoning_text(blocks) == "first thought\nsecond thought"
+
+    def test_mixed_blocks_extracts_only_thinking(self, anthropic: AnthropicProvider) -> None:
+        blocks = [
+            {"type": "thinking", "thinking": "reason A", "signature": "s"},
+            {"type": "text", "text": "visible answer"},
+            {"type": "tool_use", "id": "t1", "name": "x", "input": {}},
+            {"type": "thinking", "thinking": "reason B", "signature": "s"},
+        ]
+        assert anthropic.extract_reasoning_text(blocks) == "reason A\nreason B"
+
+    def test_thinking_block_without_thinking_field_skipped(
+        self, anthropic: AnthropicProvider
+    ) -> None:
+        blocks = [{"type": "thinking", "signature": "s"}]
+        assert anthropic.extract_reasoning_text(blocks) == ""
+
+    def test_thinking_block_with_empty_text_skipped(self, anthropic: AnthropicProvider) -> None:
+        blocks = [{"type": "thinking", "thinking": "", "signature": "s"}]
+        assert anthropic.extract_reasoning_text(blocks) == ""
+
+    def test_truncation_at_64kib_cap(self, anthropic: AnthropicProvider) -> None:
+        long_text = "x" * (_MAX_REASONING_DISPLAY_BYTES + 1024)
+        blocks = [{"type": "thinking", "thinking": long_text, "signature": "s"}]
+        result = anthropic.extract_reasoning_text(blocks)
+        assert len(result) == _MAX_REASONING_DISPLAY_BYTES
+
+    def test_just_under_cap_not_truncated(self, anthropic: AnthropicProvider) -> None:
+        text = "y" * (_MAX_REASONING_DISPLAY_BYTES - 1)
+        blocks = [{"type": "thinking", "thinking": text, "signature": "s"}]
+        assert anthropic.extract_reasoning_text(blocks) == text
+
+    def test_malformed_block_entry_skipped(self, anthropic: AnthropicProvider) -> None:
+        # A defensive sanity check — we should not crash if some
+        # entry isn't a dict (e.g. a corrupted JSON payload).
+        blocks = [
+            "not a dict",  # type: ignore[list-item]
+            {"type": "thinking", "thinking": "good one", "signature": "s"},
+        ]
+        assert anthropic.extract_reasoning_text(blocks) == "good one"  # type: ignore[arg-type]
+
+    def test_non_list_input_returns_empty(self, anthropic: AnthropicProvider) -> None:
+        # Defensive against a corrupted provider_data payload.
+        assert anthropic.extract_reasoning_text("not a list") == ""  # type: ignore[arg-type]
+        assert anthropic.extract_reasoning_text({"type": "thinking"}) == ""  # type: ignore[arg-type]
+
+
+class TestOtherProvidersDefault:
+    """Non-Anthropic providers return "" for the same fixture shapes."""
+
+    def test_openai_chat_returns_empty(self) -> None:
+        provider = OpenAIChatCompletionsProvider()
+        blocks = [{"type": "thinking", "thinking": "would-be-text", "signature": "s"}]
+        assert provider.extract_reasoning_text(blocks) == ""
+
+    def test_openai_responses_returns_empty(self) -> None:
+        provider = OpenAIResponsesProvider()
+        blocks = [{"type": "thinking", "thinking": "would-be-text", "signature": "s"}]
+        assert provider.extract_reasoning_text(blocks) == ""
+
+    def test_openai_responses_returns_empty_on_reasoning_shaped_blocks(self) -> None:
+        # Phase 3 stub — reasoning items exist but extractor returns ""
+        # until ``include=["reasoning.encrypted_content"]`` is wired.
+        provider = OpenAIResponsesProvider()
+        blocks = [{"type": "reasoning", "summary": [{"type": "summary_text", "text": "x"}]}]
+        assert provider.extract_reasoning_text(blocks) == ""
diff --git a/tests/test_reasoning_audit_log_discipline.py b/tests/test_reasoning_audit_log_discipline.py
new file mode 100644
index 00000000..630682ee
--- /dev/null
+++ b/tests/test_reasoning_audit_log_discipline.py
@@ -0,0 +1,194 @@
+"""Audit-log discipline test for reasoning text.
+
+Phase 1 of optional reasoning persistence surfaces stored thinking
+blocks on the ``/history`` payload (UI rehydration). The bytes ride
+through the helper (``extract_reasoning_for_history``), through the
+provider extractor (``AnthropicProvider.extract_reasoning_text``), and
+through the server build path (``_build_history``).
+
+This test pins the security-sensitive contract:
+
+    Reasoning text MAY land on ``msg["reasoning"]`` (UI-bound),
+    but MUST NOT appear in any ``Logger.info`` / ``warning`` /
+    ``error`` payload at any layer in the pipeline.
+
+The test mocks the standard-library ``logging.Logger`` info/warning/
+error methods, runs a thinking-bearing turn through the relevant
+extractors and history build, then asserts no captured log call's
+positional args or kwargs contain the unique marker string. Replaces
+the v4 grep-the-output approach (fragile when log strings are
+formatted) with a structural mock-and-assert (tests the actual
+contract rather than the rendered text).
+"""
+
+from __future__ import annotations
+
+import logging
+from types import SimpleNamespace
+from typing import Any
+from unittest.mock import patch
+
+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.server import _build_history
+
+_MARKER = "SECRET_REASONING_MARKER_xyz123_unlikely_collision"
+
+
+def _payload_contains_marker(args: tuple[Any, ...], kwargs: dict[str, Any]) -> bool:
+    """Walk a captured log call's args + kwargs for the marker string.
+
+    Logger.info-style calls accept a format string + positional substitution
+    args; the marker could appear in either the format string itself or
+    the substitution values. Format-time strings (``%`` substitution) are
+    NOT inspected because they're a stdlib formatting concern, not a
+    callable our pipeline reaches into. The structural check is "no
+    user-controlled marker appears in any arg slot we passed".
+    """
+    for a in args:
+        if isinstance(a, str) and _MARKER in a:
+            return True
+        # Defensive — a list/dict/exception arg might carry the marker too.
+        try:
+            if _MARKER in repr(a):
+                return True
+        except Exception:
+            continue
+    for v in kwargs.values():
+        if isinstance(v, str) and _MARKER in v:
+            return True
+        try:
+            if _MARKER in repr(v):
+                return True
+        except Exception:
+            continue
+    return False
+
+
+def _capture_log_calls():
+    """Capture every Logger.info / warning / error call into a single list."""
+    captured: list[tuple[str, tuple[Any, ...], dict[str, Any]]] = []
+
+    def make_recorder(level: str):
+        def _rec(*args: Any, **kwargs: Any) -> None:
+            captured.append((level, args, kwargs))
+
+        return _rec
+
+    return captured, [
+        patch.object(logging.Logger, "info", side_effect=make_recorder("info"), autospec=True),
+        patch.object(
+            logging.Logger, "warning", side_effect=make_recorder("warning"), autospec=True
+        ),
+        patch.object(logging.Logger, "error", side_effect=make_recorder("error"), autospec=True),
+    ]
+
+
+class TestReasoningAuditLogDiscipline:
+    """Reasoning text never lands at INFO+ severity on any logger."""
+
+    def _thinking_msg(self, text: str = _MARKER) -> dict[str, Any]:
+        return {
+            "role": "assistant",
+            "content": "Final answer.",
+            "_provider_content": [
+                {"type": "thinking", "thinking": text, "signature": "sig"},
+                {"type": "text", "text": "Final answer."},
+            ],
+        }
+
+    def test_anthropic_extractor_does_not_log_reasoning(self) -> None:
+        captured, patchers = _capture_log_calls()
+        for p in patchers:
+            p.start()
+        try:
+            provider = AnthropicProvider()
+            text = provider.extract_reasoning_text(
+                [{"type": "thinking", "thinking": _MARKER, "signature": "s"}]
+            )
+            assert text == _MARKER  # extractor IS allowed to return it
+        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.extract_reasoning_text leaked reasoning text "
+            f"into INFO+ logs: {offending}"
+        )
+
+    def test_dispatch_helper_does_not_log_reasoning(self) -> None:
+        captured, patchers = _capture_log_calls()
+        for p in patchers:
+            p.start()
+        try:
+            text = extract_reasoning_text_from_provider_content(
+                [{"type": "thinking", "thinking": _MARKER, "signature": "s"}]
+            )
+            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"extract_reasoning_text_from_provider_content leaked reasoning "
+            f"text into INFO+ logs: {offending}"
+        )
+
+    def test_list_helper_does_not_log_reasoning(self) -> None:
+        captured, patchers = _capture_log_calls()
+        for p in patchers:
+            p.start()
+        try:
+            messages = [self._thinking_msg(_MARKER)]
+            extract_reasoning_for_history(messages, persist_reasoning_flag=True)
+            assert messages[0]["reasoning"] == _MARKER  # UI-bound 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"extract_reasoning_for_history leaked reasoning text into INFO+ logs: {offending}"
+        )
+
+    def test_build_history_does_not_log_reasoning(self) -> None:
+        registry = SimpleNamespace(get_config=lambda alias: SimpleNamespace(persist_reasoning=True))
+        session = SimpleNamespace(
+            messages=[self._thinking_msg(_MARKER)],
+            _ws_id="ws-audit",
+            _registry=registry,
+            _model_alias="claude-opus-4-7",
+        )
+        captured, patchers = _capture_log_calls()
+        for p in patchers:
+            p.start()
+        try:
+            with patch(
+                "turnstone.server._load_verdict_indexes",
+                return_value=({}, {}),
+            ):
+                history = _build_history(session)
+            assert history[0]["reasoning"] == _MARKER  # UI-bound 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"_build_history leaked reasoning text into INFO+ logs: {offending}"
diff --git a/tests/test_workstream_endpoints.py b/tests/test_workstream_endpoints.py
index d8161466..d033c374 100644
--- a/tests/test_workstream_endpoints.py
+++ b/tests/test_workstream_endpoints.py
@@ -2,7 +2,9 @@
 
 from __future__ import annotations
 
+import json
 import queue
+from types import SimpleNamespace
 from typing import TYPE_CHECKING, Any
 from unittest.mock import MagicMock, patch
 
@@ -1762,3 +1764,155 @@ class TestTenantCheckOnReadEndpoints:
         assert cold_check in offloaded, (
             f"tenant_check must be invoked through asyncio.to_thread; got {offloaded}"
         )
+
+
+class TestHistoryReasoningRehydration:
+    """The lifted ``GET /v1/api/workstreams/{ws_id}/history`` surfaces
+    stored Anthropic thinking blocks on assistant messages so a page
+    refresh re-renders the reasoning bubble. Drives through the real
+    ``AnthropicProvider.extract_reasoning_text`` and the storage
+    ``reconstruct_messages`` boundary that JSON-decodes
+    ``provider_data`` into ``_provider_content``.
+    """
+
+    def test_history_handler_surfaces_reasoning_for_anthropic_thinking(self, _inject_storage):
+        ws_id = "ws-reason-1"
+        _inject_storage.register_workstream(ws_id, kind="interactive", user_id="test-user")
+        provider_data = json.dumps(
+            [
+                {"type": "thinking", "thinking": "let me reason", "signature": "s"},
+                {"type": "text", "text": "Final answer."},
+            ]
+        )
+        _inject_storage.save_message(
+            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.
+        mock_mgr = MagicMock()
+        mock_mgr.get.return_value = None
+        client = _build_history_app(mock_mgr, _inject_storage)
+
+        r = client.get(f"/v1/api/workstreams/{ws_id}/history")
+        assert r.status_code == 200
+        msgs = r.json()["messages"]
+        assistant = next(m for m in msgs if m.get("role") == "assistant")
+        assert assistant["reasoning"] == "let me reason"
+
+    def test_history_handler_strips_provider_content(self, _inject_storage):
+        ws_id = "ws-reason-2"
+        _inject_storage.register_workstream(ws_id, kind="interactive", user_id="test-user")
+        provider_data = json.dumps([{"type": "thinking", "thinking": "x", "signature": "s"}])
+        _inject_storage.save_message(ws_id, "assistant", "Answer.", provider_data=provider_data)
+        mock_mgr = MagicMock()
+        mock_mgr.get.return_value = None
+        client = _build_history_app(mock_mgr, _inject_storage)
+
+        r = client.get(f"/v1/api/workstreams/{ws_id}/history")
+        assert r.status_code == 200
+        for m in r.json()["messages"]:
+            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
+        model suppresses the reasoning field even when the data is
+        stored. ``_provider_content`` is still stripped from the wire.
+        """
+        ws_id = "ws-reason-3"
+        _inject_storage.register_workstream(ws_id, kind="interactive", user_id="test-user")
+        provider_data = json.dumps([{"type": "thinking", "thinking": "hidden", "signature": "s"}])
+        _inject_storage.save_message(ws_id, "assistant", "Answer.", provider_data=provider_data)
+        live_session = SimpleNamespace(
+            id=ws_id,
+            _registry=SimpleNamespace(
+                get_config=lambda alias: SimpleNamespace(persist_reasoning=False)
+            ),
+            _model_alias="claude-opus-4-7",
+        )
+        mock_mgr = MagicMock()
+        mock_mgr.get.return_value = live_session
+        client = _build_history_app(mock_mgr, _inject_storage)
+
+        r = client.get(f"/v1/api/workstreams/{ws_id}/history")
+        assert r.status_code == 200
+        for m in r.json()["messages"]:
+            if m.get("role") == "assistant":
+                assert "reasoning" not in m
+            assert "_provider_content" not in m
+
+    def test_history_handler_cold_workstream_resolves_via_workstream_config(self, _inject_storage):
+        """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
+        on ``app.state``. Operator flag-flip is honored uniformly
+        across live and cold workstreams.
+        """
+        ws_id = "ws-reason-cold"
+        _inject_storage.register_workstream(ws_id, kind="interactive", user_id="test-user")
+        # Simulate the model alias persisted by the rehydrate path
+        # (session_manager.py:628-629 reads it back via the same key).
+        _inject_storage.save_workstream_config(ws_id, {"model_alias": "claude-opus-4-7"})
+        provider_data = json.dumps(
+            [{"type": "thinking", "thinking": "should not surface", "signature": "s"}]
+        )
+        _inject_storage.save_message(ws_id, "assistant", "Answer.", provider_data=provider_data)
+        # No live session — handler falls back to workstream_config + registry.
+        mock_mgr = MagicMock()
+        mock_mgr.get.return_value = None
+
+        # Build the app with a global registry that reports persist=False
+        # for the saved alias.
+        cfg = _interactive_endpoint_cfg(mock_mgr)
+        handler = make_history_handler(cfg)
+        app = Starlette(
+            routes=[
+                Mount(
+                    "/v1",
+                    routes=[
+                        Route(
+                            "/api/workstreams/{ws_id}/history",
+                            handler,
+                            methods=["GET"],
+                        ),
+                    ],
+                ),
+            ],
+            middleware=[Middleware(_InjectAuthMiddleware)],
+        )
+        app.state.workstreams = mock_mgr
+        app.state.auth_storage = _inject_storage
+        app.state.registry = SimpleNamespace(
+            get_config=lambda alias: SimpleNamespace(
+                persist_reasoning=(alias != "claude-opus-4-7"),
+            )
+        )
+        client = TestClient(app)
+
+        r = client.get(f"/v1/api/workstreams/{ws_id}/history")
+        assert r.status_code == 200
+        # Flag-flip on the saved alias is honored: reasoning suppressed.
+        for m in r.json()["messages"]:
+            if m.get("role") == "assistant":
+                assert "reasoning" not in m
+            assert "_provider_content" not in m
+
+    def test_history_handler_cold_workstream_no_alias_defaults_true(self, _inject_storage):
+        """A workstream that pre-dates the rehydrate-time alias persist
+        (or one that simply has no workstream_config row) falls through
+        to the conservative default ``True``.  Reasoning surfaces.
+        """
+        ws_id = "ws-reason-cold-no-alias"
+        _inject_storage.register_workstream(ws_id, kind="interactive", user_id="test-user")
+        provider_data = json.dumps(
+            [{"type": "thinking", "thinking": "default-true wins", "signature": "s"}]
+        )
+        _inject_storage.save_message(ws_id, "assistant", "Answer.", provider_data=provider_data)
+        mock_mgr = MagicMock()
+        mock_mgr.get.return_value = None
+        client = _build_history_app(mock_mgr, _inject_storage)
+
+        r = client.get(f"/v1/api/workstreams/{ws_id}/history")
+        assert r.status_code == 200
+        assistant = next(m for m in r.json()["messages"] if m.get("role") == "assistant")
+        assert assistant["reasoning"] == "default-true wins"
diff --git a/turnstone/api/console_schemas.py b/turnstone/api/console_schemas.py
index 3c475d9e..96266d33 100644
--- a/turnstone/api/console_schemas.py
+++ b/turnstone/api/console_schemas.py
@@ -908,6 +908,8 @@ class ModelDefinitionInfo(BaseModel):
     temperature: float | None = None
     max_tokens: int | None = None
     reasoning_effort: str | None = None
+    persist_reasoning: bool = True
+    replay_reasoning_to_model: bool = False
     source: str = ""
     created_by: str = ""
     created: str = ""
@@ -926,6 +928,8 @@ class CreateModelDefinitionRequest(BaseModel):
     temperature: float | None = None
     max_tokens: int | None = None
     reasoning_effort: str | None = None
+    persist_reasoning: bool = True
+    replay_reasoning_to_model: bool = False
 
 
 class UpdateModelDefinitionRequest(BaseModel):
@@ -940,6 +944,8 @@ class UpdateModelDefinitionRequest(BaseModel):
     temperature: float | None = None
     max_tokens: int | None = None
     reasoning_effort: str | None = None
+    persist_reasoning: bool | None = None
+    replay_reasoning_to_model: bool | None = None
 
 
 class ListModelDefinitionsResponse(BaseModel):
diff --git a/turnstone/console/server.py b/turnstone/console/server.py
index 46933341..f2c61d6d 100644
--- a/turnstone/console/server.py
+++ b/turnstone/console/server.py
@@ -9904,6 +9904,9 @@ async def admin_create_model_definition(request: Request) -> JSONResponse:
         if not reasoning_effort:
             reasoning_effort = None
 
+    persist_reasoning = bool(body.get("persist_reasoning", True))
+    replay_reasoning_to_model = bool(body.get("replay_reasoning_to_model", False))
+
     storage.create_model_definition(
         definition_id=definition_id,
         alias=alias,
@@ -9918,6 +9921,8 @@ async def admin_create_model_definition(request: Request) -> JSONResponse:
         temperature=temperature,
         max_tokens=max_tokens,
         reasoning_effort=reasoning_effort,
+        persist_reasoning=persist_reasoning,
+        replay_reasoning_to_model=replay_reasoning_to_model,
     )
 
     record_audit(
@@ -10079,6 +10084,10 @@ 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 "replay_reasoning_to_model" in body:
+        updates["replay_reasoning_to_model"] = bool(body["replay_reasoning_to_model"])
 
     if updates:
         storage.update_model_definition(definition_id, **updates)
diff --git a/turnstone/console/static/admin.js b/turnstone/console/static/admin.js
index 4b311874..b88e69f1 100644
--- a/turnstone/console/static/admin.js
+++ b/turnstone/console/static/admin.js
@@ -5008,6 +5008,11 @@ function _renderModels(items) {
     if (m.max_tokens != null) overrides.push("max_tok=" + m.max_tokens);
     if (m.reasoning_effort != null)
       overrides.push("effort=" + m.reasoning_effort);
+    // 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.replay_reasoning_to_model === true) overrides.push("replay=on");
     if (overrides.length) {
       var ovrSpan = document.createElement("span");
       ovrSpan.className = "model-overrides-hint";
@@ -5196,6 +5201,8 @@ function showCreateModelModal() {
     el.style.borderColor = "";
   });
   document.getElementById("model-enabled").checked = true;
+  document.getElementById("model-persist-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;
   document.getElementById("model-detect-btn").textContent = "Detect";
@@ -5278,6 +5285,13 @@ function showEditModelModal(definitionId) {
       document.getElementById("model-capabilities").value =
         capsText === "{}" ? "" : capsText;
       document.getElementById("model-enabled").checked = m.enabled !== false;
+      // 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-replay-reasoning").checked =
+        m.replay_reasoning_to_model === true;
       _applyProviderDefaults();
     })
     .catch(function () {
@@ -5419,6 +5433,16 @@ function submitCreateModel() {
     form.reasoning_effort = null;
   }
 
+  // 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",
+  ).checked;
+  form.replay_reasoning_to_model = document.getElementById(
+    "model-replay-reasoning",
+  ).checked;
+
   var apiKey = document.getElementById("model-api-key").value;
   if (apiKey) form.api_key = apiKey;
 
diff --git a/turnstone/console/static/coordinator/coordinator.js b/turnstone/console/static/coordinator/coordinator.js
index 3fd739e9..ca2d7941 100644
--- a/turnstone/console/static/coordinator/coordinator.js
+++ b/turnstone/console/static/coordinator/coordinator.js
@@ -4195,6 +4195,20 @@
             appendUserMessageWithAttachments(text, [], { label: "user" });
           });
         } else if (role === "assistant") {
+          // Reasoning bubble (Phase 1 reasoning persistence) — render
+          // BEFORE the content card so the visual order matches the
+          // live SSE flow (reasoning_delta arrives before content_delta
+          // for thinking-enabled models). Mirrors the live ":1524" /
+          // 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
+          // a thinking lane.
+          if (typeof m.reasoning === "string" && m.reasoning.length) {
+            const rEl = appendMsg("reasoning", "", { label: "reasoning" });
+            const rBody = rEl && rEl.querySelector(".msg-body");
+            if (rBody) rBody.textContent = m.reasoning;
+          }
           // Render content BEFORE the tool batch so DOM order matches
           // chronological order (the model emits text first, then
           // dispatches tools).  Whitespace-only content (e.g. "\n\n"
diff --git a/turnstone/console/static/index.html b/turnstone/console/static/index.html
index 56703a81..ea5791c2 100644
--- a/turnstone/console/static/index.html
+++ b/turnstone/console/static/index.html
@@ -4116,7 +4116,7 @@
           placeholder='{"supports_vision": true}'
           style="font-family: var(--font-mono); font-size: 11px"
         ></textarea>
-        <div style="display: flex; gap: 20px; margin-top: 14px">
+        <div style="display: flex; gap: 20px; margin-top: 14px; flex-wrap: wrap">
           <label style="margin: 0; font-size: 12px; color: var(--fg-dim)"
             ><input
               type="checkbox"
@@ -4125,6 +4125,25 @@
               style="margin-right: 5px"
             />Enabled</label
           >
+          <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."
+            ><input
+              type="checkbox"
+              id="model-persist-reasoning"
+              checked
+              style="margin-right: 5px"
+            />Persist reasoning</label
+          >
+          <label
+            style="margin: 0; font-size: 12px; color: var(--fg-dim)"
+            title="Replay stored reasoning blocks back to the model on subsequent provider calls. Capability-dependent; off by default for cost/spec compliance."
+            ><input
+              type="checkbox"
+              id="model-replay-reasoning"
+              style="margin-right: 5px"
+            />Replay reasoning to model</label
+          >
         </div>
         <div id="model-detect-area" style="margin-top: 14px">
           <button
diff --git a/turnstone/core/history_decoration.py b/turnstone/core/history_decoration.py
index 629ee89f..459b7510 100644
--- a/turnstone/core/history_decoration.py
+++ b/turnstone/core/history_decoration.py
@@ -19,7 +19,7 @@ either an async caller (via ``asyncio.to_thread``) or a sync hook.
 from __future__ import annotations
 
 import json
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 from turnstone.core.log import get_logger
 from turnstone.core.tool_advisory import (
@@ -268,6 +268,105 @@ def extract_advisories_from_tool_envelope(
     return _entity_decode_wrapper_tags(inner), advisories
 
 
+if TYPE_CHECKING:
+    from turnstone.core.providers._anthropic import AnthropicProvider
+    from turnstone.core.providers._openai_responses import OpenAIResponsesProvider
+
+_anthropic_provider_singleton: AnthropicProvider | None = None
+_openai_responses_provider_singleton: OpenAIResponsesProvider | None = None
+
+
+def _get_anthropic_provider() -> AnthropicProvider:
+    global _anthropic_provider_singleton
+    if _anthropic_provider_singleton is None:
+        from turnstone.core.providers._anthropic import AnthropicProvider as _Anthropic
+
+        _anthropic_provider_singleton = _Anthropic()
+    return _anthropic_provider_singleton
+
+
+def _get_openai_responses_provider() -> OpenAIResponsesProvider:
+    global _openai_responses_provider_singleton
+    if _openai_responses_provider_singleton is None:
+        from turnstone.core.providers._openai_responses import (
+            OpenAIResponsesProvider as _OpenAIResp,
+        )
+
+        _openai_responses_provider_singleton = _OpenAIResp()
+    return _openai_responses_provider_singleton
+
+
+def extract_reasoning_text_from_provider_content(provider_content: Any) -> str:
+    """Dispatch reasoning extraction by first-block ``type`` field.
+
+    Routing is structural — block shape is non-overlapping across
+    providers by API design (Anthropic ``thinking``, OpenAI Responses
+    ``reasoning``, Gemini ``thought``).  Phase 1 wires Anthropic
+    extraction; OpenAI Responses returns ``""`` until Phase 3 adds the
+    ``include=["reasoning.encrypted_content"]`` request flag.  Returns
+    ``""`` for empty / missing / non-list / unknown-type input.
+
+    Pure transform — safe from any thread.  Both history surfaces
+    (interactive ``_build_history`` and lifted ``make_history_handler``)
+    call this directly.
+    """
+    if not isinstance(provider_content, list) or not provider_content:
+        return ""
+    first_block = provider_content[0]
+    if not isinstance(first_block, dict):
+        return ""
+    block_type = first_block.get("type")
+    if not isinstance(block_type, str):
+        return ""
+    if block_type == "thinking":
+        return _get_anthropic_provider().extract_reasoning_text(provider_content)
+    if block_type == "reasoning":
+        return _get_openai_responses_provider().extract_reasoning_text(provider_content)
+    return ""
+
+
+def extract_reasoning_for_history(
+    messages: list[dict[str, Any]],
+    persist_reasoning_flag: bool,
+) -> None:
+    """Surface stored reasoning text on each assistant message; strip the
+    raw provider content from the wire payload.
+
+    For the ``make_history_handler`` REST path where the response
+    payload IS the messages list returned from ``storage.load_messages``
+    — 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
+    dispatcher returned non-empty text.  Strips ``_provider_content``
+    unconditionally — the field is internal and never read by either UI.
+
+    The interactive ``_build_history`` surface DOES NOT call this
+    helper; it builds new entry dicts from scratch and calls
+    :func:`extract_reasoning_text_from_provider_content` directly per
+    assistant message, stamping ``entry["reasoning"]`` inline.  The two
+    surfaces converge on the same dispatcher; only the mutation shape
+    differs.
+
+    Pure transform.  Safe to call from ``asyncio.to_thread``.
+    """
+    for msg in messages:
+        if msg.get("role") != "assistant":
+            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
+        # 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:
+            continue
+        text = extract_reasoning_text_from_provider_content(provider_content)
+        if text:
+            msg["reasoning"] = text
+
+
 def decorate_history_messages(
     messages: list[dict[str, Any]],
     verdicts_by_call_id: dict[str, dict[str, Any]],
diff --git a/turnstone/core/model_registry.py b/turnstone/core/model_registry.py
index 2a0efacc..64402117 100644
--- a/turnstone/core/model_registry.py
+++ b/turnstone/core/model_registry.py
@@ -39,6 +39,12 @@ class ModelConfig:
     temperature: float | None = None
     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
+    # /history responses; replay_reasoning_to_model controls whether
+    # reasoning blocks ride the wire on subsequent provider calls.
+    persist_reasoning: bool = True
+    replay_reasoning_to_model: bool = False
     # Server compatibility settings for openai-compatible backends.
     # Populated from capabilities["server_compat"] during load.
     server_compat: dict[str, Any] = field(default_factory=dict)
@@ -405,6 +411,10 @@ def load_model_registry(
                 row_temperature = row.get("temperature")
                 row_max_tokens = row.get("max_tokens")
                 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_replay_reasoning = bool(row.get("replay_reasoning_to_model", False))
                 configs[alias] = ModelConfig(
                     alias=alias,
                     base_url=row_base_url,
@@ -419,6 +429,8 @@ def load_model_registry(
                     reasoning_effort=row_reasoning_effort
                     if row_reasoning_effort is not None
                     else None,
+                    persist_reasoning=row_persist_reasoning,
+                    replay_reasoning_to_model=row_replay_reasoning,
                     server_compat=row_server_compat,
                 )
         except Exception:
diff --git a/turnstone/core/providers/_anthropic.py b/turnstone/core/providers/_anthropic.py
index 9ade4d1a..611c5378 100644
--- a/turnstone/core/providers/_anthropic.py
+++ b/turnstone/core/providers/_anthropic.py
@@ -922,6 +922,36 @@ class AnthropicProvider:
             }
         )
 
+    # -- reasoning extraction ------------------------------------------------
+
+    def extract_reasoning_text(
+        self,
+        provider_blocks: list[dict[str, Any]] | None,
+    ) -> str:
+        if not isinstance(provider_blocks, list):
+            return ""
+        parts: list[str] = []
+        for block in provider_blocks:
+            if not isinstance(block, dict):
+                continue
+            if block.get("type") != "thinking":
+                continue
+            text = block.get("thinking")
+            if isinstance(text, str) and text:
+                parts.append(text)
+        if not parts:
+            return ""
+        joined = "\n".join(parts)
+        # Operator-friendly UI cap. Larger reasoning bodies are still
+        # stored verbatim in provider_data; only the rehydrated UI
+        # display payload is truncated.
+        if len(joined) > _MAX_REASONING_DISPLAY_BYTES:
+            return joined[:_MAX_REASONING_DISPLAY_BYTES]
+        return joined
+
+
+_MAX_REASONING_DISPLAY_BYTES = 64 * 1024
+
 
 def _normalize_finish_reason(reason: str) -> str:
     """Normalize Anthropic stop reasons to OpenAI-compatible strings."""
diff --git a/turnstone/core/providers/_openai_chat.py b/turnstone/core/providers/_openai_chat.py
index b6cb0e1c..4cca2a2f 100644
--- a/turnstone/core/providers/_openai_chat.py
+++ b/turnstone/core/providers/_openai_chat.py
@@ -373,3 +373,16 @@ class OpenAIChatCompletionsProvider:
     @property
     def retryable_error_names(self) -> frozenset[str]:
         return RETRYABLE_ERROR_NAMES
+
+    # -- reasoning extraction ------------------------------------------------
+
+    def extract_reasoning_text(
+        self,
+        provider_blocks: list[dict[str, Any]] | None,
+    ) -> str:
+        # OpenAI Chat (and the local-model server flavours that route
+        # through this adapter) have no first-class reasoning shape.
+        # Chat-template ``<think>`` content is captured via the inflight
+        # buffer for live UI but not persisted to ``provider_blocks``;
+        # Phase 4 may revisit.
+        return ""
diff --git a/turnstone/core/providers/_openai_responses.py b/turnstone/core/providers/_openai_responses.py
index 6c5af0a8..66ca1bb5 100644
--- a/turnstone/core/providers/_openai_responses.py
+++ b/turnstone/core/providers/_openai_responses.py
@@ -576,3 +576,15 @@ class OpenAIResponsesProvider:
     @property
     def retryable_error_names(self) -> frozenset[str]:
         return RETRYABLE_ERROR_NAMES
+
+    # -- reasoning extraction ------------------------------------------------
+
+    def extract_reasoning_text(
+        self,
+        provider_blocks: list[dict[str, Any]] | None,
+    ) -> str:
+        # Phase 3 will wire this to walk ``type=="reasoning"`` items
+        # captured via ``include=["reasoning.encrypted_content"]``.
+        # Today the request kwargs don't pass ``include`` so reasoning
+        # items in ``provider_blocks`` carry no replayable text.
+        return ""
diff --git a/turnstone/core/providers/_protocol.py b/turnstone/core/providers/_protocol.py
index 4b95148c..cbb83092 100644
--- a/turnstone/core/providers/_protocol.py
+++ b/turnstone/core/providers/_protocol.py
@@ -177,3 +177,17 @@ class LLMProvider(Protocol):
     def retryable_error_names(self) -> frozenset[str]:
         """Exception class names that should trigger retry."""
         ...
+
+    def extract_reasoning_text(
+        self,
+        provider_blocks: list[dict[str, Any]] | None,
+    ) -> str:
+        """Return concatenated reasoning text from stored ``provider_blocks``.
+
+        Providers without a first-class reasoning shape (OpenAI Chat,
+        Google) or that haven't been wired yet (OpenAI Responses pre-
+        Phase-3) return ``""``.  AnthropicProvider walks
+        ``type=="thinking"`` blocks and returns the concatenated
+        ``thinking`` text, capped at an operator-friendly size.
+        """
+        ...
diff --git a/turnstone/core/session_routes.py b/turnstone/core/session_routes.py
index e22b67ae..7b24cd96 100644
--- a/turnstone/core/session_routes.py
+++ b/turnstone/core/session_routes.py
@@ -2326,7 +2326,8 @@ def make_history_handler(cfg: SessionEndpointConfig) -> Handler:
         # shares storage with the other kind. ``cfg.list_kind`` is
         # guaranteed non-None by the misconfig gate above.
         storage = getattr(request.app.state, "auth_storage", None)
-        if mgr.get(ws_id) is None:
+        live_session = mgr.get(ws_id)
+        if live_session is None:
             if storage is None:
                 return JSONResponse({"error": cfg.not_found_label}, status_code=404)
             try:
@@ -2364,6 +2365,7 @@ def make_history_handler(cfg: SessionEndpointConfig) -> Handler:
             try:
                 from turnstone.core.history_decoration import (
                     decorate_history_messages,
+                    extract_reasoning_for_history,
                     load_verdict_indexes,
                 )
 
@@ -2376,6 +2378,49 @@ 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
+                # resolution so the operator's flag-flip takes effect
+                # uniformly — live session, storage-rehydratable cold
+                # workstream, or unknown workstream:
+                #
+                # 1. Live session in memory → read from its registry
+                #    (already-warm path).
+                # 2. Cold workstream → ``workstream_config.model_alias``
+                #    persisted at first send (see
+                #    ``session_manager.py:628`` rehydrate path) →
+                #    resolve through the kind-appropriate registry on
+                #    ``app.state``.
+                # 3. Neither available → conservative default ``True``,
+                #    matching the migration server_default and the
+                #    rehydration default in spec.
+                persist_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:
+                    try:
+                        ws_cfg = storage.load_workstream_config(ws_id) or {}
+                    except Exception:
+                        ws_cfg = {}
+                    resolved_alias = ws_cfg.get("model_alias") or ""
+                if resolved_registry is None:
+                    # Interactive server stores the registry as
+                    # ``app.state.registry``; console stores its coord
+                    # registry as ``app.state.coord_registry``.  The
+                    # lifted handler is shared, so we try both.
+                    resolved_registry = getattr(request.app.state, "registry", None) or getattr(
+                        request.app.state, "coord_registry", None
+                    )
+                if resolved_registry is not None and resolved_alias:
+                    try:
+                        persist_reasoning = bool(
+                            resolved_registry.get_config(resolved_alias).persist_reasoning
+                        )
+                    except Exception:
+                        persist_reasoning = True
+                await asyncio.to_thread(extract_reasoning_for_history, messages, persist_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 484fbdeb..c2a2c806 100644
--- a/turnstone/core/storage/_postgresql.py
+++ b/turnstone/core/storage/_postgresql.py
@@ -4167,6 +4167,8 @@ class PostgreSQLBackend:
         temperature: float | None = None,
         max_tokens: int | None = None,
         reasoning_effort: str | None = None,
+        persist_reasoning: bool = True,
+        replay_reasoning_to_model: bool = False,
     ) -> None:
         from sqlalchemy.dialects import postgresql
 
@@ -4187,6 +4189,8 @@ class PostgreSQLBackend:
                     temperature=temperature,
                     max_tokens=max_tokens,
                     reasoning_effort=reasoning_effort,
+                    persist_reasoning=1 if persist_reasoning else 0,
+                    replay_reasoning_to_model=1 if replay_reasoning_to_model else 0,
                     created_by=created_by,
                     created=now,
                     updated=now,
@@ -4205,7 +4209,7 @@ class PostgreSQLBackend:
             ).fetchone()
             if row is None:
                 return None
-            return _row_to_dict(row, "enabled")
+            return _row_to_dict(row, "enabled", "persist_reasoning", "replay_reasoning_to_model")
 
     def get_model_definition_by_alias(self, alias: str) -> dict[str, Any] | None:
 
@@ -4215,7 +4219,7 @@ class PostgreSQLBackend:
             ).fetchone()
             if row is None:
                 return None
-            return _row_to_dict(row, "enabled")
+            return _row_to_dict(row, "enabled", "persist_reasoning", "replay_reasoning_to_model")
 
     def list_model_definitions(self, enabled_only: bool = False) -> list[dict[str, Any]]:
 
@@ -4224,7 +4228,10 @@ class PostgreSQLBackend:
             if enabled_only:
                 q = q.where(model_definitions.c.enabled == 1)
             rows = conn.execute(q).fetchall()
-            return [_row_to_dict(r, "enabled") for r in rows]
+            return [
+                _row_to_dict(r, "enabled", "persist_reasoning", "replay_reasoning_to_model")
+                for r in rows
+            ]
 
     def update_model_definition(self, definition_id: str, **fields: Any) -> bool:
 
@@ -4232,6 +4239,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 "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:
             result = conn.execute(
                 sa.update(model_definitions)
diff --git a/turnstone/core/storage/_protocol.py b/turnstone/core/storage/_protocol.py
index ec5504f4..dcc222a4 100644
--- a/turnstone/core/storage/_protocol.py
+++ b/turnstone/core/storage/_protocol.py
@@ -1853,6 +1853,8 @@ class StorageBackend(Protocol):
         temperature: float | None = None,
         max_tokens: int | None = None,
         reasoning_effort: str | None = None,
+        persist_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 aa5afe54..28ad2240 100644
--- a/turnstone/core/storage/_schema.py
+++ b/turnstone/core/storage/_schema.py
@@ -660,6 +660,8 @@ 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("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),
     sa.Column("updated", sa.Text, nullable=False),
diff --git a/turnstone/core/storage/_sqlite.py b/turnstone/core/storage/_sqlite.py
index d04e7e66..8080429f 100644
--- a/turnstone/core/storage/_sqlite.py
+++ b/turnstone/core/storage/_sqlite.py
@@ -4313,6 +4313,8 @@ class SQLiteBackend:
         temperature: float | None = None,
         max_tokens: int | None = None,
         reasoning_effort: str | None = None,
+        persist_reasoning: bool = True,
+        replay_reasoning_to_model: bool = False,
     ) -> None:
 
         now = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S")
@@ -4332,6 +4334,8 @@ class SQLiteBackend:
                     "temperature": temperature,
                     "max_tokens": max_tokens,
                     "reasoning_effort": reasoning_effort,
+                    "persist_reasoning": 1 if persist_reasoning else 0,
+                    "replay_reasoning_to_model": (1 if replay_reasoning_to_model else 0),
                     "created_by": created_by,
                     "created": now,
                     "updated": now,
@@ -4349,7 +4353,7 @@ class SQLiteBackend:
             ).fetchone()
             if row is None:
                 return None
-            return _row_to_dict(row, "enabled")
+            return _row_to_dict(row, "enabled", "persist_reasoning", "replay_reasoning_to_model")
 
     def get_model_definition_by_alias(self, alias: str) -> dict[str, Any] | None:
 
@@ -4359,7 +4363,7 @@ class SQLiteBackend:
             ).fetchone()
             if row is None:
                 return None
-            return _row_to_dict(row, "enabled")
+            return _row_to_dict(row, "enabled", "persist_reasoning", "replay_reasoning_to_model")
 
     def list_model_definitions(self, enabled_only: bool = False) -> list[dict[str, Any]]:
 
@@ -4368,7 +4372,10 @@ class SQLiteBackend:
             if enabled_only:
                 q = q.where(model_definitions.c.enabled == 1)
             rows = conn.execute(q).fetchall()
-            return [_row_to_dict(r, "enabled") for r in rows]
+            return [
+                _row_to_dict(r, "enabled", "persist_reasoning", "replay_reasoning_to_model")
+                for r in rows
+            ]
 
     def update_model_definition(self, definition_id: str, **fields: Any) -> bool:
 
@@ -4376,6 +4383,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 "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:
             result = conn.execute(
                 sa.update(model_definitions)
diff --git a/turnstone/core/storage/_utils.py b/turnstone/core/storage/_utils.py
index 7061f591..0362477e 100644
--- a/turnstone/core/storage/_utils.py
+++ b/turnstone/core/storage/_utils.py
@@ -194,6 +194,8 @@ MODEL_DEFINITION_MUTABLE = frozenset(
         "temperature",
         "max_tokens",
         "reasoning_effort",
+        "persist_reasoning",
+        "replay_reasoning_to_model",
     }
 )
 PROMPT_POLICY_MUTABLE = frozenset({"name", "content", "tool_gate", "priority", "enabled"})
diff --git a/turnstone/core/storage/migrations/versions/052_model_reasoning_persistence.py b/turnstone/core/storage/migrations/versions/052_model_reasoning_persistence.py
new file mode 100644
index 00000000..5faad800
--- /dev/null
+++ b/turnstone/core/storage/migrations/versions/052_model_reasoning_persistence.py
@@ -0,0 +1,63 @@
+"""Add per-model reasoning-persistence flags to model_definitions.
+
+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``.
+* ``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
+  conservative default: spec compliance, lower per-turn cost, no behaviour
+  change vs. the pre-flag default. **Phase 1 stores the column but does
+  not consume it on the wire**; Phase 2 wires the strip branch in
+  ``_anthropic.py``'s ``_convert_messages``.
+
+Mirrors the ``enabled`` column pattern (``_schema.py:659``):
+``NOT NULL`` with an integer ``server_default`` so existing rows pick up
+the conservative defaults silently on upgrade. Distinct from
+``temperature`` / ``max_tokens`` / ``reasoning_effort`` (migration 036)
+which are nullable inherit-from-cluster sampling overrides — these are
+operator-toggle booleans, never NULL.
+
+Revision ID: 052
+Revises: 051
+Create Date: 2026-05-08
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "052"
+down_revision = "051"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    with op.batch_alter_table("model_definitions") as batch:
+        batch.add_column(
+            sa.Column(
+                "persist_reasoning",
+                sa.Integer,
+                nullable=False,
+                server_default="1",
+            )
+        )
+        batch.add_column(
+            sa.Column(
+                "replay_reasoning_to_model",
+                sa.Integer,
+                nullable=False,
+                server_default="0",
+            )
+        )
+
+
+def downgrade() -> None:
+    with op.batch_alter_table("model_definitions") as batch:
+        batch.drop_column("replay_reasoning_to_model")
+        batch.drop_column("persist_reasoning")
diff --git a/turnstone/sdk/events.py b/turnstone/sdk/events.py
index bf76723f..657da6b3 100644
--- a/turnstone/sdk/events.py
+++ b/turnstone/sdk/events.py
@@ -49,6 +49,26 @@ class ConnectedEvent(ServerEvent):
 
 @dataclass
 class HistoryEvent(ServerEvent):
+    """SSE replay payload — the per-tab message backlog on connect.
+
+    Each entry in ``messages`` is a per-message dict the frontend
+    consumes directly. Notable optional keys:
+
+    * ``role`` (``"user"`` / ``"assistant"`` / ``"tool"``)
+    * ``content`` — string for text turns, list for image / document parts
+    * ``tool_calls`` — list of ``{id, name, arguments, verdict?,
+      output_assessment?}`` (assistant turns)
+    * ``tool_call_id`` — the originating call's id (tool turns)
+    * ``reminders`` — metacognitive nudge bubbles (user / tool channels)
+    * ``advisories`` — extracted ``UserInterjection`` payloads (tool
+      turns whose envelope wrapped queued-message advisories)
+    * ``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
+      the underlying ``provider_data`` carries reasoning blocks.
+    """
+
     type: str = "history"
     messages: list[dict[str, Any]] = field(default_factory=list)
 
diff --git a/turnstone/server.py b/turnstone/server.py
index 93dffa90..99406421 100644
--- a/turnstone/server.py
+++ b/turnstone/server.py
@@ -58,6 +58,9 @@ from turnstone.core.history_decoration import (
 from turnstone.core.history_decoration import (
     extract_advisories_from_tool_envelope,
 )
+from turnstone.core.history_decoration import (
+    extract_reasoning_text_from_provider_content as _extract_reasoning_text,
+)
 from turnstone.core.history_decoration import (
     load_verdict_indexes as _load_verdict_indexes,
 )
@@ -471,6 +474,22 @@ def _build_history(
     else:
         ws_id = getattr(session, "_ws_id", "") or ""
         verdicts_by_call_id, assessments_by_call_id = _load_verdict_indexes(ws_id)
+    # 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
+    # rehydration default (Phase 1 spec).
+    persist_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)
+        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
     history = []
     for msg in session.messages:
         content = msg.get("content")
@@ -556,6 +575,15 @@ def _build_history(
                 clean_reminders.append(clean)
             if clean_reminders:
                 entry["reminders"] = clean_reminders
+        # Surface stored reasoning text on assistant messages for UI
+        # rehydration (page-refresh path).  Sourced from the in-memory
+        # ``_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:
+            reasoning_text = _extract_reasoning_text(msg.get("_provider_content"))
+            if reasoning_text:
+                entry["reasoning"] = reasoning_text
         if msg.get("tool_calls"):
             tc_entries: list[dict[str, Any]] = []
             for tc in msg["tool_calls"]:
diff --git a/turnstone/ui/static/app.js b/turnstone/ui/static/app.js
index e02cdce4..7b7881a4 100644
--- a/turnstone/ui/static/app.js
+++ b/turnstone/ui/static/app.js
@@ -1194,6 +1194,20 @@ Pane.prototype.replayHistory = function (messages) {
       }
       lastToolBlock = null;
     } else if (msg.role === "assistant") {
+      // Reasoning bubble (Phase 1 reasoning persistence) — render
+      // BEFORE the content bubble so the visual order matches the
+      // 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
+      // true and the message round-tripped a thinking lane.
+      if (msg.reasoning && msg.reasoning.length) {
+        var reasonEl = document.createElement("div");
+        reasonEl.className = "msg reasoning";
+        reasonEl.textContent = msg.reasoning;
+        self.messagesEl.appendChild(reasonEl);
+        lastToolBlock = null;
+      }
       // Render content BEFORE the tool block so the visual order
       // matches the live SSE flow (stream_text streams content first,
       // then tool_info / approve_request paints the tool block, then