diff --git a/agent/agent_runtime_helpers.py b/agent/agent_runtime_helpers.py
index 072fe507f3323..5d2c17300bd89 100644
--- a/agent/agent_runtime_helpers.py
+++ b/agent/agent_runtime_helpers.py
@@ -1678,16 +1678,33 @@ def extract_reasoning(agent, assistant_message) -> Optional[str]:
         Combined reasoning text, or None if no reasoning found
     """
     reasoning_parts = []
+    reasoning_seen: set[str] = set()
+
+    def _append_reasoning_part(value) -> None:
+        pending = [value]
+        while pending:
+            item = pending.pop()
+            if isinstance(item, str):
+                if item and item not in reasoning_seen:
+                    reasoning_seen.add(item)
+                    reasoning_parts.append(item)
+                continue
+            if isinstance(item, list):
+                pending.extend(reversed(item))
+                continue
+            if isinstance(item, dict):
+                pending.extend(
+                    item.get(key)
+                    for key in reversed(("summary", "thinking", "content", "text"))
+                )
     
     # Check direct reasoning field
     if hasattr(assistant_message, 'reasoning') and assistant_message.reasoning:
-        reasoning_parts.append(assistant_message.reasoning)
+        _append_reasoning_part(assistant_message.reasoning)
     
     # Check reasoning_content field (alternative name used by some providers)
     if hasattr(assistant_message, 'reasoning_content') and assistant_message.reasoning_content:
-        # Don't duplicate if same as reasoning
-        if assistant_message.reasoning_content not in reasoning_parts:
-            reasoning_parts.append(assistant_message.reasoning_content)
+        _append_reasoning_part(assistant_message.reasoning_content)
     
     # Check reasoning_details array (OpenRouter unified format)
     # Format: [{"type": "reasoning.summary", "summary": "...", ...}, ...]
@@ -1695,14 +1712,12 @@ def extract_reasoning(agent, assistant_message) -> Optional[str]:
         for detail in assistant_message.reasoning_details:
             if isinstance(detail, dict):
                 # Extract summary from reasoning detail object
-                summary = (
+                _append_reasoning_part(
                     detail.get('summary')
                     or detail.get('thinking')
                     or detail.get('content')
                     or detail.get('text')
                 )
-                if summary and summary not in reasoning_parts:
-                    reasoning_parts.append(summary)
 
     # Some providers embed reasoning directly inside assistant content
     # instead of returning structured reasoning fields.  Only fall back
@@ -1719,8 +1734,7 @@ def extract_reasoning(agent, assistant_message) -> Optional[str]:
             if isinstance(block, dict) and block.get("type") == "thinking":
                 thinking_text = block.get("thinking") or block.get("text") or ""
                 thinking_text = thinking_text.strip()
-                if thinking_text and thinking_text not in reasoning_parts:
-                    reasoning_parts.append(thinking_text)
+                _append_reasoning_part(thinking_text)
     if not reasoning_parts and isinstance(content, str) and content:
         inline_patterns = (
             r"<think>(.*?)</think>",
@@ -1733,8 +1747,7 @@ def extract_reasoning(agent, assistant_message) -> Optional[str]:
             flags = re.DOTALL | re.IGNORECASE
             for block in re.findall(pattern, content, flags=flags):
                 cleaned = block.strip()
-                if cleaned and cleaned not in reasoning_parts:
-                    reasoning_parts.append(cleaned)
+                _append_reasoning_part(cleaned)
     
     # Combine all reasoning parts
     if reasoning_parts:
diff --git a/agent/bedrock_adapter.py b/agent/bedrock_adapter.py
index c399081619ffa..e6e84196e51dc 100644
--- a/agent/bedrock_adapter.py
+++ b/agent/bedrock_adapter.py
@@ -784,7 +784,11 @@ def normalize_converse_response(response: Dict) -> SimpleNamespace:
         role="assistant",
         content="\n".join(text_parts) if text_parts else None,
         tool_calls=tool_calls if tool_calls else None,
-        reasoning_content="\n\n".join(reasoning_parts) if reasoning_parts else None,
+        # Bedrock exposes a mutable text projection of reasoningContent. The
+        # adapter does not retain provider-owned replay material here, so this
+        # must not be labeled as an exact reasoning_content echo.
+        reasoning="\n\n".join(reasoning_parts) if reasoning_parts else None,
+        reasoning_content=None,
     )
 
     # Build usage stats. Converse's inputTokens excludes cache read/write
@@ -979,7 +983,8 @@ def stream_converse_with_callbacks(
         role="assistant",
         content="\n".join(text_parts) if text_parts else None,
         tool_calls=tool_calls if tool_calls else None,
-        reasoning_content="\n\n".join(reasoning_parts) if reasoning_parts else None,
+        reasoning="\n\n".join(reasoning_parts) if reasoning_parts else None,
+        reasoning_content=None,
     )
 
     input_tokens = usage_data.get("inputTokens", 0)
diff --git a/agent/chat_completion_helpers.py b/agent/chat_completion_helpers.py
index e40333f0e601b..4dc09d8a57fa6 100644
--- a/agent/chat_completion_helpers.py
+++ b/agent/chat_completion_helpers.py
@@ -1387,7 +1387,6 @@ def build_api_kwargs(agent, api_messages: list, tools_for_api: list | None = Non
     )
 
 
-
 def build_assistant_message(agent, assistant_message, finish_reason: str) -> dict:
     """Build a normalized assistant message dict from an API response message.
 
@@ -1408,6 +1407,13 @@ def build_assistant_message(agent, assistant_message, finish_reason: str) -> dic
             combined = "\n\n".join(b.strip() for b in think_blocks if b.strip())
             reasoning_text = combined or None
 
+    # Mutable reasoning is safe to mask before it reaches non-streaming
+    # callbacks or persistence. Provider-native replay carriers are handled
+    # separately below and remain unchanged.
+    if isinstance(reasoning_text, str) and reasoning_text:
+        from agent.redact import redact_sensitive_text
+        reasoning_text = redact_sensitive_text(reasoning_text)
+
     if reasoning_text and agent.verbose_logging:
         logging.debug(f"Captured reasoning ({len(reasoning_text)} chars): {reasoning_text}")
 
@@ -1482,6 +1488,8 @@ def build_assistant_message(agent, assistant_message, finish_reason: str) -> dic
         if isinstance(model_extra, dict) and "reasoning_content" in model_extra:
             raw_reasoning_content = model_extra["reasoning_content"]
     if raw_reasoning_content is not None:
+        # DeepSeek, Kimi, and MiMo may require this exact provider-owned value
+        # on the next turn, so only retain the existing surrogate cleanup.
         msg["reasoning_content"] = _sanitize_surrogates(raw_reasoning_content)
     elif assistant_tool_calls and agent._needs_thinking_reasoning_pad():
         # DeepSeek v4 thinking mode and Kimi / Moonshot thinking mode
@@ -1525,8 +1533,10 @@ def build_assistant_message(agent, assistant_message, finish_reason: str) -> dic
     if hasattr(assistant_message, 'reasoning_details') and assistant_message.reasoning_details:
         # Pass reasoning_details back unmodified so providers (OpenRouter,
         # Anthropic, OpenAI) can maintain reasoning continuity across turns.
-        # Each provider may include opaque fields (signature, encrypted_content)
-        # that must be preserved exactly.
+        # Even an unsigned text/summary block is provider-owned replay state;
+        # changing it can invalidate a later request. Opaque fields such as
+        # signatures and encrypted_content make this requirement obvious, but
+        # are not the only shapes that require exact preservation.
         raw_details = assistant_message.reasoning_details
         preserved = []
         for d in raw_details:
@@ -2446,7 +2456,13 @@ def cleanup_task_resources(agent, task_id: str) -> None:
 
 
 def _build_partial_stream_stub(
-    role, full_content, full_reasoning, model_name, usage_obj, *,
+    role,
+    full_content,
+    full_reasoning,
+    model_name,
+    usage_obj,
+    *,
+    full_reasoning_content=None,
     dropped_tool_names=None,
 ):
     """Build a partial-stream-stub response for mid-stream drop scenarios.
@@ -2461,7 +2477,8 @@ def _build_partial_stream_stub(
         role=role,
         content=full_content,
         tool_calls=None,
-        reasoning_content=full_reasoning,
+        reasoning=full_reasoning,
+        reasoning_content=full_reasoning_content,
     )
     mock_choice = SimpleNamespace(
         index=0,
@@ -3024,6 +3041,7 @@ def _call_chat_completions(stream_attempt_id: int):
         model_name = None
         role = "assistant"
         reasoning_parts: list = []
+        reasoning_content_parts: list = []
         usage_obj = None
         _diag = agent._stream_diag_init()
         request_client_holder["diag"] = _diag
@@ -3103,7 +3121,8 @@ def _relay_final_response() -> dict[str, Any]:
                         "message": {
                             "role": role,
                             "content": "".join(content_parts) or None,
-                            "reasoning_content": "".join(reasoning_parts) or None,
+                            "reasoning": "".join(reasoning_parts) or None,
+                            "reasoning_content": "".join(reasoning_content_parts) or None,
                             "tool_calls": tool_calls or None,
                         },
                         "finish_reason": finish_reason or "stop",
@@ -3206,9 +3225,14 @@ def _relay_final_response() -> dict[str, Any]:
                 model_name = chunk.model
 
             # Accumulate reasoning content
-            reasoning_text = getattr(delta, "reasoning_content", None) or getattr(delta, "reasoning", None)
+            provider_reasoning_text = getattr(delta, "reasoning_content", None)
+            mutable_reasoning_text = getattr(delta, "reasoning", None)
+            reasoning_text = provider_reasoning_text or mutable_reasoning_text
             if reasoning_text:
-                reasoning_parts.append(reasoning_text)
+                if provider_reasoning_text:
+                    reasoning_content_parts.append(provider_reasoning_text)
+                if mutable_reasoning_text:
+                    reasoning_parts.append(mutable_reasoning_text)
                 _fire_first_delta()
                 agent._fire_reasoning_delta(reasoning_text)
 
@@ -3403,6 +3427,7 @@ def _relay_final_response() -> dict[str, Any]:
             finish_reason is None
             and not content_parts
             and not reasoning_parts
+            and not reasoning_content_parts
             and not tool_calls_acc
         ):
             raise EmptyStreamError(
@@ -3446,6 +3471,7 @@ def _relay_final_response() -> dict[str, Any]:
                 role, full_content,
                 "".join(reasoning_parts) or None,
                 model_name, usage_obj,
+                full_reasoning_content="".join(reasoning_content_parts) or None,
                 dropped_tool_names=_dropped_names or None,
             )
 
@@ -3468,6 +3494,7 @@ def _relay_final_response() -> dict[str, Any]:
                 role, full_content,
                 "".join(reasoning_parts) or None,
                 model_name, usage_obj,
+                full_reasoning_content="".join(reasoning_content_parts) or None,
             )
 
         effective_finish_reason = finish_reason or "stop"
@@ -3475,11 +3502,13 @@ def _relay_final_response() -> dict[str, Any]:
             effective_finish_reason = "length"
 
         full_reasoning = "".join(reasoning_parts) or None
+        full_reasoning_content = "".join(reasoning_content_parts) or None
         mock_message = SimpleNamespace(
             role=role,
             content=full_content,
             tool_calls=mock_tool_calls,
-            reasoning_content=full_reasoning,
+            reasoning=full_reasoning,
+            reasoning_content=full_reasoning_content,
         )
         mock_choice = SimpleNamespace(
             index=0,
diff --git a/agent/copilot_acp_client.py b/agent/copilot_acp_client.py
index 9cbdcd3494466..64fcd3bb99b63 100644
--- a/agent/copilot_acp_client.py
+++ b/agent/copilot_acp_client.py
@@ -487,7 +487,10 @@ def _create_chat_completion(
             content=cleaned_text,
             tool_calls=tool_calls,
             reasoning=reasoning_text or None,
-            reasoning_content=reasoning_text or None,
+            # ACP exposes a mutable reasoning transcript, not provider-owned
+            # replay state. Keep it out of reasoning_content so persistence
+            # masking cannot be bypassed by a synthetic duplicate.
+            reasoning_content=None,
             reasoning_details=None,
         )
         finish_reason = "tool_calls" if tool_calls else "stop"
diff --git a/agent/gemini_native_adapter.py b/agent/gemini_native_adapter.py
index cde63f15fc179..a07795a4cf48a 100644
--- a/agent/gemini_native_adapter.py
+++ b/agent/gemini_native_adapter.py
@@ -626,7 +626,10 @@ def translate_gemini_response(resp: Dict[str, Any], model: str) -> SimpleNamespa
         content="".join(text_pieces) if text_pieces else None,
         tool_calls=tool_calls or None,
         reasoning=reasoning,
-        reasoning_content=reasoning,
+        # Native Gemini thought text is an ordinary, mutable copy. Replay
+        # continuity is carried by thought_signature on tool calls, not by an
+        # OpenAI-compatible reasoning_content echo.
+        reasoning_content=None,
         reasoning_details=None,
     )
     choice = SimpleNamespace(index=0, message=message, finish_reason=finish_reason)
@@ -677,7 +680,6 @@ def _make_stream_chunk(
         delta_kwargs["tool_calls"] = [tool_delta]
     if reasoning:
         delta_kwargs["reasoning"] = reasoning
-        delta_kwargs["reasoning_content"] = reasoning
     delta = SimpleNamespace(**delta_kwargs)
     choice = SimpleNamespace(index=0, delta=delta, finish_reason=finish_reason)
     return _GeminiStreamChunk(
diff --git a/agent/redact.py b/agent/redact.py
index ea70246a90797..f4d47e5aff7eb 100644
--- a/agent/redact.py
+++ b/agent/redact.py
@@ -329,19 +329,39 @@ def _key_has_secret_keyword(key: str) -> bool:
     r"-----BEGIN[A-Z ]*PRIVATE KEY-----[\s\S]*?-----END[A-Z ]*PRIVATE KEY-----"
 )
 
-# Database connection strings: protocol://user:PASSWORD@host
-# Catches postgres, mysql, mongodb, redis, amqp URLs and redacts the password.
-# The userinfo and password groups forbid whitespace ([^:\s]+ / [^@\s]+) so the
-# match can never span a line break. A real DSN password never contains
-# whitespace; without this bound the greedy [^@]+ would scan past the end of a
-# code line to the next stray "@" (e.g. a Python decorator), swallowing
-# intervening lines and corrupting tool OUTPUT for any source containing a
-# postgresql:// f-string template. See issue #33801.
+# Database connection strings: dialect[+driver]://[user]:PASSWORD@host
+# Catches common SQL, document, cache, and queue database URI schemes and
+# redacts the password. The optional driver segment covers SQLAlchemy-style
+# schemes such as postgresql+psycopg, snowflake+connector, and db2+ibm_db.
+# The username may be empty for URI forms such as redis://:password@host.
+#
+# The username and password groups stop at quotes and structured-text
+# delimiters as well as URI authority delimiters. That prevents a passwordless
+# host:port string in one JSON/YAML value from scanning into a later email
+# address. URI punctuation such as commas, semicolons, and parentheses remains
+# valid inside a password; quote/bracket characters must be percent-encoded.
+#
+# A permissive password group used to scan past the end of a code line to the
+# next stray "@" (e.g. a Python decorator), swallowing intervening lines and
+# corrupting tool output for source containing a postgresql:// f-string
+# template. The leading guard also keeps a database name from matching inside
+# a longer custom scheme. See issues #33801 and #43666.
 _DB_CONNSTR_RE = re.compile(
-    r"((?:postgres(?:ql)?|mysql|mongodb(?:\+srv)?|redis|amqp)://[^:\s]+:)([^@\s]+)(@)",
+    r"(?<![a-z0-9+.-])"
+    r"((?:postgres(?:ql)?|mysqlx?|mariadb|mongodb|mssql|oracle|clickhouse|"
+    r"cockroachdb|snowflake|trino|db2|rediss?|amqps?)"
+    r"(?:\+[a-z0-9_.-]+)?://[^:/?#\s@,;'\"<>\[\]{}()\\|]*:)"
+    r"([^@/?#\s'\"<>\[\]{}\\|]+)(@)",
     re.IGNORECASE,
 )
 
+
+def _redact_db_connstr_match(match: re.Match, *, code_file: bool) -> str:
+    password = match.group(2)
+    if code_file and password.startswith("{") and password.endswith("}"):
+        return match.group(0)
+    return f"{match.group(1)}***{match.group(3)}"
+
 # Bare-token credential in a web/transport URL: ``scheme://TOKEN@host``.
 # This is the ``git remote set-url origin https://PASSWORD@github.com/...``
 # shape from issue #6396 — a single opaque credential in the userinfo position
@@ -822,15 +842,13 @@ def _redact_telegram(m):
     # forbids whitespace in the password group, so a single-line template's
     # group(2) is exactly the brace expression. See issue #33801.
     if "://" in text:
-        if code_file:
-            def _redact_db(m):
-                pw = m.group(2)
-                if pw.startswith("{") and pw.endswith("}"):
-                    return m.group(0)
-                return f"{m.group(1)}***{m.group(3)}"
-            text = _DB_CONNSTR_RE.sub(_redact_db, text)
-        else:
-            text = _DB_CONNSTR_RE.sub(lambda m: f"{m.group(1)}***{m.group(3)}", text)
+        text = _DB_CONNSTR_RE.sub(
+            lambda match: _redact_db_connstr_match(
+                match,
+                code_file=code_file,
+            ),
+            text,
+        )
 
         # Bare-token userinfo in web/transport URLs: ``scheme://TOKEN@host``.
         # The git-remote-with-embedded-password shape from #6396. Only the
diff --git a/tests/agent/test_bedrock_adapter.py b/tests/agent/test_bedrock_adapter.py
index 8994688e0f211..f098695dfba89 100644
--- a/tests/agent/test_bedrock_adapter.py
+++ b/tests/agent/test_bedrock_adapter.py
@@ -244,6 +244,39 @@ def test_text_response(self):
         assert result.usage.completion_tokens == 5
         assert result.usage.total_tokens == 15
 
+    def test_reasoning_is_not_mislabeled_as_provider_replay_content(self):
+        from agent.bedrock_adapter import normalize_converse_response
+        from agent.moa_loop import _completed_response_as_stream_chunk
+
+        result = normalize_converse_response(
+            {
+                "output": {
+                    "message": {
+                        "role": "assistant",
+                        "content": [
+                            {
+                                "reasoningContent": {
+                                    "text": "ordinary Bedrock reasoning"
+                                }
+                            },
+                            {"text": "done"},
+                        ],
+                    }
+                },
+                "stopReason": "end_turn",
+            }
+        )
+
+        message = result.choices[0].message
+        assert message.reasoning == "ordinary Bedrock reasoning"
+        assert message.reasoning_content is None
+
+        # MoA wraps a completed aggregator response as a stream chunk. Keep
+        # the provenance marker intact across that otherwise separate path.
+        delta = _completed_response_as_stream_chunk(result).choices[0].delta
+        assert delta.reasoning == "ordinary Bedrock reasoning"
+        assert delta.reasoning_content is None
+
     def test_cache_tokens_folded_into_prompt_tokens(self):
         """Converse's inputTokens excludes cache read/write tokens (unlike
         OpenAI's prompt_tokens). normalize_converse_response must add them
@@ -326,6 +359,31 @@ def test_text_stream(self):
         assert result.usage.prompt_tokens == 5
         assert result.usage.completion_tokens == 3
 
+    def test_reasoning_stream_uses_mutable_reasoning_field(self):
+        from agent.bedrock_adapter import normalize_converse_stream_events
+
+        result = normalize_converse_stream_events(
+            {
+                "stream": [
+                    {
+                        "contentBlockDelta": {
+                            "contentBlockIndex": 0,
+                            "delta": {
+                                "reasoningContent": {
+                                    "text": "ordinary streamed reasoning"
+                                }
+                            },
+                        }
+                    },
+                    {"messageStop": {"stopReason": "end_turn"}},
+                ]
+            }
+        )
+
+        message = result.choices[0].message
+        assert message.reasoning == "ordinary streamed reasoning"
+        assert message.reasoning_content is None
+
     def test_tool_use_stream(self):
         from agent.bedrock_adapter import normalize_converse_stream_events
         events = {"stream": [
diff --git a/tests/agent/test_copilot_acp_client.py b/tests/agent/test_copilot_acp_client.py
index a6b366c9c95cc..693bae62ff7c5 100644
--- a/tests/agent/test_copilot_acp_client.py
+++ b/tests/agent/test_copilot_acp_client.py
@@ -54,6 +54,30 @@ def test_stream_true_preserves_tool_call_deltas(self) -> None:
         )
         self.assertEqual(chunks[1].choices, [])
 
+    def test_reasoning_is_not_mislabeled_as_provider_replay_content(self) -> None:
+        with patch.object(
+            self.client,
+            "_run_prompt",
+            return_value=("answer", "ordinary ACP reasoning"),
+        ):
+            completion = self.client._create_chat_completion(
+                model="copilot-acp",
+                messages=[{"role": "user", "content": "question"}],
+                stream=False,
+            )
+            stream = self.client._create_chat_completion(
+                model="copilot-acp",
+                messages=[{"role": "user", "content": "question"}],
+                stream=True,
+            )
+
+        message = completion.choices[0].message
+        self.assertEqual(message.reasoning, "ordinary ACP reasoning")
+        self.assertIsNone(message.reasoning_content)
+        delta = list(stream)[0].choices[0].delta
+        self.assertEqual(delta.reasoning, "ordinary ACP reasoning")
+        self.assertIsNone(delta.reasoning_content)
+
 
     def _dispatch(self, message: dict, *, cwd: str) -> dict:
         process = _FakeProcess()
diff --git a/tests/agent/test_gemini_native_adapter.py b/tests/agent/test_gemini_native_adapter.py
index 82936d67c49e7..c5c60672e67c6 100644
--- a/tests/agent/test_gemini_native_adapter.py
+++ b/tests/agent/test_gemini_native_adapter.py
@@ -109,6 +109,7 @@ def test_translate_native_response_surfaces_reasoning_and_tool_calls():
     choice = response.choices[0]
     assert choice.finish_reason == "tool_calls"
     assert choice.message.reasoning == "thinking..."
+    assert choice.message.reasoning_content is None
     assert choice.message.tool_calls[0].function.name == "search"
     assert json.loads(choice.message.tool_calls[0].function.arguments) == {"q": "hermes"}
 
@@ -240,6 +241,33 @@ def test_stream_event_translation_emits_tool_call_delta_with_stable_index():
     assert first[-1].choices[0].finish_reason == "tool_calls"
 
 
+def test_stream_event_keeps_native_thought_out_of_reasoning_content():
+    from agent.gemini_native_adapter import translate_stream_event
+
+    chunks = translate_stream_event(
+        {
+            "candidates": [
+                {
+                    "content": {
+                        "parts": [{"thought": True, "text": "thinking..."}]
+                    },
+                    "finishReason": "STOP",
+                }
+            ]
+        },
+        model="gemini-2.5-flash",
+        tool_call_indices={},
+    )
+
+    thought_delta = next(
+        chunk.choices[0].delta
+        for chunk in chunks
+        if chunk.choices and chunk.choices[0].delta.reasoning
+    )
+    assert thought_delta.reasoning == "thinking..."
+    assert thought_delta.reasoning_content is None
+
+
 
 
 
@@ -258,4 +286,3 @@ def test_stream_event_translation_emits_tool_call_delta_with_stable_index():
 
 
 
-
diff --git a/tests/agent/test_persistence_redaction.py b/tests/agent/test_persistence_redaction.py
new file mode 100644
index 0000000000000..e19be71a5f86c
--- /dev/null
+++ b/tests/agent/test_persistence_redaction.py
@@ -0,0 +1,467 @@
+"""Persistence redaction boundaries for mutable reasoning and tool output."""
+
+import copy
+from datetime import datetime
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+import pytest
+
+
+SECRET_PASSWORD = "fake-db-password-43666"
+SECRET_URI = (
+    "postgresql+psycopg://postgres:"
+    f"{SECRET_PASSWORD}@127.0.0.1:5432/postgres"
+)
+
+
+@pytest.fixture(autouse=True)
+def _redaction_enabled(monkeypatch):
+    monkeypatch.setattr("agent.redact._REDACT_ENABLED", True)
+
+
+def _make_agent():
+    """Build the smallest test double that runs the real response builder."""
+    from run_agent import AIAgent
+
+    agent = MagicMock(spec=AIAgent)
+    agent._build_assistant_message = AIAgent._build_assistant_message.__get__(agent)
+    agent._extract_reasoning = AIAgent._extract_reasoning.__get__(agent)
+    agent._strip_think_blocks = AIAgent._strip_think_blocks.__get__(agent)
+    agent.verbose_logging = False
+    agent.reasoning_callback = None
+    agent.stream_delta_callback = None
+    agent._stream_callback = None
+    agent._needs_thinking_reasoning_pad.return_value = False
+    agent._split_responses_tool_id.return_value = (None, None)
+    agent._derive_responses_function_call_id.side_effect = (
+        lambda call_id, response_id: response_id or call_id
+    )
+    return agent
+
+
+def _api_message(content="done", **fields):
+    message = SimpleNamespace(
+        content=content,
+        tool_calls=fields.pop("tool_calls", None),
+    )
+    for key, value in fields.items():
+        setattr(message, key, value)
+    return message
+
+
+def _persist_message(tmp_path, message):
+    from hermes_state import SessionDB
+
+    db = SessionDB(db_path=tmp_path / "state.db")
+    session_id = db.create_session("redaction-test", "cli")
+    db.append_message(session_id=session_id, **message)
+    replayed = db.get_messages_as_conversation(session_id)
+    db.close()
+    return replayed
+
+
+def _make_real_persistence_agent(tmp_path):
+    """Bind the production persistence funnel to a real temporary SessionDB."""
+    from hermes_state import SessionDB
+    from run_agent import AIAgent
+
+    db = SessionDB(db_path=tmp_path / "state.db")
+    session_id = db.create_session("redaction-production-path", "cli")
+    agent = MagicMock(spec=AIAgent)
+    for method_name in (
+        "_persist_session",
+        "_drop_trailing_empty_response_scaffolding",
+        "_save_session_log",
+        "_flush_messages_to_session_db",
+        "_flush_messages_to_session_db_unlocked",
+    ):
+        setattr(agent, method_name, getattr(AIAgent, method_name).__get__(agent))
+    agent._clean_session_content = AIAgent._clean_session_content
+    agent._redact_message_content = AIAgent._redact_message_content
+
+    agent._session_db = db
+    agent._session_db_created = True
+    agent._session_persist_lock = None
+    agent._persist_disabled = False
+    agent._last_flushed_db_idx = 0
+    agent._flushed_db_message_ids = set()
+    agent._flushed_db_message_session_id = None
+    agent._db_flush_scan_prefix = None
+    agent._persist_user_message_idx = None
+    agent._persist_user_message_override = None
+    agent._persist_user_message_timestamp = None
+    agent._pending_cli_user_message = None
+    agent._active_compression_lock_holder = None
+    agent._inflight_turn_id = None
+    agent._inflight_turn_session_id = None
+
+    agent._session_json_enabled = True
+    agent.logs_dir = tmp_path / "sessions"
+    agent.logs_dir.mkdir()
+    agent.session_id = session_id
+    agent.model = "test/model"
+    agent.base_url = "https://openrouter.ai/api/v1"
+    agent.platform = "cli"
+    agent.session_start = datetime.now()
+    agent._cached_system_prompt = "test system prompt"
+    agent.tools = []
+    agent.verbose_logging = False
+    return agent, db
+
+
+def test_mutable_reasoning_is_redacted_before_callback_and_persistence(tmp_path):
+    agent = _make_agent()
+    agent.reasoning_callback = MagicMock()
+
+    built = agent._build_assistant_message(
+        _api_message(
+            reasoning=f"I connected with {SECRET_URI}",
+        ),
+        "stop",
+    )
+
+    callback_text = agent.reasoning_callback.call_args.args[0]
+    assert SECRET_PASSWORD not in callback_text
+    assert SECRET_PASSWORD not in built["reasoning"]
+    assert SECRET_PASSWORD not in built["reasoning_content"]
+
+    replayed = _persist_message(tmp_path, built)
+    assistant = next(item for item in replayed if item["role"] == "assistant")
+    assert assistant["reasoning"] == built["reasoning"]
+    assert assistant["reasoning_content"] == built["reasoning_content"]
+
+    password_bytes = SECRET_PASSWORD.encode()
+    for path in tmp_path.rglob("*"):
+        if path.is_file():
+            assert password_bytes not in path.read_bytes(), path.name
+
+
+def test_provider_reasoning_content_is_preserved_for_replay(tmp_path):
+    provider_reasoning = f"provider-owned bytes: {SECRET_URI}"
+    built = _make_agent()._build_assistant_message(
+        _api_message(
+            reasoning=f"mutable copy: {SECRET_URI}",
+            reasoning_content=provider_reasoning,
+        ),
+        "stop",
+    )
+
+    assert SECRET_PASSWORD not in built["reasoning"]
+    assert built["reasoning_content"] == provider_reasoning
+
+    replayed = _persist_message(tmp_path, built)
+    assistant = next(item for item in replayed if item["role"] == "assistant")
+    assert assistant["reasoning_content"] == provider_reasoning
+
+
+def test_mutable_reasoning_stays_redacted_through_production_persistence(tmp_path):
+    built = _make_agent()._build_assistant_message(
+        _api_message(reasoning=f"production path: {SECRET_URI}"),
+        "stop",
+    )
+    persistence_agent, db = _make_real_persistence_agent(tmp_path)
+
+    persistence_agent._persist_session([built], conversation_history=[])
+    replayed = db.get_messages_as_conversation(persistence_agent.session_id)
+    db.close()
+
+    assistant = next(item for item in replayed if item["role"] == "assistant")
+    assert SECRET_PASSWORD not in assistant["reasoning"]
+    assert SECRET_PASSWORD not in assistant["reasoning_content"]
+
+    snapshot = (
+        persistence_agent.logs_dir
+        / f"session_{persistence_agent.session_id}.json"
+    )
+    assert snapshot.exists()
+    assert SECRET_PASSWORD not in snapshot.read_text(encoding="utf-8")
+
+    password_bytes = SECRET_PASSWORD.encode()
+    for path in tmp_path.rglob("*"):
+        if path.is_file():
+            assert password_bytes not in path.read_bytes(), path.name
+
+
+def test_native_gemini_reasoning_cannot_bypass_persistence_redaction(tmp_path):
+    from agent.gemini_native_adapter import translate_gemini_response
+    from agent.transports import get_transport
+
+    response = translate_gemini_response(
+        {
+            "candidates": [
+                {
+                    "content": {
+                        "parts": [
+                            {
+                                "thought": True,
+                                "text": f"native Gemini thought: {SECRET_URI}",
+                            },
+                            {"text": "done"},
+                        ]
+                    },
+                    "finishReason": "STOP",
+                }
+            ]
+        },
+        model="gemini-2.5-flash",
+    )
+    normalized = get_transport("chat_completions").normalize_response(response)
+    built = _make_agent()._build_assistant_message(normalized, "stop")
+
+    assert SECRET_PASSWORD not in built["reasoning"]
+    assert SECRET_PASSWORD not in built["reasoning_content"]
+
+    persistence_agent, db = _make_real_persistence_agent(tmp_path)
+    persistence_agent._persist_session([built], conversation_history=[])
+    db.close()
+
+    password_bytes = SECRET_PASSWORD.encode()
+    for path in tmp_path.rglob("*"):
+        if path.is_file():
+            assert password_bytes not in path.read_bytes(), path.name
+
+
+def test_bedrock_reasoning_cannot_bypass_persistence_redaction(tmp_path):
+    from agent.bedrock_adapter import normalize_converse_response
+    from agent.transports import get_transport
+
+    response = normalize_converse_response(
+        {
+            "output": {
+                "message": {
+                    "role": "assistant",
+                    "content": [
+                        {
+                            "reasoningContent": {
+                                "text": f"Bedrock thought: {SECRET_URI}"
+                            }
+                        },
+                        {"text": "done"},
+                    ],
+                }
+            },
+            "stopReason": "end_turn",
+        }
+    )
+    normalized = get_transport("bedrock_converse").normalize_response(response)
+    built = _make_agent()._build_assistant_message(normalized, "stop")
+
+    assert SECRET_PASSWORD not in built["reasoning"]
+    assert SECRET_PASSWORD not in built["reasoning_content"]
+
+    persistence_agent, db = _make_real_persistence_agent(tmp_path)
+    persistence_agent._persist_session([built], conversation_history=[])
+    db.close()
+
+    password_bytes = SECRET_PASSWORD.encode()
+    for path in tmp_path.rglob("*"):
+        if path.is_file():
+            assert password_bytes not in path.read_bytes(), path.name
+
+
+def test_provider_reasoning_details_are_preserved_for_next_turn(tmp_path):
+    details = [
+        {
+            "type": "reasoning.summary",
+            "summary": [
+                {"type": "summary_text", "text": f"summary: {SECRET_URI}"}
+            ],
+        },
+        {
+            "type": "reasoning.text",
+            "content": {"text": f"nested content: {SECRET_URI}"},
+            "data": "provider-data",
+        },
+    ]
+    original = copy.deepcopy(details)
+
+    built = _make_agent()._build_assistant_message(
+        _api_message(reasoning=f"mutable: {SECRET_URI}", reasoning_details=details),
+        "stop",
+    )
+
+    assert details == original
+    assert built["reasoning_details"] == original
+    assert SECRET_PASSWORD not in built["reasoning"]
+
+    replayed = _persist_message(tmp_path, built)
+    assistant = next(item for item in replayed if item["role"] == "assistant")
+    assert assistant["reasoning_details"] == original
+
+    from agent.transports import get_transport
+
+    outgoing = get_transport("chat_completions").convert_messages(
+        [assistant], model="openrouter/test-model"
+    )
+    assert outgoing[0]["reasoning_details"] == original
+
+
+@pytest.mark.parametrize(
+    "detail",
+    [
+        {
+            "type": "thinking",
+            "thinking": f"signed: {SECRET_URI}",
+            "signature": "provider-signature",
+        },
+        {
+            "type": "reasoning.text",
+            "text": f"encrypted: {SECRET_URI}",
+            "encrypted_content": "provider-ciphertext",
+        },
+        {
+            "type": "reasoning.encrypted",
+            "text": f"opaque: {SECRET_URI}",
+            "data": "provider-data",
+        },
+        {
+            "type": "redacted_thinking",
+            "text": f"opaque: {SECRET_URI}",
+            "data": "provider-data",
+        },
+    ],
+)
+def test_all_provider_reasoning_detail_shapes_are_preserved(detail):
+    original = copy.deepcopy(detail)
+    built = _make_agent()._build_assistant_message(
+        _api_message(reasoning="mutable", reasoning_details=[detail]),
+        "stop",
+    )
+
+    assert detail == original
+    assert built["reasoning_details"][0] == original
+
+
+def test_sdk_shaped_reasoning_details_are_preserved():
+    dict_backed = SimpleNamespace(
+        type="reasoning.text",
+        text=f"provider text: {SECRET_URI}",
+        provider_field="keep-me",
+    )
+
+    class ModelDumpOnly:
+        __slots__ = ()
+
+        def model_dump(self):
+            return {
+                "type": "reasoning.summary",
+                "summary": f"provider summary: {SECRET_URI}",
+                "provider_field": "keep-me-too",
+            }
+
+    built = _make_agent()._build_assistant_message(
+        _api_message(
+            reasoning="mutable",
+            reasoning_details=[dict_backed, ModelDumpOnly()],
+        ),
+        "stop",
+    )
+
+    assert built["reasoning_details"] == [
+        dict_backed.__dict__,
+        ModelDumpOnly().model_dump(),
+    ]
+
+
+def test_nested_summary_is_extracted_as_mutable_redacted_reasoning():
+    details = [
+        {
+            "type": "reasoning.summary",
+            "summary": [
+                {"type": "summary_text", "text": f"summary: {SECRET_URI}"}
+            ],
+        }
+    ]
+
+    built = _make_agent()._build_assistant_message(
+        _api_message(reasoning_details=details),
+        "stop",
+    )
+
+    assert built["reasoning"].startswith("summary: ")
+    assert SECRET_PASSWORD not in built["reasoning"]
+    assert built["reasoning_details"] == details
+
+
+def test_deep_reasoning_summary_is_extracted_without_recursion_failure():
+    nested = {"text": "deep summary"}
+    for _ in range(1_500):
+        nested = {"summary": [nested]}
+
+    built = _make_agent()._build_assistant_message(
+        _api_message(
+            reasoning_details=[
+                {"type": "reasoning.summary", "summary": [nested]}
+            ]
+        ),
+        "stop",
+    )
+
+    assert built["reasoning"] == "deep summary"
+
+
+def test_tool_call_arguments_remain_exact():
+    arguments = f'{{"command":"psql {SECRET_URI}"}}'
+    tool_call = SimpleNamespace(
+        id="call-1",
+        call_id="call-1",
+        response_item_id="response-1",
+        type="function",
+        function=SimpleNamespace(name="terminal", arguments=arguments),
+        extra_content=None,
+    )
+
+    built = _make_agent()._build_assistant_message(
+        _api_message(
+            content="",
+            reasoning=f"mutable: {SECRET_URI}",
+            tool_calls=[tool_call],
+        ),
+        "tool_calls",
+    )
+
+    assert built["tool_calls"][0]["function"]["arguments"] == arguments
+    assert SECRET_PASSWORD not in built["reasoning"]
+
+
+def test_redaction_opt_out_preserves_mutable_reasoning(monkeypatch):
+    monkeypatch.setattr("agent.redact._REDACT_ENABLED", False)
+    details = [{"type": "reasoning.text", "text": f"detail: {SECRET_URI}"}]
+
+    built = _make_agent()._build_assistant_message(
+        _api_message(
+            content=f"content: {SECRET_URI}",
+            reasoning=f"reasoning: {SECRET_URI}",
+            reasoning_details=details,
+        ),
+        "stop",
+    )
+
+    assert SECRET_URI in built["content"]
+    assert SECRET_URI in built["reasoning"]
+    assert SECRET_URI in built["reasoning_details"][0]["text"]
+
+
+def test_redacted_terminal_output_leaves_no_password_in_new_database(tmp_path):
+    from agent.redact import redact_terminal_output
+
+    raw_output = f"connected successfully: {SECRET_URI}"
+    safe_output = redact_terminal_output(raw_output, "printf database-uri")
+    assert SECRET_PASSWORD not in safe_output
+
+    _persist_message(
+        tmp_path,
+        {
+            "role": "tool",
+            "content": safe_output,
+            "tool_name": "terminal",
+            "tool_call_id": "call-1",
+        },
+    )
+
+    password_bytes = SECRET_PASSWORD.encode()
+    database_files = [path for path in tmp_path.rglob("*") if path.is_file()]
+    assert database_files
+    for path in database_files:
+        assert password_bytes not in path.read_bytes(), path.name
diff --git a/tests/agent/test_redact.py b/tests/agent/test_redact.py
index 68307c7381d2a..15331229510d0 100644
--- a/tests/agent/test_redact.py
+++ b/tests/agent/test_redact.py
@@ -375,6 +375,105 @@ def test_db_connstr_password_still_redacted(self):
         assert "dbpass" not in result
 
 
+class TestDbConnstrDialectDriver:
+    """Database URI variants with drivers and encrypted schemes (#43666)."""
+
+    @pytest.mark.parametrize(
+        "text",
+        [
+            "postgresql+psycopg://postgres:s3cretpw@127.0.0.1:5432/postgres",
+            "postgresql+asyncpg://postgres:s3cretpw@db/app",
+            "mysql+pymysql://root:s3cretpw@db:3306/app",
+            "mariadb+mariadbconnector://u:s3cretpw@h/db",
+            "mssql+pyodbc://sa:s3cretpw@h/db",
+            "oracle+oracledb://u:s3cretpw@h/service",
+            "clickhouse+native://u:s3cretpw@h/default",
+            "cockroachdb+psycopg://u:s3cretpw@h/defaultdb",
+            "snowflake+connector://u:s3cretpw@account/db",
+            "trino://u:s3cretpw@h/catalog",
+            "db2+ibm_db://u:s3cretpw@h/db",
+            "mysqlx://u:s3cretpw@h/db",
+            "mongodb+srv://u:s3cretpw@cluster0.mongodb.net/db",
+            "rediss://default:s3cretpw@h:6380/0",
+            "amqps://guest:s3cretpw@h:5671/",
+        ],
+    )
+    def test_dialect_driver_password_redacted(self, text):
+        result = redact_sensitive_text(text)
+        assert "s3cretpw" not in result
+        assert ":***@" in result
+
+    @pytest.mark.parametrize(
+        "text",
+        [
+            "redis://:s3cretpw@h:6379/0",
+            "rediss://:s3cretpw@h:6380/0",
+            "amqp://:s3cretpw@h:5672/",
+            "postgresql://:s3cretpw@h/db",
+        ],
+    )
+    def test_empty_username_password_redacted(self, text):
+        result = redact_sensitive_text(text)
+        assert "s3cretpw" not in result
+        assert "://:***@" in result
+
+    def test_percent_encoded_password_redacted(self):
+        text = "postgresql+psycopg://user:p%40ss%2Fword@db.example/app"
+        result = redact_sensitive_text(text)
+        assert "p%40ss%2Fword" not in result
+        assert ":***@" in result
+
+    def test_percent_encoded_surrounding_punctuation_in_password_redacted(self):
+        text = "postgresql://user:p%2Cass%3Bword@db.example/app"
+        result = redact_sensitive_text(text)
+        assert "p%2Cass%3Bword" not in result
+        assert ":***@" in result
+
+    @pytest.mark.parametrize(
+        "password",
+        [
+            "pa,ss",
+            "pa;ss",
+            "pa(ss)",
+            "123,abc",
+            "123;abc",
+            "123(abc)",
+        ],
+    )
+    def test_raw_uri_punctuation_in_password_redacted(self, password):
+        text = f"postgresql://user:{password}@db.example/app"
+        result = redact_sensitive_text(text)
+        assert password not in result
+        assert ":***@" in result
+
+    def test_digit_leading_punctuation_password_with_explicit_port_redacted(self):
+        text = "postgresql://user:123,abc@db.example:5432"
+        result = redact_sensitive_text(text)
+        assert "123,abc" not in result
+        assert ":***@" in result
+
+    @pytest.mark.parametrize(
+        "text",
+        [
+            "postgresql+psycopg://host.example.com:5432/db",
+            "redis://h:6379/0",
+            "postgresql://host:5432/db/user@example.com",
+            "postgresql://host:5432/db?email=user@example.com",
+            "postgresql://host:5432/db#user@example.com",
+            'header {"dsn":"postgresql+psycopg://db:5432",'
+            '"email":"user@example.com"}',
+            "header {'dsn':'postgresql+psycopg://db:5432',"
+            "'email':'user@example.com'}",
+            "postgresql://host:5432; contact=user@example.com",
+            "https://user:opaqueToken@host.example.com/path",
+            "notpostgresql://user:opaqueToken@host.example.com/path",
+            "custom+postgresql://user:opaqueToken@host.example.com/path",
+        ],
+    )
+    def test_non_credential_urls_unchanged(self, text):
+        assert redact_sensitive_text(text) == text
+
+
 class TestStrictUrlCredentialRedaction:
     @pytest.mark.parametrize(
         ("text", "secret", "expected"),
@@ -650,11 +749,15 @@ class TestDbConnstrCodeOutput:
 
     def test_literal_connstr_still_redacted_with_code_file(self):
         """A real password in a literal DSN is still masked under code_file."""
-        text = "postgresql://admin:realpassword@db.internal:5432/app"
+        text = "postgresql+psycopg://admin:realpassword@db.internal:5432/app"
         result = redact_sensitive_text(text, code_file=True, force=True)
         assert "realpassword" not in result
         assert "***" in result
 
+    def test_driver_connstr_template_preserved_with_code_file(self):
+        text = 'f"postgresql+psycopg://{user}:{password}@{host}/{database}"'
+        assert redact_sensitive_text(text, code_file=True, force=True) == text
+
     def test_literal_connstr_redacted_all_schemes(self):
         for scheme, secret in [
             ("postgres", "pgsecret1234"),
@@ -829,5 +932,3 @@ def test_plural_keys_still_redacted(self):
         text = "secrets: hunter2hunter2hunter2hh"
         result = redact_sensitive_text(text)
         assert "hunter2hunter2hunter2hh" not in result
-
-
diff --git a/tests/run_agent/test_streaming.py b/tests/run_agent/test_streaming.py
index 622474839c9ac..41961b368a67d 100644
--- a/tests/run_agent/test_streaming.py
+++ b/tests/run_agent/test_streaming.py
@@ -15,14 +15,14 @@
 
 def _make_stream_chunk(
     content=None, tool_calls=None, finish_reason=None,
-    model=None, reasoning_content=None, usage=None,
+    model=None, reasoning_content=None, reasoning=None, usage=None,
 ):
     """Build a mock streaming chunk matching OpenAI's ChatCompletionChunk shape."""
     delta = SimpleNamespace(
         content=content,
         tool_calls=tool_calls,
         reasoning_content=reasoning_content,
-        reasoning=None,
+        reasoning=reasoning,
     )
     choice = SimpleNamespace(
         index=0,
@@ -514,8 +514,81 @@ def test_reasoning_callback_fires(self, mock_close, mock_create):
         assert reasoning_deltas == ["Let me think", " about this"]
         assert text_deltas == ["The answer is 42"]
         assert response.choices[0].message.reasoning_content == "Let me think about this"
+        assert response.choices[0].message.reasoning is None
         assert response.choices[0].message.content == "The answer is 42"
 
+    @patch("run_agent.AIAgent._create_request_openai_client")
+    @patch("run_agent.AIAgent._close_request_openai_client")
+    def test_mutable_reasoning_keeps_distinct_provenance(
+        self, mock_close, mock_create
+    ):
+        """delta.reasoning must not be promoted to raw replay content."""
+        from run_agent import AIAgent
+
+        chunks = [
+            _make_stream_chunk(reasoning="ordinary "),
+            _make_stream_chunk(reasoning="thinking"),
+            _make_stream_chunk(content="answer", finish_reason="stop"),
+        ]
+        mock_client = MagicMock()
+        mock_client.chat.completions.create.return_value = iter(chunks)
+        mock_create.return_value = mock_client
+
+        agent = AIAgent(
+            api_key="test-key",
+            base_url="https://example.invalid/v1",
+            model="test/model",
+            quiet_mode=True,
+            skip_context_files=True,
+            skip_memory=True,
+            stream_delta_callback=lambda _text: None,
+        )
+        setattr(agent, "api_mode", "chat_completions")
+        agent._interrupt_requested = False
+
+        response = agent._interruptible_streaming_api_call({})
+
+        message = response.choices[0].message
+        assert message.reasoning == "ordinary thinking"
+        assert message.reasoning_content is None
+
+    @patch("run_agent.AIAgent._create_request_openai_client")
+    @patch("run_agent.AIAgent._close_request_openai_client")
+    def test_mixed_reasoning_fields_keep_distinct_provenance(
+        self, mock_close, mock_create
+    ):
+        """A chunk carrying both fields must preserve each field's origin."""
+        from run_agent import AIAgent
+
+        chunks = [
+            _make_stream_chunk(
+                reasoning="mutable reasoning",
+                reasoning_content="provider replay content",
+            ),
+            _make_stream_chunk(content="answer", finish_reason="stop"),
+        ]
+        mock_client = MagicMock()
+        mock_client.chat.completions.create.return_value = iter(chunks)
+        mock_create.return_value = mock_client
+
+        agent = AIAgent(
+            api_key="test-key",
+            base_url="https://example.invalid/v1",
+            model="test/model",
+            quiet_mode=True,
+            skip_context_files=True,
+            skip_memory=True,
+            stream_delta_callback=lambda _text: None,
+        )
+        setattr(agent, "api_mode", "chat_completions")
+        agent._interrupt_requested = False
+
+        response = agent._interruptible_streaming_api_call({})
+
+        message = response.choices[0].message
+        assert message.reasoning == "mutable reasoning"
+        assert message.reasoning_content == "provider replay content"
+
 
 # ── Test: _has_stream_consumers ──────────────────────────────────────────
 
@@ -1634,4 +1707,3 @@ def test_bedrock_reasoning_models_resolve_floor(self, model_id, expected):
         from agent.chat_completion_helpers import _bedrock_reasoning_stale_floor
 
         assert _bedrock_reasoning_stale_floor(model_id) == expected
-