feat(telegram): opt-in Online/Offline bot status indicator (#49134)

Sets the Telegram bot's short description (the line under its name) to
"Online" on gateway connect and "Offline" on clean disconnect, gated
behind extra.status_indicator (off by default).

Telegram bots have no presence/online dot — that's a user-account
feature the Bot API doesn't expose for bots. The short description is
the closest available surface, so this gives users a way to tell whether
the gateway is up from the bot's profile.

- New extra.status_indicator flag (+ status_online/status_offline text
  overrides), read in __init__ via config.extra — no config-schema change.
- _set_status_indicator() helper: best-effort, swallows API errors so it
  never blocks connect/disconnect; truncates to Telegram's 120-char cap.
- Wired Online after _mark_connected(), Offline at top of disconnect()
  while the bot HTTP client is still alive.
- 9 unit tests + Telegram docs section.

Requested by @ilTrumpista, cc @Teknium.
This commit is contained in:
Teknium 2026-06-19 11:38:39 -07:00 committed by GitHub
parent 990273d90a
commit 26e76a75e5
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 214 additions and 0 deletions

View file

@ -476,6 +476,23 @@ class TelegramAdapter(BasePlatformAdapter):
self._forum_command_registered: set[int] = set()
# Lock per la registrazione sicura dei comandi nei forum supergroup
self._forum_lock = asyncio.Lock()
# Status indicator: when enabled, the bot's short description (the line
# shown under its name in the profile) is set to "Online" on connect and
# "Offline" on clean disconnect, so users can tell whether the gateway is
# up. Telegram bots have no real presence/online dot (that's a user-account
# feature), so the short description is the closest available surface.
# Off by default — this mutates the bot's GLOBAL profile, visible to all
# users. Opt in via gateway config: extra.status_indicator: true, or set
# custom strings via extra.status_online / extra.status_offline.
self._status_indicator_enabled: bool = bool(
self.config.extra.get("status_indicator", False)
)
self._status_online_text: str = str(
self.config.extra.get("status_online", "Online")
)
self._status_offline_text: str = str(
self.config.extra.get("status_offline", "Offline")
)
# DM Topics config from extra.dm_topics
self._dm_topics_config: List[Dict[str, Any]] = self.config.extra.get("dm_topics", [])
# Precomputed chat_ids that have DM topics configured (for O(1) root-DM ignore check)
@ -2245,6 +2262,13 @@ class TelegramAdapter(BasePlatformAdapter):
mode = "webhook" if self._webhook_mode else "polling"
logger.info("[%s] Connected to Telegram (%s mode)", self.name, mode)
# Surface the gateway as "Online" in the bot's short description
# (opt-in via extra.status_indicator). Non-fatal.
try:
await self._set_status_indicator(online=True)
except Exception:
pass
# Set up DM topics (Bot API 9.4 — Private Chat Topics)
# Runs after connection is established so the bot can call createForumTopic.
# Failures here are non-fatal — the bot works fine without topics.
@ -2265,8 +2289,47 @@ class TelegramAdapter(BasePlatformAdapter):
logger.error("[%s] Failed to connect to Telegram: %s", self.name, e, exc_info=True)
return False
async def _set_status_indicator(self, online: bool) -> None:
"""Set the bot's short description to the online/offline status text.
The short description is the line shown under the bot's name in its
profile. It is the closest Bot API surface to a presence indicator
bots have no real online/offline dot (that's a user-account feature).
No-op unless ``extra.status_indicator`` is enabled. Best-effort: any
failure is logged at debug and swallowed so it never blocks connect or
disconnect. The default (no language_code) description applies to every
user who doesn't have a language-specific one set.
"""
if not getattr(self, "_status_indicator_enabled", False):
return
bot = self._bot
if bot is None:
return
text = self._status_online_text if online else self._status_offline_text
# Telegram caps short_description at 120 chars.
text = text[:120]
try:
await bot.set_my_short_description(short_description=text)
logger.info("[%s] Set bot status indicator to %r", self.name, text)
except Exception as e:
logger.debug(
"[%s] Failed to set bot status indicator to %r: %s",
self.name, text, e,
)
async def disconnect(self) -> None:
"""Stop polling/webhook, cancel pending album flushes, and disconnect."""
# Mark the bot "Offline" in its short description while the bot's HTTP
# client is still alive (before app shutdown closes it). Opt-in via
# extra.status_indicator. Non-fatal. This is the clean-shutdown path;
# a hard crash leaves the last-known status, which is the expected
# limitation of a profile-text indicator.
try:
await self._set_status_indicator(online=False)
except Exception:
pass
pending_media_group_tasks = list(self._media_group_tasks.values())
for task in pending_media_group_tasks:
task.cancel()

View file

@ -0,0 +1,120 @@
"""Tests for the Telegram bot status indicator.
Telegram bots have no real online/offline presence dot (that's a user-account
feature). The closest Bot API surface is the bot's *short description* — the
line shown under the bot's name in its profile. When `extra.status_indicator`
is enabled, the adapter sets it to "Online" on connect and "Offline" on clean
disconnect so users can tell whether the gateway is up.
"""
import sys
from unittest.mock import AsyncMock, MagicMock
import pytest
from gateway.config import PlatformConfig
def _ensure_telegram_mock():
if "telegram" in sys.modules and hasattr(sys.modules["telegram"], "__file__"):
return
telegram_mod = MagicMock()
telegram_mod.ext.ContextTypes.DEFAULT_TYPE = type(None)
telegram_mod.constants.ParseMode.MARKDOWN_V2 = "MarkdownV2"
telegram_mod.constants.ChatType.GROUP = "group"
telegram_mod.constants.ChatType.SUPERGROUP = "supergroup"
telegram_mod.constants.ChatType.CHANNEL = "channel"
telegram_mod.constants.ChatType.PRIVATE = "private"
for name in ("telegram", "telegram.ext", "telegram.constants", "telegram.request"):
sys.modules.setdefault(name, telegram_mod)
_ensure_telegram_mock()
from gateway.platforms.telegram import TelegramAdapter # noqa: E402
def _make_adapter(extra):
adapter = TelegramAdapter(PlatformConfig(enabled=True, token="***", extra=extra))
adapter._bot = MagicMock()
adapter._bot.set_my_short_description = AsyncMock()
return adapter
def test_disabled_by_default():
adapter = _make_adapter(extra={})
assert adapter._status_indicator_enabled is False
def test_enabled_via_extra():
adapter = _make_adapter(extra={"status_indicator": True})
assert adapter._status_indicator_enabled is True
@pytest.mark.asyncio
async def test_disabled_is_noop():
adapter = _make_adapter(extra={"status_indicator": False})
await adapter._set_status_indicator(online=True)
adapter._bot.set_my_short_description.assert_not_called()
@pytest.mark.asyncio
async def test_online_sets_default_text():
adapter = _make_adapter(extra={"status_indicator": True})
await adapter._set_status_indicator(online=True)
adapter._bot.set_my_short_description.assert_awaited_once_with(
short_description="Online"
)
@pytest.mark.asyncio
async def test_offline_sets_default_text():
adapter = _make_adapter(extra={"status_indicator": True})
await adapter._set_status_indicator(online=False)
adapter._bot.set_my_short_description.assert_awaited_once_with(
short_description="Offline"
)
@pytest.mark.asyncio
async def test_custom_status_strings():
adapter = _make_adapter(
extra={
"status_indicator": True,
"status_online": "🟢 Gateway up",
"status_offline": "🔴 Gateway down",
}
)
await adapter._set_status_indicator(online=True)
adapter._bot.set_my_short_description.assert_awaited_once_with(
short_description="🟢 Gateway up"
)
@pytest.mark.asyncio
async def test_text_truncated_to_120_chars():
adapter = _make_adapter(
extra={"status_indicator": True, "status_online": "x" * 200}
)
await adapter._set_status_indicator(online=True)
_, kwargs = adapter._bot.set_my_short_description.call_args
assert len(kwargs["short_description"]) == 120
@pytest.mark.asyncio
async def test_noop_when_bot_is_none():
adapter = _make_adapter(extra={"status_indicator": True})
adapter._bot = None
# Must not raise even though there's no bot to call.
await adapter._set_status_indicator(online=True)
@pytest.mark.asyncio
async def test_api_failure_is_swallowed():
adapter = _make_adapter(extra={"status_indicator": True})
adapter._bot.set_my_short_description.side_effect = RuntimeError("flood wait")
# Best-effort: a Bot API failure must never propagate out of the helper,
# so it can't block connect/disconnect.
await adapter._set_status_indicator(online=True)

View file

@ -48,6 +48,37 @@ sethome - Set this chat as the home channel
```
:::
### Online/Offline status indicator (Optional)
Telegram bots have no real online/offline presence dot — that green dot is a
*user-account* feature, not something the Bot API exposes for bots. The closest
surface is the bot's **short description** (the line shown under its name in the
bot's profile).
Enable `status_indicator` and Hermes sets that short description to **Online**
when the gateway connects and **Offline** on a clean shutdown:
```yaml
gateway:
platforms:
telegram:
extra:
status_indicator: true
# Optional custom strings (defaults: "Online" / "Offline"):
status_online: "🟢 Online"
status_offline: "🔴 Offline"
```
Notes:
- The short description is **global** to the bot (visible to all users), not
per-chat. Users see it on the bot's profile page, not as a live badge inside
an open chat.
- Only a **clean** gateway shutdown (`/stop`, `disconnect`) writes "Offline".
A hard crash leaves the last-known status — the inherent limitation of a
profile-text indicator.
- Off by default, since it mutates the bot's global profile.
## Step 3: Privacy Mode (Critical for Groups)
Telegram bots have a **privacy mode** that is **enabled by default**. This is the single most common source of confusion when using bots in groups.