feat(telegram): route bot commands to slash-command handlers (chat@4.31 9c936f8)

Port upstream Telegram slash-command routing. process_update now gates a
new handle_slash_command_update before handle_incoming_message_update,
but only for update.message (not edited messages or channel posts), so a
leading /command routes to the core process_slash_command surface
instead of process_message.

parse_slash_command requires a bot_command entity at offset 0 (an
offset-7 command stays a regular message), matches @bot targeting
case-insensitively against user_name (/ping@otherbot is ignored), reads
the caption/caption_entities path when text is absent (has_text = text
is not None, so empty-string text still takes the text branch), and
splits the command from its trailing text on UTF-16-LE code-unit offsets
via a new _slice_utf16 helper rather than naive str slicing.

The event mirrors the Discord adapter: SlashCommandEvent(channel=None)
with the resolved thread id attached as channel_id for Chat to re-wrap.

Adds 7 regression tests (routes command, routes caption, ignores
otherbot, only-leading-entity, empty-string-text branch, UTF-16 split,
naive-slice divergence guard).

Author: patrick-chinchill

src/chat_sdk/adapters/telegram/adapter.py

@@ -83,6 +83,7 @@
     PostableMarkdown,
     RawMessage,
     ReactionEvent,
+    SlashCommandEvent,
     StreamChunk,
     StreamOptions,
     ThreadInfo,
@@ -1049,7 +1050,15 @@ def process_update(
             or update.get("edited_channel_post")
         )
 
-        if message_update:
+        # Slash commands are gated to fresh ``message`` updates only — edited
+        # messages and channel posts never route to the slash-command
+        # handlers. ``handle_slash_command_update`` returns ``True`` when it
+        # consumed the update, in which case the regular message path is
+        # skipped (mirrors upstream ``messageUpdate && !handledSlashCommand``).
+        message = update.get("message")
+        handled_slash_command = message is not None and self.handle_slash_command_update(message, options)
+
+        if message_update and not handled_slash_command:
             self.handle_incoming_message_update(message_update, options)
 
         if update.get("callback_query"):
@@ -1079,6 +1088,103 @@ def handle_incoming_message_update(
 
         self._chat.process_message(self, thread_id, parsed_message, options)
 
+    def handle_slash_command_update(
+        self,
+        telegram_message: TelegramMessage,
+        options: WebhookOptions | None = None,
+    ) -> bool:
+        """Route a leading ``/command`` message to the slash-command handlers.
+
+        Returns ``True`` when the update was consumed as a slash command (so
+        :meth:`process_update` skips the regular message path), and ``False``
+        otherwise. Like the Discord adapter, the event is built with
+        ``channel=None`` and the resolved thread ID is attached as
+        ``channel_id`` — ``Chat`` re-wraps it into a real ``Channel`` before
+        invoking handlers.
+        """
+        if not self._chat:
+            return False
+
+        slash_command = self.parse_slash_command(telegram_message)
+        if not slash_command:
+            return False
+
+        thread_id = self.encode_thread_id(
+            TelegramThreadId(
+                chat_id=str(telegram_message["chat"]["id"]),
+                message_thread_id=telegram_message.get("message_thread_id"),
+            )
+        )
+
+        parsed_message = self.parse_telegram_message(telegram_message, thread_id)
+        self.cache_message(parsed_message)
+
+        event = SlashCommandEvent(
+            adapter=self,
+            channel=None,  # pyrefly: ignore[bad-argument-type]  # filled in by Chat
+            user=parsed_message.author,
+            command=slash_command["command"],
+            text=slash_command["text"],
+            raw=telegram_message,
+        )
+        event.channel_id = thread_id  # type: ignore[attr-defined]
+        self._chat.process_slash_command(event, options)
+
+        return True
+
+    def parse_slash_command(
+        self,
+        telegram_message: TelegramMessage,
+    ) -> dict[str, str] | None:
+        """Extract a leading ``/command`` (and trailing text) from a message.
+
+        Returns ``None`` unless the message carries a ``bot_command`` entity
+        at offset 0 (a command at any other offset routes to
+        ``process_message``). ``@bot`` targeting is matched
+        case-insensitively against :attr:`user_name`; a command addressed to
+        another bot (``/ping@otherbot``) is ignored. Both the text/caption
+        selection (``has_text = text is not None``, so an empty-string
+        ``text`` still takes the text branch) and the command/trailing-text
+        split use UTF-16-LE offsets, matching Telegram's entity indexing.
+        """
+        has_text = telegram_message.get("text") is not None
+        text = telegram_message.get("text") if has_text else telegram_message.get("caption")
+        entities = (
+            (telegram_message.get("entities") or []) if has_text else (telegram_message.get("caption_entities") or [])
+        )
+
+        if not text:
+            return None
+
+        command_entity = next(
+            (e for e in entities if e.get("type") == "bot_command" and e.get("offset", 0) == 0),
+            None,
+        )
+
+        if not command_entity:
+            return None
+
+        raw_command = self.entity_text(text, command_entity)
+        if not raw_command.startswith("/"):
+            return None
+
+        command_without_slash = raw_command[1:]
+        at_index = command_without_slash.find("@")
+        command_name = command_without_slash if at_index == -1 else command_without_slash[:at_index]
+        target_bot = None if at_index == -1 else command_without_slash[at_index + 1 :]
+
+        if not command_name:
+            return None
+
+        if target_bot and target_bot.lower() != self._user_name.lower():
+            return None
+
+        offset = command_entity.get("offset", 0)
+        length = command_entity.get("length", 0)
+        trailing = self._slice_utf16(text, offset + length).lstrip()
+
+        return {"command": f"/{command_name}", "text": trailing}
+
     def handle_callback_query(
         self,
         callback_query: TelegramCallbackQuery,
@@ -2493,6 +2599,18 @@ def entity_text(self, text: str, entity: TelegramMessageEntity) -> str:
         utf16 = text.encode("utf-16-le")
         return utf16[offset * 2 : (offset + length) * 2].decode("utf-16-le")
 
+    @staticmethod
+    def _slice_utf16(text: str, offset: int) -> str:
+        """Return the substring of ``text`` from a UTF-16 code-unit ``offset``.
+
+        Telegram entity offsets count UTF-16 code units (matching JavaScript
+        ``String.prototype.slice``), so an astral-plane code point (e.g. an
+        emoji) advances the offset by two. Naive Python ``str`` slicing counts
+        code points and would mis-split such text — this encodes to UTF-16-LE
+        and slices by byte, mirroring :meth:`entity_text`.
+        """
+        return text.encode("utf-16-le")[offset * 2 :].decode("utf-16-le")
+
     @staticmethod
     def escape_regex(input_str: str) -> str:
         """Escape regex special characters."""

tests/test_telegram_webhook.py

@@ -218,6 +218,184 @@ async def test_returns_400_for_invalid_json(self):
         assert response["status"] == 400
 
 
+# ---------------------------------------------------------------------------
+# Slash command routing (chat@4.31 9c936f8)
+# ---------------------------------------------------------------------------
+
+
+def _slash_adapter_and_chat() -> tuple[TelegramAdapter, Any]:
+    """Wire a ``userName=mybot`` adapter to a mock chat.
+
+    ``Chat.process_slash_command`` / ``process_message`` are *synchronous*
+    methods (they spawn fire-and-forget tasks internally), and the adapter
+    calls them synchronously, so the mock uses ``MagicMock`` — an
+    ``AsyncMock`` would hand the adapter an unawaited coroutine that never
+    reflects the real (sync) call.
+    """
+    from unittest.mock import MagicMock
+
+    adapter = _make_adapter(user_name="mybot")
+    chat = MagicMock()
+    chat.process_slash_command = MagicMock()
+    chat.process_message = MagicMock()
+    adapter._chat = chat
+    return adapter, chat
+
+
+class TestTelegramSlashCommandRouting:
+    """Bot-command routing ported from the TS index.test.ts blocks."""
+
+    @pytest.mark.asyncio
+    async def test_routes_bot_command_messages_to_slash_handlers(self):
+        adapter, chat = _slash_adapter_and_chat()
+        body = json.dumps(
+            {
+                "update_id": 2,
+                "message": _sample_message(
+                    text="/ping@mybot hello world",
+                    entities=[{"type": "bot_command", "offset": 0, "length": 11}],
+                ),
+            }
+        )
+
+        response = await adapter.handle_webhook(_make_request(body))
+        assert response["status"] == 200
+
+        assert chat.process_slash_command.call_count == 1
+        chat.process_message.assert_not_called()
+
+        event = chat.process_slash_command.call_args.args[0]
+        assert event.channel_id == "telegram:123"
+        assert event.command == "/ping"
+        assert event.text == "hello world"
+        assert event.user.full_name == "User"
+        assert event.user.user_id == "456"
+
+    @pytest.mark.asyncio
+    async def test_routes_bot_command_captions_to_slash_handlers(self):
+        adapter, chat = _slash_adapter_and_chat()
+        body = json.dumps(
+            {
+                "update_id": 3,
+                "message": _sample_message(
+                    caption="/ping hello world",
+                    text=None,
+                    caption_entities=[{"type": "bot_command", "offset": 0, "length": 5}],
+                    photo=[
+                        {
+                            "file_id": "photo-1",
+                            "file_unique_id": "photo-unique-1",
+                            "height": 100,
+                            "width": 100,
+                        }
+                    ],
+                ),
+            }
+        )
+
+        response = await adapter.handle_webhook(_make_request(body))
+        assert response["status"] == 200
+
+        assert chat.process_slash_command.call_count == 1
+        chat.process_message.assert_not_called()
+
+        event = chat.process_slash_command.call_args.args[0]
+        assert event.command == "/ping"
+        assert event.text == "hello world"
+
+    @pytest.mark.asyncio
+    async def test_ignores_bot_commands_addressed_to_another_bot(self):
+        adapter, chat = _slash_adapter_and_chat()
+        body = json.dumps(
+            {
+                "update_id": 3,
+                "message": _sample_message(
+                    text="/ping@otherbot hello world",
+                    entities=[{"type": "bot_command", "offset": 0, "length": 14}],
+                ),
+            }
+        )
+
+        response = await adapter.handle_webhook(_make_request(body))
+        assert response["status"] == 200
+
+        chat.process_slash_command.assert_not_called()
+        assert chat.process_message.call_count == 1
+
+    @pytest.mark.asyncio
+    async def test_only_treats_leading_bot_command_entities_as_slash_commands(self):
+        adapter, chat = _slash_adapter_and_chat()
+        body = json.dumps(
+            {
+                "update_id": 4,
+                "message": _sample_message(
+                    text="please /ping",
+                    entities=[{"type": "bot_command", "offset": 7, "length": 5}],
+                ),
+            }
+        )
+
+        response = await adapter.handle_webhook(_make_request(body))
+        assert response["status"] == 200
+
+        chat.process_slash_command.assert_not_called()
+        assert chat.process_message.call_count == 1
+
+    @pytest.mark.asyncio
+    async def test_empty_string_text_takes_text_branch_not_caption(self):
+        """``has_text = text is not None``: empty ``text`` still uses the
+        text branch, so a caption-side ``bot_command`` entity is ignored and
+        the update routes to ``process_message`` (input-sweep regression)."""
+        adapter, chat = _slash_adapter_and_chat()
+        body = json.dumps(
+            {
+                "update_id": 5,
+                "message": _sample_message(
+                    text="",
+                    caption="/ping hello",
+                    caption_entities=[{"type": "bot_command", "offset": 0, "length": 5}],
+                ),
+            }
+        )
+
+        response = await adapter.handle_webhook(_make_request(body))
+        assert response["status"] == 200
+
+        chat.process_slash_command.assert_not_called()
+        assert chat.process_message.call_count == 1
+
+    def test_trailing_text_split_uses_utf16_offsets(self):
+        """The command/trailing-text split is computed on UTF-16 code-unit
+        offsets, not Python code points.
+
+        Telegram reports ``length`` in UTF-16 code units. The entity
+        ``/p😀g`` spans 4 code points but 5 UTF-16 units (the astral emoji is
+        a surrogate pair), so the trailing-text split starts at unit 5. A
+        naive code-point slice — ``text[5:]`` — over-advances and drops the
+        leading ``h``; the UTF-16-aware slice keeps it.
+        """
+        adapter, _ = _slash_adapter_and_chat()
+        # "/p😀g" = / p <emoji=2 units> g = 4 code points but 5 UTF-16 units.
+        result = adapter.parse_slash_command(
+            _sample_message(
+                text="/p😀g hello",
+                entities=[{"type": "bot_command", "offset": 0, "length": 5}],
+            )
+        )
+        assert result == {"command": "/p😀g", "text": "hello"}
+
+    def test_entity_text_split_naive_codepoint_would_diverge(self):
+        """Guards the UTF-16 split against a naive ``str`` slice regression.
+
+        ``_slice_utf16`` and a naive code-point slice must diverge for astral
+        text, proving the helper is load-bearing (not a no-op on ASCII)."""
+        adapter, _ = _slash_adapter_and_chat()
+        text = "/p😀g hello"
+        assert adapter._slice_utf16(text, 5) == " hello"
+        # The naive code-point slice over-advances and eats the leading "h".
+        assert text[5:] == "hello"
+
+
 # ---------------------------------------------------------------------------
 # isDM
 # ---------------------------------------------------------------------------