diff --git a/gateway/platforms/event.py b/gateway/platforms/event.py index 4e9faa4097668..44c8b12e43f72 100644 --- a/gateway/platforms/event.py +++ b/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) diff --git a/gateway/run_agent_cache.py b/gateway/run_agent_cache.py index 661b0576ed333..34d6d141b1e04 100644 --- a/gateway/run_agent_cache.py +++ b/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) diff --git a/gateway/run_busy.py b/gateway/run_busy.py index 89cadb02a82e8..c90dbad767310 100644 --- a/gateway/run_busy.py +++ b/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( diff --git a/gateway/run_goals.py b/gateway/run_goals.py index 4bb20245512c3..d78195eb8433f 100644 --- a/gateway/run_goals.py +++ b/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.""" diff --git a/gateway/run_turn.py b/gateway/run_turn.py index c980663fbf7d6..7ad565a90f546 100644 --- a/gateway/run_turn.py +++ b/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. diff --git a/gateway/slash_commands_goals.py b/gateway/slash_commands_goals.py index 2aa7317c38982..e7a3b700132c8 100644 --- a/gateway/slash_commands_goals.py +++ b/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) diff --git a/tests/gateway/test_goal_continuation_drain.py b/tests/gateway/test_goal_continuation_drain.py index 45e0747a099cf..c09dc16b32598 100644 --- a/tests/gateway/test_goal_continuation_drain.py +++ b/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 diff --git a/tests/gateway/test_goal_resume_restart.py b/tests/gateway/test_goal_resume_restart.py index 34e1639878d23..c6f655310f463 100644 --- a/tests/gateway/test_goal_resume_restart.py +++ b/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) diff --git a/tests/gateway/test_heartbeat_poller.py b/tests/gateway/test_heartbeat_poller.py index f6e5b99f47d76..0f280aa88a1b8 100644 --- a/tests/gateway/test_heartbeat_poller.py +++ b/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 diff --git a/tests/gateway/test_internal_event_pin_wiring.py b/tests/gateway/test_internal_event_pin_wiring.py index f28611f516a9a..22a2d9502e52b 100644 --- a/tests/gateway/test_internal_event_pin_wiring.py +++ b/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 = { diff --git a/tests/gateway/test_queue_consumption.py b/tests/gateway/test_queue_consumption.py index 79d323262cccb..f2a811c577345 100644 --- a/tests/gateway/test_queue_consumption.py +++ b/tests/gateway/test_queue_consumption.py @@ -248,6 +248,25 @@ def test_non_photo_media_followups_each_keep_their_own_fifo_turn(self): ] assert runner._queue_depth(session_key, adapter=adapter) == 5 + def test_synthetic_and_human_photo_keep_separate_prompt_identity(self): + """A synthetic continuation must not absorb a real human photo into its prompt identity.""" + runner, adapter = self._make_runner_and_adapter() + session_key = "telegram:user:synthetic-photo" + + synthetic = self._text_event("[goal] continue") + synthetic.preserve_prompt_pins = True + photo = self._media_event("/tmp/user.jpg", "image/jpeg", MessageType.PHOTO, text="look") + + runner._queue_or_replace_pending_event(session_key, synthetic) + runner._queue_or_replace_pending_event(session_key, photo) + + head = adapter._pending_messages[session_key] + assert head is synthetic + assert head.media_urls == [] + assert head.preserve_prompt_pins is True + assert runner._queued_events[session_key] == [photo] + assert photo.preserve_prompt_pins is False + def test_photo_burst_still_merges_into_one_head_event(self): """Control: rapid photos (and a trailing caption text) still collapse into one album event so the next turn sees the whole burst."""