"""Tests for Telegram MarkdownV2 formatting in gateway/platforms/telegram.py.

Covers: _escape_mdv2 (pure function), format_message (markdown-to-MarkdownV2
conversion pipeline), and edge cases that could produce invalid MarkdownV2
or corrupt user-visible content.
"""

import re
import sys
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock

import pytest

from gateway.config import PlatformConfig


# ---------------------------------------------------------------------------
# Mock the telegram package if it's not installed
# ---------------------------------------------------------------------------

def _ensure_telegram_mock():
    if "telegram" in sys.modules and hasattr(sys.modules["telegram"], "__file__"):
        return
    mod = MagicMock()
    mod.ext.ContextTypes.DEFAULT_TYPE = type(None)
    mod.constants.ParseMode.MARKDOWN_V2 = "MarkdownV2"
    mod.constants.ChatType.GROUP = "group"
    mod.constants.ChatType.SUPERGROUP = "supergroup"
    mod.constants.ChatType.CHANNEL = "channel"
    mod.constants.ChatType.PRIVATE = "private"
    for name in ("telegram", "telegram.ext", "telegram.constants", "telegram.request"):
        sys.modules.setdefault(name, mod)


_ensure_telegram_mock()

from plugins.platforms.telegram.adapter import (  # noqa: E402
    TelegramAdapter,
    _escape_mdv2,
    _strip_mdv2,
    _wrap_markdown_tables,
)


# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------

@pytest.fixture()
def adapter():
    config = PlatformConfig(enabled=True, token="fake-token")
    return TelegramAdapter(config)


# =========================================================================
# _escape_mdv2
# =========================================================================


class TestEscapeMdv2:
    def test_escapes_all_special_characters(self):
        special = r'_*[]()~`>#+-=|{}.!\ '
        escaped = _escape_mdv2(special)
        # Every special char should be preceded by backslash
        for ch in r'_*[]()~`>#+-=|{}.!\  ':
            if ch == ' ':
                continue
            assert f'\\{ch}' in escaped

    def test_empty_string(self):
        assert _escape_mdv2("") == ""

    def test_no_special_characters(self):
        assert _escape_mdv2("hello world 123") == "hello world 123"

    def test_backslash_escaped(self):
        assert _escape_mdv2("a\\b") == "a\\\\b"

    def test_dot_escaped(self):
        assert _escape_mdv2("v2.0") == "v2\\.0"

    def test_exclamation_escaped(self):
        assert _escape_mdv2("wow!") == "wow\\!"

    def test_mixed_text_and_specials(self):
        result = _escape_mdv2("Hello (world)!")
        assert result == "Hello \\(world\\)\\!"


# =========================================================================
# format_message - basic conversions
# =========================================================================


class TestFormatMessageBasic:
    def test_empty_string(self, adapter):
        assert adapter.format_message("") == ""

    def test_none_input(self, adapter):
        # content is falsy, returned as-is
        assert adapter.format_message(None) is None

    def test_plain_text_specials_escaped(self, adapter):
        result = adapter.format_message("Price is $5.00!")
        assert "\\." in result
        assert "\\!" in result

    def test_plain_text_no_markdown(self, adapter):
        result = adapter.format_message("Hello world")
        assert result == "Hello world"


# =========================================================================
# format_message - code blocks
# =========================================================================


class TestFormatMessageCodeBlocks:
    def test_fenced_code_block_preserved(self, adapter):
        text = "Before\n```python\nprint('hello')\n```\nAfter"
        result = adapter.format_message(text)
        # Code block contents must NOT be escaped
        assert "```python\nprint('hello')\n```" in result
        # But "After" should have no escaping needed (plain text)
        assert "After" in result

    def test_inline_code_preserved(self, adapter):
        text = "Use `my_var` here"
        result = adapter.format_message(text)
        # Inline code content must NOT be escaped
        assert "`my_var`" in result
        # The surrounding text's underscore-free content should be fine
        assert "Use" in result

    def test_code_block_special_chars_not_escaped(self, adapter):
        text = "```\nif (x > 0) { return !x; }\n```"
        result = adapter.format_message(text)
        # Inside code block, > and ! and { should NOT be escaped
        assert "if (x > 0) { return !x; }" in result

    def test_inline_code_special_chars_not_escaped(self, adapter):
        text = "Run `rm -rf ./*` carefully"
        result = adapter.format_message(text)
        assert "`rm -rf ./*`" in result

    def test_multiple_code_blocks(self, adapter):
        text = "```\nblock1\n```\ntext\n```\nblock2\n```"
        result = adapter.format_message(text)
        assert "block1" in result
        assert "block2" in result
        # "text" between blocks should be present
        assert "text" in result

    def test_inline_code_backslashes_escaped(self, adapter):
        r"""Backslashes in inline code must be escaped for MarkdownV2."""
        text = r"Check `C:\ProgramData\VMware\` path"
        result = adapter.format_message(text)
        assert r"`C:\\ProgramData\\VMware\\`" in result

    def test_fenced_code_block_backslashes_escaped(self, adapter):
        r"""Backslashes in fenced code blocks must be escaped for MarkdownV2."""
        text = "```\npath = r'C:\\Users\\test'\n```"
        result = adapter.format_message(text)
        assert r"C:\\Users\\test" in result

    def test_fenced_code_block_backticks_escaped(self, adapter):
        r"""Backticks inside fenced code blocks must be escaped for MarkdownV2."""
        text = "```\necho `hostname`\n```"
        result = adapter.format_message(text)
        assert r"echo \`hostname\`" in result

    def test_inline_code_no_double_escape(self, adapter):
        r"""Already-escaped backslashes should not be quadruple-escaped."""
        text = r"Use `\\server\share`"
        result = adapter.format_message(text)
        # \\ in input → \\\\ in output (each \ escaped once)
        assert r"`\\\\server\\share`" in result


@pytest.mark.asyncio
async def test_legacy_send_keeps_chunk_indicators_outside_fenced_code_lines(adapter):
    """Chunk markers must not corrupt Telegram MarkdownV2 code fences.

    Telegram treats a closing fenced-code line with trailing text, e.g.
    ````` (1/2)``, as malformed MarkdownV2. The bot then falls back to plain
    text, which is the user-visible duplicate/malformed preview symptom.
    """
    adapter._bot = MagicMock()
    adapter._bot.send_message = AsyncMock(
        side_effect=[SimpleNamespace(message_id=i) for i in range(1, 20)]
    )
    adapter._bot.send_chat_action = AsyncMock()
    object.__setattr__(adapter, "MAX_MESSAGE_LENGTH", 120)
    adapter._rich_messages_enabled = False

    content = (
        "Intro before code block\n"
        "```text\n"
        + ("~/.hermes/skills/github/hermes-contribution-workflow/SKILL.md\n" * 8)
        + "```\n"
        "After."
    )

    result = await adapter.send("12345", content, metadata={"expect_edits": True})

    assert result.success is True
    sent_texts = [call.kwargs["text"] for call in adapter._bot.send_message.await_args_list]
    assert len(sent_texts) > 1
    for text in sent_texts:
        for line in text.splitlines():
            assert not re.match(r"^```\s+\\?\(\d+/\d+\\?\)$", line), text
            assert not re.match(r"^```\s+\(\d+/\d+\)$", line), text


@pytest.mark.asyncio
async def test_final_send_does_not_retrigger_typing(adapter):
    """The final reply (metadata['notify']) must NOT re-arm Telegram's typing
    timer. The gateway has already torn down the refresh loop by then, so a
    re-trigger here would leave the '...typing' bubble lingering after the
    answer (Telegram has no stop-typing API). See #48678."""
    adapter._bot = MagicMock()
    adapter._bot.send_message = AsyncMock(return_value=SimpleNamespace(message_id=1))
    adapter._bot.send_chat_action = AsyncMock()
    adapter._rich_messages_enabled = False

    result = await adapter.send("12345", "All done.", metadata={"notify": True})

    assert result.success is True
    adapter._bot.send_chat_action.assert_not_called()


@pytest.mark.asyncio
async def test_intermediate_send_still_retriggers_typing(adapter):
    """Intermediate/progress sends (no notify marker) keep re-triggering typing
    so the '...typing' bubble survives across progress messages while the agent
    is still working."""
    adapter._bot = MagicMock()
    adapter._bot.send_message = AsyncMock(return_value=SimpleNamespace(message_id=1))
    adapter._bot.send_chat_action = AsyncMock()
    adapter._rich_messages_enabled = False

    result = await adapter.send("12345", "Checking:", metadata={"expect_edits": True})

    assert result.success is True
    adapter._bot.send_chat_action.assert_awaited()


# =========================================================================
# format_message - bold and italic
# =========================================================================


class TestFormatMessageBoldItalic:
    def test_bold_converted(self, adapter):
        result = adapter.format_message("This is **bold** text")
        # MarkdownV2 bold uses single *
        assert "*bold*" in result
        # Original ** should be gone
        assert "**" not in result

    def test_italic_converted(self, adapter):
        result = adapter.format_message("This is *italic* text")
        # MarkdownV2 italic uses _
        assert "_italic_" in result

    def test_bold_with_special_chars(self, adapter):
        result = adapter.format_message("**hello.world!**")
        # Content inside bold should be escaped
        assert "*hello\\.world\\!*" in result

    def test_italic_with_special_chars(self, adapter):
        result = adapter.format_message("*hello.world*")
        assert "_hello\\.world_" in result

    def test_bold_and_italic_in_same_line(self, adapter):
        result = adapter.format_message("**bold** and *italic*")
        assert "*bold*" in result
        assert "_italic_" in result

    def test_reload_mcp_summary_escapes_dynamic_server_names(self, adapter):
        content = (
            "🔄 **MCP Servers Reloaded**\n"
            "♻️ Reconnected: agent_one, tool[beta]\n"
            "➕ Added: alpha*prod\n"
            "🔧 3 tool(s) available from 2 server(s)"
        )
        result = adapter.format_message(content)
        assert "*MCP Servers Reloaded*" in result
        assert "agent\\_one" in result
        assert "tool\\[beta\\]" in result
        assert "alpha\\*prod" in result


# =========================================================================
# format_message - headers
# =========================================================================


class TestFormatMessageHeaders:
    def test_h1_converted_to_bold(self, adapter):
        result = adapter.format_message("# Title")
        # Header becomes bold in MarkdownV2
        assert "*Title*" in result
        # Hash should be removed
        assert "#" not in result

    def test_h2_converted(self, adapter):
        result = adapter.format_message("## Subtitle")
        assert "*Subtitle*" in result

    def test_header_with_inner_bold_stripped(self, adapter):
        # Headers strip redundant **...** inside
        result = adapter.format_message("## **Important**")
        # Should be *Important* not ***Important***
        assert "*Important*" in result
        count = result.count("*")
        # Should have exactly 2 asterisks (open + close)
        assert count == 2

    def test_header_with_special_chars(self, adapter):
        result = adapter.format_message("# Hello (World)!")
        assert "\\(" in result
        assert "\\)" in result
        assert "\\!" in result

    def test_multiline_headers(self, adapter):
        text = "# First\nSome text\n## Second"
        result = adapter.format_message(text)
        assert "*First*" in result
        assert "*Second*" in result
        assert "Some text" in result


# =========================================================================
# format_message - links
# =========================================================================


class TestFormatMessageLinks:
    def test_markdown_link_converted(self, adapter):
        result = adapter.format_message("[Click here](https://example.com)")
        assert "[Click here](https://example.com)" in result

    def test_link_display_text_escaped(self, adapter):
        result = adapter.format_message("[Hello!](https://example.com)")
        # The ! in display text should be escaped
        assert "Hello\\!" in result

    def test_link_url_parentheses_escaped(self, adapter):
        result = adapter.format_message("[link](https://example.com/path_(1))")
        # The ) in URL should be escaped
        assert "\\)" in result

    def test_link_with_surrounding_text(self, adapter):
        result = adapter.format_message("Visit [Google](https://google.com) today.")
        assert "[Google](https://google.com)" in result
        assert "today\\." in result


# =========================================================================
# format_message - BUG: italic regex spans newlines
# =========================================================================


class TestItalicNewlineBug:
    r"""Italic regex ``\*([^*]+)\*`` matched across newlines, corrupting content.

    This affects bullet lists using * markers and any text where * appears
    at the end of one line and start of another.
    """

    def test_bullet_list_not_corrupted(self, adapter):
        """Bullet list items using * must NOT be merged into italic."""
        text = "* Item one\n* Item two\n* Item three"
        result = adapter.format_message(text)
        # Each item should appear in the output (not eaten by italic conversion)
        assert "Item one" in result
        assert "Item two" in result
        assert "Item three" in result
        # Should NOT contain _ (italic markers) wrapping list items
        assert "_" not in result or "Item" not in result.split("_")[1] if "_" in result else True

    def test_asterisk_list_items_preserved(self, adapter):
        """Each * list item should remain as a separate line, not become italic."""
        text = "* Alpha\n* Beta"
        result = adapter.format_message(text)
        # Both items must be present in output
        assert "Alpha" in result
        assert "Beta" in result
        # The text between first * and second * must NOT become italic
        lines = result.split("\n")
        assert len(lines) >= 2

    def test_italic_does_not_span_lines(self, adapter):
        """*text on\nmultiple lines* should NOT become italic."""
        text = "Start *across\nlines* end"
        result = adapter.format_message(text)
        # Should NOT have underscore italic markers wrapping cross-line text
        # If this fails, the italic regex is matching across newlines
        assert "_across\nlines_" not in result

    def test_single_line_italic_still_works(self, adapter):
        """Normal single-line italic must still convert correctly."""
        text = "This is *italic* text"
        result = adapter.format_message(text)
        assert "_italic_" in result


# =========================================================================
# format_message - strikethrough
# =========================================================================


class TestFormatMessageStrikethrough:
    def test_strikethrough_converted(self, adapter):
        result = adapter.format_message("This is ~~deleted~~ text")
        assert "~deleted~" in result
        assert "~~" not in result

    def test_strikethrough_with_special_chars(self, adapter):
        result = adapter.format_message("~~hello.world!~~")
        assert "~hello\\.world\\!~" in result

    def test_strikethrough_in_code_not_converted(self, adapter):
        result = adapter.format_message("`~~not struck~~`")
        assert "`~~not struck~~`" in result

    def test_strikethrough_with_bold(self, adapter):
        result = adapter.format_message("**bold** and ~~struck~~")
        assert "*bold*" in result
        assert "~struck~" in result


# =========================================================================
# format_message - spoiler
# =========================================================================


class TestFormatMessageSpoiler:
    def test_spoiler_converted(self, adapter):
        result = adapter.format_message("This is ||hidden|| text")
        assert "||hidden||" in result

    def test_spoiler_with_special_chars(self, adapter):
        result = adapter.format_message("||hello.world!||")
        assert "||hello\\.world\\!||" in result

    def test_spoiler_in_code_not_converted(self, adapter):
        result = adapter.format_message("`||not spoiler||`")
        assert "`||not spoiler||`" in result

    def test_spoiler_pipes_not_escaped(self, adapter):
        """The || delimiters must not be escaped as \\|\\|."""
        result = adapter.format_message("||secret||")
        assert "\\|\\|" not in result
        assert "||secret||" in result


# =========================================================================
# format_message - blockquote
# =========================================================================


class TestFormatMessageBlockquote:
    def test_blockquote_converted(self, adapter):
        result = adapter.format_message("> This is a quote")
        assert "> This is a quote" in result
        # > must NOT be escaped
        assert "\\>" not in result

    def test_blockquote_with_special_chars(self, adapter):
        result = adapter.format_message("> Hello (world)!")
        assert "> Hello \\(world\\)\\!" in result
        assert "\\>" not in result

    def test_blockquote_multiline(self, adapter):
        text = "> Line one\n> Line two"
        result = adapter.format_message(text)
        assert "> Line one" in result
        assert "> Line two" in result
        assert "\\>" not in result

    def test_blockquote_in_code_not_converted(self, adapter):
        result = adapter.format_message("```\n> not a quote\n```")
        assert "> not a quote" in result

    def test_nested_blockquote(self, adapter):
        result = adapter.format_message(">> Nested quote")
        assert ">> Nested quote" in result
        assert "\\>" not in result

    def test_gt_in_middle_of_line_still_escaped(self, adapter):
        """Only > at line start is a blockquote; mid-line > should be escaped."""
        result = adapter.format_message("5 > 3")
        assert "\\>" in result

    def test_expandable_blockquote(self, adapter):
        """Expandable blockquote prefix **> and trailing || must NOT be escaped."""
        result = adapter.format_message("**> Hidden content||")
        assert "**>" in result
        assert "||" in result
        assert "\\*" not in result  # asterisks in prefix must not be escaped
        assert "\\>" not in result  # > in prefix must not be escaped

    def test_single_asterisk_gt_not_blockquote(self, adapter):
        """Single asterisk before > should not be treated as blockquote prefix."""
        result = adapter.format_message("*> not a quote")
        assert "\\*" in result
        assert "\\>" in result

    def test_regular_blockquote_with_pipes_escaped(self, adapter):
        """Regular blockquote ending with || should escape the pipes."""
        result = adapter.format_message("> not expandable||")
        assert "> not expandable" in result
        assert "\\|" in result
        assert "\\>" not in result


# =========================================================================
# format_message - mixed/complex
# =========================================================================


class TestFormatMessageComplex:
    def test_code_block_with_bold_outside(self, adapter):
        text = "**Note:**\n```\ncode here\n```"
        result = adapter.format_message(text)
        assert "*Note:*" in result or "*Note\\:*" in result
        assert "```\ncode here\n```" in result

    def test_bold_inside_code_not_converted(self, adapter):
        """Bold markers inside code blocks should not be converted."""
        text = "```\n**not bold**\n```"
        result = adapter.format_message(text)
        assert "**not bold**" in result

    def test_link_inside_code_not_converted(self, adapter):
        text = "`[not a link](url)`"
        result = adapter.format_message(text)
        assert "`[not a link](url)`" in result

    def test_header_after_code_block(self, adapter):
        text = "```\ncode\n```\n## Title"
        result = adapter.format_message(text)
        assert "*Title*" in result
        assert "```\ncode\n```" in result

    def test_multiple_bold_segments(self, adapter):
        result = adapter.format_message("**a** and **b** and **c**")
        assert result.count("*") >= 6  # 3 bold pairs = 6 asterisks

    def test_special_chars_in_plain_text(self, adapter):
        result = adapter.format_message("Price: $5.00 (50% off!)")
        assert "\\." in result
        assert "\\(" in result
        assert "\\)" in result
        assert "\\!" in result

    def test_empty_bold(self, adapter):
        """**** (empty bold) should not crash."""
        result = adapter.format_message("****")
        assert result is not None

    def test_empty_code_block(self, adapter):
        result = adapter.format_message("```\n```")
        assert "```" in result

    def test_placeholder_collision(self, adapter):
        """Many formatting elements should not cause placeholder collisions."""
        text = (
            "# Header\n"
            "**bold1** *italic1* `code1`\n"
            "**bold2** *italic2* `code2`\n"
            "```\nblock\n```\n"
            "[link](https://url.com)"
        )
        result = adapter.format_message(text)
        # No placeholder tokens should leak into output
        assert "\x00" not in result
        # All elements should be present
        assert "Header" in result
        assert "block" in result
        assert "url.com" in result


# =========================================================================
# _strip_mdv2 — plaintext fallback
# =========================================================================


class TestStripMdv2:
    def test_removes_escape_backslashes(self):
        assert _strip_mdv2(r"hello\.world\!") == "hello.world!"

    def test_removes_bold_markers(self):
        assert _strip_mdv2("*bold text*") == "bold text"

    def test_removes_italic_markers(self):
        assert _strip_mdv2("_italic text_") == "italic text"

    def test_removes_both_bold_and_italic(self):
        result = _strip_mdv2("*bold* and _italic_")
        assert result == "bold and italic"

    def test_preserves_snake_case(self):
        assert _strip_mdv2("my_variable_name") == "my_variable_name"

    def test_preserves_multi_underscore_identifier(self):
        assert _strip_mdv2("some_func_call here") == "some_func_call here"

    def test_plain_text_unchanged(self):
        assert _strip_mdv2("plain text") == "plain text"

    def test_empty_string(self):
        assert _strip_mdv2("") == ""

    def test_removes_strikethrough_markers(self):
        assert _strip_mdv2("~struck text~") == "struck text"

    def test_removes_spoiler_markers(self):
        assert _strip_mdv2("||hidden text||") == "hidden text"


# =========================================================================
# Markdown table auto-wrap
# =========================================================================


class TestWrapMarkdownTables:
    """_wrap_markdown_tables rewrites GFM pipe tables into Telegram-friendly
    row groups instead of leaving noisy pipe syntax in the final message."""

    def test_basic_table_rewritten_as_row_groups(self):
        text = (
            "Scores:\n\n"
            "| Player | Score |\n"
            "|--------|-------|\n"
            "| Alice  | 150   |\n"
            "| Bob    | 120   |\n"
            "\nEnd."
        )
        out = _wrap_markdown_tables(text)
        assert "**Alice**" in out
        # The heading IS the Player cell — don't repeat it as a bullet.
        assert "• Player: Alice" not in out
        assert "• Score: 150" in out
        assert "**Bob**" in out
        assert "• Score: 120" in out
        # Heading and its bullet sit on consecutive lines (no blank between).
        assert "**Alice**\n• Score: 150" in out
        # Separate row groups ARE separated by a blank line.
        assert "• Score: 150\n\n**Bob**" in out
        # Surrounding prose is preserved
        assert out.startswith("Scores:")
        assert out.endswith("End.")

    def test_bare_pipe_table_rewritten(self):
        """Tables without outer pipes (GFM allows this) are still detected."""
        text = "head1 | head2\n--- | ---\na | b\nc | d"
        out = _wrap_markdown_tables(text)
        assert out.startswith("**a**")
        # No duplicate first bullet — heading 'a' already shows the head1 value.
        assert "• head1: a" not in out
        assert "• head2: b" in out
        assert "**c**" in out

    def test_alignment_separators(self):
        """Separator rows with :--- / ---: / :---: alignment markers match."""
        text = (
            "| Name | Age | City |\n"
            "|:-----|----:|:----:|\n"
            "| Ada  |  30 | NYC  |"
        )
        out = _wrap_markdown_tables(text)
        assert "**Ada**" in out
        # 'Ada' is the heading (first cell); skip the redundant Name bullet.
        assert "• Name: Ada" not in out
        assert "• Age: 30" in out
        assert "• City: NYC" in out
        # All three lines pack tightly with single newlines.
        assert "**Ada**\n• Age: 30\n• City: NYC" in out

    def test_two_consecutive_tables_rewritten_separately(self):
        text = (
            "| A | B |\n"
            "|---|---|\n"
            "| 1 | 2 |\n"
            "\n"
            "| X | Y |\n"
            "|---|---|\n"
            "| 9 | 8 |"
        )
        out = _wrap_markdown_tables(text)
        assert out.count("**1**") == 1
        assert out.count("**9**") == 1
        # Headings duplicate first cells (no row-label col) — skip those bullets.
        assert "• A: 1" not in out
        assert "• X: 9" not in out
        assert "• B: 2" in out
        assert "• Y: 8" in out

    def test_plain_text_with_pipes_not_wrapped(self):
        """A bare pipe in prose must NOT trigger wrapping."""
        text = "Use the | pipe operator to chain commands."
        assert _wrap_markdown_tables(text) == text

    def test_horizontal_rule_not_wrapped(self):
        """A lone '---' horizontal rule must not be mistaken for a separator."""
        text = "Section A\n\n---\n\nSection B"
        assert _wrap_markdown_tables(text) == text

    def test_existing_code_block_with_pipes_left_alone(self):
        """A table already inside a fenced code block must not be re-wrapped."""
        text = (
            "```\n"
            "| a | b |\n"
            "|---|---|\n"
            "| 1 | 2 |\n"
            "```"
        )
        assert _wrap_markdown_tables(text) == text

    def test_no_pipe_character_short_circuits(self):
        text = "Plain **bold** text with no table."
        assert _wrap_markdown_tables(text) == text

    def test_no_dash_short_circuits(self):
        text = "a | b\nc | d"  # has pipes but no '-' separator row
        assert _wrap_markdown_tables(text) == text

    def test_single_column_separator_not_matched(self):
        """Single-column tables (rare) are not detected — we require at
        least one internal pipe in the separator row to avoid false
        positives on formatting rules."""
        text = "| a |\n| - |\n| b |"
        assert _wrap_markdown_tables(text) == text

    def test_row_group_uses_single_newlines_within_group(self):
        """Regression: each bullet within a row-group must be separated by
        a single newline, not a blank line.  Telegram renders blank lines
        as paragraph breaks, which previously left every bullet floating in
        its own paragraph and made multi-column tables unreadable.

        Mirrors the exact pattern that produced the screenshot bug report:
        a five-column comparison table with no row-label column.
        """
        text = (
            "| Play | Capital | Build | $/day | Risk |\n"
            "|---|---|---|---|---|\n"
            "| A. Copy Hands (HK/SZ) | $5-10k | 2 wk | $30-70 | Low |\n"
            "| B. NO-sweeper        | $50-100k | 3 wk | $300-1000 | Med |"
        )
        out = _wrap_markdown_tables(text)

        # No bullet sits inside its own paragraph: the substring "\n\n• "
        # would mean a blank line precedes a bullet, which is the bug.
        assert "\n\n• " not in out

        # The two row-groups DO have a paragraph break between them.
        groups = [g for g in out.split("\n\n") if g.strip()]
        assert len(groups) == 2
        # Heading + 4 bullets per group means each group is exactly 5 lines.
        for group in groups:
            line_count = group.count("\n") + 1
            assert line_count == 5, (
                "Each row-group should be 5 lines (heading + 4 bullets), "
                f"got {line_count}:\n{group}"
            )

    def test_row_label_column_preserves_first_bullet(self):
        """When the table has a row-label column (data rows have one more
        cell than the header row), the heading comes from the label cell
        and is distinct from any header — so every header→value bullet is
        kept, including the first one."""
        text = (
            "|        | Score | Rank |\n"
            "|--------|-------|------|\n"
            "| Alice  | 150   | 1    |\n"
            "| Bob    | 120   | 2    |\n"
        )
        out = _wrap_markdown_tables(text)
        assert "**Alice**" in out
        # No header to duplicate against — both bullets stay.
        assert "• Score: 150" in out
        assert "• Rank: 1" in out
        assert "**Alice**\n• Score: 150\n• Rank: 1" in out


class TestFormatMessageTables:
    """End-to-end: pipe tables become readable Telegram-native text instead
    of escaped pipe syntax or fenced code blocks."""

    def test_table_rendered_as_bullets(self, adapter):
        text = (
            "Data:\n\n"
            "| Col1 | Col2 |\n"
            "|------|------|\n"
            "| A    | B    |\n"
        )
        out = adapter.format_message(text)
        assert "*A*" in out
        # Heading 'A' duplicates the Col1 value — skip that bullet.
        assert "• Col1: A" not in out
        assert "• Col2: B" in out
        assert "```" not in out
        assert "\\|" not in out

    def test_text_after_table_still_formatted(self, adapter):
        text = (
            "| A | B |\n"
            "|---|---|\n"
            "| 1 | 2 |\n"
            "\n"
            "Nice **work** team!"
        )
        out = adapter.format_message(text)
        # MarkdownV2 bold conversion still happens outside the table
        assert "*work*" in out
        # Exclamation outside fence is escaped
        assert "\\!" in out
        assert "*1*" in out
        # Heading '1' is also the A-column value — skip the redundant bullet.
        assert "• A: 1" not in out
        assert "• B: 2" in out

    def test_multiple_tables_in_single_message(self, adapter):
        text = (
            "First:\n"
            "| A | B |\n"
            "|---|---|\n"
            "| 1 | 2 |\n"
            "\n"
            "Second:\n"
            "| X | Y |\n"
            "|---|---|\n"
            "| 9 | 8 |\n"
        )
        out = adapter.format_message(text)
        assert out.count("*1*") == 1
        assert out.count("*9*") == 1
        assert "• Y: 8" in out


@pytest.mark.asyncio
async def test_send_escapes_chunk_indicator_for_markdownv2(adapter):
    adapter.MAX_MESSAGE_LENGTH = 80
    adapter._bot = MagicMock()

    sent_texts = []

    async def _fake_send_message(**kwargs):
        sent_texts.append(kwargs["text"])
        msg = MagicMock()
        msg.message_id = len(sent_texts)
        return msg

    adapter._bot.send_message = AsyncMock(side_effect=_fake_send_message)

    content = ("**bold** chunk content " * 12).strip()
    result = await adapter.send("123", content)

    assert result.success is True
    assert len(sent_texts) > 1
    assert re.search(r" \\\([0-9]+/[0-9]+\\\)$", sent_texts[0])
    assert re.search(r" \\\([0-9]+/[0-9]+\\\)$", sent_texts[-1])


# =========================================================================
# edit_message — streaming Markdown safety
# =========================================================================


class TestEditMessageStreamingSafety:
    @pytest.mark.asyncio
    async def test_non_final_edit_uses_plain_text_without_markdown(self):
        adapter = TelegramAdapter(PlatformConfig(enabled=True, token="fake-token"))
        adapter._bot = MagicMock()
        adapter._bot.edit_message_text = AsyncMock()

        result = await adapter.edit_message("123", "456", "partial **bold", finalize=False)

        assert result.success is True
        adapter._bot.edit_message_text.assert_awaited_once_with(
            chat_id=123,
            message_id=456,
            text="partial **bold",
        )

    @pytest.mark.asyncio
    async def test_final_edit_uses_markdownv2_with_plain_fallback(self):
        adapter = TelegramAdapter(PlatformConfig(enabled=True, token="fake-token"))
        adapter._bot = MagicMock()
        adapter._bot.edit_message_text = AsyncMock(side_effect=[Exception("bad markdown"), None])

        result = await adapter.edit_message("123", "456", "final **bold**", finalize=True)

        assert result.success is True
        first_call = adapter._bot.edit_message_text.await_args_list[0].kwargs
        second_call = adapter._bot.edit_message_text.await_args_list[1].kwargs
        assert "parse_mode" in first_call
        assert first_call["text"] == "final *bold*"
        assert second_call == {
            "chat_id": 123,
            "message_id": 456,
            "text": "final bold",
        }

    @pytest.mark.asyncio
    async def test_message_too_long_splits_into_continuations_not_silent_truncation(self):
        """When edit_message_text exceeds Telegram's 4096 UTF-16 limit on
        finalize, the adapter must split the content across the existing
        message + new continuation messages so the user gets the full reply.
        Previously the adapter best-effort truncated the content with '…' and
        returned success=True, dropping everything past the truncation
        boundary (#19537)."""
        adapter = TelegramAdapter(PlatformConfig(enabled=True, token="fake-token"))
        adapter._bot = MagicMock()
        adapter._bot.edit_message_text = AsyncMock()
        # Continuation sends return monotonically increasing message ids.
        _next_id = [1000]
        async def _fake_send(**kwargs):
            _next_id[0] += 1
            return SimpleNamespace(message_id=_next_id[0])
        adapter._bot.send_message = AsyncMock(side_effect=_fake_send)

        # 6000-char content well over the 4096 UTF-16 limit.
        oversized = "x" * 6000
        result = await adapter.edit_message("123", "456", oversized, finalize=True)

        # Adapter reports success with continuations populated.
        assert result.success is True
        assert result.error is None
        assert len(result.continuation_message_ids) >= 1, (
            "expected at least one continuation message"
        )
        # The reported message_id is the LAST visible message (the final
        # continuation), so subsequent edits target the most recent.
        assert result.message_id == result.continuation_message_ids[-1]
        # Original message_id (456) was edited with chunk 1.
        first_edit = adapter._bot.edit_message_text.call_args
        assert first_edit.kwargs["message_id"] == 456
        # Continuations were sent threaded as replies for visual grouping.
        assert adapter._bot.send_message.await_count == len(result.continuation_message_ids)

    @pytest.mark.asyncio
    async def test_message_too_long_continuations_preserve_topic_metadata(self):
        """Overflow continuations should stay in the originating Telegram topic."""
        adapter = TelegramAdapter(PlatformConfig(enabled=True, token="fake-token"))
        adapter._bot = MagicMock()
        adapter._bot.edit_message_text = AsyncMock()
        sent_kwargs = []

        async def _fake_send(**kwargs):
            sent_kwargs.append(kwargs)
            return SimpleNamespace(message_id=1000 + len(sent_kwargs))

        adapter._bot.send_message = AsyncMock(side_effect=_fake_send)

        result = await adapter.edit_message(
            "-100123",
            "456",
            "x" * 6000,
            finalize=True,
            metadata={"thread_id": "17585"},
        )

        assert result.success is True
        assert sent_kwargs, "expected at least one overflow continuation"
        assert all(kwargs.get("message_thread_id") == 17585 for kwargs in sent_kwargs)
        assert sent_kwargs[0]["reply_to_message_id"] == 456

    @pytest.mark.asyncio
    async def test_mid_stream_overflow_truncates_instead_of_splitting(self):
        """During streaming (finalize=False), oversized content must be
        truncated to fit one message rather than split into continuations.
        Splitting mid-stream creates new message IDs that the stream consumer
        then edits with the full accumulated text, causing an infinite
        duplication loop (#48648)."""
        adapter = TelegramAdapter(PlatformConfig(enabled=True, token="fake-token"))
        adapter._bot = MagicMock()
        adapter._bot.edit_message_text = AsyncMock()
        _next_id = [1000]

        async def _fake_send(**kwargs):
            _next_id[0] += 1
            return SimpleNamespace(message_id=_next_id[0])

        adapter._bot.send_message = AsyncMock(side_effect=_fake_send)

        oversized = "x" * 6000
        result = await adapter.edit_message("123", "456", oversized, finalize=False)

        # Must NOT create continuation messages during streaming.
        assert result.success is True
        assert adapter._bot.send_message.await_count == 0, (
            "mid-stream overflow must not send continuation messages"
        )
        # message_id stays the original — no shift to a new ID.
        assert result.message_id == "456"
        # The edit must contain truncated content (≤ MAX_MESSAGE_LENGTH).
        edited_text = adapter._bot.edit_message_text.call_args.kwargs["text"]
        assert len(edited_text) <= adapter.MAX_MESSAGE_LENGTH

    @pytest.mark.asyncio
    async def test_saturated_preview_dedups_repeat_oversized_edits(self):
        """Once a mid-stream preview saturates at the truncation cap, further
        oversized edits truncate to the SAME text — re-sending them is a
        visual no-op that burns Telegram's flood budget (~1 edit/0.8s for the
        rest of a long stream ⇒ flood control with 200s+ penalties, hanging
        final delivery). The adapter must skip identical saturated previews
        without an API call."""
        adapter = TelegramAdapter(PlatformConfig(enabled=True, token="fake-token"))
        adapter._bot = MagicMock()
        adapter._bot.edit_message_text = AsyncMock()
        adapter._bot.send_message = AsyncMock()

        # First oversized edit: delivers the truncated preview (1 API call).
        r1 = await adapter.edit_message("123", "456", "x" * 6000, finalize=False)
        assert r1.success is True
        assert adapter._bot.edit_message_text.await_count == 1

        # Stream keeps growing within the same chunk count: previews truncate
        # identically — no API calls. (7000 and 8000 chars both truncate to
        # the same "…(1/2)" preview; 9000 crosses into "(1/3)" — a real change
        # that SHOULD be delivered, at most one edit per ~4096 chars of growth
        # instead of one per 0.8s tick.)
        for grow in (7000, 8000):
            r = await adapter.edit_message("123", "456", "x" * grow, finalize=False)
            assert r.success is True
            assert r.message_id == "456"
        assert adapter._bot.edit_message_text.await_count == 1, (
            "identical saturated previews must not be re-sent"
        )
        # Chunk-count boundary: marker changes (1/2 → 1/3) — one real edit.
        await adapter.edit_message("123", "456", "x" * 9000, finalize=False)
        assert adapter._bot.edit_message_text.await_count == 2
        # ...and saturates again at the new marker.
        await adapter.edit_message("123", "456", "x" * 9100, finalize=False)
        assert adapter._bot.edit_message_text.await_count == 2

        # A DIFFERENT oversized prefix (content changed within the first 4096)
        # must still go through.
        await adapter.edit_message("123", "456", "y" * 9100, finalize=False)
        assert adapter._bot.edit_message_text.await_count == 3

    @pytest.mark.asyncio
    async def test_saturated_preview_state_cleared_on_finalize(self):
        """finalize=True delivers full content (split) and clears saturation
        state, so a reused message id can't be masked by stale dedup."""
        adapter = TelegramAdapter(PlatformConfig(enabled=True, token="fake-token"))
        adapter._bot = MagicMock()
        adapter._bot.edit_message_text = AsyncMock()
        _next_id = [1000]

        async def _fake_send(**kwargs):
            _next_id[0] += 1
            return SimpleNamespace(message_id=_next_id[0])

        adapter._bot.send_message = AsyncMock(side_effect=_fake_send)

        await adapter.edit_message("123", "456", "x" * 6000, finalize=False)
        assert ("123", "456") in adapter._last_overflow_preview

        result = await adapter.edit_message("123", "456", "x" * 6000, finalize=True)
        assert result.success is True
        # Finalize split-delivered (edit + continuation) and cleared the state.
        assert adapter._bot.send_message.await_count >= 1
        assert ("123", "456") not in adapter._last_overflow_preview

    @pytest.mark.asyncio
    async def test_saturation_state_cleared_when_content_shrinks(self):
        """A same-id edit back under the cap (segment reset) clears saturation
        state so later oversized edits aren't wrongly deduped."""
        adapter = TelegramAdapter(PlatformConfig(enabled=True, token="fake-token"))
        adapter._bot = MagicMock()
        adapter._bot.edit_message_text = AsyncMock()
        adapter._bot.send_message = AsyncMock()

        await adapter.edit_message("123", "456", "x" * 6000, finalize=False)
        assert ("123", "456") in adapter._last_overflow_preview
        await adapter.edit_message("123", "456", "short", finalize=False)
        assert ("123", "456") not in adapter._last_overflow_preview
        # Oversized again → must be delivered, not deduped.
        await adapter.edit_message("123", "456", "x" * 6000, finalize=False)
        assert adapter._bot.edit_message_text.await_count == 3

    @pytest.mark.asyncio
    async def test_mid_stream_reactive_overflow_retries_truncated_edit(self):
        """If Telegram rejects a streaming edit as too long, retry with a
        one-message preview instead of splitting into continuations."""
        adapter = TelegramAdapter(PlatformConfig(enabled=True, token="fake-token"))
        adapter._bot = MagicMock()
        adapter._bot.edit_message_text = AsyncMock(
            side_effect=[Exception("Bad Request: message is too long"), None]
        )
        adapter._bot.send_message = AsyncMock()

        content = "x" * adapter.MAX_MESSAGE_LENGTH
        result = await adapter.edit_message("123", "456", content, finalize=False)

        assert result.success is True
        assert result.message_id == "456"
        adapter._bot.send_message.assert_not_called()
        assert adapter._bot.edit_message_text.await_count == 2
        retry_text = adapter._bot.edit_message_text.await_args_list[1].kwargs["text"]
        assert len(retry_text) <= adapter.MAX_MESSAGE_LENGTH


# =========================================================================
# Telegram guest mention gating
# =========================================================================


def _guest_test_adapter(*, guest_mode=True, require_mention=True, allowed_chats=None):
    config = PlatformConfig(
        enabled=True,
        token="fake-token",
        extra={
            "guest_mode": guest_mode,
            "require_mention": require_mention,
            "allowed_chats": allowed_chats or ["-100200"],
        },
    )
    adapter = object.__new__(TelegramAdapter)
    adapter.config = config
    adapter._bot = SimpleNamespace(id=999, username="hermes_bot")
    adapter._mention_patterns = adapter._compile_mention_patterns()
    # PR db50af910 added a TELEGRAM_ALLOWED_USERS allowlist gate to
    # _should_process_message. These tests aren't exercising the auth
    # gate — they're exercising the guest-mode mention/allowed_chats
    # logic that runs after — so stub the user authz to always allow.
    adapter._is_callback_user_authorized = lambda *_a, **_kw: True
    return adapter


def _guest_group_message(text, *, chat_id=-100201, entities=None, reply_to_bot=False):
    reply_to_message = SimpleNamespace(from_user=SimpleNamespace(id=999)) if reply_to_bot else None
    return SimpleNamespace(
        text=text,
        caption=None,
        entities=entities or [],
        caption_entities=[],
        message_thread_id=None,
        chat=SimpleNamespace(id=chat_id, type="group"),
        from_user=SimpleNamespace(id=111),
        reply_to_message=reply_to_message,
    )


def _guest_mention_entity(text, mention="@hermes_bot"):
    return SimpleNamespace(type="mention", offset=text.index(mention), length=len(mention))


class TestTelegramGuestMentionGating:
    def test_guest_mode_allows_explicit_mention_outside_allowed_chats(self):
        adapter = _guest_test_adapter(guest_mode=True, allowed_chats=["-100200"])
        text = "please help @hermes_bot"
        message = _guest_group_message(
            text,
            chat_id=-100201,
            entities=[_guest_mention_entity(text)],
        )

        assert adapter._should_process_message(message) is True

    def test_guest_mode_does_not_allow_reply_outside_allowed_chats(self):
        adapter = _guest_test_adapter(guest_mode=True, allowed_chats=["-100200"])
        message = _guest_group_message("replying without mention", chat_id=-100201, reply_to_bot=True)

        assert adapter._should_process_message(message) is False

    def test_guest_mode_disabled_keeps_allowed_chats_as_hard_gate_for_mentions(self):
        adapter = _guest_test_adapter(guest_mode=False, allowed_chats=["-100200"])
        text = "please help @hermes_bot"
        message = _guest_group_message(
            text,
            chat_id=-100201,
            entities=[_guest_mention_entity(text)],
        )

        assert adapter._should_process_message(message) is False

    def test_guest_mode_allows_bot_command_entity_outside_allowed_chats(self):
        """``/cmd@botname`` is a ``bot_command`` entity, not ``mention``."""
        adapter = _guest_test_adapter(guest_mode=True, allowed_chats=["-100200"])
        text = "/status@hermes_bot"
        message = _guest_group_message(
            text,
            chat_id=-100201,
            entities=[SimpleNamespace(type="bot_command", offset=0, length=len(text))],
        )

        assert adapter._should_process_message(message) is True

    def test_guest_mode_allows_text_mention_entity_outside_allowed_chats(self):
        """MessageEntity(type=text_mention) tags a user by ID — recognised as mention."""
        adapter = _guest_test_adapter(guest_mode=True, allowed_chats=["-100200"])
        message = _guest_group_message(
            "hey there",
            chat_id=-100201,
            entities=[SimpleNamespace(type="text_mention", offset=0, length=3, user=SimpleNamespace(id=999))],
        )

        assert adapter._should_process_message(message) is True

    def test_guest_mode_allows_mention_in_caption_outside_allowed_chats(self):
        """Media caption @mention should bypass allowed_chats via guest_mode."""
        adapter = _guest_test_adapter(guest_mode=True, allowed_chats=["-100200"])
        text = "look @hermes_bot"
        message = _guest_group_message(
            text="",
            chat_id=-100201,
            entities=[],
        )
        message.caption = text
        message.caption_entities = [_guest_mention_entity(text)]

        assert adapter._should_process_message(message) is True
