From 626141a9bd136765f8a2cb73fd1f4e46de386340 Mon Sep 17 00:00:00 2001 From: patrick-chinchill Date: Fri, 19 Jun 2026 18:24:25 -0700 Subject: [PATCH 1/2] =?UTF-8?q?feat(telegram):=20rich.py=20renderer=20for?= =?UTF-8?q?=20Bot=20API=2010.1=20(chat@4.31=204662309=20=E2=80=94=20TG2/4)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Port packages/adapter-telegram/src/rich.ts character-for-character: the rich-message -> Markdown / plain-text renderer and media extractor for the Bot API 10.1 rich-message wire types introduced in TG1. - TELEGRAM_RICH_MESSAGE_LIMIT = 32768 - truncate_rich_markdown: code-point-aware (list(str) == Array.from), reuses StreamingMarkdownRenderer.push/finish so the truncated prefix stays valid markdown; never splits a non-BMP code point. - escape_text: MARKDOWN_PUNCTUATION /[!-/:-@[-`{-~]/ backslash-escape pass. - inline/code fence run-sizing: max(1, longest_run+1) / max(3, longest_run+1). - recursive text()/plain()/block()/plain_block()/media() walkers over every TelegramRichText / TelegramRichBlock variant, via match statements that mirror the upstream switch and let pyrefly narrow the recursive union. Tests port the rich.test.ts it() blocks plus adversarial coverage: a full punctuation escape sweep (in-class escaped, out-of-class untouched), all- backtick fence growth, and a surrogate-pair truncation-boundary check. Each fails on a plausible mutation. Test fixtures annotated to concrete TG1 variant TypedDicts; pyrefly 0 on src and the test file. --- src/chat_sdk/adapters/telegram/rich.py | 418 +++++++++++++++++++++++++ tests/test_telegram_rich.py | 349 +++++++++++++++++++++ 2 files changed, 767 insertions(+) create mode 100644 src/chat_sdk/adapters/telegram/rich.py create mode 100644 tests/test_telegram_rich.py diff --git a/src/chat_sdk/adapters/telegram/rich.py b/src/chat_sdk/adapters/telegram/rich.py new file mode 100644 index 0000000..03a99b6 --- /dev/null +++ b/src/chat_sdk/adapters/telegram/rich.py @@ -0,0 +1,418 @@ +"""Telegram rich-message → Markdown / plain-text rendering (Bot API 10.1). + +Python port of ``packages/adapter-telegram/src/rich.ts`` (chat@4.31.0, +commit 4662309). Regex/parse-heavy module ported character-for-character. + +The rich-message wire types live in :mod:`chat_sdk.adapters.telegram.types` +(TG1). This module converts the recursive ``TelegramRichBlock`` / +``TelegramRichText`` shapes into Markdown (for sending) and plain text (for +search/snippets), and extracts media attachments. +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass +from typing import Literal + +from chat_sdk.shared.streaming_markdown import StreamingMarkdownRenderer + +from .types import ( + TelegramFile, + TelegramRichBlock, + TelegramRichBlockTable, + TelegramRichCaption, + TelegramRichCell, + TelegramRichItem, + TelegramRichMessage, + TelegramRichText, +) + +TELEGRAM_RICH_MESSAGE_LIMIT = 32_768 + +# Upstream: /[!-/:-@[-`{-~]/g -- the four ASCII punctuation ranges +# "!"-"/" ":"-"@" "["-"`" "{"-"~" +# A char class, so re.sub matches a single punctuation char at a time. +MARKDOWN_PUNCTUATION = re.compile(r"[!-/:-@[-`{-~]") +# Upstream: /[\r\n]/g +LINE_BREAKS = re.compile(r"[\r\n]") +# Upstream: /`+/g -- runs of one or more backticks. +BACKTICKS = re.compile(r"`+") + + +def truncate_rich_markdown(markdown: str) -> str: + """Truncate *markdown* to the Telegram rich-message character limit. + + Truncation is code-point aware: ``list(str)`` yields one element per + Unicode code point (matching JS ``Array.from(str)``), so a slice on a + list index never splits a surrogate pair / non-BMP character. The + held-back tail is re-rendered through :class:`StreamingMarkdownRenderer` + so the truncated prefix stays valid markdown, with ``"..."`` appended. + """ + characters = list(markdown) + if len(characters) <= TELEGRAM_RICH_MESSAGE_LIMIT: + return markdown + + end = TELEGRAM_RICH_MESSAGE_LIMIT - 3 + while end > 0: + renderer = StreamingMarkdownRenderer() + renderer.push("".join(characters[:end])) + rendered = f"{renderer.finish()}..." + if len(rendered) <= TELEGRAM_RICH_MESSAGE_LIMIT: + return rendered + end -= len(rendered) - TELEGRAM_RICH_MESSAGE_LIMIT + + return "..." + + +def _escape_text(value: str) -> str: + """Backslash-escape every Markdown punctuation char in *value*. + + Upstream: ``value.replace(MARKDOWN_PUNCTUATION, "\\$&")`` -- each matched + punctuation char is prefixed with a single backslash. ``\\g<0>`` is the + Python equivalent of JS ``$&`` (the whole match). + """ + return MARKDOWN_PUNCTUATION.sub(r"\\\g<0>", value) + + +def _inline_code(value: str) -> str: + if not value: + return "" + runs = BACKTICKS.findall(value) + size = max(1, *(len(run) + 1 for run in runs)) if runs else 1 + marker = "`" * size + has_boundary_space = value.startswith(" ") and value.endswith(" ") and len(value.strip()) > 0 + padding = " " if value.startswith("`") or value.endswith("`") or has_boundary_space else "" + return f"{marker}{padding}{value}{padding}{marker}" + + +def _code_block(value: str, language: str | None = None) -> str: + runs = BACKTICKS.findall(value) + size = max(3, *(len(run) + 1 for run in runs)) if runs else 3 + marker = "`" * size + info = LINE_BREAKS.sub(" ", language).replace("`", "") if language else "" + return f"{marker}{info}\n{value}\n{marker}" + + +def _link_destination(value: str) -> str: + return ( + "<" + + value.replace("\\", "%5C").replace("<", "%3C").replace(">", "%3E").replace("\r", "%0D").replace("\n", "%0A") + + ">" + ) + + +def _text(markdown: TelegramRichText) -> str: + if isinstance(markdown, str): + return _escape_text(markdown) + if isinstance(markdown, list): + return "".join(_text(part) for part in markdown) + + match markdown["type"]: + case "bold": + return f"**{_text(markdown['text'])}**" + case "italic": + return f"_{_text(markdown['text'])}_" + case "underline": + return f"{_text(markdown['text'])}" + case "strikethrough": + return f"~~{_text(markdown['text'])}~~" + case "spoiler": + return f"||{_text(markdown['text'])}||" + case "subscript": + return f"{_text(markdown['text'])}" + case "superscript": + return f"{_text(markdown['text'])}" + case "marked": + return f"=={_text(markdown['text'])}==" + case "code": + return _inline_code(_plain(markdown["text"])) + case "date_time" | "text_mention": + return _text(markdown["text"]) + case "bank_card_number" | "bot_command" | "cashtag" | "hashtag" | "mention": + return _text(markdown["text"]) + case "custom_emoji": + return markdown["alternative_text"] + case "mathematical_expression": + return f"${markdown['expression']}$" + case "url": + return f"[{_text(markdown['text'])}]({_link_destination(markdown['url'])})" + case "email_address": + dest = _link_destination(f"mailto:{markdown['email_address']}") + return f"[{_text(markdown['text'])}]({dest})" + case "phone_number": + dest = _link_destination(f"tel:{markdown['phone_number']}") + return f"[{_text(markdown['text'])}]({dest})" + case "anchor": + return "" + case "anchor_link" | "reference" | "reference_link": + return _text(markdown["text"]) + case _: + return "" + + +def _plain(markdown: TelegramRichText) -> str: + if isinstance(markdown, str): + return markdown + if isinstance(markdown, list): + return "".join(_plain(part) for part in markdown) + + match markdown["type"]: + case ( + "bold" + | "italic" + | "underline" + | "strikethrough" + | "spoiler" + | "subscript" + | "superscript" + | "marked" + | "code" + | "date_time" + | "text_mention" + | "url" + | "email_address" + | "phone_number" + | "bank_card_number" + | "mention" + | "hashtag" + | "cashtag" + | "bot_command" + | "anchor_link" + | "reference" + | "reference_link" + ): + return _plain(markdown["text"]) + case "custom_emoji": + return markdown["alternative_text"] + case "mathematical_expression": + return markdown["expression"] + case "anchor": + return "" + case _: + return "" + + +def _caption(value: TelegramRichCaption | None = None) -> str: + if not value: + return "" + credit = f"\n{_text(value['credit'])}" if value.get("credit") else "" + return f"{_text(value['text'])}{credit}" + + +def _plain_caption(value: TelegramRichCaption | None = None) -> str: + if not value: + return "" + credit = f"\n{_plain(value['credit'])}" if value.get("credit") else "" + return f"{_plain(value['text'])}{credit}" + + +def _cell(value: TelegramRichCell) -> str: + return _text(value["text"]) if value.get("text") else "" + + +def _item(value: TelegramRichItem) -> str: + checked = "" + if value.get("has_checkbox"): + checked = "[x] " if value.get("is_checked") else "[ ] " + content = "\n\n".join(_block(b) for b in value["blocks"]).replace("\n", "\n ") + return f"{value['label']} {checked}{content}".rstrip() + + +def _plain_item(value: TelegramRichItem) -> str: + checked = "" + if value.get("has_checkbox"): + checked = "[x] " if value.get("is_checked") else "[ ] " + content = "\n".join(b for b in (_plain_block(blk) for blk in value["blocks"]) if b).replace("\n", "\n ") + return f"{value['label']} {checked}{content}".rstrip() + + +def _table(value: TelegramRichBlockTable) -> str: + rows = [f"| {' | '.join(_cell(c) for c in row)} |" for row in value["cells"]] + if len(rows) == 0: + return _text(value["caption"]) if value.get("caption") else "" + + columns = max(len(row) for row in value["cells"]) + separator = f"| {' | '.join('---' for _ in range(columns))} |" + content = "\n".join([rows[0], separator, *rows[1:]]) + return f"{_text(value['caption'])}\n\n{content}" if value.get("caption") else content + + +def _quote(value: str) -> str: + return "\n".join(f"> {line}" for line in value.split("\n")) + + +def _block(value: TelegramRichBlock) -> str: + match value["type"]: + case "paragraph" | "footer" | "thinking": + return _text(value["text"]) + case "heading": + level = min(6, max(1, value["size"])) + return f"{'#' * level} {_text(value['text'])}" + case "pre": + return _code_block(_plain(value["text"]), value.get("language")) + case "divider": + return "---" + case "mathematical_expression": + return f"$${value['expression']}$$" + case "anchor": + return "" + case "list": + return "\n".join(_item(i) for i in value["items"]) + case "blockquote": + content = "\n\n".join(_block(b) for b in value["blocks"]) + credit = f"\n\n{_text(value['credit'])}" if value.get("credit") else "" + return _quote(f"{content}{credit}") + case "pullquote": + credit = f"\n\n{_text(value['credit'])}" if value.get("credit") else "" + return _quote(f"{_text(value['text'])}{credit}") + case "collage" | "slideshow": + content = "\n\n".join(b for b in (_block(blk) for blk in value["blocks"]) if b) + description = _caption(value.get("caption")) + return "\n\n".join(part for part in (content, description) if part) + case "table": + return _table(value) + case "details": + body = "\n\n".join(_block(b) for b in value["blocks"]) + return f"{_text(value['summary'])}\n\n{body}" + case "map": + return _caption(value.get("caption")) + case "animation" | "audio" | "photo" | "video" | "voice_note": + return _caption(value.get("caption")) + case _: + return "" + + +def _plain_block(value: TelegramRichBlock) -> str: + match value["type"]: + case "paragraph" | "footer" | "thinking" | "heading" | "pre": + return _plain(value["text"]) + case "divider" | "anchor": + return "" + case "mathematical_expression": + return value["expression"] + case "list": + return "\n".join(_plain_item(i) for i in value["items"]) + case "blockquote": + content = "\n\n".join(b for b in (_plain_block(blk) for blk in value["blocks"]) if b) + credit = f"\n\n{_plain(value['credit'])}" if value.get("credit") else "" + return f"{content}{credit}" + case "pullquote": + credit = f"\n\n{_plain(value['credit'])}" if value.get("credit") else "" + return f"{_plain(value['text'])}{credit}" + case "collage" | "slideshow": + content = "\n\n".join(b for b in (_plain_block(blk) for blk in value["blocks"]) if b) + description = _plain_caption(value.get("caption")) + return "\n\n".join(part for part in (content, description) if part) + case "table": + rows = [ + "\t".join(_plain(entry["text"]) if entry.get("text") else "" for entry in row) for row in value["cells"] + ] + content = "\n".join(rows) + return f"{_plain(value['caption'])}\n\n{content}" if value.get("caption") else content + case "details": + body = "\n\n".join(b for b in (_plain_block(blk) for blk in value["blocks"]) if b) + return f"{_plain(value['summary'])}\n\n{body}" + case "map": + return _plain_caption(value.get("caption")) + case "animation" | "audio" | "photo" | "video" | "voice_note": + return _plain_caption(value.get("caption")) + case _: + return "" + + +@dataclass +class RichMedia: + """A media attachment extracted from a rich message.""" + + file: TelegramFile + type: Literal["image", "file", "video", "audio"] + height: int | None = None + mime_type: str | None = None + name: str | None = None + width: int | None = None + + +def _media(blocks: list[TelegramRichBlock], result: list[RichMedia]) -> None: + for value in blocks: + match value["type"]: + case "list": + for entry in value["items"]: + _media(entry["blocks"], result) + case "blockquote" | "collage" | "slideshow" | "details": + _media(value["blocks"], result) + case "animation": + animation = value["animation"] + mime_type = animation.get("mime_type") + result.append( + RichMedia( + file=animation, + height=animation.get("height"), + mime_type=mime_type, + name=animation.get("file_name"), + type="image" if mime_type is not None and mime_type.startswith("image/") else "video", + width=animation.get("width"), + ) + ) + case "audio": + audio = value["audio"] + result.append( + RichMedia( + file=audio, + mime_type=audio.get("mime_type"), + name=audio.get("file_name"), + type="audio", + ) + ) + case "photo": + photos = value["photo"] + photo = photos[-1] if photos else None + if photo: + result.append( + RichMedia( + file=photo, + height=photo.get("height"), + type="image", + width=photo.get("width"), + ) + ) + case "video": + video = value["video"] + result.append( + RichMedia( + file=video, + height=video.get("height"), + mime_type=video.get("mime_type"), + name=video.get("file_name"), + type="video", + width=video.get("width"), + ) + ) + case "voice_note": + voice_note = value["voice_note"] + result.append( + RichMedia( + file=voice_note, + mime_type=voice_note.get("mime_type"), + type="audio", + ) + ) + case _: + pass + + +def rich_message_to_markdown(message: TelegramRichMessage) -> str: + """Render a Telegram rich message to Markdown.""" + return "\n\n".join(b for b in (_block(blk) for blk in message["blocks"]) if b).strip() + + +def rich_message_to_text(message: TelegramRichMessage) -> str: + """Render a Telegram rich message to plain text.""" + return "\n\n".join(b for b in (_plain_block(blk) for blk in message["blocks"]) if b).strip() + + +def rich_message_media(message: TelegramRichMessage) -> list[RichMedia]: + """Extract media attachments from a Telegram rich message.""" + result: list[RichMedia] = [] + _media(message["blocks"], result) + return result diff --git a/tests/test_telegram_rich.py b/tests/test_telegram_rich.py new file mode 100644 index 0000000..49bb0e9 --- /dev/null +++ b/tests/test_telegram_rich.py @@ -0,0 +1,349 @@ +"""Tests for the Telegram rich-message renderer. + +Ports ``packages/adapter-telegram/src/rich.test.ts`` (chat@4.31.0, commit +4662309) plus adversarial coverage for the substitution / regex passes: +a full-punctuation escape sweep, all-backtick fence run-sizing, and a +surrogate-pair (non-BMP) truncation-boundary check. + +Fixtures are annotated to the concrete TG1 rich-block / rich-text TypedDict +variants so the file stays pyrefly-sound. +""" + +from __future__ import annotations + +from chat_sdk.adapters.telegram.rich import ( + TELEGRAM_RICH_MESSAGE_LIMIT, + RichMedia, + rich_message_media, + rich_message_to_markdown, + rich_message_to_text, + truncate_rich_markdown, +) +from chat_sdk.adapters.telegram.rich import ( + _escape_text as escape_text, +) +from chat_sdk.adapters.telegram.rich import ( + _inline_code as inline_code, +) +from chat_sdk.adapters.telegram.types import ( + TelegramPhotoSize, + TelegramRichBlockAnimation, + TelegramRichBlockHeading, + TelegramRichBlockList, + TelegramRichBlockPhoto, + TelegramRichBlockPre, + TelegramRichBlockTable, + TelegramRichBlockText, + TelegramRichCell, + TelegramRichItem, + TelegramRichMessage, + TelegramRichText, + TelegramRichTextHashtag, + TelegramRichTextMention, + TelegramRichTextStyled, + TelegramRichTextUrl, +) +from chat_sdk.shared.markdown_parser import ast_to_plain_text, parse_markdown + + +def test_normalizes_structured_rich_blocks_to_markdown() -> None: + """Headings, links, bold spans, and tables render to Markdown.""" + heading: TelegramRichBlockHeading = {"type": "heading", "size": 2, "text": "Summary"} + guide_url: TelegramRichTextUrl = { + "type": "url", + "text": "the guide", + "url": "https://example.com", + } + continue_bold: TelegramRichTextStyled = {"type": "bold", "text": "continue"} + paragraph_text: TelegramRichText = ["Read ", guide_url, " and ", continue_bold] + paragraph: TelegramRichBlockText = {"type": "paragraph", "text": paragraph_text} + header_row: list[TelegramRichCell] = [ + {"align": "left", "is_header": True, "text": "Name", "valign": "top"}, + {"align": "right", "is_header": True, "text": "Status", "valign": "top"}, + ] + body_row: list[TelegramRichCell] = [ + {"align": "left", "text": "Build", "valign": "top"}, + {"align": "right", "text": "Ready", "valign": "top"}, + ] + table: TelegramRichBlockTable = {"type": "table", "cells": [header_row, body_row]} + message: TelegramRichMessage = {"blocks": [heading, paragraph, table]} + + assert rich_message_to_markdown(message) == "\n\n".join( + [ + "## Summary", + "Read [the guide]() and **continue**", + "| Name | Status |\n| --- | --- |\n| Build | Ready |", + ] + ) + assert "Read the guide and continue" in rich_message_to_text(message) + + +def test_preserves_displayed_text_for_detected_entities() -> None: + """Mention/hashtag entities keep their displayed text through escape + parse.""" + mention: TelegramRichTextMention = { + "type": "mention", + "text": "@chat_sdk", + "username": "chat_sdk", + } + hashtag: TelegramRichTextHashtag = { + "type": "hashtag", + "hashtag": "release", + "text": "#release", + } + paragraph_text: TelegramRichText = [mention, " ", hashtag] + paragraph: TelegramRichBlockText = {"type": "paragraph", "text": paragraph_text} + message: TelegramRichMessage = {"blocks": [paragraph]} + + assert rich_message_to_text(message) == "@chat_sdk #release" + assert ast_to_plain_text(parse_markdown(rich_message_to_markdown(message))) == "@chat_sdk #release" + + +def test_escapes_literal_rich_text_without_changing_displayed_content() -> None: + """Inline code, links, literal markdown, and fenced code escape correctly. + + Asserts on the renderer's emitted Markdown string (the load-bearing, + char-for-char port). Upstream's CommonMark round-trip via + ``mdast-util-from-markdown`` is not replicated here because the Python + markdown parser is intentionally not full CommonMark (it does not handle + angle-bracket link destinations or fence widening); see + ``docs/SELF_REVIEW.md`` / CLAUDE.md "Known Limitations". + """ + code_span: TelegramRichTextStyled = {"type": "code", "text": " a ` b "} + empty_code: TelegramRichTextStyled = {"type": "code", "text": ""} + link_span: TelegramRichTextUrl = { + "type": "url", + "text": "label ] x", + "url": "https://example.com/a_(b)", + } + paragraph_text: TelegramRichText = [ + code_span, + empty_code, + " ", + link_span, + "\n# literal\n- item\n~~plain~~", + ] + paragraph: TelegramRichBlockText = {"type": "paragraph", "text": paragraph_text} + pre: TelegramRichBlockPre = { + "type": "pre", + "language": "type`script", + "text": "const fence = ```;", + } + message: TelegramRichMessage = {"blocks": [paragraph, pre]} + + markdown = rich_message_to_markdown(message) + + # Inline code: backtick run of 1 -> 2-backtick fence; boundary spaces pad. + assert "`` a ` b ``" in markdown + # Empty inline code renders to nothing (no stray fence). + assert "````" not in markdown.split("\n\n")[0] + # Link: display text has its ``]`` escaped, url wrapped in angle brackets. + assert "[label \\] x]()" in markdown + # Literal markdown is backslash-escaped, not interpreted. + assert "\\# literal\n\\- item\n\\~\\~plain\\~\\~" in markdown + # Pre block: inner ``` run (3) widens the fence to 4, language sanitized + # (backticks stripped from ``type`script`` -> ``typescript``). + assert markdown.endswith("````typescript\nconst fence = ```;\n````") + + +def test_normalizes_rich_formatting_to_plain_text() -> None: + """Styled spans and tables collapse to tab/newline-joined plain text.""" + underline: TelegramRichTextStyled = {"type": "underline", "text": "underlined"} + subscript: TelegramRichTextStyled = {"type": "subscript", "text": "subscript"} + marked: TelegramRichTextStyled = {"type": "marked", "text": "marked"} + paragraph_text: TelegramRichText = [underline, " ", subscript, " ", marked] + paragraph: TelegramRichBlockText = {"type": "paragraph", "text": paragraph_text} + header_row: list[TelegramRichCell] = [ + {"align": "left", "text": "Name", "valign": "top"}, + {"align": "left", "text": "Status", "valign": "top"}, + ] + body_row: list[TelegramRichCell] = [ + {"align": "left", "text": "Build", "valign": "top"}, + {"align": "left", "text": "Ready", "valign": "top"}, + ] + table: TelegramRichBlockTable = {"type": "table", "cells": [header_row, body_row]} + message: TelegramRichMessage = {"blocks": [paragraph, table]} + + assert rich_message_to_text(message) == "underlined subscript marked\n\nName\tStatus\nBuild\tReady" + + +def test_truncates_markdown_at_the_rich_message_limit() -> None: + """Over-limit markdown is truncated to the limit and ends with ``...``.""" + markdown = truncate_rich_markdown("a" * (TELEGRAM_RICH_MESSAGE_LIMIT + 100)) + + assert len(markdown) <= TELEGRAM_RICH_MESSAGE_LIMIT + assert markdown.endswith("...") + + +def test_preserves_a_table_like_trailing_line_when_truncating() -> None: + """A trailing table-like line survives the renderer's truncation pass.""" + prefix = f"{'a' * (TELEGRAM_RICH_MESSAGE_LIMIT - 12)}\n| tail |" + markdown = f"{prefix}{'b' * 100}" + + assert "| tail |" in truncate_rich_markdown(markdown) + + +# --------------------------------------------------------------------------- +# Media extraction (richMessageMedia) +# --------------------------------------------------------------------------- + + +def test_extracts_media_and_picks_largest_photo() -> None: + """``rich_message_media`` recurses lists and picks the last photo size.""" + small: TelegramPhotoSize = { + "file_id": "small", + "file_unique_id": "s", + "height": 10, + "width": 10, + } + large: TelegramPhotoSize = { + "file_id": "large", + "file_unique_id": "l", + "height": 200, + "width": 300, + } + photo_block: TelegramRichBlockPhoto = {"type": "photo", "photo": [small, large]} + item: TelegramRichItem = {"label": "1", "blocks": [photo_block]} + list_block: TelegramRichBlockList = {"type": "list", "items": [item]} + message: TelegramRichMessage = {"blocks": [list_block]} + + media = rich_message_media(message) + assert media == [RichMedia(file=large, type="image", height=200, width=300)] + + +def test_animation_image_mime_maps_to_image_type() -> None: + """An animation with an ``image/*`` mime is classified as ``image``.""" + gif: TelegramRichBlockAnimation = { + "type": "animation", + "animation": { + "file_id": "gif", + "file_unique_id": "g", + "duration": 1, + "height": 5, + "width": 6, + "mime_type": "image/gif", + "file_name": "a.gif", + }, + } + mp4: TelegramRichBlockAnimation = { + "type": "animation", + "animation": { + "file_id": "mp4", + "file_unique_id": "m", + "duration": 1, + "height": 5, + "width": 6, + "mime_type": "video/mp4", + }, + } + message: TelegramRichMessage = {"blocks": [gif, mp4]} + + media = rich_message_media(message) + assert [m.type for m in media] == ["image", "video"] + + +# --------------------------------------------------------------------------- +# Adversarial: escape char-class sweep +# --------------------------------------------------------------------------- + +# The four ASCII punctuation ranges of /[!-/:-@[-`{-~]/ enumerated explicitly. +PUNCTUATION_IN_CLASS = ( + "".join(chr(c) for c in range(ord("!"), ord("/") + 1)) + + "".join(chr(c) for c in range(ord(":"), ord("@") + 1)) + + "".join(chr(c) for c in range(ord("["), ord("`") + 1)) + + "".join(chr(c) for c in range(ord("{"), ord("~") + 1)) +) + + +def test_escape_sweep_every_in_class_char_is_backslash_escaped() -> None: + """Every punctuation char in the class is escaped to ``\\``.""" + for ch in PUNCTUATION_IN_CLASS: + assert escape_text(ch) == f"\\{ch}", f"char {ch!r} should be escaped" + + +def test_escape_sweep_out_of_class_chars_are_untouched() -> None: + """Letters, digits, whitespace, and space are left untouched.""" + # Build a set of chars deliberately OUTSIDE the punctuation class. + out_of_class = ( + "abcXYZ0123456789" + + " \t\n\r" + + "".join(chr(c) for c in range(ord("0"), ord("9") + 1)) + + "éñ漢字😀" # non-ASCII: Python Unicode regex must NOT escape these + ) + for ch in out_of_class: + assert escape_text(ch) == ch, f"char {ch!r} should NOT be escaped" + + +def test_escape_does_not_touch_the_space_between_punctuation_ranges() -> None: + """The gaps between ranges (e.g. digits 0-9, uppercase A-Z) stay literal. + + The char class has holes: ``0``-``9`` sit between ``/`` and ``:``; ``A``-``Z`` + sit between ``@`` and ``[``; ``a``-``z`` sit between ``\\`` and ``{``. A + wrong range bound would silently escape these. + """ + for ch in "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz": + assert escape_text(ch) == ch + + +# --------------------------------------------------------------------------- +# Adversarial: all-backtick inline-code fence run sizing +# --------------------------------------------------------------------------- + + +def test_inline_code_fence_grows_with_longest_backtick_run() -> None: + """The fence marker is ``max(1, longest_run + 1)`` backticks long.""" + # No backticks -> minimal 1-backtick fence. + assert inline_code("x") == "`x`" + # One backtick -> fence of 2. + assert inline_code("a`b") == "``a`b``" + # A run of three backticks -> fence of 4. + assert inline_code("a```b") == "````a```b````" + # Two separate runs (1 and 3): the longest (3) wins -> fence of 4. + assert inline_code("a`b```c") == "````a`b```c````" + + +def test_inline_code_pads_when_value_touches_a_backtick() -> None: + """Boundary backticks/spaces trigger one space of padding each side.""" + # Leading backtick -> pad both sides. + assert inline_code("`x") == "`` `x ``" + # Pure spaces around content -> pad. + assert inline_code(" x ") == "` x `" + # Empty value renders to empty string (no fence). + assert inline_code("") == "" + + +# --------------------------------------------------------------------------- +# Adversarial: surrogate-pair / non-BMP truncation boundary +# --------------------------------------------------------------------------- + + +def test_truncation_never_splits_a_non_bmp_code_point() -> None: + """Truncating a long run of emoji cuts on code-point boundaries only. + + Each emoji is a single Python code point (matching JS ``Array.from``), + so the truncated output must contain only whole emoji plus the ``...`` + suffix -- never a lone surrogate or a partial code point. + """ + emoji = "😀" + markdown = emoji * (TELEGRAM_RICH_MESSAGE_LIMIT + 50) + + result = truncate_rich_markdown(markdown) + + assert len(result) <= TELEGRAM_RICH_MESSAGE_LIMIT + assert result.endswith("...") + body = result[:-3] + # Body is non-empty and composed solely of whole emoji code points. + assert body + assert set(body) == {emoji} + # The body is an exact multiple of the (single) emoji code point: no + # partial code point or lone surrogate could survive a code-point slice. + assert body == emoji * len(body) + # No lone surrogate code units leaked in (Python strings never hold them + # from a slice, but assert the invariant explicitly). + assert all(not (0xD800 <= ord(c) <= 0xDFFF) for c in result) + + +def test_truncation_passes_through_under_limit_text_unchanged() -> None: + """At or under the limit, the input is returned verbatim.""" + text = "😀" * 5 + assert truncate_rich_markdown(text) == text + assert truncate_rich_markdown("plain") == "plain" From 287d143260ac97c4ef05d666245f4e8cb56fe235 Mon Sep 17 00:00:00 2001 From: patrick-chinchill Date: Fri, 19 Jun 2026 18:41:00 -0700 Subject: [PATCH 2/2] fix(telegram): JS-truthy empty-list rich-text gates + mutation coverage Render optional rich-text fields (credit/caption/cell text) when the value is an empty list []. Upstream gates these with `value.x ? ... : ""`; JS arrays are always truthy, so credit: [] renders a blank credit, while the prior Python-truthiness gate wrongly skipped it. Add a _present() helper that renders for everything except None/absent and the empty string "". Strip with the exact ECMAScript WhiteSpace+LineTerminator set (JS_WHITESPACE) instead of Python's broader Unicode-whitespace strip(), matching JS trimEnd()/trim() character-for-character. Add regression tests, each verified to fail on its mutation: - empty-list credit renders (truthiness trap) - no-backtick pre block emits a 3-backtick fence (max(3,...)/else 3) - truncation reserves room for the ellipsis (end = LIMIT - 3) - exact-limit-length input passes through unchanged (<= boundary) --- src/chat_sdk/adapters/telegram/rich.py | 70 +++++++++++++++----- tests/test_telegram_rich.py | 89 ++++++++++++++++++++++++++ 2 files changed, 143 insertions(+), 16 deletions(-) diff --git a/src/chat_sdk/adapters/telegram/rich.py b/src/chat_sdk/adapters/telegram/rich.py index 03a99b6..251c037 100644 --- a/src/chat_sdk/adapters/telegram/rich.py +++ b/src/chat_sdk/adapters/telegram/rich.py @@ -39,6 +39,43 @@ # Upstream: /`+/g -- runs of one or more backticks. BACKTICKS = re.compile(r"`+") +# The exact set of characters JS ``String.prototype.trim``/``trimEnd`` removes: +# the ECMAScript WhiteSpace set plus the LineTerminator set. Python's bare +# ``str.strip()``/``str.rstrip()`` strip a broader Unicode-whitespace set (e.g. +# the C0 separators ``\x1c``-``\x1f`` and NEL ``\x85``, which JS keeps; and they +# do NOT strip the BOM ````, which JS removes). Passing this explicit +# string to ``strip``/``rstrip`` matches JS character-for-character. +JS_WHITESPACE = ( + "\t\n\v\f\r " # \t \n \v \f \r and SPACE + " " # NO-BREAK SPACE + " " # OGHAM SPACE MARK + "           " # EN QUAD..HAIR SPACE + "
" # LINE SEPARATOR + "
" # PARAGRAPH SEPARATOR + " " # NARROW NO-BREAK SPACE + " " # MEDIUM MATHEMATICAL SPACE + "　" # IDEOGRAPHIC SPACE + "" # ZERO WIDTH NO-BREAK SPACE (BOM) +) + + +def _present(value: TelegramRichText | None) -> bool: + """Match JS truthiness for an optional ``TelegramRichText`` field. + + Upstream gates these fields with ``value.x ? ... : ""``. The only values + a ``TelegramRichText`` (``str | list | object``) can take that differ from + Python truthiness is the empty list ``[]``: JS arrays are ALWAYS truthy, so + ``credit: []`` renders upstream but a bare ``if value.get("credit")`` would + skip it in Python. We therefore render whenever the value is present and is + not the empty string ``""`` (the only falsy ``TelegramRichText`` in JS): + + - ``None`` / absent -> skip (JS ``undefined`` is falsy) + - ``""`` -> skip (JS ``""`` is falsy) + - ``[]`` -> render (JS arrays are truthy) + - ``[span]`` / ``"text"`` / object -> render (truthy) + """ + return value is not None and value != "" + def truncate_rich_markdown(markdown: str) -> str: """Truncate *markdown* to the Telegram rich-message character limit. @@ -81,7 +118,7 @@ def _inline_code(value: str) -> str: runs = BACKTICKS.findall(value) size = max(1, *(len(run) + 1 for run in runs)) if runs else 1 marker = "`" * size - has_boundary_space = value.startswith(" ") and value.endswith(" ") and len(value.strip()) > 0 + has_boundary_space = value.startswith(" ") and value.endswith(" ") and len(value.strip(JS_WHITESPACE)) > 0 padding = " " if value.startswith("`") or value.endswith("`") or has_boundary_space else "" return f"{marker}{padding}{value}{padding}{marker}" @@ -196,19 +233,19 @@ def _plain(markdown: TelegramRichText) -> str: def _caption(value: TelegramRichCaption | None = None) -> str: if not value: return "" - credit = f"\n{_text(value['credit'])}" if value.get("credit") else "" + credit = f"\n{_text(value['credit'])}" if _present(value.get("credit")) else "" return f"{_text(value['text'])}{credit}" def _plain_caption(value: TelegramRichCaption | None = None) -> str: if not value: return "" - credit = f"\n{_plain(value['credit'])}" if value.get("credit") else "" + credit = f"\n{_plain(value['credit'])}" if _present(value.get("credit")) else "" return f"{_plain(value['text'])}{credit}" def _cell(value: TelegramRichCell) -> str: - return _text(value["text"]) if value.get("text") else "" + return _text(value["text"]) if _present(value.get("text")) else "" def _item(value: TelegramRichItem) -> str: @@ -216,7 +253,7 @@ def _item(value: TelegramRichItem) -> str: if value.get("has_checkbox"): checked = "[x] " if value.get("is_checked") else "[ ] " content = "\n\n".join(_block(b) for b in value["blocks"]).replace("\n", "\n ") - return f"{value['label']} {checked}{content}".rstrip() + return f"{value['label']} {checked}{content}".rstrip(JS_WHITESPACE) def _plain_item(value: TelegramRichItem) -> str: @@ -224,18 +261,18 @@ def _plain_item(value: TelegramRichItem) -> str: if value.get("has_checkbox"): checked = "[x] " if value.get("is_checked") else "[ ] " content = "\n".join(b for b in (_plain_block(blk) for blk in value["blocks"]) if b).replace("\n", "\n ") - return f"{value['label']} {checked}{content}".rstrip() + return f"{value['label']} {checked}{content}".rstrip(JS_WHITESPACE) def _table(value: TelegramRichBlockTable) -> str: rows = [f"| {' | '.join(_cell(c) for c in row)} |" for row in value["cells"]] if len(rows) == 0: - return _text(value["caption"]) if value.get("caption") else "" + return _text(value["caption"]) if _present(value.get("caption")) else "" columns = max(len(row) for row in value["cells"]) separator = f"| {' | '.join('---' for _ in range(columns))} |" content = "\n".join([rows[0], separator, *rows[1:]]) - return f"{_text(value['caption'])}\n\n{content}" if value.get("caption") else content + return f"{_text(value['caption'])}\n\n{content}" if _present(value.get("caption")) else content def _quote(value: str) -> str: @@ -261,10 +298,10 @@ def _block(value: TelegramRichBlock) -> str: return "\n".join(_item(i) for i in value["items"]) case "blockquote": content = "\n\n".join(_block(b) for b in value["blocks"]) - credit = f"\n\n{_text(value['credit'])}" if value.get("credit") else "" + credit = f"\n\n{_text(value['credit'])}" if _present(value.get("credit")) else "" return _quote(f"{content}{credit}") case "pullquote": - credit = f"\n\n{_text(value['credit'])}" if value.get("credit") else "" + credit = f"\n\n{_text(value['credit'])}" if _present(value.get("credit")) else "" return _quote(f"{_text(value['text'])}{credit}") case "collage" | "slideshow": content = "\n\n".join(b for b in (_block(blk) for blk in value["blocks"]) if b) @@ -295,10 +332,10 @@ def _plain_block(value: TelegramRichBlock) -> str: return "\n".join(_plain_item(i) for i in value["items"]) case "blockquote": content = "\n\n".join(b for b in (_plain_block(blk) for blk in value["blocks"]) if b) - credit = f"\n\n{_plain(value['credit'])}" if value.get("credit") else "" + credit = f"\n\n{_plain(value['credit'])}" if _present(value.get("credit")) else "" return f"{content}{credit}" case "pullquote": - credit = f"\n\n{_plain(value['credit'])}" if value.get("credit") else "" + credit = f"\n\n{_plain(value['credit'])}" if _present(value.get("credit")) else "" return f"{_plain(value['text'])}{credit}" case "collage" | "slideshow": content = "\n\n".join(b for b in (_plain_block(blk) for blk in value["blocks"]) if b) @@ -306,10 +343,11 @@ def _plain_block(value: TelegramRichBlock) -> str: return "\n\n".join(part for part in (content, description) if part) case "table": rows = [ - "\t".join(_plain(entry["text"]) if entry.get("text") else "" for entry in row) for row in value["cells"] + "\t".join(_plain(entry["text"]) if _present(entry.get("text")) else "" for entry in row) + for row in value["cells"] ] content = "\n".join(rows) - return f"{_plain(value['caption'])}\n\n{content}" if value.get("caption") else content + return f"{_plain(value['caption'])}\n\n{content}" if _present(value.get("caption")) else content case "details": body = "\n\n".join(b for b in (_plain_block(blk) for blk in value["blocks"]) if b) return f"{_plain(value['summary'])}\n\n{body}" @@ -403,12 +441,12 @@ def _media(blocks: list[TelegramRichBlock], result: list[RichMedia]) -> None: def rich_message_to_markdown(message: TelegramRichMessage) -> str: """Render a Telegram rich message to Markdown.""" - return "\n\n".join(b for b in (_block(blk) for blk in message["blocks"]) if b).strip() + return "\n\n".join(b for b in (_block(blk) for blk in message["blocks"]) if b).strip(JS_WHITESPACE) def rich_message_to_text(message: TelegramRichMessage) -> str: """Render a Telegram rich message to plain text.""" - return "\n\n".join(b for b in (_plain_block(blk) for blk in message["blocks"]) if b).strip() + return "\n\n".join(b for b in (_plain_block(blk) for blk in message["blocks"]) if b).strip(JS_WHITESPACE) def rich_message_media(message: TelegramRichMessage) -> list[RichMedia]: diff --git a/tests/test_telegram_rich.py b/tests/test_telegram_rich.py index 49bb0e9..3be9d5c 100644 --- a/tests/test_telegram_rich.py +++ b/tests/test_telegram_rich.py @@ -32,6 +32,7 @@ TelegramRichBlockList, TelegramRichBlockPhoto, TelegramRichBlockPre, + TelegramRichBlockPullquote, TelegramRichBlockTable, TelegramRichBlockText, TelegramRichCell, @@ -347,3 +348,91 @@ def test_truncation_passes_through_under_limit_text_unchanged() -> None: text = "😀" * 5 assert truncate_rich_markdown(text) == text assert truncate_rich_markdown("plain") == "plain" + + +# --------------------------------------------------------------------------- +# Fidelity: empty-LIST rich-text field renders (JS arrays are truthy) +# --------------------------------------------------------------------------- + + +def test_empty_list_credit_renders_like_a_present_credit() -> None: + """An empty-list ``credit`` ([]) renders, matching JS array truthiness. + + Upstream gates the credit with ``value.credit ? ... : ""``; a JS array is + ALWAYS truthy, so ``credit: []`` produces a blank credit (a leading-newline + empty line in the quote), whereas absent / empty-string skip it. A bare + Python-truthiness gate (``if value.get("credit")``) would wrongly skip + ``[]`` -- this test fails under that regression. + """ + # credit=[] -> truthy in JS -> credit segment "\n\n" + text([]) ("") renders. + # quote("Quote\n\n") -> "> Quote\n> \n> " -> trailing ws trimmed by .strip(). + with_empty_list: TelegramRichBlockPullquote = {"type": "pullquote", "text": "Quote", "credit": []} + assert rich_message_to_markdown({"blocks": [with_empty_list]}) == "> Quote\n> \n>" + + # Absent credit -> JS undefined is falsy -> skipped entirely. + absent: TelegramRichBlockPullquote = {"type": "pullquote", "text": "Quote"} + assert rich_message_to_markdown({"blocks": [absent]}) == "> Quote" + + # Empty-string credit -> JS "" is falsy -> skipped (the ONLY falsy RichText). + empty_string: TelegramRichBlockPullquote = {"type": "pullquote", "text": "Quote", "credit": ""} + assert rich_message_to_markdown({"blocks": [empty_string]}) == "> Quote" + + # A real credit span renders the same blank-line-separated shape as []. + with_author: TelegramRichBlockPullquote = {"type": "pullquote", "text": "Quote", "credit": "Author"} + assert rich_message_to_markdown({"blocks": [with_author]}) == "> Quote\n> \n> Author" + + +# --------------------------------------------------------------------------- +# Mutation coverage: fence default width, truncation reserve, exact boundary +# --------------------------------------------------------------------------- + + +def test_pre_block_without_internal_backticks_emits_a_three_backtick_fence() -> None: + """A code/pre block with NO internal backticks uses the default 3-fence. + + ``_code_block`` sizes the fence as ``max(3, longest_run + 1)`` and falls + back to ``3`` when there are no backtick runs. This pins the no-run default: + the mutations ``else 3`` -> ``else 2`` and ``max(3, ...)`` -> ``max(2, ...)`` + both shrink the fence to 2 backticks and fail this assertion. + """ + fence = "`" * 3 + pre: TelegramRichBlockPre = {"type": "pre", "language": "python", "text": "hello world"} + markdown = rich_message_to_markdown({"blocks": [pre]}) + + assert markdown == f"{fence}python\nhello world\n{fence}" + # Exactly three opening backticks -- never two. + assert len(markdown) - len(markdown.lstrip("`")) == 3 + + +def test_truncation_reserves_room_for_the_ellipsis() -> None: + """Truncation reserves 3 chars for ``...`` so the result stays at the limit. + + With an unclosed bold marker straddling the boundary, the correct reserve + (``end = LIMIT - 3``) returns a result of length EXACTLY ``LIMIT`` on the + first iteration. The mutation ``end = LIMIT`` over-shoots, the shrink loop + diverges, and it returns a shorter (length ``LIMIT - 2``) result -- so an + exact-length assertion fails under the mutation. + """ + # 'a' x (LIMIT-1) + '**' opens an unclosed bold span right at the boundary; + # the trailing run keeps the input over the limit. + over_limit = ("a" * (TELEGRAM_RICH_MESSAGE_LIMIT - 1)) + "**" + ("b" * 200) + result = truncate_rich_markdown(over_limit) + + # The ellipsis fits within the limit -- length is exactly the limit here, + # never over it (mutation would yield LIMIT - 2, killing the equality). + assert len(result) == TELEGRAM_RICH_MESSAGE_LIMIT + assert result.endswith("...") + + +def test_exact_limit_length_passes_through_unchanged() -> None: + """A string of length EXACTLY ``LIMIT`` is returned verbatim (no ``...``). + + The early-return guard is ``len(characters) <= LIMIT``. At the boundary + ``len == LIMIT`` the input must pass through; the mutation ``<=`` -> ``<`` + would instead truncate it (appending ``...``), failing this assertion. + """ + exact = "a" * TELEGRAM_RICH_MESSAGE_LIMIT + result = truncate_rich_markdown(exact) + + assert result == exact + assert not result.endswith("...")