diff --git a/gateway/platforms/telegram.py b/gateway/platforms/telegram.py index 02e6beb000f..0334fdca5f2 100644 --- a/gateway/platforms/telegram.py +++ b/gateway/platforms/telegram.py @@ -18,6 +18,10 @@ logger = logging.getLogger(__name__) try: from telegram import Update, Bot, Message, InlineKeyboardButton, InlineKeyboardMarkup + try: + from telegram import LinkPreviewOptions + except ImportError: + LinkPreviewOptions = None from telegram.ext import ( Application, CommandHandler, @@ -36,6 +40,7 @@ except ImportError: Message = Any InlineKeyboardButton = Any InlineKeyboardMarkup = Any + LinkPreviewOptions = None Application = Any CommandHandler = Any CallbackQueryHandler = Any @@ -137,6 +142,7 @@ class TelegramAdapter(BasePlatformAdapter): self._webhook_mode: bool = False self._mention_patterns = self._compile_mention_patterns() self._reply_to_mode: str = getattr(config, 'reply_to_mode', 'first') or 'first' + self._disable_link_previews: bool = self._coerce_bool_extra("disable_link_previews", False) # Buffer rapid/album photo updates so Telegram image bursts are handled # as a single MessageEvent instead of self-interrupting multiple turns. self._media_batch_delay_seconds = float(os.getenv("HERMES_TELEGRAM_MEDIA_BATCH_DELAY_SECONDS", "0.8")) @@ -202,6 +208,26 @@ class TelegramAdapter(BasePlatformAdapter): pass return isinstance(error, OSError) + def _coerce_bool_extra(self, key: str, default: bool = False) -> bool: + value = self.config.extra.get(key) if getattr(self.config, "extra", None) else None + if value is None: + return default + if isinstance(value, str): + lowered = value.strip().lower() + if lowered in ("true", "1", "yes", "on"): + return True + if lowered in ("false", "0", "no", "off"): + return False + return default + return bool(value) + + def _link_preview_kwargs(self) -> Dict[str, Any]: + if not self._disable_link_previews: + return {} + if LinkPreviewOptions is not None: + return {"link_preview_options": LinkPreviewOptions(is_disabled=True)} + return {"disable_web_page_preview": True} + async def _handle_polling_network_error(self, error: Exception) -> None: """Reconnect polling after a transient network interruption. @@ -856,6 +882,7 @@ class TelegramAdapter(BasePlatformAdapter): parse_mode=ParseMode.MARKDOWN_V2, reply_to_message_id=reply_to_id, message_thread_id=effective_thread_id, + **self._link_preview_kwargs(), ) except Exception as md_error: # Markdown parsing failed, try plain text @@ -868,6 +895,7 @@ class TelegramAdapter(BasePlatformAdapter): parse_mode=None, reply_to_message_id=reply_to_id, message_thread_id=effective_thread_id, + **self._link_preview_kwargs(), ) else: raise @@ -1055,6 +1083,7 @@ class TelegramAdapter(BasePlatformAdapter): text=text, parse_mode=ParseMode.MARKDOWN, reply_markup=keyboard, + **self._link_preview_kwargs(), ) return SendResult(success=True, message_id=str(msg.message_id)) except Exception as e: @@ -1114,6 +1143,7 @@ class TelegramAdapter(BasePlatformAdapter): "text": text, "parse_mode": ParseMode.MARKDOWN, "reply_markup": keyboard, + **self._link_preview_kwargs(), } if thread_id: kwargs["message_thread_id"] = int(thread_id) @@ -1184,6 +1214,7 @@ class TelegramAdapter(BasePlatformAdapter): parse_mode=ParseMode.MARKDOWN, reply_markup=keyboard, message_thread_id=int(thread_id) if thread_id else None, + **self._link_preview_kwargs(), ) # Store picker state keyed by chat_id diff --git a/tests/gateway/test_telegram_approval_buttons.py b/tests/gateway/test_telegram_approval_buttons.py index ec5bbd47ee1..93b5f82eef9 100644 --- a/tests/gateway/test_telegram_approval_buttons.py +++ b/tests/gateway/test_telegram_approval_buttons.py @@ -50,9 +50,9 @@ from gateway.platforms.telegram import TelegramAdapter from gateway.config import Platform, PlatformConfig -def _make_adapter(): +def _make_adapter(extra=None): """Create a TelegramAdapter with mocked internals.""" - config = PlatformConfig(enabled=True, token="test-token") + config = PlatformConfig(enabled=True, token="test-token", extra=extra or {}) adapter = TelegramAdapter(config) adapter._bot = AsyncMock() adapter._app = MagicMock() @@ -134,6 +134,23 @@ class TestTelegramExecApproval: ) assert result.success is False + @pytest.mark.asyncio + async def test_disable_link_previews_sets_preview_kwargs(self): + adapter = _make_adapter(extra={"disable_link_previews": True}) + mock_msg = MagicMock() + mock_msg.message_id = 42 + adapter._bot.send_message = AsyncMock(return_value=mock_msg) + + await adapter.send_exec_approval( + chat_id="12345", command="ls", session_key="s" + ) + + kwargs = adapter._bot.send_message.call_args[1] + assert ( + kwargs.get("disable_web_page_preview") is True + or kwargs.get("link_preview_options") is not None + ) + @pytest.mark.asyncio async def test_truncates_long_command(self): adapter = _make_adapter()