fix(gateway): internal events keep the session-context and channel prompt pins

gateway/run_agent_cache.py

@@ -618,19 +618,54 @@ 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)
         if 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

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.

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:

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"

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