From 6e5ce4fa07cc4c2fa0407279a039d0e8d92b76ee Mon Sep 17 00:00:00 2001 From: rookiestar28 Date: Tue, 28 Apr 2026 19:08:04 +0800 Subject: [PATCH] feat(connector): preserve Telegram topic delivery --- README.md | 1 + connector/platforms/telegram_polling.py | 88 +++++++++++++++- docs/connector.md | 3 +- tests/test_f72_telegram_topic_delivery.py | 123 ++++++++++++++++++++++ 4 files changed, 211 insertions(+), 4 deletions(-) create mode 100644 tests/test_f72_telegram_topic_delivery.py diff --git a/README.md b/README.md index acb95fd..342e0c7 100644 --- a/README.md +++ b/README.md @@ -670,6 +670,7 @@ The connector currently remains an **optional attached subsystem inside this rep - **Run Jobs**: Submit templates via chat commands. - **Approvals**: Approve/Reject paused workflows from your phone. - **Secure**: Outbound-only for Telegram/Discord. LINE/WhatsApp/WeChat/KakaoTalk/Slack require inbound HTTPS (webhook), while Slack can also use Socket Mode and Feishu can run in either webhook or long-connection mode with a dedicated callback ingress path. +- **Telegram topics**: Forum topic commands keep their topic context for immediate replies and delayed result delivery. - **WeChat encrypted mode**: Official Account encrypted webhook mode is supported when AES settings are configured. - **KakaoTalk response safety**: QuickReply limits and safe fallback handling are enforced for reliable payload behavior. - **Slack multi-workspace and interactive mode**: Workspace installs can be handled through connector-managed OAuth install/callback routes with per-workspace token binding, fail-closed health diagnostics, and signed interactive callback handling for action payloads. diff --git a/connector/platforms/telegram_polling.py b/connector/platforms/telegram_polling.py index 0c5c19e..020dd3b 100644 --- a/connector/platforms/telegram_polling.py +++ b/connector/platforms/telegram_polling.py @@ -5,6 +5,7 @@ Long-polling implementation for Telegram Bot API. import asyncio import logging +import re import time from typing import Optional @@ -14,6 +15,7 @@ from ..router import CommandRouter from ..state import ConnectorState logger = logging.getLogger(__name__) +_THREAD_ID_RE = re.compile(r"^\d{1,10}$") def _import_aiohttp(): @@ -24,6 +26,21 @@ def _import_aiohttp(): return aiohttp +def _normalize_message_thread_id(value) -> Optional[int]: + if value is None or isinstance(value, bool): + return None + text = str(value).strip() + if not _THREAD_ID_RE.fullmatch(text): + return None + try: + thread_id = int(text) + except ValueError: + return None + if thread_id <= 0: + return None + return thread_id + + class TelegramPolling: def __init__(self, config: ConnectorConfig, router: CommandRouter): self.config = config @@ -125,6 +142,9 @@ class TelegramPolling: user_id = from_obj.get("id") username = from_obj.get("username") or sender_chat.get("username") or "unknown" text = message["text"] + message_thread_id = _normalize_message_thread_id( + message.get("message_thread_id") + ) # Security Check is_allowed = False @@ -149,24 +169,72 @@ class TelegramPolling: message_id=str(message["message_id"]), text=text, timestamp=time.time(), + thread_id=str(message_thread_id or ""), ) try: resp = await self.router.handle(req) - await self._send_response(chat_id, resp) + await self._send_response( + chat_id, + resp, + delivery_context=( + {"thread_id": req.thread_id} if req.thread_id else None + ), + ) except Exception as e: logger.exception(f"Error handling command: {e}") await self._send_response( - chat_id, CommandResponse(text="[Error] Internal processing error.") + chat_id, + CommandResponse(text="[Error] Internal processing error."), + delivery_context=( + {"thread_id": req.thread_id} if req.thread_id else None + ), ) - async def _send_response(self, chat_id: int, resp: CommandResponse): + def _thread_id_from_context( + self, delivery_context: Optional[dict] + ) -> Optional[int]: + context = delivery_context or {} + return _normalize_message_thread_id(context.get("thread_id")) + + async def _send_thread_diagnostic(self, chat_id, raw_thread_id) -> None: + preview = str(raw_thread_id or "")[:32] + logger.warning("Invalid Telegram message_thread_id ignored: %r", preview) + if not self.session: + return + url = f"{self.base_url}/sendMessage" + payload = { + "chat_id": chat_id, + "text": "[OpenClaw] Invalid Telegram thread/topic id; delivery used the parent chat.", + } + try: + async with self.session.post(url, json=payload) as r: + if r.status != 200: + logger.error( + f"Failed to send Telegram thread diagnostic: {r.status} {await r.text()}" + ) + except Exception as e: + logger.error(f"Telegram thread diagnostic exception: {e}") + + async def _send_response( + self, + chat_id: int, + resp: CommandResponse, + delivery_context: Optional[dict] = None, + ): url = f"{self.base_url}/sendMessage" payload = { "chat_id": chat_id, # Remediation: Plain text only, no parse_mode "text": resp.text, } + thread_id = self._thread_id_from_context(delivery_context) + if thread_id is not None: + payload["message_thread_id"] = thread_id + elif delivery_context and delivery_context.get("thread_id"): + await self._send_thread_diagnostic( + chat_id, delivery_context.get("thread_id") + ) try: async with self.session.post(url, json=payload) as r: if r.status != 200: @@ -193,6 +261,13 @@ class TelegramPolling: url = f"{self.base_url}/sendPhoto" data = aiohttp.FormData() data.add_field("chat_id", channel_id) + thread_id = self._thread_id_from_context(delivery_context) + if thread_id is not None: + data.add_field("message_thread_id", str(thread_id)) + elif delivery_context and delivery_context.get("thread_id"): + await self._send_thread_diagnostic( + channel_id, delivery_context.get("thread_id") + ) if caption: data.add_field("caption", caption) @@ -220,6 +295,13 @@ class TelegramPolling: # Using simplified direct call url = f"{self.base_url}/sendMessage" payload = {"chat_id": channel_id, "text": text} + thread_id = self._thread_id_from_context(delivery_context) + if thread_id is not None: + payload["message_thread_id"] = thread_id + elif delivery_context and delivery_context.get("thread_id"): + await self._send_thread_diagnostic( + channel_id, delivery_context.get("thread_id") + ) try: async with self.session.post(url, json=payload) as r: if r.status != 200: diff --git a/docs/connector.md b/docs/connector.md index e0c6059..0c0708a 100644 --- a/docs/connector.md +++ b/docs/connector.md @@ -74,7 +74,7 @@ When backend multi-tenant mode is enabled (`OPENCLAW_MULTI_TENANT_ENABLED=1`): ## Supported Platforms -- **Telegram**: Long-polling (instant response). +- **Telegram**: Long-polling (instant response), including forum topic reply context. - **Discord**: Gateway WebSocket (instant response). - **LINE**: Webhook (requires inbound HTTPS). - **WhatsApp**: Webhook (requires inbound HTTPS). @@ -115,6 +115,7 @@ Set the following environment variables (or put them in a `.env` file if you use - `OPENCLAW_CONNECTOR_TELEGRAM_TOKEN`: Your Bot Token (from @BotFather). - `OPENCLAW_CONNECTOR_TELEGRAM_ALLOWED_USERS`: Comma-separated list of User IDs (e.g. `123456, 789012`). - `OPENCLAW_CONNECTOR_TELEGRAM_ALLOWED_CHATS`: Comma-separated list of Chat/Group IDs. +- Telegram forum topics are preserved when Telegram provides `message_thread_id`; command replies and delayed result delivery are sent back to the same topic. Manually configured delivery contexts must use numeric topic/thread IDs. **Discord:** diff --git a/tests/test_f72_telegram_topic_delivery.py b/tests/test_f72_telegram_topic_delivery.py new file mode 100644 index 0000000..e656b23 --- /dev/null +++ b/tests/test_f72_telegram_topic_delivery.py @@ -0,0 +1,123 @@ +import unittest +from unittest.mock import AsyncMock + +from connector.config import ConnectorConfig +from connector.contract import CommandResponse +from connector.platforms.telegram_polling import TelegramPolling + + +class _FakeResponse: + status = 200 + + async def text(self): + return "ok" + + +class _FakePostContext: + def __init__(self, response): + self.response = response + + async def __aenter__(self): + return self.response + + async def __aexit__(self, exc_type, exc, tb): + return False + + +class _FakeSession: + def __init__(self): + self.posts = [] + + def post(self, url, **kwargs): + self.posts.append((url, kwargs)) + return _FakePostContext(_FakeResponse()) + + +class _FakeRouter: + def __init__(self): + self.requests = [] + + async def handle(self, req): + self.requests.append(req) + return CommandResponse(text="topic reply") + + +def _form_field_value(form_data, name): + for headers, _extra, value in form_data._fields: + if headers.get("name") == name: + return value + return None + + +class TestTelegramTopicDelivery(unittest.IsolatedAsyncioTestCase): + def _server(self): + cfg = ConnectorConfig() + cfg.telegram_bot_token = "token" + cfg.telegram_allowed_chats = [-100123] + router = _FakeRouter() + server = TelegramPolling(cfg, router) + server.session = _FakeSession() + return server, router + + async def test_inbound_topic_update_preserves_thread_and_replies_to_topic(self): + server, router = self._server() + + await server._process_update( + { + "update_id": 1, + "message": { + "message_id": 77, + "message_thread_id": 456, + "chat": {"id": -100123, "type": "supergroup"}, + "from": {"id": 42, "username": "alice"}, + "text": "/status", + }, + } + ) + + self.assertEqual(router.requests[0].thread_id, "456") + _url, kwargs = server.session.posts[-1] + self.assertEqual(kwargs["json"]["message_thread_id"], 456) + + async def test_send_message_includes_valid_thread_id(self): + server, _router = self._server() + + await server.send_message( + "-100123", + "done", + delivery_context={"thread_id": "456"}, + ) + + _url, kwargs = server.session.posts[-1] + self.assertEqual(kwargs["json"]["message_thread_id"], 456) + + async def test_send_image_includes_valid_thread_id(self): + server, _router = self._server() + + await server.send_image( + "-100123", + b"image", + filename="out.png", + delivery_context={"thread_id": "456"}, + ) + + _url, kwargs = server.session.posts[-1] + self.assertEqual(_form_field_value(kwargs["data"], "message_thread_id"), "456") + + async def test_malformed_thread_id_is_diagnostic_not_telegram_parameter(self): + server, _router = self._server() + server._send_thread_diagnostic = AsyncMock() + + await server.send_message( + "-100123", + "done", + delivery_context={"thread_id": "abc123"}, + ) + + _url, kwargs = server.session.posts[-1] + self.assertNotIn("message_thread_id", kwargs["json"]) + server._send_thread_diagnostic.assert_awaited_once() + + +if __name__ == "__main__": + unittest.main()