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

agent/agent_runtime_helpers.py

@@ -37,6 +37,7 @@
 from agent.trajectory import convert_scratchpad_to_think
 from agent.credential_pool import STATUS_EXHAUSTED
 from agent.error_classifier import FailoverReason
+from agent.turn_context import drop_stale_api_content
 from utils import base_url_host_matches, base_url_hostname, env_var_enabled, atomic_json_write
 
 logger = logging.getLogger(__name__)
@@ -587,6 +588,10 @@ def _is_codex_interim(m: Dict) -> bool:
                     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.
+                drop_stale_api_content(prev)
                 repairs += 1
                 continue
         merged.append(msg)

agent/chat_completion_helpers.py

@@ -30,6 +30,7 @@
 from hermes_constants import PARTIAL_STREAM_STUB_ID, FINISH_REASON_LENGTH
 from agent.error_classifier import FailoverReason
 from agent.errors import EmptyStreamError
+from agent.turn_context import substitute_api_content
 from agent.gemini_native_adapter import is_native_gemini_base_url
 from agent.model_metadata import is_local_endpoint
 from agent.message_content import flatten_message_text
@@ -1905,6 +1906,15 @@ 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.
+            substitute_api_content(api_msg)
             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

@@ -33,6 +33,7 @@
     estimate_messages_tokens_rough,
 )
 from agent.redact import redact_sensitive_text
+from agent.turn_context import drop_stale_api_content
 
 logger = logging.getLogger(__name__)
 
@@ -655,6 +656,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; drop it so replay can't resend the pre-rewrite bytes.
+        drop_stale_api_content(new_msg)
         result.append(new_msg)
         changed = True
 
@@ -3464,6 +3468,10 @@ def compress(self, messages: List[Dict[str, Any]], current_tokens: int = None, f
                 # 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.
+                drop_stale_api_content(msg)
                 _merge_summary_into_tail = False
             compressed.append(msg)
 

agent/conversation_loop.py

@@ -32,9 +32,12 @@
 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 _perform_api_call(next_api_kwargs):
             # 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

@@ -22,6 +22,7 @@
 
 from agent.tool_dispatch_helpers import make_tool_result_message
 from agent.tool_result_classification import tool_may_have_side_effect
+from agent.turn_context import drop_stale_api_content
 
 logger = logging.getLogger(__name__)
 
@@ -311,6 +312,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.
+                drop_stale_api_content(redacted)
                 cleaned.append(redacted)
                 continue
         cleaned.append(msg)

agent/transports/chat_completions.py

@@ -187,6 +187,7 @@ def convert_messages(
                 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,13 +230,15 @@ def mutable_msg() -> dict[str, Any]:
                 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)
                 out_msg.pop("codex_message_items", None)
                 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).