diff --git a/gateway/run_agent_cache.py b/gateway/run_agent_cache.py index 22db90dbb0a8..e186721223cc 100644 --- a/gateway/run_agent_cache.py +++ b/gateway/run_agent_cache.py @@ -618,12 +618,22 @@ def _voice_channel_sidecar_note(self, event, source: SessionSource, session_key: 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 an existing pin verbatim; with no pin yet it renders and pins as usual.""" _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 and _eph_pin is not None: + return _eph_pin[1] + _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 +641,31 @@ def _pinned_session_context_prompt(self, context, redact_pii: bool, session_key: self._session_state(session_key).conversation.ephemeral_pin = (_eph_key, text) return text + def _pinned_channel_inputs( + self, session_key: Optional[str], channel_prompt: Optional[str], source: SessionSource, *, internal: bool, + ): + """``(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.""" + if not session_key: + return channel_prompt, source + if not internal: + self._session_state(session_key).conversation.channel_pin = (channel_prompt, source.parent_chat_id) + 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 source.parent_chat_id: + from gateway.session_identity import replace_source + source = replace_source(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 95aa81223f7b..719067e41c1c 100644 --- a/gateway/run_turn.py +++ b/gateway/run_turn.py @@ -2085,7 +2085,9 @@ async def _hmwa_prepare_turn(self, event, source, session_entry, session_key, _q # 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=event.internal, + ) # 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). @@ -2203,12 +2205,16 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g # 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.channel_prompt, source, internal=event.internal, + ) 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, @@ -3870,7 +3876,9 @@ async def _run_agent_queued_followup( 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.channel_prompt, next_source, internal=pending_event.internal, + ) 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 9e36c35e4b5c..918327bf30bf 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 000000000000..fc4f966955a9 --- /dev/null +++ b/tests/gateway/test_internal_event_pin_wiring.py @@ -0,0 +1,209 @@ +"""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. A helper-level unit test would stay green if +the call site in ``_handle_message_with_agent`` stopped forwarding +``event.internal``, so 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.run_turn_runner import TurnRunner +from gateway.session import SessionEntry, SessionSource +from gateway.turn_context import TurnContext + +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) + + # Internal-first on a fresh session (kanban/API-server wake, startup resume): + # with no pin to reuse it must still render AND pin, or every internal-only + # turn re-renders and loses verbatim reuse. + await _drive(runner, ((True, _wake_source()),)) + assert runner._peek_session_state(KEY).conversation.ephemeral_pin is not None + + await _drive(runner, ((False, _human_source()), (True, _wake_source()), (False, _human_source()))) + + seen = [kw["context_prompt"] for kw in calls[1:]] + 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" + + +# --------------------------------------------------------------------------- +# The other ephemeral prompt components must not toggle across the sequence either. +# --------------------------------------------------------------------------- + +THREAD_ID = "1552000000000000001" +PARENT_ID = "1513247605675790346" + + +def _effective_ephemeral(runner, kw) -> str: + """The real TurnRunner combiner applied to what reached ``_run_agent``.""" + ctx = TurnContext(source=kw["source"], context_prompt=kw["context_prompt"], channel_prompt=kw.get("channel_prompt")) + return TurnRunner(runner, ctx)._combined_ephemeral_prompt() + + +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" diff --git a/tests/gateway/test_queued_final_ledger.py b/tests/gateway/test_queued_final_ledger.py index 764c733a49ad..9764ce25abb0 100644 --- a/tests/gateway/test_queued_final_ledger.py +++ b/tests/gateway/test_queued_final_ledger.py @@ -60,8 +60,10 @@ def _rows(): def _source(*, chat_id=CHAT, thread_id=None, chat_type="dm"): + # parent_chat_id is a real SessionSource field; the queued-followup path reads it to pin the + # channel inputs, so the stub carries it like every real source does. return SimpleNamespace(platform=Platform.TELEGRAM, chat_id=chat_id, thread_id=thread_id, - chat_type=chat_type) + chat_type=chat_type, parent_chat_id=None) def _telegram_adapter(send_result=None):