feat(telegram): support Bot API 10 guest messages

cli-config.yaml.example

@@ -818,6 +818,7 @@ platform_toolsets:
 #     # guest_mode lets explicit @mentions from non-allowlisted groups through.
 #     # Default false; ordinary messages, replies, and regex wake words stay blocked.
 #     guest_mode: false
+#     guest_thinking_text: "💭 Thinking..."  # Bot API 10.0 guest reply placeholder
 #     # allowed_chats: ["-1001234567890"]
 #     extra:
 #       disable_link_previews: false  # Set true to suppress Telegram URL previews in bot messages

gateway/config.py

@@ -1141,6 +1141,8 @@ def _merge_platform_map(source_platforms: Any) -> None:
                     bridged["exclusive_bot_mentions"] = platform_cfg["exclusive_bot_mentions"]
                 if plat == Platform.TELEGRAM and "observe_unmentioned_group_messages" in platform_cfg:
                     bridged["observe_unmentioned_group_messages"] = platform_cfg["observe_unmentioned_group_messages"]
+                if plat == Platform.TELEGRAM and "guest_thinking_text" in platform_cfg:
+                    bridged["guest_thinking_text"] = platform_cfg["guest_thinking_text"]
                 if "dm_policy" in platform_cfg:
                     bridged["dm_policy"] = platform_cfg["dm_policy"]
                 if "allow_from" in platform_cfg:

gateway/platforms/base.py

@@ -62,10 +62,11 @@ def _thread_metadata_for_source(source, reply_to_message_id: str | None = None)
     synthetic/resumed sends that have no reply anchor fall back to Telegram's
     ``direct_messages_topic_id`` when the Bot API supports it.
     """
+    metadata = dict(getattr(source, "platform_metadata", None) or {})
     thread_id = getattr(source, "thread_id", None)
     if thread_id is None:
-        return None
-    metadata = {"thread_id": thread_id}
+        return metadata or None
+    metadata["thread_id"] = thread_id
     if _platform_name(getattr(source, "platform", None)) == "telegram" and getattr(source, "chat_type", None) == "dm":
         metadata["telegram_dm_topic_reply_fallback"] = True
         tid = str(thread_id)

gateway/run.py

@@ -472,6 +472,39 @@ def render_notice_line(notice) -> str:
     return str(getattr(notice, "text", "") or "").strip()
 
 
+
+def _is_telegram_guest_metadata(metadata: Optional[Dict[str, Any]]) -> bool:
+    return bool(
+        metadata
+        and (
+            metadata.get("telegram_guest_query_id")
+            or metadata.get("telegram_guest_inline_message_id")
+        )
+    )
+
+
+def _should_suppress_final_send_after_stream(
+    *,
+    response_text: str,
+    response_previewed: bool,
+    response_transformed: bool,
+    stream_consumer: Any,
+    metadata: Optional[Dict[str, Any]],
+) -> bool:
+    if not response_text or response_text == "(empty)":
+        return False
+    if response_transformed:
+        return False
+    streamed = bool(
+        stream_consumer
+        and getattr(stream_consumer, "final_response_sent", False)
+    )
+    content_delivered = bool(
+        stream_consumer
+        and getattr(stream_consumer, "final_content_delivered", False)
+    )
+    return bool(streamed or content_delivered)
+
 async def _send_or_update_status_coro(adapter, chat_id, status_key, content, metadata):
     """Route a status message through adapter.send_or_update_status when supported.
 
@@ -13874,13 +13907,35 @@ def _thread_metadata_for_source(
         reply_to_message_id: Optional[str] = None,
     ) -> Optional[Dict[str, Any]]:
         """Build the metadata dict platforms need for thread-aware replies."""
-        return self._thread_metadata_for_target(
+        metadata = self._thread_metadata_for_target(
             getattr(source, "platform", None),
             getattr(source, "chat_id", None),
             getattr(source, "thread_id", None),
             chat_type=getattr(source, "chat_type", None),
             reply_to_message_id=reply_to_message_id or getattr(source, "message_id", None),
         )
+        platform_metadata = getattr(source, "platform_metadata", None)
+        if platform_metadata:
+            merged = dict(platform_metadata)
+            if metadata:
+                merged.update(metadata)
+            return merged
+        return metadata
+
+    def _progress_metadata_for_source(
+        self,
+        source,
+        reply_to_message_id: Optional[str],
+        progress_thread_id: Optional[str],
+    ) -> Optional[Dict[str, Any]]:
+        """Build metadata for status/progress messages, preserving platform metadata."""
+        if not progress_thread_id:
+            return self._thread_metadata_for_source(source, reply_to_message_id)
+        if str(progress_thread_id) == str(getattr(source, "thread_id", None)):
+            return self._thread_metadata_for_source(source, reply_to_message_id)
+        metadata = dict(getattr(source, "platform_metadata", None) or {})
+        metadata["thread_id"] = str(progress_thread_id)
+        return metadata
 
     def _thread_metadata_for_target(
         self,
@@ -16969,11 +17024,11 @@ def progress_callback(event_type: str, tool_name: str = None, preview: str = Non
         _progress_thread_id = _resolve_progress_thread_id(
             source.platform, source.thread_id, event_message_id,
         )
-        _progress_metadata = (
-            self._thread_metadata_for_source(source, event_message_id)
-            if _progress_thread_id == source.thread_id
-            else {"thread_id": _progress_thread_id}
-        ) if _progress_thread_id else None
+        _progress_metadata = self._progress_metadata_for_source(
+            source,
+            event_message_id,
+            _progress_thread_id,
+        )
         _progress_metadata = _non_conversational_metadata(_progress_metadata, platform=source.platform)
         _progress_reply_to = (
             event_message_id
@@ -17427,7 +17482,11 @@ def _event_callback_sync(event_type: str, context: dict) -> None:
                 "reply_to_message_id": event_message_id,
             }
         else:
-            _status_thread_metadata = self._thread_metadata_for_source(source, event_message_id) if _progress_thread_id else None
+            _status_thread_metadata = self._progress_metadata_for_source(
+                source,
+                event_message_id,
+                _progress_thread_id,
+            )
 
         def _status_callback_sync(event_type: str, message: str) -> None:
             if not _status_adapter or not _run_still_current():
@@ -19534,7 +19593,14 @@ def _stream_confirmed_final_delivery(
                 _final,
                 previewed=_previewed,
             )
-            if not _is_empty_sentinel and not _transformed and (_streamed or _content_delivered):
+            _suppress_final_send = _should_suppress_final_send_after_stream(
+                response_text=_final,
+                response_previewed=_previewed,
+                response_transformed=_transformed,
+                stream_consumer=_sc,
+                metadata=self._thread_metadata_for_source(source, event_message_id),
+            )
+            if _suppress_final_send:
                 logger.info(
                     "Suppressing normal final send for session %s: final delivery already confirmed (streamed=%s previewed=%s content_delivered=%s).",
                     session_key or "?",

gateway/session.py

@@ -22,6 +22,11 @@
 logger = logging.getLogger(__name__)
 
 
+def _session_key_component(value: Any) -> str:
+    """Return a safe single component for colon-delimited session keys."""
+    return str(value).replace(":", "_")
+
+
 def _now() -> datetime:
     """Return the current local time."""
     return datetime.now()
@@ -180,6 +185,7 @@ class SessionSource:
     # None => the gateway's active/default profile. Drives both session-key
     # namespacing and the per-turn config/credential scope.
     profile: Optional[str] = None
+    platform_metadata: Optional[Dict[str, Any]] = None  # Ephemeral platform routing data
 
     # Internal, wire-INVISIBLE trust signal: True when this event was delivered
     # to the gateway over the per-instance-authenticated relay WebSocket (the
@@ -889,15 +895,25 @@ def build_session_key(
     """
     ns = _session_key_namespace(profile)
     platform = source.platform.value
+    platform_metadata = getattr(source, "platform_metadata", None)
+    session_key_suffix = None
+    if isinstance(platform_metadata, dict):
+        raw_suffix = platform_metadata.get("session_key_suffix")
+        if raw_suffix:
+            session_key_suffix = _session_key_component(raw_suffix)
     if source.chat_type == "dm":
         dm_chat_id = source.chat_id
         if source.platform == Platform.WHATSAPP:
             dm_chat_id = canonical_whatsapp_identifier(source.chat_id)
 
         if dm_chat_id:
             if source.thread_id:
-                return f"{ns}:{platform}:dm:{dm_chat_id}:{source.thread_id}"
-            return f"{ns}:{platform}:dm:{dm_chat_id}"
+                key = f"{ns}:{platform}:dm:{dm_chat_id}:{source.thread_id}"
+            else:
+                key = f"{ns}:{platform}:dm:{dm_chat_id}"
+            if session_key_suffix:
+                key = f"{key}:{session_key_suffix}"
+            return key
         # No chat_id — fall back to the sender's own identifier before the
         # bare per-platform sink.  Without this, every DM from every user that
         # arrives without a chat_id (non-standard adapters / synthetic sources)
@@ -912,11 +928,19 @@ def build_session_key(
             )
         if dm_participant_id:
             if source.thread_id:
-                return f"{ns}:{platform}:dm:{dm_participant_id}:{source.thread_id}"
-            return f"{ns}:{platform}:dm:{dm_participant_id}"
+                key = f"{ns}:{platform}:dm:{dm_participant_id}:{source.thread_id}"
+            else:
+                key = f"{ns}:{platform}:dm:{dm_participant_id}"
+            if session_key_suffix:
+                key = f"{key}:{session_key_suffix}"
+            return key
         if source.thread_id:
-            return f"{ns}:{platform}:dm:{source.thread_id}"
-        return f"{ns}:{platform}:dm"
+            key = f"{ns}:{platform}:dm:{source.thread_id}"
+        else:
+            key = f"{ns}:{platform}:dm"
+        if session_key_suffix:
+            key = f"{key}:{session_key_suffix}"
+        return key
 
     participant_id = source.user_id_alt or source.user_id
     if participant_id and source.platform == Platform.WHATSAPP:
@@ -940,6 +964,8 @@ def build_session_key(
 
     if isolate_user and participant_id:
         key_parts.append(str(participant_id))
+    if session_key_suffix:
+        key_parts.append(str(session_key_suffix))
 
     return ":".join(key_parts)
 
@@ -1306,13 +1332,19 @@ def _recover_session_from_db(
         if not callable(finder):
             return None
         try:
+            platform_metadata = getattr(source, "platform_metadata", None)
+            exact_only = bool(
+                isinstance(platform_metadata, dict)
+                and platform_metadata.get("session_key_suffix")
+            )
             recovered = finder(
                 source=source.platform.value,
                 user_id=source.user_id,
                 session_key=session_key,
                 chat_id=source.chat_id,
                 chat_type=source.chat_type,
                 thread_id=source.thread_id,
+                exact_only=exact_only,
             )
         except Exception as exc:
             logger.debug("Gateway session DB recovery failed for %s: %s", session_key, exc)

hermes_cli/config.py

@@ -2413,6 +2413,7 @@ def _ensure_hermes_home_managed(home: Path):
         "reactions": False,            # Add 👀/✅/❌ reactions to messages during processing
         "channel_prompts": {},         # Per-chat/topic ephemeral system prompts (topics inherit from parent group)
         "allowed_chats": "",           # If set, bot ONLY responds in these group/supergroup chat IDs (whitelist)
+        "guest_thinking_text": "💭 Thinking...",  # Placeholder text for Bot API 10.0 guest-message replies
         "extra": {
             "rich_messages": False,     # Bot API 10.1 rich messages (tables/task lists/details/math) render natively; set True to opt in. Default stays legacy MarkdownV2 because rich messages can be hard to copy as plain text in Telegram clients.
             "rich_drafts": False,       # Experimental Bot API 10.1 rich draft previews during Telegram DM streaming. Default off because Telegram Desktop/macOS can visually overlay rich draft frames until the chat redraws.

hermes_state.py

@@ -1948,6 +1948,7 @@ def find_latest_gateway_session_for_peer(
         chat_id: Optional[str] = None,
         chat_type: Optional[str] = None,
         thread_id: Optional[str] = None,
+        exact_only: bool = False,
     ) -> Optional[Dict[str, Any]]:
         """Find the latest recoverable gateway session for a routing peer.
 
@@ -1978,6 +1979,8 @@ def find_latest_gateway_session_for_peer(
             ).fetchone()
             if row is not None:
                 return dict(row)
+            if exact_only:
+                return None
 
             # Conservative fallback for rows created by current code but with a
             # temporarily-missing exact key: still require the complete peer