diff --git a/cli-config.yaml.example b/cli-config.yaml.example index f058705cfe21..bec44ad58e0c 100644 --- a/cli-config.yaml.example +++ b/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 diff --git a/gateway/config.py b/gateway/config.py index d6f84b2405ba..13088507898a 100644 --- a/gateway/config.py +++ b/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: diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py index 1025964dc43b..25a4a7bc827d 100644 --- a/gateway/platforms/base.py +++ b/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) diff --git a/gateway/run.py b/gateway/run.py index 8256c283d4cf..d75bd72d87db 100644 --- a/gateway/run.py +++ b/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 "?", diff --git a/gateway/session.py b/gateway/session.py index 9d20c204a83b..c786d706c67c 100644 --- a/gateway/session.py +++ b/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,6 +895,12 @@ 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: @@ -896,8 +908,12 @@ def build_session_key( 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,6 +1332,11 @@ 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, @@ -1313,6 +1344,7 @@ def _recover_session_from_db( 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) diff --git a/hermes_cli/config.py b/hermes_cli/config.py index 81d66e99eba0..0c9bdae5d3f9 100644 --- a/hermes_cli/config.py +++ b/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. diff --git a/hermes_state.py b/hermes_state.py index a2895b09c7a2..54a5954c33d0 100644 --- a/hermes_state.py +++ b/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 diff --git a/plugins/platforms/telegram/adapter.py b/plugins/platforms/telegram/adapter.py index 40f1ab094238..f2f982f77c0d 100644 --- a/plugins/platforms/telegram/adapter.py +++ b/plugins/platforms/telegram/adapter.py @@ -16,6 +16,7 @@ import html as _html import re import threading +import uuid from datetime import datetime, timezone from typing import Dict, List, Optional, Set, Any @@ -166,6 +167,10 @@ async def _shutdown_abandoned_app(app) -> None: ContextTypes, filters, ) + try: + from telegram.ext import TypeHandler + except ImportError: + TypeHandler = None from telegram.constants import ParseMode, ChatType from telegram.request import HTTPXRequest TELEGRAM_AVAILABLE = True @@ -181,6 +186,7 @@ async def _shutdown_abandoned_app(app) -> None: CommandHandler = Any CallbackQueryHandler = Any TelegramMessageHandler = Any + TypeHandler = None HTTPXRequest = Any filters = None ParseMode = None @@ -253,7 +259,7 @@ def check_telegram_requirements() -> bool: global TELEGRAM_AVAILABLE, Update, Bot, Message, InlineKeyboardButton global InlineKeyboardMarkup, LinkPreviewOptions, Application global CommandHandler, CallbackQueryHandler, TelegramMessageHandler - global ContextTypes, filters, ParseMode, ChatType, HTTPXRequest + global TypeHandler, ContextTypes, filters, ParseMode, ChatType, HTTPXRequest if TELEGRAM_AVAILABLE: return True try: @@ -274,6 +280,10 @@ def check_telegram_requirements() -> bool: MessageHandler as _MH, ContextTypes as _CT, filters as _filters, ) + try: + from telegram.ext import TypeHandler as _TH + except ImportError: + _TH = None from telegram.constants import ParseMode as _PM, ChatType as _CtT from telegram.request import HTTPXRequest as _HR except ImportError: @@ -288,6 +298,7 @@ def check_telegram_requirements() -> bool: CommandHandler = _CH CallbackQueryHandler = _CQH TelegramMessageHandler = _MH + TypeHandler = _TH ContextTypes = _CT filters = _filters ParseMode = _PM @@ -674,6 +685,164 @@ def _should_drop_delayed_delivery(self) -> bool: """ return bool(getattr(self, "_drop_delayed_deliveries", False)) + @staticmethod + def _telegram_allowed_updates() -> list[str]: + """Return PTB's known update types plus Bot API 10.0 guest messages.""" + try: + allowed = list(Update.ALL_TYPES) + except Exception: + allowed = [] + if "guest_message" not in allowed: + allowed.append("guest_message") + return allowed + + @staticmethod + def _metadata_is_guest(metadata: Optional[Dict[str, Any]]) -> bool: + return bool( + metadata + and ( + metadata.get("telegram_guest_query_id") + or metadata.get("telegram_guest_inline_message_id") + ) + ) + + @staticmethod + def _guest_inline_message_id(metadata: Optional[Dict[str, Any]]) -> Optional[str]: + if not metadata: + return None + inline_id = metadata.get("telegram_guest_inline_message_id") + return str(inline_id) if inline_id else None + + @staticmethod + def _guest_query_id(metadata: Optional[Dict[str, Any]]) -> Optional[str]: + if not metadata: + return None + query_id = metadata.get("telegram_guest_query_id") + return str(query_id) if query_id else None + + def _guest_single_message_content(self, content: str) -> str: + """Clamp Bot API guest replies to one editable Telegram message.""" + if utf16_len(content) <= self.MAX_MESSAGE_LENGTH: + return content + note = "\n\n[Response truncated: Telegram guest mode allows only one message.]" + budget = self.MAX_MESSAGE_LENGTH - utf16_len(note) + if budget <= 0: + return note[-self.MAX_MESSAGE_LENGTH:] + from gateway.platforms.base import _prefix_within_utf16_limit + return _prefix_within_utf16_limit(content, budget).rstrip() + note + + def _guest_inline_result(self, text: str, *, use_markdown: bool = False) -> Dict[str, Any]: + message_text = self._guest_single_message_content(text) + content: Dict[str, Any] = {"message_text": message_text} + if use_markdown: + content = { + "message_text": self._guest_single_message_content(self.format_message(message_text)), + "parse_mode": ParseMode.MARKDOWN_V2, + } + return { + "type": "article", + "id": uuid.uuid4().hex, + "title": "Hermes response", + "input_message_content": content, + } + + def _guest_thinking_text(self) -> str: + text = ( + self.config.extra.get("guest_thinking_text") + or os.getenv("TELEGRAM_GUEST_THINKING_TEXT") + or "💭 Thinking..." + ) + return str(text) + + async def _answer_guest_query( + self, + guest_query_id: str, + text: str = "Thinking...", + *, + use_markdown: bool = False, + ) -> SendResult: + """Answer a Bot API 10.0 guest query and return its inline message id.""" + if not self._bot: + return SendResult(success=False, error="Not connected") + result = self._guest_inline_result(text, use_markdown=use_markdown) + + async def _call_answer(result_payload: Dict[str, Any]) -> Any: + answer = getattr(self._bot, "answer_guest_query", None) or getattr( + self._bot, "answerGuestQuery", None + ) + if callable(answer): + return await answer(guest_query_id=guest_query_id, result=result_payload) + post = getattr(self._bot, "_post", None) + if not callable(post): + raise RuntimeError("python-telegram-bot does not expose answerGuestQuery") + return await post( + "answerGuestQuery", + {"guest_query_id": guest_query_id, "result": result_payload}, + ) + + try: + try: + raw = await _call_answer(result) + except Exception as fmt_err: + if not ( + use_markdown + and ( + "parse" in str(fmt_err).lower() + or "markdown" in str(fmt_err).lower() + ) + ): + raise + raw = await _call_answer(self._guest_inline_result(text, use_markdown=False)) + + inline_id = raw.get("inline_message_id") if isinstance(raw, dict) else getattr(raw, "inline_message_id", None) + if not inline_id: + return SendResult(success=False, error="answerGuestQuery returned no inline_message_id") + return SendResult(success=True, message_id=str(inline_id), raw_response=raw) + except Exception as exc: + logger.error("[%s] Failed to answer Telegram guest query: %s", self.name, exc, exc_info=True) + return SendResult(success=False, error=str(exc)) + + async def _edit_guest_inline_message( + self, + inline_message_id: str, + content: str, + *, + finalize: bool = False, + ) -> SendResult: + """Edit the single inline message created by answerGuestQuery.""" + if not self._bot: + return SendResult(success=False, error="Not connected") + content = self._guest_single_message_content(content) + try: + formatted = self._guest_single_message_content(self.format_message(content)) + try: + await self._bot.edit_message_text( + inline_message_id=inline_message_id, + text=formatted, + parse_mode=ParseMode.MARKDOWN_V2, + ) + except Exception as fmt_err: + if "not modified" in str(fmt_err).lower(): + return SendResult(success=True, message_id=inline_message_id) + await self._bot.edit_message_text( + inline_message_id=inline_message_id, + text=content, + ) + return SendResult(success=True, message_id=inline_message_id) + except Exception as exc: + err = str(exc).lower() + if "not modified" in err: + return SendResult(success=True, message_id=inline_message_id) + retryable = any( + marker in err + for marker in ( + "connecterror", "connect error", "connection error", "networkerror", + "network error", "timed out", "temporarily unavailable", "httpx", + ) + ) + logger.error("[%s] Failed to edit Telegram guest inline message %s: %s", self.name, inline_message_id, exc, exc_info=True) + return SendResult(success=False, error=str(exc), retryable=retryable) + def _notification_kwargs( self, metadata: Optional[Dict[str, Any]] ) -> Dict[str, Any]: @@ -1942,7 +2111,7 @@ async def _start_polling_resilient(self, *, drop_pending_updates: bool, error_ca raise RuntimeError("Telegram application/updater not initialized") try: await self._app.updater.start_polling( - allowed_updates=Update.ALL_TYPES, + allowed_updates=self._telegram_allowed_updates(), drop_pending_updates=drop_pending_updates, error_callback=error_callback, ) @@ -2041,7 +2210,7 @@ async def _handle_polling_network_error(self, error: Exception) -> None: if not app: raise RuntimeError("Telegram application was torn down during reconnect") await app.updater.start_polling( - allowed_updates=Update.ALL_TYPES, + allowed_updates=self._telegram_allowed_updates(), drop_pending_updates=False, error_callback=self._polling_error_callback_ref, ) @@ -2436,7 +2605,7 @@ async def _handle_polling_conflict(self, error: Exception) -> None: if not app: raise RuntimeError("Telegram application was torn down during conflict reconnect") await app.updater.start_polling( - allowed_updates=Update.ALL_TYPES, + allowed_updates=self._telegram_allowed_updates(), drop_pending_updates=False, error_callback=self._polling_error_callback_ref, ) @@ -3095,6 +3264,8 @@ def _with_limits(httpx_kwargs: Optional[dict] = None) -> dict: self._bot = self._app.bot # Register handlers + if TypeHandler is not None: + self._app.add_handler(TypeHandler(Update, self._handle_guest_update), group=1) self._app.add_handler(TelegramMessageHandler( filters.TEXT & ~filters.COMMAND, self._handle_text_message @@ -3206,7 +3377,7 @@ def _with_limits(httpx_kwargs: Optional[dict] = None) -> dict: url_path=webhook_path, webhook_url=webhook_url, secret_token=webhook_secret, - allowed_updates=Update.ALL_TYPES, + allowed_updates=self._telegram_allowed_updates(), # Webhooks are push-based — Telegram does not hold a # server-side getUpdates queue, so this flag is a no-op # in practice. Mirror the polling path's reconnect @@ -3486,6 +3657,21 @@ async def send( # Skip whitespace-only text to prevent Telegram 400 empty-text errors. if not content or not content.strip(): return SendResult(success=True, message_id=None) + + if self._metadata_is_guest(metadata): + inline_id = self._guest_inline_message_id(metadata) + if inline_id: + return await self._edit_guest_inline_message(inline_id, content) + query_id = self._guest_query_id(metadata) + if query_id: + result = await self._answer_guest_query( + query_id, + content, + use_markdown=True, + ) + if result.success and result.message_id and metadata is not None: + metadata["telegram_guest_inline_message_id"] = result.message_id + return result try: # Bot API 10.1 rich fast-path: send the raw agent markdown via @@ -3851,6 +4037,14 @@ async def edit_message( if not self._bot: return SendResult(success=False, error="Not connected") + if self._metadata_is_guest(metadata): + inline_id = self._guest_inline_message_id(metadata) or str(message_id) + return await self._edit_guest_inline_message( + inline_id, + content, + finalize=finalize, + ) + # Rich finalize (Bot API 10.1): when the completed content has # constructs the legacy MarkdownV2 edit degrades (tables → bullet # lists, task lists,
, block math) and rich is available, @@ -7350,6 +7544,131 @@ def _effective_update_message(self, update: Update) -> Optional[Message]: """ return getattr(update, "effective_message", None) or getattr(update, "message", None) + @staticmethod + def _obj_field(obj: Any, name: str, default: Any = None) -> Any: + value = getattr(obj, name, default) + if value is not default: + return value + api_kwargs = getattr(obj, "api_kwargs", None) + if isinstance(api_kwargs, dict): + return api_kwargs.get(name, default) + if isinstance(obj, dict): + return obj.get(name, default) + return default + + def _extract_guest_message(self, update: Update) -> Optional[Message]: + guest = getattr(update, "guest_message", None) + if guest is not None: + return guest + api_kwargs = getattr(update, "api_kwargs", None) + raw = api_kwargs.get("guest_message") if isinstance(api_kwargs, dict) else None + if raw is None: + return None + try: + de_json = getattr(Message, "de_json", None) + if callable(de_json): + return de_json(raw, self._bot) + except Exception: + logger.debug("[%s] Failed to parse raw Telegram guest_message", self.name, exc_info=True) + return raw + + def _guest_caller_user(self, message: Message) -> Any: + caller = self._obj_field(message, "guest_bot_caller_user") + if caller is None: + caller = getattr(message, "from_user", None) + return caller + + def _guest_query_id_from_message(self, message: Message) -> Optional[str]: + query_id = self._obj_field(message, "guest_query_id") + return str(query_id).strip() if query_id else None + + def _guest_user_id(self, user: Any) -> Optional[str]: + user_id = self._obj_field(user, "id") + return str(user_id).strip() if user_id is not None else None + + def _guest_user_name(self, user: Any) -> Optional[str]: + full_name = self._obj_field(user, "full_name") + if full_name: + return str(full_name) + first_name = self._obj_field(user, "first_name") + last_name = self._obj_field(user, "last_name") + name = " ".join(str(part) for part in (first_name, last_name) if part) + if name: + return name + username = self._obj_field(user, "username") + return str(username) if username else None + + def _telegram_guest_allowed_users(self) -> set[str]: + """Return user allowlist entries that may invoke Bot API guest mode.""" + raw_values = [ + self.config.extra.get("allow_from"), + self.config.extra.get("group_allow_from"), + os.getenv("TELEGRAM_ALLOWED_USERS", ""), + os.getenv("TELEGRAM_GROUP_ALLOWED_USERS", ""), + os.getenv("GATEWAY_ALLOWED_USERS", ""), + ] + allowed: set[str] = set() + for raw in raw_values: + if raw is None: + continue + values = raw if isinstance(raw, list) else str(raw).split(",") + allowed.update(str(value).strip() for value in values if str(value).strip()) + return allowed + + def _is_guest_caller_authorized(self, caller_id: str) -> bool: + allowed = self._telegram_guest_allowed_users() + if not allowed: + return False + if "*" in allowed: + return True + check_ids = {caller_id} + if "@" in caller_id: + check_ids.add(caller_id.split("@", 1)[0]) + return bool(check_ids & allowed) + + async def _handle_guest_update(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: + """Handle Telegram Bot API 10.0 guest messages.""" + msg = self._extract_guest_message(update) + if not msg: + return + text = getattr(msg, "text", None) or getattr(msg, "caption", None) or "" + if not text: + return + query_id = self._guest_query_id_from_message(msg) + if not query_id: + logger.warning("[%s] Dropping Telegram guest_message without guest_query_id", self.name) + return + + caller = self._guest_caller_user(msg) + caller_id = self._guest_user_id(caller) + if not caller_id: + logger.warning("[%s] Dropping Telegram guest_message without caller user id", self.name) + return + + caller_name = self._guest_user_name(caller) + if not self._is_guest_caller_authorized(caller_id): + logger.warning("[%s] Unauthorized Telegram guest_message caller: %s (%s)", self.name, caller_id, caller_name) + return + + event = self._build_message_event(msg, MessageType.TEXT, update_id=getattr(update, "update_id", None)) + event.text = self._clean_bot_trigger_text(text) + event.source.user_id = caller_id + event.source.user_name = caller_name + # Guest query/update/message identifiers can remain stable across + # multiple turns in the same Telegram guest conversation. Give every + # guest turn its own Hermes session so transcript context never leaks + # between guest invocations. + turn_nonce = uuid.uuid4().hex[:12] + event.source.platform_metadata = { + "telegram_guest_query_id": query_id, + "session_key_suffix": f"telegram_guest_{query_id}_{turn_nonce}", + } + ack = await self._answer_guest_query(query_id, self._guest_thinking_text()) + if not ack.success: + return + event.source.platform_metadata["telegram_guest_inline_message_id"] = ack.message_id + await self.handle_message(event) + async def _handle_text_message(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Handle incoming text messages. @@ -7357,6 +7676,8 @@ async def _handle_text_message(self, update: Update, context: ContextTypes.DEFAU rapid successive text messages from the same user/chat and aggregate them into a single MessageEvent before dispatching. """ + if self._extract_guest_message(update) is not None: + return msg = self._effective_update_message(update) if not msg or not msg.text: return @@ -7385,6 +7706,8 @@ async def _handle_text_message(self, update: Update, context: ContextTypes.DEFAU async def _handle_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Handle incoming command messages.""" + if self._extract_guest_message(update) is not None: + return msg = self._effective_update_message(update) if not msg or not msg.text: return @@ -7407,6 +7730,8 @@ async def _handle_command(self, update: Update, context: ContextTypes.DEFAULT_TY async def _handle_location_message(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Handle incoming location/venue pin messages.""" + if self._extract_guest_message(update) is not None: + return msg = self._effective_update_message(update) if not msg: return diff --git a/pyproject.toml b/pyproject.toml index 963210075e0a..0e8bafca199a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -158,7 +158,7 @@ modal = ["modal==1.3.4"] daytona = ["daytona==0.155.0"] hindsight = ["hindsight-client==0.6.1"] dev = ["debugpy==1.8.20", "pytest==9.0.2", "pytest-asyncio==1.3.0", "mcp==1.26.0", "starlette==1.0.1", "ty==0.0.21", "ruff==0.15.10", "setuptools==81.0.0"] # starlette: CVE-2026-48710; setuptools: latest <82 (torch >=2.11 caps setuptools<82) -messaging = ["python-telegram-bot[webhooks]==22.6", "discord.py[voice]==2.7.1", "aiohttp==3.14.1", "brotlicffi==1.2.0.1", "slack-bolt==1.27.0", "slack-sdk==3.40.1", "qrcode==7.4.2"] # aiohttp 3.14.1: CVE-2026-34513/34518/34519/34520/34525 + 34993(RCE)/47265 +messaging = ["python-telegram-bot[webhooks]==22.8", "discord.py[voice]==2.7.1", "aiohttp==3.14.1", "brotlicffi==1.2.0.1", "slack-bolt==1.27.0", "slack-sdk==3.40.1", "qrcode==7.4.2"] # aiohttp 3.14.1: CVE-2026-34513/34518/34519/34520/34525 + 34993(RCE)/47265 cron = [] # croniter is now a core dependency; this extra kept for back-compat slack = ["slack-bolt==1.27.0", "slack-sdk==3.40.1", "aiohttp==3.14.1"] matrix = ["mautrix[encryption]==0.21.0", "aiosqlite==0.22.1", "asyncpg==0.31.0", "aiohttp-socks==0.11.0", "aiohttp==3.14.1"] # aiohttp 3.14.1: CVE-2026-34993(RCE)/47265 + 34513/34518/34519/34520/34525 (mautrix/aiohttp-socks only cap aiohttp<4 / >=3.10, so pin the patched floor directly) @@ -229,7 +229,7 @@ vertex = ["google-auth==2.55.1"] azure-identity = ["azure-identity==1.25.3"] termux = [ # Baseline Android / Termux path for reliable fresh installs. - "python-telegram-bot[webhooks]==22.6", + "python-telegram-bot[webhooks]==22.8", "hermes-agent[cron]", "hermes-agent[cli]", "hermes-agent[pty]", diff --git a/tests/gateway/test_session_store_runtime_stale_guard.py b/tests/gateway/test_session_store_runtime_stale_guard.py index 57f8c624bf25..08f50c77c492 100644 --- a/tests/gateway/test_session_store_runtime_stale_guard.py +++ b/tests/gateway/test_session_store_runtime_stale_guard.py @@ -202,3 +202,20 @@ def test_force_new_skips_stale_check(self, tmp_path): assert result.session_id != "sid_old" db.get_session.assert_not_called() + + def test_suffix_scoped_source_disables_peer_fallback_recovery(self, tmp_path): + source = SessionSource( + platform=Platform.TELEGRAM, + chat_id="-100", + chat_type="group", + user_id="123", + platform_metadata={"session_key_suffix": "telegram_guest_q1_turn2"}, + ) + db = _db_returning({}) + db.find_latest_gateway_session_for_peer.return_value = None + store = _make_store_with_db(tmp_path, db) + + store.get_or_create_session(source) + + db.find_latest_gateway_session_for_peer.assert_called_once() + assert db.find_latest_gateway_session_for_peer.call_args.kwargs["exact_only"] is True diff --git a/tests/gateway/test_telegram_format.py b/tests/gateway/test_telegram_format.py index 18dec441eced..5056737307a5 100644 --- a/tests/gateway/test_telegram_format.py +++ b/tests/gateway/test_telegram_format.py @@ -12,7 +12,7 @@ import pytest -from gateway.config import PlatformConfig +from gateway.config import Platform, PlatformConfig # --------------------------------------------------------------------------- @@ -1120,9 +1120,13 @@ def _guest_test_adapter(*, guest_mode=True, require_mention=True, allowed_chats= "guest_mode": guest_mode, "require_mention": require_mention, "allowed_chats": allowed_chats or ["-100200"], + # Keep these unit tests isolated from TELEGRAM_ALLOWED_TOPICS set by + # config-bridge tests in the same process. + "allowed_topics": [], }, ) adapter = object.__new__(TelegramAdapter) + adapter.platform = Platform.TELEGRAM adapter.config = config adapter._bot = SimpleNamespace(id=999, username="hermes_bot") adapter._mention_patterns = adapter._compile_mention_patterns() diff --git a/tests/gateway/test_telegram_guest_botapi10.py b/tests/gateway/test_telegram_guest_botapi10.py new file mode 100644 index 000000000000..6de71adb9225 --- /dev/null +++ b/tests/gateway/test_telegram_guest_botapi10.py @@ -0,0 +1,191 @@ +import pytest +from types import SimpleNamespace +from unittest.mock import AsyncMock + +from gateway.config import Platform, PlatformConfig +from gateway.platforms.base import MessageEvent, MessageType, SendResult +from gateway.run import _should_suppress_final_send_after_stream +from gateway.session import SessionSource, build_session_key +from plugins.platforms.telegram.adapter import TelegramAdapter + + +def _adapter(*, allow_from=None, guest_thinking_text=None): + adapter = object.__new__(TelegramAdapter) + adapter.platform = Platform.TELEGRAM + extra = {"allow_from": allow_from or []} + if guest_thinking_text is not None: + extra["guest_thinking_text"] = guest_thinking_text + adapter.config = PlatformConfig(enabled=True, token="token", extra=extra) + adapter._bot = SimpleNamespace( + answer_guest_query=AsyncMock(return_value=SimpleNamespace(inline_message_id="inline-1")), + edit_message_text=AsyncMock(), + ) + adapter._message_handler = AsyncMock() + return adapter + + +def _guest_message(*, text="hello", query_id="q1", caller_id=123, caller_name="Guest User"): + caller = SimpleNamespace(id=caller_id, full_name=caller_name, first_name="Guest", last_name="User") + return SimpleNamespace( + text=text, + caption=None, + guest_query_id=query_id, + guest_bot_caller_user=caller, + from_user=caller, + chat=SimpleNamespace(id=-100, title="source chat", type="group"), + message_id=77, + date=None, + reply_to_message=None, + ) + + +def _event_for(msg): + return MessageEvent( + text=msg.text, + message_type=MessageType.TEXT, + source=SessionSource( + platform=Platform.TELEGRAM, + chat_id=str(msg.chat.id), + chat_type="group", + user_id=str(msg.from_user.id), + user_name=msg.from_user.full_name, + message_id=str(msg.message_id), + ), + raw_message=msg, + message_id=str(msg.message_id), + ) + + +def test_guest_allowed_updates_includes_guest_message(): + assert "guest_message" in TelegramAdapter._telegram_allowed_updates() + + +@pytest.mark.asyncio +async def test_authorized_guest_message_creates_event_with_metadata(monkeypatch): + adapter = _adapter(allow_from=["123"], guest_thinking_text="working") + msg = _guest_message() + update = SimpleNamespace(update_id=42, guest_message=msg) + built = _event_for(msg) + adapter._build_message_event = lambda *args, **kwargs: built + adapter._clean_bot_trigger_text = lambda text: text + adapter.handle_message = AsyncMock() + + await adapter._handle_guest_update(update, SimpleNamespace()) + + adapter._bot.answer_guest_query.assert_awaited_once() + kwargs = adapter._bot.answer_guest_query.await_args.kwargs + assert kwargs["guest_query_id"] == "q1" + assert kwargs["result"]["input_message_content"]["message_text"] == "working" + adapter.handle_message.assert_awaited_once_with(built) + assert built.source.user_id == "123" + assert built.source.user_name == "Guest User" + assert built.source.platform_metadata["telegram_guest_query_id"] == "q1" + assert built.source.platform_metadata["telegram_guest_inline_message_id"] == "inline-1" + assert built.source.platform_metadata["session_key_suffix"].startswith("telegram_guest_q1_") + + +@pytest.mark.asyncio +async def test_guest_messages_with_same_query_id_get_distinct_session_suffixes(): + adapter = _adapter(allow_from=["123"]) + suffixes = [] + + for text in ("first", "second"): + msg = _guest_message(text=text, query_id="q1") + built = _event_for(msg) + adapter._build_message_event = lambda *args, built=built, **kwargs: built + adapter._clean_bot_trigger_text = lambda text: text + adapter.handle_message = AsyncMock() + + await adapter._handle_guest_update(SimpleNamespace(update_id=42, guest_message=msg), SimpleNamespace()) + suffixes.append(built.source.platform_metadata["session_key_suffix"]) + + assert suffixes[0] != suffixes[1] + + +@pytest.mark.asyncio +async def test_guest_update_without_query_id_is_dropped(): + adapter = _adapter(allow_from=["123"]) + msg = _guest_message(query_id=None) + adapter._build_message_event = lambda *args, **kwargs: _event_for(msg) + adapter.handle_message = AsyncMock() + + await adapter._handle_guest_update(SimpleNamespace(update_id=1, guest_message=msg), SimpleNamespace()) + + adapter._bot.answer_guest_query.assert_not_called() + adapter.handle_message.assert_not_called() + + +@pytest.mark.asyncio +async def test_unauthorized_guest_caller_is_dropped(): + adapter = _adapter(allow_from=["999"]) + msg = _guest_message(caller_id=123) + adapter._build_message_event = lambda *args, **kwargs: _event_for(msg) + adapter.handle_message = AsyncMock() + + await adapter._handle_guest_update(SimpleNamespace(update_id=1, guest_message=msg), SimpleNamespace()) + + adapter._bot.answer_guest_query.assert_not_called() + adapter.handle_message.assert_not_called() + + +@pytest.mark.asyncio +async def test_first_guest_send_answers_query_and_mutates_inline_metadata(): + adapter = _adapter(allow_from=["123"]) + metadata = {"telegram_guest_query_id": "q1"} + + result = await adapter.send("-100", "final **answer**", metadata=metadata) + + assert result.success + assert result.message_id == "inline-1" + assert metadata["telegram_guest_inline_message_id"] == "inline-1" + adapter._bot.answer_guest_query.assert_awaited_once() + adapter._bot.edit_message_text.assert_not_called() + + +@pytest.mark.asyncio +async def test_subsequent_guest_send_edits_inline_message_not_normal_send(): + adapter = _adapter(allow_from=["123"]) + metadata = {"telegram_guest_query_id": "q1", "telegram_guest_inline_message_id": "inline-1"} + + result = await adapter.send("-100", "updated", metadata=metadata) + + assert result.success + adapter._bot.answer_guest_query.assert_not_called() + adapter._bot.edit_message_text.assert_awaited_once() + assert adapter._bot.edit_message_text.await_args.kwargs["inline_message_id"] == "inline-1" + + +@pytest.mark.asyncio +async def test_guest_edit_uses_inline_message_id(): + adapter = _adapter(allow_from=["123"]) + metadata = {"telegram_guest_inline_message_id": "inline-1"} + + result = await adapter.edit_message("-100", "ignored-normal-id", "edited", finalize=True, metadata=metadata) + + assert result.success + adapter._bot.edit_message_text.assert_awaited_once() + assert adapter._bot.edit_message_text.await_args.kwargs["inline_message_id"] == "inline-1" + + +def test_guest_session_key_suffix_is_included(): + source = SessionSource( + platform=Platform.TELEGRAM, + chat_id="-100", + chat_type="group", + user_id="123", + platform_metadata={"session_key_suffix": "telegram_guest:q1:42"}, + ) + + assert build_session_key(source).endswith(":123:telegram_guest_q1_42") + + +def test_guest_final_send_suppression_follows_stream_delivery_state(): + stream_consumer = SimpleNamespace(final_response_sent=True, final_content_delivered=False) + + assert _should_suppress_final_send_after_stream( + response_text="done", + response_previewed=False, + response_transformed=False, + stream_consumer=stream_consumer, + metadata={"telegram_guest_query_id": "q1"}, + ) diff --git a/tests/test_hermes_state.py b/tests/test_hermes_state.py index 707272f10045..38a5a20e6be6 100644 --- a/tests/test_hermes_state.py +++ b/tests/test_hermes_state.py @@ -4997,6 +4997,15 @@ def test_gateway_session_peer_round_trip_and_recovery(db): ) assert recovered["id"] == "gw-session" + assert db.find_latest_gateway_session_for_peer( + source="telegram", + user_id="user-1", + session_key="agent:main:telegram:dm:other-key", + chat_id="chat-1", + chat_type="dm", + exact_only=True, + ) is None + def test_gateway_session_recovery_reopens_legacy_agent_close_rows(db): db.create_session( diff --git a/uv.lock b/uv.lock index b51daeca10d3..41da3bd4c6e8 100644 --- a/uv.lock +++ b/uv.lock @@ -1821,8 +1821,8 @@ requires-dist = [ { name = "python-dotenv", specifier = "==1.2.2" }, { name = "python-multipart", specifier = ">=0.0.9,<1" }, { name = "python-multipart", marker = "extra == 'web'", specifier = "==0.0.27" }, - { name = "python-telegram-bot", extras = ["webhooks"], marker = "extra == 'messaging'", specifier = "==22.6" }, - { name = "python-telegram-bot", extras = ["webhooks"], marker = "extra == 'termux'", specifier = "==22.6" }, + { name = "python-telegram-bot", extras = ["webhooks"], marker = "extra == 'messaging'", specifier = "==22.8" }, + { name = "python-telegram-bot", extras = ["webhooks"], marker = "extra == 'termux'", specifier = "==22.8" }, { name = "pywinpty", marker = "sys_platform == 'win32'", specifier = ">=2.0.0,<3" }, { name = "pyyaml", specifier = "==6.0.3" }, { name = "qrcode", marker = "extra == 'dingtalk'", specifier = "==7.4.2" }, @@ -3497,14 +3497,14 @@ wheels = [ [[package]] name = "python-telegram-bot" -version = "22.6" +version = "22.8" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "httpx" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/cd/9b/8df90c85404166a6631e857027866263adb27440d8af1dbeffbdc4f0166c/python_telegram_bot-22.6.tar.gz", hash = "sha256:50ae8cc10f8dff01445628687951020721f37956966b92a91df4c1bf2d113742", size = 1503761, upload-time = "2026-01-24T13:57:00.269Z" } +sdist = { url = "https://files.pythonhosted.org/packages/ba/77/153517bb1ac1bba670c6fb1dbf09e1fd0730494b1705934e715391413a0d/python_telegram_bot-22.8.tar.gz", hash = "sha256:f9d3847fcb23ee603477e442800b33bb4adf851a73e0619d2050be879decf1ef", size = 1551700, upload-time = "2026-06-12T08:10:29.1Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/13/97/7298f0e1afe3a1ae52ff4c5af5087ed4de319ea73eb3b5c8c4dd4e76e708/python_telegram_bot-22.6-py3-none-any.whl", hash = "sha256:e598fe171c3dde2dfd0f001619ee9110eece66761a677b34719fb18934935ce0", size = 737267, upload-time = "2026-01-24T13:56:58.06Z" }, + { url = "https://files.pythonhosted.org/packages/60/7c/ed7d4dd94280bd434173cae9f7a7aedaaab9af128ae4f494423a5687c820/python_telegram_bot-22.8-py3-none-any.whl", hash = "sha256:42373918097f1b837cc4e717d588c19ea79651497ec712bb5b0c76e5e63c50e1", size = 769397, upload-time = "2026-06-12T08:10:27.066Z" }, ] [package.optional-dependencies]