fix(telegram): escape dynamic markdown in callback flows

Use MarkdownV2 formatting for Telegram callback follow-ups and interactive prompts where dynamic names or user text can break legacy Markdown parsing. Add regression coverage for reload-mcp, model picker, approval callbacks, and update prompts.

tests/gateway/test_telegram_approval_buttons.py

@@ -195,6 +195,29 @@ class TestTelegramExecApproval:
             or kwargs.get("link_preview_options") is not None
         )
 
+    @pytest.mark.asyncio
+    async def test_send_update_prompt_escapes_dynamic_prompt(self):
+        adapter = _make_adapter()
+        sent = {}
+
+        async def mock_send_message(**kwargs):
+            sent.update(kwargs)
+            return SimpleNamespace(message_id=55)
+
+        adapter._bot.send_message = AsyncMock(side_effect=mock_send_message)
+
+        result = await adapter.send_update_prompt(
+            chat_id="12345",
+            prompt="Fix [issue]_1 and verify *markdown*",
+            default="alpha_beta",
+            metadata={"thread_id": "999"},
+        )
+
+        assert result.success is True
+        assert "MARKDOWN_V2" in repr(sent["parse_mode"])
+        assert "Fix \\[issue\\]\\_1" in sent["text"]
+        assert "alpha\\_beta" in sent["text"]
+
     @pytest.mark.asyncio
     async def test_truncates_long_command(self):
         adapter = _make_adapter()
@@ -210,9 +233,6 @@ class TestTelegramExecApproval:
         kwargs = adapter._bot.send_message.call_args[1]
         assert "..." in kwargs["text"]
         assert len(kwargs["text"]) < 5000
-
-
-# ===========================================================================
 # _handle_callback_query — approval button clicks
 # ===========================================================================
 
@@ -251,6 +271,34 @@ class TestTelegramApprovalCallback:
         # State should be cleaned up
         assert 1 not in adapter._approval_state
 
+    @pytest.mark.asyncio
+    async def test_approval_callback_escapes_dynamic_user_name(self):
+        adapter = _make_adapter()
+        adapter._approval_state[3] = "agent:main:telegram:group:12345:99"
+
+        query = AsyncMock()
+        query.data = "ea:once:3"
+        query.message = MagicMock()
+        query.message.chat_id = 12345
+        query.from_user = MagicMock()
+        query.from_user.first_name = "Alice_Bob"
+        query.answer = AsyncMock()
+        query.edit_message_text = AsyncMock()
+
+        update = MagicMock()
+        update.callback_query = query
+        context = MagicMock()
+        query.from_user.id = "12345"
+
+        with patch.dict(os.environ, {"TELEGRAM_ALLOWED_USERS": "*"}, clear=False):
+            with patch("tools.approval.resolve_gateway_approval", return_value=1):
+                await adapter._handle_callback_query(update, context)
+
+        edit_kwargs = query.edit_message_text.call_args[1]
+        assert "MARKDOWN_V2" in repr(edit_kwargs["parse_mode"])
+        assert "Alice\\_Bob" in edit_kwargs["text"]
+        assert "Approved once" in edit_kwargs["text"]
+
     @pytest.mark.asyncio
     async def test_deny_button(self):
         adapter = _make_adapter()

tests/gateway/test_telegram_format.py

@@ -210,6 +210,19 @@ class TestFormatMessageBoldItalic:
         assert "*bold*" in result
         assert "_italic_" in result
 
+    def test_reload_mcp_summary_escapes_dynamic_server_names(self, adapter):
+        content = (
+            "🔄 **MCP Servers Reloaded**\n"
+            "♻️ Reconnected: agent_one, tool[beta]\n"
+            "➕ Added: alpha*prod\n"
+            "🔧 3 tool(s) available from 2 server(s)"
+        )
+        result = adapter.format_message(content)
+        assert "*MCP Servers Reloaded*" in result
+        assert "agent\\_one" in result
+        assert "tool\\[beta\\]" in result
+        assert "alpha\\*prod" in result
+
 
 # =========================================================================
 # format_message - headers

tests/gateway/test_telegram_model_picker.py

@@ -43,6 +43,65 @@ def _make_adapter():
 
 
 class TestTelegramModelPicker:
+    @pytest.mark.asyncio
+    async def test_send_model_picker_escapes_dynamic_provider_label(self):
+        adapter = _make_adapter()
+        sent = {}
+
+        async def mock_send_message(**kwargs):
+            sent.update(kwargs)
+            return SimpleNamespace(message_id=101)
+
+        adapter._bot.send_message = AsyncMock(side_effect=mock_send_message)
+
+        result = await adapter.send_model_picker(
+            chat_id="12345",
+            providers=[
+                {"slug": "provider_one", "name": "Provider One", "total_models": 1, "is_current": True}
+            ],
+            current_model="model_1",
+            current_provider="provider_one",
+            session_key="s",
+            on_model_selected=AsyncMock(),
+            metadata={"thread_id": "99999"},
+        )
+
+        assert result.success is True
+        assert "MARKDOWN_V2" in repr(sent["parse_mode"])
+        assert "provider\\_one" in sent["text"]
+        assert "`model_1`" in sent["text"]
+
+    @pytest.mark.asyncio
+    async def test_back_button_escapes_dynamic_provider_label(self):
+        adapter = _make_adapter()
+        adapter._model_picker_state["12345"] = {
+            "providers": [{"slug": "provider_one", "name": "Provider One", "total_models": 1, "is_current": True}],
+            "current_model": "model_1",
+            "current_provider": "provider_one",
+            "session_key": "s",
+            "on_model_selected": AsyncMock(),
+            "msg_id": 42,
+        }
+
+        query = AsyncMock()
+        query.data = "mb"
+        query.message = MagicMock()
+        query.message.chat_id = 12345
+        query.from_user = MagicMock()
+        query.answer = AsyncMock()
+        query.edit_message_text = AsyncMock()
+
+        update = MagicMock()
+        update.callback_query = query
+        context = MagicMock()
+
+        await adapter._handle_model_picker_callback(query, "mb", "12345")
+
+        edit_kwargs = query.edit_message_text.call_args[1]
+        assert "MARKDOWN_V2" in repr(edit_kwargs["parse_mode"])
+        assert "provider\\_one" in edit_kwargs["text"]
+        assert "`model_1`" in edit_kwargs["text"]
+
     @pytest.mark.asyncio
     async def test_retries_without_thread_when_thread_not_found(self):
         adapter = _make_adapter()