fix(gateway): persist internal prompt pins across restart

gateway/run_agent_cache.py

@@ -618,8 +618,88 @@ 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 _rehydrate_prompt_pins(
+        self, session_key: Optional[str], expected_session_id: Optional[str] = None,
+    ) -> None:
+        """Restore durable prompt pins lazily for the first internal turn after a gateway restart."""
+        if not session_key:
+            return
+        state = self._session_state(session_key)
+        if state.conversation.ephemeral_pin is not None and state.conversation.channel_pin is not None:
+            return
+        store = getattr(self, "session_store", None)
+        getter = getattr(store, "get_prompt_pin", None)
+        if not callable(getter):
+            return
+        try:
+            persisted = getter(session_key, expected_session_id=expected_session_id)
+        except Exception:
+            logger.debug("Failed to read persisted prompt pin for %s", session_key, exc_info=True)
+            return
+        if not isinstance(persisted, dict):
+            return
+        context_key, context_prompt = persisted.get("context_key"), persisted.get("context_prompt")
+        channel_prompt, parent_chat_id = persisted.get("channel_prompt"), persisted.get("parent_chat_id")
+        if isinstance(context_key, str) and context_key and isinstance(context_prompt, str):
+            if state.conversation.ephemeral_pin is None:
+                state.conversation.ephemeral_pin = (context_key, context_prompt)
+        if (
+            (channel_prompt is None or isinstance(channel_prompt, str))
+            and (parent_chat_id is None or isinstance(parent_chat_id, str))
+            and state.conversation.channel_pin is None
+        ):
+            state.conversation.channel_pin = (channel_prompt, parent_chat_id)
+
+    def _prompt_pin_snapshot(self, session_key: Optional[str]) -> Optional[Dict[str, Any]]:
+        """Current coherent pair of context/channel pins, ready for durable storage."""
+        state = self._peek_session_state(session_key) if session_key else None
+        if state is None:
+            return None
+        ephemeral_pin, channel_pin = state.conversation.ephemeral_pin, state.conversation.channel_pin
+        if not (
+            isinstance(ephemeral_pin, tuple) and len(ephemeral_pin) == 2
+            and isinstance(ephemeral_pin[0], str) and isinstance(ephemeral_pin[1], str)
+            and isinstance(channel_pin, tuple) and len(channel_pin) == 2
+        ):
+            return None
+        channel_prompt, parent_chat_id = channel_pin
+        if channel_prompt is not None and not isinstance(channel_prompt, str):
+            return None
+        if parent_chat_id is not None and not isinstance(parent_chat_id, str):
+            return None
+        return {
+            "version": 1,
+            "context_key": ephemeral_pin[0],
+            "context_prompt": ephemeral_pin[1],
+            "channel_prompt": channel_prompt,
+            "parent_chat_id": parent_chat_id,
+        }
+
+    async def _persist_prompt_pins(
+        self, session_key: Optional[str], expected_session_id: Optional[str],
+    ) -> None:
+        """Persist changed human-turn pins before agent execution; failures stay cache-only."""
+        snapshot = self._prompt_pin_snapshot(session_key)
+        if snapshot is None or not session_key:
+            return
+        store = getattr(self, "session_store", None)
+        getter = getattr(store, "get_prompt_pin", None)
+        try:
+            if callable(getter) and getter(session_key) == snapshot:
+                return
+        except Exception:
+            logger.debug("Failed to compare persisted prompt pin for %s", session_key, exc_info=True)
+        try:
+            await self.async_session_store.set_prompt_pin(
+                session_key, snapshot, expected_session_id=expected_session_id,
+            )
+        except Exception:
+            # 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)
+
     def _pinned_session_context_prompt(
         self, context, redact_pii: bool, session_key: Optional[str], *, internal: bool = False,
+        expected_session_id: Optional[str] = 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).
@@ -631,6 +711,10 @@ def _pinned_session_context_prompt(
         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 None:
+            self._rehydrate_prompt_pins(session_key, expected_session_id)
+            _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)
@@ -643,6 +727,7 @@ def _pinned_session_context_prompt(
 
     def _pinned_channel_inputs(
         self, session_key: Optional[str], channel_prompt: Optional[str], source: SessionSource, *, internal: bool,
+        expected_session_id: Optional[str] = None,
     ):
         """``(channel_prompt, source)`` for this turn's agent run.
 
@@ -658,6 +743,10 @@ def _pinned_channel_inputs(
             return channel_prompt, source
         state = self._peek_session_state(session_key)
         pin = state.conversation.channel_pin if state else None
+        if pin is None:
+            self._rehydrate_prompt_pins(session_key, expected_session_id)
+            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

gateway/run_startup.py

@@ -1767,7 +1767,9 @@ async def _process_handoff(self, row: Dict[str, Any], profile_name: Optional[str
         # Ensure a session_store entry exists for this key; switch_session then re-points it.
         await self.async_session_store.get_or_create_session(dest.source)
         # switch_session ends the prior session and reopens the CLI session under the new key.
-        switched = await self.async_session_store.switch_session(session_key, cli_session_id)
+        switched = await self.async_session_store.switch_session(
+            session_key, cli_session_id, preserve_prompt_pin=False,
+        )
         if switched is None:
             raise RuntimeError(f"could not switch session key {session_key} → {cli_session_id}")
         # Evict the cached AIAgent (rebuild against the CLI session_id, like /resume) and clear stale

gateway/run_turn.py

@@ -2077,6 +2077,7 @@ async def _hmwa_prepare_turn(self, event, source, session_entry, session_key, _q
         # 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, internal=event.internal,
+            expected_session_id=session_entry.session_id,
         )
 
         # Per-turn notes ride the user message via the api_content sidecar, NOT context_prompt
@@ -2198,7 +2199,12 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g
             # 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,
+                expected_session_id=_run_start_session_id,
             )
+            if not event.internal:
+                # Persist the coherent context+channel pair before execution: a crash during the
+                # human turn may be followed by an internal startup-resume on the next process.
+                await self._persist_prompt_pins(session_key, _run_start_session_id)
             agent_result = await self._run_agent(
                 message=message_text, context_prompt=prepared.context_prompt, history=history, source=_turn_source,
                 session_id=_run_start_session_id, session_key=session_key,

gateway/session.py

@@ -477,6 +477,32 @@ def sanitize_model_override(override: Optional[Dict[str, Any]]) -> Optional[Dict
     return cleaned or None
 
 
+_PROMPT_PIN_VERSION = 1
+
+
+def sanitize_prompt_pin(pin: Any) -> Optional[Dict[str, Any]]:
+    """Validated durable snapshot of the exact ephemeral inputs reused by internal turns."""
+    if not isinstance(pin, dict) or pin.get("version") != _PROMPT_PIN_VERSION:
+        return None
+    context_key = pin.get("context_key")
+    context_prompt = pin.get("context_prompt")
+    channel_prompt = pin.get("channel_prompt")
+    parent_chat_id = pin.get("parent_chat_id")
+    if not isinstance(context_key, str) or not context_key or not isinstance(context_prompt, str):
+        return None
+    if channel_prompt is not None and not isinstance(channel_prompt, str):
+        return None
+    if parent_chat_id is not None and not isinstance(parent_chat_id, str):
+        return None
+    return {
+        "version": _PROMPT_PIN_VERSION,
+        "context_key": context_key,
+        "context_prompt": context_prompt,
+        "channel_prompt": channel_prompt,
+        "parent_chat_id": parent_chat_id,
+    }
+
+
 @dataclass
 class SessionEntry:
     """Routing-index entry: maps a session key to its current session ID and metadata."""
@@ -533,6 +559,9 @@ class SessionEntry:
     # "default" spelled out). The key namespace only says where the turn RUNS; after a restart this is
     # what says which bot may deliver to it. None = unknown (row predates the field, or standalone).
     transport_profile: Optional[str] = None
+    # Exact session-context/channel inputs from the last human turn. Append-only dataclass field so
+    # older positional construction of transport_profile keeps its meaning.
+    prompt_pin: Optional[Dict[str, Any]] = None
 
     # Fields (de)serialized verbatim, in wire order (``from_dict`` reads them with
     # ``data.get(name, <dataclass default>)``), split around the three ISO-datetime/token keys.
@@ -562,6 +591,10 @@ def to_dict(self) -> Dict[str, Any]:
         if self.model_override:
             # Defence-in-depth against an unsanitized dict stored directly.
             result["model_override"] = sanitize_model_override(self.model_override)
+        if self.prompt_pin:
+            # Same defence-in-depth: routing JSON must never preserve malformed pin state.
+            if pin := sanitize_prompt_pin(self.prompt_pin):
+                result["prompt_pin"] = pin
         if self.transport_profile:
             result["transport_profile"] = self.transport_profile
         if self.origin:
@@ -604,6 +637,7 @@ def from_dict(cls, data: Dict[str, Any]) -> "SessionEntry":
             last_resume_marked_at=_parse_iso(data.get("last_resume_marked_at")),
             active_turn_token=token, active_turn_started_at=started_at,
             model_override=sanitize_model_override(data.get("model_override")),
+            prompt_pin=sanitize_prompt_pin(data.get("prompt_pin")),
             transport_profile=transport_profile if isinstance(transport_profile, str) and transport_profile else None,
             **plain,
         )
@@ -1110,6 +1144,48 @@ def get_model_override(self, session_key: str) -> Optional[Dict[str, str]]:
             entry = self._entry_locked(session_key)
             return dict(entry.model_override) if entry and entry.model_override else None
 
+    def set_prompt_pin(
+        self, session_key: str, pin: Optional[Dict[str, Any]], *,
+        expected_session_id: Optional[str] = None,
+    ) -> bool:
+        """Persist effective prompt inputs without letting a stale turn cross a boundary.
+
+        The candidate is durable before the in-memory entry is published. The expected session id
+        is the turn's launch identity: a concurrent reset or resume makes this a no-op instead of
+        writing the old conversation's system bytes onto the new one.
+        """
+        cleaned = sanitize_prompt_pin(pin)
+        if pin is not None and cleaned is None:
+            return False
+        with self._lock:
+            entry = self._entry_locked(session_key)
+            if entry is None:
+                return False
+            if expected_session_id is not None and entry.session_id != expected_session_id:
+                return False
+            if entry.prompt_pin == cleaned:
+                return True
+            candidate = entry.to_dict()
+            if cleaned is None:
+                candidate.pop("prompt_pin", None)
+            else:
+                candidate["prompt_pin"] = cleaned
+            self._save_entry(session_key, entry_data=candidate, lock_held=True)
+            entry.prompt_pin = dict(cleaned) if cleaned is not None else None
+            return True
+
+    def get_prompt_pin(
+        self, session_key: str, *, expected_session_id: Optional[str] = None,
+    ) -> Optional[Dict[str, Any]]:
+        """Return the prompt pin only while the route still owns the caller's session."""
+        with self._lock:
+            entry = self._entry_locked(session_key)
+            if entry is None:
+                return None
+            if expected_session_id is not None and entry.session_id != expected_session_id:
+                return None
+            return dict(entry.prompt_pin) if entry.prompt_pin else None
+
     def reset_session(self, session_key: str, display_name: Optional[str] = None) -> Optional[SessionEntry]:
         """Force reset a session, creating a new session ID."""
         with self._lock:
@@ -1224,13 +1300,16 @@ def remove_by_session_id(self, session_id: str) -> int:
     # restart-resume freshness gate (#85709).
     def switch_session(
         self, session_key: str, target_session_id: str, *, expected_session_id: Optional[str] = None,
+        preserve_prompt_pin: bool = True,
     ) -> Optional[SessionEntry]:
         """Point a session key at an existing session ID (``/resume``): ends the current row and
         reopens the target so resume matches the CLI.
 
         ``expected_session_id`` makes the repoint a compare-and-swap: ``None`` is returned when
         the key no longer points at that session, so a caller that resolved against a snapshot
         across an await (async-delegation re-pin) cannot overwrite a concurrent /new or /resume.
+        Prompt pins follow non-boundary repoints by default; /resume opts out explicitly because it
+        starts a different conversation on the same routing key.
         """
         with self._lock:
             old_entry = self._entry_locked(session_key)
@@ -1247,6 +1326,10 @@ def switch_session(
             new_entry = self._replace_route_locked(
                 session_key, old_entry, target_session_id, _now(),
                 display_name=old_entry.display_name, model_override=old_entry.model_override,
+                prompt_pin=(
+                    dict(old_entry.prompt_pin)
+                    if preserve_prompt_pin and old_entry.prompt_pin is not None else None
+                ),
             )
 
         if self._db_for_key(session_key) and old_entry.session_id:

gateway/slash_commands_session.py

@@ -893,7 +893,9 @@ async def _handle_resume_command(self, event: MessageEvent) -> str:
         if current_entry.session_id == target_id:
             return t("gateway.resume.already_on", name=name)
         self._release_running_agent_state(session_key)
-        new_entry = await self.async_session_store.switch_session(session_key, target_id)
+        new_entry = await self.async_session_store.switch_session(
+            session_key, target_id, preserve_prompt_pin=False,
+        )
         if not new_entry:
             return t("gateway.resume.switch_failed")
         # Conversation boundary: all conversation-scoped state + security state in one funnel call.

tests/gateway/test_handoff_secondary_profile_adapter.py

@@ -67,7 +67,7 @@ def _make_multiplex_runner():
         platform=Platform.TELEGRAM, chat_type="dm",
     ))
 
-    async def _switch(key, sid):
+    async def _switch(key, sid, *, preserve_prompt_pin=True):
         captured["session_key"] = key
         return SessionEntry(
             session_key=key, session_id=sid,

tests/gateway/test_handoff_thread_session_key.py

@@ -29,7 +29,7 @@
 
 from gateway.config import GatewayConfig, HomeChannel, Platform, PlatformConfig
 from gateway.run import GatewayRunner
-from gateway.session import SessionSource, build_session_key
+from gateway.session import SessionEntry, SessionSource, build_session_key
 
 
 def _organic_discord_thread_key(thread_id: str, parent_id: str, user_id: str) -> str:
@@ -212,3 +212,50 @@ def test_matrix_handoff_key_matches_the_thread_reply_key(is_dm, monkeypatch):
     _dest, handoff = _handoff_destination(Platform.MATRIX, room_id, "$seed", None, adapter)
     assert handoff == _organic_matrix_reply_key(adapter, room_id, "$seed")
     assert handoff.split(":")[3] == ("dm" if is_dm else "group")
+
+@pytest.mark.asyncio
+async def test_cli_handoff_drops_previous_route_prompt_pin():
+    """CLI handoff is a conversation boundary: the synthetic internal turn must not inherit
+    prompt bytes from whatever conversation previously occupied the destination routing key."""
+    runner = object.__new__(GatewayRunner)
+    source = SessionSource(
+        platform=Platform.DISCORD,
+        chat_id="handoff-thread",
+        chat_type="thread",
+        user_id="system:handoff",
+        thread_id="handoff-thread",
+    )
+    dest = SimpleNamespace(
+        source=source,
+        platform_name="discord",
+        home=SimpleNamespace(chat_id="parent-channel"),
+        effective_thread_id="handoff-thread",
+        transport=SimpleNamespace(send=AsyncMock()),
+    )
+    runner._handoff_resolve_destination = AsyncMock(return_value=dest)
+    runner._handoff_session_key = MagicMock(return_value="agent:main:discord:thread:handoff-thread:handoff-thread")
+    runner._evict_cached_agent = MagicMock()
+    runner._release_running_agent_state = MagicMock()
+    runner._handle_message = AsyncMock(return_value=None)
+
+    store = MagicMock()
+    store.get_or_create_session.return_value = SessionEntry(
+        session_key="agent:main:discord:thread:handoff-thread:handoff-thread",
+        session_id="previous-gateway-session",
+        created_at=None,
+        updated_at=None,
+    )
+    store.switch_session.return_value = SessionEntry(
+        session_key="agent:main:discord:thread:handoff-thread:handoff-thread",
+        session_id="cli-session",
+        created_at=None,
+        updated_at=None,
+    )
+    runner.session_store = store
+
+    await runner._process_handoff(
+        {"id": "cli-session", "title": "work", "handoff_platform": "discord"}
+    )
+
+    assert store.switch_session.call_args.kwargs["preserve_prompt_pin"] is False
+