refactor: shared helpers for api_content sidecar pop/drop/extract

Deduplicates the sidecar handling across 9 sites:
- substitute_api_content(): 2 API-bound pop+substitute sites
  (chat_completion_helpers, transport — conversation_loop keeps its
  inline pop because the current-turn compose fallback needs the value)
- drop_stale_api_content(): 4 rewrite-drop sites
  (context_compressor x2, replay_cleanup, agent_runtime_helpers)
- extract_api_content_sidecar(): 3 gateway forwarding sites
  (gateway/session, gateway/slash_commands, cli_commands_mixin)

Also: restore the eager _pending_cli_user_message clear on the early
row-creation try (was lost when the crash-persist moved after prefetch —
a crash in compression between the two tries would leave a stale staged
input), and fix a comment indentation nit in run_agent.py.

Co-authored-by: Soju06 <qlskssk@gmail.com>

agent/agent_runtime_helpers.py

@@ -37,6 +37,7 @@ from agent.tool_dispatch_helpers import _trajectory_normalize_msg, make_tool_res
 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__)
@@ -590,7 +591,7 @@ def repair_message_sequence(agent, messages: List[Dict]) -> int:
                 # 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)
+                drop_stale_api_content(prev)
                 repairs += 1
                 continue
         merged.append(msg)

agent/chat_completion_helpers.py

@@ -30,6 +30,7 @@ from hermes_cli.timeouts import get_provider_request_timeout, get_provider_stale
 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
@@ -1937,13 +1938,7 @@ def handle_max_iterations(agent, messages: list, api_call_count: int) -> str:
             # 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
+            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 @@ from agent.model_metadata import (
     estimate_messages_tokens_rough,
 )
 from agent.redact import redact_sensitive_text
+from agent.turn_context import drop_stale_api_content
 
 logger = logging.getLogger(__name__)
 
@@ -656,8 +657,8 @@ def _strip_historical_media(messages: List[Dict[str, Any]]) -> List[Dict[str, An
         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)
+        # 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
 
@@ -3470,7 +3471,7 @@ This compaction should PRIORITISE preserving all information related to the focu
                 # 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)
+                drop_stale_api_content(msg)
                 _merge_summary_into_tail = False
             compressed.append(msg)
 

agent/replay_cleanup.py

@@ -22,6 +22,7 @@ from typing import Any, Dict, List
 
 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__)
 
@@ -315,7 +316,7 @@ def strip_stale_dangerous_confirmations(
                 # 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)
+                drop_stale_api_content(redacted)
                 cleaned.append(redacted)
                 continue
         cleaned.append(msg)

agent/turn_context.py

@@ -28,7 +28,7 @@ import logging
 import threading
 import uuid
 from dataclasses import dataclass
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, Mapping, Optional
 
 from agent.conversation_compression import conversation_history_after_compression
 from agent.iteration_budget import IterationBudget
@@ -76,6 +76,51 @@ def compose_user_api_content(
     return content + "\n\n" + "\n\n".join(injections)
 
 
+def substitute_api_content(api_msg: Dict[str, Any]) -> Optional[str]:
+    """Pop the ``api_content`` sidecar and substitute it into ``content``.
+
+    Used at every API-bound message-build site (the ``api_messages`` build in
+    ``conversation_loop``, the max-iterations summary in
+    ``chat_completion_helpers``, the chat-completions transport). The sidecar
+    carries the exact bytes previously sent to the API for this message when
+    they differ from the clean stored content; substituting it here keeps the
+    provider prompt-cache prefix byte-stable across turns.
+
+    Returns the popped sidecar string (for callers that need the value for
+    current-turn composition logic) or ``None`` when absent.
+    """
+    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
+    return sidecar
+
+
+def drop_stale_api_content(msg: Dict[str, Any]) -> None:
+    """Drop the ``api_content`` sidecar from a message whose content was rewritten.
+
+    Called from every content-rewrite path (historical image strip,
+    merge-summary-into-tail, consecutive-user repair merge, stale-confirmation
+    redaction). Replaying the pre-rewrite sidecar would resend exactly what
+    the rewrite removed, so it must be dropped — the cost is one cache
+    boundary miss, never wrong content.
+    """
+    msg.pop("api_content", None)
+
+
+def extract_api_content_sidecar(msg: Mapping[str, Any]) -> Optional[str]:
+    """Extract the ``api_content`` sidecar from a message dict for persistence.
+
+    Shared by the gateway/branch forwarding sites that copy the sidecar into a
+    new row. Returns the string sidecar or ``None`` when absent/non-string.
+    """
+    v = msg.get("api_content")
+    return v if isinstance(v, str) else None
+
+
 def reanchor_current_turn_user_idx(messages: List[Any], user_message: Any) -> int:
     """Locate this turn's user message after compaction rebuilt ``messages``.
 
@@ -466,6 +511,14 @@ def build_turn_context(
             agent.session_id or "none",
             exc_info=True,
         )
+    finally:
+        # Clear the staged CLI input eagerly (as the pre-refactor code did)
+        # so a crash in preflight compression — which runs between this row
+        # create and the late crash-persist below — doesn't leave a stale
+        # _pending_cli_user_message that the next turn would mistake for a
+        # fresh staged 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.

gateway/session.py

@@ -95,6 +95,7 @@ from .whatsapp_identity import (
     normalize_whatsapp_identifier,  # noqa: F401 - re-exported for gateway.session callers
 )
 from utils import atomic_replace
+from agent.turn_context import extract_api_content_sidecar
 
 # Session keys/ids flow into filesystem paths downstream (e.g.
 # ``sessions_dir / f"{session_id}.json"`` in hermes_state, request-dump
@@ -2586,11 +2587,7 @@ class SessionStore:
             # 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
-            ),
+            api_content=extract_api_content_sidecar(message),
         )
 
     # Maximum in-memory pending messages per session before dropping the

gateway/slash_commands.py

@@ -31,6 +31,7 @@ from typing import Any, Optional, Union
 
 from agent.account_usage import fetch_account_usage, render_account_usage_lines
 from agent.i18n import t
+from agent.turn_context import extract_api_content_sidecar
 from gateway.config import HomeChannel, Platform, PlatformConfig
 from gateway.platforms.base import EphemeralReply, MessageEvent, MessageType
 from gateway.session import (
@@ -4107,11 +4108,7 @@ class GatewaySlashCommandsMixin:
                     # 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
-                    ),
+                    api_content=extract_api_content_sidecar(msg),
                 )
             except Exception:
                 pass  # Best-effort copy

hermes_cli/cli_commands_mixin.py

@@ -28,6 +28,7 @@ from rich.markup import escape as _escape
 from rich.panel import Panel
 
 from hermes_constants import display_hermes_home, is_termux as _is_termux_environment
+from agent.turn_context import extract_api_content_sidecar
 from hermes_cli.browser_connect import (
     DEFAULT_BROWSER_CDP_URL,
     discover_local_cdp_url,
@@ -956,11 +957,7 @@ class CLICommandsMixin:
                     # 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
-                    ),
+                    api_content=extract_api_content_sidecar(msg),
                 )
             except Exception:
                 pass  # Best-effort copy

run_agent.py

@@ -1940,7 +1940,7 @@ class AIAgent:
                     _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.