feat(send): [[plain]] directive + --plain flag for unformatted sends

hermes_cli/send_cmd.py

@@ -384,6 +384,7 @@ def cmd_send(args: argparse.Namespace) -> None:
         "action": "send",
         "target": target,
         "message": message,
+        "plain": getattr(args, "plain", False),
     }
 
     result = send_message_tool(tool_args)
@@ -470,6 +471,16 @@ def register_send_subparser(subparsers) -> argparse.ArgumentParser:
         help="Prepend a subject/header line before the message body.",
     )
 
+    parser.add_argument(
+        "-p",
+        "--plain",
+        action="store_true",
+        default=False,
+        help="Send as plain text (Telegram): no HTML/MarkdownV2 parsing. Use for "
+        "status/alert text containing <placeholders>, commit messages, or markdown "
+        "metacharacters that the formatter would mangle. Emojis and bare URLs still render.",
+    )
+
     parser.add_argument(
         "-l",
         "--list",

tests/hermes_cli/test_send_cmd.py

@@ -64,6 +64,49 @@ def fake_tool(monkeypatch):
 # ---------------------------------------------------------------------------
 
 
+def test_plain_flag_sets_plain_true_in_tool_args(fake_tool):
+    """--plain must arrive in the send_message_tool arg dict as plain=True,
+    which is what drives parse_mode=None on the Telegram side."""
+    args = _parse(["--to", "telegram", "--plain", "status: <ok>"])
+    with pytest.raises(SystemExit) as exc:
+        send_cmd.cmd_send(args)
+    assert exc.value.code == 0
+    assert len(fake_tool.calls) == 1
+    call = fake_tool.calls[0]
+    assert call["action"] == "send"
+    assert call["target"] == "telegram"
+    assert call["message"] == "status: <ok>"
+    # The literal placeholder is passed through verbatim by the CLI.
+    assert call["plain"] is True
+
+
+def test_plain_short_flag_sets_plain_true(fake_tool):
+    """The -p short form is equivalent to --plain."""
+    args = _parse(["--to", "telegram", "-p", "hello"])
+    with pytest.raises(SystemExit):
+        send_cmd.cmd_send(args)
+    assert fake_tool.calls[0]["plain"] is True
+
+
+def test_default_send_has_plain_false(fake_tool):
+    """Without --plain the arg dict carries plain=False (anti-tautology:
+    proves the True cases above are driven by the flag, not a constant)."""
+    args = _parse(["--to", "telegram", "hello world"])
+    with pytest.raises(SystemExit) as exc:
+        send_cmd.cmd_send(args)
+    assert exc.value.code == 0
+    assert fake_tool.calls[0]["plain"] is False
+
+
+def test_plain_flag_default_is_false_on_parser():
+    """The parser default for the flag itself is False."""
+    args = _parse(["--to", "telegram", "hello"])
+    assert args.plain is False
+    args_plain = _parse(["--to", "telegram", "--plain", "hello"])
+    assert args_plain.plain is True
+
+
+
 
 
 

tests/tools/test_send_message_target_parse.py

@@ -62,6 +62,7 @@ def test_send_message_routes_whatsapp_group_jid_without_home_fallback() -> None:
         thread_id=None,
         media_files=[],
         force_document=False,
+        force_plain=False,
     )
 
 
@@ -116,6 +117,7 @@ def test_resolved_opaque_plugin_target_uses_directory_id() -> None:
         thread_id=None,
         media_files=[],
         force_document=False,
+        force_plain=False,
     )
 
 

tests/tools/test_send_message_tool.py

@@ -315,6 +315,7 @@ def test_ntfy_topic_target_bypasses_channel_directory(self):
             thread_id=None,
             media_files=[],
             force_document=False,
+            force_plain=False,
         )
 
 
@@ -354,6 +355,108 @@ def test_media_tag_outside_allowed_roots_is_not_sent(self, tmp_path, monkeypatch
             thread_id=None,
             media_files=[],
             force_document=False,
+            force_plain=False,
+        )
+
+    def test_plain_directive_strips_and_forces_plain(self):
+        """[[plain]] prefix is stripped from the message and force_plain=True
+        is threaded through to _send_to_platform (angle-bracket text intact)."""
+        config, telegram_cfg = _make_config()
+
+        with patch("gateway.config.load_gateway_config", return_value=config), \
+             patch("tools.interrupt.is_interrupted", return_value=False), \
+             patch("model_tools._run_async", side_effect=_run_async_immediately), \
+             patch("tools.send_message_tool._send_to_platform", new=AsyncMock(return_value={"success": True})) as send_mock, \
+             patch("gateway.mirror.mirror_to_session", return_value=True):
+            result = json.loads(
+                send_message_tool(
+                    {
+                        "action": "send",
+                        "target": "telegram:12345",
+                        "message": "[[plain]]Deploy of <service> failed",
+                    }
+                )
+            )
+
+        assert result["success"] is True
+        send_mock.assert_awaited_once_with(
+            Platform.TELEGRAM,
+            telegram_cfg,
+            "12345",
+            # The [[plain]] directive is stripped; the literal <service>
+            # placeholder survives untouched in the forwarded message.
+            "Deploy of <service> failed",
+            thread_id=None,
+            media_files=[],
+            force_document=False,
+            force_plain=True,
+        )
+
+    def test_plain_arg_forces_plain_without_directive(self):
+        """The CLI --plain flag arrives as args['plain']=True and forces
+        force_plain=True even with no [[plain]] directive in the text."""
+        config, telegram_cfg = _make_config()
+
+        with patch("gateway.config.load_gateway_config", return_value=config), \
+             patch("tools.interrupt.is_interrupted", return_value=False), \
+             patch("model_tools._run_async", side_effect=_run_async_immediately), \
+             patch("tools.send_message_tool._send_to_platform", new=AsyncMock(return_value={"success": True})) as send_mock, \
+             patch("gateway.mirror.mirror_to_session", return_value=True):
+            result = json.loads(
+                send_message_tool(
+                    {
+                        "action": "send",
+                        "target": "telegram:12345",
+                        "message": "status: <ok>",
+                        "plain": True,
+                    }
+                )
+            )
+
+        assert result["success"] is True
+        send_mock.assert_awaited_once_with(
+            Platform.TELEGRAM,
+            telegram_cfg,
+            "12345",
+            "status: <ok>",
+            thread_id=None,
+            media_files=[],
+            force_document=False,
+            force_plain=True,
+        )
+
+    def test_no_plain_directive_leaves_force_plain_false(self):
+        """Default path: no directive, no flag => force_plain=False and the
+        message text is unchanged (regression guard against over-stripping)."""
+        config, telegram_cfg = _make_config()
+
+        with patch("gateway.config.load_gateway_config", return_value=config), \
+             patch("tools.interrupt.is_interrupted", return_value=False), \
+             patch("model_tools._run_async", side_effect=_run_async_immediately), \
+             patch("tools.send_message_tool._send_to_platform", new=AsyncMock(return_value={"success": True})) as send_mock, \
+             patch("gateway.mirror.mirror_to_session", return_value=True):
+            result = json.loads(
+                send_message_tool(
+                    {
+                        "action": "send",
+                        "target": "telegram:12345",
+                        "message": "plain [[note]] body",
+                    }
+                )
+            )
+
+        assert result["success"] is True
+        send_mock.assert_awaited_once_with(
+            Platform.TELEGRAM,
+            telegram_cfg,
+            "12345",
+            # A lookalike [[note]] token is NOT the [[plain]] directive and is
+            # left in place; only [[plain]] triggers stripping.
+            "plain [[note]] body",
+            thread_id=None,
+            media_files=[],
+            force_document=False,
+            force_plain=False,
         )
 
     def test_top_level_send_failure_redacts_query_token(self):
@@ -779,6 +882,57 @@ def test_html_message_uses_html_parse_mode(self, monkeypatch):
         assert kwargs["parse_mode"] == "HTML"
         assert kwargs["text"] == "<b>Hello</b> world"
 
+    def test_force_plain_sends_with_no_parse_mode(self, monkeypatch):
+        """force_plain=True => parse_mode=None so nothing is interpreted."""
+        bot = self._make_bot()
+        _install_telegram_mock(monkeypatch, bot)
+
+        asyncio.run(
+            _send_telegram("tok", "123", "*not* _markdown_", force_plain=True)
+        )
+
+        bot.send_message.assert_awaited_once()
+        kwargs = bot.send_message.await_args.kwargs
+        # The actual parse_mode passed to the Bot API is None (not HTML/MarkdownV2).
+        assert kwargs["parse_mode"] is None
+        # Text is sent verbatim: no MarkdownV2 escaping of * or _.
+        assert kwargs["text"] == "*not* _markdown_"
+
+    def test_force_plain_sends_angle_bracket_placeholder_verbatim(self, monkeypatch):
+        """A literal <placeholder> is NOT treated as HTML when force_plain."""
+        bot = self._make_bot()
+        _install_telegram_mock(monkeypatch, bot)
+
+        raw = "Deploy of <service> to <region> failed: <error-code>"
+        asyncio.run(_send_telegram("tok", "123", raw, force_plain=True))
+
+        bot.send_message.assert_awaited_once()
+        kwargs = bot.send_message.await_args.kwargs
+        # Critically: even though the text contains angle-bracket tokens that
+        # the HTML auto-detector would otherwise match, force_plain wins and
+        # parse_mode is None, so Telegram receives the raw, unescaped text.
+        assert kwargs["parse_mode"] is None
+        assert kwargs["text"] == raw
+
+    def test_force_plain_beats_html_autodetection(self, monkeypatch):
+        """force_plain overrides the <tag> HTML heuristic (anti-tautology).
+
+        Without force_plain the same text takes the HTML branch (parse_mode
+        HTML); with it, parse_mode must be None. Asserting both directions
+        proves the flag actually drives the parse_mode, not a constant.
+        """
+        bot = self._make_bot()
+        _install_telegram_mock(monkeypatch, bot)
+
+        text = "status: <ok>"
+
+        asyncio.run(_send_telegram("tok", "123", text, force_plain=False))
+        assert bot.send_message.await_args.kwargs["parse_mode"] == "HTML"
+
+        bot.send_message.reset_mock()
+        asyncio.run(_send_telegram("tok", "123", text, force_plain=True))
+        assert bot.send_message.await_args.kwargs["parse_mode"] is None
+
 
     def test_transient_bad_gateway_retries_text_send(self, monkeypatch):
         bot = self._make_bot()

tools/send_message_tool.py

@@ -435,6 +435,15 @@ def _handle_send(args):
     # JPGs where Telegram's sendPhoto recompresses to 1280px).
     force_document_attachments = "[[as_document]]" in message
 
+    # Capture [[plain]] directive: send with NO parse_mode (plain text), bypassing
+    # both HTML auto-detection and MarkdownV2 conversion. Used by status notifiers
+    # (cron heartbeats, escalation alerts) whose dynamic content legitimately
+    # contains <placeholder> tokens, commit messages, and markdown metacharacters
+    # that the HTML/MarkdownV2 parsers misread and mangle. Stripped before send.
+    force_plain = args.get("plain", False) or "[[plain]]" in message
+    if "[[plain]]" in message:
+        message = message.replace("[[plain]]", "").strip()
+
     media_files, cleaned_message = BasePlatformAdapter.extract_media(message)
     media_files = BasePlatformAdapter.filter_media_delivery_paths(media_files)
     mirror_text = cleaned_message.strip() or _describe_media_for_mirror(media_files)
@@ -487,6 +496,7 @@ def _handle_send(args):
             "thread_id": thread_id,
             "media_files": media_files,
             "force_document": force_document_attachments,
+            "force_plain": force_plain,
         }
         # Preserve the exact built-in call contract; only custom handlers need
         # the complete typed request.
@@ -922,7 +932,7 @@ async def _send_via_adapter(
     }
 
 
-async def _send_to_platform(platform, pconfig, chat_id, message, thread_id=None, media_files=None, force_document=False, args=None):
+async def _send_to_platform(platform, pconfig, chat_id, message, thread_id=None, media_files=None, force_document=False, force_plain=False, args=None):
     """Route a message to the appropriate platform sender.
 
     Long messages are automatically chunked to fit within platform limits
@@ -1016,6 +1026,7 @@ async def _send_to_platform(platform, pconfig, chat_id, message, thread_id=None,
             thread_id=thread_id,
             disable_link_previews=disable_link_previews,
             force_document=force_document,
+            force_plain=force_plain,
         )
 
     # --- Discord: chunked delivery via the registry's standalone_sender_fn.
@@ -1343,7 +1354,7 @@ def _is_telegram_thread_not_found(error: Exception) -> bool:
     return "thread not found" in str(error).lower()
 
 
-async def _send_telegram(token, chat_id, message, media_files=None, thread_id=None, disable_link_previews=False, force_document=False):
+async def _send_telegram(token, chat_id, message, media_files=None, thread_id=None, disable_link_previews=False, force_document=False, force_plain=False):
     """Send via Telegram Bot API (one-shot, no polling needed).
 
     Applies markdown→MarkdownV2 formatting (same as the gateway adapter)
@@ -1359,7 +1370,16 @@ async def _send_telegram(token, chat_id, message, media_files=None, thread_id=No
         # Inspired by github.com/ashaney — PR #1568.
         _has_html = bool(re.search(r'<[a-zA-Z/][^>]*>', message))
 
-        if _has_html:
+        if force_plain:
+            # Explicit plain-text: no parse_mode at all. Bypasses both HTML
+            # auto-detection (which false-positives on literal <placeholder>
+            # tokens like <x>/<file>) and MarkdownV2 escaping. Emojis and bare
+            # URLs still render/auto-link; only formatting syntax is inert.
+            # Used by status notifiers via [[plain]] / args["plain"].
+            formatted = message
+            send_parse_mode = None
+            _has_html = False
+        elif _has_html:
             formatted = message
             send_parse_mode = ParseMode.HTML
         else: