fix(gateway): route restart internals away from shared chats

gateway/platforms/base.py

@@ -23,6 +23,7 @@
 from abc import ABC, abstractmethod
 from urllib.parse import urlsplit
 
+from gateway.session import is_shared_audience
 from utils import normalize_proxy_url
 
 logger = logging.getLogger(__name__)
@@ -57,6 +58,7 @@
 _HISTORY_MEDIA_LOOKUP_ADMISSION = threading.BoundedSemaphore(
     _HISTORY_MEDIA_LOOKUP_MAX_WORKERS
 )
+_PRIVATE_REPLY_PUBLIC_FALLBACK = "Gateway acknowledged the request; operational details are private."
 
 
 def _platform_name(platform) -> str:
@@ -2667,6 +2669,22 @@ def _invalidate_pending_stt_cache(event: MessageEvent) -> None:
             delattr(event, attr)
 
 
+class PrivateReply(EphemeralReply):
+    """System reply whose full text must not default to a shared origin chat."""
+
+    public_fallback: Optional[str]
+
+    def __new__(
+        cls,
+        text: str,
+        ttl_seconds: Optional[int] = None,
+        public_fallback: Optional[str] = None,
+    ):
+        instance = super().__new__(cls, text, ttl_seconds=ttl_seconds)
+        instance.public_fallback = public_fallback
+        return instance
+
+
 def merge_pending_message_event(
     pending_messages: Dict[str, MessageEvent],
     session_key: str,
@@ -4193,6 +4211,69 @@ async def send_private_notice(
             metadata=metadata,
         )
 
+    def _supports_private_notice_delivery(self) -> bool:
+        """Return True when ``send_private_notice`` is not the public fallback."""
+        method = getattr(type(self), "send_private_notice", None)
+        if method is not BasePlatformAdapter.send_private_notice:
+            return True
+        # Tests and plugin shims sometimes install an instance method after
+        # construction. Treat that as explicit support, but keep the class
+        # default from leaking confidential text through its public fallback.
+        return "send_private_notice" in getattr(self, "__dict__", {})
+
+    async def _send_private_reply_or_fallback(
+        self,
+        event: MessageEvent,
+        reply: PrivateReply,
+        *,
+        reply_to: Optional[str] = None,
+        metadata: Optional[Dict[str, Any]] = None,
+    ) -> SendResult:
+        """Deliver a private reply without falling back to shared full text."""
+        source = event.source
+        if source is None:
+            logger.warning("[%s] Dropping private reply: event has no source", self.name)
+            return SendResult(success=False, error="missing message source")
+        if not is_shared_audience(source):
+            return await self._send_with_retry(
+                chat_id=source.chat_id,
+                content=reply.text,
+                reply_to=reply_to,
+                metadata=metadata,
+            )
+
+        if self._supports_private_notice_delivery() and source.user_id:
+            try:
+                result = await self.send_private_notice(
+                    chat_id=source.chat_id,
+                    user_id=source.user_id,
+                    content=reply.text,
+                    reply_to=reply_to,
+                    metadata=metadata,
+                )
+                if result is None or getattr(result, "success", True) is not False:
+                    return result
+                logger.warning(
+                    "[%s] send_private_notice failed for private reply; "
+                    "using public fallback: %s",
+                    self.name,
+                    getattr(result, "error", "send returned success=False"),
+                )
+            except Exception as exc:
+                logger.warning(
+                    "[%s] send_private_notice raised for private reply; "
+                    "using public fallback: %s",
+                    self.name,
+                    exc,
+                )
+
+        return await self._send_with_retry(
+            chat_id=source.chat_id,
+            content=reply.public_fallback or _PRIVATE_REPLY_PUBLIC_FALLBACK,
+            reply_to=reply_to,
+            metadata=metadata,
+        )
+
     async def send_typing(self, chat_id: str, metadata=None) -> None:
         """
         Send a typing indicator.
@@ -5961,13 +6042,28 @@ async def handle_message(self, event: MessageEvent) -> None:
                     response = await self._message_handler(event)
                     _text, _eph_ttl = self._unwrap_ephemeral(response)
                     if _text:
-                        _r = await self._send_with_retry(
-                            chat_id=event.source.chat_id,
-                            content=_text,
-                            reply_to=_reply_anchor_for_event(event),
-                            metadata=_mark_notify_metadata(_thread_meta),
-                        )
-                        if _eph_ttl > 0 and _r.success and _r.message_id:
+                        _reply_anchor = _reply_anchor_for_event(event)
+                        _metadata = _mark_notify_metadata(_thread_meta)
+                        if isinstance(response, PrivateReply):
+                            _r = await self._send_private_reply_or_fallback(
+                                event,
+                                response,
+                                reply_to=_reply_anchor,
+                                metadata=_metadata,
+                            )
+                        else:
+                            _r = await self._send_with_retry(
+                                chat_id=event.source.chat_id,
+                                content=_text,
+                                reply_to=_reply_anchor,
+                                metadata=_metadata,
+                            )
+                        if (
+                            _eph_ttl > 0
+                            and getattr(_r, "success", False)
+                            and getattr(_r, "message_id", None)
+                            and not isinstance(response, PrivateReply)
+                        ):
                             self._schedule_ephemeral_delete(
                                 chat_id=event.source.chat_id,
                                 message_id=_r.message_id,
@@ -6014,13 +6110,28 @@ async def handle_message(self, event: MessageEvent) -> None:
                         response = await self._message_handler(event)
                         _text, _eph_ttl = self._unwrap_ephemeral(response)
                         if _text:
-                            _r = await self._send_with_retry(
-                                chat_id=event.source.chat_id,
-                                content=_text,
-                                reply_to=_reply_anchor_for_event(event),
-                                metadata=_mark_notify_metadata(_thread_meta),
-                            )
-                            if _eph_ttl > 0 and _r.success and _r.message_id:
+                            _reply_anchor = _reply_anchor_for_event(event)
+                            _metadata = _mark_notify_metadata(_thread_meta)
+                            if isinstance(response, PrivateReply):
+                                _r = await self._send_private_reply_or_fallback(
+                                    event,
+                                    response,
+                                    reply_to=_reply_anchor,
+                                    metadata=_metadata,
+                                )
+                            else:
+                                _r = await self._send_with_retry(
+                                    chat_id=event.source.chat_id,
+                                    content=_text,
+                                    reply_to=_reply_anchor,
+                                    metadata=_metadata,
+                                )
+                            if (
+                                _eph_ttl > 0
+                                and getattr(_r, "success", False)
+                                and getattr(_r, "message_id", None)
+                                and not isinstance(response, PrivateReply)
+                            ):
                                 self._schedule_ephemeral_delete(
                                     chat_id=event.source.chat_id,
                                     message_id=_r.message_id,
@@ -6162,6 +6273,7 @@ async def _stop_typing_task() -> None:
             # Call the handler (this can take a while with tool calls)
             response = await self._message_handler(event)
             is_ephemeral_response = isinstance(response, EphemeralReply)
+            private_response = response if isinstance(response, PrivateReply) else None
 
             # Slash-command handlers may return an EphemeralReply sentinel to
             # request that their reply message auto-delete after a TTL (used
@@ -6192,7 +6304,23 @@ async def _stop_typing_task() -> None:
                 )
                 response = None
             if not response:
+                private_response = None
                 logger.debug("[%s] Handler returned empty/None response for %s", self.name, event.source.chat_id)
+
+            # Private replies are already classified by the handler as
+            # owner-private/backend-internal. Route them before normal text
+            # and media handling can default them back to the origin chat.
+            if response and private_response is not None:
+                _final_thread_metadata = _mark_notify_metadata(_thread_metadata)
+                result = await self._send_private_reply_or_fallback(
+                    event,
+                    private_response,
+                    reply_to=_reply_anchor_for_event(event),
+                    metadata=_final_thread_metadata,
+                )
+                _record_delivery(result)
+                response = None
+
             if response:
                 # Capture [[as_document]] before extract_media strips it, so the
                 # dispatch partition below can route image-extension files

gateway/run.py

@@ -2390,6 +2390,7 @@ def _platform_has_bot_credential(platform: "Platform", platform_config: "Platfor
     build_session_context_prompt,
     build_channel_continuity_note,
     build_session_key,
+    is_shared_audience,
     is_shared_multi_user_session,
     neutralize_untrusted_inline_text,
 )
@@ -21656,6 +21657,8 @@ async def _send_restart_notification(self) -> Optional[tuple[str, str, Optional[
             platform_str = data.get("platform")
             chat_id = data.get("chat_id")
             chat_type = data.get("chat_type")
+            user_id = data.get("user_id")
+            user_name = data.get("user_name")
             thread_id = data.get("thread_id")
             message_id = data.get("message_id")
 
@@ -21693,12 +21696,70 @@ async def _send_restart_notification(self) -> Optional[tuple[str, str, Optional[
                     metadata["user_id"] = str(data["user_id"])
                 if data.get("scope_id"):
                     metadata["scope_id"] = str(data["scope_id"])
-            result = await transport.send(
-                platform,
-                str(chat_id),
-                "♻ Gateway restarted successfully. Your session continues.",
-                metadata=_non_conversational_metadata(metadata, platform=platform),
+            source = SessionSource(
+                platform=platform,
+                chat_id=str(chat_id),
+                chat_type=chat_type or "dm",
+                user_id=str(user_id) if user_id else None,
+                user_name=str(user_name) if user_name else None,
+                thread_id=str(thread_id) if thread_id else None,
+                message_id=str(message_id) if message_id else None,
+                scope_id=str(data["scope_id"]) if data.get("scope_id") else None,
+                delivered_via_upstream_relay=(
+                    data.get("delivered_via_upstream_relay") is True
+                ),
             )
+            private_text = "♻ Gateway restarted successfully. Your session continues."
+            metadata = _non_conversational_metadata(metadata, platform=platform)
+            result = None
+            if is_shared_audience(source):
+                if (
+                    source.user_id
+                    and getattr(
+                        transport.adapter,
+                        "_supports_private_notice_delivery",
+                        lambda: False,
+                    )()
+                ):
+                    try:
+                        result = await transport.adapter.send_private_notice(
+                            str(chat_id),
+                            source.user_id,
+                            private_text,
+                            metadata=metadata,
+                        )
+                    except Exception as exc:
+                        logger.warning(
+                            "Private restart notification to %s:%s user %s raised; "
+                            "using public fallback: %s",
+                            platform_str,
+                            chat_id,
+                            source.user_id,
+                            exc,
+                        )
+                    if result is not None and getattr(result, "success", True) is False:
+                        logger.warning(
+                            "Private restart notification to %s:%s user %s was not delivered: %s",
+                            platform_str,
+                            chat_id,
+                            source.user_id,
+                            getattr(result, "error", "send returned success=False"),
+                        )
+                        result = None
+                if result is None:
+                    result = await transport.send(
+                        platform,
+                        str(chat_id),
+                        "♻ Gateway restarted successfully. Operational details were kept private.",
+                        metadata=metadata,
+                    )
+            else:
+                result = await transport.send(
+                    platform,
+                    str(chat_id),
+                    private_text,
+                    metadata=metadata,
+                )
             # adapter.send() catches provider errors (e.g. "Chat not found")
             # and returns SendResult(success=False) rather than raising, so
             # we must inspect the result before claiming success — otherwise

gateway/session.py

@@ -309,6 +309,17 @@ def from_dict(cls, data: Dict[str, Any]) -> "SessionSource":
         )
 
 
+_PRIVATE_CHAT_TYPES = frozenset({"dm", "direct", "private", "c2c"})
+
+
+def is_shared_audience(source: Optional[SessionSource]) -> bool:
+    """Return True when a source is readable by more than the owner/requester."""
+    if source is None or source.platform == Platform.LOCAL:
+        return False
+    chat_type = str(getattr(source, "chat_type", "") or "").lower()
+    return chat_type not in _PRIVATE_CHAT_TYPES
+
+
 
 @dataclass
 class SessionContext:

gateway/slash_commands.py

@@ -33,7 +33,7 @@
 from agent.i18n import t
 from agent.turn_context import extract_api_content_sidecar
 from gateway.config import HomeChannel, Platform, PlatformConfig, persist_home_channel
-from gateway.platforms.base import EphemeralReply, MessageEvent, MessageType
+from gateway.platforms.base import EphemeralReply, MessageEvent, MessageType, PrivateReply
 from gateway.session import (
     AsyncSessionStore,
     SessionSource,
@@ -1547,8 +1547,8 @@ async def _handle_restart_command(self, event: MessageEvent) -> Union[str, Ephem
         if self._restart_requested or self._draining:
             count = self._running_agent_count()
             if count:
-                return t("gateway.draining", count=count)
-            return EphemeralReply(t("gateway.restart.in_progress"))
+                return PrivateReply(t("gateway.draining", count=count), ttl_seconds=0)
+            return PrivateReply(t("gateway.restart.in_progress"), ttl_seconds=0)
 
         # Save the requester's routing info so the new gateway process can
         # notify them once it comes back online.
@@ -1558,10 +1558,12 @@ async def _handle_restart_command(self, event: MessageEvent) -> Union[str, Ephem
                 "chat_id": event.source.chat_id,
                 "chat_type": event.source.chat_type,
             }
+            if event.source.user_id:
+                notify_data["user_id"] = event.source.user_id
+            if event.source.user_name:
+                notify_data["user_name"] = event.source.user_name
             if event.source.delivered_via_upstream_relay is True:
                 notify_data["delivered_via_upstream_relay"] = True
-                if event.source.user_id:
-                    notify_data["user_id"] = event.source.user_id
                 if event.source.scope_id:
                     notify_data["scope_id"] = event.source.scope_id
             if event.source.thread_id:
@@ -1628,8 +1630,8 @@ async def _handle_restart_command(self, event: MessageEvent) -> Union[str, Ephem
         else:
             self.request_restart(detached=True, via_service=False)
         if active_agents:
-            return t("gateway.draining", count=active_agents)
-        return EphemeralReply(t("gateway.restart.restarting"))
+            return PrivateReply(t("gateway.draining", count=active_agents), ttl_seconds=0)
+        return PrivateReply(t("gateway.restart.restarting"), ttl_seconds=0)
 
     async def _handle_version_command(self, event: MessageEvent) -> str:
         """Handle /version — show the running Hermes Agent version."""