mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-07-21 16:18:55 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 8

perf(gateway): byte-stable session context prompts

Browse source 
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.
This commit is contained in:
Soju06 2026-07-14 08:53:42 +00:00 • committed by kshitij
parent 94f8166dc8
commit c0c76a4715
6 changed files with 920 additions and 22 deletions
Download patch file Download diff file Expand all files Collapse all files

63
agent/turn_context.py
View file

@ -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()

11
gateway/config.py
View file

@ -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."""

241
gateway/run.py
View file

@ -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] = []

10
gateway/session.py
View file

@ -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(

207
tests/agent/test_gateway_turn_sidecar.py Normal file
View file

@ -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

410
tests/gateway/test_prompt_tail_freeze.py Normal file
View file

@ -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)