Telegram message formatting and MarkdownV2 chunk indicator handling

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #1478 fix(telegram): escape chunk indicators in MarkdownV2

  merged to main

  files: gateway/platforms/telegram.py, tests/gateway/test_telegram_format.py

  Show the diff

  • gateway/platforms/telegram.py

    @@ -322,6 +322,14 @@ async def send(
                 # Format and split message if needed
                 formatted = self.format_message(content)
                 chunks = self.truncate_message(formatted, self.MAX_MESSAGE_LENGTH)
    +            if len(chunks) > 1:
    +                # truncate_message appends a raw " (1/2)" suffix. Escape the
    +                # MarkdownV2-special parentheses so Telegram doesn't reject the
    +                # chunk and fall back to plain text.
    +                chunks = [
    +                    re.sub(r" \((\d+)/(\d+)\)$", r" \\(\1/\2\\)", chunk)
    +                    for chunk in chunks
    +                ]
                 
                 message_ids = []
                 thread_id = metadata.get("thread_id") if metadata else None

  • tests/gateway/test_telegram_format.py

    @@ -7,7 +7,7 @@
     
     import re
     import sys
    -from unittest.mock import MagicMock
    +from unittest.mock import AsyncMock, MagicMock
     
     import pytest
     
    @@ -392,3 +392,27 @@ def test_plain_text_unchanged(self):
     
         def test_empty_string(self):
             assert _strip_mdv2("") == ""
    +
    +
    [redacted:email]
    +async def test_send_escapes_chunk_indicator_for_markdownv2(adapter):
    +    adapter.MAX_MESSAGE_LENGTH = 80
    +    adapter._bot = MagicMock()
    +
    +    sent_texts = []
    +
    +    async def _fake_send_message(**kwargs):
    +        sent_texts.append(kwargs["text"])
    +        msg = MagicMock()
    +        msg.message_id = len(sent_texts)
    +        return msg
    +
    +    adapter._bot.send_message = AsyncMock(side_effect=_fake_send_message)
    +
    +    content = ("**bold** chunk content " * 12).strip()
    +    result = await adapter.send("123", content)
    +
    +    assert result.success is True
    +    assert len(sent_texts) > 1
    +    assert re.search(r" \\\([0-9]+/[0-9]+\\\)$", sent_texts[0])
    +    assert re.search(r" \\\([0-9]+/[0-9]+\\\)$", sent_texts[-1])