fix(gateway): preserve prompt pins across synthetic goal/heartbeat/resume turns

gateway/platforms/event.py

@@ -92,6 +92,9 @@ class MessageEvent:
     # knows the message was meant for someone else); None means unknown and keeps the visible
     # fallback, like True.
     reply_expected: Optional[bool] = None
+    # Synthetic continuations can remain ordinary authorized/activity-touching turns while carrying
+    # no fresh system-prompt identity. Appended here to preserve MessageEvent positional arguments.
+    preserve_prompt_pins: bool = False
 
     # Process-local admission receipt, never routing metadata or execution acknowledgement.
     _gateway_accepted: bool = field(default=False, init=False, repr=False, compare=False)

gateway/run_agent_cache.py

@@ -620,8 +620,8 @@ def _voice_channel_sidecar_note(self, event, source: SessionSource, session_key:
         return f"[Voice channel now: {vc_now or 'not connected to a voice channel'}]"
 
     async def _rehydrate_prompt_pins(self, session_key: str, expected_session_id: Optional[str]) -> None:
-        """Adopt the durable pin snapshot for an internal turn when this process holds no pins for
-        *session_key* (a restart). Eviction clears only ``ephemeral_pin`` and keeps ``channel_pin``,
+        """Adopt the durable pin snapshot for a pin-preserving turn (internal wake or synthetic
+        goal/heartbeat continuation) when this process holds no pins for *session_key* (a restart). Eviction clears only ``ephemeral_pin`` and keeps ``channel_pin``,
         so an evicted agent still re-renders instead of reviving the snapshot."""
         state = self._peek_session_state(session_key)
         if state is not None and (
@@ -662,25 +662,34 @@ async def _persist_prompt_pins(self, session_key: Optional[str], expected_sessio
             # Durability protects cache continuity; a store outage must not block the user turn.
             logger.debug("Failed to persist prompt pin for %s", session_key, exc_info=True)
 
+    @staticmethod
+    def _event_preserves_prompt_pins(event: Any) -> bool:
+        """True when *event* must reuse established prompt-pin identity without becoming internal."""
+        return bool(
+            getattr(event, "internal", False)
+            or getattr(event, "preserve_prompt_pins", False)
+        )
+
     def _pinned_session_context_prompt(
-        self, context, redact_pii: bool, session_key: Optional[str], *, internal: bool = False,
+        self, context, redact_pii: bool, session_key: Optional[str], *,
+        internal: bool = False, preserve_pin: Optional[bool] = None,
     ) -> 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).
 
-        ``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 (restored by ``_rehydrate_prompt_pins`` after a restart); with
-        no pin yet it renders and pins as usual. The pin records the ``privacy.redact_pii`` it was
+        Pin-preserving events include internal wakes plus synthetic goal/heartbeat continuations
+        that intentionally carry no fresh channel/session metadata. Re-rendering either shape
+        re-keys the pin and the next human turn re-keys it back (A→B→A), rewriting already-sent
+        system bytes each time. With no established pin yet, render and pin as usual. The pin records the ``privacy.redact_pii`` it was
         rendered under: bytes from another privacy policy are never reused, even by an internal
         event."""
+        if preserve_pin is None:
+            preserve_pin = internal
         _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 _eph_pin is not None and _eph_pin[2] != redact_pii:
             _eph_pin = None
-        if internal and _eph_pin is not None:
+        if preserve_pin 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:
@@ -691,18 +700,21 @@ def _pinned_session_context_prompt(
         return text
 
     def _pinned_channel_inputs(
-        self, session_key: Optional[str], channel_prompt: Optional[str], source: SessionSource, *, internal: bool,
+        self, session_key: Optional[str], channel_prompt: Optional[str], source: SessionSource, *,
+        internal: bool = False, preserve_pin: Optional[bool] = None,
     ):
         """``(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 (``_rehydrate_prompt_pins`` restores both after a restart)."""
+        prompt (looked up by chat/thread/``parent_chat_id``). Pin-preserving events carry no
+        authoritative channel inputs of their own, so they reuse the established human-turn pin
+        without changing authorization semantics. Ordinary turns record their current inputs
+        (``_rehydrate_prompt_pins`` restores both after a restart)."""
+        if preserve_pin is None:
+            preserve_pin = internal
         if not session_key:
             return channel_prompt, source
-        if not internal:
+        if not preserve_pin:
             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)

gateway/run_busy.py

@@ -386,6 +386,8 @@ def _queue_or_replace_pending_event(self, session_key: str, event: MessageEvent)
         existing = pending_slot.get(session_key) if isinstance(pending_slot, dict) else None
         same_security_context = existing is not None and (
             getattr(existing, "internal", False) == getattr(event, "internal", False)
+            and getattr(existing, "preserve_prompt_pins", False)
+            == getattr(event, "preserve_prompt_pins", False)
             and getattr(existing, "allow_gateway_control", True)
             == getattr(event, "allow_gateway_control", True)
             and all(

gateway/run_goals.py

@@ -107,7 +107,10 @@ def _synthetic_prompt_event(source: Any, text: str, *, internal: bool = False) -
         would quote that stale message (Telegram DM topics route anchorless via the topic id).
         """
         source = dataclasses.replace(source, message_id=None) if getattr(source, "message_id", None) else source
-        return MessageEvent(text=text, message_type=MessageType.TEXT, source=source, internal=internal)
+        return MessageEvent(
+            text=text, message_type=MessageType.TEXT, source=source, internal=internal,
+            preserve_prompt_pins=True,
+        )
 
     def _register_heartbeat_watch(self, quick_key: str, source: Any, session_id: str) -> None:
         """Track the canonical route and start the restart-recoverable poller."""

gateway/run_turn.py

@@ -2057,10 +2057,11 @@ 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.
-        if event.internal and session_key:
+        if session_key and self._event_preserves_prompt_pins(event):
             await self._rehydrate_prompt_pins(session_key, session_entry.session_id)
         context_prompt = self._pinned_session_context_prompt(
-            context, _redact_pii, session_key, internal=event.internal,
+            context, _redact_pii, session_key,
+            preserve_pin=self._event_preserves_prompt_pins(event),
         )
 
         # Per-turn notes ride the user message via the api_content sidecar, NOT context_prompt
@@ -2175,9 +2176,10 @@ 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).
+            # Internal wakes and synthetic continuations reuse established prompt-pin inputs.
             _turn_channel_prompt, _turn_source = self._pinned_channel_inputs(
-                session_key, event.channel_prompt, source, internal=event.internal,
+                session_key, event.channel_prompt, source,
+                preserve_pin=self._event_preserves_prompt_pins(event),
             )
             if not event.internal:
                 # Persist the coherent context+channel pair before execution: a crash during the
@@ -3837,7 +3839,8 @@ async def _run_agent_queued_followup(
             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, next_source = self._pinned_channel_inputs(
-                next_session_key, pending_event.channel_prompt, next_source, internal=pending_event.internal,
+                next_session_key, pending_event.channel_prompt, next_source,
+                preserve_pin=self._event_preserves_prompt_pins(pending_event),
             )
             if not pending_event.internal:
                 # A drained human turn re-pins its channel inputs; make them durable like a first turn.

gateway/slash_commands_goals.py

@@ -78,13 +78,16 @@ def _enqueue_goal_turn(
         try:
             adapter, quick_key = self._adapter_and_key_for(event)
             if text and adapter and quick_key:
-                turn = MessageEvent(
-                    text=text,
-                    message_type=MessageType.TEXT,
-                    source=event.source,
-                    message_id=event.message_id if kickoff else None,
-                    channel_prompt=event.channel_prompt if kickoff else None,
-                )
+                if kickoff:
+                    turn = MessageEvent(
+                        text=text,
+                        message_type=MessageType.TEXT,
+                        source=event.source,
+                        message_id=event.message_id,
+                        channel_prompt=event.channel_prompt,
+                    )
+                else:
+                    turn = self._synthetic_prompt_event(event.source, text)
                 self._enqueue_fifo(quick_key, turn, adapter)
         except Exception as exc:
             logger.debug("goal %s failed: %s", label, exc)

tests/gateway/test_goal_continuation_drain.py

@@ -202,3 +202,4 @@ async def test_runner_goal_hook_enqueues_into_the_key_the_adapter_drains(hermes_
         f"drains: pending keys={list(adapter._pending_messages)} "
         f"expected={adapter_key}"
     )
+    assert getattr(adapter._pending_messages[adapter_key], "preserve_prompt_pins", False) is True

tests/gateway/test_goal_resume_restart.py

@@ -172,6 +172,18 @@ def _resume_event() -> MessageEvent:
 
 
 class TestGatewayResumeRestartsWork:
+    def test_goal_kickoff_keeps_current_prompt_identity(self):
+        runner, adapter = _make_runner()
+        event = _resume_event()
+        event.channel_prompt = "Channel hint."
+
+        runner._enqueue_goal_turn(event, "kickoff", label="test kickoff", kickoff=True)
+
+        pending = adapter._pending_messages[_GW_KEY]
+        assert pending.message_id == event.message_id
+        assert pending.channel_prompt == "Channel hint."
+        assert pending.preserve_prompt_pins is False
+
     @pytest.mark.asyncio
     async def test_resume_after_budget_exhaustion_enqueues_continuation(
         self, hermes_home
@@ -188,6 +200,9 @@ async def test_resume_after_budget_exhaustion_enqueues_continuation(
             "— otherwise the goal sits idle until the next real user message"
         )
         assert pending.text.startswith("[Continuing toward your standing goal]")
+        assert pending.internal is False
+        assert pending.channel_prompt is None
+        assert getattr(pending, "preserve_prompt_pins", False) is True
         # The pause/clear stale-work guard must recognize the queued turn as
         # a synthetic goal continuation so it can be cleaned up on /goal pause.
         assert GatewayRunner._is_goal_continuation_event(pending)

tests/gateway/test_heartbeat_poller.py

@@ -78,6 +78,7 @@ async def handler(event):
             await runner._heartbeat_poll_once(watch)
         assert len(received) == 1
         assert not received[0].internal  # authorization and emergency-stop still apply
+        assert getattr(received[0], "preserve_prompt_pins", False) is True
         # Gateway-stamped provenance: the turn may end silently (#113031) and is not a reply to
         # the message that registered the watch (#112149).
         assert display_kind_for_event(received[0]) == INTERNAL_NOTIFICATION_DISPLAY_KIND

tests/gateway/test_internal_event_pin_wiring.py

@@ -11,6 +11,7 @@
 
 from __future__ import annotations
 
+import dataclasses
 from datetime import datetime
 from unittest.mock import AsyncMock, MagicMock
 
@@ -320,6 +321,77 @@ async def test_event_backed_followup_overrides_inherited_channel_prompt(monkeypa
     assert len(calls) == 1
     assert calls[0]["channel_prompt"] == "Queued event prompt.", "event prompt must win over the inherited one"
 
+
+@pytest.mark.asyncio
+async def test_non_internal_synthetic_event_preserves_all_prompt_pins(monkeypatch):
+    """#126109: non-internal synthetic turns inherit both session-context and channel pins."""
+    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)
+
+    source = _human_thread_source()
+    await _drive(runner, ((False, source),), channel_prompt="Channel hint.")
+
+    synthetic = runner._synthetic_prompt_event(source, "[Continuing toward your standing goal]")
+    assert synthetic.internal is False
+    assert synthetic.channel_prompt is None
+    assert synthetic.source.message_id is None
+    assert synthetic.preserve_prompt_pins is True
+    # pre_gateway_dispatch rewrite hooks rebuild events with dataclasses.replace.
+    synthetic = dataclasses.replace(synthetic, text=synthetic.text + " (rewritten)")
+    assert synthetic.preserve_prompt_pins is True
+
+    await runner._handle_message_with_agent(synthetic, synthetic.source, KEY, 1)
+    await _drive(runner, ((False, source),), channel_prompt="Channel hint.")
+
+    assert [call["channel_prompt"] for call in calls] == ["Channel hint."] * 3
+    ephemeral = [_effective_ephemeral(runner, call) for call in calls]
+    assert "Triggering message: provided per-turn" in ephemeral[0]
+    assert ephemeral[0] == ephemeral[1] == ephemeral[2]
+    assert runner._peek_session_state(KEY).conversation.channel_pin == ("Channel hint.", PARENT_ID)
+
+
+@pytest.mark.asyncio
+async def test_queued_synthetic_event_preserves_channel_pin(monkeypatch):
+    """In-band FIFO continuations consume the same preservation marker."""
+    runner = _make_runner(monkeypatch)
+    calls: list[dict] = []
+    _capture(runner, calls)
+    runner._run_agent_deliver_first_response = AsyncMock()
+    runner._refresh_agent_cache_message_count = AsyncMock()
+    runner._prepare_profile_scoped_inbound_message_text = AsyncMock(return_value="continue")
+    runner._session_key_for_source = lambda source: KEY
+
+    source = _human_thread_source()
+    await _drive(runner, ((False, source),), channel_prompt="Channel hint.")
+    first = calls[0]
+    turn_ctx = TurnContext(
+        source=first["source"],
+        context_prompt=first["context_prompt"],
+        channel_prompt=first["channel_prompt"],
+        session_key=KEY,
+        session_id="sess-wiring",
+        run_generation=1,
+        history=[],
+    )
+    adapter = MagicMock()
+    adapter._active_sessions = {}
+    pending = runner._synthetic_prompt_event(source, "[goal] continue")
+
+    await runner._run_agent_queued_followup(
+        turn_ctx, adapter, pending.text, pending, "done",
+        {"final_response": "done", "messages": []}, None,
+    )
+
+    assert len(calls) == 2
+    assert calls[1]["channel_prompt"] == "Channel hint."
+    assert runner._peek_session_state(KEY).conversation.channel_pin == ("Channel hint.", PARENT_ID)
+
 @pytest.mark.asyncio
 async def test_first_internal_event_after_restart_rehydrates_durable_prompt_pins(monkeypatch):
     config = GatewayConfig()
@@ -362,6 +434,37 @@ async def test_first_internal_event_after_restart_rehydrates_durable_prompt_pins
     assert _effective_ephemeral(after, first_internal) == _effective_ephemeral(before, human)
 
 
+@pytest.mark.asyncio
+async def test_first_synthetic_event_after_restart_rehydrates_durable_prompt_pins(monkeypatch):
+    """#126109 across a restart: a goal/heartbeat continuation is usually the first turn a
+    restarted gateway runs for a session, and it is non-internal, so it must adopt the durable
+    pins on the same decision the in-process reuse makes (``preserve_prompt_pins``)."""
+    config = GatewayConfig()
+    config.platforms[Platform.DISCORD] = PlatformConfig(
+        enabled=True,
+        channel_overrides={PARENT_ID: ChannelOverride(system_prompt="Parent persona.")},
+    )
+    durable: dict = {}
+    before = _make_runner(monkeypatch, config, durable_prompt_pin=durable)
+    calls_before: list[dict] = []
+    _capture(before, calls_before)
+    await _drive(before, ((False, _human_thread_source()),), channel_prompt="Channel hint.")
+    assert durable["value"]["channel_prompt"] == "Channel hint."
+
+    after = _make_runner(monkeypatch, config, durable_prompt_pin=durable)
+    calls_after: list[dict] = []
+    _capture(after, calls_after)
+    synthetic = after._synthetic_prompt_event(_human_thread_source(), "[Continuing toward your standing goal]")
+    assert synthetic.internal is False and synthetic.preserve_prompt_pins is True
+    await after._handle_message_with_agent(synthetic, synthetic.source, KEY, 1)
+    await _drive(after, ((False, _human_thread_source()),), channel_prompt="Channel hint.")
+
+    human, first_synthetic, next_human = calls_before[0], calls_after[0], calls_after[1]
+    assert [human["channel_prompt"], first_synthetic["channel_prompt"], next_human["channel_prompt"]] == ["Channel hint."] * 3
+    assert _effective_ephemeral(after, first_synthetic) == _effective_ephemeral(before, human)
+    assert _effective_ephemeral(after, next_human) == _effective_ephemeral(before, human)
+
+
 @pytest.mark.asyncio
 async def test_human_first_after_restart_ignores_stale_durable_prompt_pin(monkeypatch):
     durable = {