diff --git a/gateway/run.py b/gateway/run.py
index a1d0f9929d3aa..ef626504dc4bc 100644
--- a/gateway/run.py
+++ b/gateway/run.py
@@ -1452,21 +1452,33 @@ def _select_cached_agent_history(
     persisted_history: List[Dict[str, Any]],
     live_history: Any,
 ) -> List[Dict[str, Any]]:
-    """Prefer a cached agent's live in-memory transcript over a shorter
-    persisted one.
+    """Prefer a cached live transcript only when it is longer and contains at
+    least one real, non-ephemeral unpersisted row.
 
     Guards the FTS write-corruption case (#50502): when message writes fail
     silently through corrupt FTS triggers, the next turn reloads a stale/empty
     ``conversation_history`` from disk even though the same cached ``AIAgent``
-    still holds the full live ``_session_messages``. Replacing the live
-    transcript with that shorter persisted copy causes immediate same-session
-    amnesia. When the live transcript is strictly longer, keep it.
+    still holds unpersisted real rows in ``_session_messages``. Replacing those
+    rows with the shorter persisted copy causes immediate same-session amnesia.
+    Length alone does not trigger retention.
 
     Returns ``persisted_history`` unchanged unless the live copy is a longer
-    list, in which case a copy of the live transcript is returned.
+    list containing at least one real transcript row without the intrinsic
+    ``_db_persisted`` marker. A longer all-durable list can be an expected
+    replay-filtering delta (for example, cleanup of an interrupted read-only
+    tool block). Deliberately unpersisted retry scaffolding is ignored.
     """
     if isinstance(live_history, list) and len(live_history) > len(persisted_history):
-        return list(live_history)
+        from run_agent import _is_ephemeral_scaffolding
+
+        has_unpersisted_row = any(
+            isinstance(message, dict)
+            and not message.get("_db_persisted")
+            and not _is_ephemeral_scaffolding(message)
+            for message in live_history
+        )
+        if has_unpersisted_row:
+            return list(live_history)
     return persisted_history
 
 
diff --git a/tests/gateway/test_cached_agent_history_guard.py b/tests/gateway/test_cached_agent_history_guard.py
new file mode 100644
index 0000000000000..2ee152a6b65bb
--- /dev/null
+++ b/tests/gateway/test_cached_agent_history_guard.py
@@ -0,0 +1,67 @@
+"""Regression tests for cached gateway transcript selection."""
+
+from gateway.run import _build_gateway_agent_history, _select_cached_agent_history
+
+
+def test_select_cached_history_keeps_expected_replay_cleanup():
+    """A durable read-only tool tail removed by stock replay cleanup stays removed."""
+    live = [
+        {"role": "user", "content": "check status", "_db_persisted": True},
+        {
+            "role": "assistant",
+            "content": "",
+            "tool_calls": [
+                {
+                    "id": "call-1",
+                    "type": "function",
+                    "function": {"name": "web_search", "arguments": "{}"},
+                }
+            ],
+            "_db_persisted": True,
+        },
+        {
+            "role": "tool",
+            "tool_call_id": "call-1",
+            "content": "[Command interrupted]",
+            "_db_persisted": True,
+        },
+    ]
+
+    persisted, observed = _build_gateway_agent_history(live)
+
+    assert observed is None
+    assert len(persisted) < len(live)
+    assert persisted == [{"role": "user", "content": "check status"}]
+    assert _select_cached_agent_history(persisted, live) is persisted
+
+
+def test_select_cached_history_preserves_live_when_a_real_row_is_unpersisted():
+    persisted = [
+        {"role": "user", "content": "hello"},
+        {"role": "assistant", "content": "hi"},
+    ]
+    live = [
+        {"role": "user", "content": "hello", "_db_persisted": True},
+        {"role": "assistant", "content": "hi", "_db_persisted": True},
+        {"role": "user", "content": "not written"},
+    ]
+
+    selected = _select_cached_agent_history(persisted, live)
+
+    assert selected is not persisted
+    assert selected == live
+
+
+def test_select_cached_history_ignores_unmarked_ephemeral_scaffolding():
+    """Internal retry nudges are deliberately unpersisted, not evidence of DB lag."""
+    persisted = [{"role": "user", "content": "ship the fix"}]
+    live = [
+        {"role": "user", "content": "ship the fix", "_db_persisted": True},
+        {
+            "role": "user",
+            "content": "Verify your work before stopping.",
+            "_pre_verify_synthetic": True,
+        },
+    ]
+
+    assert _select_cached_agent_history(persisted, live) is persisted