From c0c76a47153398953c718ca729bc5192da1e63ac Mon Sep 17 00:00:00 2001 From: Soju06 Date: Tue, 14 Jul 2026 08:53:42 +0000 Subject: [PATCH] perf(gateway): byte-stable session context prompts The per-message ephemeral context prompt re-renders every turn, and any byte change (Discord auto-thread rename, reset notes, voice channel state) both breaks the provider prompt-cache prefix at the head of every request and changes the gateway agent-cache signature, forcing a full agent rebuild per message. Pin the rendered block per session keyed by a hash of exactly the fields it renders, so only a real input change (rename, topic edit, /sethome, redact_pii flip) re-renders; deliver one-shot per-turn facts (auto-reset note, first-contact intro, voice-channel changes) on the current user message via the api_content sidecar instead of the system prompt; sort get_connected_platforms for byte-stable ordering. --- agent/turn_context.py | 63 ++++ gateway/config.py | 11 +- gateway/run.py | 241 +++++++++++-- gateway/session.py | 10 + tests/agent/test_gateway_turn_sidecar.py | 207 ++++++++++++ tests/gateway/test_prompt_tail_freeze.py | 410 +++++++++++++++++++++++ 6 files changed, 920 insertions(+), 22 deletions(-) create mode 100644 tests/agent/test_gateway_turn_sidecar.py create mode 100644 tests/gateway/test_prompt_tail_freeze.py diff --git a/agent/turn_context.py b/agent/turn_context.py index b40329628931..8807407278c9 100644 --- a/agent/turn_context.py +++ b/agent/turn_context.py @@ -121,6 +121,46 @@ def extract_api_content_sidecar(msg: Mapping[str, Any]) -> Optional[str]: return v if isinstance(v, str) else None +def consume_gateway_turn_context_notes(agent: Any) -> str: + """Pop the gateway's per-turn must-deliver notes off the agent (one-shot). + + The gateway relocates volatile per-turn facts OUT of the ephemeral system + prompt (auto-reset notes, the first-contact intro, voice-channel changes) + and delivers them on the current user message via the api_content sidecar + instead, so the composed system prompt stays byte-stable turn-over-turn. + It stages the rendered notes on ``agent._gateway_turn_context_notes`` + right before ``run_conversation``; this consumes them so a cached agent + can never replay a stale note on a later turn. + """ + notes = getattr(agent, "_gateway_turn_context_notes", "") or "" + if hasattr(agent, "_gateway_turn_context_notes"): + try: + agent._gateway_turn_context_notes = "" + except Exception: + pass + return notes if isinstance(notes, str) else "" + + +def append_notes_to_multimodal_content(content: Any, notes: str) -> bool: + """Deliver must-deliver notes on a multimodal (list) user message. + + ``compose_user_api_content`` returns ``None`` for non-string content, so + sidecar-borne facts would silently drop on image/attachment turns. For + gateway must-deliver notes we instead append a text part to the content + list in place — the part becomes durable message content (persisted and + replayed as-is), which keeps the wire and the transcript byte-identical. + + Returns ``True`` when a part was appended. + """ + if not notes or not isinstance(content, list): + return False + try: + content.append({"type": "text", "text": notes}) + return True + except Exception: + return False + + def reanchor_current_turn_user_idx(messages: List[Any], user_message: Any) -> int: """Locate this turn's user message after compaction rebuilt ``messages``. @@ -702,6 +742,29 @@ def build_turn_context( except Exception as exc: logger.warning("pre_llm_call hook failed: %s", exc) + # Gateway must-deliver notes (auto-reset note, first-contact intro, + # voice-channel change) ride the same user-message injection channel as + # plugin context so the ephemeral system prompt can stay byte-stable. + # One-shot: staged by the gateway right before this turn, consumed here. + # Multimodal (list) content can't take the string sidecar — append a + # durable text part instead of dropping the fact. + _gateway_notes = consume_gateway_turn_context_notes(agent) + if _gateway_notes: + _gw_turn_content = ( + messages[current_turn_user_idx].get("content") + if 0 <= current_turn_user_idx < len(messages) + and isinstance(messages[current_turn_user_idx], dict) + else None + ) + if isinstance(_gw_turn_content, list): + append_notes_to_multimodal_content(_gw_turn_content, _gateway_notes) + else: + plugin_user_context = ( + plugin_user_context + "\n\n" + _gateway_notes + if plugin_user_context + else _gateway_notes + ) + # Per-turn file-mutation verifier state. agent._turn_failed_file_mutations = {} agent._turn_file_mutation_paths = set() diff --git a/gateway/config.py b/gateway/config.py index f321a322b8a4..bc2c92387632 100644 --- a/gateway/config.py +++ b/gateway/config.py @@ -857,14 +857,21 @@ class GatewayConfig: ) def get_connected_platforms(self) -> List[Platform]: - """Return list of platforms that are enabled and configured.""" + """Return list of platforms that are enabled and configured. + + Sorted by platform value so the rendered "Connected Platforms" list + (and the home-channel blocks derived from it) is byte-stable across + gateway restarts and mid-process platform registration — dict + insertion order is not a stable contract and a reorder busts the + prompt cache without any semantic change. + """ connected = [] for platform, config in self.platforms.items(): if not config.enabled: continue if self._is_platform_connected(platform, config): connected.append(platform) - return connected + return sorted(connected, key=lambda p: str(p.value)) def _is_platform_connected(self, platform: Platform, config: PlatformConfig) -> bool: """Check whether a single platform is sufficiently configured.""" diff --git a/gateway/run.py b/gateway/run.py index c6e4e4de0c70..a2981ae2469a 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -3021,6 +3021,9 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew _session_model_overrides: Dict[str, Dict[str, str]] = {} _pending_one_turn_model_restores: Dict[str, Dict[str, Any]] = {} _session_reasoning_overrides: Dict[str, Dict[str, Any]] = {} + _pending_turn_sidecar_notes: Dict[str, List[str]] = {} + _session_ephemeral_pin: Dict[str, tuple] = {} + _session_vc_last: Dict[str, str] = {} _startup_restore_in_progress: bool = False # Loop-liveness heartbeat / shutdown-watchdog handles (#66892). Class-level # defaults so partial construction in tests doesn't blow up on access; the @@ -3216,6 +3219,18 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew # Per-session reasoning effort overrides from /reasoning. # Key: session_key, Value: parsed reasoning config dict. self._session_reasoning_overrides: Dict[str, Dict[str, Any]] = {} + # Per-turn must-deliver notes relocated out of the ephemeral system + # prompt (auto-reset note, first-contact intro, voice-channel change). + # Staged by _handle_message_with_agent, consumed once by run_sync and + # delivered on the current user message (api_content sidecar). + self._pending_turn_sidecar_notes: Dict[str, List[str]] = {} + # Pinned session-context bytes keyed by the renderer-input change + # key. Key hit → reuse pinned bytes verbatim; key miss → re-render + # + re-pin (a legitimate cache bust). + self._session_ephemeral_pin: Dict[str, tuple] = {} + # Last voice-channel context delivered per session — the VC note is + # injected only when the live state differs from this value. + self._session_vc_last: Dict[str, str] = {} self._kanban_notifier_profile = self._active_profile_name() # Teams meeting pipeline runtime (bound later when msgraph_webhook adapter exists). self._teams_pipeline_runtime = None @@ -8332,6 +8347,14 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew self._set_session_reasoning_override(key, None) if hasattr(self, "_pending_model_notes"): self._pending_model_notes.pop(key, None) + # Session keys are source-derived and REUSED by the + # next conversation: a staged-but-never-consumed + # sidecar note (turn aborted between staging and + # run_sync) must not leak into a future session's + # first user message. + _psn = getattr(self, "_pending_turn_sidecar_notes", None) + if isinstance(_psn, dict): + _psn.pop(key, None) # Clear per-session model cache so a resumed turn # resolves from current config, not a stale fallback # cached before the session went idle (mirrors /new @@ -12034,10 +12057,24 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew except Exception: pass - # Build the context prompt to inject - context_prompt = build_session_context_prompt(context, redact_pii=_redact_pii) - - # If the previous session expired and was auto-reset, prepend a notice + # Build the context prompt to inject. The render is pinned per + # session, keyed by a hash of the exact renderer inputs + # (_ephemeral_change_key). A key hit reuses the pinned bytes verbatim + # so the composed system prompt cannot drift turn-over-turn; a key + # miss (thread rename, /sethome, redact_pii flip, ...) re-renders + # once — the only legitimate cache busts. + context_prompt = self._pinned_session_context_prompt( + context, _redact_pii, session_key + ) + + # Per-turn must-deliver notes. These used to be appended to + # context_prompt (the ephemeral system prompt), which guaranteed a + # turn1→turn2 system-prompt diff and a full agent rebuild. They now + # ride the current user message via the api_content sidecar instead + # (staged below, consumed in run_sync → build_turn_context). + turn_sidecar_notes: List[str] = [] + + # If the previous session expired and was auto-reset, deliver a notice # so the agent knows this is a fresh conversation (not an intentional /reset). if _was_auto_reset: reset_reason = getattr(session_entry, 'auto_reset_reason', None) or 'idle' @@ -12049,7 +12086,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew context_note = "[System note: The previous gateway session could not be recovered after a restart (API recovery timed out). This is a fresh conversation — use /resume to restore history if needed.]" else: context_note = "[System note: The user's previous session expired due to inactivity. This is a fresh conversation with no prior context.]" - context_prompt = context_note + "\n\n" + context_prompt + turn_sidecar_notes.append(context_note) # Send a user-facing notification explaining the reset, unless: # - notifications are disabled in config @@ -12559,11 +12596,14 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew "Session hygiene auto-compress failed: %s", e ) - # First-message onboarding -- only on the very first interaction ever + # First-message onboarding -- only on the very first interaction ever. + # Delivered on the current user message (sidecar), NOT the ephemeral + # system prompt: present-on-turn-1/absent-on-turn-2 was a guaranteed + # system-prompt diff and agent rebuild. if not history and not await self.async_session_store.has_any_sessions(): # Default first-contact note: a brief self-introduction. _intro_note = ( - "\n\n[System note: This is the user's very first message ever. " + "[System note: This is the user's very first message ever. " "Briefly introduce yourself and mention that /help shows available commands. " "Keep the introduction concise -- one or two sentences max.]" ) @@ -12586,16 +12626,16 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew profile_build_mode(_onb_cfg) == "ask" and not is_seen(_onb_cfg, PROFILE_BUILD_FLAG) ): - context_prompt += profile_build_directive() + turn_sidecar_notes.append(profile_build_directive().strip()) mark_seen(_hermes_home / "config.yaml", PROFILE_BUILD_FLAG) else: - context_prompt += _intro_note + turn_sidecar_notes.append(_intro_note) except Exception as _pb_err: logger.debug( "Profile-build onboarding directive failed, using plain intro: %s", _pb_err, ) - context_prompt += _intro_note + turn_sidecar_notes.append(_intro_note) # One-time prompt if no home channel is set for this platform # Skip for webhooks - they deliver directly to configured targets (github_comment, etc.) @@ -12653,17 +12693,18 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew await self._deliver_platform_notice(source, notice) # ----------------------------------------------------------------- - # Voice channel awareness — inject current voice channel state - # into context so the agent knows who is in the channel and who - # is speaking, without needing a separate tool call. + # Voice channel awareness — deliver current voice channel state so + # the agent knows who is in the channel and who is speaking, without + # needing a separate tool call. Delivered on the current user + # message and ONLY when it changed since the previous turn: the + # member/speaking serialization differs essentially every turn, and + # appending it to the ephemeral system prompt forced a full agent + # rebuild + prompt-cache re-key per message. The system prompt + # carries a static pointer line instead (gateway/session.py). # ----------------------------------------------------------------- - if source.platform == Platform.DISCORD: - adapter = self.adapters.get(Platform.DISCORD) - guild_id = self._get_guild_id(event) - if guild_id and adapter and hasattr(adapter, "get_voice_channel_context"): - vc_context = adapter.get_voice_channel_context(guild_id) - if vc_context: - context_prompt += f"\n\n{vc_context}" + _vc_note = self._voice_channel_sidecar_note(event, source, session_key) + if _vc_note: + turn_sidecar_notes.append(_vc_note) # ----------------------------------------------------------------- # Auto-analyze images sent by the user @@ -12722,6 +12763,13 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew except Exception as _ts_err: logger.debug("Message timestamp injection failed (non-fatal): %s", _ts_err) + # Stage the collected must-deliver notes for this turn's agent run + # (one-shot; consumed in run_sync). Staged AFTER the message_text + # early-out above so an aborted turn cannot leak its notes into the + # next turn's user message. + if turn_sidecar_notes and session_key: + self._set_pending_turn_sidecar_notes(session_key, turn_sidecar_notes) + # Bind this gateway run generation to the adapter's active-session # event so deferred post-delivery callbacks can be released by the # same run that registered them. @@ -17638,6 +17686,141 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew cached[0], cached[1], _live, _snapshot_sid, ) + def _set_pending_turn_sidecar_notes(self, session_key: str, notes: List[str]) -> None: + """Stage per-turn must-deliver notes for the next agent run (one-shot).""" + if not session_key or not notes: + return + if not hasattr(self, "_pending_turn_sidecar_notes"): + self._pending_turn_sidecar_notes = {} + self._pending_turn_sidecar_notes[session_key] = list(notes) + + def _consume_pending_turn_sidecar_notes(self, session_key: str) -> List[str]: + if not session_key: + return [] + notes = getattr(self, "_pending_turn_sidecar_notes", None) + if not isinstance(notes, dict): + return [] + staged = notes.pop(session_key, None) + return list(staged) if isinstance(staged, list) else [] + + def _voice_channel_sidecar_note(self, event, source: SessionSource, session_key: str) -> Optional[str]: + """Return a ``[Voice channel now: ...]`` note when VC state changed. + + Compares the live Discord voice-channel context against the last + value delivered for this session and returns a note only on change + (including leaving the channel). Unchanged state returns ``None`` so + the per-turn member/speaking serialization cannot churn the prompt. + """ + if source.platform != Platform.DISCORD: + return None + adapter = self.adapters.get(Platform.DISCORD) + guild_id = self._get_guild_id(event) + if not (guild_id and adapter and hasattr(adapter, "get_voice_channel_context")): + return None + try: + vc_now = adapter.get_voice_channel_context(guild_id) or "" + except Exception: + logger.debug("voice-channel context read failed", exc_info=True) + return None + if not hasattr(self, "_session_vc_last"): + self._session_vc_last = {} + vc_prev = self._session_vc_last.get(session_key) if session_key else None + if session_key: + self._session_vc_last[session_key] = vc_now + if vc_now == (vc_prev if vc_prev is not None else ""): + return None + if not vc_now: + return "[Voice channel now: not connected to a voice channel]" + return f"[Voice channel now: {vc_now}]" + + def _pinned_session_context_prompt( + self, context, redact_pii: bool, session_key: Optional[str] + ) -> str: + """Return the session-context prompt, pinned per session. + + Key hit → the pinned bytes are reused VERBATIM (immunizes the + composed system prompt against renderer nondeterminism); key miss → + re-render ``build_session_context_prompt`` and re-pin (a legitimate + cache bust: rename, topic edit, /sethome, redact_pii flip, ...). + """ + if not hasattr(self, "_session_ephemeral_pin"): + self._session_ephemeral_pin = {} + _eph_key = self._ephemeral_change_key(context, redact_pii) + _eph_pin = self._session_ephemeral_pin.get(session_key) if session_key else None + if _eph_pin is not None and _eph_pin[0] == _eph_key: + return _eph_pin[1] + text = build_session_context_prompt(context, redact_pii=redact_pii) + if session_key: + self._session_ephemeral_pin[session_key] = (_eph_key, text) + return text + + @staticmethod + def _ephemeral_change_key(context, redact_pii: bool) -> str: + """Hash the exact inputs ``build_session_context_prompt`` renders. + + This key decides when the pinned per-session context-prompt bytes are + reused verbatim vs re-rendered. The maintained invariant (guarded by + the parity test in tests/gateway/test_prompt_tail_freeze.py): any + input whose change alters the rendered bytes MUST appear here — + omission means a stale pinned prompt (cosmetic staleness); inclusion + of an extra field only costs a spurious re-render. + """ + import hashlib + + src = context.source + platform = src.platform.value if src.platform else "" + + discord_ids: tuple = () + discord_tools = "" + if src.platform == Platform.DISCORD: + from gateway.session import _discord_tools_loaded + + discord_tools = "1" if _discord_tools_loaded() else "0" + discord_ids = ( + str(src.guild_id or ""), + str(src.parent_chat_id or ""), + str(src.thread_id or ""), + str(src.chat_id or ""), + # Only PRESENCE is rendered (the id itself is delivered + # per-turn in the user message) — keying on the value would + # re-render every message for zero byte change. + "1" if src.message_id else "0", + ) + + try: + from hermes_constants import display_hermes_home + + home_display = str(display_hermes_home()) + except Exception: + home_display = "" + + key_tuple = ( + platform, + str(src.chat_id or ""), + str(src.thread_id or ""), + str(src.chat_type or ""), + str(src.chat_name or ""), + str(src.chat_topic or ""), + str(src.user_name or ""), + str(src.user_id or ""), + str(getattr(src, "profile", None) or ""), + bool(context.shared_multi_user_session), + discord_ids, + discord_tools, + tuple(p.value for p in context.connected_platforms), + tuple( + ( + p.value, + str(getattr(hc, "name", "") or ""), + str(getattr(hc, "chat_id", "") or ""), + ) + for p, hc in context.home_channels.items() + ), + bool(redact_pii), + home_display, + ) + return hashlib.sha256(repr(key_tuple).encode("utf-8")).hexdigest() + def _evict_cached_agent(self, session_key: str) -> None: """Remove a cached agent for a session (called on /new, /model, etc). @@ -17662,6 +17845,16 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew ``_agent_cache_lock`` on slow socket teardown — mirrors the cap-enforcer and idle-sweeper paths. """ + # Prompt-stability state rides the agent-cache lifecycle: a fresh + # agent must re-render its session-context bytes (the pin) and re-see + # the current voice-channel state once. + _pin_store = getattr(self, "_session_ephemeral_pin", None) + if isinstance(_pin_store, dict): + _pin_store.pop(session_key, None) + _vc_store = getattr(self, "_session_vc_last", None) + if isinstance(_vc_store, dict): + _vc_store.pop(session_key, None) + _lock = getattr(self, "_agent_cache_lock", None) evicted = None if _lock: @@ -19986,6 +20179,14 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew agent.reasoning_config = reasoning_config agent.service_tier = self._service_tier agent.request_overrides = turn_route.get("request_overrides") or {} + # Must-deliver notes for THIS turn ride the current user message + # (api_content sidecar), never the system prompt: staged by + # _handle_message_with_agent (auto-reset note, first-contact + # intro, voice-channel change). Assigned unconditionally so a + # reused cached agent never replays a stale note. + agent._gateway_turn_context_notes = "\n\n".join( + self._consume_pending_turn_sidecar_notes(session_key) + ) _bg_review_release = threading.Event() _bg_review_pending: list[str] = [] diff --git a/gateway/session.py b/gateway/session.py index bed8f31b2bfc..0168d940ffd4 100644 --- a/gateway/session.py +++ b/gateway/session.py @@ -564,6 +564,16 @@ def build_session_context_prompt( "Do not promise to perform these actions. If the user asks, explain " "that you can only read messages sent directly to you and respond." ) + # Static (never per-turn): live voice-channel state used to be + # appended here and changed bytes every turn the bot sat in a voice + # channel, busting the prompt cache. It now arrives on the current + # user message as a `[Voice channel now: ...]` note, injected only + # when it actually changed. + lines.append("") + lines.append( + "Voice-channel state, when relevant, appears in the current " + "message as a `[Voice channel now: ...]` note." + ) elif context.source.platform == Platform.BLUEBUBBLES: lines.append("") lines.append( diff --git a/tests/agent/test_gateway_turn_sidecar.py b/tests/agent/test_gateway_turn_sidecar.py new file mode 100644 index 000000000000..c9233959319a --- /dev/null +++ b/tests/agent/test_gateway_turn_sidecar.py @@ -0,0 +1,207 @@ +"""Gateway must-deliver notes on the current user message. + +The gateway relocates per-turn volatile facts OUT of the ephemeral system +prompt — auto-reset notes, the first-contact intro, voice-channel changes — +and stages them on ``agent._gateway_turn_context_notes``. +``build_turn_context`` consumes them once and delivers them through the same +api_content sidecar channel as plugin context (string content), or as an +appended text part on multimodal (list) content, where the string sidecar +cannot apply and the fact would otherwise silently drop. +""" + +from __future__ import annotations + +import types +from unittest.mock import patch + +import pytest + +from agent.turn_context import ( + append_notes_to_multimodal_content, + build_turn_context, + compose_user_api_content, + consume_gateway_turn_context_notes, +) + + +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_api_content_sidecar.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 = "discord" + 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 + + 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): + pass + + +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: str(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 + + +RESET_NOTE = ( + "[System note: The user's previous session expired due to inactivity. " + "This is a fresh conversation with no prior context.]" +) +VC_NOTE = "[Voice channel now: dev-vc (2 members)]" + + +class TestConsumeIsOneShot: + def test_consume_clears_the_attribute(self): + agent = _FakeAgent() + agent._gateway_turn_context_notes = RESET_NOTE + assert consume_gateway_turn_context_notes(agent) == RESET_NOTE + assert consume_gateway_turn_context_notes(agent) == "" + + def test_absent_attribute_is_empty(self): + assert consume_gateway_turn_context_notes(_FakeAgent()) == "" + + def test_non_string_value_is_empty(self): + agent = _FakeAgent() + agent._gateway_turn_context_notes = ["not-a-string"] + assert consume_gateway_turn_context_notes(agent) == "" + + +class TestStringContentSidecarDelivery: + def test_notes_ride_the_api_content_sidecar(self): + """String user message: the note lands in the API copy only — the + stored content stays clean and the sidecar persists the exact sent + bytes (replay keeps them byte-stable in history).""" + agent = _FakeAgent() + agent._gateway_turn_context_notes = RESET_NOTE + with patch("hermes_cli.plugins.invoke_hook", return_value=[]): + ctx = _build(agent) + msg = ctx.messages[ctx.current_turn_user_idx] + assert msg["content"] == "hello" + assert msg["api_content"] == "hello\n\n" + RESET_NOTE + # The composed bytes match what conversation_loop would send. + assert msg["api_content"] == compose_user_api_content( + "hello", ctx.ext_prefetch_cache, ctx.plugin_user_context + ) + # Consumed: a later turn on the same cached agent replays nothing. + assert agent._gateway_turn_context_notes == "" + + def test_notes_append_after_plugin_context(self): + agent = _FakeAgent() + agent._gateway_turn_context_notes = VC_NOTE + 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["api_content"] == "hello\n\nPLUGIN-CTX\n\n" + VC_NOTE + + def test_no_notes_means_no_stamp(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] + + +class TestMultimodalFallback: + def test_notes_appended_as_text_part_on_list_content(self): + """Multimodal turns can't take the string sidecar + (compose_user_api_content returns None for lists) — the must-deliver + fact is appended as a durable text part instead of dropping.""" + agent = _FakeAgent() + agent._gateway_turn_context_notes = RESET_NOTE + content = [ + {"type": "text", "text": "look at this"}, + {"type": "image_url", "image_url": {"url": "https://x/img.png"}}, + ] + with patch("hermes_cli.plugins.invoke_hook", return_value=[]): + ctx = _build(agent, user_message=content) + msg = ctx.messages[ctx.current_turn_user_idx] + assert msg["content"][-1] == {"type": "text", "text": RESET_NOTE} + # No string sidecar for list content. + assert "api_content" not in msg + + def test_helper_appends_only_to_lists(self): + content = [{"type": "text", "text": "hi"}] + assert append_notes_to_multimodal_content(content, "NOTE") is True + assert content[-1] == {"type": "text", "text": "NOTE"} + assert append_notes_to_multimodal_content("string", "NOTE") is False + assert append_notes_to_multimodal_content(content, "") is False diff --git a/tests/gateway/test_prompt_tail_freeze.py b/tests/gateway/test_prompt_tail_freeze.py new file mode 100644 index 000000000000..285c1ccd6bcd --- /dev/null +++ b/tests/gateway/test_prompt_tail_freeze.py @@ -0,0 +1,410 @@ +"""Byte-stable gateway system prompts (the ephemeral session-context pin). + +The composed system prompt used to change bytes nearly every gateway turn: +the "## Current Session Context" block was re-rendered from live platform +state per message (thread renames, voice-channel member/speaking state, +one-shot onboarding and auto-reset notes). Every byte change re-keys the +provider prompt cache AND changes the gateway agent-cache signature, forcing +a full AIAgent rebuild per message. + +The fix pins the rendered session-context bytes per session keyed by a hash +of the exact renderer inputs (``_ephemeral_change_key``) and relocates +must-deliver per-turn facts onto the current user message (the api_content +sidecar), so a key hit reuses the pinned bytes verbatim. + +The maintained invariant — every rendered input appears in the change key — +is guarded by the parity test below. +""" + +from __future__ import annotations + +import hashlib +from types import SimpleNamespace +from unittest.mock import MagicMock + +import pytest + +from gateway.config import GatewayConfig, HomeChannel, Platform, PlatformConfig +from gateway.session import ( + SessionContext, + SessionSource, + build_session_context_prompt, +) + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _make_runner(**attrs): + from gateway.run import GatewayRunner + + runner = object.__new__(GatewayRunner) + runner._session_ephemeral_pin = {} + runner._session_vc_last = {} + runner._pending_turn_sidecar_notes = {} + runner._session_model_overrides = {} + runner._session_reasoning_overrides = {} + runner.adapters = {} + runner.session_store = MagicMock() + for key, value in attrs.items(): + setattr(runner, key, value) + return runner + + +def _make_context( + *, + platform: Platform = Platform.DISCORD, + chat_id: str = "111222333", + chat_name: str = "general", + chat_type: str = "channel", + thread_id: str | None = "444555666", + parent_chat_id: str | None = "111222333", + chat_topic: str | None = "ops chatter", + user_name: str | None = "pix", + user_id: str | None = "9001", + guild_id: str | None = "777888999", + message_id: str | None = "1357", + shared_multi_user: bool = False, + connected: list[Platform] | None = None, + home_channels: dict | None = None, +) -> SessionContext: + source = SessionSource( + platform=platform, + chat_id=chat_id, + chat_name=chat_name, + chat_type=chat_type, + user_id=user_id, + user_name=user_name, + thread_id=thread_id, + chat_topic=chat_topic, + parent_chat_id=parent_chat_id, + scope_id=guild_id, + message_id=message_id, + ) + connected = connected if connected is not None else [Platform.DISCORD, Platform.TELEGRAM] + if home_channels is None: + home_channels = { + Platform.DISCORD: HomeChannel( + platform=Platform.DISCORD, chat_id="111222333", name="general" + ), + } + return SessionContext( + source=source, + connected_platforms=connected, + home_channels=home_channels, + shared_multi_user_session=shared_multi_user, + ) + + +@pytest.fixture(autouse=True) +def _stable_discord_tools(monkeypatch): + """Pin the config/env-dependent renderer gate so key<->render parity is + evaluated on the same footing in every environment.""" + monkeypatch.setattr("gateway.session._discord_tools_loaded", lambda: True) + + +def _key(runner, context, redact_pii=False): + return runner._ephemeral_change_key(context, redact_pii) # noqa: SLF001 + + +def _render(context, redact_pii=False): + return build_session_context_prompt(context, redact_pii=redact_pii) + + +# --------------------------------------------------------------------------- +# 1. Parity: key <-> render (the maintained invariant) +# --------------------------------------------------------------------------- + +class TestEphemeralChangeKeyParity: + # Single-field mutations spanning every rendered input. For each: + # if the rendered bytes change, the key MUST change (staleness guard). + _MUTATIONS = [ + ("chat_name", dict(chat_name="renamed-thread")), + ("chat_topic", dict(chat_topic="new topic")), + ("chat_topic_cleared", dict(chat_topic=None)), + ("thread_id", dict(thread_id="000111222")), + ("thread_cleared", dict(thread_id=None, parent_chat_id=None)), + ("chat_type", dict(chat_type="group")), + ("user_name", dict(user_name="somebody-else")), + ("user_name_cleared", dict(user_name=None)), + ("user_id", dict(user_name=None, user_id="1234")), + ("shared_multi_user", dict(shared_multi_user=True)), + ("guild_id", dict(guild_id="123123123")), + ("parent_chat_id", dict(parent_chat_id="999000111")), + ("chat_id", dict(chat_id="999999999", parent_chat_id="999999999")), + ("platform", dict(platform=Platform.TELEGRAM)), + ("connected_platforms", dict(connected=[Platform.DISCORD])), + ( + "home_channel_renamed", + dict( + home_channels={ + Platform.DISCORD: HomeChannel( + platform=Platform.DISCORD, chat_id="111222333", name="ops-home" + ) + } + ), + ), + ( + "home_channel_added", + dict( + home_channels={ + Platform.DISCORD: HomeChannel( + platform=Platform.DISCORD, chat_id="111222333", name="general" + ), + Platform.TELEGRAM: HomeChannel( + platform=Platform.TELEGRAM, chat_id="tg1", name="tg-home" + ), + } + ), + ), + ("message_id_cleared", dict(message_id=None)), + ] + + @pytest.mark.parametrize("name,mutation", _MUTATIONS) + def test_render_change_implies_key_change(self, name, mutation): + runner = _make_runner() + base = _make_context() + mutated = _make_context(**mutation) + + render_changed = _render(base) != _render(mutated) + key_changed = _key(runner, base) != _key(runner, mutated) + + if render_changed: + assert key_changed, ( + f"mutation {name!r} changed the rendered bytes but not the " + "change key — the pin would serve STALE context" + ) + + def test_redact_pii_flip_changes_key(self): + # PII redaction only rewrites bytes on pii-safe platforms; the key + # must react wherever the render does. + runner = _make_runner() + ctx = _make_context(platform=Platform.TELEGRAM, thread_id=None, parent_chat_id=None) + assert _render(ctx, False) != _render(ctx, True) + assert _key(runner, ctx, False) != _key(runner, ctx, True) + + def test_discord_tools_gate_flip_changes_key(self, monkeypatch): + runner = _make_runner() + ctx = _make_context() + render_on, key_on = _render(ctx), _key(runner, ctx) + monkeypatch.setattr("gateway.session._discord_tools_loaded", lambda: False) + assert _render(ctx) != render_on + assert _key(runner, ctx) != key_on + + def test_message_id_value_change_is_not_a_bust(self): + """Only message-id PRESENCE renders (the id itself rides the user + message) — a new id every turn must not re-render.""" + runner = _make_runner() + a = _make_context(message_id="1357") + b = _make_context(message_id="2468") + assert _render(a) == _render(b) + assert _key(runner, a) == _key(runner, b) + + def test_key_is_deterministic(self): + runner = _make_runner() + ctx = _make_context() + assert _key(runner, ctx) == _key(runner, ctx) + + +# --------------------------------------------------------------------------- +# 2. The pin: reuse verbatim on hit, exactly one legit bust on change +# --------------------------------------------------------------------------- + +class TestSessionContextPin: + def test_pin_hit_returns_identical_object(self): + runner = _make_runner() + ctx = _make_context() + first = runner._pinned_session_context_prompt(ctx, False, "sk") # noqa: SLF001 + second = runner._pinned_session_context_prompt(_make_context(), False, "sk") # noqa: SLF001 + # Identity, not just equality: the pinned bytes are reused verbatim, + # immunizing against renderer nondeterminism. + assert second is first + + def test_auto_thread_rename_busts_exactly_once(self): + """Turn 1: placeholder title. Turn 2: gateway auto-rename lands (one + legit bust — Source line AND origin delivery line move together). + Turn 3+: byte-stable.""" + runner = _make_runner() + t1 = runner._pinned_session_context_prompt( # noqa: SLF001 + _make_context(chat_name="new-chat-1357"), False, "sk" + ) + t2 = runner._pinned_session_context_prompt( # noqa: SLF001 + _make_context(chat_name="Fixing the flaky deploy"), False, "sk" + ) + t3 = runner._pinned_session_context_prompt( # noqa: SLF001 + _make_context(chat_name="Fixing the flaky deploy"), False, "sk" + ) + assert t1 != t2 + assert t3 is t2 + assert "Fixing the flaky deploy" in t2 + + def test_eviction_drops_pin_and_vc_state(self): + runner = _make_runner( + _agent_cache={}, _running_agents={}, + ) + runner._session_ephemeral_pin["sk"] = ("k", "text") + runner._session_vc_last["sk"] = "vc" + runner._evict_cached_agent("sk") # noqa: SLF001 + assert "sk" not in runner._session_ephemeral_pin + assert "sk" not in runner._session_vc_last + + def test_no_session_key_never_pins(self): + runner = _make_runner() + ctx = _make_context() + out = runner._pinned_session_context_prompt(ctx, False, None) # noqa: SLF001 + assert out == _render(ctx) + assert runner._session_ephemeral_pin == {} + + +# --------------------------------------------------------------------------- +# 3. Two-turn byte test: composed system prompt sha256 + codex cache key +# --------------------------------------------------------------------------- + +def _compose(context_prompt: str) -> str: + """Compose base + ephemeral exactly like conversation_loop does.""" + base = "BASE IDENTITY PROMPT\n" + "x" * 8000 + return (base + "\n\n" + context_prompt).strip() + + +class TestComposedPromptByteStability: + def test_turn2_equals_turn3_sha256(self): + runner = _make_runner() + name = "Fixing the flaky deploy" + t2 = _compose( + runner._pinned_session_context_prompt( # noqa: SLF001 + _make_context(chat_name=name), False, "sk" + ) + ) + t3 = _compose( + runner._pinned_session_context_prompt( # noqa: SLF001 + _make_context(chat_name=name), False, "sk" + ) + ) + assert hashlib.sha256(t2.encode()).hexdigest() == hashlib.sha256(t3.encode()).hexdigest() + + def test_codex_cache_key_constant_across_turns(self): + """The codex transport content-addresses its prompt cache key from + (instructions + tools); pinned ephemeral bytes keep it warm.""" + from agent.transports.codex import _content_cache_key + + runner = _make_runner() + tools = [{"type": "function", "name": "read_file"}] + keys = [ + _content_cache_key( + _compose( + runner._pinned_session_context_prompt( # noqa: SLF001 + _make_context(), False, "sk" + ) + ), + tools, + ) + for _ in range(3) + ] + assert keys[0] is not None + assert len(set(keys)) == 1 + + +# --------------------------------------------------------------------------- +# 4. Voice-channel sidecar note: only-when-changed +# --------------------------------------------------------------------------- + +def _source(): + return SessionSource( + platform=Platform.DISCORD, chat_id="c1", chat_type="channel", user_id="u1" + ) + + +class _VcAdapter: + def __init__(self, value): + self.value = value + + def get_voice_channel_context(self, guild_id): + return self.value + + +def _vc_runner(vc_value): + adapter = _VcAdapter(vc_value) + runner = _make_runner(adapters={Platform.DISCORD: adapter}) + return runner, adapter + + +def _vc_event(): + return SimpleNamespace(raw_message=SimpleNamespace(guild_id="777")) + + +class TestVoiceChannelSidecarNote: + def test_first_sighting_injects(self): + runner, _ = _vc_runner("**Voice:** dev-vc (2 members)") + note = runner._voice_channel_sidecar_note(_vc_event(), _source(), "sk") # noqa: SLF001 + assert note == "[Voice channel now: **Voice:** dev-vc (2 members)]" + + def test_unchanged_state_injects_nothing(self): + runner, _ = _vc_runner("**Voice:** dev-vc (2 members)") + assert runner._voice_channel_sidecar_note(_vc_event(), _source(), "sk") # noqa: SLF001 + assert runner._voice_channel_sidecar_note(_vc_event(), _source(), "sk") is None # noqa: SLF001 + + def test_member_change_injects_again(self): + runner, adapter = _vc_runner("**Voice:** dev-vc (2 members)") + runner._voice_channel_sidecar_note(_vc_event(), _source(), "sk") # noqa: SLF001 + adapter.value = "**Voice:** dev-vc (3 members)" + note = runner._voice_channel_sidecar_note(_vc_event(), _source(), "sk") # noqa: SLF001 + assert note == "[Voice channel now: **Voice:** dev-vc (3 members)]" + + def test_leaving_channel_injects_disconnect_note(self): + runner, adapter = _vc_runner("**Voice:** dev-vc (2 members)") + runner._voice_channel_sidecar_note(_vc_event(), _source(), "sk") # noqa: SLF001 + adapter.value = "" + note = runner._voice_channel_sidecar_note(_vc_event(), _source(), "sk") # noqa: SLF001 + assert note == "[Voice channel now: not connected to a voice channel]" + + def test_never_in_channel_injects_nothing(self): + runner, _ = _vc_runner("") + assert runner._voice_channel_sidecar_note(_vc_event(), _source(), "sk") is None # noqa: SLF001 + + def test_non_discord_platform_is_noop(self): + runner, _ = _vc_runner("**Voice:** dev-vc") + src = SessionSource(platform=Platform.TELEGRAM, chat_id="c", user_id="u") + assert runner._voice_channel_sidecar_note(_vc_event(), src, "sk") is None # noqa: SLF001 + + +# --------------------------------------------------------------------------- +# 5. Sidecar note staging: one-shot per turn +# --------------------------------------------------------------------------- + +class TestSidecarNoteStaging: + def test_set_then_consume_once(self): + runner = _make_runner() + runner._set_pending_turn_sidecar_notes("sk", ["[System note: reset]"]) # noqa: SLF001 + assert runner._consume_pending_turn_sidecar_notes("sk") == ["[System note: reset]"] # noqa: SLF001 + assert runner._consume_pending_turn_sidecar_notes("sk") == [] # noqa: SLF001 + + def test_empty_inputs_are_noops(self): + runner = _make_runner() + runner._set_pending_turn_sidecar_notes("", ["x"]) # noqa: SLF001 + runner._set_pending_turn_sidecar_notes("sk", []) # noqa: SLF001 + assert runner._consume_pending_turn_sidecar_notes("sk") == [] # noqa: SLF001 + assert runner._consume_pending_turn_sidecar_notes("") == [] # noqa: SLF001 + + +# --------------------------------------------------------------------------- +# 6. Connected platforms: stable order +# --------------------------------------------------------------------------- + +class TestConnectedPlatformsOrder: + def test_sorted_regardless_of_insertion_order(self): + cfg_a = GatewayConfig( + platforms={ + Platform.TELEGRAM: PlatformConfig(enabled=True, token="t"), + Platform.DISCORD: PlatformConfig(enabled=True, token="d"), + } + ) + cfg_b = GatewayConfig( + platforms={ + Platform.DISCORD: PlatformConfig(enabled=True, token="d"), + Platform.TELEGRAM: PlatformConfig(enabled=True, token="t"), + } + ) + assert cfg_a.get_connected_platforms() == cfg_b.get_connected_platforms() + values = [p.value for p in cfg_a.get_connected_platforms()] + assert values == sorted(values)