diff --git a/CHANGELOG.md b/CHANGELOG.md
index ab26504e..7c66f9d9 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -11,6 +11,8 @@
 
 - **Refine selected transcript text directly in the composer.** Selecting text inside a chat message now shows a "Refine" action alongside "Reply with selection" in the floating selection toolbar. Refine seeds the selected text into the composer as an editable draft (a quoted block plus a localized "Refine instruction:" line, caret ready) so you can shape a follow-up before sending — the ChatGPT/Notion-style quote-to-edit flow. The transfer is one-way and marker-free (no internal sentinel reaches the composer or the sent message), preserves any existing draft and attachments, collapses to 44×44 icon-only buttons on phones, and covers all 15 locales. Thanks @rodboev. (#6410)
 
+- **Agent replay is now byte-accurate without ever exposing internal provider text to the browser.** The exact provider-facing reply text (`api_content`) is preserved through the internal session store so agent-to-agent replay is faithful, while every client-facing surface (HTTP responses, SSE journal + runner streams, shares) strips it and its provenance aliases through the public projection. Includes a maintainer hardening pass: session redaction is selective (transcript fields are scrubbed; operational fields such as the workspace path pass through unchanged, so a workspace path that merely contains credential-shaped text is no longer masked into an invalid path), and the runner-backed SSE relay now projects payloads before emit so `api_content` cannot leak through that path. Thanks @franksong2702. (#6757)
+
 - **Transparent Stream copy buttons now confirm success in place.** The tool-event and thinking copy controls flash a check for ~1.5s after a successful copy — the same feedback the normal chat copy buttons already give — then revert to the copy glyph. The confirmation is now clearly visible without hover (important on touch), stays correct across live-row reconciliation, cleans up if its row is torn down mid-feedback, and is never serialized into a restored session snapshot. Thanks @Stacey2911. (#6037)
 
 - **Kanban boards can now set a default workspace path.** The board settings modal gains a "Default workspace path" field (a type-or-pick combobox sourced from your saved workspaces, so you can choose a known path or type a custom one — validated server-side). Tasks on the board without an explicit workspace path inherit this default. The board switcher chip is now always shown (even with a single board) and its menu item is now "Board settings…", enabled for the default board. Thanks @rodboev. (#6055, #5932)
diff --git a/api/helpers.py b/api/helpers.py
index d285cfe4..417bd668 100644
--- a/api/helpers.py
+++ b/api/helpers.py
@@ -14,6 +14,13 @@ from api.config import IMAGE_EXTS, MD_EXTS
 
 logger = logging.getLogger(__name__)
 
+_PUBLIC_MESSAGE_INTERNAL_FIELDS = frozenset({
+    "api_content",
+    "_state_db_row_id",
+    "_db_row_id",
+    "state_db_row_id",
+})
+
 
 # Treat stalled/closed HTTP clients as normal disconnects.  Long-lived SSE
 # connections often end this way when a browser tab sleeps, a phone switches
@@ -949,10 +956,7 @@ def _redact_value(v, *, _enabled: bool | None = None):
     if isinstance(v, str):
         return _redact_text(v, _enabled=_enabled)
     if isinstance(v, dict):
-        return {
-            key: _redact_value(value, _enabled=_enabled)
-            for key, value in v.items()
-        }
+        return {key: _redact_value(value, _enabled=_enabled) for key, value in v.items()}
     if isinstance(v, list):
         return [_redact_value(item, _enabled=_enabled) for item in v]
     return v
@@ -985,26 +989,207 @@ def _redact_message_content_part(part, *, _enabled: bool):
     return result
 
 
+def _scrub_alias_record(record):
+    """Copy one schema record while removing private replay aliases.
+
+    This deliberately copies only one record level.  Values such as a tool's
+    ``args`` or a function's opaque ``arguments`` are business payloads, not
+    nested WebUI records, so parsing or recursively walking them would corrupt
+    valid user data.
+    """
+    if not isinstance(record, dict):
+        return _copy_json_value(record)
+    return {
+        key: _copy_json_value(value)
+        for key, value in record.items()
+        if key not in _PUBLIC_MESSAGE_INTERNAL_FIELDS
+    }
+
+
+def _scrub_content_part(part):
+    """Scrub one canonical ``messages[*].content[*]`` part."""
+    return _scrub_alias_record(part)
+
+
+def _scrub_function_record(function):
+    """Scrub a tool-call function envelope without parsing arguments."""
+    return _scrub_alias_record(function)
+
+
+def _scrub_tool_call_record(tool_call):
+    """Scrub one canonical tool-call record and its function envelope."""
+    result = _scrub_alias_record(tool_call)
+    if not isinstance(tool_call, dict):
+        return result
+    function = tool_call.get("function")
+    if isinstance(function, dict):
+        result["function"] = _scrub_function_record(function)
+    return result
+
+
+def _scrub_message_record(message, *, preserve_api_content: bool = False):
+    """Scrub one message record along its authoritative nested schema paths."""
+    if not isinstance(message, dict):
+        return _copy_json_value(message)
+    result = _scrub_alias_record(message)
+    if preserve_api_content and isinstance(message.get("api_content"), str) and message.get("api_content"):
+        result["api_content"] = message["api_content"]
+    content = message.get("content")
+    if isinstance(content, list):
+        result["content"] = [_scrub_content_part(part) for part in content]
+    tool_calls = message.get("tool_calls")
+    if isinstance(tool_calls, list):
+        result["tool_calls"] = [_scrub_tool_call_record(call) for call in tool_calls]
+    return result
+
+
+def _scrub_message_records(messages, *, preserve_api_content: bool = False):
+    if not isinstance(messages, list):
+        return _copy_json_value(messages)
+    return [
+        _scrub_message_record(message, preserve_api_content=preserve_api_content)
+        for message in messages
+    ]
+
+
+def _scrub_tool_call_records(tool_calls):
+    if not isinstance(tool_calls, list):
+        return _copy_json_value(tool_calls)
+    return [_scrub_tool_call_record(call) for call in tool_calls]
+
+
+def _scrub_runtime_journal_snapshot(snapshot):
+    """Scrub only the snapshot's real message/tool-call arrays.
+
+    A live tool's ``args`` dictionary is intentionally opaque.  In particular,
+    ``tool_calls[].args.messages[]`` is ordinary business input even though the
+    field name happens to be ``messages``; it is not a transcript container.
+    """
+    if not isinstance(snapshot, dict):
+        return _copy_json_value(snapshot)
+    result = _copy_json_value(snapshot)
+    if isinstance(snapshot.get("messages"), list):
+        result["messages"] = _scrub_message_records(snapshot["messages"])
+    if isinstance(snapshot.get("tool_calls"), list):
+        result["tool_calls"] = _scrub_tool_call_records(snapshot["tool_calls"])
+    return result
+
+
+def scrub_internal_replay_fields(
+    value,
+    *,
+    preserve_message_api_content: bool = False,
+    message_records: bool | None = None,
+):
+    """Copy runtime data and scrub aliases at authoritative schema positions.
+
+    ``message_records`` selects the shape of a bare list (messages versus
+    session-level tool calls).  A session-shaped dictionary follows only its
+    named ``messages``, ``context_messages``, ``tool_calls``, and
+    ``runtime_journal_snapshot`` fields.  No generic key-name recursion is
+    performed, so arbitrary nested tool arguments remain byte-for-byte intact.
+    The Agent boundary may retain a message record's trusted ``api_content``;
+    public/import boundaries use the default, stripping it everywhere.
+    """
+    if isinstance(value, list):
+        if message_records is False:
+            return _scrub_tool_call_records(value)
+        return _scrub_message_records(
+            value,
+            preserve_api_content=preserve_message_api_content,
+        )
+    if not isinstance(value, dict):
+        return _copy_json_value(value)
+    result = _copy_json_value(value)
+    for key, child in value.items():
+        if key in {"messages", "context_messages"} and isinstance(child, list):
+            result[key] = _scrub_message_records(
+                child,
+                preserve_api_content=preserve_message_api_content,
+            )
+        elif key == "tool_calls" and isinstance(child, list):
+            result[key] = _scrub_tool_call_records(child)
+        elif key == "runtime_journal_snapshot" and isinstance(child, dict):
+            result[key] = _scrub_runtime_journal_snapshot(child)
+    return result
+
+
+def _public_message_projection(message, *, _enabled: bool):
+    """Return one public transcript message without internal replay fields."""
+    message = scrub_internal_replay_fields([message], message_records=True)[0]
+    if not isinstance(message, dict):
+        return _redact_value(message, _enabled=_enabled)
+    item = {}
+    allow_native_image = message.get("role") == "user"
+    for key, value in message.items():
+        if key in _PUBLIC_MESSAGE_INTERNAL_FIELDS:
+            continue
+        if allow_native_image and key == "content" and isinstance(value, list):
+            item[key] = [
+                _redact_message_content_part(part, _enabled=_enabled)
+                for part in value
+            ]
+        else:
+            item[key] = _redact_value(value, _enabled=_enabled)
+    return item
+
+
 def _redact_messages(messages, *, _enabled: bool):
     if not isinstance(messages, list):
         return _redact_value(messages, _enabled=_enabled)
-    redacted = []
-    for message in messages:
-        if not isinstance(message, dict):
-            redacted.append(_redact_value(message, _enabled=_enabled))
-            continue
-        item = {}
-        allow_native_image = message.get("role") == "user"
-        for key, value in message.items():
-            if allow_native_image and key == "content" and isinstance(value, list):
-                item[key] = [
-                    _redact_message_content_part(part, _enabled=_enabled)
-                    for part in value
-                ]
-            else:
-                item[key] = _redact_value(value, _enabled=_enabled)
-        redacted.append(item)
-    return redacted
+    return [_public_message_projection(message, _enabled=_enabled) for message in messages]
+
+
+def _redact_tool_calls(tool_calls, *, _enabled: bool):
+    scrubbed = scrub_internal_replay_fields(tool_calls, message_records=False)
+    return _redact_value(scrubbed, _enabled=_enabled)
+
+
+def _redact_nested_message_containers(value, *, _enabled: bool):
+    """Redact only the runtime snapshot's authoritative message arrays."""
+    scrubbed = scrub_internal_replay_fields(value)
+    if not isinstance(scrubbed, dict):
+        return _redact_value(scrubbed, _enabled=_enabled)
+    result = {}
+    for key, child in scrubbed.items():
+        if key == "messages" and isinstance(child, list):
+            result[key] = _redact_messages(child, _enabled=_enabled)
+        elif key == "tool_calls" and isinstance(child, list):
+            result[key] = _redact_tool_calls(child, _enabled=_enabled)
+        else:
+            result[key] = _redact_value(child, _enabled=_enabled)
+    return result
+
+
+def public_session_projection(session_dict: dict) -> dict:
+    """Return a public session payload with redaction and alias stripping.
+
+    Callers use this for every response/export/SSE session payload.  It never
+    mutates the in-memory session or the caller's dictionary.
+    """
+    return redact_session_data(session_dict)
+
+
+def strip_public_internal_fields(value, *, message_records: bool = False):
+    """Deep-copy imported records through the shared schema scrubber.
+
+    JSON import uses this before constructing or saving a ``Session``.  The
+    Four replay aliases belong to a message/content-part/tool-call/function
+    record itself; matching names inside user content or tool arguments are
+    ordinary JSON and must be preserved.  This is intentionally independent of
+    the credential-redaction setting: caller-supplied provider sidecars must
+    never become durable WebUI session state.
+    """
+    return scrub_internal_replay_fields(value, message_records=message_records)
+
+
+def _copy_json_value(value):
+    """Deep-copy JSON-shaped data without applying message-field filtering."""
+    if isinstance(value, dict):
+        return {key: _copy_json_value(child) for key, child in value.items()}
+    if isinstance(value, list):
+        return [_copy_json_value(child) for child in value]
+    return value
 
 
 def redact_session_data(session_dict: dict) -> dict:
@@ -1022,20 +1207,28 @@ def redact_session_data(session_dict: dict) -> dict:
     """
     from api.config import load_settings
     _enabled = bool(load_settings().get("api_redact_enabled", True))
-    result = dict(session_dict)
-    if isinstance(result.get('title'), str):
-        result['title'] = _redact_text(result['title'], _enabled=_enabled)
-    if 'messages' in result:
-        result['messages'] = _redact_messages(result['messages'], _enabled=_enabled)
-    if 'tool_calls' in result:
-        result['tool_calls'] = _redact_value(result['tool_calls'], _enabled=_enabled)
-    if 'todo_state' in result:
-        result['todo_state'] = _redact_value(result['todo_state'], _enabled=_enabled)
-    if 'runtime_journal_snapshot' in result:
-        result['runtime_journal_snapshot'] = _redact_value(
-            result['runtime_journal_snapshot'],
-            _enabled=_enabled,
-        )
+    if not isinstance(session_dict, dict):
+        return {}
+    result = {}
+    for key, value in session_dict.items():
+        if key in _PUBLIC_MESSAGE_INTERNAL_FIELDS:
+            continue
+        if key == 'title' and isinstance(value, str):
+            result[key] = _redact_text(value, _enabled=_enabled)
+        elif key in {'messages', 'context_messages'}:
+            result[key] = _redact_messages(value, _enabled=_enabled)
+        elif key == 'tool_calls' and isinstance(value, list):
+            result[key] = _redact_tool_calls(value, _enabled=_enabled)
+        elif key in {'todo_state', 'runtime_journal_snapshot'}:
+            result[key] = _redact_nested_message_containers(value, _enabled=_enabled)
+        else:
+            # Operational fields (workspace path, ids, config, timestamps, etc.)
+            # are NOT credential-masked: a valid workspace path may legitimately
+            # contain a credential-shaped component, and masking it would corrupt
+            # the authoritative value the client echoes back on the next send.
+            # Deep-copy them through unchanged (only transcript-bearing fields
+            # above carry free-form user/model text worth redacting).
+            result[key] = _copy_json_value(value)
     return result
 
 
diff --git a/api/models.py b/api/models.py
index c1762474..6fdc7774 100644
--- a/api/models.py
+++ b/api/models.py
@@ -7986,6 +7986,11 @@ def get_state_db_session_messages(
                 'codex_reasoning_items',
                 'reasoning_content',
                 'codex_message_items',
+                # Hermes Agent stores the exact provider-facing text here when
+                # it differs from the clean transcript content.  Keep this
+                # sidecar in the WebUI's internal history; the provider-safe
+                # projection strips it before any direct API request.
+                'api_content',
             ]
             id_col = ['id'] if 'id' in available else []
             selected = id_col + ['role', 'content', 'timestamp'] + [c for c in optional if c in available]
@@ -8090,6 +8095,11 @@ def get_state_db_session_messages(
                     'content': row['content'],
                     'timestamp': row['timestamp'],
                 }
+                # ``id`` is the durable SQLite row identity, not the WebUI's
+                # session-local stable message id.  Keep it in a private
+                # provenance field so duplicate reconciliation can align a
+                # state.db sidecar without changing the existing ``id`` key
+                # used by WebUI transcript merge/dedup logic.
                 for col in optional:
                     if col not in row.keys():
                         continue
@@ -8099,6 +8109,16 @@ def get_state_db_session_messages(
                     if col in {'tool_calls', 'reasoning_details', 'codex_reasoning_items', 'codex_message_items'}:
                         value = _json_loads_if_string(value)
                     msg[col] = value
+                # Keep durable provenance only alongside a real Agent replay
+                # sidecar. Ordinary state.db rows must retain their historic
+                # shape and must not acquire internal bookkeeping fields.
+                if (
+                    id_col
+                    and row['id'] is not None
+                    and isinstance(msg.get('api_content'), str)
+                    and msg['api_content']
+                ):
+                    msg['_state_db_row_id'] = row['id']
                 if msg.get('role') == 'tool' and msg.get('tool_name') and not msg.get('name'):
                     msg['name'] = msg['tool_name']
                 msgs.append(msg)
@@ -8131,6 +8151,8 @@ def get_state_db_session_message_prefix_summary(
         before_ts = float(before_timestamp)
     except (TypeError, ValueError):
         return None
+    if not math.isfinite(before_ts):
+        return None
 
     if isinstance(profile, str) and profile:
         db_path = _get_profile_home(profile) / 'state.db'
@@ -8219,12 +8241,13 @@ def get_state_db_session_message_keys_before_timestamp(
             available = {str(row['name']) for row in cur.fetchall()}
             if not {'id', 'session_id', 'role', 'content', 'timestamp', 'tool_calls'}.issubset(available):
                 return None
+            api_content_select = ", api_content" if "api_content" in available else ""
             cur.execute(
-                """
+                f"""
                 SELECT
                     COALESCE(role, '') AS role,
                     COALESCE(content, '') AS content,
-                    tool_calls
+                    tool_calls{api_content_select}
                 FROM messages
                 WHERE session_id = ? AND timestamp IS NOT NULL AND timestamp < ?
                 ORDER BY timestamp ASC, id ASC
@@ -8237,6 +8260,7 @@ def get_state_db_session_message_keys_before_timestamp(
                         "role": row["role"],
                         "content": row["content"],
                         "tool_calls": _json_loads_if_string(row["tool_calls"]),
+                        "api_content": row["api_content"] if "api_content" in available else None,
                     }
                 )
                 for row in cur.fetchall()
@@ -8299,6 +8323,11 @@ def _normalized_message_timestamp_for_key(value):
         timestamp = float(value)
     except (TypeError, ValueError):
         return str(value)
+    if not math.isfinite(timestamp):
+        # Keep malformed metadata from crashing key construction.  The
+        # reconciliation layer separately marks this row invalid so it cannot
+        # fall through to a weaker metadata-free match.
+        return f"<invalid:{value!r}>"
     # Truncate to second-level granularity so that sub-second drift between
     # the sidecar JSON write and the state.db created_at write does not cause
     # the legacy dedup key to differ for the same logical message.
@@ -8312,9 +8341,29 @@ def _message_timestamp_as_float(msg):
     if value is None or value == "":
         return None
     try:
-        return float(value)
+        timestamp = float(value)
     except (TypeError, ValueError):
         return None
+    return timestamp if math.isfinite(timestamp) else None
+
+
+def _session_message_api_content_key(msg: dict):
+    """Return the exact trusted provider sidecar used in duplicate identity."""
+    if not isinstance(msg, dict):
+        return None
+    value = msg.get("api_content")
+    return value if isinstance(value, str) and value else None
+
+
+def _session_message_key_with_sidecar(base_key: tuple, msg: dict) -> tuple:
+    """Append provider sidecar identity only when one is actually present.
+
+    The no-sidecar key shape is an internal compatibility surface used by
+    reconciliation tests and callers.  A present sidecar must extend that
+    identity so different provider bytes cannot collapse into one duplicate.
+    """
+    sidecar = _session_message_api_content_key(msg)
+    return base_key if sidecar is None else (*base_key, sidecar)
 
 
 def _session_message_merge_key(msg: dict):
@@ -8322,7 +8371,9 @@ def _session_message_merge_key(msg: dict):
         return ("non_dict", repr(msg))
     message_identity = msg.get("id") or msg.get("message_id")
     if message_identity:
-        return ("message_id", str(message_identity))
+        return _session_message_key_with_sidecar(
+            ("message_id", str(message_identity)), msg
+        )
     # Include tool_calls so assistant messages that invoke different tools
     # (but share identical empty content and same-second timestamp) are not
     # collapsed by the merge-key guard at line ~4216.  Without this,
@@ -8331,7 +8382,7 @@ def _session_message_merge_key(msg: dict):
     # every state.db tool-call after the first one registered by the sidecar.
     _tc = msg.get("tool_calls")
     _tc_key = json.dumps(_tc, sort_keys=True, default=str) if _tc else ""
-    return (
+    return _session_message_key_with_sidecar((
         "legacy",
         str(msg.get("role") or ""),
         str(msg.get("content") or ""),
@@ -8339,7 +8390,7 @@ def _session_message_merge_key(msg: dict):
         str(msg.get("tool_call_id") or ""),
         str(msg.get("tool_name") or msg.get("name") or ""),
         _tc_key,
-    )
+    ), msg)
 
 
 def _session_messages_have_prefix(messages, prefix) -> bool:
@@ -8386,6 +8437,585 @@ def _merge_session_display_metadata(target: dict | None, source: dict | None) ->
             target[key] = copy.deepcopy(value)
 
 
+def _state_db_row_identity_details(message: dict | None) -> tuple[str | None, bool]:
+    """Return ``(row_id, valid)`` for private state.db provenance aliases.
+
+    A message carrying two different aliases is contradictory provenance.  It
+    must not silently fall through to timestamp/sequence matching, because that
+    would turn an identity conflict into a guessed provider-side payload.
+    """
+    if not isinstance(message, dict):
+        return None, True
+    values = set()
+    for key in ("_state_db_row_id", "_db_row_id", "state_db_row_id"):
+        if key not in message or message.get(key) in (None, ""):
+            continue
+        value = message.get(key)
+        if isinstance(value, bool):
+            return None, False
+        if isinstance(value, int):
+            normalized = str(value) if value >= 0 else None
+        elif isinstance(value, float):
+            normalized = str(int(value)) if math.isfinite(value) and value >= 0 and value.is_integer() else None
+        elif isinstance(value, str):
+            text = value.strip()
+            normalized = text if text.isdigit() else None
+            if normalized is not None:
+                normalized = str(int(normalized))
+        else:
+            normalized = None
+        if normalized is None:
+            return None, False
+        values.add(normalized)
+    if len(values) > 1:
+        return None, False
+    return (next(iter(values)) if values else None), True
+
+
+def _state_db_row_identity(message: dict | None):
+    """Return durable state.db provenance without treating WebUI ``id`` as it."""
+    identity, valid = _state_db_row_identity_details(message)
+    return identity if valid else None
+
+
+def _stable_message_identity_details(message: dict | None) -> tuple[str | None, bool]:
+    """Return a canonical stable message id and whether its aliases are valid."""
+    if not isinstance(message, dict):
+        return None, True
+    values = set()
+    for key in ("id", "message_id"):
+        if key not in message or message.get(key) in (None, ""):
+            continue
+        value = message.get(key)
+        if isinstance(value, bool):
+            return None, False
+        if isinstance(value, (int, str)):
+            normalized = str(value).strip()
+        elif isinstance(value, float):
+            normalized = str(value) if math.isfinite(value) else ""
+        else:
+            normalized = ""
+        if not normalized:
+            return None, False
+        values.add(normalized)
+    if len(values) > 1:
+        return None, False
+    return (next(iter(values)) if values else None), True
+
+
+def _message_identity_compatible(target: dict | None, source: dict | None) -> bool:
+    """Return whether private identities do not contradict one another."""
+    target_stable, target_stable_valid = _stable_message_identity_details(target)
+    source_stable, source_stable_valid = _stable_message_identity_details(source)
+    if not target_stable_valid or not source_stable_valid:
+        return False
+    if target_stable is not None and source_stable is not None and target_stable != source_stable:
+        return False
+    target_row_id, target_row_id_valid = _state_db_row_identity_details(target)
+    source_row_id, source_row_id_valid = _state_db_row_identity_details(source)
+    if not target_row_id_valid or not source_row_id_valid:
+        return False
+    if target_row_id is not None and source_row_id is not None and target_row_id != source_row_id:
+        return False
+    return _visible_content_compatible(target, source)
+
+
+def _message_exact_timestamp(message: dict | None):
+    """Return a numeric transcript timestamp, preserving ``0`` as valid."""
+    if not isinstance(message, dict):
+        return None
+    parsed, valid = _message_exact_timestamp_details(message)
+    return parsed if valid else None
+
+
+def _message_exact_timestamp_details(message: dict | None) -> tuple[float | None, bool]:
+    """Return ``(timestamp, valid)`` while distinguishing absent metadata."""
+    if not isinstance(message, dict):
+        return None, True
+    for key in ("timestamp", "_ts"):
+        if key not in message or message.get(key) in (None, ""):
+            continue
+        try:
+            parsed = float(message.get(key))
+        except (TypeError, ValueError):
+            return None, False
+        if not math.isfinite(parsed):
+            return None, False
+        return parsed, True
+    return None, True
+
+
+def _message_sidecar_role(message: dict | None):
+    if not isinstance(message, dict):
+        return None
+    role = str(message.get("role") or "").strip().lower()
+    return role if role in {"user", "assistant"} else None
+
+
+_WORKSPACE_PREFIX_RE = re.compile(r"^\s*\[Workspace(?:::v1)?:[^\]]+\]\s*")
+
+
+def _message_visible_content_key(message: dict | None):
+    """Return a canonical visible-content key for sidecar reconciliation."""
+    if not isinstance(message, dict):
+        return None
+    role = _message_sidecar_role(message)
+    if role is None:
+        return None
+
+    def _normalize(value, *, strip_prefix=False):
+        if isinstance(value, str):
+            text = value
+            if strip_prefix:
+                text = _WORKSPACE_PREFIX_RE.sub("", text, count=1)
+            return " ".join(text.split())
+        if isinstance(value, list):
+            return [_normalize(item, strip_prefix=strip_prefix) for item in value]
+        if isinstance(value, dict):
+            return {
+                key: _normalize(child, strip_prefix=False)
+                for key, child in value.items()
+                if key not in {"api_content", "_state_db_row_id", "_db_row_id", "state_db_row_id"}
+            }
+        return value
+
+    normalized = _normalize(message.get("content"), strip_prefix=(role == "user"))
+    try:
+        content = json.dumps(normalized, ensure_ascii=False, sort_keys=True, default=str)
+    except (TypeError, ValueError):
+        content = repr(normalized)
+    return role, content
+
+
+def _visible_content_compatible(target: dict | None, source: dict | None) -> bool:
+    """Return True only when role and visible content agree exactly."""
+    return _message_visible_content_key(target) == _message_visible_content_key(source)
+
+
+def _copy_api_content_sidecar(target: dict | None, source: dict | None) -> bool:
+    """Copy a non-empty internal sidecar without replacing an existing one."""
+    if not isinstance(target, dict) or not isinstance(source, dict):
+        return False
+    target_role = _message_sidecar_role(target)
+    source_role = _message_sidecar_role(source)
+    if target_role is None or target_role != source_role:
+        return False
+    if not _visible_content_compatible(target, source):
+        return False
+    if target.get("api_content") not in (None, ""):
+        return True
+    api_content = source.get("api_content")
+    if isinstance(api_content, str) and api_content:
+        target["api_content"] = api_content
+        return True
+    return False
+
+
+def _reconcile_api_content_sidecars(sidecar_messages: list, state_messages: list) -> None:
+    """Attach state.db ``api_content`` to the matching sidecar transcript rows.
+
+    Matching is intentionally stricter than visible transcript dedupe. A
+    unique stable message id is preferred, followed by a unique durable
+    state.db row id, exact timestamps, same-second timestamp drift, and finally
+    a mutually-unique role/content pair that may have mixed timestamp metadata.
+    Duplicate, malformed, or contradictory provenance is consumed without a
+    fallback guess. Repeated visible rows without unique provenance remain
+    unattached. If both sides already carry different non-empty provider
+    sidecars, neither is attached; the append-only merge preserves those rows
+    as distinct.
+    """
+    sidecar = [message for message in sidecar_messages or () if isinstance(message, dict)]
+    state = [
+        message
+        for message in state_messages or ()
+        if isinstance(message, dict)
+        and _message_sidecar_role(message) is not None
+        and isinstance(message.get("api_content"), str)
+        and message.get("api_content")
+    ]
+    if not sidecar or not state:
+        return
+
+    used_targets: set[int] = set()
+    used_sources: set[int] = set()
+
+    # Invalid provenance is never treated as absent metadata. Consume those
+    # rows up front so no weaker tier can attach an arbitrary provider sidecar.
+    for index, message in enumerate(sidecar):
+        _, row_id_valid = _state_db_row_identity_details(message)
+        _, timestamp_valid = _message_exact_timestamp_details(message)
+        _, stable_id_valid = _stable_message_identity_details(message)
+        if not row_id_valid or not timestamp_valid or not stable_id_valid:
+            used_targets.add(index)
+    for source_index, message in enumerate(state):
+        _, row_id_valid = _state_db_row_identity_details(message)
+        _, timestamp_valid = _message_exact_timestamp_details(message)
+        _, stable_id_valid = _stable_message_identity_details(message)
+        if not row_id_valid or not timestamp_valid or not stable_id_valid:
+            used_sources.add(source_index)
+
+    # 1. Stable message identity. This is the strongest WebUI-side identity
+    # and must run before timestamp/content fallbacks. Duplicate or conflicting
+    # aliases are consumed above and cannot fall through to a guess.
+    targets_by_stable_id = {}
+    state_by_stable_id = {}
+    for index, message in enumerate(sidecar):
+        if index in used_targets:
+            continue
+        stable_id, valid = _stable_message_identity_details(message)
+        if valid and stable_id is not None:
+            targets_by_stable_id.setdefault(stable_id, []).append(index)
+    for source_index, message in enumerate(state):
+        if source_index in used_sources:
+            continue
+        stable_id, valid = _stable_message_identity_details(message)
+        if valid and stable_id is not None:
+            state_by_stable_id.setdefault(stable_id, []).append(source_index)
+    stable_ids = list(targets_by_stable_id)
+    stable_ids.extend(
+        stable_id
+        for stable_id in state_by_stable_id
+        if stable_id not in targets_by_stable_id
+    )
+    duplicate_stable_target_indexes = set()
+    duplicate_stable_source_indexes = set()
+    for stable_id in stable_ids:
+        target_indexes = targets_by_stable_id.get(stable_id, [])
+        source_indexes = state_by_stable_id.get(stable_id, [])
+        if len(target_indexes) > 1 or len(source_indexes) > 1:
+            # Durable row ids are stronger provenance than a repeated stable
+            # message id. Defer this bucket's quarantine until after the row-id
+            # tier so unique row-id pairs can still resolve it one-to-one.
+            duplicate_stable_target_indexes.update(target_indexes)
+            duplicate_stable_source_indexes.update(source_indexes)
+
+    for stable_id in stable_ids:
+        target_indexes = [
+            index
+            for index in targets_by_stable_id.get(stable_id, ())
+            if index not in used_targets
+        ]
+        source_indexes = [
+            index
+            for index in state_by_stable_id.get(stable_id, ())
+            if index not in used_sources
+        ]
+        # A singleton stable id present on only one side remains eligible for
+        # stronger durable-row, timestamp, or content matching below.
+        if len(target_indexes) != 1 or len(source_indexes) != 1:
+            continue
+        target_index = target_indexes[0]
+        source_index = source_indexes[0]
+        if not _message_identity_compatible(sidecar[target_index], state[source_index]):
+            used_targets.add(target_index)
+            used_sources.add(source_index)
+            continue
+        target_api_content = _session_message_api_content_key(sidecar[target_index])
+        source_api_content = _session_message_api_content_key(state[source_index])
+        used_targets.add(target_index)
+        used_sources.add(source_index)
+        if (
+            target_api_content is not None
+            and source_api_content is not None
+            and target_api_content != source_api_content
+        ):
+            continue
+        _copy_api_content_sidecar(sidecar[target_index], state[source_index])
+
+    # 2. Durable row identity. This path is unambiguous only when every
+    # alias agrees, each row id occurs once on each side, and visible content
+    # is compatible.  Any duplicate/conflicting id is consumed and rejected;
+    # it must not fall through to a weaker tier.  A unique row id with
+    # conflicting non-empty sidecars is left for the append-only merge, which
+    # can preserve both authoritative payloads without guessing.
+    targets_by_row_id = {}
+    for index, message in enumerate(sidecar):
+        if index in used_targets:
+            continue
+        row_id, valid = _state_db_row_identity_details(message)
+        if not valid:
+            used_targets.add(index)
+        elif row_id is not None:
+            targets_by_row_id.setdefault(row_id, []).append(index)
+    state_by_row_id = {}
+    for source_index, message in enumerate(state):
+        if source_index in used_sources:
+            continue
+        row_id, valid = _state_db_row_identity_details(message)
+        if not valid:
+            used_sources.add(source_index)
+        elif row_id is not None:
+            state_by_row_id.setdefault(row_id, []).append(source_index)
+
+    # Reject duplicate source buckets independently of the target map.  A
+    # source-only duplicate id has no target loop iteration to consume it, so
+    # without this pre-pass it would descend into timestamp/metadata-free
+    # matching and attach one arbitrary provider sidecar.
+    for source_indexes in state_by_row_id.values():
+        if len(source_indexes) > 1:
+            used_sources.update(source_indexes)
+
+    for row_id, original_target_indexes in targets_by_row_id.items():
+        original_source_indexes = state_by_row_id.get(row_id, [])
+        target_indexes = [
+            index for index in original_target_indexes if index not in used_targets
+        ]
+        source_indexes = [
+            index for index in original_source_indexes if index not in used_sources
+        ]
+        if len(original_target_indexes) != 1 or len(original_source_indexes) != 1:
+            used_targets.update(target_indexes)
+            used_sources.update(source_indexes)
+            continue
+        if len(target_indexes) != 1 or len(source_indexes) != 1:
+            continue
+        target_index = target_indexes[0]
+        source_index = source_indexes[0]
+        if not _message_identity_compatible(sidecar[target_index], state[source_index]):
+            # Row ids never override contradictory stable ids, aliases, roles,
+            # or visible content. Quarantine both sides before fallback.
+            used_targets.add(target_index)
+            used_sources.add(source_index)
+            continue
+        target_api_content = _session_message_api_content_key(sidecar[target_index])
+        source_api_content = _session_message_api_content_key(state[source_index])
+        if (
+            target_api_content is not None
+            and source_api_content is not None
+            and target_api_content != source_api_content
+        ):
+            # A durable row id is not sufficient to reconcile two already
+            # authoritative provider payloads. Quarantine both rows so a
+            # weaker timestamp/content tier cannot attach one payload to the
+            # other row.
+            used_targets.add(target_index)
+            used_sources.add(source_index)
+            continue
+        used_targets.add(target_index)
+        used_sources.add(source_index)
+        _copy_api_content_sidecar(sidecar[target_index], state[source_index])
+
+    # Any duplicate stable-id rows left unresolved by the authoritative row-id
+    # tier are ambiguous and must not fall through to timestamps/content.
+    used_targets.update(
+        index for index in duplicate_stable_target_indexes if index not in used_targets
+    )
+    used_sources.update(
+        index for index in duplicate_stable_source_indexes if index not in used_sources
+    )
+
+    # 3. Exact role + timestamp. Equal timestamps are not provenance; resolve
+    # only mutually-unique compatible pairs inside a bucket and never guess by
+    # list order when a relationship is ambiguous.
+    sidecar_by_exact_timestamp = {}
+    state_by_exact_timestamp = {}
+    for index, message in enumerate(sidecar):
+        if index in used_targets:
+            continue
+        role = _message_sidecar_role(message)
+        timestamp = _message_exact_timestamp(message)
+        _, valid = _state_db_row_identity_details(message)
+        if valid and role is not None and timestamp is not None:
+            sidecar_by_exact_timestamp.setdefault((role, timestamp), []).append(index)
+    for source_index, message in enumerate(state):
+        if source_index in used_sources:
+            continue
+        role = _message_sidecar_role(message)
+        timestamp = _message_exact_timestamp(message)
+        _, valid = _state_db_row_identity_details(message)
+        if valid and role is not None and timestamp is not None:
+            state_by_exact_timestamp.setdefault((role, timestamp), []).append((source_index, message))
+    for key, state_rows in state_by_exact_timestamp.items():
+        target_indexes = [
+            index
+            for index in sidecar_by_exact_timestamp.get(key, ())
+            if index not in used_targets
+        ]
+        if not target_indexes:
+            # No exact-timestamp target is not an ambiguity: leave the source
+            # row available for the documented same-second/content tiers.
+            continue
+
+        # Use the same mutually-unique compatibility graph as the same-second
+        # tier. This resolves distinct content pairs inside a repeated exact
+        # timestamp bucket without guessing by list order.
+        compatible_targets_by_source = {}
+        for source_index, source_message in state_rows:
+            if source_index in used_sources:
+                continue
+            compatible_targets_by_source[source_index] = [
+                target_index
+                for target_index in target_indexes
+                if target_index not in used_targets
+                and _message_identity_compatible(sidecar[target_index], source_message)
+                and (
+                    _session_message_api_content_key(sidecar[target_index]) is None
+                    or _session_message_api_content_key(sidecar[target_index])
+                    == _session_message_api_content_key(source_message)
+                )
+            ]
+
+        compatible_sources_by_target = {}
+        for source_index, candidates in compatible_targets_by_source.items():
+            for target_index in candidates:
+                compatible_sources_by_target.setdefault(target_index, []).append(source_index)
+
+        for source_index, candidates in compatible_targets_by_source.items():
+            if len(candidates) != 1:
+                continue
+            target_index = candidates[0]
+            if len(compatible_sources_by_target.get(target_index, ())) != 1:
+                continue
+            if source_index in used_sources or target_index in used_targets:
+                continue
+            used_sources.add(source_index)
+            used_targets.add(target_index)
+            _copy_api_content_sidecar(sidecar[target_index], state[source_index])
+
+    # 4. Same-second sub-second drift. The sidecar JSON and state.db can
+    # record one logical turn with different fractional timestamps while still
+    # landing in the same wall-clock second. Do not use the truncated second
+    # as identity by itself: require a unique *compatible* visible row on each
+    # side. A candidate is compatible only when role + normalized visible
+    # content and all available provenance agree, and neither side carries a
+    # conflicting provider sidecar.
+    # This keeps repeated text, contradictory provenance, and conflicting wire
+    # payloads fail-closed instead of guessing by list order.
+    sidecar_by_same_second = {}
+    state_by_same_second = {}
+    for index, message in enumerate(sidecar):
+        if index in used_targets:
+            continue
+        role = _message_sidecar_role(message)
+        timestamp = _message_exact_timestamp(message)
+        row_id, valid = _state_db_row_identity_details(message)
+        if valid and role is not None and timestamp is not None:
+            second = _normalized_message_timestamp_for_key(timestamp)
+            sidecar_by_same_second.setdefault((role, second), []).append(index)
+    for source_index, message in enumerate(state):
+        if source_index in used_sources:
+            continue
+        role = _message_sidecar_role(message)
+        timestamp = _message_exact_timestamp(message)
+        row_id, valid = _state_db_row_identity_details(message)
+        if valid and role is not None and timestamp is not None:
+            second = _normalized_message_timestamp_for_key(timestamp)
+            state_by_same_second.setdefault((role, second), []).append(source_index)
+
+    for key, state_indexes in state_by_same_second.items():
+        target_indexes = [
+            index
+            for index in sidecar_by_same_second.get(key, ())
+            if index not in used_targets
+        ]
+        if not target_indexes:
+            continue
+
+        compatible_targets_by_source = {}
+        for source_index in state_indexes:
+            if source_index in used_sources:
+                continue
+            source_message = state[source_index]
+            compatible_targets_by_source[source_index] = [
+                target_index
+                for target_index in target_indexes
+                if target_index not in used_targets
+                and _message_identity_compatible(sidecar[target_index], source_message)
+                and (
+                    _session_message_api_content_key(sidecar[target_index]) is None
+                    or _session_message_api_content_key(sidecar[target_index])
+                    == _session_message_api_content_key(source_message)
+                )
+            ]
+
+        compatible_sources_by_target = {}
+        for source_index, candidates in compatible_targets_by_source.items():
+            for target_index in candidates:
+                compatible_sources_by_target.setdefault(target_index, []).append(source_index)
+
+        # Accept only one-to-one compatible pairs.  This deliberately does not
+        # zip a bucket: repeated same-text rows or equal timestamps remain
+        # distinct until durable provenance resolves them.
+        for source_index, candidates in compatible_targets_by_source.items():
+            if len(candidates) != 1:
+                continue
+            target_index = candidates[0]
+            if len(compatible_sources_by_target.get(target_index, ())) != 1:
+                continue
+            if source_index in used_sources or target_index in used_targets:
+                continue
+            used_sources.add(source_index)
+            used_targets.add(target_index)
+            _copy_api_content_sidecar(sidecar[target_index], state[source_index])
+
+    # 5. Role + visible-content fallback tolerates mixed timestamp metadata,
+    # but only when the candidate relationship is mutually unique. Repeated
+    # identical rows have no principled ordering and remain unattached.
+    sidecar_by_visible_content = {}
+    state_by_visible_content = {}
+    for index, message in enumerate(sidecar):
+        if index in used_targets:
+            continue
+        role = _message_sidecar_role(message)
+        _, row_id_valid = _state_db_row_identity_details(message)
+        _, stable_id_valid = _stable_message_identity_details(message)
+        _, timestamp_valid = _message_exact_timestamp_details(message)
+        if (
+            row_id_valid
+            and stable_id_valid
+            and timestamp_valid
+            and role is not None
+            and _message_visible_content_key(message) is not None
+        ):
+            sidecar_by_visible_content.setdefault(_message_visible_content_key(message), []).append(index)
+    for source_index, message in enumerate(state):
+        if source_index in used_sources:
+            continue
+        role = _message_sidecar_role(message)
+        _, row_id_valid = _state_db_row_identity_details(message)
+        _, stable_id_valid = _stable_message_identity_details(message)
+        _, timestamp_valid = _message_exact_timestamp_details(message)
+        if (
+            row_id_valid
+            and stable_id_valid
+            and timestamp_valid
+            and role is not None
+            and _message_visible_content_key(message) is not None
+        ):
+            state_by_visible_content.setdefault(_message_visible_content_key(message), []).append(source_index)
+    for visible_key, target_indexes in sidecar_by_visible_content.items():
+        source_indexes = state_by_visible_content.get(visible_key, [])
+        if len(target_indexes) != 1 or len(source_indexes) != 1:
+            continue
+        target_index = target_indexes[0]
+        source_index = source_indexes[0]
+        target_timestamp, target_timestamp_valid = _message_exact_timestamp_details(
+            sidecar[target_index]
+        )
+        source_timestamp, source_timestamp_valid = _message_exact_timestamp_details(
+            state[source_index]
+        )
+        if (
+            not target_timestamp_valid
+            or not source_timestamp_valid
+            or (target_timestamp is not None and source_timestamp is not None)
+        ):
+            continue
+        if not _message_identity_compatible(sidecar[target_index], state[source_index]):
+            continue
+        target_api_content = _session_message_api_content_key(sidecar[target_index])
+        source_api_content = _session_message_api_content_key(state[source_index])
+        if (
+            target_api_content is not None
+            and source_api_content is not None
+            and target_api_content != source_api_content
+        ):
+            continue
+        used_targets.add(target_index)
+        used_sources.add(source_index)
+        _copy_api_content_sidecar(sidecar[target_index], state[source_index])
+
+
 def _session_message_dedup_key(msg: dict):
     """Like _session_message_merge_key but preserves full-precision timestamp.
 
@@ -8398,13 +9028,15 @@ def _session_message_dedup_key(msg: dict):
         return ("non_dict", repr(msg))
     message_identity = msg.get("id") or msg.get("message_id")
     if message_identity:
-        return ("message_id", str(message_identity))
+        return _session_message_key_with_sidecar(
+            ("message_id", str(message_identity)), msg
+        )
     # Include tool_calls in the key so assistant messages that carry
     # different tool invocations (but identical empty content/timestamp)
     # are never collapsed into one.  (#3346 regression)
     _tc = msg.get("tool_calls")
     _tc_key = json.dumps(_tc, sort_keys=True, default=str) if _tc else ""
-    return (
+    return _session_message_key_with_sidecar((
         "legacy",
         str(msg.get("role") or ""),
         str(msg.get("content") or ""),
@@ -8412,7 +9044,7 @@ def _session_message_dedup_key(msg: dict):
         str(msg.get("tool_call_id") or ""),
         str(msg.get("tool_name") or msg.get("name") or ""),
         _tc_key,
-    )
+    ), msg)
 
 
 def _normalized_session_message_content(msg: dict) -> str:
@@ -8450,12 +9082,12 @@ def _session_message_content_key(msg: dict):
         content = " ".join(
             _strip_workspace_prefix(content, include_legacy=True).split()
         )
-    return (
+    return _session_message_key_with_sidecar((
         role,
         content,
         str(msg.get("tool_call_id") or ""),
         str(msg.get("tool_name") or msg.get("name") or ""),
-    )
+    ), msg)
 
 
 def _session_message_visible_key(msg: dict):
@@ -8467,11 +9099,11 @@ def _session_message_visible_key(msg: dict):
     # ("assistant", "") and the merge treats state.db rows as replays.
     _tc = msg.get("tool_calls")
     _tc_key = json.dumps(_tc, sort_keys=True, default=str) if _tc else ""
-    return (
+    return _session_message_key_with_sidecar((
         str(msg.get("role") or ""),
         _normalized_session_message_content(msg),
         _tc_key,
-    )
+    ), msg)
 
 
 def _build_visible_duplicate_lookup(visible_keys: set[tuple]) -> dict:
@@ -8496,6 +9128,7 @@ def _matching_visible_duplicate(visible_key: tuple, visible_keys: set[tuple], lo
         return visible_key
     role = visible_key[0]
     content = visible_key[1] if len(visible_key) > 1 else ""
+    sidecar = visible_key[3] if len(visible_key) > 3 else None
     if not content:
         return None
     if lookup is None:
@@ -8505,7 +9138,8 @@ def _matching_visible_duplicate(visible_key: tuple, visible_keys: set[tuple], lo
     for existing_key in lookup.get("by_role", {}).get(role, []):
         existing_role = existing_key[0]
         existing_content = existing_key[1] if len(existing_key) > 1 else ""
-        if role != existing_role or not existing_content:
+        existing_sidecar = existing_key[3] if len(existing_key) > 3 else None
+        if role != existing_role or sidecar != existing_sidecar or not existing_content:
             continue
         # Exact visible-key equality was checked above. For very large payloads
         # (tool logs / request dumps), Python-in substring and fuzzy-token
@@ -8582,6 +9216,12 @@ def state_db_delta_after_context(sidecar_context: list, state_messages: list) ->
     if not sidecar_context or not state_messages:
         return state_messages
 
+    # Attach trusted state.db sidecars before computing context keys.  A missing
+    # sidecar is not a wildcard: once attached, it participates in every
+    # duplicate/prefix decision below and distinct provider bytes remain
+    # distinct rows.
+    _reconcile_api_content_sidecars(sidecar_context, state_messages)
+
     # Recovered interrupted turns are special: the visible interruption marker
     # is synthetic, so the recovered user turn should still count as a mirrored
     # prefix when it is the actual aligned prefix row.
@@ -8836,6 +9476,46 @@ def merge_session_messages_append_only(
     """
     sidecar_messages = list(sidecar_messages or [])
     state_messages = list(state_messages or [])
+    _reconcile_api_content_sidecars(sidecar_messages, state_messages)
+    # The reconciler's quarantine sets are invocation-local. Mirror the
+    # identity-bucket guards here because this append-only merge has its own
+    # row-id fast path that must not re-admit a row isolated above.
+    sidecar_row_id_counts = collections.Counter()
+    state_row_id_counts = collections.Counter()
+    for message in sidecar_messages:
+        row_id, row_id_valid = _state_db_row_identity_details(message)
+        if row_id_valid and row_id is not None:
+            sidecar_row_id_counts[row_id] += 1
+    for message in state_messages:
+        row_id, row_id_valid = _state_db_row_identity_details(message)
+        if row_id_valid and row_id is not None:
+            state_row_id_counts[row_id] += 1
+
+    def _row_id_fast_path_allowed(existing, incoming) -> bool:
+        if not isinstance(existing, dict) or not isinstance(incoming, dict):
+            return False
+        existing_row_id, existing_row_id_valid = _state_db_row_identity_details(existing)
+        incoming_row_id, incoming_row_id_valid = _state_db_row_identity_details(incoming)
+        if (
+            not existing_row_id_valid
+            or not incoming_row_id_valid
+            or existing_row_id is None
+            or existing_row_id != incoming_row_id
+        ):
+            return False
+        if (
+            sidecar_row_id_counts.get(existing_row_id, 0) > 1
+            or state_row_id_counts.get(existing_row_id, 0) > 1
+        ):
+            return False
+        _, existing_timestamp_valid = _message_exact_timestamp_details(existing)
+        _, incoming_timestamp_valid = _message_exact_timestamp_details(incoming)
+        return (
+            existing_timestamp_valid
+            and incoming_timestamp_valid
+            and _message_identity_compatible(existing, incoming)
+        )
+
     # Per-invocation cache keyed by message identity. Sidecar/state message objects
     # are retained for this call, and this function does not mutate key-defining
     # fields before each helper call.
@@ -8977,6 +9657,8 @@ def merge_session_messages_append_only(
     merged_by_message_key = {}
     merged_by_dedup_key = {}
     merged_by_visible_key = {}
+    merged_by_row_id = {}
+    ambiguous_row_ids = set()
     max_sidecar_timestamp = None
 
     def _remember_merged_message(message):
@@ -8985,6 +9667,12 @@ def merge_session_messages_append_only(
         merged_by_message_key.setdefault(_cached_message_key(message, "merge"), message)
         merged_by_dedup_key.setdefault(_cached_message_key(message, "dedup"), message)
         merged_by_visible_key.setdefault(_cached_message_key(message, "visible"), message)
+        row_id, row_id_valid = _state_db_row_identity_details(message)
+        if row_id_valid and row_id is not None:
+            if row_id in merged_by_row_id:
+                ambiguous_row_ids.add(row_id)
+            else:
+                merged_by_row_id[row_id] = message
 
     for msg in sidecar_messages:
         timestamp = _message_timestamp_as_float(msg)
@@ -9051,6 +9739,32 @@ def merge_session_messages_append_only(
             # are caught by the dedup guard (#3346).
             seen_dedup_keys.add(dedup_key)
             continue
+        row_id, row_id_valid = _state_db_row_identity_details(msg)
+        row_id_sidecar_conflict = False
+        existing = (
+            merged_by_row_id.get(row_id)
+            if row_id_valid and row_id is not None
+            else None
+        )
+        if (
+            row_id_valid
+            and row_id is not None
+            and existing is not None
+            and row_id not in ambiguous_row_ids
+            and _row_id_fast_path_allowed(existing, msg)
+        ):
+            existing_api_content = _session_message_api_content_key(existing)
+            incoming_api_content = _session_message_api_content_key(msg)
+            row_id_sidecar_conflict = (
+                existing_api_content is not None
+                and incoming_api_content is not None
+                and existing_api_content != incoming_api_content
+            )
+            if not row_id_sidecar_conflict:
+                if existing_api_content is None and incoming_api_content is not None:
+                    _copy_api_content_sidecar(existing, msg)
+                _merge_session_display_metadata(existing, msg)
+                continue
         # Skip rows ABOVE the watermark only while the sidecar has NOT advanced
         # past the watermark. Because Session.save() no longer auto-clears the
         # watermark, an unconditional `timestamp > watermark` skip would become
@@ -9185,6 +9899,7 @@ def merge_session_messages_append_only(
             and max_sidecar_timestamp is not None
             and timestamp is not None
             and timestamp <= max_sidecar_timestamp
+            and not row_id_sidecar_conflict
         ):
             # When a truncation watermark is active and the sidecar holds only
             # the edited user checkpoint, state.db may contain an assistant/tool
diff --git a/api/routes.py b/api/routes.py
index 6143b39d..1b902280 100644
--- a/api/routes.py
+++ b/api/routes.py
@@ -2891,6 +2891,8 @@ from api.helpers import (
     _security_headers,
     _sanitize_error,
     redact_session_data,
+    public_session_projection,
+    strip_public_internal_fields,
     _redact_text,
     _CLIENT_DISCONNECT_ERRORS,
 )
@@ -9623,6 +9625,7 @@ from api.models import (
     get_state_db_session_message_keys_before_timestamp,
     get_state_db_session_summary,
     merge_session_messages_append_only,
+    _reconcile_api_content_sidecars,
     _enrich_sidebar_lineage_metadata,
     _active_stream_ids,
     _evict_sessions_over_cap,
@@ -13283,7 +13286,7 @@ def handle_get(handler, parsed) -> bool:
             }
             attach_todo_state(sess, msgs)
             sess = _merge_cli_sidebar_metadata(sess, cli_meta)
-            return j(handler, {"session": redact_session_data(sess)})
+            return j(handler, {"session": public_session_projection(sess)})
 
     if parsed.path == "/api/session/lineage/report":
         sid = parse_qs(parsed.query).get("session_id", [""])[0]
@@ -14411,7 +14414,9 @@ def handle_post(handler, parsed) -> bool:
                     "created_at": share_meta["share_created_at"],
                     "updated_at": share_meta["share_updated_at"],
                 },
-                "session": response_session.compact() | {"messages": response_session.messages},
+                "session": public_session_projection(
+                    response_session.compact() | {"messages": response_session.messages}
+                ),
             },
         )
 
@@ -14450,7 +14455,9 @@ def handle_post(handler, parsed) -> bool:
             handler,
             {
                 "ok": True,
-                "session": response_session.compact() | {"messages": response_session.messages},
+                "session": public_session_projection(
+                    response_session.compact() | {"messages": response_session.messages}
+                ),
             },
         )
 
@@ -14580,7 +14587,9 @@ def handle_post(handler, parsed) -> bool:
                 profile=getattr(s, "profile", None),
                 session_id=getattr(s, "session_id", None),
             )
-        payload = {"session": s.compact() | {"messages": s.messages}}
+        payload = {
+            "session": public_session_projection(s.compact() | {"messages": s.messages})
+        }
         if worktree_skipped:
             # Config-default worktree was skipped (non-git workspace); tell the
             # client the session is plain so the UI doesn't assume isolation.
@@ -14677,7 +14686,14 @@ def handle_post(handler, parsed) -> bool:
                 session_id=getattr(copied_session, "session_id", None),
             )
 
-            return j(handler, {"session": copied_session.compact() | {"messages": copied_session.messages}})
+            return j(
+                handler,
+                {
+                    "session": public_session_projection(
+                        copied_session.compact() | {"messages": copied_session.messages}
+                    )
+                },
+            )
         except Exception as e:
             return bad(handler, str(e))
 
@@ -15076,7 +15092,10 @@ def handle_post(handler, parsed) -> bool:
             except Exception:
                 logger.debug("Failed to close workspace terminal after workspace update")
         set_last_workspace(new_ws)
-        return j(handler, {"session": s.compact() | {"messages": s.messages}})
+        return j(
+            handler,
+            {"session": public_session_projection(s.compact() | {"messages": s.messages})},
+        )
     if parsed.path == "/api/session/worktree/remove":
         sid = body.get("session_id", "")
         if not sid or not isinstance(sid, str) or not sid.strip():
@@ -15353,7 +15372,13 @@ def handle_post(handler, parsed) -> bool:
         from api.config import _evict_session_agent
         _evict_session_agent(body["session_id"])
         return j(
-            handler, {"ok": True, "session": s.compact() | {"messages": s.messages}}
+            handler,
+            {
+                "ok": True,
+                "session": public_session_projection(
+                    s.compact() | {"messages": s.messages}
+                ),
+            },
         )
 
     if parsed.path == "/api/session/branch":
@@ -15455,11 +15480,21 @@ def handle_post(handler, parsed) -> bool:
         from api.session_ops import truncate_context_for_display_keep
 
         fork_keep = keep_count if keep_count is not None else len(source_messages)
-        forked_context = truncate_context_for_display_keep(
-            getattr(source, "context_messages", None),
-            source_messages,
-            fork_keep,
+        forked_context = copy.deepcopy(
+            truncate_context_for_display_keep(
+                getattr(source, "context_messages", None),
+                source_messages,
+                fork_keep,
+            )
         )
+        # `truncate_context_for_display_keep` aligns rows by visible display
+        # identity but intentionally does not carry provider replay sidecars.
+        # Reconcile only the retained prefix before constructing the branch so
+        # its persisted model context receives the same conservative
+        # api_content bytes as the forked display messages.  The helper mutates
+        # the freshly aligned context copy; the source session and
+        # forked_messages ownership remain untouched.
+        _reconcile_api_content_sidecars(forked_context, forked_messages)
         branch = Session(
             workspace=source.workspace,
             model=source.model,
@@ -17028,7 +17063,10 @@ def _handle_session_export(handler, parsed):
     active_profile = get_active_profile_name()
     if not _profiles_match(getattr(s, "profile", None), active_profile):
         return bad(handler, "Session not found", 404)
-    safe = redact_session_data(s.__dict__)
+    # ``public_session_projection`` supersedes the narrower
+    # ``redact_session_data`` path so export context_messages uses the same
+    # alias-stripping boundary as the visible transcript.
+    safe = public_session_projection(s.__dict__)
     qs = parse_qs(parsed.query)
     fmt = qs.get("format", ["json"])[0].lower()
     if fmt == "html":
@@ -17694,6 +17732,29 @@ def _runner_event_payload(entry: dict):
     return entry
 
 
+def _project_runner_event_payload(payload):
+    """Strip internal replay fields (api_content, row-id aliases) from a runner
+    SSE payload before it is relayed to the browser.
+
+    Runner-backed SSE relays adapter payloads verbatim; a terminal event can
+    carry a full ``session`` object (with per-message ``api_content`` sidecars)
+    or be session/message-shaped itself. Neither must reach a client, so route
+    the transcript-bearing shapes through the same public projection every other
+    session emitter uses. Non-session payloads pass through unchanged.
+    """
+    if not isinstance(payload, dict):
+        return payload
+    # Terminal events wrap the session under a "session" key.
+    if isinstance(payload.get("session"), dict):
+        projected = dict(payload)
+        projected["session"] = public_session_projection(payload["session"])
+        return projected
+    # Payload is itself session/message-shaped (has a messages transcript).
+    if "messages" in payload or "context_messages" in payload:
+        return public_session_projection(payload)
+    return payload
+
+
 def _runner_event_id(run_id: str, entry: dict) -> str | None:
     event_id = entry.get("event_id") or entry.get("id")
     if event_id:
@@ -17740,7 +17801,7 @@ def _stream_runner_run_events(handler, run_id: str, cursor: str | None = None) -
                 if not isinstance(entry, dict):
                     continue
                 event = _runner_event_name(entry)
-                _sse_with_id(handler, event, _runner_event_payload(entry), _runner_event_id(run_id, entry))
+                _sse_with_id(handler, event, _project_runner_event_payload(_runner_event_payload(entry)), _runner_event_id(run_id, entry))
                 emitted = True
                 if event in SSE_RELAY_CLOSE_EVENTS:
                     terminal = True
@@ -22793,7 +22854,7 @@ def _handle_chat_sync(handler, body):
                 _merge_display_messages_after_agent_result,
                 _restore_display_reasoning_metadata,
                 _restore_reasoning_metadata,
-                _sanitize_messages_for_api,
+                _sanitize_messages_for_agent,
                 _compact_session_image_parts_for_persistence,
                 _context_messages_for_new_turn,
                 _workspace_context_prefix,
@@ -22824,7 +22885,7 @@ def _handle_chat_sync(handler, body):
             result = agent.run_conversation(
                 user_message=workspace_ctx + msg,
                 system_message=workspace_system_msg,
-                conversation_history=_sanitize_messages_for_api(
+                conversation_history=_sanitize_messages_for_agent(
                     _previous_context_messages,
                     cfg=get_config(),
                     effective_model=_model,
@@ -22906,7 +22967,7 @@ def _handle_chat_sync(handler, body):
         {
             "answer": result.get("final_response") or "",
             "status": "done" if result.get("completed", True) else "partial",
-            "session": s.compact() | {"messages": s.messages},
+            "session": public_session_projection(s.compact() | {"messages": s.messages}),
             "result": {k: v for k, v in result.items() if k != "messages"},
         },
     )
@@ -26080,6 +26141,11 @@ def _normalize_message_for_import_refresh(message: object) -> object:
     normalized = dict(message)
     normalized.pop("timestamp", None)
     normalized.pop("_ts", None)
+    # These are WebUI/Agent replay bookkeeping aliases at the message's top
+    # level.  Strip only those exact keys; nested business payloads are opaque
+    # and must remain part of the semantic import comparison.
+    for key in ("api_content", "_state_db_row_id", "_db_row_id", "state_db_row_id"):
+        normalized.pop(key, None)
     return normalized
 
 
@@ -26238,19 +26304,21 @@ def _handle_session_import_cli(handler, body):
         return j(
             handler,
             {
-                "session": existing.compact()
-                | {
-                    "messages": existing.messages,
-                    "is_cli_session": (False if _existing_is_sa else True),
-                    # Greptile #4911 follow-up: read read_only from
-                    # the persisted Session, NOT from cli_meta.  This
-                    # refresh path is for an already-WebUI-owned
-                    # session; the WebUI's persisted view is the
-                    # source of truth for the response, not the
-                    # foreign store's current value.  (Mirrors the
-                    # GET /api/session fix.)
-                    "read_only": bool(getattr(existing, "read_only", False)),
-                },
+                "session": public_session_projection(
+                    existing.compact()
+                    | {
+                        "messages": existing.messages,
+                        "is_cli_session": (False if _existing_is_sa else True),
+                        # Greptile #4911 follow-up: read read_only from
+                        # the persisted Session, NOT from cli_meta.  This
+                        # refresh path is for an already-WebUI-owned
+                        # session; the WebUI's persisted view is the
+                        # source of truth for the response, not the
+                        # foreign store's current value.  (Mirrors the
+                        # GET /api/session fix.)
+                        "read_only": bool(getattr(existing, "read_only", False)),
+                    }
+                ),
                 "imported": False,
             },
         )
@@ -26334,7 +26402,13 @@ def _handle_session_import_cli(handler, body):
             "messages": msgs,
             "tool_calls": [],
         }
-        return j(handler, {"session": session_payload, "imported": False})
+        return j(
+            handler,
+            {
+                "session": public_session_projection(session_payload),
+                "imported": False,
+            },
+        )
 
     s = import_cli_session(
         sid,
@@ -26379,11 +26453,13 @@ def _handle_session_import_cli(handler, body):
     return j(
         handler,
         {
-            "session": s.compact()
-            | {
-                "messages": msgs,
-                "is_cli_session": True,
-            },
+            "session": public_session_projection(
+                s.compact()
+                | {
+                    "messages": msgs,
+                    "is_cli_session": True,
+                }
+            ),
             "imported": True,
         },
     )
@@ -26393,9 +26469,15 @@ def _handle_session_import(handler, body):
     """Import a session from a JSON export. Creates a new session with a new ID."""
     if not body or not isinstance(body, dict):
         return bad(handler, "Request body must be a JSON object")
-    messages = body.get("messages")
+    messages = strip_public_internal_fields(
+        body.get("messages"),
+        message_records=True,
+    )
     if not isinstance(messages, list):
         return bad(handler, 'JSON must contain a "messages" array')
+    raw_tool_calls = body.get("tool_calls", [])
+    if not isinstance(raw_tool_calls, list):
+        return bad(handler, 'JSON "tool_calls" must be an array')
     title = body.get("title", "Imported session")
     try:
         workspace = str(resolve_trusted_workspace(body.get("workspace", str(DEFAULT_WORKSPACE))))
@@ -26407,7 +26489,7 @@ def _handle_session_import(handler, body):
         workspace=workspace,
         model=model,
         messages=messages,
-        tool_calls=body.get("tool_calls", []),
+        tool_calls=strip_public_internal_fields(raw_tool_calls),
         profile=get_active_profile_name(),
     )
     s.pinned = body.get("pinned", False)
@@ -26417,7 +26499,13 @@ def _handle_session_import(handler, body):
         _evict_sessions_over_cap()  # #4765: safe LRU eviction (never active/unsaved)
     s.save()
     publish_session_list_changed("session_import")
-    return j(handler, {"ok": True, "session": s.compact() | {"messages": s.messages}})
+    return j(
+        handler,
+        {
+            "ok": True,
+            "session": public_session_projection(s.compact() | {"messages": s.messages}),
+        },
+    )
 
 
 def _mask_secrets(obj):
diff --git a/api/streaming.py b/api/streaming.py
index 0e607ff0..ddc25de4 100644
--- a/api/streaming.py
+++ b/api/streaming.py
@@ -51,7 +51,11 @@ from api.config import (
     _main_model_request_overrides,
     PROCESS_SESSION_INDEX, PROCESS_SESSION_INDEX_LOCK,
 )
-from api.helpers import redact_session_data, _redact_text
+from api.helpers import (
+    redact_session_data,
+    scrub_internal_replay_fields,
+    _redact_text,
+)
 from api.compression_anchor import is_context_compression_marker, visible_messages_for_anchor
 from api.compression_recovery import stamp_compression_exhausted_recovery
 from api.metering import meter
@@ -129,6 +133,13 @@ def _redacted_session_payload_with_full_messages(session, *, tool_calls=None) ->
         return None
 
 
+def _ephemeral_session_payload(session_id: str, messages) -> dict:
+    """Project the non-persistent ``/btw`` terminal session for public SSE."""
+    return redact_session_data(
+        {'session_id': session_id, 'messages': messages if isinstance(messages, list) else []}
+    )
+
+
 def _cancel_event_payload(
     message: str = "Cancelled by user",
     *,
@@ -4975,6 +4986,7 @@ def _sanitize_messages_for_api(
     effective_model: str | None = None,
     effective_provider: str | None = None,
     effective_base_url: str | None = None,
+    preserve_api_content: bool = False,
     requested_provider: str = "",
 ):
     """Return a deep copy of messages with only API-safe fields.
@@ -4995,7 +5007,18 @@ def _sanitize_messages_for_api(
     remaining replay gap where an older native image in the saved transcript kept
     causing 400s on every later text-only turn (#2297).
     """
-    strip_native_images = cfg is not None and _resolve_image_input_mode(cfg, effective_provider or "", effective_model or "", requested_provider=requested_provider) == "text"
+    strip_native_images = cfg is not None and _resolve_image_input_mode(
+        cfg,
+        effective_provider or "",
+        effective_model or "",
+        requested_provider=requested_provider,
+    ) == "text"
+    allowed_keys = _API_SAFE_MSG_KEYS
+    if preserve_api_content:
+        # ``api_content`` is an internal Agent replay sidecar.  It is opt-in
+        # here because direct provider/compression projections must continue to
+        # reject unknown bookkeeping fields.
+        allowed_keys = _API_SAFE_MSG_KEYS | {"api_content"}
     # First pass: collect all tool_call_ids declared by assistant messages.
     # Handles both OpenAI ('id') and Anthropic ('call_id') field names.
     valid_tool_call_ids: set = set()
@@ -5041,7 +5064,16 @@ def _sanitize_messages_for_api(
             if not tid or tid not in valid_tool_call_ids:
                 # Orphaned tool result — skip to avoid 400 from strict providers.
                 continue
-        sanitized = {k: v for k, v in msg.items() if k in _API_SAFE_MSG_KEYS}
+        sanitized = {k: v for k, v in msg.items() if k in allowed_keys}
+        sanitized = scrub_internal_replay_fields(
+            [sanitized],
+            preserve_message_api_content=preserve_api_content,
+            message_records=True,
+        )[0]
+        if sanitized.get("role") not in {"user", "assistant"}:
+            sanitized.pop("api_content", None)
+        elif not isinstance(sanitized.get("api_content"), str) or not sanitized.get("api_content"):
+            sanitized.pop("api_content", None)
         # Drop empty tool_calls — strict providers (DeepSeek, newer OpenAI)
         # reject tool_calls: [] with HTTP 400 even when no orphaned calls exist.
         if 'tool_calls' in sanitized and not sanitized['tool_calls']:
@@ -5125,6 +5157,33 @@ def _sanitize_messages_for_api(
     return final
 
 
+def _sanitize_messages_for_agent(
+    messages,
+    *,
+    cfg: dict = None,
+    effective_model: str | None = None,
+    effective_provider: str | None = None,
+    effective_base_url: str | None = None,
+    requested_provider: str = "",
+):
+    """Build the internal Agent replay projection with ``api_content`` intact.
+
+    ``api_content`` is a durable provider-facing sidecar, not a direct-provider
+    payload field.  Keep this opt-in at one named boundary so every Agent
+    history call uses the same contract while ordinary API/compression callers
+    continue to use the default-stripping sanitizer.
+    """
+    return _sanitize_messages_for_api(
+        messages,
+        cfg=cfg,
+        effective_model=effective_model,
+        effective_provider=effective_provider,
+        effective_base_url=effective_base_url,
+        preserve_api_content=True,
+        requested_provider=requested_provider,
+    )
+
+
 def _api_safe_message_positions(messages):
     """Return [(original_index, sanitized_message)] for API-safe messages."""
     valid_tool_call_ids: set = set()
@@ -5157,6 +5216,10 @@ def _api_safe_message_positions(messages):
             if not tid or tid not in valid_tool_call_ids:
                 continue
         sanitized = {k: v for k, v in msg.items() if k in _API_SAFE_MSG_KEYS}
+        sanitized = scrub_internal_replay_fields(
+            [sanitized],
+            message_records=True,
+        )[0]
         if 'tool_calls' in sanitized and not sanitized['tool_calls']:
             del sanitized['tool_calls']
         if is_recovered:
@@ -5241,7 +5304,11 @@ def _deduplicate_context_messages(messages):
         if _is_compressed_context_tool_result_summary_message(msg) and not msg.get('tool_call_id'):
             deduped.append(msg)
             continue
-        key = _message_identity(msg)
+        # Context ownership is provider-facing: two rows with identical visible
+        # text but different durable ``api_content`` sidecars are distinct turns.
+        # Keep the display identity unchanged so ordinary transcript dedup still
+        # collapses the same visible row.
+        key = _message_replay_key(msg)
         if isinstance(msg, dict) and msg.get('role') == 'user' and key is not None:
             user_exact_key = (
                 key,
@@ -5620,11 +5687,12 @@ def _message_identity(msg):
     )
 
 
-def _messages_have_prefix(messages, prefix):
+def _messages_have_prefix(messages, prefix, *, key_fn=None):
+    key_fn = key_fn or _message_identity
     if len(messages or []) < len(prefix or []):
         return False
     for idx, expected in enumerate(prefix or []):
-        if _message_identity((messages or [])[idx]) != _message_identity(expected):
+        if key_fn((messages or [])[idx]) != key_fn(expected):
             return False
     return True
 
@@ -5632,16 +5700,30 @@ def _messages_have_prefix(messages, prefix):
 def _message_replay_key(msg):
     """Return a stable comparison key for replay/overlap de-duplication."""
     identity = _message_identity(msg)
+    # ``api_content`` is a provider-facing replay sidecar.  It must participate
+    # in context/replay overlap identity or two same-visible turns can collapse
+    # before the Agent sees the original wire bytes.  Keep synthetic/adjacent
+    # partial collapse on the existing identity path; partial rows are display
+    # bookkeeping rather than durable provider turns.
+    raw_sidecar = (
+        msg.get("api_content")
+        if isinstance(msg, dict) and not msg.get("_partial")
+        else None
+    )
+    sidecar = raw_sidecar if isinstance(raw_sidecar, str) and raw_sidecar else None
     if identity is not None:
+        if sidecar is not None:
+            return (*identity, sidecar)
         return identity
     if not isinstance(msg, dict):
         return None
-    return (
+    key = (
         str(msg.get('role') or ''),
         _message_text(msg.get('content', '')),
         str(msg.get('tool_call_id') or ''),
         json.dumps(msg.get('tool_calls') or [], sort_keys=True, ensure_ascii=False),
     )
+    return (*key, sidecar) if sidecar is not None else key
 
 
 def _strip_replayed_prefix(existing_messages, candidates):
@@ -5674,6 +5756,20 @@ def _looks_like_replayed_session_arc_summary(previous_msg, candidate_msg):
         return False
     if previous_msg.get('role') != candidate_msg.get('role'):
         return False
+    previous_api_content = previous_msg.get("api_content")
+    candidate_api_content = candidate_msg.get("api_content")
+    normalized_previous_api_content = (
+        " ".join(previous_api_content.split())
+        if isinstance(previous_api_content, str) and previous_api_content
+        else None
+    )
+    normalized_candidate_api_content = (
+        " ".join(candidate_api_content.split())
+        if isinstance(candidate_api_content, str) and candidate_api_content
+        else None
+    )
+    if normalized_previous_api_content != normalized_candidate_api_content:
+        return False
     previous_text = " ".join(_message_text(previous_msg.get('content', '')).split())
     candidate_text = " ".join(_message_text(candidate_msg.get('content', '')).split())
     if len(previous_text) < 2000 or len(candidate_text) < 2000:
@@ -5730,7 +5826,11 @@ def _dedupe_replayed_context_messages(previous_context, result_messages, msg_tex
     if not previous_context or not result_messages:
         return result_messages
     previous_user_tail = _stale_user_tail_candidate(_last_user_row(previous_context))
-    if not _messages_have_prefix(result_messages, previous_context):
+    if not _messages_have_prefix(
+        result_messages,
+        previous_context,
+        key_fn=_message_replay_key,
+    ):
         # Agent-side role-sequence repair can replace the last prior user row
         # with a repaired current-user row. In that shape the result no longer
         # has `previous_context` as an exact prefix, but it should still be
@@ -5739,7 +5839,11 @@ def _dedupe_replayed_context_messages(previous_context, result_messages, msg_tex
             msg_text
             and len(previous_context) >= 1
             and len(result_messages) >= len(previous_context)
-            and _messages_have_prefix(result_messages, previous_context[:-1])
+            and _messages_have_prefix(
+                result_messages,
+                previous_context[:-1],
+                key_fn=_message_replay_key,
+            )
         ):
             boundary_idx = len(previous_context) - 1
             boundary_row = result_messages[boundary_idx]
@@ -5758,6 +5862,11 @@ def _dedupe_replayed_context_messages(previous_context, result_messages, msg_tex
                     # history in previous_context untouched.
                     cleaned_boundary = copy.deepcopy(boundary_row)
                     cleaned_boundary['content'] = msg_text
+                    # The stale row's provider-facing payload describes the
+                    # discarded prefix + current turn, not the rewritten
+                    # visible turn.  Do not let those obsolete bytes survive
+                    # into Agent context.
+                    cleaned_boundary.pop('api_content', None)
                     candidates = [cleaned_boundary] + result_messages[boundary_idx + 1:]
                 else:
                     candidates = result_messages[boundary_idx:]
@@ -6099,6 +6208,10 @@ def _strip_stale_user_merge_from_messages(
         ):
             cleaned = copy.deepcopy(msg) if isinstance(msg, dict) else {'role': 'user', 'content': msg_text}
             cleaned['content'] = msg_text
+            # The stale row's provider-facing payload describes the discarded
+            # merged prefix, so it is invalid once the visible content is
+            # rewritten to the submitted turn.
+            cleaned.pop('api_content', None)
             out.append(cleaned)
         else:
             out.append(msg)
@@ -9761,7 +9874,7 @@ def _run_agent_streaming(
             _run_conversation_kwargs = dict(
                 user_message=user_message,
                 system_message=workspace_system_msg,
-                conversation_history=_sanitize_messages_for_api(
+                conversation_history=_sanitize_messages_for_agent(
                     _previous_context_messages,
                     cfg=_cfg,
                     effective_model=resolved_model,
@@ -9852,8 +9965,15 @@ def _run_agent_streaming(
                     if isinstance(_m, dict) and _m.get('role') == 'assistant':
                         _answer = str(_m.get('content', ''))
                         break
+                # /btw is intentionally non-persistent, but its terminal SSE
+                # payload is still public output.  Project the ephemeral
+                # session before enqueueing it so raw Agent ``api_content`` or
+                # provenance aliases cannot cross the wire.
+                _ephemeral_session = _ephemeral_session_payload(
+                    session_id, result.get('messages', [])
+                )
                 put('done', {
-                    'session': {'session_id': session_id, 'messages': result.get('messages', [])},
+                    'session': _ephemeral_session,
                     'usage': {'input_tokens': 0, 'output_tokens': 0},
                     'ephemeral': True,
                     'answer': _answer,
@@ -10257,7 +10377,7 @@ def _run_agent_streaming(
                                 _heal_kwargs = dict(
                                     user_message=user_message,
                                     system_message=workspace_system_msg,
-                                    conversation_history=_sanitize_messages_for_api(
+                                    conversation_history=_sanitize_messages_for_agent(
                                         _previous_context_messages,
                                         cfg=_cfg,
                                         effective_model=resolved_model,
@@ -11497,7 +11617,7 @@ def _run_agent_streaming(
                         _heal_kwargs2 = dict(
                             user_message=user_message,
                             system_message=workspace_system_msg,
-                            conversation_history=_sanitize_messages_for_api(
+                            conversation_history=_sanitize_messages_for_agent(
                                 _previous_context_messages,
                                 cfg=_cfg,
                                 effective_model=resolved_model,
diff --git a/tests/test_chat_start_claim_cli_session.py b/tests/test_chat_start_claim_cli_session.py
index 81fd6fb3..a907e70c 100644
--- a/tests/test_chat_start_claim_cli_session.py
+++ b/tests/test_chat_start_claim_cli_session.py
@@ -153,7 +153,7 @@ def test_get_session_route_uses_shared_synthesiser():
     src = ROUTES_PY.read_text(encoding="utf-8")
     # Find the /api/session GET block (not /api/sessions).
     block = re.search(
-        r'if parsed\.path == "/api/session":.*?return j\(handler, \{"session": redact_session_data\(sess\)\}\)',
+        r'if parsed\.path == "/api/session":.*?return j\(handler, \{"session": public_session_projection\(sess\)\}\)',
         src,
         re.DOTALL,
     )
@@ -185,7 +185,7 @@ def test_get_session_preserves_cli_read_only_flag():
     sets to True for BOTH the explicit AND the source-refused
     refusal paths."""
     block = re.search(
-        r'if parsed\.path == "/api/session":.*?return j\(handler, \{"session": redact_session_data\(sess\)\}\)\)?',
+        r'if parsed\.path == "/api/session":.*?return j\(handler, \{"session": public_session_projection\(sess\)\}\)\)?',
         ROUTES_PY.read_text(encoding="utf-8"),
         re.DOTALL,
     )
diff --git a/tests/test_issue6751_api_content_agent_replay.py b/tests/test_issue6751_api_content_agent_replay.py
new file mode 100644
index 00000000..e78850cd
--- /dev/null
+++ b/tests/test_issue6751_api_content_agent_replay.py
@@ -0,0 +1,1797 @@
+"""Regression coverage for #6751 api_content replay preservation.
+
+The fixture models the durable Agent ``messages`` rows that carry a clean
+visible ``content`` value plus the exact provider-facing ``api_content``
+sidecar.  The two user rows intentionally render the same text while keeping
+different timestamps and sidecars so a content-only reconciliation can be
+observed if it crosses the rows.
+"""
+
+import sqlite3
+import sys
+import json
+from types import SimpleNamespace
+
+import pytest
+
+
+pytestmark = pytest.mark.requires_agent_modules
+
+
+def _write_state_db(path, sid):
+    conn = sqlite3.connect(path)
+    conn.execute(
+        """
+        CREATE TABLE messages (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            session_id TEXT,
+            role TEXT,
+            content TEXT,
+            timestamp REAL,
+            api_content TEXT
+        )
+        """
+    )
+    conn.executemany(
+        """
+        INSERT INTO messages (session_id, role, content, timestamp, api_content)
+        VALUES (?, ?, ?, ?, ?)
+        """,
+        [
+            (sid, "user", "same visible text", 100.0, "[Workspace::v1: /one]\\nsame visible text"),
+            (sid, "user", "same visible text", 200.0, "[Workspace::v1: /two]\\nsame visible text"),
+        ],
+    )
+    conn.commit()
+    conn.close()
+
+
+def test_issue6751_state_db_sidecars_survive_reconcile_and_internal_agent_projection(
+    monkeypatch, tmp_path
+):
+    """Distinct same-text rows retain their own sidecar through every boundary."""
+    import api.models as models
+    from api.streaming import _sanitize_messages_for_agent
+
+    sid = "issue6751-replay"
+    db_path = tmp_path / "state.db"
+    _write_state_db(db_path, sid)
+    monkeypatch.setattr(models, "_active_state_db_path", lambda: db_path)
+
+    state_messages = models.get_state_db_session_messages(sid)
+    assert [message["api_content"] for message in state_messages] == [
+        "[Workspace::v1: /one]\\nsame visible text",
+        "[Workspace::v1: /two]\\nsame visible text",
+    ]
+    assert [message["_state_db_row_id"] for message in state_messages] == [1, 2]
+
+    sidecar_messages = [
+        {"role": "user", "content": "same visible text", "timestamp": 100.0},
+        {"role": "user", "content": "same visible text", "timestamp": 200.0},
+    ]
+    merged = models.merge_session_messages_append_only(sidecar_messages, state_messages)
+    assert [message.get("api_content") for message in merged] == [
+        "[Workspace::v1: /one]\\nsame visible text",
+        "[Workspace::v1: /two]\\nsame visible text",
+    ]
+
+    internal_history = _sanitize_messages_for_agent(merged)
+    assert [message.get("api_content") for message in internal_history] == [
+        "[Workspace::v1: /one]\\nsame visible text",
+        "[Workspace::v1: /two]\\nsame visible text",
+    ]
+
+
+def test_issue6751_reconciliation_prioritizes_row_id_and_preserves_existing_sidecar():
+    from api.models import merge_session_messages_append_only
+
+    sidecar = [
+        {
+            "role": "user",
+            "content": "same visible text",
+            "timestamp": 200.0,
+            "_state_db_row_id": 2,
+            "api_content": "already-authoritative",
+        },
+        {
+            "role": "user",
+            "content": "same visible text",
+            "timestamp": 100.0,
+            "_state_db_row_id": 1,
+        },
+    ]
+    state = [
+        {
+            "role": "user",
+            "content": "same visible text",
+            "timestamp": 100.0,
+            "_state_db_row_id": 1,
+            "api_content": "[Workspace::v1: /one]\\nsame visible text",
+        },
+        {
+            "role": "user",
+            "content": "same visible text",
+            "timestamp": 200.0,
+            "_state_db_row_id": 2,
+            "api_content": "[Workspace::v1: /two]\\nsame visible text",
+        },
+    ]
+
+    merged = merge_session_messages_append_only(sidecar, state)
+
+    assert [message["api_content"] for message in merged] == [
+        "already-authoritative",
+        "[Workspace::v1: /one]\\nsame visible text",
+        "[Workspace::v1: /two]\\nsame visible text",
+    ]
+
+
+def test_issue6751_reconciliation_consumes_each_state_source_once_on_timestamp_collision():
+    from api.models import merge_session_messages_append_only
+
+    sidecar = [
+        {
+            "role": "user",
+            "content": "same visible text",
+            "timestamp": 100.0,
+            "_state_db_row_id": 1,
+        },
+        {"role": "user", "content": "same visible text", "timestamp": 100.0},
+    ]
+    state = [
+        {
+            "role": "user",
+            "content": "same visible text",
+            "timestamp": 100.0,
+            "_state_db_row_id": 1,
+            "api_content": "wire-one",
+        },
+        {
+            "role": "user",
+            "content": "same visible text",
+            "timestamp": 100.0,
+            "api_content": "wire-two",
+        },
+    ]
+
+    merged = merge_session_messages_append_only(sidecar, state)
+
+    assert [message.get("api_content") for message in merged] == ["wire-one", "wire-two"]
+
+
+def test_issue6751_reconciliation_fails_closed_when_row_id_roles_disagree():
+    from api.models import merge_session_messages_append_only
+
+    sidecar = [
+        {
+            "role": "user",
+            "content": "same visible text",
+            "timestamp": 100.0,
+            "_state_db_row_id": 1,
+        }
+    ]
+    state = [
+        {
+            "role": "assistant",
+            "content": "same visible text",
+            "timestamp": 100.0,
+            "_state_db_row_id": 1,
+            "api_content": "assistant-wire",
+        }
+    ]
+
+    merged = merge_session_messages_append_only(sidecar, state)
+
+    assert "api_content" not in merged[0]
+
+
+def test_issue6751_duplicate_durable_row_ids_fail_closed():
+    """A row id shared by multiple sidecar targets must never guess a target."""
+    from api.models import merge_session_messages_append_only
+
+    sidecar = [
+        {"role": "user", "content": "same", "_state_db_row_id": 7},
+        {"role": "user", "content": "same", "_state_db_row_id": 7},
+    ]
+    state = [
+        {
+            "role": "user",
+            "content": "same",
+            "_state_db_row_id": 7,
+            "api_content": "wire",
+        }
+    ]
+
+    merged = merge_session_messages_append_only(sidecar, state)
+
+    assert [message.get("api_content") for message in merged[:2]] == [None, None]
+
+
+def test_issue6751_conflicting_row_id_aliases_fail_closed():
+    """Contradictory private aliases are invalid provenance, not a fallback hint."""
+    from api.models import merge_session_messages_append_only
+
+    sidecar = [
+        {
+            "role": "assistant",
+            "content": "answer",
+            "_state_db_row_id": 7,
+            "_db_row_id": 8,
+        }
+    ]
+    state = [
+        {
+            "role": "assistant",
+            "content": "answer",
+            "_state_db_row_id": 7,
+            "api_content": "wire",
+        }
+    ]
+
+    merged = merge_session_messages_append_only(sidecar, state)
+
+    assert "api_content" not in merged[0]
+
+
+def test_issue6751_incompatible_visible_content_fails_closed():
+    """A matching durable id cannot override an incompatible visible turn."""
+    from api.models import merge_session_messages_append_only
+
+    sidecar = [
+        {"role": "user", "content": "visible one", "_state_db_row_id": 7}
+    ]
+    state = [
+        {
+            "role": "user",
+            "content": "visible two",
+            "_state_db_row_id": 7,
+            "api_content": "wire-two",
+        }
+    ]
+
+    merged = merge_session_messages_append_only(sidecar, state)
+
+    assert "api_content" not in merged[0]
+
+
+def test_issue6751_ambiguous_metadata_free_sequence_fails_closed():
+    from api.models import merge_session_messages_append_only
+
+    sidecar = [
+        {"role": "assistant", "content": "same answer"},
+        {"role": "assistant", "content": "same answer"},
+    ]
+    state = [
+        {"role": "assistant", "content": "same answer", "api_content": "wire-one"},
+        {"role": "assistant", "content": "same answer", "api_content": "wire-two"},
+    ]
+
+    merged = merge_session_messages_append_only(sidecar, state)
+
+    assert merged[:2] == sidecar
+    assert [message.get("api_content") for message in merged] == [
+        None,
+        None,
+        "wire-one",
+        "wire-two",
+    ]
+
+
+def test_issue6751_partial_metadata_bucket_uses_unique_role_content_pairs():
+    from api.models import merge_session_messages_append_only
+
+    sidecar = [
+        {"role": "assistant", "content": "first"},
+        {"role": "assistant", "content": "second"},
+    ]
+    state = [
+        {"role": "assistant", "content": "first", "timestamp": 10.0, "api_content": "wire-first"},
+        {"role": "assistant", "content": "second", "api_content": "wire-second"},
+    ]
+
+    merged = merge_session_messages_append_only(sidecar, state)
+
+    assert merged[:2] == sidecar
+    assert [message.get("api_content") for message in merged] == [
+        "wire-first",
+        "wire-second",
+    ]
+
+
+def test_issue6751_state_db_reader_preserves_shape_without_api_content(monkeypatch, tmp_path):
+    import api.models as models
+
+    db_path = tmp_path / "state.db"
+    conn = sqlite3.connect(db_path)
+    conn.execute(
+        "CREATE TABLE messages (id INTEGER PRIMARY KEY, session_id TEXT, role TEXT, content TEXT, timestamp REAL)"
+    )
+    conn.execute(
+        "INSERT INTO messages VALUES (7, ?, ?, ?, ?)",
+        ("shape-session", "user", "ordinary", 1.0),
+    )
+    conn.commit()
+    conn.close()
+    monkeypatch.setattr(models, "_active_state_db_path", lambda: db_path)
+
+    assert models.get_state_db_session_messages("shape-session") == [
+        {"role": "user", "content": "ordinary", "timestamp": 1.0}
+    ]
+
+
+def test_issue6751_public_message_projection_strips_internal_replay_fields(monkeypatch):
+    import api.helpers as helpers
+    import api.config as config
+
+    monkeypatch.setattr(config, "load_settings", lambda: {"api_redact_enabled": False})
+    safe = helpers.redact_session_data(
+        {
+            "messages": [
+                {
+                    "role": "user",
+                    "content": "visible",
+                    "api_content": "wire",
+                    "_state_db_row_id": 7,
+                    "_db_row_id": 8,
+                    "state_db_row_id": 9,
+                }
+            ]
+        }
+    )
+
+    assert safe["messages"] == [{"role": "user", "content": "visible"}]
+
+
+def test_issue6751_public_session_projection_strips_context_aliases(monkeypatch):
+    import api.config as config
+    from api.helpers import public_session_projection
+
+    monkeypatch.setattr(config, "load_settings", lambda: {"api_redact_enabled": False})
+    safe = public_session_projection(
+        {
+            "messages": [{"role": "user", "content": "visible", "api_content": "wire"}],
+            "context_messages": [
+                {
+                    "role": "assistant",
+                    "content": "answer",
+                    "api_content": "provider bytes",
+                    "_state_db_row_id": 7,
+                    "_db_row_id": 8,
+                    "state_db_row_id": 9,
+                }
+            ],
+            "runtime_journal_snapshot": {
+                "messages": [{"role": "assistant", "api_content": "nested"}]
+            },
+        }
+    )
+
+    assert safe["messages"] == [{"role": "user", "content": "visible"}]
+    assert safe["context_messages"] == [{"role": "assistant", "content": "answer"}]
+    assert safe["runtime_journal_snapshot"]["messages"] == [{"role": "assistant"}]
+
+
+def test_issue6751_public_projection_preserves_non_message_alias_keys(monkeypatch):
+    import api.config as config
+    from api.helpers import public_session_projection
+
+    monkeypatch.setattr(config, "load_settings", lambda: {"api_redact_enabled": False})
+    safe = public_session_projection(
+        {
+            "messages": [
+                {
+                    "role": "user",
+                    "content": {
+                        "api_content": "user-json-field",
+                        "_db_row_id": 41,
+                    },
+                }
+            ],
+            "tool_calls": [
+                {
+                    "name": "tool",
+                    "api_content": "tool-payload-field",
+                    "_db_row_id": 42,
+                }
+            ],
+        }
+    )
+
+    assert safe["messages"][0]["content"] == {
+        "api_content": "user-json-field",
+        "_db_row_id": 41,
+    }
+    assert "api_content" not in safe["tool_calls"][0]
+    assert "_db_row_id" not in safe["tool_calls"][0]
+
+
+def test_issue6751_provider_projection_still_strips_api_content_by_default():
+    from api.streaming import _sanitize_messages_for_api
+
+    sanitized = _sanitize_messages_for_api(
+        [{
+            "role": "user",
+            "content": "same visible text",
+            "api_content": "[Workspace::v1: /tmp]\\nsame visible text",
+        }]
+    )
+
+    assert sanitized == [{"role": "user", "content": "same visible text"}]
+
+
+def test_issue6751_sync_chat_agent_receives_original_api_content_bytes(monkeypatch, tmp_path):
+    """The production sync route must hand the Agent the original wire text."""
+    import api.config as config
+    import api.models as models
+    import api.routes as routes
+
+    state_dir = tmp_path / "state"
+    session_dir = state_dir / "sessions"
+    session_dir.mkdir(parents=True)
+    monkeypatch.setattr(models, "SESSION_DIR", session_dir)
+    monkeypatch.setattr(models, "SESSION_INDEX_FILE", state_dir / "session_index.json")
+    monkeypatch.setattr(routes, "SESSION_INDEX_FILE", state_dir / "session_index.json")
+    monkeypatch.setattr(routes, "get_session", models.get_session)
+    monkeypatch.setattr(routes, "title_from", models.title_from)
+    monkeypatch.setattr(config, "get_config", lambda: {"model": "test-model", "provider": "test-provider"})
+    monkeypatch.setattr(routes, "get_config", lambda: {"model": "test-model", "provider": "test-provider"})
+    monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda value: tmp_path)
+    monkeypatch.setattr(routes, "load_settings", lambda: {})
+    monkeypatch.setattr(routes, "_resolve_cli_toolsets", lambda: [])
+
+    original_wire = "[Workspace::v1: /original-workspace]\nturn-two request"
+    context = [
+        {
+            "role": "user",
+            "content": "turn-two request",
+            "timestamp": 2.0,
+            "api_content": original_wire,
+        },
+        {"role": "assistant", "content": "turn-two answer", "timestamp": 3.0},
+    ]
+    session = models.Session(
+        session_id="issue6751_sync_agent",
+        workspace=str(tmp_path),
+        messages=list(context),
+        context_messages=list(context),
+        model="test-model",
+        model_provider="test-provider",
+    )
+    session.save(touch_updated_at=False)
+
+    captured = {}
+
+    class FakeAgent:
+        def __init__(self, **_kwargs):
+            pass
+
+        def run_conversation(self, **kwargs):
+            captured["history"] = list(kwargs.get("conversation_history") or [])
+            return {
+                "messages": captured["history"]
+                + [
+                    {"role": "user", "content": kwargs["persist_user_message"]},
+                    {"role": "assistant", "content": "ok"},
+                ],
+                "final_response": "ok",
+                "completed": True,
+            }
+
+    monkeypatch.setitem(sys.modules, "run_agent", SimpleNamespace(AIAgent=FakeAgent))
+
+    class Handler:
+        def __init__(self):
+            self.headers = {}
+            self.status = None
+            self.wfile = SimpleNamespace(write=lambda _body: None)
+
+        def send_response(self, status):
+            self.status = status
+
+        def send_header(self, _name, _value):
+            return None
+
+        def end_headers(self):
+            return None
+
+    handler = Handler()
+    routes._handle_chat_sync(
+        handler,
+        {
+            "session_id": session.session_id,
+            "message": "follow-up",
+            "workspace": str(tmp_path),
+        },
+    )
+
+    assert handler.status == 200
+    assert [message.get("api_content") for message in captured["history"] if message.get("role") == "user"] == [
+        original_wire
+    ]
+
+
+def test_issue6751_json_import_strips_internal_aliases_before_persistence(monkeypatch, tmp_path):
+    """Caller-supplied provider bytes never become durable imported context."""
+    from collections import OrderedDict
+
+    import api.config as config
+    import api.models as models
+    import api.routes as routes
+
+    state_dir = tmp_path / "state"
+    session_dir = state_dir / "sessions"
+    session_dir.mkdir(parents=True)
+    sessions = OrderedDict()
+    monkeypatch.setattr(models, "SESSION_DIR", session_dir)
+    monkeypatch.setattr(models, "SESSION_INDEX_FILE", state_dir / "session_index.json")
+    monkeypatch.setattr(models, "SESSIONS", sessions)
+    monkeypatch.setattr(routes, "SESSIONS", sessions)
+    monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda value: tmp_path)
+    monkeypatch.setattr(routes, "get_active_profile_name", lambda: "default")
+    monkeypatch.setattr(routes, "publish_session_list_changed", lambda *args, **kwargs: None)
+    monkeypatch.setattr(config, "load_settings", lambda: {"api_redact_enabled": False})
+    captured = {}
+
+    def fake_json(handler, payload, status=200, **_kwargs):
+        captured["payload"] = payload
+        captured["status"] = status
+        return True
+
+    monkeypatch.setattr(routes, "j", fake_json)
+    routes._handle_session_import(
+        None,
+        {
+            "title": "imported",
+            "workspace": str(tmp_path),
+            "messages": [
+                {
+                    "role": "user",
+                    "content": {
+                        "visible": "visible",
+                        "api_content": "user-json-field",
+                        "_db_row_id": 41,
+                    },
+                    "api_content": "caller provider bytes",
+                    "_state_db_row_id": 1,
+                    "_db_row_id": 2,
+                    "state_db_row_id": 3,
+                }
+            ],
+            "context_messages": [
+                {"role": "assistant", "content": "answer", "api_content": "hidden"}
+            ],
+            "tool_calls": [
+                {
+                    "name": "tool",
+                    "api_content": "nested hidden",
+                    "_db_row_id": 3,
+                    "arguments": {
+                        "api_content": "tool-argument-field",
+                        "_db_row_id": 42,
+                    },
+                }
+            ],
+        },
+    )
+
+    assert captured["status"] == 200
+    imported = next(iter(sessions.values()))
+    assert imported.messages == [
+        {
+            "role": "user",
+            "content": {
+                "visible": "visible",
+                "api_content": "user-json-field",
+                "_db_row_id": 41,
+            },
+        }
+    ]
+    assert imported.context_messages == []
+    assert imported.tool_calls == [
+        {
+            "name": "tool",
+            "arguments": {
+                "api_content": "tool-argument-field",
+                "_db_row_id": 42,
+            },
+        }
+    ]
+    persisted = json.loads(imported.path.read_text(encoding="utf-8"))
+    aliases = ("api_content", "_state_db_row_id", "_db_row_id", "state_db_row_id")
+    assert all(alias not in persisted["messages"][0] for alias in aliases)
+    assert all(alias not in persisted["tool_calls"][0] for alias in aliases)
+    assert persisted["messages"][0]["content"]["api_content"] == "user-json-field"
+    assert persisted["messages"][0]["content"]["_db_row_id"] == 41
+    assert persisted["tool_calls"][0]["arguments"]["api_content"] == "tool-argument-field"
+    assert persisted["tool_calls"][0]["arguments"]["_db_row_id"] == 42
+
+
+def test_issue6751_json_import_nested_tool_calls_are_removed_at_agent_boundary(
+    monkeypatch, tmp_path
+):
+    """Import, reload, and the sync Agent boundary must not pass nested aliases."""
+    from collections import OrderedDict
+
+    import api.config as config
+    import api.models as models
+    import api.routes as routes
+
+    state_dir = tmp_path / "state"
+    session_dir = state_dir / "sessions"
+    session_dir.mkdir(parents=True)
+    sessions = OrderedDict()
+    monkeypatch.setattr(models, "SESSION_DIR", session_dir)
+    monkeypatch.setattr(models, "SESSION_INDEX_FILE", state_dir / "session_index.json")
+    monkeypatch.setattr(models, "SESSIONS", sessions)
+    monkeypatch.setattr(routes, "SESSION_INDEX_FILE", state_dir / "session_index.json")
+    monkeypatch.setattr(routes, "SESSIONS", sessions)
+    monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda value: tmp_path)
+    monkeypatch.setattr(routes, "get_active_profile_name", lambda: "default")
+    monkeypatch.setattr(routes, "publish_session_list_changed", lambda *args, **kwargs: None)
+    monkeypatch.setattr(config, "load_settings", lambda: {"api_redact_enabled": False})
+    monkeypatch.setattr(config, "get_config", lambda: {"model": "test-model", "provider": "test-provider"})
+    monkeypatch.setattr(routes, "get_config", lambda: {"model": "test-model", "provider": "test-provider"})
+    monkeypatch.setattr(routes, "_resolve_cli_toolsets", lambda: [])
+    captured_response = {}
+
+    def fake_json(handler, payload, status=200, **_kwargs):
+        captured_response["payload"] = payload
+        captured_response["status"] = status
+        return True
+
+    _real_routes_json = routes.j
+    monkeypatch.setattr(routes, "j", fake_json)
+    routes._handle_session_import(
+        None,
+        {
+            "title": "nested-import",
+            "workspace": str(tmp_path),
+            "messages": [
+                {
+                    "role": "assistant",
+                    "content": [
+                        {
+                            "type": "text",
+                            "text": "answer",
+                            "api_content": "ordinary content field",
+                        }
+                    ],
+                    "_state_db_row_id": 1,
+                    "tool_calls": [
+                        {
+                            "id": "call-1",
+                            "api_content": "nested provider bytes",
+                            "_db_row_id": 2,
+                            "args": {
+                                "messages": [
+                                    {"api_content": "import-business-payload"}
+                                ]
+                            },
+                            "function": {
+                                "name": "lookup",
+                                "arguments": '{"api_content":"ordinary argument field"}',
+                            },
+                        }
+                    ],
+                },
+                {
+                    "role": "tool",
+                    "tool_call_id": "call-1",
+                    "content": "lookup result",
+                },
+            ],
+            "tool_calls": [
+                {
+                    "id": "call-1",
+                    "api_content": "session provider bytes",
+                    "state_db_row_id": 3,
+                    "arguments": {"_db_row_id": 4},
+                }
+            ],
+        },
+    )
+
+    assert captured_response["status"] == 200
+    imported = next(iter(sessions.values()))
+    # Reload from the persisted sidecar to cover the import -> disk -> load path.
+    reloaded = models.Session.load(imported.session_id)
+    assert reloaded is not None
+    nested = reloaded.messages[0]
+    aliases = ("api_content", "_state_db_row_id", "_db_row_id", "state_db_row_id")
+    assert all(alias not in nested for alias in aliases)
+    assert all(alias not in nested["tool_calls"][0] for alias in aliases)
+    assert "api_content" not in nested["content"][0]
+    assert nested["tool_calls"][0]["function"]["arguments"] == (
+        '{"api_content":"ordinary argument field"}'
+    )
+    assert nested["tool_calls"][0]["args"]["messages"] == [
+        {"api_content": "import-business-payload"}
+    ]
+    assert reloaded.tool_calls == [{"id": "call-1", "arguments": {"_db_row_id": 4}}]
+
+    # Restore the real response writer for the production sync route below.
+    monkeypatch.setattr(routes, "j", _real_routes_json)
+
+    captured_agent = {}
+
+    class FakeAgent:
+        def __init__(self, **_kwargs):
+            pass
+
+        def run_conversation(self, **kwargs):
+            captured_agent["history"] = list(kwargs.get("conversation_history") or [])
+            return {
+                "messages": captured_agent["history"]
+                + [
+                    {"role": "user", "content": kwargs["persist_user_message"]},
+                    {"role": "assistant", "content": "ok"},
+                ],
+                "final_response": "ok",
+                "completed": True,
+            }
+
+    monkeypatch.setitem(sys.modules, "run_agent", SimpleNamespace(AIAgent=FakeAgent))
+    monkeypatch.setattr(routes, "get_session", lambda _sid: reloaded)
+
+    class Handler:
+        def __init__(self):
+            self.headers = {}
+            self.status = None
+            self.wfile = SimpleNamespace(write=lambda _body: None)
+
+        def send_response(self, status):
+            self.status = status
+
+        def send_header(self, _name, _value):
+            return None
+
+        def end_headers(self):
+            return None
+
+    handler = Handler()
+    routes._handle_chat_sync(
+        handler,
+        {
+            "session_id": reloaded.session_id,
+            "message": "follow-up",
+            "workspace": str(tmp_path),
+        },
+    )
+
+    assert handler.status == 200
+    agent_history_row = next(
+        message for message in captured_agent["history"] if message.get("role") == "assistant"
+    )
+    assert all(alias not in agent_history_row for alias in aliases)
+    assert all(alias not in agent_history_row["tool_calls"][0] for alias in aliases)
+
+
+def test_issue6751_json_import_rejects_non_list_session_tool_calls(monkeypatch, tmp_path):
+    """Session-level tool_calls are a list schema, not an arbitrary JSON value."""
+    from collections import OrderedDict
+
+    import api.config as config
+    import api.helpers as helpers
+    import api.models as models
+    import api.routes as routes
+
+    state_dir = tmp_path / "state"
+    session_dir = state_dir / "sessions"
+    session_dir.mkdir(parents=True)
+    sessions = OrderedDict()
+    monkeypatch.setattr(models, "SESSION_DIR", session_dir)
+    monkeypatch.setattr(models, "SESSION_INDEX_FILE", state_dir / "session_index.json")
+    monkeypatch.setattr(models, "SESSIONS", sessions)
+    monkeypatch.setattr(routes, "SESSION_INDEX_FILE", state_dir / "session_index.json")
+    monkeypatch.setattr(routes, "SESSIONS", sessions)
+    monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda value: tmp_path)
+    monkeypatch.setattr(config, "load_settings", lambda: {"api_redact_enabled": False})
+    captured = {}
+
+    def fake_json(handler, payload, status=200, **_kwargs):
+        captured["payload"] = payload
+        captured["status"] = status
+        return True
+
+    monkeypatch.setattr(routes, "j", fake_json)
+    monkeypatch.setattr(helpers, "j", fake_json)
+    routes._handle_session_import(
+        None,
+        {
+            "title": "invalid-tool-calls",
+            "workspace": str(tmp_path),
+            "messages": [],
+            "tool_calls": {"api_content": "must-not-be-a-session-list"},
+        },
+    )
+
+    assert captured["status"] == 400
+    assert sessions == OrderedDict()
+
+
+def test_issue6751_context_dedup_keeps_distinct_api_content_histories():
+    from api.streaming import _deduplicate_context_messages
+
+    history = [
+        {"role": "user", "content": "same visible", "api_content": "wire-one"},
+        {"role": "user", "content": "same visible", "api_content": "wire-two"},
+    ]
+
+    deduped = _deduplicate_context_messages(history)
+
+    assert [message["api_content"] for message in deduped] == ["wire-one", "wire-two"]
+
+
+def test_issue6751_context_dedup_ignores_malformed_api_content():
+    from api.streaming import _deduplicate_context_messages
+
+    history = [
+        {"role": "user", "content": "same visible", "api_content": {"bad": True}},
+        {"role": "user", "content": "same visible"},
+    ]
+
+    assert _deduplicate_context_messages(history) == history[:1]
+
+
+def test_issue6751_source_only_duplicate_durable_id_does_not_fallback_attach():
+    from api.models import merge_session_messages_append_only
+
+    sidecar = [{"role": "user", "content": "same", "timestamp": 100.0}]
+    state = [
+        {
+            "role": "user",
+            "content": "same",
+            "timestamp": 100.0,
+            "_state_db_row_id": 7,
+            "api_content": "wire-one",
+        },
+        {
+            "role": "user",
+            "content": "same",
+            "timestamp": 200.0,
+            "_state_db_row_id": 7,
+            "api_content": "wire-two",
+        },
+    ]
+
+    merged = merge_session_messages_append_only(sidecar, state)
+
+    assert merged[0].get("api_content") is None
+
+
+def test_issue6751_ephemeral_terminal_sse_projects_agent_messages(monkeypatch):
+    import api.config as config
+    from api.streaming import _ephemeral_session_payload
+
+    monkeypatch.setattr(config, "load_settings", lambda: {"api_redact_enabled": False})
+    payload = _ephemeral_session_payload(
+        "ephemeral-sid",
+        [
+            {
+                "role": "assistant",
+                "content": "visible",
+                "api_content": "raw provider bytes",
+                "_state_db_row_id": 7,
+                "_db_row_id": 8,
+                "state_db_row_id": 9,
+            }
+        ],
+    )
+
+    assert payload == {
+        "session_id": "ephemeral-sid",
+        "messages": [{"role": "assistant", "content": "visible"}],
+    }
+
+
+def test_issue6751_schema_scrubber_preserves_tool_argument_business_payload(monkeypatch):
+    """Only authoritative message/tool schemas are scrubbed; args are opaque."""
+    import api.config as config
+    from api.helpers import public_session_projection
+
+    monkeypatch.setattr(config, "load_settings", lambda: {"api_redact_enabled": False})
+    payload = public_session_projection(
+        {
+            "messages": [
+                {
+                    "role": "assistant",
+                    "content": [
+                        {
+                            "type": "text",
+                            "text": "visible",
+                            "api_content": "content-part-internal",
+                        }
+                    ],
+                    "api_content": "message-wire",
+                    "_state_db_row_id": 7,
+                    "tool_calls": [
+                        {
+                            "id": "call-1",
+                            "api_content": "tool-call-internal",
+                            "function": {
+                                "name": "lookup",
+                                "api_content": "function-internal",
+                                "arguments": '{"api_content":"opaque-argument"}',
+                            },
+                            "args": {
+                                "messages": [
+                                    {
+                                        "api_content": "business-payload",
+                                        "_db_row_id": 41,
+                                    }
+                                ]
+                            },
+                        }
+                    ],
+                }
+            ],
+            "runtime_journal_snapshot": {
+                "messages": [
+                    {
+                        "role": "assistant",
+                        "content": "snapshot-visible",
+                        "api_content": "snapshot-wire",
+                    }
+                ],
+                "tool_calls": [
+                    {
+                        "name": "lookup",
+                        "args": {
+                            "messages": [
+                                {
+                                    "api_content": "snapshot-business-payload",
+                                }
+                            ]
+                        },
+                    }
+                ],
+            },
+        }
+    )
+
+    message = payload["messages"][0]
+    assert "api_content" not in message
+    assert "_state_db_row_id" not in message
+    assert "api_content" not in message["content"][0]
+    tool_call = message["tool_calls"][0]
+    assert "api_content" not in tool_call
+    assert "api_content" not in tool_call["function"]
+    assert tool_call["function"]["arguments"] == '{"api_content":"opaque-argument"}'
+    assert tool_call["args"]["messages"][0] == {
+        "api_content": "business-payload",
+        "_db_row_id": 41,
+    }
+    assert payload["runtime_journal_snapshot"]["messages"] == [
+        {"role": "assistant", "content": "snapshot-visible"}
+    ]
+    assert payload["runtime_journal_snapshot"]["tool_calls"][0]["args"]["messages"][0] == {
+        "api_content": "snapshot-business-payload"
+    }
+
+
+def test_issue6751_agent_schema_scrubber_keeps_message_sidecar_only():
+    from api.streaming import _sanitize_messages_for_agent
+
+    messages = [
+        {
+            "role": "assistant",
+            "content": [
+                {"type": "text", "text": "visible", "api_content": "part-internal"}
+            ],
+            "api_content": "message-wire",
+            "tool_calls": [
+                {
+                    "id": "call-1",
+                    "api_content": "tool-call-internal",
+                    "function": {
+                        "name": "lookup",
+                        "api_content": "function-internal",
+                        "arguments": '{"api_content":"opaque-argument"}',
+                    },
+                    "args": {"messages": [{"api_content": "business-payload"}]},
+                }
+            ],
+        },
+        {"role": "tool", "tool_call_id": "call-1", "content": "result"},
+    ]
+
+    sanitized = _sanitize_messages_for_agent(messages)
+
+    assistant = sanitized[0]
+    assert assistant["api_content"] == "message-wire"
+    assert "api_content" not in assistant["content"][0]
+    tool_call = assistant["tool_calls"][0]
+    assert "api_content" not in tool_call
+    assert "api_content" not in tool_call["function"]
+    assert tool_call["function"]["arguments"] == '{"api_content":"opaque-argument"}'
+    assert tool_call["args"]["messages"] == [{"api_content": "business-payload"}]
+
+
+def test_issue6751_session_arc_fuzzy_replay_requires_equal_api_content():
+    from api.streaming import _looks_like_replayed_session_arc_summary
+
+    summary_a = {
+        "role": "user",
+        "content": "[Session Arc Summary]\n" + "same recovered context\n" * 260,
+        "api_content": "wire-one",
+    }
+    summary_b = dict(summary_a, api_content="wire-two")
+
+    assert not _looks_like_replayed_session_arc_summary(summary_a, summary_b)
+    assert _looks_like_replayed_session_arc_summary(summary_a, dict(summary_a))
+
+
+@pytest.mark.parametrize("bad_row_id", ["invalid-row-id", "nan", "inf", float("nan"), float("inf")])
+def test_issue6751_invalid_row_id_rejects_metadata_free_fallback(bad_row_id):
+    from api.models import merge_session_messages_append_only
+
+    sidecar = [{"role": "user", "content": "same"}]
+    state = [
+        {
+            "role": "user",
+            "content": "same",
+            "_state_db_row_id": bad_row_id,
+            "api_content": "wire",
+        }
+    ]
+
+    merged = merge_session_messages_append_only(sidecar, state)
+
+    assert merged[0].get("api_content") is None
+
+
+@pytest.mark.parametrize("bad_timestamp", ["invalid-time", "nan", "inf", float("nan"), float("inf")])
+def test_issue6751_invalid_timestamp_rejects_metadata_free_fallback(bad_timestamp):
+    from api.models import merge_session_messages_append_only
+
+    sidecar = [{"role": "user", "content": "same"}]
+    state = [
+        {
+            "role": "user",
+            "content": "same",
+            "timestamp": bad_timestamp,
+            "api_content": "wire",
+        }
+    ]
+
+    merged = merge_session_messages_append_only(sidecar, state)
+
+    assert merged[0].get("api_content") is None
+
+
+def test_issue6751_distinct_state_sidecars_are_not_deduplicated():
+    from api.models import merge_session_messages_append_only
+
+    merged = merge_session_messages_append_only(
+        [],
+        [
+            {"role": "user", "content": "same", "api_content": "wire-one"},
+            {"role": "user", "content": "same", "api_content": "wire-two"},
+        ],
+    )
+
+    assert [message["api_content"] for message in merged] == ["wire-one", "wire-two"]
+
+
+def test_issue6751_same_row_id_different_sidecars_remain_distinct():
+    """A durable row id cannot override distinct provider payload identity."""
+    from api.models import merge_session_messages_append_only
+
+    sidecar = [
+        {
+            "role": "user",
+            "content": "same visible",
+            "timestamp": 100.0,
+            "_state_db_row_id": 7,
+            "api_content": "wire-one",
+        }
+    ]
+    state = [
+        {
+            "role": "user",
+            "content": "same visible",
+            "timestamp": 100.0,
+            "_state_db_row_id": 7,
+            "api_content": "wire-two",
+        }
+    ]
+
+    merged = merge_session_messages_append_only(sidecar, state)
+
+    assert [message.get("api_content") for message in merged] == [
+        "wire-one",
+        "wire-two",
+    ]
+
+
+def test_issue6751_unique_same_second_timestamp_drift_reconciles_before_merge_identity():
+    """One logical turn with sub-second storage drift must not persist twice."""
+    from api.models import merge_session_messages_append_only
+
+    sidecar = [
+        {"role": "user", "content": "same visible", "timestamp": 100.1}
+    ]
+    state = [
+        {
+            "role": "user",
+            "content": "same visible",
+            "timestamp": 100.8,
+            "api_content": "provider wire",
+        }
+    ]
+
+    merged = merge_session_messages_append_only(sidecar, state)
+
+    assert len(merged) == 1
+    assert merged[0]["api_content"] == "provider wire"
+
+
+def test_issue6751_same_second_drift_with_repeated_text_remains_ambiguous():
+    """Sub-second proximity alone cannot choose between repeated user turns."""
+    from api.models import merge_session_messages_append_only
+
+    sidecar = [
+        {"role": "user", "content": "same visible", "timestamp": 100.1},
+        {"role": "user", "content": "same visible", "timestamp": 100.2},
+    ]
+    state = [
+        {
+            "role": "user",
+            "content": "same visible",
+            "timestamp": 100.8,
+            "api_content": "provider wire",
+        }
+    ]
+
+    merged = merge_session_messages_append_only(sidecar, state)
+
+    assert len(merged) == 3
+    assert [message.get("api_content") for message in merged[:2]] == [None, None]
+    assert merged[2]["api_content"] == "provider wire"
+
+
+def test_issue6751_stale_user_cleanup_drops_obsolete_api_content_on_both_paths():
+    """Rewritten visible content must never retain the superseded wire payload."""
+    from api.streaming import (
+        _dedupe_replayed_context_messages,
+        _strip_stale_user_merge_from_messages,
+    )
+
+    previous = [{"role": "user", "content": "old turn"}]
+    stale = {
+        "role": "user",
+        "content": "old turn\n\nnew turn",
+        "api_content": "obsolete provider bytes",
+    }
+
+    stripped = _strip_stale_user_merge_from_messages(
+        [stale],
+        "new turn",
+        "old turn",
+        previous_context=previous,
+    )
+    reconciled = _dedupe_replayed_context_messages(
+        previous,
+        [stale, {"role": "assistant", "content": "answer"}],
+        "new turn",
+    )
+
+    assert stripped[0]["content"] == "new turn"
+    assert "api_content" not in stripped[0]
+    rewritten = next(
+        message
+        for message in reconciled
+        if message.get("role") == "user" and message.get("content") == "new turn"
+    )
+    assert "api_content" not in rewritten
+
+
+def test_issue6751_cli_refresh_ignores_internal_aliases_and_appends_new_rows(monkeypatch):
+    """A pre-upgrade import remains a semantic prefix of enriched fresh rows."""
+    import api.routes as routes
+
+    existing = SimpleNamespace(
+        session_id="issue6751-cli-refresh",
+        profile="default",
+        messages=[{"role": "user", "content": "first", "timestamp": 1}],
+        source_tag="cli",
+        raw_source="cli",
+        session_source="cli",
+        source_label="CLI",
+        parent_session_id=None,
+        read_only=False,
+        compact=lambda: {"session_id": "issue6751-cli-refresh"},
+    )
+    saved = []
+    existing.save = lambda **kwargs: saved.append(kwargs)
+    fresh = [
+        {
+            "role": "user",
+            "content": "first",
+            "timestamp": 1.9,
+            "api_content": "provider wire",
+            "_state_db_row_id": 1,
+            "_db_row_id": 1,
+            "state_db_row_id": 1,
+        },
+        {"role": "assistant", "content": "new CLI reply", "timestamp": 2.0},
+    ]
+
+    monkeypatch.setattr(routes.Session, "load", lambda _sid: existing)
+    monkeypatch.setattr(routes, "_session_visible_to_active_profile", lambda *_args: True)
+    monkeypatch.setattr(routes, "_resolve_cli_import_metadata", lambda *_args, **_kwargs: {})
+    monkeypatch.setattr(routes, "get_cli_session_messages", lambda *_args, **_kwargs: fresh)
+    monkeypatch.setattr(routes, "_is_subagent_child_session_id", lambda _sid: False)
+    monkeypatch.setattr(routes, "publish_session_list_changed", lambda *_args, **_kwargs: None)
+    monkeypatch.setattr(routes, "public_session_projection", lambda payload: payload)
+    captured = {}
+    monkeypatch.setattr(
+        routes,
+        "j",
+        lambda _handler, payload, **_kwargs: captured.setdefault("payload", payload),
+    )
+
+    routes._handle_session_import_cli(None, {"session_id": existing.session_id})
+
+    assert existing.messages == fresh
+    assert saved == [{"touch_updated_at": False}]
+    assert captured["payload"]["imported"] is False
+
+
+def test_issue6751_branch_endpoint_reconciles_provider_sidecars_into_first_agent_context(
+    monkeypatch,
+):
+    """The branch must persist and execute from the same reconciled prefix."""
+    from collections import OrderedDict
+
+    import api.routes as routes
+
+    source_messages = [
+        {
+            "role": "user",
+            "content": "visible request",
+            "timestamp": 1.0,
+            "api_content": "[Workspace::v1: /original]\nvisible request",
+        },
+        {
+            "role": "assistant",
+            "content": "visible answer",
+            "timestamp": 2.0,
+            "api_content": "provider answer bytes",
+        },
+    ]
+    source = SimpleNamespace(
+        _branch_source_readonly=True,
+        session_id="issue6751-parent",
+        title="Parent",
+        workspace="/tmp/workspace",
+        model="test-model",
+        model_provider="test-provider",
+        profile="default",
+        messages=source_messages,
+        context_messages=[
+            {"role": "user", "content": "visible request", "timestamp": 1.0},
+            {"role": "assistant", "content": "visible answer", "timestamp": 2.0},
+        ],
+        project_id=None,
+        personality=None,
+        enabled_toolsets=None,
+        context_length=None,
+        threshold_tokens=None,
+        gateway_routing=None,
+        context_engine=None,
+        context_engine_state={},
+    )
+    captured = {}
+
+    class Branch:
+        def __init__(self, **kwargs):
+            self.__dict__.update(kwargs)
+            self.session_id = "issue6751-branch"
+            captured["branch"] = self
+
+        def save(self):
+            captured["saved"] = True
+
+    monkeypatch.setattr(routes, "_check_csrf", lambda _handler: True)
+    monkeypatch.setattr(routes, "read_body", lambda _handler: {"session_id": source.session_id})
+    monkeypatch.setattr(routes, "_guard_request_session_visibility", lambda *_args, **_kwargs: True)
+    monkeypatch.setattr(routes, "_load_branch_source_or_refuse", lambda *_args: source)
+    monkeypatch.setattr(routes, "_session_requires_cli_metadata_lookup", lambda _source: False)
+    monkeypatch.setattr(routes, "_is_messaging_session_record", lambda _record: False)
+    monkeypatch.setattr(routes, "Session", Branch)
+    monkeypatch.setattr(routes, "SESSIONS", OrderedDict())
+    monkeypatch.setattr(routes, "_evict_sessions_over_cap", lambda: None)
+    monkeypatch.setattr(routes, "publish_session_list_changed", lambda *_args, **_kwargs: None)
+    monkeypatch.setattr(routes, "j", lambda _handler, payload, **_kwargs: payload)
+    handler = SimpleNamespace(
+        client_address=("127.0.0.1", 12345),
+        headers={},
+        command="POST",
+        path="/api/session/branch",
+    )
+
+    routes.handle_post(
+        handler,
+        SimpleNamespace(path="/api/session/branch", query=""),
+    )
+
+    branch = captured["branch"]
+    assert captured["saved"] is True
+    assert branch.messages == source_messages
+    assert [message.get("api_content") for message in branch.context_messages] == [
+        "[Workspace::v1: /original]\nvisible request",
+        "provider answer bytes",
+    ]
+
+
+def test_issue6751_real_state_reader_unique_row_id_reconciles_timestamp_drift(
+    monkeypatch, tmp_path
+):
+    """The real state.db reader's unique row identity must prevent a replay duplicate."""
+    import api.models as models
+
+    sid = "issue6751-reader-row-id"
+    db_path = tmp_path / "state.db"
+    conn = sqlite3.connect(db_path)
+    conn.execute(
+        """
+        CREATE TABLE messages (
+            id INTEGER PRIMARY KEY,
+            session_id TEXT,
+            role TEXT,
+            content TEXT,
+            timestamp REAL,
+            api_content TEXT
+        )
+        """
+    )
+    conn.execute(
+        "INSERT INTO messages VALUES (?, ?, ?, ?, ?, ?)",
+        (7, sid, "user", "same visible", 100.8, "provider wire"),
+    )
+    conn.commit()
+    conn.close()
+    monkeypatch.setattr(models, "_active_state_db_path", lambda: db_path)
+
+    state_messages = models.get_state_db_session_messages(sid)
+    assert state_messages == [
+        {
+            "role": "user",
+            "content": "same visible",
+            "timestamp": 100.8,
+            "api_content": "provider wire",
+            "_state_db_row_id": 7,
+        }
+    ]
+
+    merged = models.merge_session_messages_append_only(
+        [
+            {
+                "role": "user",
+                "content": "same visible",
+                "timestamp": 100.1,
+                "id": "display-turn-1",
+            }
+        ],
+        state_messages,
+    )
+
+    assert len(merged) == 1
+    assert merged[0]["api_content"] == "provider wire"
+
+
+def test_issue6751_reconciliation_prefers_unique_stable_message_id_before_weaker_tiers():
+    from api.models import merge_session_messages_append_only
+
+    sidecar = [
+        {
+            "role": "assistant",
+            "content": "same answer",
+            "timestamp": 10.0,
+            "id": "turn-a",
+        },
+        {
+            "role": "assistant",
+            "content": "same answer",
+            "timestamp": 20.0,
+            "id": "turn-b",
+        },
+    ]
+    state = [
+        {
+            "role": "assistant",
+            "content": "same answer",
+            "id": "turn-b",
+            "api_content": "wire-b",
+        },
+        {
+            "role": "assistant",
+            "content": "same answer",
+            "id": "turn-a",
+            "api_content": "wire-a",
+        },
+    ]
+
+    merged = merge_session_messages_append_only(sidecar, state)
+
+    assert len(merged) == 2
+    assert [(message["id"], message.get("api_content")) for message in merged] == [
+        ("turn-a", "wire-a"),
+        ("turn-b", "wire-b"),
+    ]
+
+
+def test_issue6751_nan_timestamps_cannot_be_reaccepted_by_row_id_tier():
+    """Invalid timestamp provenance remains isolated even when row ids agree."""
+    from api.models import merge_session_messages_append_only
+
+    sidecar = [
+        {
+            "role": "user",
+            "content": "same request",
+            "timestamp": float("nan"),
+            "_state_db_row_id": 42,
+        }
+    ]
+    state = [
+        {
+            "role": "user",
+            "content": "same request",
+            "timestamp": float("nan"),
+            "_state_db_row_id": 42,
+            "api_content": "must-not-attach",
+        }
+    ]
+
+    merged = merge_session_messages_append_only(sidecar, state)
+
+    assert merged[0].get("api_content") is None
+
+
+def test_issue6751_unique_row_id_resolves_duplicate_stable_bucket():
+    """A unique durable row pair outranks an otherwise ambiguous stable id."""
+    from api.models import merge_session_messages_append_only
+
+    sidecar = [
+        {
+            "role": "user",
+            "content": "same request",
+            "id": "duplicate-turn",
+            "_state_db_row_id": 51,
+        },
+        {
+            "role": "user",
+            "content": "same request",
+            "id": "duplicate-turn",
+            "_state_db_row_id": 52,
+        },
+    ]
+    state = [
+        {
+            "role": "user",
+            "content": "same request",
+            "id": "duplicate-turn",
+            "_state_db_row_id": 51,
+            "api_content": "row-51-wire",
+        }
+    ]
+
+    merged = merge_session_messages_append_only(sidecar, state)
+
+    assert len(merged) == 2
+    assert [message.get("api_content") for message in merged] == ["row-51-wire", None]
+
+
+def test_issue6751_row_id_identity_conflict_is_quarantined_before_fallback():
+    """A row-id match cannot override contradictory stable message identity."""
+    from api.models import merge_session_messages_append_only
+
+    sidecar = [
+        {
+            "role": "assistant",
+            "content": "same answer",
+            "timestamp": 100.0,
+            "id": "turn-a",
+            "_state_db_row_id": 7,
+        },
+        {
+            "role": "assistant",
+            "content": "same answer",
+            "timestamp": 200.0,
+        },
+    ]
+    state = [
+        {
+            "role": "assistant",
+            "content": "same answer",
+            "timestamp": 200.0,
+            "id": "turn-b",
+            "_state_db_row_id": 7,
+            "api_content": "turn-b-wire",
+        }
+    ]
+
+    merged = merge_session_messages_append_only(sidecar, state)
+
+    assert [message.get("api_content") for message in merged[:2]] == [None, None]
+
+
+def test_issue6751_row_id_sidecar_conflict_cannot_fall_through_by_timestamp():
+    """Conflicting authoritative payloads quarantine both row-id candidates."""
+    from api.models import merge_session_messages_append_only
+
+    sidecar = [
+        {
+            "role": "user",
+            "content": "same request",
+            "timestamp": 100.0,
+            "_state_db_row_id": 7,
+            "api_content": "existing-wire",
+        },
+        {
+            "role": "user",
+            "content": "same request",
+            "timestamp": 200.0,
+        },
+    ]
+    state = [
+        {
+            "role": "user",
+            "content": "same request",
+            "timestamp": 200.0,
+            "_state_db_row_id": 7,
+            "api_content": "conflicting-wire",
+        }
+    ]
+
+    merged = merge_session_messages_append_only(sidecar, state)
+
+    assert [message.get("api_content") for message in merged[:2]] == [
+        "existing-wire",
+        None,
+    ]
+
+
+def test_issue6751_exact_timestamp_bucket_matches_mutually_unique_content_pairs():
+    """Equal timestamps do not hide unique compatible message pairs."""
+    from api.models import merge_session_messages_append_only
+
+    sidecar = [
+        {"role": "assistant", "content": "answer one", "timestamp": 300.0},
+        {"role": "assistant", "content": "answer two", "timestamp": 300.0},
+    ]
+    state = [
+        {
+            "role": "assistant",
+            "content": "answer two",
+            "timestamp": 300.0,
+            "api_content": "wire-two",
+        },
+        {
+            "role": "assistant",
+            "content": "answer one",
+            "timestamp": 300.0,
+            "api_content": "wire-one",
+        },
+    ]
+
+    merged = merge_session_messages_append_only(sidecar, state)
+
+    assert len(merged) == 2
+    assert [message.get("api_content") for message in merged] == [
+        "wire-one",
+        "wire-two",
+    ]
+
+
+def test_issue6751_incompatible_exact_timestamp_pair_keeps_mixed_fallback_open():
+    """An incompatible exact-time row must not consume a later unique match."""
+    from api.models import merge_session_messages_append_only
+
+    sidecar = [
+        {"role": "assistant", "content": "different answer", "timestamp": 400.0},
+        {"role": "assistant", "content": "matching answer"},
+    ]
+    state = [
+        {
+            "role": "assistant",
+            "content": "matching answer",
+            "timestamp": 400.0,
+            "api_content": "matching-wire",
+        }
+    ]
+
+    merged = merge_session_messages_append_only(sidecar, state)
+
+    assert len(merged) == 2
+    assert merged[0].get("api_content") is None
+    assert merged[1]["api_content"] == "matching-wire"
+
+
+def test_issue6751_conflicting_stable_aliases_cannot_be_reaccepted_by_row_id_tier():
+    """Contradictory stable aliases stay isolated from weaker identity tiers."""
+    from api.models import merge_session_messages_append_only
+
+    sidecar = [
+        {
+            "role": "assistant",
+            "content": "same answer",
+            "id": "stable-a",
+            "message_id": "stable-b",
+            "_state_db_row_id": 43,
+        }
+    ]
+    state = [
+        {
+            "role": "assistant",
+            "content": "same answer",
+            "_state_db_row_id": 43,
+            "api_content": "must-not-attach",
+        }
+    ]
+
+    merged = merge_session_messages_append_only(sidecar, state)
+
+    assert merged[0].get("api_content") is None
+
+
+@pytest.mark.parametrize(
+    ("sidecar_timestamp", "state_timestamp"),
+    [(0, 1), (1, 0)],
+)
+def test_issue6751_conflicting_present_timestamps_block_role_content_fallback(
+    sidecar_timestamp, state_timestamp
+):
+    """Two present timestamps that miss time tiers must fail closed."""
+    from api.models import merge_session_messages_append_only
+
+    sidecar = [
+        {
+            "role": "assistant",
+            "content": "same answer",
+            "timestamp": sidecar_timestamp,
+        }
+    ]
+    state = [
+        {
+            "role": "assistant",
+            "content": "same answer",
+            "timestamp": state_timestamp,
+            "api_content": "must-not-attach",
+        }
+    ]
+
+    merged = merge_session_messages_append_only(sidecar, state)
+
+    assert merged[0].get("api_content") is None
+
+
+def test_issue6751_reconciliation_role_content_fallback_tolerates_mixed_timestamps():
+    from api.models import merge_session_messages_append_only
+
+    sidecar = [{"role": "assistant", "content": "one answer", "timestamp": 30.0}]
+    state = [{"role": "assistant", "content": "one answer", "api_content": "wire"}]
+
+    merged = merge_session_messages_append_only(sidecar, state)
+
+    assert len(merged) == 1
+    assert merged[0]["api_content"] == "wire"
+
+
+def test_issue6751_repeated_identical_rows_without_provenance_stay_unmatched():
+    from api.models import merge_session_messages_append_only
+
+    sidecar = [
+        {"role": "assistant", "content": "same answer", "timestamp": 30.0},
+        {"role": "assistant", "content": "same answer"},
+    ]
+    state = [
+        {"role": "assistant", "content": "same answer", "api_content": "wire-a"},
+        {"role": "assistant", "content": "same answer", "api_content": "wire-b"},
+    ]
+
+    merged = merge_session_messages_append_only(sidecar, state)
+
+    assert [message.get("api_content") for message in merged[:2]] == [None, None]
+    assert [message.get("api_content") for message in merged[2:]] == [
+        "wire-a",
+        "wire-b",
+    ]
+
+
+@pytest.mark.parametrize(
+    ("keep_count", "expected_message_count", "expected_context_sidecars"),
+    [
+        (0, 0, []),
+        (1, 1, ["[Workspace::v1: /fork]\nrequest"]),
+        (2, 2, ["[Workspace::v1: /fork]\nrequest", "provider answer bytes"]),
+    ],
+)
+def test_issue6751_branch_keep_count_retains_final_assistant_sidecar(
+    monkeypatch, keep_count, expected_message_count, expected_context_sidecars
+):
+    """Fork prefixes keep the same reconciled provider payload as the display rows."""
+    from collections import OrderedDict
+
+    import api.routes as routes
+
+    source_messages = [
+        {
+            "role": "user",
+            "content": "request",
+            "timestamp": 1.0,
+            "id": "turn-user",
+            "api_content": "[Workspace::v1: /fork]\nrequest",
+        },
+        {
+            "role": "assistant",
+            "content": "answer",
+            "timestamp": 2.0,
+            "id": "turn-assistant",
+            "api_content": "provider answer bytes",
+        },
+    ]
+    source = SimpleNamespace(
+        _branch_source_readonly=True,
+        session_id="issue6751-keep-count-parent",
+        title="Parent",
+        workspace="/tmp/workspace",
+        model="test-model",
+        model_provider="test-provider",
+        profile="default",
+        messages=source_messages,
+        context_messages=[
+            {"role": "user", "content": "request", "id": "turn-user"},
+            {"role": "assistant", "content": "answer", "id": "turn-assistant"},
+        ],
+        project_id=None,
+        personality=None,
+        enabled_toolsets=None,
+        context_length=None,
+        threshold_tokens=None,
+        gateway_routing=None,
+        context_engine=None,
+        context_engine_state={},
+    )
+    captured = {}
+
+    class Branch:
+        def __init__(self, **kwargs):
+            self.__dict__.update(kwargs)
+            self.session_id = f"issue6751-keep-count-{keep_count}"
+            captured["branch"] = self
+
+        def save(self):
+            captured["saved"] = True
+
+    monkeypatch.setattr(routes, "_check_csrf", lambda _handler: True)
+    monkeypatch.setattr(
+        routes,
+        "read_body",
+        lambda _handler: {"session_id": source.session_id, "keep_count": keep_count},
+    )
+    monkeypatch.setattr(routes, "_guard_request_session_visibility", lambda *_args, **_kwargs: True)
+    monkeypatch.setattr(routes, "_load_branch_source_or_refuse", lambda *_args: source)
+    monkeypatch.setattr(routes, "_session_requires_cli_metadata_lookup", lambda _source: False)
+    monkeypatch.setattr(routes, "_is_messaging_session_record", lambda _record: False)
+    monkeypatch.setattr(routes, "Session", Branch)
+    monkeypatch.setattr(routes, "SESSIONS", OrderedDict())
+    monkeypatch.setattr(routes, "_evict_sessions_over_cap", lambda: None)
+    monkeypatch.setattr(routes, "publish_session_list_changed", lambda *_args, **_kwargs: None)
+    monkeypatch.setattr(routes, "j", lambda _handler, payload, **_kwargs: payload)
+    handler = SimpleNamespace(
+        client_address=("127.0.0.1", 12345),
+        headers={},
+        command="POST",
+        path="/api/session/branch",
+    )
+
+    routes.handle_post(
+        handler,
+        SimpleNamespace(path="/api/session/branch", query=""),
+    )
+
+    branch = captured["branch"]
+    assert len(branch.messages) == expected_message_count
+    assert [message.get("api_content") for message in branch.context_messages] == expected_context_sidecars
diff --git a/tests/test_issue6757_redaction_and_runner_sse_fixes.py b/tests/test_issue6757_redaction_and_runner_sse_fixes.py
new file mode 100644
index 00000000..bd7f104b
--- /dev/null
+++ b/tests/test_issue6757_redaction_and_runner_sse_fixes.py
@@ -0,0 +1,109 @@
+"""Regression coverage for the two maintainer fixes folded into PR #6757.
+
+1. redact_session_data must NOT credential-mask operational session fields such
+   as the workspace path — a valid path can legitimately contain a
+   credential-shaped component, and masking it corrupts the authoritative value
+   the client echoes back on the next send (which then fails validation).
+2. Runner-backed SSE payloads must be projected before relay so an internal
+   `api_content` replay sidecar never reaches the browser.
+"""
+from __future__ import annotations
+
+
+def test_redact_session_data_preserves_credential_shaped_workspace_path():
+    from api.helpers import redact_session_data
+
+    # A real workspace path whose leaf looks like a secret token.
+    workdir = "/home/user/keys/sk-abcdefghijklmnopqrstuvwxyz012345/project"
+    session = {
+        "id": "sess-1",
+        "workspace": workdir,
+        "workdir": workdir,
+        "cwd": workdir,
+        "messages": [{"role": "user", "content": "hello"}],
+    }
+    out = redact_session_data(session)
+    # Operational path fields survive verbatim (not masked).
+    assert out["workspace"] == workdir
+    assert out["workdir"] == workdir
+    assert out["cwd"] == workdir
+    # And the internal replay/id aliases are still stripped from messages.
+    assert "api_content" not in out["messages"][0]
+
+
+def test_redact_session_data_still_strips_api_content_from_messages():
+    from api.helpers import redact_session_data
+
+    session = {
+        "id": "sess-2",
+        "workspace": "/home/user/project",
+        "messages": [
+            {
+                "role": "assistant",
+                "content": "clean transcript text",
+                "api_content": "PROVIDER REPLAY BYTES — must never leak",
+                "_state_db_row_id": 42,
+                "_db_row_id": 42,
+                "state_db_row_id": 42,
+            }
+        ],
+        "context_messages": [
+            {"role": "user", "content": "ctx", "api_content": "ctx replay bytes"}
+        ],
+    }
+    out = redact_session_data(session)
+    msg = out["messages"][0]
+    assert "api_content" not in msg
+    assert "_state_db_row_id" not in msg
+    assert "_db_row_id" not in msg
+    assert "state_db_row_id" not in msg
+    assert msg["content"] == "clean transcript text"
+    # context_messages sidecars are stripped too.
+    assert "api_content" not in out["context_messages"][0]
+
+
+def test_project_runner_event_payload_strips_api_content_from_session():
+    from api.routes import _project_runner_event_payload
+
+    payload = {
+        "status": "done",
+        "session": {
+            "id": "sess-3",
+            "messages": [
+                {
+                    "role": "assistant",
+                    "content": "visible",
+                    "api_content": "REPLAY BYTES — must not reach browser",
+                }
+            ],
+        },
+    }
+    out = _project_runner_event_payload(payload)
+    assert "api_content" not in out["session"]["messages"][0]
+    assert out["session"]["messages"][0]["content"] == "visible"
+    # Non-transcript fields on the envelope are preserved.
+    assert out["status"] == "done"
+
+
+def test_project_runner_event_payload_strips_api_content_from_message_shaped():
+    from api.routes import _project_runner_event_payload
+
+    payload = {
+        "id": "sess-4",
+        "messages": [
+            {"role": "assistant", "content": "hi", "api_content": "REPLAY — leak?"}
+        ],
+    }
+    out = _project_runner_event_payload(payload)
+    assert "api_content" not in out["messages"][0]
+
+
+def test_project_runner_event_payload_passes_non_session_through():
+    from api.routes import _project_runner_event_payload
+
+    # A token/delta event with no transcript must pass through untouched.
+    payload = {"type": "token", "delta": "hello", "seq": 7}
+    assert _project_runner_event_payload(payload) == payload
+    # Non-dict payloads pass through unchanged.
+    assert _project_runner_event_payload("raw") == "raw"
+    assert _project_runner_event_payload(None) is None
diff --git a/tests/test_native_image_attachments.py b/tests/test_native_image_attachments.py
index b3118d32..7b572dc5 100644
--- a/tests/test_native_image_attachments.py
+++ b/tests/test_native_image_attachments.py
@@ -17,6 +17,7 @@ from api.streaming import (
     _attachment_name,
     _build_native_multimodal_message,
     _NATIVE_IMAGE_MAX_BYTES,
+    _sanitize_messages_for_agent,
     _sanitize_messages_for_api,
 )
 from api.routes import _normalize_chat_attachments
@@ -355,18 +356,68 @@ class TestBuildNativeMultimodalMessage:
         assert sanitized == [{'role': 'user', 'content': content}]
 
     def test_sync_chat_history_sanitizer_receives_config(self):
-        """#2398: fallback POST /api/chat must use the text-mode history sanitizer too."""
-        src = Path('api/routes.py').read_text()
-        assert 'conversation_history=_sanitize_messages_for_api(' in src, (
-            'The legacy synchronous /api/chat endpoint must sanitize history through '
-            '_sanitize_messages_for_api.'
-        )
-        assert 'cfg=get_config(),' in src, (
-            'The legacy synchronous /api/chat endpoint must pass current config into '
-            '_sanitize_messages_for_api so historical image_url parts are stripped '
-            'for text-mode providers just like the streaming endpoint.'
+        """#2398/#6751: Agent history strips native images but keeps wire sidecars."""
+        original_user_wire = '[Workspace::v1: /original]\ndescribe this image'
+        original_assistant_wire = '[Workspace::v1: /original]\nIt is a chart.'
+        history = [
+            {
+                'role': 'user',
+                'content': [
+                    {'type': 'text', 'text': 'describe this image'},
+                    {'type': 'image_url', 'image_url': {'url': 'data:image/png;base64,AAA='}},
+                ],
+                'api_content': original_user_wire,
+            },
+            {
+                'role': 'assistant',
+                'content': 'It is a chart.',
+                'api_content': original_assistant_wire,
+            },
+        ]
+        cfg = {'agent': {'image_input_mode': 'text'}}
+
+        sanitized = _sanitize_messages_for_agent(history, cfg=cfg)
+
+        assert sanitized[0]['content'] == 'describe this image'
+        assert 'image_url' not in str(sanitized)
+        assert sanitized[0]['api_content'] == original_user_wire
+        assert sanitized[1]['api_content'] == original_assistant_wire
+
+    def test_agent_history_sanitizer_forwards_requested_provider(self, monkeypatch):
+        """The Agent wrapper must preserve master image-routing provenance."""
+        import api.streaming as streaming
+
+        captured = {}
+
+        def resolve_mode(cfg, active_provider, active_model, *, requested_provider):
+            captured.update({
+                'cfg': cfg,
+                'active_provider': active_provider,
+                'active_model': active_model,
+                'requested_provider': requested_provider,
+            })
+            return 'native'
+
+        monkeypatch.setattr(streaming, '_resolve_image_input_mode', resolve_mode)
+        history = [{'role': 'user', 'content': 'hello', 'api_content': 'wire'}]
+        cfg = {'agent': {'image_input_mode': 'auto'}}
+
+        sanitized = _sanitize_messages_for_agent(
+            history,
+            cfg=cfg,
+            effective_model='model-a',
+            effective_provider='custom',
+            requested_provider='custom:gateway-a',
         )
 
+        assert captured == {
+            'cfg': cfg,
+            'active_provider': 'custom',
+            'active_model': 'model-a',
+            'requested_provider': 'custom:gateway-a',
+        }
+        assert sanitized == history
+
     def test_fake_png_rejected_by_magic_bytes(self):
         """A file named .png that is not actually an image must be rejected."""
         with TemporaryDirectory() as d: