fix(channels): send Telegram messages as rich only when content has rich constructs (#5470)

backend/app/channels/telegram.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import asyncio
 import logging
+import re
 import threading
 import time
 from collections.abc import Coroutine
@@ -42,6 +43,35 @@ MAX_TRACKED_STREAM_MESSAGES = 256
 # Indirection so tests can patch the clock without touching the global time module.
 _monotonic = time.monotonic
 
+# Rich Messages only earn their keep when the text carries a construct that
+# renders natively: a fenced code block, emphasis, a table, a task list,
+# <details>, block math, or a link. Structured command/error replies are plain
+# text with none of these, so they stay plain and their newlines and
+# <placeholder> tokens survive verbatim instead of being collapsed by the
+# rich parser.
+#
+# The patterns are deliberately *well-formed*, not "contains this character":
+# a table must be a line that leads with a pipe, a link must be [text](url),
+# so a command line like "/goal [condition|clear]" (brackets + a mid-line pipe)
+# never trips the detector.
+_TELEGRAM_RICH_CONSTRUCT_RE = re.compile(
+    r"(?m)"
+    r"^\s*(`{3,}|~{3,})"  # fenced code block
+    r"|^\s*[-*+]\s+\[[ xX]\]"  # task list item
+    r"|^\s*\|.*\|"  # table row (line leads with a pipe)
+    r"|^\s*\|?\s*:?-{2,}\s*\|"  # table separator row (needs a pipe, so "--flag" stays plain)
+    r"|\[[^\]]*\]\("  # markdown link [text](url)
+    r"|\*\*[^*\n]+\*\*"  # bold
+    r"|\*(?!\s)[^*\n]+?(?<!\s)\*"  # italic (tight delimiters, so "2 * 3" stays plain)
+    r"|<details"  # collapsible details
+    r"|\$\$"  # block math
+)
+
+
+def _has_rich_constructs(text: str) -> bool:
+    """Whether *text* contains a construct that needs native rich rendering."""
+    return _TELEGRAM_RICH_CONSTRUCT_RE.search(text) is not None
+
 
 def _load_telegram_input_file(path, filename: str):
     from telegram import InputFile
@@ -310,7 +340,11 @@ class TelegramChannel(Channel):
         return False
 
     def _can_send_rich(self, text: str) -> bool:
-        return bool(self.config.get("rich_messages")) and 0 < len(text) <= TELEGRAM_MAX_RICH_MESSAGE_LENGTH
+        # Rich Messages are used only when rich_messages is on and the text
+        # actually contains a rich construct. Structured command/error replies
+        # are plain text with none, so they stay plain and their newlines and
+        # <placeholder> tokens are not collapsed into one line.
+        return bool(self.config.get("rich_messages")) and 0 < len(text) <= TELEGRAM_MAX_RICH_MESSAGE_LENGTH and _has_rich_constructs(text)
 
     async def _edit_rich_message(self, chat_id: int, message_id: int, text: str) -> bool:
         """Replace a streamed preview with a persistent Telegram Rich Message."""

backend/tests/test_channels.py

@@ -10317,6 +10317,76 @@ class TestTelegramStreaming:
 
         _run(go())
 
+    def test_plain_command_reply_stays_plain_when_enabled(self):
+        """A plain command/error reply (no rich construct) must stay plain text
+        even when rich_messages is on, so newlines and <placeholder> tokens
+        survive. The text deliberately carries a bracketed-pipe token
+        ([condition|clear]) to prove the construct detector stays well-formed."""
+
+        async def go():
+            ch, bot = self._make_channel_with_bot()
+            ch.config["rich_messages"] = True
+            help_text = "Available commands:\n/goal [condition|clear] — Set or clear a goal\n/agent use <name> — Start with an agent"
+
+            await ch.send(OutboundMessage(channel_name="telegram", chat_id="12345", thread_id="t1", text=help_text, is_final=True))
+
+            assert bot.rich == []
+            assert [message["text"] for message in bot.sent] == [help_text]
+
+        _run(go())
+
+    def test_plain_flag_list_stays_plain_when_enabled(self):
+        """A reply whose lines merely *start* with ``--`` (CLI flag lists,
+        signature separators) must stay plain: a table-separator row needs a
+        pipe, so a bare ``--verbose`` line is not a GFM delimiter row."""
+
+        async def go():
+            ch, bot = self._make_channel_with_bot()
+            ch.config["rich_messages"] = True
+            flag_list = "Options:\n--verbose\n--help"
+
+            await ch.send(OutboundMessage(channel_name="telegram", chat_id="12345", thread_id="t1", text=flag_list, is_final=True))
+
+            assert bot.rich == []
+            assert [message["text"] for message in bot.sent] == [flag_list]
+
+        _run(go())
+
+    def test_plain_arithmetic_stays_plain_when_enabled(self):
+        """A reply with spaced asterisks used as multiplication (``2 * 3 * 4``)
+        must stay plain: italic emphasis needs tight, non-space delimiters, so
+        the spans between the asterisks are not handed to the rich parser."""
+
+        async def go():
+            ch, bot = self._make_channel_with_bot()
+            ch.config["rich_messages"] = True
+            arithmetic = "Compute: 2 * 3 * 4 = 24"
+
+            await ch.send(OutboundMessage(channel_name="telegram", chat_id="12345", thread_id="t1", text=arithmetic, is_final=True))
+
+            assert bot.rich == []
+            assert [message["text"] for message in bot.sent] == [arithmetic]
+
+        _run(go())
+
+    def test_plain_command_reply_stays_plain_on_stream_edit(self, monkeypatch):
+        """A streamed-then-final plain reply (no rich construct) must not be
+        replaced by a rich edit of the in-flight placeholder."""
+
+        async def go():
+            ch, bot = self._make_channel_with_bot()
+            ch.config["rich_messages"] = True
+            monkeypatch.setattr("app.channels.telegram._monotonic", lambda: 1000.0)
+
+            await ch._send_running_reply("12345", 42)
+            await ch.send(OutboundMessage(channel_name="telegram", chat_id="12345", thread_id="t1", text="Available commands:\n/new — new", is_final=True, thread_ts="42"))
+
+            assert bot.rich == []
+            # Final text is applied as a plain edit of the streamed placeholder.
+            assert [message["text"] for message in bot.edited] == ["Available commands:\n/new — new"]
+
+        _run(go())
+
     def test_final_replaces_plain_stream_with_rich_message(self, monkeypatch):
         async def go():
             ch, bot = self._make_channel_with_bot()