diff --git a/docs/configuration/sinks/telegram.rst b/docs/configuration/sinks/telegram.rst
index c25f54625..4842b16da 100644
--- a/docs/configuration/sinks/telegram.rst
+++ b/docs/configuration/sinks/telegram.rst
@@ -15,9 +15,13 @@ Robusta can report issues and events in your Kubernetes cluster to Telegram conv
 
 .. note::
 
-    Tables are sent as file attachments to Telegram because it's too big for most Telegram chat clients.
+    Small tables (for example Alert labels) are included in the finding message as a
+    collapsible quote, including when ``send_files`` is ``False``. If a message is
+    longer than Telegram's 4096-character limit, it is split across sequential
+    messages rather than attached as a ``.txt`` file. ``send_files`` still attaches
+    images and other ``FileBlock`` files (graphs, logs), not table text.
 
-    In addition, 2-way interactivity (``CallbackBlock``) isn't implemented yet.
+    2-way interactivity (``CallbackBlock``) isn't implemented yet.
 
 Getting your Bot token
 ------------------------------------------------
diff --git a/src/robusta/core/sinks/telegram/telegram_client.py b/src/robusta/core/sinks/telegram/telegram_client.py
index fb23d34a4..d183734bc 100644
--- a/src/robusta/core/sinks/telegram/telegram_client.py
+++ b/src/robusta/core/sinks/telegram/telegram_client.py
@@ -5,6 +5,7 @@
 import requests
 
 from robusta.core.reporting.utils import PNG_SUFFIX, SVG_SUFFIX, convert_svg_to_png, is_image
+from robusta.core.sinks.telegram.telegram_html import split_telegram_html
 
 TELEGRAM_BASE_URL = os.environ.get("TELEGRAM_BASE_URL", "https://api.telegram.org")
 
@@ -16,13 +17,24 @@ def __init__(self, chat_id: Union[int, str], thread_id: int, bot_token: str):
         self.bot_token = bot_token
 
     def send_message(self, message: str, disable_links_preview: bool = True):
+        """Send one or more HTML sendMessage calls. Oversized text is split, never attached as a file."""
+        chunks = split_telegram_html(message)
+        if not chunks:
+            return
+        for chunk in chunks:
+            self._send_message_chunk(chunk, disable_links_preview=disable_links_preview)
+
+    def _send_message_chunk(self, message: str, disable_links_preview: bool = True):
         url = f"{TELEGRAM_BASE_URL}/bot{self.bot_token}/sendMessage"
         message_json = {
             "chat_id": self.chat_id,
             "message_thread_id": self.thread_id,
-            "disable_web_page_preview": disable_links_preview,
-            "parse_mode": "Markdown",
+            # HTML is required for <blockquote expandable>. Do not use MarkdownV2
+            # (open PR robusta-dev/robusta#2105 / issue #1982); that path cannot
+            # express expandable quotes. Related UX: #2137.
+            "parse_mode": "HTML",
             "text": message,
+            "link_preview_options": {"is_disabled": disable_links_preview},
         }
         response = requests.post(url, json=message_json)
 
@@ -33,13 +45,16 @@ def send_message(self, message: str, disable_links_preview: bool = True):
 
     def send_file(self, file_name: str, contents: bytes):
         file_type = "Photo" if is_image(file_name) else "Document"
-        url = f"{TELEGRAM_BASE_URL}/bot{self.bot_token}/send{file_type}?chat_id={self.chat_id}"
+        url = f"{TELEGRAM_BASE_URL}/bot{self.bot_token}/send{file_type}"
         if file_name.endswith(SVG_SUFFIX):
             contents = convert_svg_to_png(contents)
             file_name = file_name.replace(SVG_SUFFIX, PNG_SUFFIX)
 
+        data = {"chat_id": self.chat_id}
+        if self.thread_id is not None:
+            data["message_thread_id"] = self.thread_id
         files = {file_type.lower(): (file_name, contents)}
-        response = requests.post(url, files=files)
+        response = requests.post(url, data=data, files=files)
 
         if response.status_code != 200:
             logging.error(
diff --git a/src/robusta/core/sinks/telegram/telegram_html.py b/src/robusta/core/sinks/telegram/telegram_html.py
new file mode 100644
index 000000000..50ef5ac7b
--- /dev/null
+++ b/src/robusta/core/sinks/telegram/telegram_html.py
@@ -0,0 +1,245 @@
+"""Telegram HTML helpers.
+
+Issue #2167 requires parse_mode=HTML and <blockquote expandable>. This module is
+intentionally separate from any MarkdownV2 transformer (PR #2105 / issue #1982).
+Related Telegram UX: #2137. Tables are never routed through
+Transformer.tableblock_to_fileblocks (column-count helper used by other sinks;
+telegram historically filed every table after PR #245).
+"""
+
+import html
+import re
+from typing import List
+
+from robusta.core.reporting.base import BaseBlock
+from robusta.core.reporting.blocks import (
+    DividerBlock,
+    FileBlock,
+    HeaderBlock,
+    JsonBlock,
+    KubernetesDiffBlock,
+    ListBlock,
+    MarkdownBlock,
+    TableBlock,
+)
+
+try:
+    from tabulate import tabulate
+except ImportError:
+
+    def tabulate(*args, **kwargs):
+        raise ImportError("Please install tabulate to use the TableBlock")
+
+
+TELEGRAM_MESSAGE_CHAR_LIMIT = 4096
+TELEGRAM_MIN_DOCUMENT_TEXT_BYTES = 1024
+# Room to close/reopen a nested tag stack when splitting a long message.
+_TAG_CLOSE_RESERVE = 80
+
+_TAG_PATTERN = re.compile(r"</?([a-zA-Z][a-zA-Z0-9-]*)(?:\s[^>]*)?/?>")
+_SLACK_LINK_PATTERN = re.compile(r"<([^<|>\s]+)\|([^>]+)>")
+_MD_LINK_PATTERN = re.compile(r"\[([^\]]+)\]\(([^)]+)\)")
+_CODE_PATTERN = re.compile(r"`([^`]+)`")
+_BOLD_DOUBLE_PATTERN = re.compile(r"\*\*(.+?)\*\*")
+_BOLD_SINGLE_PATTERN = re.compile(r"\*(?!\s)([^*]+?)\*")
+
+
+def escape_telegram_html(text: str) -> str:
+    """Escape &, <, and > in user or table text for Telegram HTML parse_mode."""
+    return html.escape(str(text), quote=False)
+
+
+def telegram_html_link(text: str, url: str) -> str:
+    """Build an <a href> tag with escaped text and URL."""
+    return f'<a href="{html.escape(url, quote=True)}">{escape_telegram_html(text)}</a>'
+
+
+def markdown_to_telegram_html(text: str) -> str:
+    """Convert finding markdown (including Slack links) to Telegram HTML.
+
+    User text is HTML-escaped first. Only Telegram-supported tags are emitted.
+    """
+    if not text:
+        return ""
+
+    slack_links = []
+
+    def _stash_slack_link(match: re.Match) -> str:
+        slack_links.append((match.group(2), match.group(1)))
+        return f"\x00SLACK{len(slack_links) - 1}\x00"
+
+    working = _SLACK_LINK_PATTERN.sub(_stash_slack_link, text)
+    escaped = escape_telegram_html(working)
+
+    md_links = []
+
+    def _stash_md_link(match: re.Match) -> str:
+        md_links.append((match.group(1), match.group(2)))
+        return f"\x00MDLINK{len(md_links) - 1}\x00"
+
+    escaped = _MD_LINK_PATTERN.sub(_stash_md_link, escaped)
+
+    code_spans = []
+
+    def _stash_code(match: re.Match) -> str:
+        code_spans.append(match.group(1))
+        return f"\x00CODE{len(code_spans) - 1}\x00"
+
+    escaped = _CODE_PATTERN.sub(_stash_code, escaped)
+    escaped = _BOLD_DOUBLE_PATTERN.sub(r"<b>\1</b>", escaped)
+    escaped = _BOLD_SINGLE_PATTERN.sub(r"<b>\1</b>", escaped)
+
+    for index, code_text in enumerate(code_spans):
+        escaped = escaped.replace(f"\x00CODE{index}\x00", f"<code>{code_text}</code>")
+
+    for index, (link_text, url) in enumerate(md_links):
+        # link_text and url were escaped with quote=False; re-escape the href.
+        href = html.escape(html.unescape(url), quote=True)
+        escaped = escaped.replace(f"\x00MDLINK{index}\x00", f'<a href="{href}">{link_text}</a>')
+
+    for index, (link_text, url) in enumerate(slack_links):
+        escaped = escaped.replace(
+            f"\x00SLACK{index}\x00",
+            telegram_html_link(link_text, url),
+        )
+
+    return escaped
+
+
+def table_block_to_telegram_html(block: TableBlock) -> str:
+    """Render a TableBlock as a collapsible Telegram quote."""
+    table_text = tabulate(block.render_rows(), headers=block.headers, tablefmt="presto")
+    body = f"<pre>{escape_telegram_html(table_text)}</pre>"
+    if block.table_name:
+        body = f"{markdown_to_telegram_html(block.table_name)}\n{body}"
+    return f"<blockquote expandable>{body}</blockquote>"
+
+
+def block_to_telegram_html(block: BaseBlock) -> str:
+    """Render a reporting block as Telegram HTML. FileBlocks are omitted."""
+    if isinstance(block, FileBlock):
+        return ""
+    if isinstance(block, TableBlock):
+        return table_block_to_telegram_html(block)
+    if isinstance(block, MarkdownBlock):
+        return markdown_to_telegram_html(block.text) if block.text else ""
+    if isinstance(block, DividerBlock):
+        return "-------------------"
+    if isinstance(block, JsonBlock):
+        return f"<pre>{escape_telegram_html(block.json_str)}</pre>"
+    if isinstance(block, HeaderBlock):
+        return f"<b>{escape_telegram_html(block.text)}</b>"
+    if isinstance(block, ListBlock):
+        return "\n".join(f"• {escape_telegram_html(item)}" for item in block.items)
+    if isinstance(block, KubernetesDiffBlock):
+        lines = []
+        for diff in block.diffs:
+            path = escape_telegram_html(".".join(str(part) for part in diff.path))
+            old = escape_telegram_html(str(diff.other_value))
+            new = escape_telegram_html(str(diff.value))
+            lines.append(f"<b>{path}</b>: {old} ==> {new}")
+        return "\n".join(lines)
+    return ""
+
+
+def _tag_name(opening_tag: str) -> str:
+    match = re.match(r"</?([a-zA-Z][a-zA-Z0-9-]*)", opening_tag)
+    return match.group(1).lower() if match else ""
+
+
+def _open_tag_stack(html_text: str) -> List[str]:
+    """Return unmatched opening tags in document order (full tag strings)."""
+    stack: List[str] = []
+    for match in _TAG_PATTERN.finditer(html_text):
+        raw = match.group(0)
+        name = match.group(1).lower()
+        if raw.startswith("</"):
+            for index in range(len(stack) - 1, -1, -1):
+                if _tag_name(stack[index]) == name:
+                    stack.pop(index)
+                    break
+        elif raw.endswith("/>"):
+            continue
+        else:
+            stack.append(raw)
+    return stack
+
+
+def _avoid_split_inside_entity(text: str, index: int) -> int:
+    """Backtrack so `&amp;` / `&lt;` / `&#123;` are not split across chunks."""
+    if index <= 0 or index >= len(text):
+        return index
+    amp = text.rfind("&", 0, index)
+    if amp == -1:
+        return index
+    semicolon = text.find(";", amp, min(len(text), amp + 16))
+    if semicolon == -1:
+        return amp
+    if amp < index <= semicolon:
+        return amp
+    return index
+
+
+def _find_split_index(text: str, limit: int) -> int:
+    """Choose a split index at or before limit that is not inside a tag or entity."""
+    if len(text) <= limit:
+        return len(text)
+    window = text[:limit]
+    last_open = window.rfind("<")
+    last_close = window.rfind(">")
+    if last_open > last_close:
+        return _avoid_split_inside_entity(text, last_open if last_open > 0 else limit)
+
+    newline = window.rfind("\n")
+    if newline >= limit // 4:
+        return _avoid_split_inside_entity(text, newline + 1)
+    space = window.rfind(" ")
+    if space >= limit // 4:
+        return _avoid_split_inside_entity(text, space + 1)
+    return _avoid_split_inside_entity(text, limit)
+
+
+def split_telegram_html(text: str, limit: int = TELEGRAM_MESSAGE_CHAR_LIMIT) -> List[str]:
+    """Split HTML into sequential chunks that each fit Telegram's sendMessage limit.
+
+    Unclosed tags are closed at the end of a chunk and reopened on the next one.
+    Never used as a reason to send a .txt document.
+    """
+    if not text:
+        return []
+
+    chunks: List[str] = []
+    remaining = text
+    while remaining:
+        if len(remaining) <= limit:
+            chunks.append(remaining)
+            break
+
+        content_limit = max(limit - _TAG_CLOSE_RESERVE, 1)
+        split_at = _find_split_index(remaining, content_limit)
+        if split_at <= 0:
+            split_at = min(content_limit, len(remaining))
+
+        chunk = remaining[:split_at]
+        remaining = remaining[split_at:]
+        stack = _open_tag_stack(chunk)
+        if stack:
+            close = "".join(f"</{_tag_name(tag)}>" for tag in reversed(stack))
+            reopen = "".join(stack)
+            while chunk and len(chunk) + len(close) > limit:
+                remaining = chunk[-1] + remaining
+                chunk = chunk[:-1]
+                stack = _open_tag_stack(chunk)
+                close = "".join(f"</{_tag_name(tag)}>" for tag in reversed(stack))
+                reopen = "".join(stack)
+            chunk = chunk + close
+            remaining = reopen + remaining
+        if chunk:
+            chunks.append(chunk)
+
+    return chunks
+
+
+def should_send_text_as_document(contents: bytes) -> bool:
+    """Return True only when text is large enough to justify sendDocument."""
+    return len(contents) >= TELEGRAM_MIN_DOCUMENT_TEXT_BYTES
diff --git a/src/robusta/core/sinks/telegram/telegram_sink.py b/src/robusta/core/sinks/telegram/telegram_sink.py
index fd5a10c1e..57f48f9c4 100644
--- a/src/robusta/core/sinks/telegram/telegram_sink.py
+++ b/src/robusta/core/sinks/telegram/telegram_sink.py
@@ -1,13 +1,15 @@
-from enum import Enum
-
-from tabulate import tabulate
-
 from robusta.core.reporting.base import BaseBlock, Finding, FindingSeverity, FindingStatus
-from robusta.core.reporting.blocks import FileBlock, MarkdownBlock, TableBlock
+from robusta.core.reporting.blocks import FileBlock
+from robusta.core.reporting.utils import is_image
 from robusta.core.sinks.sink_base import SinkBase
 from robusta.core.sinks.telegram.telegram_client import TelegramClient
+from robusta.core.sinks.telegram.telegram_html import (
+    block_to_telegram_html,
+    escape_telegram_html,
+    markdown_to_telegram_html,
+    telegram_html_link,
+)
 from robusta.core.sinks.telegram.telegram_sink_params import TelegramSinkConfigWrapper
-from robusta.core.sinks.transformer import Transformer
 
 SEVERITY_EMOJI_MAP = {
     FindingSeverity.INFO: "\U0001F7E2",
@@ -16,7 +18,6 @@
 }
 INVESTIGATE_ICON = "\U0001F50E"
 SILENCE_ICON = "\U0001F515"
-VIDEO_ICON = "\U0001F3AC"
 
 
 class TelegramSink(SinkBase):
@@ -32,17 +33,20 @@ def write_finding(self, finding: Finding, platform_enabled: bool):
         self.__send_telegram_message(finding, platform_enabled)
 
     def __send_telegram_message(self, finding: Finding, platform_enabled: bool):
-        self.client.send_message(self.__get_message_text(finding, platform_enabled))
-        if self.send_files:
-            for enrichment in finding.enrichments:
-                file_blocks = [block for block in enrichment.blocks if isinstance(block, FileBlock)]
-                for block in file_blocks:
+        has_graph_or_image = self.send_files and self.__finding_has_graph_or_image(finding)
+        self.client.send_message(
+            self.__get_message_text(finding, platform_enabled),
+            disable_links_preview=not has_graph_or_image,
+        )
+        # Tables are already in the HTML message. send_files only attaches real
+        # FileBlock images/files. When send_files is false, tables still inline
+        # (or split across sendMessage); they are never dropped or sent as .txt.
+        if not self.send_files:
+            return
+        for enrichment in finding.enrichments:
+            for block in enrichment.blocks:
+                if isinstance(block, FileBlock):
                     self.client.send_file(file_name=block.filename, contents=block.contents)
-                table_blocks = [block for block in enrichment.blocks if isinstance(block, TableBlock)]
-                for block in table_blocks:
-                    table_text = tabulate(block.render_rows(), headers=block.headers, tablefmt="presto")
-                    table_name = block.table_name if block.table_name else "table"
-                    self.client.send_file(file_name=f"{table_name}.txt", contents=table_text.encode("utf-8"))
 
     def __get_message_text(self, finding: Finding, platform_enabled: bool):
         status: FindingStatus = (
@@ -56,44 +60,57 @@ def __get_message_text(self, finding: Finding, platform_enabled: bool):
         if actions_content:
             message_content += actions_content
 
-        blocks = [MarkdownBlock(text=f"*Source:* `{self.cluster_name}`\n\n")]
+        message_content += f"<b>Source:</b> <code>{escape_telegram_html(self.cluster_name)}</code>\n\n"
 
-        # first add finding description block
         if finding.description:
-            blocks.append(MarkdownBlock(finding.description))
+            message_content += markdown_to_telegram_html(finding.description) + "\n"
 
         for enrichment in finding.enrichments:
-            blocks.extend([block for block in enrichment.blocks if self.__is_telegram_text_block(block)])
-
-        for block in blocks:
-            block_text = Transformer.to_standard_markdown([block])
-            if len(block_text) + len(message_content) >= 4096:  # telegram message size limit
-                break
-            message_content += block_text + "\n"
+            for block in enrichment.blocks:
+                if not self.__is_telegram_text_block(block):
+                    continue
+                block_text = block_to_telegram_html(block)
+                if block_text:
+                    message_content += block_text + "\n"
 
         return message_content
 
     def _get_actions_block(self, finding: Finding, platform_enabled: bool):
-        actions_content = ""
+        actions = []
         if platform_enabled:
-            actions_content += (
-                f"[{INVESTIGATE_ICON} Investigate]({finding.get_investigate_uri(self.account_id, self.cluster_name)}) "
+            actions.append(
+                telegram_html_link(
+                    f"{INVESTIGATE_ICON} Investigate",
+                    finding.get_investigate_uri(self.account_id, self.cluster_name),
+                )
             )
             if finding.add_silence_url:
-                actions_content += f"[{SILENCE_ICON} Silence]({finding.get_prometheus_silence_url(self.account_id, self.cluster_name)})"
+                actions.append(
+                    telegram_html_link(
+                        f"{SILENCE_ICON} Silence",
+                        finding.get_prometheus_silence_url(self.account_id, self.cluster_name),
+                    )
+                )
 
         for link in finding.links:
-            actions_content = f"[{link.link_text}]({link.url})"
+            actions.append(telegram_html_link(link.link_text, link.url))
 
-        if actions_content:
-            actions_content += "\n\n"
+        if not actions:
+            return ""
 
-        return actions_content
+        return " ".join(actions) + "\n\n"
 
     @classmethod
     def __is_telegram_text_block(cls, block: BaseBlock) -> bool:
-        # enrichments text tables are too big for mobile device
-        return not (isinstance(block, FileBlock) or isinstance(block, TableBlock))
+        return not isinstance(block, FileBlock)
+
+    @classmethod
+    def __finding_has_graph_or_image(cls, finding: Finding) -> bool:
+        for enrichment in finding.enrichments:
+            for block in enrichment.blocks:
+                if isinstance(block, FileBlock) and is_image(block.filename):
+                    return True
+        return False
 
     @classmethod
     def __build_telegram_title(
@@ -101,4 +118,4 @@ def __build_telegram_title(
     ) -> str:
         icon = SEVERITY_EMOJI_MAP.get(severity, "")
         status_str: str = f"{status.to_emoji()} {status.name.lower()} - " if add_silence_url else ""
-        return f"{status_str}{icon} {severity.name} - *{title}*\n\n"
+        return f"{status_str}{icon} {severity.name} - <b>{escape_telegram_html(title)}</b>\n\n"
diff --git a/src/robusta/core/sinks/telegram/telegram_sink_params.py b/src/robusta/core/sinks/telegram/telegram_sink_params.py
index cf1d002b0..f435613db 100644
--- a/src/robusta/core/sinks/telegram/telegram_sink_params.py
+++ b/src/robusta/core/sinks/telegram/telegram_sink_params.py
@@ -8,7 +8,7 @@ class TelegramSinkParams(SinkBaseParams):
     bot_token: str
     chat_id: Union[int, str]
     thread_id: int = None
-    send_files: bool = True  # Change to False, to omit file attachments
+    send_files: bool = True  # Images and FileBlock files only; tables are always inline
 
     @classmethod
     def _get_sink_type(cls):
diff --git a/tests/test_telegram_sink.py b/tests/test_telegram_sink.py
new file mode 100644
index 000000000..631c67a2b
--- /dev/null
+++ b/tests/test_telegram_sink.py
@@ -0,0 +1,400 @@
+from unittest.mock import MagicMock, patch
+
+from robusta.core.reporting.base import Finding, FindingSeverity, Link
+from robusta.core.reporting.blocks import FileBlock, MarkdownBlock, TableBlock
+from robusta.core.sinks.telegram.telegram_client import TelegramClient
+from robusta.core.sinks.telegram.telegram_html import (
+    TELEGRAM_MESSAGE_CHAR_LIMIT,
+    TELEGRAM_MIN_DOCUMENT_TEXT_BYTES,
+    escape_telegram_html,
+    markdown_to_telegram_html,
+    should_send_text_as_document,
+    split_telegram_html,
+    table_block_to_telegram_html,
+)
+from robusta.core.sinks.telegram.telegram_sink import TelegramSink
+from robusta.core.sinks.telegram.telegram_sink_params import TelegramSinkConfigWrapper, TelegramSinkParams
+
+
+class MockRegistry:
+    def get_global_config(self) -> dict:
+        return {
+            "account_id": "test-account",
+            "cluster_name": "test-cluster",
+            "signing_key": "test-signing-key",
+        }
+
+
+def _sink(send_files: bool = True) -> TelegramSink:
+    config = TelegramSinkConfigWrapper(
+        telegram_sink=TelegramSinkParams(
+            name="telegram_sink",
+            bot_token="test-token",
+            chat_id=123456,
+            thread_id=1,
+            send_files=send_files,
+        )
+    )
+    return TelegramSink(config, MockRegistry())
+
+
+def _finding_with_table(table_name: str = "*Alert labels*", rows=None, headers=None, description="Pod is crashing"):
+    finding = Finding(
+        title="CrashLoopBackOff",
+        description=description,
+        aggregation_key="CrashLoopBackOff",
+        severity=FindingSeverity.HIGH,
+    )
+    finding.add_enrichment(
+        [
+            TableBlock(
+                rows=rows
+                or [
+                    ["alertname", "CrashLoopBackOff"],
+                    ["namespace", "default"],
+                    ["pod", "demo"],
+                ],
+                headers=headers or ["label", "value"],
+                table_name=table_name,
+            )
+        ]
+    )
+    return finding
+
+
+def test_escape_telegram_html_escapes_ampersand_lt_gt():
+    assert escape_telegram_html("a & b < c > d") == "a &amp; b &lt; c &gt; d"
+
+
+def test_markdown_to_telegram_html_escapes_user_text_and_applies_formatting():
+    html = markdown_to_telegram_html("See *status* of `app & svc` at [docs](https://example.com/?q=a&b=1)")
+    assert "<b>status</b>" in html
+    assert "<code>app &amp; svc</code>" in html
+    assert 'href="https://example.com/?q=a&amp;b=1"' in html
+    assert "&amp;" in html
+    assert "<script>" not in html
+
+
+def test_markdown_code_span_keeps_literal_asterisks():
+    html = markdown_to_telegram_html("use `*pod*` not *pod*")
+    assert "<code>*pod*</code>" in html
+    assert "<code><b>pod</b></code>" not in html
+    assert "<b>pod</b>" in html
+
+
+def test_table_block_uses_expandable_blockquote_and_escapes_cells():
+    block = TableBlock(rows=[["cpu", "80% & idle"]], headers=["name", "value"], table_name="*Labels*")
+    html = table_block_to_telegram_html(block)
+    assert "<blockquote expandable>" in html
+    assert "</blockquote>" in html
+    assert "<pre>" in html
+    assert "&amp;" in html
+    assert "<b>Labels</b>" in html
+    label_at = html.find("<b>Labels</b>")
+    quote_at = html.find("<blockquote expandable>")
+    assert quote_at != -1 and label_at != -1
+    assert quote_at < label_at < html.find("</blockquote>")
+
+
+def test_should_not_send_document_for_text_under_1kb():
+    assert should_send_text_as_document(b"x" * (TELEGRAM_MIN_DOCUMENT_TEXT_BYTES - 1)) is False
+    assert should_send_text_as_document(b"x" * TELEGRAM_MIN_DOCUMENT_TEXT_BYTES) is True
+
+
+def test_split_telegram_html_keeps_short_text_in_one_chunk():
+    assert split_telegram_html("hello") == ["hello"]
+
+
+def test_split_telegram_html_splits_over_limit_and_reopens_tags():
+    inner = "row\n" * 2000
+    html = f"<blockquote expandable><pre>{inner}</pre></blockquote>"
+    chunks = split_telegram_html(html)
+    assert len(chunks) > 1
+    assert all(len(chunk) <= TELEGRAM_MESSAGE_CHAR_LIMIT for chunk in chunks)
+    assert chunks[0].startswith("<blockquote expandable>")
+    assert chunks[0].endswith("</blockquote>") or "</pre>" in chunks[0]
+    assert any("<blockquote expandable>" in chunk for chunk in chunks[1:])
+
+
+def test_split_telegram_html_does_not_split_named_entities():
+    # content_limit is TELEGRAM_MESSAGE_CHAR_LIMIT - 80; land inside &amp;
+    prefix = "x" * (TELEGRAM_MESSAGE_CHAR_LIMIT - 80 - 1)
+    html = prefix + "&amp;tail" + "y" * 50
+    chunks = split_telegram_html(html)
+    assert len(chunks) >= 2
+    assert all("&amp" not in chunk or "&amp;" in chunk for chunk in chunks)
+    assert "".join(chunks) == html
+    assert not any(chunk.endswith("&") or chunk.endswith("&a") or chunk.endswith("&am") for chunk in chunks)
+
+
+@patch("robusta.core.sinks.telegram.telegram_client.requests.post")
+def test_send_message_uses_html_parse_mode_and_link_preview_options(mock_post):
+    mock_post.return_value.status_code = 200
+    client = TelegramClient(chat_id=123456, thread_id=1, bot_token="test-token")
+    client.send_message("hello <b>world</b>")
+
+    mock_post.assert_called_once()
+    payload = mock_post.call_args.kwargs["json"]
+    assert payload["parse_mode"] == "HTML"
+    assert payload["parse_mode"] != "MarkdownV2"
+    assert "MarkdownV2" not in payload.values()
+    assert payload["link_preview_options"] == {"is_disabled": True}
+    assert payload["text"] == "hello <b>world</b>"
+    assert "disable_web_page_preview" not in payload
+
+
+@patch("robusta.core.sinks.telegram.telegram_client.requests.post")
+def test_send_message_splits_large_text_into_sequential_sendmessage_calls(mock_post):
+    mock_post.return_value.status_code = 200
+    client = TelegramClient(chat_id=123456, thread_id=1, bot_token="test-token")
+    long_text = "A" * (TELEGRAM_MESSAGE_CHAR_LIMIT + 50)
+    client.send_message(long_text)
+
+    assert mock_post.call_count >= 2
+    texts = [call.kwargs["json"]["text"] for call in mock_post.call_args_list]
+    assert all(len(text) <= TELEGRAM_MESSAGE_CHAR_LIMIT for text in texts)
+    assert "".join(texts) == long_text
+    for args in mock_post.call_args_list:
+        payload = args.kwargs["json"]
+        assert payload["parse_mode"] == "HTML"
+        assert payload["parse_mode"] != "MarkdownV2"
+        url = args.args[0] if args.args else args.kwargs.get("url", "")
+        assert "sendMessage" in (url or mock_post.call_args_list[0].args[0])
+        assert "sendDocument" not in str(args)
+
+
+@patch("robusta.core.sinks.telegram.telegram_client.requests.post")
+def test_send_message_never_uses_markdown_v2(mock_post):
+    mock_post.return_value.status_code = 200
+    client = TelegramClient(chat_id=123456, thread_id=1, bot_token="test-token")
+    client.send_message("plain")
+    payload = mock_post.call_args.kwargs["json"]
+    assert payload["parse_mode"] == "HTML"
+    assert payload.get("parse_mode") != "MarkdownV2"
+
+
+def test_small_tableblock_is_inlined_and_does_not_call_send_file():
+    sink = _sink(send_files=True)
+    finding = _finding_with_table()
+    sink.client = MagicMock()
+
+    sink.write_finding(finding, platform_enabled=False)
+
+    sink.client.send_file.assert_not_called()
+    sink.client.send_message.assert_called_once()
+    message = sink.client.send_message.call_args.args[0]
+    assert "<blockquote expandable>" in message
+    assert "CrashLoopBackOff" in message
+    assert "namespace" in message
+    assert ".txt" not in message
+
+
+def test_small_table_under_1kb_never_sent_as_document_even_when_send_files_true():
+    sink = _sink(send_files=True)
+    finding = _finding_with_table()
+    sink.client = MagicMock()
+
+    sink.write_finding(finding, platform_enabled=False)
+
+    assert sink.client.send_file.call_count == 0
+    message = sink.client.send_message.call_args.args[0]
+    encoded = message.encode("utf-8")
+    assert len(finding.enrichments[0].blocks[0].to_table_string().encode("utf-8")) < TELEGRAM_MIN_DOCUMENT_TEXT_BYTES
+    assert encoded  # table lives in the HTML message
+
+
+def test_html_special_characters_in_title_and_table_are_escaped():
+    sink = _sink()
+    finding = Finding(
+        title="CPU > 90% & <critical>",
+        description="Usage of app & db",
+        aggregation_key="HighCPU",
+        severity=FindingSeverity.HIGH,
+    )
+    finding.add_enrichment(
+        [TableBlock(rows=[["msg", "a < b & c > d"]], headers=["k", "v"], table_name="Details")]
+    )
+    sink.client = MagicMock()
+
+    sink.write_finding(finding, platform_enabled=False)
+
+    message = sink.client.send_message.call_args.args[0]
+    assert "CPU &gt; 90% &amp; &lt;critical&gt;" in message
+    assert "app &amp; db" in message
+    assert "a &lt; b &amp; c &gt; d" in message
+    assert "<blockquote expandable>" in message
+
+
+def test_large_table_is_still_inlined_not_attached_as_txt():
+    rows = [[f"label{i}", f"value{i}"] for i in range(80)]
+    sink = _sink(send_files=True)
+    finding = _finding_with_table(rows=rows)
+    table_bytes = finding.enrichments[0].blocks[0].to_table_string().encode("utf-8")
+    assert len(table_bytes) >= TELEGRAM_MIN_DOCUMENT_TEXT_BYTES
+    sink.client = MagicMock()
+
+    sink.write_finding(finding, platform_enabled=False)
+
+    sink.client.send_file.assert_not_called()
+    message = sink.client.send_message.call_args.args[0]
+    assert "<blockquote expandable>" in message
+    assert "label0" in message
+
+
+def test_fileblock_images_still_use_send_file_when_send_files_true():
+    sink = _sink(send_files=True)
+    finding = _finding_with_table()
+    finding.add_enrichment([FileBlock(filename="graph.png", contents=b"fake-png-bytes")])
+    sink.client = MagicMock()
+
+    sink.write_finding(finding, platform_enabled=False)
+
+    sink.client.send_file.assert_called_once_with(file_name="graph.png", contents=b"fake-png-bytes")
+    message = sink.client.send_message.call_args.args[0]
+    assert "<blockquote expandable>" in message
+    assert sink.client.send_message.call_args.kwargs["disable_links_preview"] is False
+
+
+def test_send_files_false_skips_fileblocks_but_still_inlines_tables():
+    sink = _sink(send_files=False)
+    finding = _finding_with_table()
+    finding.add_enrichment([FileBlock(filename="graph.png", contents=b"fake-png-bytes")])
+    sink.client = MagicMock()
+
+    sink.write_finding(finding, platform_enabled=False)
+
+    sink.client.send_file.assert_not_called()
+    message = sink.client.send_message.call_args.args[0]
+    assert "<blockquote expandable>" in message
+    assert "CrashLoopBackOff" in message
+    assert ".txt" not in message
+    assert sink.client.send_message.call_args.kwargs["disable_links_preview"] is True
+
+
+def test_send_files_false_large_text_splits_sendmessage_never_txt():
+    sink = _sink(send_files=False)
+    rows = [[f"label{i}", f"value{i}"] for i in range(80)]
+    finding = _finding_with_table(rows=rows, description="z" * 5000)
+    table_bytes = finding.enrichments[0].blocks[0].to_table_string().encode("utf-8")
+    assert len(table_bytes) >= TELEGRAM_MIN_DOCUMENT_TEXT_BYTES
+
+    with patch("robusta.core.sinks.telegram.telegram_client.requests.post") as mock_post:
+        mock_post.return_value.status_code = 200
+        sink.write_finding(finding, platform_enabled=False)
+
+    assert mock_post.call_count >= 2
+    for args in mock_post.call_args_list:
+        url = args.args[0]
+        payload = args.kwargs.get("json") or {}
+        assert url.endswith("/sendMessage")
+        assert "sendDocument" not in url
+        assert payload.get("parse_mode") == "HTML"
+        assert payload.get("parse_mode") != "MarkdownV2"
+        assert len(payload["text"]) <= TELEGRAM_MESSAGE_CHAR_LIMIT
+    combined = "".join(args.kwargs["json"]["text"] for args in mock_post.call_args_list)
+    assert "<blockquote expandable>" in combined
+    assert "label0" in combined
+
+
+@patch("robusta.core.sinks.telegram.telegram_client.requests.post")
+def test_send_file_passes_message_thread_id(mock_post):
+    mock_post.return_value.status_code = 200
+    client = TelegramClient(chat_id=123456, thread_id=99, bot_token="test-token")
+    client.send_file(file_name="graph.png", contents=b"fake-png-bytes")
+
+    mock_post.assert_called_once()
+    url = mock_post.call_args.args[0]
+    assert url.endswith("/sendPhoto")
+    data = mock_post.call_args.kwargs["data"]
+    assert data["chat_id"] == 123456
+    assert data["message_thread_id"] == 99
+    files = mock_post.call_args.kwargs["files"]
+    assert "photo" in files
+
+
+@patch("robusta.core.sinks.telegram.telegram_client.requests.post")
+def test_send_file_omits_thread_id_when_unset(mock_post):
+    mock_post.return_value.status_code = 200
+    client = TelegramClient(chat_id=123456, thread_id=None, bot_token="test-token")
+    client.send_file(file_name="notes.log", contents=b"logfile")
+
+    data = mock_post.call_args.kwargs["data"]
+    assert data["chat_id"] == 123456
+    assert "message_thread_id" not in data
+    url = mock_post.call_args.args[0]
+    assert url.endswith("/sendDocument")
+
+
+def test_sink_params_do_not_add_markdownv2_parse_mode():
+    # Stay independent of robusta-dev/robusta#2105, which adds a MarkdownV2 parse_mode.
+    assert "parse_mode" not in TelegramSinkParams.__fields__
+    params = TelegramSinkParams(name="tg", bot_token="t", chat_id=1)
+    assert getattr(params, "parse_mode", None) != "MarkdownV2"
+
+
+def test_overflow_text_does_not_fall_back_to_txt_document():
+    sink = _sink(send_files=True)
+    finding = _finding_with_table(description="x" * 5000)
+    sink.client = MagicMock()
+
+    sink.write_finding(finding, platform_enabled=False)
+
+    sink.client.send_file.assert_not_called()
+    sink.client.send_message.assert_called_once()
+    message = sink.client.send_message.call_args.args[0]
+    assert len(message) > TELEGRAM_MESSAGE_CHAR_LIMIT
+
+
+@patch("robusta.core.sinks.telegram.telegram_client.requests.post")
+def test_sink_large_message_splits_sendmessage_via_client(mock_post):
+    mock_post.return_value.status_code = 200
+    sink = _sink(send_files=True)
+    finding = _finding_with_table(description="y" * 5000)
+
+    sink.write_finding(finding, platform_enabled=False)
+
+    assert mock_post.call_count >= 2
+    for args in mock_post.call_args_list:
+        payload = args.kwargs["json"]
+        assert payload["parse_mode"] == "HTML"
+        assert len(payload["text"]) <= TELEGRAM_MESSAGE_CHAR_LIMIT
+        url = args.args[0]
+        assert url.endswith("/sendMessage")
+    assert all("sendDocument" not in (args.args[0]) for args in mock_post.call_args_list)
+    assert all("sendDocument" not in str(args) for args in mock_post.call_args_list)
+
+
+def test_actions_and_source_use_html_not_markdown():
+    sink = _sink()
+    finding = Finding(
+        title="Alert",
+        description="desc",
+        aggregation_key="Alert",
+        severity=FindingSeverity.HIGH,
+        add_silence_url=True,
+    )
+    finding.add_link(Link(url="https://example.com/runbook", name="Runbook"))
+    sink.client = MagicMock()
+
+    sink.write_finding(finding, platform_enabled=True)
+
+    message = sink.client.send_message.call_args.args[0]
+    assert "<b>Alert</b>" in message
+    assert "<b>Source:</b>" in message
+    assert "<code>test-cluster</code>" in message
+    assert "<a href=" in message
+    assert "[Investigate]" not in message
+    assert "*Alert*" not in message
+    assert "MarkdownV2" not in message
+
+
+def test_markdown_blocks_in_enrichments_are_html():
+    sink = _sink()
+    finding = Finding(title="t", aggregation_key="k", description=None)
+    finding.add_enrichment([MarkdownBlock("check `foo & bar`")])
+    sink.client = MagicMock()
+
+    sink.write_finding(finding, platform_enabled=False)
+
+    message = sink.client.send_message.call_args.args[0]
+    assert "<code>foo &amp; bar</code>" in message