feat: compact Telegram PR messages

.github/actions/prepare-telegram-issue-message/prepare.py

@@ -1,10 +1,12 @@
 #!/usr/bin/env python3
 import json
 import os
+import re
 import subprocess
 import uuid
 
 ISSUE_BODY_LIMIT = 500
+MARKDOWN_V2_SPECIAL_CHARACTERS = frozenset("_*[]()~`>#+-=|{}.!\\")
 
 
 def gh_json(*arguments):
@@ -19,15 +21,55 @@ def truncate(value, limit):
     return value[: limit - 3].rstrip() + "..."
 
 
+def escape_markdown(value):
+    return "".join(f"\\{character}" if character in MARKDOWN_V2_SPECIAL_CHARACTERS else character for character in str(value))
+
+
+def escape_link_url(value):
+    return str(value).replace("\\", "\\\\").replace(")", "\\)")
+
+
+def body_excerpt(value, limit=ISSUE_BODY_LIMIT):
+    lines = value.strip().splitlines()
+    while lines and not lines[0].strip():
+        lines.pop(0)
+    if not lines:
+        return ""
+
+    heading = re.fullmatch(r"\s{0,3}#{1,6}\s+(.+?)\s*#*\s*", lines[0])
+    if heading:
+        lines = lines[1:]
+        while lines and not lines[0].strip():
+            lines.pop(0)
+        paragraph = []
+        for line in lines:
+            if not line.strip():
+                break
+            paragraph.append(line.strip())
+        parts = [heading.group(1), " ".join(paragraph)]
+        return truncate("\n".join(part for part in parts if part), limit)
+
+    paragraph = []
+    for line in lines:
+        if not line.strip():
+            break
+        paragraph.append(line.strip())
+    return truncate(" ".join(paragraph), limit)
+
+
 def format_closed(repository, issue, actor):
-    lines = [
-        f"🟢 {repository} — issue closed",
-        f"#{issue['number']} {issue['title']}",
+    label = escape_markdown(f"Issue #{issue['number']}")
+    link = f"[{label}]({escape_link_url(issue['url'])})"
+    parts = [
+        escape_markdown(repository),
+        f"{link} closed",
+        f"*{escape_markdown(issue['title'])}*",
+        escape_markdown(actor),
     ]
-    if body := truncate(issue.get("body") or "", ISSUE_BODY_LIMIT):
-        lines.append(body)
-    lines.append(f"{actor} · {issue['url']}")
-    return "\n".join(lines)
+    message = " · ".join(parts)
+    if excerpt := body_excerpt(issue.get("body") or ""):
+        message += f"\n{escape_markdown(excerpt)}"
+    return message
 
 
 def write_output(message):

.github/actions/prepare-telegram-pr-message/prepare.py

@@ -6,6 +6,7 @@
 from datetime import datetime, timezone
 
 DIGEST_PR_LIMIT = 10
+MARKDOWN_V2_SPECIAL_CHARACTERS = frozenset("_*[]()~`>#+-=|{}.!\\")
 TELEGRAM_MESSAGE_LIMIT = 4096
 
 
@@ -18,27 +19,39 @@ def author_login(pull_request):
     return (pull_request.get("author") or {}).get("login") or "ghost"
 
 
+def escape_markdown(value):
+    return "".join(f"\\{character}" if character in MARKDOWN_V2_SPECIAL_CHARACTERS else character for character in str(value))
+
+
+def escape_link_url(value):
+    return str(value).replace("\\", "\\\\").replace(")", "\\)")
+
+
+def pull_request_link(pull_request):
+    label = escape_markdown(f"PR #{pull_request['number']}")
+    url = escape_link_url(pull_request["url"])
+    return f"[{label}]({url})"
+
+
 def format_opened(repository, pull_request, event_action):
     if pull_request.get("isDraft"):
         return ""
 
     events = {
-        "ready_for_review": ("🆗", "PR ready for review"),
-        "reopened": ("🆙", "PR reopened"),
+        "ready_for_review": "ready",
+        "reopened": "reopened",
     }
-    emoji, event = events.get(event_action, ("🆕", "PR opened"))
+    event = events.get(event_action, "opened")
     return (
-        f"{emoji} {repository} — {event}\n"
-        f"#{pull_request['number']} {pull_request['title']}\n"
-        f"{author_login(pull_request)} · {pull_request['url']}"
+        f"{escape_markdown(repository)} · {pull_request_link(pull_request)} {event} · "
+        f"*{escape_markdown(pull_request['title'])}* · {escape_markdown(author_login(pull_request))}"
     )
 
 
 def format_merged(repository, pull_request):
     return (
-        f"🔀 {repository} — PR merged\n"
-        f"#{pull_request['number']} {pull_request['title']}\n"
-        f"{author_login(pull_request)} · {pull_request['url']}"
+        f"{escape_markdown(repository)} · {pull_request_link(pull_request)} merged · "
+        f"*{escape_markdown(pull_request['title'])}* · {escape_markdown(author_login(pull_request))}"
     )
 
 
@@ -52,24 +65,29 @@ def format_digest(repository, pull_requests, now=None):
         return ""
 
     now = now or datetime.now(timezone.utc)
-    header = f"🔠 {repository} — {len(pull_requests)} open PR(s)"
+    count = len(pull_requests)
+    header = f"{escape_markdown(repository)} · {count} open {'PR' if count == 1 else 'PRs'}"
     lines = []
     for pull_request in pull_requests[:DIGEST_PR_LIMIT]:
         age = max(0, (now - parse_github_time(pull_request["createdAt"])).days)
         line = (
-            f"#{pull_request['number']} {pull_request['title']}\n"
-            f"{author_login(pull_request)} · {age}d · {pull_request['url']}"
+            f"{pull_request_link(pull_request)} *{escape_markdown(pull_request['title'])}* · "
+            f"{escape_markdown(author_login(pull_request))} · {age}d"
         )
         remaining = len(pull_requests) - len(lines) - 1
-        suffix = f"...and {remaining} more · https://github.com/{repository}/pulls" if remaining else ""
+        suffix = (
+            f"[{escape_markdown(f'...and {remaining} more')}](https://github.com/{escape_link_url(repository)}/pulls)"
+            if remaining
+            else ""
+        )
         candidate = "\n".join((header, *lines, line, suffix)).rstrip()
         if len(candidate) > TELEGRAM_MESSAGE_LIMIT:
             break
         lines.append(line)
 
     remaining = len(pull_requests) - len(lines)
     if remaining:
-        lines.append(f"...and {remaining} more · https://github.com/{repository}/pulls")
+        lines.append(f"[{escape_markdown(f'...and {remaining} more')}](https://github.com/{escape_link_url(repository)}/pulls)")
 
     return "\n".join((header, *lines))
 

.github/actions/send-telegram-message/action.yml

@@ -1,5 +1,5 @@
 name: Send Telegram message
-description: Send a plain-text message to a Telegram chat.
+description: Send a message to a Telegram chat.
 inputs:
   message:
     description: Message text.
@@ -10,6 +10,9 @@ inputs:
   telegram-chat-id:
     description: Telegram chat ID.
     required: true
+  parse-mode:
+    description: Optional Telegram parse mode.
+    default: ""
 runs:
   using: composite
   steps:
@@ -18,6 +21,7 @@ runs:
         MESSAGE: ${{ inputs.message }}
         BOT_TOKEN: ${{ inputs.telegram-bot-token }}
         CHAT_ID: ${{ inputs.telegram-chat-id }}
+        PARSE_MODE: ${{ inputs.parse-mode }}
       run: |
         set -euo pipefail
         curl -fsSL \
@@ -28,4 +32,6 @@ runs:
           -d "$(jq -n \
             --arg c "$CHAT_ID" \
             --arg t "$MESSAGE" \
-            '{chat_id: $c, text: $t, link_preview_options: {is_disabled: true}}')"
+            --arg p "$PARSE_MODE" \
+            '{chat_id: $c, text: $t, link_preview_options: {is_disabled: true}}
+            + if $p == "" then {} else {parse_mode: $p} end')"

.github/workflows/notify-telegram-issue-shared.yml

@@ -23,3 +23,4 @@ jobs:
           message: ${{ steps.prepare.outputs.message }}
           telegram-bot-token: ${{ secrets.TELEGRAM_BOT_TOKEN }}
           telegram-chat-id: ${{ secrets.TELEGRAM_CHAT_ID }}
+          parse-mode: MarkdownV2

.github/workflows/notify-telegram-pr-shared.yml

@@ -27,3 +27,4 @@ jobs:
           message: ${{ steps.prepare.outputs.message }}
           telegram-bot-token: ${{ secrets.TELEGRAM_BOT_TOKEN }}
           telegram-chat-id: ${{ secrets.TELEGRAM_CHAT_ID }}
+          parse-mode: MarkdownV2

test/test_telegram_notifications.py

@@ -38,14 +38,13 @@ def test_formats_opened_pull_request(self):
 
         self.assertEqual(
             message,
-            "🆕 owner/repo — PR opened\n"
-            "#42 feat: add notifications\n"
-            "octocat · https://github.com/owner/repo/pull/42",
+            "owner/repo · [PR \\#42](https://github.com/owner/repo/pull/42) opened · "
+            "*feat: add notifications* · octocat",
         )
 
     def test_formats_ready_and_reopened_pull_request_actions(self):
-        self.assertIn("🆗 owner/repo — PR ready for review", PR_TELEGRAM.format_opened("owner/repo", self.pull_request, "ready_for_review"))
-        self.assertIn("🆙 owner/repo — PR reopened", PR_TELEGRAM.format_opened("owner/repo", self.pull_request, "reopened"))
+        self.assertIn("[PR \\#42](https://github.com/owner/repo/pull/42) ready", PR_TELEGRAM.format_opened("owner/repo", self.pull_request, "ready_for_review"))
+        self.assertIn("[PR \\#42](https://github.com/owner/repo/pull/42) reopened", PR_TELEGRAM.format_opened("owner/repo", self.pull_request, "reopened"))
 
     def test_skips_draft_pull_request(self):
         self.pull_request["isDraft"] = True
@@ -55,8 +54,20 @@ def test_skips_draft_pull_request(self):
     def test_formats_merged_pull_request(self):
         message = PR_TELEGRAM.format_merged("owner/repo", self.pull_request)
 
-        self.assertIn("🔀 owner/repo — PR merged", message)
-        self.assertIn("#42 feat: add notifications", message)
+        self.assertEqual(
+            message,
+            "owner/repo · [PR \\#42](https://github.com/owner/repo/pull/42) merged · "
+            "*feat: add notifications* · octocat",
+        )
+
+    def test_escapes_pull_request_values_for_telegram_markdown(self):
+        self.pull_request["title"] = "fix: escape [markup] & text!"
+        self.pull_request["author"] = {"login": "dependabot[bot]"}
+
+        message = PR_TELEGRAM.format_opened("owner/repo", self.pull_request, "opened")
+
+        self.assertIn(r"*fix: escape \[markup\] & text\!*", message)
+        self.assertIn(r"dependabot\[bot\]", message)
 
     def test_open_pull_request_digest_excludes_drafts_and_reports_age(self):
         draft = dict(self.pull_request, number=41, isDraft=True)
@@ -66,8 +77,8 @@ def test_open_pull_request_digest_excludes_drafts_and_reports_age(self):
             now=datetime(2026, 8, 12, 12, tzinfo=timezone.utc),
         )
 
-        self.assertIn("1 open PR(s)", message)
-        self.assertTrue(message.startswith("🔠 owner/repo"))
+        self.assertIn("1 open PR", message)
+        self.assertTrue(message.startswith("owner/repo"))
         self.assertIn("octocat · 2d", message)
         self.assertNotIn("#41", message)
         self.assertNotIn("more", message)
@@ -80,17 +91,18 @@ def test_open_pull_request_digest_lists_at_most_ten_pull_requests(self):
             self.pull_request
             | {
                 "number": number,
+                "url": f"https://github.com/owner/repo/pull/{number}",
                 "createdAt": "2026-08-10T12:00:00Z",
             }
             for number in range(1, 13)
         ]
 
         message = PR_TELEGRAM.format_digest("owner/repo", pull_requests)
 
-        self.assertIn("12 open PR(s)", message)
-        self.assertIn("#10 feat: add notifications", message)
-        self.assertNotIn("#11 feat: add notifications", message)
-        self.assertIn("...and 2 more · https://github.com/owner/repo/pulls", message)
+        self.assertIn("12 open PRs", message)
+        self.assertIn(r"[PR \#10](https://github.com/owner/repo/pull/10) *feat: add notifications*", message)
+        self.assertNotIn(r"[PR \#11](https://github.com/owner/repo/pull/11) *feat: add notifications*", message)
+        self.assertIn(r"[\.\.\.and 2 more](https://github.com/owner/repo/pulls)", message)
 
     def test_open_pull_request_digest_stays_within_telegram_limit(self):
         pull_requests = [
@@ -106,7 +118,7 @@ def test_open_pull_request_digest_stays_within_telegram_limit(self):
         message = PR_TELEGRAM.format_digest("owner/repo", pull_requests)
 
         self.assertLessEqual(len(message), PR_TELEGRAM.TELEGRAM_MESSAGE_LIMIT)
-        self.assertIn("more · https://github.com/owner/repo/pulls", message)
+        self.assertIn("more]", message)
 
     def test_writes_multiline_github_output(self):
         with tempfile.NamedTemporaryFile() as output, patch.dict(os.environ, {"GITHUB_OUTPUT": output.name}):
@@ -128,7 +140,11 @@ def test_closed_issue_includes_body(self):
 
         message = ISSUE_TELEGRAM.format_closed("owner/repo", issue, "octocat")
 
-        self.assertIn("#12 Move notifications\nResolution summary.\noctocat ·", message)
+        self.assertEqual(
+            message,
+            "owner/repo · [Issue \\#12](https://github.com/owner/repo/issues/12) closed · "
+            "*Move notifications* · octocat\nResolution summary\\.",
+        )
 
     def test_closed_issue_truncates_long_body(self):
         issue = {
@@ -139,10 +155,38 @@ def test_closed_issue_truncates_long_body(self):
         }
 
         message = ISSUE_TELEGRAM.format_closed("owner/repo", issue, "octocat")
-        body = message.splitlines()[2]
+        body = ISSUE_TELEGRAM.truncate(issue["body"], ISSUE_TELEGRAM.ISSUE_BODY_LIMIT)
 
         self.assertEqual(len(body), ISSUE_TELEGRAM.ISSUE_BODY_LIMIT)
         self.assertTrue(body.endswith("..."))
+        self.assertTrue(message.endswith(r"\.\.\."))
+
+    def test_closed_issue_uses_only_first_body_paragraph(self):
+        issue = {
+            "number": 12,
+            "title": "Fix [alerts]",
+            "body": "First line.\nContinued *line*.\n\nIgnored paragraph.",
+            "url": "https://github.com/owner/repo/issues/12",
+        }
+
+        message = ISSUE_TELEGRAM.format_closed("owner/repo", issue, "dependabot[bot]")
+
+        self.assertIn(r"*Fix \[alerts\]*", message)
+        self.assertTrue(message.endswith("dependabot\\[bot\\]\nFirst line\\. Continued \\*line\\*\\."))
+        self.assertNotIn("Ignored", message)
+
+    def test_closed_issue_includes_leading_heading_and_first_paragraph(self):
+        issue = {
+            "number": 12,
+            "title": "Move notifications",
+            "body": "## Resolution\n\nFirst line.\nContinued line.\n\nIgnored paragraph.",
+            "url": "https://github.com/owner/repo/issues/12",
+        }
+
+        message = ISSUE_TELEGRAM.format_closed("owner/repo", issue, "octocat")
+
+        self.assertTrue(message.endswith("\nResolution\nFirst line\\. Continued line\\."))
+        self.assertNotIn("Ignored", message)
 
 
 class TelegramWorkflowTest(unittest.TestCase):
@@ -166,6 +210,7 @@ def test_pr_workflow_prepares_and_sends_supported_events(self):
                 "$/.github/actions/send-telegram-message",
             ],
         )
+        self.assertEqual(steps[-1]["with"]["parse-mode"], "MarkdownV2")
 
     def test_issue_workflow_only_sends_closed_issue_notification(self):
         workflow = self.load_workflow("notify-telegram-issue-shared.yml")
@@ -181,6 +226,7 @@ def test_issue_workflow_only_sends_closed_issue_notification(self):
                 "$/.github/actions/send-telegram-message",
             ],
         )
+        self.assertEqual(job["steps"][-1]["with"]["parse-mode"], "MarkdownV2")
 
     def test_baseline_caller_passes_only_explicit_telegram_secrets(self):
         path = BASELINE_ROOT / ".github" / "workflows" / "notify-telegram-pr.yml"
@@ -207,6 +253,8 @@ def test_internal_actions_separate_formatting_and_delivery(self):
         self.assertNotIn("api.telegram.org", pr_action)
         self.assertNotIn("api.telegram.org", issue_action)
         self.assertIn("inputs:\n  message:", send_action)
+        self.assertIn("parse-mode:", send_action)
+        self.assertIn("parse_mode", send_action)
         self.assertIn("--output /dev/null", send_action)
         self.assertIn("api.telegram.org", send_action)