feat(cache): persist the exact bytes sent to the API in an api_content sidecar

The first LLM call of every gateway turn gets ~0% provider prompt-cache
hit rate (in-turn calls: 97-99%) because the bytes sent for a turn's
user message are not the bytes replayed next turn: memory-prefetch and
pre_llm_call context are injected into the API copy only, the #48677
persist override writes cleaned content to the DB row, and
get_messages_as_conversation sanitize/strips user and assistant content
on load. Any of these diverges the request prefix at that message and
re-prefills everything after it — measured 27.9s for the first call vs
2.4-5.8s cached at a median ~156k-token context.

Persist what you send: a nullable messages.api_content column stores the
exact content string sent to the API when it differs from the clean
stored content, and replay substitutes it verbatim (no sanitize, no
strip). The injection composition lives in one helper
(turn_context.compose_user_api_content); the turn prologue stamps its
output onto the live user message, the api_messages build sends the
stamped bytes, and every outgoing copy pops the field so it never
reaches a provider. The crash-resilience user-turn persist moves after
prefetch/pre_llm_call so the user row is written once with its final
sidecar; _ensure_db_session stays before preflight compression (session
rotation needs the parent row under PRAGMA foreign_keys=ON). The
current-turn index trackers are re-anchored after compaction rebuilds
the message list, in-place preflight compaction backfills the stamp onto
the already-inserted row, and gateway replay forwards the sidecar only
when the replay pipeline did not rewrite the content. Rewrite paths that
would leave stale bytes (historical image strip, merge-summary-into-
tail, consecutive-user repair merge, stale-confirmation redaction) drop
the sidecar; the chat-completions transport and the max-iterations
summary path strip it defensively. codex_app_server and MoA turns are
excluded from stamping because their wire bytes differ from the
composition. A missing or dropped sidecar degrades to today's behavior
(one cache-boundary miss), never to wrong content.

agent/agent_runtime_helpers.py

@@ -587,6 +587,10 @@ def repair_message_sequence(agent, messages: List[Dict]) -> int:
                     if prev_content and new_content
                     else (prev_content or new_content)
                 )
+                # Merged content invalidates the api_content sidecar (exact
+                # bytes previously sent for the pre-merge message) — drop it
+                # so replay can't substitute stale bytes.
+                prev.pop("api_content", None)
                 repairs += 1
                 continue
         merged.append(msg)

agent/chat_completion_helpers.py

@@ -1929,6 +1929,21 @@ def handle_max_iterations(agent, messages: list, api_call_count: int) -> str:
             # and every Hermes-internal underscore-prefixed scaffolding key.
             for schema_foreign in ("tool_name", "codex_reasoning_items", "codex_message_items", "timestamp"):
                 api_msg.pop(schema_foreign, None)
+            # api_content (the persist-what-you-send sidecar) carries the
+            # exact bytes every main-loop call sent for this message —
+            # substitute it before dropping the key (Hermes bookkeeping,
+            # never a provider field), mirroring the loop's api_messages
+            # build. Popping without substituting would send CLEAN content
+            # here, diverging the summary request's prefix at the EARLIEST
+            # sidecar-carrying message and re-prefilling the whole transcript
+            # at exactly the moment the context is largest.
+            _sidecar = api_msg.pop("api_content", None)
+            if (
+                isinstance(_sidecar, str)
+                and _sidecar
+                and api_msg.get("role") in ("user", "assistant")
+            ):
+                api_msg["content"] = _sidecar
             for internal_key in [k for k in api_msg if isinstance(k, str) and k.startswith("_")]:
                 api_msg.pop(internal_key, None)
             if _needs_sanitize:

agent/context_compressor.py

@@ -655,6 +655,9 @@ def _strip_historical_media(messages: List[Dict[str, Any]]) -> List[Dict[str, An
             continue
         new_msg = msg.copy()
         new_msg["content"] = _strip_images_from_content(content)
+        # Content rewritten → the api_content sidecar (exact bytes previously
+        # sent) is stale; replaying it would resend the pre-rewrite bytes.
+        new_msg.pop("api_content", None)
         result.append(new_msg)
         changed = True
 
@@ -3464,6 +3467,10 @@ This compaction should PRIORITISE preserving all information related to the focu
                 # Mark the merged message so frontends can identify it as
                 # containing a compression summary prefix.
                 msg[COMPRESSED_SUMMARY_METADATA_KEY] = True
+                # Content rewritten → the api_content sidecar (exact bytes
+                # previously sent) is stale; drop it so replay can't resend
+                # the pre-merge bytes without the summary.
+                msg.pop("api_content", None)
                 _merge_summary_into_tail = False
             compressed.append(msg)
 

agent/conversation_loop.py

@@ -32,9 +32,12 @@ from agent.conversation_compression import conversation_history_after_compressio
 from agent.display import KawaiiSpinner
 from agent.error_classifier import FailoverReason, classify_api_error
 from agent.iteration_budget import IterationBudget
-from agent.turn_context import build_turn_context
+from agent.turn_context import (
+    build_turn_context,
+    compose_user_api_content,
+    reanchor_current_turn_user_idx,
+)
 from agent.turn_retry_state import TurnRetryState
-from agent.memory_manager import build_memory_context_block
 from agent.message_sanitization import (
     close_interrupted_tool_sequence,
     _repair_tool_call_arguments,
@@ -629,8 +632,8 @@ def run_conversation(
     # ── Per-turn setup (the prologue) ──
     # All once-per-turn setup — stdio guarding, retry-counter resets, user
     # message sanitization, todo/nudge hydration, system-prompt restore-or-
-    # build, crash-resilience persistence, preflight compression, the
-    # ``pre_llm_call`` plugin hook, and external-memory prefetch — lives in
+    # build, preflight compression, the ``pre_llm_call`` plugin hook,
+    # external-memory prefetch, and crash-resilience persistence — lives in
     # ``build_turn_context``.  It mutates ``agent`` exactly as the inline code
     # did and returns the locals the loop below reads back.  See
     # ``agent/turn_context.py``.
@@ -650,6 +653,9 @@ def run_conversation(
         set_session_context=set_session_context,
         set_current_write_origin=set_current_write_origin,
         ra=_ra,
+        # MoA turns append per-call aggregated context to the API copy of the
+        # user message, so no byte-stable api_content sidecar can be stamped.
+        moa_active=bool(moa_config),
     )
     user_message = _ctx.user_message
     original_user_message = _ctx.original_user_message
@@ -858,23 +864,51 @@ def run_conversation(
         for idx, msg in enumerate(messages):
             api_msg = msg.copy()
 
+            # api_content is the persistence sidecar carrying the exact bytes
+            # sent to the API for this message when they differ from the clean
+            # stored content (see compose_user_api_content in turn_context).
+            # It is bookkeeping, never a provider field — pop it from EVERY
+            # outgoing copy.
+            _api_content = api_msg.pop("api_content", None)
+
             # Inject ephemeral context into the current turn's user message.
             # Sources: memory manager prefetch + plugin pre_llm_call hooks
             # with target="user_message" (the default).  Both are
             # API-call-time only — the original message in `messages` is
-            # never mutated, so nothing leaks into session persistence.
+            # never mutated beyond the api_content stamp, so nothing leaks
+            # into the clean transcript content.
             if idx == current_turn_user_idx and msg.get("role") == "user":
-                _injections = []
-                if _ext_prefetch_cache:
-                    _fenced = build_memory_context_block(_ext_prefetch_cache)
-                    if _fenced:
-                        _injections.append(_fenced)
-                if _plugin_user_context:
-                    _injections.append(_plugin_user_context)
-                if _injections:
-                    _base = api_msg.get("content", "")
-                    if isinstance(_base, str):
-                        api_msg["content"] = _base + "\n\n" + "\n\n".join(_injections)
+                if isinstance(_api_content, str) and _api_content:
+                    # Stamped by the prologue from the same composition —
+                    # reuse it so the persisted sidecar and the wire cannot
+                    # drift, and so every pass this turn sends identical
+                    # bytes (composed from msg["content"], never from a
+                    # previously-injected copy).
+                    api_msg["content"] = _api_content
+                else:
+                    # Callers that bypass the prologue stamping: compose live.
+                    _composed = compose_user_api_content(
+                        api_msg.get("content", ""),
+                        _ext_prefetch_cache,
+                        _plugin_user_context,
+                    )
+                    if _composed is not None:
+                        api_msg["content"] = _composed
+            elif (
+                isinstance(_api_content, str)
+                and _api_content
+                and msg.get("role") in ("user", "assistant")
+            ):
+                # Historical message: replay the exact bytes sent when it was
+                # live, so the provider prompt-cache prefix stays byte-stable
+                # instead of diverging at the injection point and
+                # re-prefilling everything after it. User rows carry the
+                # prefetch/plugin injection sidecar; user AND assistant rows
+                # can carry a sanitize-divergence sidecar (content that
+                # ``get_messages_as_conversation``'s sanitize_context/strip
+                # would rewrite on reload — see the capture in
+                # ``_flush_messages_to_session_db``).
+                api_msg["content"] = _api_content
 
             # For ALL assistant messages, pass reasoning back to the API
             # This ensures multi-turn reasoning context is preserved
@@ -4352,6 +4386,16 @@ def run_conversation(
             # to fit the context window.
             retry_count += 1
             _retry.restart_with_compressed_messages = False
+            # In-loop compression rebuilt `messages` with fresh compaction
+            # copies, so the pre-compression current-turn index is stale.
+            # Re-anchor exactly like the prologue does: a stale index that
+            # lands on a historical user message would make the live-compose
+            # fallback inject this turn's prefetch into that message on the
+            # wire only, diverging the next turn's replayed prefix there.
+            current_turn_user_idx = reanchor_current_turn_user_idx(
+                messages, user_message
+            )
+            agent._persist_user_message_idx = current_turn_user_idx
             continue
 
         if _retry.restart_with_rebuilt_messages:

agent/replay_cleanup.py

@@ -311,6 +311,11 @@ def strip_stale_dangerous_confirmations(
                 )
                 redacted = dict(msg)
                 redacted["content"] = _EXPIRED_CONFIRMATION_SENTINEL
+                # Drop the api_content sidecar: it carries the exact bytes
+                # previously sent — i.e. the dangerous confirmation this
+                # redaction exists to expire. Replaying it verbatim would
+                # undo the redaction on the wire.
+                redacted.pop("api_content", None)
                 cleaned.append(redacted)
                 continue
         cleaned.append(msg)

agent/transports/chat_completions.py

@@ -187,6 +187,7 @@ class ChatCompletionsTransport(ProviderTransport):
                 or "tool_name" in msg
                 or "effect_disposition" in msg
                 or "timestamp" in msg  # #47868 — strict providers reject this
+                or "api_content" in msg  # persist-what-you-send sidecar
             ):
                 needs_sanitize = True
                 break
@@ -229,6 +230,7 @@ class ChatCompletionsTransport(ProviderTransport):
                 or "tool_name" in msg
                 or "effect_disposition" in msg
                 or "timestamp" in msg  # #47868 — leak into strict providers
+                or "api_content" in msg  # persist-what-you-send sidecar
             ):
                 out_msg = mutable_msg()
                 out_msg.pop("codex_reasoning_items", None)
@@ -236,6 +238,7 @@ class ChatCompletionsTransport(ProviderTransport):
                 out_msg.pop("tool_name", None)
                 out_msg.pop("effect_disposition", None)
                 out_msg.pop("timestamp", None)  # #47868 — leak into strict providers
+                out_msg.pop("api_content", None)  # persist-what-you-send sidecar
 
 
             # Drop all Hermes-internal scaffolding markers (``_``-prefixed).

agent/turn_context.py

@@ -3,8 +3,10 @@
 ``run_conversation`` opened with ~470 lines of straight-line setup before the
 tool-calling loop ever started: stdio guarding, runtime-main wiring, retry-counter
 resets, user-message sanitization, todo/nudge-counter hydration, system-prompt
-restore-or-build, crash-resilience persistence, preflight context compression, the
-``pre_llm_call`` plugin hook, and external-memory prefetch.
+restore-or-build, session-row creation (before compression, whose DB writes
+reference the row), preflight context compression, the ``pre_llm_call`` plugin
+hook, external-memory prefetch, and crash-resilience persistence (last, so the
+user row is written once with its final ``api_content`` sidecar).
 
 All of that is *prologue* — it runs once per turn, has no back-references into the
 loop, and produces a fixed set of values the loop then consumes. ``TurnContext``
@@ -30,6 +32,7 @@ from typing import Any, Dict, List, Optional
 
 from agent.conversation_compression import conversation_history_after_compression
 from agent.iteration_budget import IterationBudget
+from agent.memory_manager import build_memory_context_block
 from agent.model_metadata import (
     estimate_messages_tokens_rough,
     estimate_request_tokens_rough,
@@ -38,6 +41,67 @@ from agent.model_metadata import (
 logger = logging.getLogger(__name__)
 
 
+def compose_user_api_content(
+    content: Any,
+    ext_prefetch_cache: str,
+    plugin_user_context: str,
+) -> Optional[str]:
+    """Compose the API-bound content of the current turn's user message.
+
+    Sources: memory-manager prefetch + ``pre_llm_call`` plugin context with
+    target="user_message" (the default). Both are appended to the *API copy*
+    of the user message only — the stored content stays clean.
+
+    This is the single source of that composition. The prologue stamps the
+    result onto the live message as ``api_content`` (persisted alongside the
+    clean content) and the ``api_messages`` build in ``conversation_loop``
+    sends the same helper's output, so the persisted sidecar can never drift
+    from the bytes on the wire — which is the whole prompt-cache invariant:
+    what turn N sends must be what turn N+1 replays.
+
+    Returns ``None`` when nothing is injected (multimodal/non-string content,
+    or no ephemeral context), meaning the message is sent as-is.
+    """
+    if not isinstance(content, str):
+        return None
+    injections = []
+    if ext_prefetch_cache:
+        fenced = build_memory_context_block(ext_prefetch_cache)
+        if fenced:
+            injections.append(fenced)
+    if plugin_user_context:
+        injections.append(plugin_user_context)
+    if not injections:
+        return None
+    return content + "\n\n" + "\n\n".join(injections)
+
+
+def reanchor_current_turn_user_idx(messages: List[Any], user_message: Any) -> int:
+    """Locate this turn's user message after compaction rebuilt ``messages``.
+
+    Compression replaces list entries with fresh copies (and may append a
+    todo-snapshot user message or a restored user turn AFTER the surviving
+    copy of the current turn's message), so a pre-compression index is
+    meaningless. Prefer the LAST user message whose content exactly matches
+    this turn's text — the surviving copy in the common case — so the
+    injection stamp and the #48677 persist override can't land on a
+    todo-snapshot or historical row. Fall back to the last user message when
+    no exact match survives (merge-summary-into-tail rewrites the content but
+    the trackers still need a live anchor). Returns -1 when the list has no
+    user message at all.
+    """
+    fallback = -1
+    for i in range(len(messages) - 1, -1, -1):
+        msg = messages[i]
+        if not (isinstance(msg, dict) and msg.get("role") == "user"):
+            continue
+        if fallback < 0:
+            fallback = i
+        if msg.get("content") == user_message:
+            return i
+    return fallback
+
+
 def _compression_made_progress(
     orig_len: int, new_len: int, orig_tokens: int, new_tokens: int
 ) -> bool:
@@ -133,6 +197,7 @@ def build_turn_context(
     set_session_context,
     set_current_write_origin,
     ra,
+    moa_active: bool = False,
 ) -> TurnContext:
     """Run the once-per-turn setup and return the loop's input context.
 
@@ -379,38 +444,34 @@ def build_turn_context(
 
     # Create the DB session row now that _cached_system_prompt is populated, so
     # the persisted snapshot is written non-NULL on the first turn (Issue
-    # #45499). Keep row creation and the marker-based append in the same
-    # per-agent critical section as CLI close persistence.
+    # #45499). Idempotent: _ensure_db_session() no-ops once the row exists.
+    # Must run BEFORE preflight compression: in-place compaction inserts
+    # message rows referencing this session (archive_and_compact), and
+    # rotation creates a child with parent_session_id pointing at it — with
+    # PRAGMA foreign_keys=ON, a missing parent row fails both INSERTs on a
+    # fresh oversized first turn. The user-turn crash persist itself runs
+    # LATER (after memory prefetch / pre_llm_call), so the row is written
+    # once with its final api_content — both steps take the same per-agent
+    # persist lock as CLI close persistence.
     persist_lock = getattr(agent, "_session_persist_lock", None)
-
-    def _ensure_and_persist() -> None:
-        agent._ensure_db_session()
-        agent._persist_session(messages, conversation_history)
-
-    # Crash-resilience: persist the inbound user turn as soon as the session row exists.
     try:
         if persist_lock is None:
-            _ensure_and_persist()
+            agent._ensure_db_session()
         else:
             with persist_lock:
-                _ensure_and_persist()
+                agent._ensure_db_session()
     except Exception:
         logger.warning(
-            "Early turn-start session persistence failed for session=%s",
+            "Turn-start session row creation failed for session=%s",
             agent.session_id or "none",
             exc_info=True,
         )
-    finally:
-        # Keep an unmarked staged input available to a later close retry if the
-        # normal persistence attempt failed. Once the marker is present, the
-        # close path must no longer treat it as a pre-worker UI input.
-        if not isinstance(pending_cli_message, dict) or pending_cli_message.get("_db_persisted"):
-            agent._pending_cli_user_message = None
 
     # ── Preflight context compression ──
     # Gate the (expensive) full token estimate behind a cheap pre-check.
     # See ``_should_run_preflight_estimate`` for the OR semantics that fix
     # issue #27405 (a few very large messages slipping past the count gate).
+    _preflight_compressed = False
     if agent.compression_enabled and _should_run_preflight_estimate(
         messages,
         agent.context_compressor.protect_first_n,
@@ -478,6 +539,7 @@ def build_turn_context(
                 getattr(agent, "codex_app_server_auto_compaction", "native"),
             )
         elif _compressor.should_compress(_preflight_tokens):
+            _preflight_compressed = True
             logger.info(
                 "Preflight compression: ~%s tokens >= %s threshold (model %s, ctx %s)",
                 f"{_preflight_tokens:,}",
@@ -521,6 +583,19 @@ def build_turn_context(
                 if not _compressor.should_compress(_preflight_tokens):
                     break
 
+    if _preflight_compressed:
+        # Compression rebuilt the list (tail messages are fresh compaction
+        # copies), so the pre-compression index of this turn's user message
+        # is stale. Re-anchor both index trackers: the api_content stamp
+        # below, the loop's injection site, and the flush's persist-override
+        # row (#48677) must all target the surviving dict, not a stale
+        # position. Exact-content match first so a todo-snapshot user message
+        # appended after the tail can't steal the anchor.
+        current_turn_user_idx = reanchor_current_turn_user_idx(
+            messages, user_message
+        )
+        agent._persist_user_message_idx = current_turn_user_idx
+
     # Plugin hook: pre_llm_call (context injected into user message, not system prompt).
     plugin_user_context = ""
     try:
@@ -610,6 +685,92 @@ def build_turn_context(
         except Exception:
             pass
 
+    # ── api_content sidecar: persist what you send ──
+    # The prefetch/plugin context above is injected into the API copy of this
+    # turn's user message, never into the stored content — so on the next
+    # turn the message would replay WITHOUT the injection, diverging the
+    # request prefix at this point and re-prefilling everything after it
+    # (the whole previous turn's assistant/tool chain). Stamp the exact
+    # API-bound bytes on the live dict, only when they differ from the clean
+    # content, so the crash persist below writes both in the same row and
+    # replay can reproduce the sent prefix byte-for-byte. Guarded by the
+    # same predicate the api_messages build uses, so the stamped bytes are
+    # exactly the bytes the loop sends. codex_app_server turns bypass the
+    # api_messages build entirely (the codex thread gets the plain user
+    # message), so stamping there would persist bytes that were never sent.
+    # MoA turns append per-call aggregated reference context to the same API
+    # copy AFTER this composition, so the stamped bytes would never match the
+    # wire either — skip the stamp rather than persist provably wrong "exact
+    # sent bytes" (MoA keeps its pre-sidecar cache behavior).
+    if (
+        not moa_active
+        and getattr(agent, "api_mode", None) != "codex_app_server"
+        and 0 <= current_turn_user_idx < len(messages)
+        and messages[current_turn_user_idx].get("role") == "user"
+    ):
+        _turn_user_msg = messages[current_turn_user_idx]
+        _api_content = compose_user_api_content(
+            _turn_user_msg.get("content", ""), ext_prefetch_cache, plugin_user_context
+        )
+        if _api_content is not None and _api_content != _turn_user_msg.get("content"):
+            _turn_user_msg["api_content"] = _api_content
+            # In-place preflight compaction has ALREADY inserted this turn's
+            # user row (archive_and_compact runs before prefetch/pre_llm_call
+            # can compose the sidecar), and the crash persist below identity-
+            # skips every compacted dict (they are all in the rebound
+            # conversation_history) — so the stamp would never reach the DB.
+            # Backfill it onto the freshly-inserted row directly. Rotation
+            # mode needs nothing here: its compacted copies flush to the
+            # child session after this stamp.
+            if _preflight_compressed and bool(
+                getattr(agent, "_last_compaction_in_place", False)
+            ):
+                _db = getattr(agent, "_session_db", None)
+                if _db is not None:
+                    try:
+                        _db.set_latest_user_api_content(
+                            agent.session_id,
+                            _turn_user_msg.get("content"),
+                            _api_content,
+                        )
+                    except Exception:
+                        logger.warning(
+                            "in-place compaction api_content backfill failed "
+                            "for session=%s",
+                            agent.session_id or "none",
+                            exc_info=True,
+                        )
+
+    # Crash-resilience: persist the inbound user turn before the first LLM
+    # call. Runs after preflight compression (which rewrites history anyway)
+    # and after prefetch/pre_llm_call, so the user row is written once with
+    # its final api_content instead of being re-written mid-turn.
+    # Keep row creation and the marker-based append in the same per-agent
+    # critical section as CLI close persistence, and retry the row create if
+    # the pre-compression attempt above failed transiently.
+    def _ensure_and_persist() -> None:
+        agent._ensure_db_session()
+        agent._persist_session(messages, conversation_history)
+
+    try:
+        if persist_lock is None:
+            _ensure_and_persist()
+        else:
+            with persist_lock:
+                _ensure_and_persist()
+    except Exception:
+        logger.warning(
+            "Early turn-start session persistence failed for session=%s",
+            agent.session_id or "none",
+            exc_info=True,
+        )
+    finally:
+        # Keep an unmarked staged input available to a later close retry if the
+        # normal persistence attempt failed. Once the marker is present, the
+        # close path must no longer treat it as a pre-worker UI input.
+        if not isinstance(pending_cli_message, dict) or pending_cli_message.get("_db_persisted"):
+            agent._pending_cli_user_message = None
+
     return TurnContext(
         user_message=user_message,
         original_user_message=original_user_message,

gateway/run.py

@@ -849,6 +849,23 @@ def _build_replay_entry(
     providers.
     """
     entry: Dict[str, Any] = {"role": role, "content": content}
+    # api_content sidecar (persist-what-you-send, prompt-cache stability):
+    # forward the exact bytes previously sent to the API for this message so
+    # the agent's api_messages build can substitute them and keep the request
+    # prefix byte-stable across turns. Forward ONLY when this replay pipeline
+    # did not rewrite the content (timestamp injection, auto-continue strip,
+    # mirror prefix): a rewritten clean content means the pipeline decided
+    # different bytes must replay — resending the stored sidecar would
+    # reintroduce exactly what was stripped. Dropping it costs one cache
+    # boundary; resending stripped noise is a behavior regression.
+    _sidecar = msg.get("api_content")
+    if (
+        role in ("user", "assistant")
+        and isinstance(_sidecar, str)
+        and _sidecar
+        and content == msg.get("content")
+    ):
+        entry["api_content"] = _sidecar
     if role == "assistant":
         for _rkey in _ASSISTANT_REPLAY_FIELDS:
             if _rkey not in msg:

gateway/session.py

@@ -2582,6 +2582,15 @@ class SessionStore:
             platform_message_id=(message.get("platform_message_id") or message.get("message_id")),
             observed=bool(message.get("observed")),
             timestamp=message.get("timestamp"),
+            # api_content sidecar: the exact bytes sent to the API for
+            # this message (prompt-cache-stable replay). Must survive
+            # any gateway-side persistence path or the next turn's
+            # replay diverges at this row.
+            api_content=(
+                message.get("api_content")
+                if isinstance(message.get("api_content"), str)
+                else None
+            ),
         )
 
     # Maximum in-memory pending messages per session before dropping the

gateway/slash_commands.py

@@ -4104,6 +4104,14 @@ class GatewaySlashCommandsMixin:
                     reasoning_details=msg.get("reasoning_details"),
                     codex_reasoning_items=msg.get("codex_reasoning_items"),
                     codex_message_items=msg.get("codex_message_items"),
+                    # Keep the api_content sidecar so the branch's first turn
+                    # replays the parent's exact wire bytes (warm provider
+                    # prompt cache) instead of a full cold prefill.
+                    api_content=(
+                        msg.get("api_content")
+                        if isinstance(msg.get("api_content"), str)
+                        else None
+                    ),
                 )
             except Exception:
                 pass  # Best-effort copy

hermes_cli/cli_commands_mixin.py

@@ -953,6 +953,14 @@ class CLICommandsMixin:
                     tool_calls=msg.get("tool_calls"),
                     tool_call_id=msg.get("tool_call_id"),
                     reasoning=msg.get("reasoning"),
+                    # Keep the api_content sidecar so the branch's first turn
+                    # replays the parent's exact wire bytes (warm provider
+                    # prompt cache) instead of a full cold prefill.
+                    api_content=(
+                        msg.get("api_content")
+                        if isinstance(msg.get("api_content"), str)
+                        else None
+                    ),
                 )
             except Exception:
                 pass  # Best-effort copy

hermes_state.py

@@ -817,7 +817,8 @@ CREATE TABLE IF NOT EXISTS messages (
     platform_message_id TEXT,
     observed INTEGER DEFAULT 0,
     active INTEGER NOT NULL DEFAULT 1,
-    compacted INTEGER NOT NULL DEFAULT 0
+    compacted INTEGER NOT NULL DEFAULT 0,
+    api_content TEXT
 );
 
 CREATE TABLE IF NOT EXISTS session_model_usage (
@@ -4138,6 +4139,7 @@ class SessionDB:
         observed: bool = False,
         effect_disposition: Optional[str] = None,
         timestamp: Any = None,
+        api_content: Optional[str] = None,
     ) -> int:
         """
         Append a message to a session. Returns the message row ID.
@@ -4150,6 +4152,11 @@ class SessionDB:
         independent of the SQLite autoincrement primary key and is used by
         platform-specific flows like yuanbao's recall guard to redact a
         message by its platform-side identifier.
+
+        ``api_content`` is the exact content string sent to the API for this
+        message when it differs from ``content`` (ephemeral memory/plugin
+        injections, persist overrides).  It is a byte-fidelity sidecar for
+        prompt-cache-stable replay — stored verbatim, never sanitized.
         """
         # Serialize structured fields to JSON before entering the write txn
         reasoning_details_json = (
@@ -4189,8 +4196,8 @@ class SessionDB:
                 """INSERT INTO messages (session_id, role, content, tool_call_id,
                    tool_calls, tool_name, effect_disposition, timestamp, token_count, finish_reason,
                    reasoning, reasoning_content, reasoning_details, codex_reasoning_items,
-                   codex_message_items, platform_message_id, observed, active)
-                   VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
+                   codex_message_items, platform_message_id, observed, active, api_content)
+                   VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
                 (
                     session_id,
                     role,
@@ -4210,6 +4217,7 @@ class SessionDB:
                     platform_message_id,
                     1 if observed else 0,
                     1,
+                    api_content if isinstance(api_content, str) else None,
                 ),
             )
             msg_id = cursor.lastrowid
@@ -4278,12 +4286,14 @@ class SessionDB:
                 msg.get("platform_message_id") or msg.get("message_id")
             )
 
+            api_content = msg.get("api_content")
+
             conn.execute(
                 """INSERT INTO messages (session_id, role, content, tool_call_id,
                    tool_calls, tool_name, effect_disposition, timestamp, token_count, finish_reason,
                    reasoning, reasoning_content, reasoning_details, codex_reasoning_items,
-                   codex_message_items, platform_message_id, observed, active)
-                   VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
+                   codex_message_items, platform_message_id, observed, active, api_content)
+                   VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
                 (
                     session_id,
                     role,
@@ -4303,6 +4313,7 @@ class SessionDB:
                     platform_msg_id,
                     1 if msg.get("observed") else 0,
                     1,
+                    api_content if isinstance(api_content, str) else None,
                 ),
             )
             inserted += 1
@@ -4427,6 +4438,37 @@ class SessionDB:
 
         return self._execute_write(_do)
 
+    def set_latest_user_api_content(
+        self, session_id: str, content: Any, api_content: str
+    ) -> int:
+        """Backfill the ``api_content`` sidecar onto the newest ACTIVE user row.
+
+        In-place preflight compaction (:meth:`archive_and_compact`) inserts the
+        current turn's user row BEFORE the turn prologue composes the
+        prefetch/plugin sidecar, and the subsequent crash persist identity-skips
+        every compacted dict — without this backfill the stamped sidecar would
+        never land in the DB and any reload would replay clean content,
+        re-introducing the prompt-cache divergence the sidecar exists to close.
+
+        The ``content`` match is a defensive guard: if the newest active user
+        row is not the message the caller stamped (racing rewrite, unexpected
+        tail shape), nothing is written. Returns the number of rows updated
+        (0 or 1).
+        """
+        encoded = self._encode_content(content)
+
+        def _do(conn):
+            cursor = conn.execute(
+                "UPDATE messages SET api_content = ? WHERE id = ("
+                "SELECT id FROM messages "
+                "WHERE session_id = ? AND role = 'user' AND active = 1 "
+                "ORDER BY id DESC LIMIT 1"
+                ") AND content IS ?",
+                (api_content, session_id, encoded),
+            )
+            return cursor.rowcount
+
+        return self._execute_write(_do)
 
     def get_messages(
         self,
@@ -4800,7 +4842,8 @@ class SessionDB:
             rows = self._conn.execute(
                 "SELECT role, content, tool_call_id, tool_calls, tool_name, effect_disposition, "
                 "finish_reason, reasoning, reasoning_content, reasoning_details, "
-                "codex_reasoning_items, codex_message_items, platform_message_id, observed, timestamp "
+                "codex_reasoning_items, codex_message_items, platform_message_id, observed, timestamp, "
+                "api_content "
                 f"FROM messages WHERE session_id IN ({placeholders})"
                 # Order by AUTOINCREMENT id (true insertion order), NOT timestamp:
                 # append_message stamps rows with time.time(), which is not
@@ -4851,6 +4894,14 @@ class SessionDB:
             if row["role"] in {"user", "assistant"} and isinstance(content, str):
                 content = sanitize_context(content).strip()
             msg = {"role": row["role"], "content": content}
+            # api_content is the byte-fidelity sidecar: the exact string sent
+            # to the API when it differed from the clean content. Returned
+            # VERBATIM — no sanitize_context, no strip — because the replay
+            # path substitutes it for content to keep the provider prompt
+            # cache prefix byte-stable across turns. Cleaning it here would
+            # re-introduce the divergence it exists to remove.
+            if row["api_content"]:
+                msg["api_content"] = row["api_content"]
             if row["timestamp"]:
                 msg["timestamp"] = row["timestamp"]
             if row["tool_call_id"]:

run_agent.py

@@ -155,7 +155,10 @@ from agent.model_metadata import (
 )
 from agent.usage_pricing import normalize_usage
 # Re-exported for tests that monkeypatch these symbols on run_agent.
-from agent.context_compressor import ContextCompressor  # noqa: F401
+from agent.context_compressor import (  # noqa: F401
+    COMPRESSED_SUMMARY_METADATA_KEY,
+    ContextCompressor,
+)
 from agent.retry_utils import jittered_backoff  # noqa: F401
 from agent.prompt_builder import (  # noqa: F401  # re-exported via _ra() / mock.patch("run_agent.<name>") / from run_agent import <name>
     DEFAULT_AGENT_IDENTITY,
@@ -1928,9 +1931,16 @@ class AIAgent:
                     continue
                 role = msg.get("role", "unknown")
                 content = msg.get("content")
+                # api_content sidecar: the exact bytes sent to the API when
+                # they differ from the clean content (stamped by the turn
+                # prologue for prefetch/plugin injections). Written verbatim
+                # so replay can reproduce the sent prefix byte-for-byte.
+                _row_api_content = msg.get("api_content")
+                if not isinstance(_row_api_content, str):
+                    _row_api_content = None
                 _row_timestamp = msg.get("timestamp")
                 # Apply the persist override to THIS row's written values only
-                # (never to the live dict). A multimodal override is a complete
+# (never to the live dict). A multimodal override is a complete
                 # clean replacement for an API-local noted payload. Preserve the
                 # historical text-only guard for a list payload, though: a plain
                 # text override must not erase its image/audio transcript summary.
@@ -1943,12 +1953,55 @@ class AIAgent:
                     _ov_idx == _msg_idx or msg is pending_cli_message
                 )
                 if is_current_turn_user and msg.get("role") == "user":
-                    if _ov_content is not None and (
-                        not isinstance(content, list) or isinstance(_ov_content, list)
+                    # Preflight compaction can re-anchor the override index at
+                    # a message whose content was MERGED with the compaction
+                    # summary (merge-summary-into-tail). Overwriting that with
+                    # the clean gateway text would silently drop the summary
+                    # from the durable transcript. The wire is already
+                    # consistent — the merge popped the sidecar and the merged
+                    # content is what gets sent — so keep it.
+                    if (
+                        _ov_content is not None
+                        and (not isinstance(content, list) or isinstance(_ov_content, list))
+                        and not msg.get(COMPRESSED_SUMMARY_METADATA_KEY)
                     ):
+                        # The live content is what the API call sends; the
+                        # override is the cleaned transcript value. If they
+                        # differ and no injection already stamped the sidecar,
+                        # keep the sent bytes in api_content so replay matches
+                        # the wire (#48677 divergence, closed for the cache
+                        # prefix too).
+                        if (
+                            _row_api_content is None
+                            and isinstance(content, str)
+                            and content != _ov_content
+                        ):
+                            _row_api_content = content
                         content = _ov_content
                     if _ov_timestamp is not None:
                         _row_timestamp = _ov_timestamp
+                # Store the sidecar only when it actually differs.
+                if _row_api_content == content:
+                    _row_api_content = None
+                # Load-time sanitize divergence: get_messages_as_conversation
+                # replays user/assistant rows through
+                # ``sanitize_context(content).strip()``, so content that
+                # sanitize would rewrite (echoed/pasted <memory-context>
+                # fences or system notes) replays different bytes after a
+                # session reload even though THIS turn sent it verbatim.
+                # Capture the sent bytes in the sidecar so a reloaded session
+                # replays what was actually on the wire. Compared in wire form
+                # (both sides .strip()-ed — the api_messages build strips
+                # every outgoing content string) so plain surrounding
+                # whitespace doesn't grow redundant sidecars.
+                if (
+                    _row_api_content is None
+                    and role in ("user", "assistant")
+                    and isinstance(content, str)
+                    and content
+                    and sanitize_context(content).strip() != content.strip()
+                ):
+                    _row_api_content = content
                 # Persist multimodal tool results as their text summary only —
                 # base64 images would bloat the session DB and aren't useful
                 # for cross-session replay.
@@ -1985,6 +2038,7 @@ class AIAgent:
                     codex_reasoning_items=msg.get("codex_reasoning_items") if role == "assistant" else None,
                     codex_message_items=msg.get("codex_message_items") if role == "assistant" else None,
                     timestamp=_row_timestamp,
+                    api_content=_row_api_content,
                 )
                 msg[_DB_PERSISTED_MARKER] = True
             # The intrinsic markers are now the sole source of truth. Reset the

scripts/release.py

@@ -83,6 +83,7 @@ LEGACY_AUTHOR_MAP = {
     "neo@neodeMac-mini.local": "neo-claw-bot",  # PR #58465 salvage (moa: drop empty user turns from advisory view)
     "2024104039@mails.szu.edu.cn": "pixel4039",  # PR #64420 salvage (streaming: retry zero-chunk streams)
     "marceloparra.hm@gmail.com": "marcelohildebrand",  # PR #42346 salvage (lmstudio: JIT load mode)
+    "qlskssk@gmail.com": "Soju06",  # agent turn-latency perf PRs
     "m.guttmann@journaway.com": "mguttmann",  # PR #63738 salvage (Anthropic setup-token pool auth normalization)
     "VrtxOmega@pm.me": "VrtxOmega",  # PR #43809 salvage (desktop: WSL folder-picker path bridge)
     "gn00742754@gmail.com": "SemonCat",  # PR #56786 salvage (Slack Agent View manifests and Assistant APIs)

tests/gateway/test_replay_entry_fields.py

@@ -251,3 +251,68 @@ class TestBuildReplayEntry:
         assert "timestamp" not in entry
         assert "internal_marker" not in entry
         assert "tool_call_id" not in entry
+
+
+class TestReplayEntryApiContentSidecar:
+    """The api_content sidecar (persist-what-you-send) must survive the
+    gateway's transcript→agent_history rebuild, or the whole prompt-cache
+    fix is inert on gateway platforms — but only when this pipeline did not
+    rewrite the content (a rewrite means different bytes must replay)."""
+
+    def test_user_forwards_api_content_when_content_unchanged(self):
+        msg = {"role": "user", "content": "hi", "api_content": "hi\n\nCTX"}
+        entry = _build_replay_entry("user", "hi", msg)
+        assert entry["api_content"] == "hi\n\nCTX"
+        assert entry["content"] == "hi"
+
+    def test_assistant_forwards_api_content_when_content_unchanged(self):
+        msg = {"role": "assistant", "content": "a", "api_content": "a <memory-context>"}
+        entry = _build_replay_entry("assistant", "a", msg)
+        assert entry["api_content"] == "a <memory-context>"
+
+    def test_dropped_when_pipeline_rewrote_content(self):
+        """Timestamp injection / auto-continue strip / mirror prefix change
+        the replayed content — resending the stored sidecar would
+        reintroduce exactly what was stripped."""
+        msg = {"role": "user", "content": "hi", "api_content": "hi\n\nCTX"}
+        entry = _build_replay_entry("user", "[Tue 12:00] hi", msg)
+        assert "api_content" not in entry
+
+    def test_tool_role_never_forwards(self):
+        msg = {"role": "tool", "content": "r", "api_content": "r+X"}
+        entry = _build_replay_entry("tool", "r", msg)
+        assert "api_content" not in entry
+
+    def test_non_string_or_empty_sidecar_ignored(self):
+        for bad in (None, "", 42, ["x"]):
+            msg = {"role": "user", "content": "hi", "api_content": bad}
+            entry = _build_replay_entry("user", "hi", msg)
+            assert "api_content" not in entry
+
+
+class TestGatewayHistoryBuildForwardsSidecar:
+    def test_end_to_end_history_build_keeps_sidecar(self):
+        from gateway.run import _build_gateway_agent_history
+
+        history = [
+            {"role": "user", "content": "hi", "api_content": "hi\n\nCTX", "timestamp": 123.0},
+            {"role": "assistant", "content": "hello"},
+        ]
+        agent_history, _obs = _build_gateway_agent_history(history)
+        assert agent_history[0]["api_content"] == "hi\n\nCTX"
+
+    def test_mirror_prefix_drops_sidecar(self):
+        from gateway.run import _build_gateway_agent_history
+
+        history = [
+            {
+                "role": "user",
+                "content": "hi",
+                "api_content": "hi\n\nCTX",
+                "mirror": True,
+                "mirror_source": "other",
+            },
+        ]
+        agent_history, _obs = _build_gateway_agent_history(history)
+        assert agent_history[0]["content"].startswith("[Delivered from other]")
+        assert "api_content" not in agent_history[0]