From 9c18b383de07b4b65474b7ffbabcfd08f22d9e07 Mon Sep 17 00:00:00 2001 From: Kyzcreig <9063726+Kyzcreig@users.noreply.github.com> Date: Thu, 24 Sep 2026 17:03:17 -0700 Subject: [PATCH] fix(gateway): internal events must not re-key the session-context / channel prompt pin Internal events (kanban wakes, delegation completions, watch notifications) carry a source rebuilt from the persisted origin: no chat_name/user_name/message_id/parent_chat_id, and channel_prompt=None. _pinned_session_context_prompt re-rendered from that source, so every internal turn re-keyed the pin and the next human turn re-keyed it back (A->B->A). Each flip rewrote already-sent system bytes and collapsed the prompt cache to the static prefix. The same toggle happened through channel_prompt and parent-keyed channel_overrides in the ephemeral system prompt. - _pinned_session_context_prompt(internal=True) reuses the existing pin verbatim and never re-pins. - Human turns record (channel_prompt, parent_chat_id) in ConversationState.channel_pin; internal turns reuse them (main run and the queued follow-up path). tests/gateway/test_internal_event_pin_wiring.py drives the real _handle_message_with_agent human->internal->human with _run_agent stubbed. Both tests fail on main and pass here; dropping internal= at the call site makes both fail again. (cherry picked from commit 3b38527ae7f0f498ca265fe9f64fcbc1c8c7765c) --- gateway/run_agent_cache.py | 42 +++- gateway/run_turn.py | 14 +- gateway/session_state.py | 2 + .../gateway/test_internal_event_pin_wiring.py | 213 ++++++++++++++++++ 4 files changed, 264 insertions(+), 7 deletions(-) create mode 100644 tests/gateway/test_internal_event_pin_wiring.py diff --git a/gateway/run_agent_cache.py b/gateway/run_agent_cache.py index 22db90dbb0..8462ff251c 100644 --- a/gateway/run_agent_cache.py +++ b/gateway/run_agent_cache.py @@ -4,6 +4,7 @@ for GatewayRunner (MRO mixin). ``gateway.run`` internals are imported lazily ins from __future__ import annotations +import dataclasses import importlib import logging import threading @@ -618,12 +619,22 @@ class GatewayAgentCacheMixin: return None return f"[Voice channel now: {vc_now or 'not connected to a voice channel'}]" - def _pinned_session_context_prompt(self, context, redact_pii: bool, session_key: Optional[str]) -> str: + def _pinned_session_context_prompt( + self, context, redact_pii: bool, session_key: Optional[str], *, internal: bool = False, + ) -> str: """Session-context prompt pinned per session: key hit → pinned bytes reused VERBATIM (immune - to renderer nondeterminism); key miss → re-render and re-pin (rename, topic edit, /sethome).""" - _eph_key = self._ephemeral_change_key(context, redact_pii) + to renderer nondeterminism); key miss → re-render and re-pin (rename, topic edit, /sethome). + + ``internal`` events (kanban wakes, delegation completions, watch notifications) carry a + source rebuilt from the persisted origin, without chat_name/user_name/message_id. Rendering + from it re-keyed the pin, and the next human turn re-keyed it back (A→B→A), rewriting + already-sent system bytes each time. An internal event is never a real metadata change, so + it reuses the existing pin verbatim and never re-pins.""" _pin_state = self._peek_session_state(session_key) if session_key else None _eph_pin = _pin_state.conversation.ephemeral_pin if _pin_state else None + if internal: + return _eph_pin[1] if _eph_pin is not None else build_session_context_prompt(context, redact_pii=redact_pii) + _eph_key = self._ephemeral_change_key(context, redact_pii) 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) @@ -631,6 +642,31 @@ class GatewayAgentCacheMixin: self._session_state(session_key).conversation.ephemeral_pin = (_eph_key, text) return text + def _pinned_channel_inputs(self, session_key, event, source): + """``(channel_prompt, source)`` for this turn's agent run. + + The ephemeral system prompt also appends ``channel_prompt`` and the ``channel_overrides`` + prompt (looked up by chat/thread/``parent_chat_id``). Internal events carry + ``channel_prompt=None`` and a source without ``parent_chat_id``, so they dropped both and + toggled the system prompt like the context pin did. Human turns record their inputs; + internal turns reuse them.""" + channel_prompt = getattr(event, "channel_prompt", None) + if not session_key: + return channel_prompt, source + if not getattr(event, "internal", False): + self._session_state(session_key).conversation.channel_pin = ( + channel_prompt, getattr(source, "parent_chat_id", None), + ) + return channel_prompt, source + state = self._peek_session_state(session_key) + pin = state.conversation.channel_pin if state else None + if pin is None: + return channel_prompt, source + pinned_prompt, pinned_parent = pin + if pinned_parent and not getattr(source, "parent_chat_id", None): + source = dataclasses.replace(source, parent_chat_id=pinned_parent) + return pinned_prompt, source + @staticmethod def _ephemeral_change_key(context, redact_pii: bool) -> str: """Hash the exact inputs ``build_session_context_prompt`` renders. Invariant diff --git a/gateway/run_turn.py b/gateway/run_turn.py index 80b01432b7..405ffe4653 100644 --- a/gateway/run_turn.py +++ b/gateway/run_turn.py @@ -2087,7 +2087,9 @@ class GatewayTurnMixin: # The context prompt render is pinned per session, keyed by a hash of the renderer inputs, so # the system prompt cannot drift turn-over-turn; a miss (thread rename, /sethome) re-renders. - context_prompt = self._pinned_session_context_prompt(context, _redact_pii, session_key) + context_prompt = self._pinned_session_context_prompt( + context, _redact_pii, session_key, internal=bool(getattr(event, "internal", False)), + ) # Per-turn notes ride the user message via the api_content sidecar, NOT context_prompt # (appending to the ephemeral system prompt forced a full agent rebuild). @@ -2205,12 +2207,14 @@ class GatewayTurnMixin: # Admission/typing is not execution. All routing, authorization and # turn preparation gates have passed when the agent runner is entered. event._heartbeat_execution_started = True + # Internal events reuse the last human turn's channel inputs (see _pinned_channel_inputs). + _turn_channel_prompt, _turn_source = self._pinned_channel_inputs(session_key, event, source) agent_result = await self._run_agent( - message=message_text, context_prompt=prepared.context_prompt, history=history, source=source, + message=message_text, context_prompt=prepared.context_prompt, history=history, source=_turn_source, session_id=_run_start_session_id, session_key=session_key, run_generation=run_generation, event_message_id=self._reply_anchor_for_event(event), inbound_message_id=str(event.message_id) if event.message_id else None, - channel_prompt=event.channel_prompt, moa_config=getattr(event, "_moa_config", None), + channel_prompt=_turn_channel_prompt, moa_config=getattr(event, "_moa_config", None), persist_user_message=prepared.persist_user_message, persist_user_timestamp=prepared.persist_user_timestamp, persist_user_display_kind=prepared.persist_user_display_kind, @@ -3872,7 +3876,9 @@ class GatewayTurnMixin: next_persist_message = strip_discord_triggering_note(pending_event, next_message) next_message_id = self._reply_anchor_for_event(pending_event) next_inbound_id = str(pending_event.message_id) if getattr(pending_event, "message_id", None) else None - next_channel_prompt = getattr(pending_event, "channel_prompt", None) + next_channel_prompt, next_source = self._pinned_channel_inputs( + next_session_key, pending_event, next_source, + ) next_message_type = getattr(pending_event, "message_type", None) # Clear the prior turn's streaming-TTS completion marker so the recursive turn isn't suppressed. diff --git a/gateway/session_state.py b/gateway/session_state.py index 9e36c35e4b..918327bf30 100644 --- a/gateway/session_state.py +++ b/gateway/session_state.py @@ -50,6 +50,8 @@ class ConversationState: queued_events: List[Any] = field(default_factory=list) # /queue overflow FIFO (head in adapter) sidecar_notes: List[str] = field(default_factory=list) # one-shot must-deliver notes ephemeral_pin: Optional[Tuple[Any, ...]] = None # pinned session-context (change_key, text) + # (channel_prompt, parent_chat_id) of the last non-internal turn; internal events reuse it + channel_pin: Optional[Tuple[Optional[str], Optional[str]]] = None vc_last: Optional[str] = None # last voice-channel context delivered def clear(self) -> None: diff --git a/tests/gateway/test_internal_event_pin_wiring.py b/tests/gateway/test_internal_event_pin_wiring.py new file mode 100644 index 0000000000..e28c695032 --- /dev/null +++ b/tests/gateway/test_internal_event_pin_wiring.py @@ -0,0 +1,213 @@ +"""Call-site wiring guard for the internal-event session-context pin. + +``_pinned_session_context_prompt(..., internal=True)`` reuses the existing pin +verbatim so an internal event (kanban wake, delegation completion, watch +notification) cannot re-key it. The unit tests in test_prompt_tail_freeze.py +exercise that helper directly, so they stay green if the call site in +``_handle_message_with_agent`` stops forwarding ``event.internal``. This test drives the REAL handler through +human -> internal -> human on one session and asserts the context prompt that +reaches ``_run_agent`` is byte-identical on all three turns. +""" + +from __future__ import annotations + +from datetime import datetime +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from gateway.config import ChannelOverride, GatewayConfig, Platform, PlatformConfig +from gateway.platforms.base import MessageEvent +from gateway.session import SessionEntry, SessionSource + +KEY = "agent:main:discord:group:1513247605675790346:117431298246705156" +_ORIGIN = dict( + platform=Platform.DISCORD, + chat_id="1513247605675790346", + chat_type="group", + user_id="117431298246705156", + scope_id="1480524732964278294", +) + + +def _human_source() -> SessionSource: + return SessionSource( + **_ORIGIN, + chat_name="Guild / #general", + user_name="Ace", + message_id="1552671843494666330", + ) + + +def _wake_source() -> SessionSource: + # kanban_watchers._push_wake shape: rebuilt from the persisted origin, so + # no chat_name / user_name / message_id. + return SessionSource(**_ORIGIN) + + +def _make_runner(monkeypatch, config: GatewayConfig | None = None): + import agent.model_metadata as mm + import gateway.run as gr + import gateway.session as gs + + monkeypatch.setattr(gs, "_discord_tools_loaded", lambda: True) + monkeypatch.setattr(gr, "_resolve_runtime_agent_kwargs", lambda: {"api_key": "fake"}) + monkeypatch.setattr(mm, "get_model_context_length", lambda *a, **k: 100_000) + + r = gr.GatewayRunner(config or GatewayConfig()) + r.adapters = {} + r._running_agents = {} + r._running_agents_ts = {} + r._pending_messages = {} + r._pending_approvals = {} + r._is_user_authorized = lambda s: True + r._set_session_env = lambda c: None + r._handle_active_session_busy_message = AsyncMock(return_value=False) + r._session_db = MagicMock() + r._recover_telegram_topic_thread_id = lambda s: None + r._cache_session_source = lambda k, s: None + r._is_session_run_current = lambda k, g: True + r._begin_session_run_generation = lambda k: 1 + r._reply_anchor_for_event = lambda e: None + r._get_guild_id = lambda e: None + r._should_send_voice_reply = lambda *a, **k: False + r.hooks = MagicMock() + r.hooks.emit = AsyncMock() + # The turn lease is released by the real turn tail, which _run_agent's + # stub skips; disable leasing so three sequential turns can run. + r._turn_leases = None + + store = MagicMock() + store.get_or_create_session.return_value = SessionEntry( + session_key=KEY, + session_id="sess-wiring", + created_at=datetime(2026, 1, 1), + updated_at=datetime(2026, 1, 2), + platform=Platform.DISCORD, + chat_type="group", + ) + store.load_transcript.return_value = [] + store.has_platform_message_id.return_value = False + r.session_store = store + return r + + +def _capture(runner, sink: list): + async def fake_run_agent(**kw): + sink.append(kw) + return { + "final_response": "ok", + "messages": [], + "tools": [], + "history_offset": 0, + "last_prompt_tokens": 0, + } + + runner._run_agent = fake_run_agent + + +async def _drive(runner, turns, *, channel_prompt=None): + for internal, src in turns: + event = MessageEvent( + text="[kanban] wake" if internal else "hi", + source=src, + message_id=None if internal else src.message_id, + internal=internal, + # Adapters resolve channel_prompts onto human events; the kanban + # wake is built without one. + channel_prompt=None if internal else channel_prompt, + ) + await runner._handle_message_with_agent(event, src, KEY, 1) + + +@pytest.mark.asyncio +async def test_internal_event_reuses_pin_through_real_handler(monkeypatch): + runner = _make_runner(monkeypatch) + calls: list[dict] = [] + _capture(runner, calls) + + await _drive(runner, ((False, _human_source()), (True, _wake_source()), (False, _human_source()))) + + seen = [kw["context_prompt"] for kw in calls] + assert len(seen) == 3, f"_run_agent reached {len(seen)}/3 turns" + # The human render names the chat; the wake-shaped source cannot. If the + # internal turn re-rendered, its bytes would differ and the next human + # turn would re-key back (A->B->A). + assert "Guild / #general" in seen[0] + assert seen[0] == seen[1] == seen[2], "internal event re-keyed the session-context pin" + + +# --------------------------------------------------------------------------- +# Caveat C2: the other ephemeral components must not toggle either. +# --------------------------------------------------------------------------- + +THREAD_ID = "1552000000000000001" +PARENT_ID = "1513247605675790346" + + +def _effective_ephemeral(runner, kw) -> str: + """Mirror TurnRunner.run_sync's combined_ephemeral from _run_agent's inputs.""" + src = kw["source"] + combined = kw["context_prompt"] or "" + event_cp = (kw.get("channel_prompt") or "").strip() + if event_cp: + combined = (combined + "\n\n" + event_cp).strip() + cfg_cp = runner._get_system_prompt_for_channel( + src.platform, + src.chat_id or "", + thread_id=getattr(src, "thread_id", None), + parent_id=getattr(src, "parent_chat_id", None), + ) + if cfg_cp: + combined = (combined + "\n\n" + cfg_cp).strip() + return combined + + +def _thread_origin() -> dict: + return dict( + platform=Platform.DISCORD, + chat_id=THREAD_ID, + chat_type="thread", + thread_id=THREAD_ID, + user_id="117431298246705156", + scope_id="1480524732964278294", + ) + + +def _human_thread_source() -> SessionSource: + return SessionSource( + **_thread_origin(), + parent_chat_id=PARENT_ID, + chat_name="Guild / #dev / build thread", + user_name="Ace", + message_id="1552671843494666331", + ) + + +def _wake_thread_source() -> SessionSource: + # _push_wake rebuilds chat_id/thread_id from the subscription; it has no + # parent_chat_id, chat_name, user_name or message_id. + return SessionSource(**_thread_origin()) + + +@pytest.mark.asyncio +async def test_internal_event_keeps_channel_prompt_and_parent_override(monkeypatch): + config = GatewayConfig() + config.platforms[Platform.DISCORD] = PlatformConfig( + enabled=True, + channel_overrides={PARENT_ID: ChannelOverride(system_prompt="Parent persona.")}, + ) + runner = _make_runner(monkeypatch, config) + calls: list[dict] = [] + _capture(runner, calls) + + await _drive( + runner, + ((False, _human_thread_source()), (True, _wake_thread_source()), (False, _human_thread_source())), + channel_prompt="Channel hint.", + ) + + assert len(calls) == 3 + eph = [_effective_ephemeral(runner, kw) for kw in calls] + assert "Channel hint." in eph[0] and "Parent persona." in eph[0] + assert eph[0] == eph[1] == eph[2], "internal event toggled the channel ephemeral components"