From 39efad89a8c26256fe897866d1f5d5229eb42a98 Mon Sep 17 00:00:00 2001 From: kshitijk4poor <82637225+kshitijk4poor@users.noreply.github.com> Date: Sun, 19 Jul 2026 08:12:28 +0530 Subject: [PATCH] refactor: shared helpers for api_content sidecar pop/drop/extract MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- agent/agent_runtime_helpers.py | 3 +- agent/chat_completion_helpers.py | 9 ++---- agent/context_compressor.py | 7 ++-- agent/replay_cleanup.py | 3 +- agent/turn_context.py | 55 +++++++++++++++++++++++++++++++- gateway/session.py | 7 ++-- gateway/slash_commands.py | 7 ++-- hermes_cli/cli_commands_mixin.py | 7 ++-- run_agent.py | 2 +- 9 files changed, 71 insertions(+), 29 deletions(-) diff --git a/agent/agent_runtime_helpers.py b/agent/agent_runtime_helpers.py index 5fb4924b97e05..e54cfff71946c 100644 --- a/agent/agent_runtime_helpers.py +++ b/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) diff --git a/agent/chat_completion_helpers.py b/agent/chat_completion_helpers.py index 820aa6661bd7f..04c59ae7fd1d8 100644 --- a/agent/chat_completion_helpers.py +++ b/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: diff --git a/agent/context_compressor.py b/agent/context_compressor.py index 285fc1f7fcfac..f94285663b26e 100644 --- a/agent/context_compressor.py +++ b/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) diff --git a/agent/replay_cleanup.py b/agent/replay_cleanup.py index ddb067b8259c5..780fe761bbdaf 100644 --- a/agent/replay_cleanup.py +++ b/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) diff --git a/agent/turn_context.py b/agent/turn_context.py index 2e12234921a1d..b403296289318 100644 --- a/agent/turn_context.py +++ b/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. diff --git a/gateway/session.py b/gateway/session.py index 5addc834d0963..bed8f31b2bfc6 100644 --- a/gateway/session.py +++ b/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 diff --git a/gateway/slash_commands.py b/gateway/slash_commands.py index 870a2a0aabb8e..dccf4ae8be0d4 100644 --- a/gateway/slash_commands.py +++ b/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 diff --git a/hermes_cli/cli_commands_mixin.py b/hermes_cli/cli_commands_mixin.py index 1627d2b875008..4e65155b60228 100644 --- a/hermes_cli/cli_commands_mixin.py +++ b/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 diff --git a/run_agent.py b/run_agent.py index d10208a93da29..7280e29422605 100644 --- a/run_agent.py +++ b/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.