From 7b3dcee928e6a8c61269c02ca5ee95dd04d25e3e Mon Sep 17 00:00:00 2001 From: Soju06 Date: Tue, 14 Jul 2026 07:48:41 +0000 Subject: [PATCH] feat(cache): persist the exact bytes sent to the API in an api_content sidecar MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 4 + agent/chat_completion_helpers.py | 15 + agent/context_compressor.py | 7 + agent/conversation_loop.py | 76 +- agent/replay_cleanup.py | 5 + agent/transports/chat_completions.py | 3 + agent/turn_context.py | 199 +++- gateway/run.py | 17 + gateway/session.py | 9 + gateway/slash_commands.py | 8 + hermes_cli/cli_commands_mixin.py | 8 + hermes_state.py | 63 +- run_agent.py | 62 +- scripts/release.py | 1 + tests/agent/test_api_content_sidecar.py | 1035 +++++++++++++++++++++ tests/gateway/test_replay_entry_fields.py | 65 ++ 16 files changed, 1532 insertions(+), 45 deletions(-) create mode 100644 tests/agent/test_api_content_sidecar.py diff --git a/agent/agent_runtime_helpers.py b/agent/agent_runtime_helpers.py index 36712ec48281..5fb4924b97e0 100644 --- a/agent/agent_runtime_helpers.py +++ b/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) diff --git a/agent/chat_completion_helpers.py b/agent/chat_completion_helpers.py index c3632bffc41e..820aa6661bd7 100644 --- a/agent/chat_completion_helpers.py +++ b/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: diff --git a/agent/context_compressor.py b/agent/context_compressor.py index 0194c829886d..285fc1f7fcfa 100644 --- a/agent/context_compressor.py +++ b/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) diff --git a/agent/conversation_loop.py b/agent/conversation_loop.py index 7a7e6bf33547..83c534087773 100644 --- a/agent/conversation_loop.py +++ b/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: diff --git a/agent/replay_cleanup.py b/agent/replay_cleanup.py index 19ed38e059b7..ddb067b8259c 100644 --- a/agent/replay_cleanup.py +++ b/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) diff --git a/agent/transports/chat_completions.py b/agent/transports/chat_completions.py index d1ee4ef83b52..086883eca1c7 100644 --- a/agent/transports/chat_completions.py +++ b/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). diff --git a/agent/turn_context.py b/agent/turn_context.py index ade5d12763d8..2e12234921a1 100644 --- a/agent/turn_context.py +++ b/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, diff --git a/gateway/run.py b/gateway/run.py index 2b27dda67d2b..551f4a78ea93 100644 --- a/gateway/run.py +++ b/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: diff --git a/gateway/session.py b/gateway/session.py index 02fdd0043aeb..5addc834d096 100644 --- a/gateway/session.py +++ b/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 diff --git a/gateway/slash_commands.py b/gateway/slash_commands.py index dc708638db8c..870a2a0aabb8 100644 --- a/gateway/slash_commands.py +++ b/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 diff --git a/hermes_cli/cli_commands_mixin.py b/hermes_cli/cli_commands_mixin.py index 77aaa523942c..1627d2b87500 100644 --- a/hermes_cli/cli_commands_mixin.py +++ b/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 diff --git a/hermes_state.py b/hermes_state.py index 86618cbed48f..c8caa9e1ef27 100644 --- a/hermes_state.py +++ b/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"]: diff --git a/run_agent.py b/run_agent.py index ccb61a572aed..d10208a93da2 100644 --- a/run_agent.py +++ b/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.") / from run_agent import 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 + # 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 diff --git a/scripts/release.py b/scripts/release.py index 822d54797023..ae59b48ae1a2 100755 --- a/scripts/release.py +++ b/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) diff --git a/tests/agent/test_api_content_sidecar.py b/tests/agent/test_api_content_sidecar.py new file mode 100644 index 000000000000..b78fab9f84f2 --- /dev/null +++ b/tests/agent/test_api_content_sidecar.py @@ -0,0 +1,1035 @@ +"""Tests for the ``api_content`` sidecar ("persist what you send"). + +The first LLM call of every turn used to miss the provider prompt cache +because the bytes sent to the API diverged from the bytes replayed from the +persisted transcript: memory-prefetch / plugin context is injected into the +API copy of the current turn's user message only, and the persist +user-message override (#48677) writes cleaned content to the DB row. The fix +persists the EXACT sent content in a nullable ``messages.api_content`` column +and replays it verbatim (no sanitize, no strip). + +Covers: SessionDB round-trip and auto-migration, the shared composition +helper, prologue stamping order, the flush-override sidecar, and the +end-to-end wire invariant (turn N+1 replays turn N's bytes) against an +in-process mock provider. +""" + +from __future__ import annotations + +import json +import os +import shutil +import sqlite3 +import sys +import tempfile +import threading +import types +from http.server import BaseHTTPRequestHandler, HTTPServer +from unittest.mock import MagicMock, patch + +import pytest + +from agent.memory_manager import build_memory_context_block +from agent.turn_context import build_turn_context, compose_user_api_content +from hermes_state import SessionDB + + +# --------------------------------------------------------------------------- +# compose_user_api_content — the single source of the injection composition +# --------------------------------------------------------------------------- + +class TestComposeUserApiContent: + def test_none_when_nothing_to_inject(self): + assert compose_user_api_content("hello", "", "") is None + + def test_none_for_multimodal_content(self): + blocks = [{"type": "text", "text": "hi"}] + assert compose_user_api_content(blocks, "mem", "ctx") is None + + def test_composes_memory_block_and_plugin_context(self): + out = compose_user_api_content("hello", "likes tea", "PLUGIN-CTX") + fenced = build_memory_context_block("likes tea") + assert out == "hello" + "\n\n" + fenced + "\n\n" + "PLUGIN-CTX" + + def test_plugin_context_only(self): + assert compose_user_api_content("hello", "", "CTX") == "hello\n\nCTX" + + def test_deterministic_across_calls(self): + a = compose_user_api_content("hello", "likes tea", "CTX") + b = compose_user_api_content("hello", "likes tea", "CTX") + assert a == b + + +# --------------------------------------------------------------------------- +# SessionDB: schema, round-trip, verbatim replay +# --------------------------------------------------------------------------- + +class TestSessionDbSidecar: + def _open(self, tmp_path): + db = SessionDB(db_path=tmp_path / "state.db") + db.create_session("s1", source="cli") + return db + + def test_round_trip_is_verbatim_not_sanitized(self, tmp_path): + """api_content must bypass sanitize_context/strip on load — clean + content loses the block and outer whitespace, the + sidecar must not.""" + db = self._open(tmp_path) + sent = " hello\n\n\nrecalled\n\n" + try: + db.append_message("s1", "user", content="hello", api_content=sent) + msgs = db.get_messages_as_conversation("s1") + assert msgs[0]["content"] == "hello" + assert msgs[0]["api_content"] == sent # byte-for-byte + finally: + db.close() + + def test_absent_when_null(self, tmp_path): + db = self._open(tmp_path) + try: + db.append_message("s1", "user", content="hello") + msgs = db.get_messages_as_conversation("s1") + assert "api_content" not in msgs[0] + finally: + db.close() + + def test_get_messages_exposes_column(self, tmp_path): + db = self._open(tmp_path) + try: + db.append_message("s1", "user", content="hello", api_content="hello+ctx") + rows = db.get_messages("s1") + assert rows[0]["api_content"] == "hello+ctx" + finally: + db.close() + + def test_insert_message_rows_carries_sidecar(self, tmp_path): + """replace_messages (compaction/rewrite flows) preserves the sidecar + from message dicts.""" + db = self._open(tmp_path) + try: + db.replace_messages( + "s1", + [ + {"role": "user", "content": "hello", "api_content": "hello+ctx"}, + {"role": "assistant", "content": "hi"}, + ], + ) + msgs = db.get_messages_as_conversation("s1") + assert msgs[0]["api_content"] == "hello+ctx" + assert "api_content" not in msgs[1] + finally: + db.close() + + +class TestAutoMigration: + def test_reconciliation_adds_api_content_column(self, tmp_path): + """Opening a pre-sidecar DB adds the column declaratively (the + SCHEMA_SQL diff in _reconcile_columns), no version-gated migration.""" + db_path = tmp_path / "legacy.db" + conn = sqlite3.connect(str(db_path)) + conn.executescript(""" + CREATE TABLE schema_version (version INTEGER NOT NULL); + + CREATE TABLE sessions ( + id TEXT PRIMARY KEY, + source TEXT NOT NULL, + parent_session_id TEXT, + started_at REAL NOT NULL + ); + + CREATE TABLE messages ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + session_id TEXT NOT NULL, + role TEXT NOT NULL, + content TEXT, + tool_call_id TEXT, + tool_calls TEXT, + tool_name TEXT, + timestamp REAL NOT NULL, + token_count INTEGER, + finish_reason TEXT, + reasoning TEXT + ); + """) + conn.execute( + "INSERT INTO messages (session_id, role, content, timestamp) " + "VALUES (?, ?, ?, ?)", + ("s1", "user", "old row", 1000.0), + ) + conn.commit() + conn.close() + + db = SessionDB(db_path=db_path) + try: + cols = { + row[1] + for row in db._conn.execute('PRAGMA table_info("messages")').fetchall() + } + assert "api_content" in cols + # Old rows read back NULL (no key); new writes round-trip. + db.append_message("s1", "user", content="new", api_content="new+ctx") + msgs = db.get_messages_as_conversation("s1") + assert "api_content" not in msgs[0] + assert msgs[1]["api_content"] == "new+ctx" + finally: + db.close() + + +# --------------------------------------------------------------------------- +# Prologue stamping (build_turn_context) +# --------------------------------------------------------------------------- + +class _FakeTodoStore: + def has_items(self): + return True + + +class _FakeGuardrails: + def reset_for_turn(self): + pass + + +class _FakeAgent: + """Minimal stand-in covering only what the prologue touches + (mirrors tests/agent/test_turn_context.py).""" + + def __init__(self): + self.session_id = "sess-1" + self.model = "test/model" + self.provider = "openrouter" + self.base_url = "https://openrouter.ai/api/v1" + self.api_key = "sk-x" + self.api_mode = "chat_completions" + self.platform = "cli" + self.quiet_mode = True + self.max_iterations = 90 + self.tools = [] + self.valid_tool_names = set() + self._skip_mcp_refresh = True + self.compression_enabled = False + self.context_compressor = types.SimpleNamespace( + protect_first_n=2, protect_last_n=2 + ) + self._cached_system_prompt = "SYSTEM" + self._memory_store = None + self._memory_manager = None + self._memory_nudge_interval = 0 + self._turns_since_memory = 0 + self._user_turn_count = 0 + self._todo_store = _FakeTodoStore() + self._tool_guardrails = _FakeGuardrails() + self._compression_warning = None + self._interrupt_requested = False + self._memory_write_origin = "assistant_tool" + self._stream_context_scrubber = None + self._stream_think_scrubber = None + # Captures the user message's api_content at persist time, proving + # the stamp lands BEFORE the early persist writes the row. + self.api_content_at_persist = "" + + def _ensure_db_session(self): + pass + + def _restore_primary_runtime(self): + pass + + def _cleanup_dead_connections(self): + return False + + def _emit_status(self, _msg): + pass + + def _replay_compression_warning(self): + pass + + def _hydrate_todo_store(self, *_a, **_k): + pass + + def _safe_print(self, *_a, **_k): + pass + + def _persist_session(self, messages, _history=None): + self.api_content_at_persist = messages[-1].get("api_content") + + +def _build(agent, **overrides): + kwargs = dict( + agent=agent, + user_message="hello", + system_message=None, + conversation_history=None, + task_id=None, + stream_callback=None, + persist_user_message=None, + restore_or_build_system_prompt=lambda *a, **k: None, + install_safe_stdio=lambda: None, + sanitize_surrogates=lambda s: s, + summarize_user_message_for_log=lambda s: s, + set_session_context=lambda _sid: None, + set_current_write_origin=lambda _o: None, + ra=lambda: types.SimpleNamespace(_set_interrupt=lambda *a, **k: None), + ) + kwargs.update(overrides) + return build_turn_context(**kwargs) + + +@pytest.fixture(autouse=True) +def _stub_runtime_main(): + with patch("agent.auxiliary_client.set_runtime_main", lambda *a, **k: None): + yield + + +class TestPrologueStamping: + def test_stamps_api_content_from_plugin_context(self): + agent = _FakeAgent() + with patch( + "hermes_cli.plugins.invoke_hook", + return_value=[{"context": "PLUGIN-CTX"}], + ): + ctx = _build(agent) + msg = ctx.messages[ctx.current_turn_user_idx] + assert msg["content"] == "hello" # clean content untouched + assert msg["api_content"] == compose_user_api_content( + "hello", ctx.ext_prefetch_cache, ctx.plugin_user_context + ) + assert msg["api_content"] == "hello\n\nPLUGIN-CTX" + # The early persist saw the stamped sidecar (written in one insert). + assert agent.api_content_at_persist == "hello\n\nPLUGIN-CTX" + + def test_no_stamp_without_injections(self): + agent = _FakeAgent() + with patch("hermes_cli.plugins.invoke_hook", return_value=[]): + ctx = _build(agent) + assert "api_content" not in ctx.messages[ctx.current_turn_user_idx] + assert agent.api_content_at_persist is None + + def test_no_stamp_for_codex_app_server(self): + """codex_app_server turns bypass the api_messages build, so the + injected bytes are never sent — stamping would persist a lie.""" + agent = _FakeAgent() + agent.api_mode = "codex_app_server" + with patch( + "hermes_cli.plugins.invoke_hook", + return_value=[{"context": "PLUGIN-CTX"}], + ): + ctx = _build(agent) + assert "api_content" not in ctx.messages[ctx.current_turn_user_idx] + + +# --------------------------------------------------------------------------- +# Flush: persist-override rows keep the sent bytes in the sidecar (#48677) +# --------------------------------------------------------------------------- + +class TestFlushOverrideSidecar: + def _make_agent(self, db, sid): + from run_agent import AIAgent + + agent = AIAgent( + api_key="test-key", + base_url="https://openrouter.ai/api/v1", + quiet_mode=True, + skip_context_files=True, + skip_memory=True, + session_db=db, + session_id=sid, + ) + agent._session_db_created = True + return agent + + def test_override_moves_sent_bytes_to_sidecar(self, tmp_path): + db = SessionDB(db_path=tmp_path / "state.db") + sid = "sess-ov" + db.create_session(session_id=sid, source="cli") + try: + agent = self._make_agent(db, sid) + live = "[gateway note] observed context\n\nactual question" + messages = [{"role": "user", "content": live}] + agent._persist_user_message_idx = 0 + agent._persist_user_message_override = "actual question" + agent._persist_user_message_timestamp = None + agent._flush_messages_to_session_db(messages, None) + + msgs = db.get_messages_as_conversation(sid) + assert msgs[0]["content"] == "actual question" + assert msgs[0]["api_content"] == live + # The live dict is never mutated by the flush. + assert messages[0]["content"] == live + finally: + db.close() + + def test_stamped_sidecar_wins_over_override_derivation(self, tmp_path): + """When the prologue already stamped api_content (injections), the + flush must keep those bytes — they are what actually went out.""" + db = SessionDB(db_path=tmp_path / "state.db") + sid = "sess-ov2" + db.create_session(session_id=sid, source="cli") + try: + agent = self._make_agent(db, sid) + messages = [ + { + "role": "user", + "content": "live text", + "api_content": "live text\n\nPLUGIN-CTX", + } + ] + agent._persist_user_message_idx = 0 + agent._persist_user_message_override = "clean text" + agent._persist_user_message_timestamp = None + agent._flush_messages_to_session_db(messages, None) + + msgs = db.get_messages_as_conversation(sid) + assert msgs[0]["content"] == "clean text" + assert msgs[0]["api_content"] == "live text\n\nPLUGIN-CTX" + finally: + db.close() + + +# --------------------------------------------------------------------------- +# End-to-end wire invariant against an in-process mock provider +# --------------------------------------------------------------------------- + +class _MockHandler(BaseHTTPRequestHandler): + captured_requests: list = [] + response_queue: list = [] + + def do_POST(self): # noqa: N802 (http.server API) + length = int(self.headers.get("Content-Length", 0)) + req = json.loads(self.rfile.read(length).decode()) + type(self).captured_requests.append(req) + is_stream = req.get("stream") is True + if type(self).response_queue: + resp = type(self).response_queue.pop(0) + else: + resp = _text_resp("DONE") + msg = resp["choices"][0]["message"] + if is_stream: + content = msg.get("content") or "" + tcs = msg.get("tool_calls") + self.send_response(200) + self.send_header("Content-Type", "text/event-stream") + self.end_headers() + chunks = [{"id": "m", "choices": [{"index": 0, "delta": {"role": "assistant", "content": ""}, "finish_reason": None}]}] + if content: + chunks.append({"id": "m", "choices": [{"index": 0, "delta": {"content": content}, "finish_reason": None}]}) + if tcs: + for ti, tc in enumerate(tcs): + chunks.append({"id": "m", "choices": [{"index": 0, "delta": {"tool_calls": [{ + "index": ti, "id": tc["id"], "type": "function", + "function": {"name": tc["function"]["name"], "arguments": tc["function"]["arguments"]}}]}, "finish_reason": None}]}) + chunks.append({"id": "m", "choices": [{"index": 0, "delta": {}, "finish_reason": "tool_calls" if tcs else "stop"}]}) + for c in chunks: + self.wfile.write(f"data: {json.dumps(c)}\n\n".encode()) + self.wfile.write(b"data: [DONE]\n\n") + self.wfile.flush() + else: + body = json.dumps(resp).encode() + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def log_message(self, *a, **kw): + pass + + +def _tc_resp(name: str, args: str = "{}") -> dict: + return { + "id": "m", + "choices": [{"index": 0, "message": { + "role": "assistant", "content": "", + "tool_calls": [{"id": "call_1", "type": "function", + "function": {"name": name, "arguments": args}}]}, + "finish_reason": "tool_calls"}], + "usage": {"prompt_tokens": 10, "completion_tokens": 0, "total_tokens": 10}, + } + + +def _text_resp(text: str) -> dict: + return { + "id": "m", + "choices": [{"index": 0, "message": {"role": "assistant", "content": text}, "finish_reason": "stop"}], + "usage": {"prompt_tokens": 10, "completion_tokens": 0, "total_tokens": 10}, + } + + +@pytest.fixture() +def wire_env(): + """Mock provider + isolated HERMES_HOME + a shared SessionDB. + + Yields (make_agent, handler, db, sid): ``make_agent()`` builds a fresh + AIAgent bound to the shared DB/session, so a second call models a + process-restart turn N+1 that reloads history from the store. + """ + _MockHandler.captured_requests = [] + _MockHandler.response_queue = [] + srv = HTTPServer(("127.0.0.1", 0), _MockHandler) + port = srv.server_address[1] + t = threading.Thread(target=srv.serve_forever, daemon=True) + t.start() + + test_home = tempfile.mkdtemp(prefix="hermes_api_content_") + os.makedirs(os.path.join(test_home, ".hermes")) + prev_home = os.environ.get("HERMES_HOME") + os.environ["HERMES_HOME"] = os.path.join(test_home, ".hermes") + + from run_agent import AIAgent + + from pathlib import Path + + db = SessionDB(db_path=Path(test_home) / "state.db") + sid = "sess-wire" + + def make_agent(): + agent = AIAgent( + api_key="test-key", base_url=f"http://127.0.0.1:{port}/v1", + provider="openai-compat", model="test-model", + max_iterations=10, enabled_toolsets=[], + quiet_mode=True, skip_context_files=True, skip_memory=True, + save_trajectories=False, platform="cli", + session_db=db, session_id=sid, + ) + agent.valid_tool_names = {"read_file"} + return agent + + try: + with patch( + "hermes_cli.plugins.invoke_hook", + side_effect=lambda hook, **kw: ( + [{"context": "PLUGIN-CTX"}] if hook == "pre_llm_call" else [] + ), + ): + yield make_agent, _MockHandler, db, sid + finally: + srv.shutdown() + db.close() + shutil.rmtree(test_home, ignore_errors=True) + if prev_home is None: + os.environ.pop("HERMES_HOME", None) + else: + os.environ["HERMES_HOME"] = prev_home + + +def _chat_requests(handler) -> list: + # The model context-length probe also hits the mock; keep only + # chat-completions payloads. + return [r for r in handler.captured_requests if "messages" in r] + + +def _user_messages(req: dict) -> list: + return [m for m in req.get("messages", []) if m.get("role") == "user"] + + +class TestWireInvariant: + def test_injection_sent_stamped_and_stable_within_turn(self, wire_env): + """The current turn's user message goes out with the injected context, + the sidecar equals the sent bytes exactly, the field never reaches the + wire, and every pass within the turn sends identical bytes.""" + make_agent, handler, db, sid = wire_env + agent = make_agent() + # Two API calls in one turn: tool call, then final text. + handler.response_queue.append(_tc_resp("read_file", '{"file_path": "/nonexistent-path"}')) + handler.response_queue.append(_text_resp("done")) + + agent.run_conversation("hello please", conversation_history=[], task_id="t") + + reqs = _chat_requests(handler) + assert len(reqs) == 2 + sent_1 = _user_messages(reqs[0])[0]["content"] + sent_2 = _user_messages(reqs[1])[0]["content"] + assert sent_1 == "hello please\n\nPLUGIN-CTX" + assert sent_2 == sent_1 # repeated builds: identical bytes + + # The sidecar never reaches the provider. + for req in reqs: + for m in req.get("messages", []): + assert "api_content" not in m + + # Persisted row: clean content + exact sent bytes in the sidecar. + user_rows = [r for r in db.get_messages(sid) if r["role"] == "user"] + assert user_rows[0]["content"] == "hello please" + assert user_rows[0]["api_content"] == sent_1 + + def test_next_turn_replays_previous_turn_bytes(self, wire_env): + """The cache invariant: the serialized user message replayed in turn + N+1 (history reloaded from the store) EQUALS the bytes turn N sent.""" + make_agent, handler, db, sid = wire_env + + # ── Turn N ── + agent1 = make_agent() + agent1.run_conversation("hello please", conversation_history=[], task_id="t1") + turn_n_user = _user_messages(_chat_requests(handler)[0])[0] + turn_n_bytes = json.dumps(turn_n_user, sort_keys=True) + + # ── Turn N+1: fresh agent, history reloaded from the store ── + history = db.get_messages_as_conversation(sid) + # The stored history carries the sidecar, not the injected content. + assert history[0]["content"] == "hello please" + assert history[0]["api_content"] == turn_n_user["content"] + + handler.captured_requests = [] + agent2 = make_agent() + agent2.run_conversation( + "second question", conversation_history=history, task_id="t2" + ) + + replayed = _user_messages(_chat_requests(handler)[0])[0] + assert json.dumps(replayed, sort_keys=True) == turn_n_bytes + + # And the new current-turn message got its own injection + sidecar. + current = _user_messages(_chat_requests(handler)[0])[-1] + assert current["content"] == "second question\n\nPLUGIN-CTX" + + +# --------------------------------------------------------------------------- +# Review fixes: re-anchoring, MoA, in-place compaction backfill, override +# guard, sanitize-divergence capture, max-iterations replay, replay cleanup +# --------------------------------------------------------------------------- + +from agent.turn_context import reanchor_current_turn_user_idx + + +class TestReanchorCurrentTurnUserIdx: + def test_exact_match_beats_later_todo_snapshot(self): + """compress_context can append a todo-snapshot USER message after the + surviving current-turn copy — the anchor must stay on the real turn.""" + messages = [ + {"role": "assistant", "content": "summary"}, + {"role": "user", "content": "hello"}, + {"role": "user", "content": "## Current TODOs\n- [ ] thing"}, + ] + assert reanchor_current_turn_user_idx(messages, "hello") == 1 + + def test_most_recent_duplicate_wins(self): + messages = [ + {"role": "user", "content": "ok"}, + {"role": "assistant", "content": "a"}, + {"role": "user", "content": "ok"}, + ] + assert reanchor_current_turn_user_idx(messages, "ok") == 2 + + def test_falls_back_to_last_user_without_exact_match(self): + """Merge-summary-into-tail rewrites the content; the trackers still + need a live anchor.""" + messages = [ + {"role": "user", "content": "[prior context]\nsummary\nhello"}, + {"role": "assistant", "content": "a"}, + ] + assert reanchor_current_turn_user_idx(messages, "hello") == 0 + + def test_minus_one_when_no_user_message(self): + messages = [{"role": "assistant", "content": "a"}] + assert reanchor_current_turn_user_idx(messages, "hello") == -1 + assert reanchor_current_turn_user_idx([], "hello") == -1 + + def test_non_dict_entries_ignored(self): + messages = ["junk", {"role": "user", "content": "hello"}, None] + assert reanchor_current_turn_user_idx(messages, "hello") == 1 + + +class TestPrologueMoaAndInPlaceBackfill: + def test_no_stamp_for_moa_turns(self): + """MoA appends per-call aggregated context to the API copy AFTER the + composition — a stamped sidecar would persist bytes that never match + the wire.""" + agent = _FakeAgent() + with patch( + "hermes_cli.plugins.invoke_hook", + return_value=[{"context": "PLUGIN-CTX"}], + ): + ctx = _build(agent, moa_active=True) + assert "api_content" not in ctx.messages[ctx.current_turn_user_idx] + + def test_inplace_compaction_backfills_sidecar_into_db(self): + """In-place preflight compaction inserts the current-turn user row + BEFORE the stamp (archive_and_compact), and the crash persist + identity-skips every compacted dict — the stamp must be pushed into + the existing row directly.""" + agent = _FakeAgent() + agent.compression_enabled = True + agent._session_db = MagicMock() + + calls = {"n": 0} + + def _should_compress(_tokens): + calls["n"] += 1 + return calls["n"] == 1 # compress once, then stop + + agent.context_compressor = types.SimpleNamespace( + protect_first_n=0, + protect_last_n=0, + threshold_tokens=1, + context_length=1000, + last_prompt_tokens=-1, + should_compress=_should_compress, + should_defer_preflight_to_real_usage=lambda _t: False, + get_active_compression_failure_cooldown=lambda: None, + ) + + def _compress(messages, _system, approx_tokens=None, task_id=None): + # Emulate compress_context in in_place mode: archive_and_compact + # already inserted these rows (api_content=NULL — the stamp has + # not happened yet), fresh copies replace the live dicts. + agent._last_compaction_in_place = True + return ( + [ + {"role": "assistant", "content": "compaction summary"}, + dict(messages[-1]), # surviving current-turn user copy + ], + "SYSTEM", + ) + + agent._compress_context = _compress + + big = "x" * 4000 + history = [ + {"role": "user", "content": big}, + {"role": "assistant", "content": big}, + ] + with patch( + "hermes_cli.plugins.invoke_hook", + return_value=[{"context": "PLUGIN-CTX"}], + ): + ctx = _build(agent, conversation_history=history) + + msg = ctx.messages[ctx.current_turn_user_idx] + assert msg["content"] == "hello" + assert msg["api_content"] == "hello\n\nPLUGIN-CTX" + agent._session_db.set_latest_user_api_content.assert_called_once_with( + "sess-1", "hello", "hello\n\nPLUGIN-CTX" + ) + + +class TestSetLatestUserApiContent: + def _open(self, tmp_path): + db = SessionDB(db_path=tmp_path / "state.db") + db.create_session("s1", source="cli") + return db + + def test_updates_newest_active_user_row(self, tmp_path): + db = self._open(tmp_path) + try: + db.append_message("s1", "user", content="q1") + db.append_message("s1", "assistant", content="a1") + db.append_message("s1", "user", content="q2") + assert db.set_latest_user_api_content("s1", "q2", "q2\n\nCTX") == 1 + msgs = db.get_messages_as_conversation("s1") + assert "api_content" not in msgs[0] + assert msgs[2]["api_content"] == "q2\n\nCTX" + finally: + db.close() + + def test_content_mismatch_writes_nothing(self, tmp_path): + db = self._open(tmp_path) + try: + db.append_message("s1", "user", content="q1") + assert db.set_latest_user_api_content("s1", "other", "other+CTX") == 0 + msgs = db.get_messages_as_conversation("s1") + assert "api_content" not in msgs[0] + finally: + db.close() + + +class TestFlushCompressedSummaryOverrideGuard: + def _make_agent(self, db, sid): + from run_agent import AIAgent + + agent = AIAgent( + api_key="test-key", + base_url="https://openrouter.ai/api/v1", + quiet_mode=True, + skip_context_files=True, + skip_memory=True, + session_db=db, + session_id=sid, + ) + agent._session_db_created = True + return agent + + def test_override_skipped_for_compression_merged_row(self, tmp_path): + """The #48677 override must not replace a compression-merged user + message: that would silently drop the compaction summary from the + durable clean transcript.""" + from agent.context_compressor import COMPRESSED_SUMMARY_METADATA_KEY + + db = SessionDB(db_path=tmp_path / "state.db") + sid = "sess-merged" + db.create_session(session_id=sid, source="cli") + try: + agent = self._make_agent(db, sid) + merged = "[prior context]\ncompaction summary\n\nactual question" + messages = [ + { + "role": "user", + "content": merged, + COMPRESSED_SUMMARY_METADATA_KEY: True, + } + ] + agent._persist_user_message_idx = 0 + agent._persist_user_message_override = "actual question" + agent._persist_user_message_timestamp = None + agent._flush_messages_to_session_db(messages, None) + + msgs = db.get_messages_as_conversation(sid) + assert msgs[0]["content"] == merged # summary survives + assert "api_content" not in msgs[0] # wire == row, no sidecar + finally: + db.close() + + +class TestFlushSanitizeDivergenceCapture: + def _make_agent(self, db, sid): + from run_agent import AIAgent + + agent = AIAgent( + api_key="test-key", + base_url="https://openrouter.ai/api/v1", + quiet_mode=True, + skip_context_files=True, + skip_memory=True, + session_db=db, + session_id=sid, + ) + agent._session_db_created = True + return agent + + def test_user_content_sanitize_would_rewrite_is_captured(self, tmp_path): + """get_messages_as_conversation strips fences on + load; the sent bytes must survive in the sidecar so a reloaded + session replays what was actually on the wire.""" + from agent.memory_manager import sanitize_context + + db = SessionDB(db_path=tmp_path / "state.db") + sid = "sess-fence" + db.create_session(session_id=sid, source="cli") + try: + agent = self._make_agent(db, sid) + raw = "what does a literal tag do?" + assert sanitize_context(raw) != raw # test precondition + messages = [ + {"role": "user", "content": raw}, + {"role": "assistant", "content": "it fences recalled memory "}, + ] + agent._flush_messages_to_session_db(messages, None) + + msgs = db.get_messages_as_conversation(sid) + assert msgs[0]["content"] == sanitize_context(raw).strip() + assert msgs[0]["api_content"] == raw + assert msgs[1]["api_content"] == ( + "it fences recalled memory " + ) + finally: + db.close() + + def test_plain_whitespace_padding_not_captured(self, tmp_path): + """The api build strips every outgoing content string, so mere + surrounding whitespace replays identically — no sidecar bloat.""" + db = SessionDB(db_path=tmp_path / "state.db") + sid = "sess-ws" + db.create_session(session_id=sid, source="cli") + try: + agent = self._make_agent(db, sid) + messages = [{"role": "user", "content": " hello \n"}] + agent._flush_messages_to_session_db(messages, None) + msgs = db.get_messages_as_conversation(sid) + assert "api_content" not in msgs[0] + finally: + db.close() + + +class TestMaxIterationsSummaryReplay: + def test_summary_request_substitutes_sidecar_bytes(self): + """The forced-summary request must replay the same bytes every + main-loop call sent — popping the sidecar without substituting sends + CLEAN content and diverges the prefix at the earliest injected + message, exactly when the context is largest.""" + from run_agent import AIAgent + from agent.chat_completion_helpers import handle_max_iterations + + agent = AIAgent( + api_key="test-key", + base_url="https://openrouter.ai/api/v1", + quiet_mode=True, + skip_context_files=True, + skip_memory=True, + ) + agent._cached_system_prompt = "SYS" + + captured = {} + + class _Completions: + def create(self, **kwargs): + captured.update(kwargs) + return "RAW-RESPONSE" + + client = types.SimpleNamespace( + chat=types.SimpleNamespace(completions=_Completions()) + ) + transport = types.SimpleNamespace( + normalize_response=lambda _r: types.SimpleNamespace(content="SUMMARY") + ) + + messages = [ + {"role": "user", "content": "q1", "api_content": "q1\n\nPLUGIN-CTX"}, + {"role": "assistant", "content": "a1"}, + ] + with patch.object( + agent, "_ensure_primary_openai_client", return_value=client + ), patch.object(agent, "_get_transport", return_value=transport): + out = handle_max_iterations(agent, messages, 5) + + assert out == "SUMMARY" + sent_users = [ + m for m in captured["messages"] if m.get("role") == "user" + ] + assert sent_users[0]["content"] == "q1\n\nPLUGIN-CTX" + for m in captured["messages"]: + assert "api_content" not in m + # The live history dict is never mutated. + assert messages[0]["content"] == "q1" + assert messages[0]["api_content"] == "q1\n\nPLUGIN-CTX" + + +class TestSessionRowExistsBeforePreflightCompaction: + """Moving the crash persist after prefetch/pre_llm_call (one write with + the final sidecar) must NOT delay session-row creation: with PRAGMA + foreign_keys=ON, in-place ``archive_and_compact`` inserts message rows + referencing ``sessions(id)`` and rotation creates a child whose + ``parent_session_id`` points at the current row — both run during + preflight compression, which a fresh oversized first turn reaches + before the delayed persist. Drives the real ``compress_context`` path + against a real, empty SessionDB.""" + + def _make_agent(self, db, sid, *, in_place): + from run_agent import AIAgent + + with patch.dict(os.environ, {"OPENROUTER_API_KEY": "test-key"}): + agent = AIAgent( + api_key="test-key", + base_url="https://openrouter.ai/api/v1", + model="test/model", + platform="cli", + quiet_mode=True, + session_db=db, + session_id=sid, + skip_context_files=True, + skip_memory=True, + ) + # Fresh first turn: no session row has been created yet. + assert not agent._session_db_created + agent._cached_system_prompt = "SYSTEM" + agent.compression_enabled = True + agent.compression_in_place = in_place + agent._compression_feasibility_checked = True + + calls = {"n": 0} + + def _should_compress(_tokens): + calls["n"] += 1 + return calls["n"] == 1 # compress once, then stop + + seen = {} + compacted = [ + {"role": "assistant", "content": "[CONTEXT COMPACTION] summary"}, + {"role": "user", "content": "hello"}, + ] + + def _compress(_messages, **_kwargs): + # The ordering invariant under test: the row must already exist + # when compression starts — the DB writes right after this call + # (archive_and_compact / child create_session) reference it. + seen["row_at_compress"] = db.get_session(sid) + return compacted + + compressor = MagicMock() + compressor.protect_first_n = 0 + compressor.protect_last_n = 0 + compressor.threshold_tokens = 1 + compressor.context_length = 1000 + compressor.last_prompt_tokens = -1 + compressor.should_compress = _should_compress + compressor.should_defer_preflight_to_real_usage = lambda _t: False + compressor.get_active_compression_failure_cooldown = lambda: None + compressor.compress = _compress + compressor.compression_count = 1 + compressor._last_summary_error = None + compressor._last_compress_aborted = False + compressor._last_summary_auth_failure = False + compressor._last_aux_model_failure_model = None + compressor._last_aux_model_failure_error = None + agent.context_compressor = compressor + return agent, seen + + def _oversized_history(self): + big = "x" * 4000 + return [ + {"role": "user", "content": big}, + {"role": "assistant", "content": big}, + ] + + def test_in_place_first_turn_compaction_persists(self, tmp_path): + db = SessionDB(db_path=tmp_path / "state.db") + sid = "sess-fresh-inplace" + try: + agent, seen = self._make_agent(db, sid, in_place=True) + with patch("hermes_cli.plugins.invoke_hook", return_value=[]): + ctx = _build(agent, conversation_history=self._oversized_history()) + + # The row was created before compression started — without it the + # FK on messages.session_id rejects the compacted rows and the + # compaction write is silently lost. + assert seen["row_at_compress"] is not None + assert agent.session_id == sid # no rotation + assert agent._last_compaction_in_place is True + contents = [m["content"] for m in db.get_messages(sid)] + assert "[CONTEXT COMPACTION] summary" in contents + # And the live context is the compacted set. + assert ctx.messages[ctx.current_turn_user_idx]["content"] == "hello" + finally: + db.close() + + def test_rotation_first_turn_compaction_creates_child(self, tmp_path): + db = SessionDB(db_path=tmp_path / "state.db") + sid = "sess-fresh-rot" + try: + agent, seen = self._make_agent(db, sid, in_place=False) + with patch("hermes_cli.plugins.invoke_hook", return_value=[]): + _build(agent, conversation_history=self._oversized_history()) + + # The parent row existed before compression started — the child + # INSERT's parent_session_id FK needs it; without it + # create_session raises and the orphan-avoidance path rolls the + # id back, so the rotation never happens. + assert seen["row_at_compress"] is not None + child = agent.session_id + assert child != sid + child_row = db.get_session(child) + assert child_row is not None + assert child_row["parent_session_id"] == sid + parent_row = db.get_session(sid) + assert parent_row is not None + assert parent_row["end_reason"] == "compression" + finally: + db.close() + + +class TestStaleConfirmationRedactionDropsSidecar: + def test_redaction_pops_api_content(self): + """The expired-confirmation redaction rewrites content — replaying + the sidecar verbatim would resend the dangerous bytes it just + redacted.""" + from agent.replay_cleanup import strip_stale_dangerous_confirmations + + history = [ + { + "role": "user", + "content": "confirm forced restart", + "api_content": "confirm forced restart\n\nPLUGIN-CTX", + "timestamp": 1000.0, + } + ] + cleaned = strip_stale_dangerous_confirmations( + history, now=1000.0 + 999999.0 + ) + assert cleaned[0]["content"] != "confirm forced restart" + assert "api_content" not in cleaned[0] diff --git a/tests/gateway/test_replay_entry_fields.py b/tests/gateway/test_replay_entry_fields.py index c0891d3721fc..526c761044a1 100644 --- a/tests/gateway/test_replay_entry_fields.py +++ b/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 "} + entry = _build_replay_entry("assistant", "a", msg) + assert entry["api_content"] == "a " + + 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]