feat(gateway): add Telegram guest mention mode

This commit is contained in:
Nikita Nosov 2026-05-08 11:49:55 +00:00 committed by kshitij
parent 369cee018d
commit 55f518e521
6 changed files with 168 additions and 25 deletions

View file

@ -23,10 +23,10 @@ from gateway.config import Platform, PlatformConfig
# Telegram
# ---------------------------------------------------------------------------
def _make_telegram_adapter(*, allowed_chats=None, require_mention=None):
def _make_telegram_adapter(*, allowed_chats=None, require_mention=None, guest_mode=False):
from gateway.platforms.telegram import TelegramAdapter
extra = {}
extra = {"guest_mode": guest_mode}
if allowed_chats is not None:
extra["allowed_chats"] = allowed_chats
if require_mention is not None:

View file

@ -7,6 +7,7 @@ or corrupt user-visible content.
import re
import sys
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock
import pytest
@ -757,3 +758,72 @@ class TestEditMessageStreamingSafety:
"message_id": 456,
"text": "final **bold**",
}
# =========================================================================
# Telegram guest mention gating
# =========================================================================
def _guest_test_adapter(*, guest_mode=True, require_mention=True, allowed_chats=None):
config = PlatformConfig(
enabled=True,
token="fake-token",
extra={
"guest_mode": guest_mode,
"require_mention": require_mention,
"allowed_chats": allowed_chats or ["-100200"],
},
)
adapter = object.__new__(TelegramAdapter)
adapter.config = config
adapter._bot = SimpleNamespace(id=999, username="hermes_bot")
adapter._mention_patterns = adapter._compile_mention_patterns()
return adapter
def _guest_group_message(text, *, chat_id=-100201, entities=None, reply_to_bot=False):
reply_to_message = SimpleNamespace(from_user=SimpleNamespace(id=999)) if reply_to_bot else None
return SimpleNamespace(
text=text,
caption=None,
entities=entities or [],
caption_entities=[],
message_thread_id=None,
chat=SimpleNamespace(id=chat_id, type="group"),
from_user=SimpleNamespace(id=111),
reply_to_message=reply_to_message,
)
def _guest_mention_entity(text, mention="@hermes_bot"):
return SimpleNamespace(type="mention", offset=text.index(mention), length=len(mention))
class TestTelegramGuestMentionGating:
def test_guest_mode_allows_explicit_mention_outside_allowed_chats(self):
adapter = _guest_test_adapter(guest_mode=True, allowed_chats=["-100200"])
text = "please help @hermes_bot"
message = _guest_group_message(
text,
chat_id=-100201,
entities=[_guest_mention_entity(text)],
)
assert adapter._should_process_message(message) is True
def test_guest_mode_does_not_allow_reply_outside_allowed_chats(self):
adapter = _guest_test_adapter(guest_mode=True, allowed_chats=["-100200"])
message = _guest_group_message("replying without mention", chat_id=-100201, reply_to_bot=True)
assert adapter._should_process_message(message) is False
def test_guest_mode_disabled_keeps_allowed_chats_as_hard_gate_for_mentions(self):
adapter = _guest_test_adapter(guest_mode=False, allowed_chats=["-100200"])
text = "please help @hermes_bot"
message = _guest_group_message(
text,
chat_id=-100201,
entities=[_guest_mention_entity(text)],
)
assert adapter._should_process_message(message) is False

View file

@ -12,6 +12,8 @@ def _make_adapter(
ignored_threads=None,
allow_from=None,
group_allow_from=None,
allowed_chats=None,
guest_mode=None,
):
from gateway.platforms.telegram import TelegramAdapter
@ -28,6 +30,10 @@ def _make_adapter(
extra["allow_from"] = allow_from
if group_allow_from is not None:
extra["group_allow_from"] = group_allow_from
if allowed_chats is not None:
extra["allowed_chats"] = allowed_chats
if guest_mode is not None:
extra["guest_mode"] = guest_mode
adapter = object.__new__(TelegramAdapter)
adapter.platform = Platform.TELEGRAM
@ -150,6 +156,36 @@ def test_free_response_chats_bypass_mention_requirement():
assert adapter._should_process_message(_group_message("hello everyone", chat_id=-201)) is False
def test_guest_mode_allows_only_direct_mentions_outside_allowed_chats():
adapter = _make_adapter(
require_mention=True,
allowed_chats=["-200"],
guest_mode=True,
mention_patterns=[r"^\s*chompy\b"],
)
mentioned = _group_message(
"hi @hermes_bot",
chat_id=-201,
entities=[_mention_entity("hi @hermes_bot")],
)
assert adapter._should_process_message(mentioned) is True
assert adapter._should_process_message(_group_message("reply", chat_id=-201, reply_to_bot=True)) is False
assert adapter._should_process_message(_group_message("chompy status", chat_id=-201)) is False
assert adapter._should_process_message(_group_message("hello", chat_id=-201)) is False
def test_guest_mode_defaults_to_false_for_allowed_chat_bypass():
adapter = _make_adapter(require_mention=True, allowed_chats=["-200"], guest_mode=False)
mentioned = _group_message(
"hi @hermes_bot",
chat_id=-201,
entities=[_mention_entity("hi @hermes_bot")],
)
assert adapter._should_process_message(mentioned) is False
def test_ignored_threads_drop_group_messages_before_other_gates():
adapter = _make_adapter(require_mention=False, free_response_chats=["-200"], ignored_threads=[31, "42"])
@ -179,6 +215,7 @@ def test_config_bridges_telegram_group_settings(monkeypatch, tmp_path):
(hermes_home / "config.yaml").write_text(
"telegram:\n"
" require_mention: true\n"
" guest_mode: true\n"
" mention_patterns:\n"
" - \"^\\\\s*chompy\\\\b\"\n"
" free_response_chats:\n"
@ -189,14 +226,19 @@ def test_config_bridges_telegram_group_settings(monkeypatch, tmp_path):
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.delenv("TELEGRAM_REQUIRE_MENTION", raising=False)
monkeypatch.delenv("TELEGRAM_MENTION_PATTERNS", raising=False)
monkeypatch.delenv("TELEGRAM_GUEST_MODE", 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 __import__("os").environ["TELEGRAM_GUEST_MODE"] == "true"
assert json.loads(__import__("os").environ["TELEGRAM_MENTION_PATTERNS"]) == [r"^\s*chompy\b"]
assert __import__("os").environ["TELEGRAM_FREE_RESPONSE_CHATS"] == "-123"
tg_cfg = config.platforms.get(Platform.TELEGRAM)
assert tg_cfg is not None
assert tg_cfg.extra.get("guest_mode") is True
def test_config_bridges_telegram_user_allowlists(monkeypatch, tmp_path):