feat(connector): preserve Telegram topic delivery

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.

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:

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:**
 

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()