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]