feat(telegram): add group mention gating and regex triggers (#3870)

Adds Discord-style mention gating for Telegram groups:
- telegram.require_mention: gate group messages (default: false)
- telegram.mention_patterns: regex wake-word triggers
- telegram.free_response_chats: bypass gating for specific chats

When require_mention is enabled, group messages are accepted only for:
- slash commands
- replies to the bot
- @botusername mentions
- regex wake-word pattern matches

DMs remain unrestricted. @mention text is stripped before passing to
the agent. Invalid regex patterns are ignored with a warning.

Config bridges follow the existing Discord pattern (yaml → env vars).

Cherry-picked and adapted from PR #1977 by mcleay. Fixed ChatType
comparison to work without python-telegram-bot installed (uses string
matching instead of enum, consistent with other entity_type checks).

Co-authored-by: mcleay <mcleay@users.noreply.github.com>

gateway/config.py

@@ -515,6 +515,10 @@ def load_gateway_config() -> GatewayConfig:
                     )
                 if "reply_prefix" in platform_cfg:
                     bridged["reply_prefix"] = platform_cfg["reply_prefix"]
+                if "require_mention" in platform_cfg:
+                    bridged["require_mention"] = platform_cfg["require_mention"]
+                if "mention_patterns" in platform_cfg:
+                    bridged["mention_patterns"] = platform_cfg["mention_patterns"]
                 if not bridged:
                     continue
                 plat_data = platforms_data.setdefault(plat.value, {})
@@ -539,6 +543,20 @@ def load_gateway_config() -> GatewayConfig:
                     os.environ["DISCORD_FREE_RESPONSE_CHANNELS"] = str(frc)
                 if "auto_thread" in discord_cfg and not os.getenv("DISCORD_AUTO_THREAD"):
                     os.environ["DISCORD_AUTO_THREAD"] = str(discord_cfg["auto_thread"]).lower()
+
+            # Telegram settings → env vars (env vars take precedence)
+            telegram_cfg = yaml_cfg.get("telegram", {})
+            if isinstance(telegram_cfg, dict):
+                if "require_mention" in telegram_cfg and not os.getenv("TELEGRAM_REQUIRE_MENTION"):
+                    os.environ["TELEGRAM_REQUIRE_MENTION"] = str(telegram_cfg["require_mention"]).lower()
+                if "mention_patterns" in telegram_cfg and not os.getenv("TELEGRAM_MENTION_PATTERNS"):
+                    import json as _json
+                    os.environ["TELEGRAM_MENTION_PATTERNS"] = _json.dumps(telegram_cfg["mention_patterns"])
+                frc = telegram_cfg.get("free_response_chats")
+                if frc is not None and not os.getenv("TELEGRAM_FREE_RESPONSE_CHATS"):
+                    if isinstance(frc, list):
+                        frc = ",".join(str(v) for v in frc)
+                    os.environ["TELEGRAM_FREE_RESPONSE_CHATS"] = str(frc)
     except Exception as e:
         logger.warning(
             "Failed to process config.yaml — falling back to .env / gateway.json values. "

gateway/platforms/telegram.py

@@ -8,6 +8,7 @@ Uses python-telegram-bot library for:
 """
 
 import asyncio
+import json
 import logging
 import os
 import re
@@ -122,6 +123,7 @@ class TelegramAdapter(BasePlatformAdapter):
         super().__init__(config, Platform.TELEGRAM)
         self._app: Optional[Application] = None
         self._bot: Optional[Bot] = None
+        self._mention_patterns = self._compile_mention_patterns()
         self._reply_to_mode: str = getattr(config, 'reply_to_mode', 'first') or 'first'
         # Buffer rapid/album photo updates so Telegram image bursts are handled
         # as a single MessageEvent instead of self-interrupting multiple turns.
@@ -1325,6 +1327,148 @@ class TelegramAdapter(BasePlatformAdapter):
 
         return text
     
+    # ── Group mention gating ──────────────────────────────────────────────
+
+    def _telegram_require_mention(self) -> bool:
+        """Return whether group chats should require an explicit bot trigger."""
+        configured = self.config.extra.get("require_mention")
+        if configured is not None:
+            if isinstance(configured, str):
+                return configured.lower() in ("true", "1", "yes", "on")
+            return bool(configured)
+        return os.getenv("TELEGRAM_REQUIRE_MENTION", "false").lower() in ("true", "1", "yes", "on")
+
+    def _telegram_free_response_chats(self) -> set[str]:
+        raw = self.config.extra.get("free_response_chats")
+        if raw is None:
+            raw = os.getenv("TELEGRAM_FREE_RESPONSE_CHATS", "")
+        if isinstance(raw, list):
+            return {str(part).strip() for part in raw if str(part).strip()}
+        return {part.strip() for part in str(raw).split(",") if part.strip()}
+
+    def _compile_mention_patterns(self) -> List[re.Pattern]:
+        """Compile optional regex wake-word patterns for group triggers."""
+        patterns = self.config.extra.get("mention_patterns")
+        if patterns is None:
+            raw = os.getenv("TELEGRAM_MENTION_PATTERNS", "").strip()
+            if raw:
+                try:
+                    loaded = json.loads(raw)
+                except Exception:
+                    loaded = [part.strip() for part in raw.splitlines() if part.strip()]
+                    if not loaded:
+                        loaded = [part.strip() for part in raw.split(",") if part.strip()]
+                patterns = loaded
+
+        if patterns is None:
+            return []
+        if isinstance(patterns, str):
+            patterns = [patterns]
+        if not isinstance(patterns, list):
+            logger.warning(
+                "[%s] telegram mention_patterns must be a list or string; got %s",
+                self.name,
+                type(patterns).__name__,
+            )
+            return []
+
+        compiled: List[re.Pattern] = []
+        for pattern in patterns:
+            if not isinstance(pattern, str) or not pattern.strip():
+                continue
+            try:
+                compiled.append(re.compile(pattern, re.IGNORECASE))
+            except re.error as exc:
+                logger.warning("[%s] Invalid Telegram mention pattern %r: %s", self.name, pattern, exc)
+        if compiled:
+            logger.info("[%s] Loaded %d Telegram mention pattern(s)", self.name, len(compiled))
+        return compiled
+
+    def _is_group_chat(self, message: Message) -> bool:
+        chat = getattr(message, "chat", None)
+        if not chat:
+            return False
+        chat_type = str(getattr(chat, "type", "")).split(".")[-1].lower()
+        return chat_type in ("group", "supergroup")
+
+    def _is_reply_to_bot(self, message: Message) -> bool:
+        if not self._bot or not getattr(message, "reply_to_message", None):
+            return False
+        reply_user = getattr(message.reply_to_message, "from_user", None)
+        return bool(reply_user and getattr(reply_user, "id", None) == getattr(self._bot, "id", None))
+
+    def _message_mentions_bot(self, message: Message) -> bool:
+        if not self._bot:
+            return False
+
+        bot_username = (getattr(self._bot, "username", None) or "").lstrip("@").lower()
+        bot_id = getattr(self._bot, "id", None)
+
+        def _iter_sources():
+            yield getattr(message, "text", None) or "", getattr(message, "entities", None) or []
+            yield getattr(message, "caption", None) or "", getattr(message, "caption_entities", None) or []
+
+        for source_text, entities in _iter_sources():
+            if bot_username and f"@{bot_username}" in source_text.lower():
+                return True
+            for entity in entities:
+                entity_type = str(getattr(entity, "type", "")).split(".")[-1].lower()
+                if entity_type == "mention" and bot_username:
+                    offset = int(getattr(entity, "offset", -1))
+                    length = int(getattr(entity, "length", 0))
+                    if offset < 0 or length <= 0:
+                        continue
+                    if source_text[offset:offset + length].strip().lower() == f"@{bot_username}":
+                        return True
+                elif entity_type == "text_mention":
+                    user = getattr(entity, "user", None)
+                    if user and getattr(user, "id", None) == bot_id:
+                        return True
+        return False
+
+    def _message_matches_mention_patterns(self, message: Message) -> bool:
+        if not self._mention_patterns:
+            return False
+        for candidate in (getattr(message, "text", None), getattr(message, "caption", None)):
+            if not candidate:
+                continue
+            for pattern in self._mention_patterns:
+                if pattern.search(candidate):
+                    return True
+        return False
+
+    def _clean_bot_trigger_text(self, text: Optional[str]) -> Optional[str]:
+        if not text or not self._bot or not getattr(self._bot, "username", None):
+            return text
+        username = re.escape(self._bot.username)
+        cleaned = re.sub(rf"(?i)@{username}\b[,:\-]*\s*", "", text).strip()
+        return cleaned or text
+
+    def _should_process_message(self, message: Message, *, is_command: bool = False) -> bool:
+        """Apply Telegram group trigger rules.
+
+        DMs remain unrestricted. Group/supergroup messages are accepted when:
+        - the chat is explicitly allowlisted in ``free_response_chats``
+        - ``require_mention`` is disabled
+        - the message is a command
+        - the message replies to the bot
+        - the bot is @mentioned
+        - the text/caption matches a configured regex wake-word pattern
+        """
+        if not self._is_group_chat(message):
+            return True
+        if str(getattr(getattr(message, "chat", None), "id", "")) in self._telegram_free_response_chats():
+            return True
+        if not self._telegram_require_mention():
+            return True
+        if is_command:
+            return True
+        if self._is_reply_to_bot(message):
+            return True
+        if self._message_mentions_bot(message):
+            return True
+        return self._message_matches_mention_patterns(message)
+
     async def _handle_text_message(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
         """Handle incoming text messages.
 
@@ -1334,14 +1478,19 @@ class TelegramAdapter(BasePlatformAdapter):
         """
         if not update.message or not update.message.text:
             return
+        if not self._should_process_message(update.message):
+            return
 
         event = self._build_message_event(update.message, MessageType.TEXT)
+        event.text = self._clean_bot_trigger_text(event.text)
         self._enqueue_text_event(event)
     
     async def _handle_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
         """Handle incoming command messages."""
         if not update.message or not update.message.text:
             return
+        if not self._should_process_message(update.message, is_command=True):
+            return
         
         event = self._build_message_event(update.message, MessageType.COMMAND)
         await self.handle_message(event)
@@ -1350,6 +1499,8 @@ class TelegramAdapter(BasePlatformAdapter):
         """Handle incoming location/venue pin messages."""
         if not update.message:
             return
+        if not self._should_process_message(update.message):
+            return
 
         msg = update.message
         venue = getattr(msg, "venue", None)
@@ -1493,6 +1644,8 @@ class TelegramAdapter(BasePlatformAdapter):
         """Handle incoming media messages, downloading images to local cache."""
         if not update.message:
             return
+        if not self._should_process_message(update.message):
+            return
         
         msg = update.message
         
@@ -1516,7 +1669,7 @@ class TelegramAdapter(BasePlatformAdapter):
         
         # Add caption as text
         if msg.caption:
-            event.text = msg.caption
+            event.text = self._clean_bot_trigger_text(msg.caption)
         
         # Handle stickers: describe via vision tool with caching
         if msg.sticker:

tests/gateway/test_telegram_group_gating.py

@@ -0,0 +1,110 @@
+import json
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+from gateway.config import Platform, PlatformConfig, load_gateway_config
+
+
+def _make_adapter(require_mention=None, free_response_chats=None, mention_patterns=None):
+    from gateway.platforms.telegram import TelegramAdapter
+
+    extra = {}
+    if require_mention is not None:
+        extra["require_mention"] = require_mention
+    if free_response_chats is not None:
+        extra["free_response_chats"] = free_response_chats
+    if mention_patterns is not None:
+        extra["mention_patterns"] = mention_patterns
+
+    adapter = object.__new__(TelegramAdapter)
+    adapter.platform = Platform.TELEGRAM
+    adapter.config = PlatformConfig(enabled=True, token="***", extra=extra)
+    adapter._bot = SimpleNamespace(id=999, username="hermes_bot")
+    adapter._message_handler = AsyncMock()
+    adapter._pending_text_batches = {}
+    adapter._pending_text_batch_tasks = {}
+    adapter._text_batch_delay_seconds = 0.01
+    adapter._mention_patterns = adapter._compile_mention_patterns()
+    return adapter
+
+
+def _group_message(text="hello", *, chat_id=-100, reply_to_bot=False, entities=None, caption=None, caption_entities=None):
+    reply_to_message = None
+    if reply_to_bot:
+        reply_to_message = SimpleNamespace(from_user=SimpleNamespace(id=999))
+    return SimpleNamespace(
+        text=text,
+        caption=caption,
+        entities=entities or [],
+        caption_entities=caption_entities or [],
+        chat=SimpleNamespace(id=chat_id, type="group"),
+        reply_to_message=reply_to_message,
+    )
+
+
+def _mention_entity(text, mention="@hermes_bot"):
+    offset = text.index(mention)
+    return SimpleNamespace(type="mention", offset=offset, length=len(mention))
+
+
+def test_group_messages_can_be_opened_via_config():
+    adapter = _make_adapter(require_mention=False)
+
+    assert adapter._should_process_message(_group_message("hello everyone")) is True
+
+
+def test_group_messages_can_require_direct_trigger_via_config():
+    adapter = _make_adapter(require_mention=True)
+
+    assert adapter._should_process_message(_group_message("hello everyone")) is False
+    assert adapter._should_process_message(_group_message("hi @hermes_bot", entities=[_mention_entity("hi @hermes_bot")])) is True
+    assert adapter._should_process_message(_group_message("replying", reply_to_bot=True)) is True
+    assert adapter._should_process_message(_group_message("/status"), is_command=True) is True
+
+
+def test_free_response_chats_bypass_mention_requirement():
+    adapter = _make_adapter(require_mention=True, free_response_chats=["-200"])
+
+    assert adapter._should_process_message(_group_message("hello everyone", chat_id=-200)) is True
+    assert adapter._should_process_message(_group_message("hello everyone", chat_id=-201)) is False
+
+
+def test_regex_mention_patterns_allow_custom_wake_words():
+    adapter = _make_adapter(require_mention=True, mention_patterns=[r"^\s*chompy\b"])
+
+    assert adapter._should_process_message(_group_message("chompy status")) is True
+    assert adapter._should_process_message(_group_message("   chompy help")) is True
+    assert adapter._should_process_message(_group_message("hey chompy")) is False
+
+
+def test_invalid_regex_patterns_are_ignored():
+    adapter = _make_adapter(require_mention=True, mention_patterns=[r"(", r"^\s*chompy\b"])
+
+    assert adapter._should_process_message(_group_message("chompy status")) is True
+    assert adapter._should_process_message(_group_message("hello everyone")) is False
+
+
+def test_config_bridges_telegram_group_settings(monkeypatch, tmp_path):
+    hermes_home = tmp_path / ".hermes"
+    hermes_home.mkdir()
+    (hermes_home / "config.yaml").write_text(
+        "telegram:\n"
+        "  require_mention: true\n"
+        "  mention_patterns:\n"
+        "    - \"^\\\\s*chompy\\\\b\"\n"
+        "  free_response_chats:\n"
+        "    - \"-123\"\n",
+        encoding="utf-8",
+    )
+
+    monkeypatch.setenv("HERMES_HOME", str(hermes_home))
+    monkeypatch.delenv("TELEGRAM_REQUIRE_MENTION", raising=False)
+    monkeypatch.delenv("TELEGRAM_MENTION_PATTERNS", raising=False)
+    monkeypatch.delenv("TELEGRAM_FREE_RESPONSE_CHATS", raising=False)
+
+    config = load_gateway_config()
+
+    assert config is not None
+    assert __import__("os").environ["TELEGRAM_REQUIRE_MENTION"] == "true"
+    assert json.loads(__import__("os").environ["TELEGRAM_MENTION_PATTERNS"]) == [r"^\s*chompy\b"]
+    assert __import__("os").environ["TELEGRAM_FREE_RESPONSE_CHATS"] == "-123"

website/docs/user-guide/messaging/telegram.md

@@ -161,9 +161,35 @@ Configure the TTS provider in your `config.yaml` under the `tts.provider` key.
 Hermes Agent works in Telegram group chats with a few considerations:
 
 - **Privacy mode** determines what messages the bot can see (see [Step 3](#step-3-privacy-mode-critical-for-groups))
-- When privacy mode is on, **@mention the bot** (e.g., `@my_hermes_bot what's the weather?`) or **reply to its messages** to interact
-- When privacy mode is off (or bot is admin), the bot sees all messages and can participate naturally
 - `TELEGRAM_ALLOWED_USERS` still applies — only authorized users can trigger the bot, even in groups
+- You can keep the bot from responding to ordinary group chatter with `telegram.require_mention: true`
+- With `telegram.require_mention: true`, group messages are accepted when they are:
+  - slash commands
+  - replies to one of the bot's messages
+  - `@botusername` mentions
+  - matches for one of your configured regex wake words in `telegram.mention_patterns`
+- If `telegram.require_mention` is left unset or false, Hermes keeps the previous open-group behavior and responds to normal group messages it can see
+
+### Example group trigger configuration
+
+Add this to `~/.hermes/config.yaml`:
+
+```yaml
+telegram:
+  require_mention: true
+  mention_patterns:
+    - "^\\s*chompy\\b"
+```
+
+This example allows all the usual direct triggers plus messages that begin with `chompy`, even if they do not use an `@mention`.
+
+### Notes on `mention_patterns`
+
+- Patterns use Python regular expressions
+- Matching is case-insensitive
+- Patterns are checked against both text messages and media captions
+- Invalid regex patterns are ignored with a warning in the gateway logs rather than crashing the bot
+- If you want a pattern to match only at the start of a message, anchor it with `^`
 
 ## Private Chat Topics (Bot API 9.4)