From a9c73f178033399980087f18dea18200ad6a24ab Mon Sep 17 00:00:00 2001
From: Tayo <contact+tayo@rezhajulio.id>
Date: Sat, 17 Oct 2026 05:09:07 +0700
Subject: [PATCH 1/8] Fix: detect rich-message spam bypassing text-based
 filters

Rich messages (Message.rich_message) arrive with empty text while their
payload hides in page blocks, so the AI spam monitor's text/caption-only
filter never classified them and the inline-keyboard URL check (which
only inspects reply_markup) missed their URL buttons. This let the
2026-10-17 adult-spam forward (t.me/base6d1ef16_bot) sit in the group
for hours until a manual admin delete.

- New bot.services.rich_message: duck-typed walker extracting plain
  text (headings, paragraphs, table cells, button labels, custom emoji
  alt text) and deduplicated URLs (buttons, text links, login/web-app)
  from rich message blocks.
- Group 1 (inline keyboard spam) now also auto-deletes rich messages
  carrying non-whitelisted URLs, with its own notification wording.
- ai_spam_monitor now includes rich text + URLs in the classified text,
  so URL-less rich spam still reaches the classifier and admin alerts.
- 21 new tests; full suite 1320 passed, ruff + mypy clean.
---
 src/bot/constants.py                |  16 ++
 src/bot/handlers/ai_spam_monitor.py |  17 +-
 src/bot/handlers/anti_spam.py       |  44 ++++-
 src/bot/services/rich_message.py    |  89 +++++++++
 tests/test_rich_message.py          | 279 ++++++++++++++++++++++++++++
 5 files changed, 439 insertions(+), 6 deletions(-)
 create mode 100644 src/bot/services/rich_message.py
 create mode 100644 tests/test_rich_message.py

diff --git a/src/bot/constants.py b/src/bot/constants.py
index fb0a585..8693934 100644
--- a/src/bot/constants.py
+++ b/src/bot/constants.py
@@ -361,6 +361,22 @@ def format_hours_display(hours: int) -> str:
     "📌 [Peraturan Grup]({rules_link})"
 )
 
+# Rich message spam notification
+RICH_MESSAGE_SPAM_NOTIFICATION = (
+    "🚫 *Spam Terdeteksi*\n\n"
+    "Pesan dari {user_mention} telah dihapus karena mengandung "
+    "pesan rich text dengan tautan mencurigakan.\n\n"
+    "Pengguna telah dibatasi.\n\n"
+    "📌 [Peraturan Grup]({rules_link})"
+)
+
+RICH_MESSAGE_SPAM_NOTIFICATION_NO_RESTRICT = (
+    "🚫 *Spam Terdeteksi*\n\n"
+    "Pesan dari {user_mention} telah dihapus karena mengandung "
+    "pesan rich text dengan tautan mencurigakan.\n\n"
+    "📌 [Peraturan Grup]({rules_link})"
+)
+
 # Contact spam notification
 CONTACT_SPAM_NOTIFICATION = (
     "🚫 *Kontak Dihapus*\n\n"
diff --git a/src/bot/handlers/ai_spam_monitor.py b/src/bot/handlers/ai_spam_monitor.py
index e21c381..ec1f161 100644
--- a/src/bot/handlers/ai_spam_monitor.py
+++ b/src/bot/handlers/ai_spam_monitor.py
@@ -56,6 +56,7 @@
     daily_budget_exhausted,
     try_spend_budget,
 )
+from bot.services.rich_message import extract_rich_text, extract_rich_urls
 from bot.services.telegram_utils import (
     is_user_admin_in_group,
     is_user_admin_or_trusted,
@@ -66,8 +67,20 @@
 logger = logging.getLogger(__name__)
 
 def _classifiable_text(message: Message) -> str:
-    """Return the text available for classification: body text or media caption."""
-    return message.text or message.caption or ""
+    """Return the text available for classification.
+
+    Covers body text, media captions, and rich message content: rich
+    messages arrive with empty ``text`` while their payload (headings,
+    table cells, button labels and URLs) lives in ``rich_message``
+    blocks, so that content is extracted too.
+    """
+    parts = [message.text or "", message.caption or ""]
+    rich_message = getattr(message, "rich_message", None)
+    rich_text = extract_rich_text(rich_message)
+    if rich_text:
+        parts.append(rich_text)
+    parts.extend(extract_rich_urls(rich_message))
+    return " ".join(part for part in parts if part)
 
 
 def ai_spam_filter(update: Update) -> bool:
diff --git a/src/bot/handlers/anti_spam.py b/src/bot/handlers/anti_spam.py
index b29cb7b..9f25b22 100644
--- a/src/bot/handlers/anti_spam.py
+++ b/src/bot/handlers/anti_spam.py
@@ -4,6 +4,7 @@
 This module enforces anti-spam rules including:
 - Contact card spam detection (all members)
 - Inline keyboard URL spam detection (all members)
+- Rich message URL spam detection (all members)
 - Probation enforcement for new users (forwards, links, external replies, stories, media)
 """
 
@@ -23,10 +24,13 @@
     NEW_USER_SPAM_RESTRICTION,
     NEW_USER_SPAM_WARNING,
     RESTRICTED_PERMISSIONS,
+    RICH_MESSAGE_SPAM_NOTIFICATION,
+    RICH_MESSAGE_SPAM_NOTIFICATION_NO_RESTRICT,
     format_hours_display,
 )
 from bot.database.service import get_database
 from bot.group_config import get_group_config_for_update
+from bot.services.rich_message import extract_rich_urls
 from bot.services.telegram_utils import (
     get_user_mention,
     is_url_whitelisted,
@@ -221,6 +225,24 @@ def has_non_whitelisted_inline_keyboard_urls(message: Message) -> bool:
     return False
 
 
+def has_non_whitelisted_rich_message_urls(message: Message) -> bool:
+    """Check if a message's rich content carries non-whitelisted URLs.
+
+    Rich messages (``Message.rich_message``) cannot be composed from
+    regular Telegram clients either, so the same policy as inline
+    keyboard spam applies: URL buttons and text links pointing to
+    non-whitelisted domains are treated as spam.
+
+    Args:
+        message: Telegram message to check.
+
+    Returns:
+        bool: True if any rich block URL is not whitelisted.
+    """
+    urls = extract_rich_urls(getattr(message, "rich_message", None))
+    return any(not is_url_whitelisted(url) for url in urls)
+
+
 def has_contact(message: Message) -> bool:
     """
     Check if a message contains a contact card.
@@ -364,16 +386,30 @@ async def handle_inline_keyboard_spam(
     update: Update, context: HandlerContext
 ) -> None:
     """
-    Handle spam messages containing inline keyboard buttons with non-whitelisted URLs.
+    Handle spam messages containing non-whitelisted URLs in buttons.
 
-    Regular users cannot create inline keyboards from the Telegram client.
-    Any group message with inline keyboard URL buttons pointing to
-    non-whitelisted domains is treated as spam.
+    Regular users cannot create inline keyboards — or rich messages —
+    from the Telegram client. Any group message with inline keyboard
+    URL buttons or rich message blocks pointing to non-whitelisted
+    domains is treated as spam: the message is deleted and the sender
+    restricted.
 
     Args:
         update: Telegram update containing the message.
         context: Bot context with helper methods.
     """
+    message = update.message or update.edited_message
+    if message is not None and has_non_whitelisted_rich_message_urls(message):
+        await _handle_group_spam(
+            update,
+            context,
+            detector=has_non_whitelisted_rich_message_urls,
+            label="rich message spam",
+            should_restrict=lambda cfg: True,
+            template_restricted=RICH_MESSAGE_SPAM_NOTIFICATION,
+            template_no_restrict=RICH_MESSAGE_SPAM_NOTIFICATION_NO_RESTRICT,
+        )
+        return
     await _handle_group_spam(
         update,
         context,
diff --git a/src/bot/services/rich_message.py b/src/bot/services/rich_message.py
new file mode 100644
index 0000000..770f7ea
--- /dev/null
+++ b/src/bot/services/rich_message.py
@@ -0,0 +1,89 @@
+"""Extract searchable text and URLs from Telegram rich messages.
+
+Rich messages (``Message.rich_message``, Bot API 10.1+/aiogram 3.29+)
+carry their payload in page blocks instead of ``message.text``. Spam
+filters that only read ``text``/``caption`` are blind to them — a rich
+spam message arrives with empty text while its links and buttons hide
+in the block tree. This module walks that tree and returns plain text
+and URLs for the spam detectors.
+
+The walker is duck-typed on purpose: aiogram exposes ~25 block types
+and a ~30-variant rich-text union, and a newly added variant must not
+silently become invisible to the filters again.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+# Attributes that hold nested blocks, cells, buttons, or list items.
+_CONTAINER_ATTRS = ("blocks", "cells", "buttons", "items")
+# Text-bearing attributes besides the generic ``text``.
+_TEXT_ATTRS = ("caption", "summary", "alternative_text")
+
+
+def _walk(node: Any, texts: list[str], urls: list[str]) -> None:
+    """Recursively collect text fragments and URLs from a rich node."""
+    if node is None:
+        return
+    if isinstance(node, str):
+        stripped = node.strip()
+        if stripped:
+            texts.append(stripped)
+        return
+    if isinstance(node, (list, tuple)):
+        for item in node:
+            _walk(item, texts, urls)
+        return
+
+    url = getattr(node, "url", None)
+    if isinstance(url, str) and url.strip():
+        urls.append(url.strip())
+    for nested in ("login_url", "web_app"):
+        nested_url = getattr(getattr(node, nested, None), "url", None)
+        if isinstance(nested_url, str) and nested_url.strip():
+            urls.append(nested_url.strip())
+
+    for attr in ("text",) + _TEXT_ATTRS:
+        value = getattr(node, attr, None)
+        if value is not None:
+            _walk(value, texts, urls)
+
+    for attr in _CONTAINER_ATTRS:
+        value = getattr(node, attr, None)
+        if value:
+            _walk(value, texts, urls)
+
+
+def extract_rich_text(rich_message: Any | None) -> str:
+    """Return searchable plain text from a rich message, or "" when none.
+
+    Covers headings, paragraphs, table cells, button labels, and custom
+    emoji alternative text. URLs are intentionally excluded here — use
+    :func:`extract_rich_urls` for those.
+    """
+    if rich_message is None:
+        return ""
+    texts: list[str] = []
+    try:
+        _walk(rich_message, texts, [])
+    except Exception:
+        logger.warning("extract_rich_text: failed to walk rich message", exc_info=True)
+        return ""
+    return " ".join(texts)
+
+
+def extract_rich_urls(rich_message: Any | None) -> list[str]:
+    """Return deduplicated URLs from a rich message's buttons and text links."""
+    if rich_message is None:
+        return []
+    urls: list[str] = []
+    try:
+        _walk(rich_message, [], urls)
+    except Exception:
+        logger.warning("extract_rich_urls: failed to walk rich message", exc_info=True)
+        return []
+    return list(dict.fromkeys(urls))
diff --git a/tests/test_rich_message.py b/tests/test_rich_message.py
new file mode 100644
index 0000000..98fca64
--- /dev/null
+++ b/tests/test_rich_message.py
@@ -0,0 +1,279 @@
+"""Tests for rich message text/URL extraction and rich spam detection."""
+
+from datetime import UTC, datetime
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from aiogram.types import (
+    Chat,
+    Message,
+    RichBlockButtons,
+    RichBlockParagraph,
+    RichBlockSectionHeading,
+    RichBlockTable,
+    RichBlockTableCell,
+    RichMessage,
+    RichMessageButton,
+    RichTextBold,
+    RichTextCustomEmoji,
+    RichTextUrl,
+    User,
+)
+
+from bot.dispatch import AppState, HandlerContext, StopPropagation
+from bot.group_config import GroupConfig
+from bot.handlers.ai_spam_monitor import _classifiable_text
+from bot.handlers.anti_spam import (
+    handle_inline_keyboard_spam,
+    has_non_whitelisted_rich_message_urls,
+)
+from bot.services.rich_message import extract_rich_text, extract_rich_urls
+
+SPAM_URL = "https://t.me/base6d1ef16_bot?startapp=138762"
+
+
+def _spam_blocks():
+    """Rich blocks mirroring the 2026-10-17 adult-spam incident."""
+    return [
+        RichBlockSectionHeading(text="HOT 2026 DROP", size=1),
+        RichBlockParagraph(
+            text=[
+                RichTextBold(text="FRESH LEAKS"),
+                " ",
+                RichTextCustomEmoji(
+                    custom_emoji_id="5375191900686927814", alternative_text="🍑"
+                ),
+            ]
+        ),
+        RichBlockTable(
+            cells=[
+                [
+                    RichBlockTableCell(
+                        text=RichTextUrl(text="BIKINI SHOWS", url=SPAM_URL),
+                        align="center",
+                        valign="middle",
+                    ),
+                    RichBlockTableCell(
+                        text="EXCLUSIVE PICS", align="center", valign="middle"
+                    ),
+                ]
+            ]
+        ),
+        RichBlockButtons(
+            buttons=[
+                RichMessageButton(text="OPEN LINK", url=SPAM_URL),
+                RichMessageButton(text="ENTER NOW", url=SPAM_URL),
+            ]
+        ),
+    ]
+
+
+def _make_message(blocks=None, text=None):
+    return Message(
+        message_id=1,
+        date=datetime.now(UTC),
+        chat=Chat(id=-100123456, type="supergroup"),
+        text=text,
+        rich_message=RichMessage(blocks=blocks) if blocks is not None else None,
+    )
+
+
+class TestExtractRichText:
+    def test_extracts_heading_paragraph_table_and_buttons(self):
+        text = extract_rich_text(RichMessage(blocks=_spam_blocks()))
+        for expected in (
+            "HOT 2026 DROP",
+            "FRESH LEAKS",
+            "🍑",
+            "BIKINI SHOWS",
+            "EXCLUSIVE PICS",
+            "OPEN LINK",
+            "ENTER NOW",
+        ):
+            assert expected in text
+
+    def test_none_returns_empty(self):
+        assert extract_rich_text(None) == ""
+
+    def test_empty_blocks_returns_empty(self):
+        assert extract_rich_text(RichMessage(blocks=[])) == ""
+
+    def test_plain_string_blocks(self):
+        text = extract_rich_text(
+            RichMessage(blocks=[RichBlockParagraph(text="hello world")])
+        )
+        assert text == "hello world"
+
+    def test_walk_failure_returns_empty(self):
+        class _Boom:
+            @property
+            def url(self):
+                raise RuntimeError("boom")
+
+        assert extract_rich_text(_Boom()) == ""
+
+
+class TestExtractRichUrls:
+    def test_extracts_button_and_text_link_urls_deduped(self):
+        urls = extract_rich_urls(RichMessage(blocks=_spam_blocks()))
+        assert urls == [SPAM_URL]
+
+    def test_none_returns_empty(self):
+        assert extract_rich_urls(None) == []
+
+    def test_no_urls_returns_empty(self):
+        assert (
+            extract_rich_urls(
+                RichMessage(blocks=[RichBlockParagraph(text="no links here")])
+            )
+            == []
+        )
+
+    def test_walk_failure_returns_empty(self):
+        class _Boom:
+            @property
+            def url(self):
+                raise RuntimeError("boom")
+
+        assert extract_rich_urls(_Boom()) == []
+
+
+class TestHasNonWhitelistedRichMessageUrls:
+    def test_spam_rich_message_detected(self):
+        assert has_non_whitelisted_rich_message_urls(_make_message(_spam_blocks())) is True
+
+    def test_whitelisted_rich_urls_ignored(self):
+        blocks = [
+            RichBlockButtons(
+                buttons=[
+                    RichMessageButton(
+                        text="Docs", url="https://docs.python.org/3/library/"
+                    )
+                ]
+            )
+        ]
+        assert has_non_whitelisted_rich_message_urls(_make_message(blocks)) is False
+
+    def test_no_rich_message_ignored(self):
+        assert has_non_whitelisted_rich_message_urls(_make_message()) is False
+
+    def test_rich_without_urls_ignored(self):
+        blocks = [RichBlockParagraph(text="just text, no links")]
+        assert has_non_whitelisted_rich_message_urls(_make_message(blocks)) is False
+
+
+class TestClassifiableTextRich:
+    def test_rich_content_included(self):
+        text = _classifiable_text(_make_message(_spam_blocks()))
+        assert "HOT 2026 DROP" in text
+        assert "BIKINI SHOWS" in text
+        assert SPAM_URL in text
+
+    def test_plain_text_unchanged(self):
+        assert _classifiable_text(_make_message(text="hello")) == "hello"
+
+    def test_caption_unchanged(self):
+        msg = Message(
+            message_id=1,
+            date=datetime.now(UTC),
+            chat=Chat(id=1, type="private"),
+            caption="a caption",
+        )
+        assert _classifiable_text(msg) == "a caption"
+
+    def test_empty_message_stays_empty(self):
+        assert _classifiable_text(_make_message()) == ""
+
+    def test_text_and_rich_combined(self):
+        text = _classifiable_text(_make_message(_spam_blocks(), text="intro"))
+        assert text.startswith("intro")
+        assert "HOT 2026 DROP" in text
+
+
+class TestHandleInlineKeyboardSpamRich:
+    """Rich-message spam takes the auto-delete path at group=1."""
+
+    @pytest.fixture
+    def mock_update(self):
+        update = MagicMock()
+        update.message = MagicMock(spec=Message)
+        update.message.from_user = MagicMock(spec=User)
+        update.message.from_user.id = 12345
+        update.message.from_user.is_bot = False
+        update.message.from_user.full_name = "Spam User"
+        update.message.from_user.username = "spamuser"
+        update.message.delete = AsyncMock()
+        update.message.reply_markup = None
+        update.edited_message = None
+        return update
+
+    @pytest.fixture
+    def mock_context(self):
+        bot = MagicMock()
+        bot.send_message = AsyncMock()
+        bot.restrict_chat_member = AsyncMock()
+        state = AppState()
+        state.group_admin_ids = {}
+        state.trusted_user_ids = set()
+        return HandlerContext(bot=bot, state=state, args=[])
+
+    @pytest.fixture
+    def group_config(self):
+        return GroupConfig(
+            group_id=-100123456,
+            warning_topic_id=123,
+            rules_link="https://example.com/rules",
+        )
+
+    async def test_rich_spam_deleted_restricted_notified(
+        self, mock_update, mock_context, group_config
+    ):
+        mock_update.message.rich_message = RichMessage(blocks=_spam_blocks())
+
+        with patch(
+            "bot.handlers.anti_spam.get_group_config_for_update",
+            return_value=group_config,
+        ):
+            with pytest.raises(StopPropagation):
+                await handle_inline_keyboard_spam(mock_update, mock_context)
+
+        mock_update.message.delete.assert_called_once()
+        mock_context.bot.restrict_chat_member.assert_called_once()
+        mock_context.bot.send_message.assert_called_once()
+        sent_text = mock_context.bot.send_message.call_args.kwargs["text"]
+        assert "rich text" in sent_text
+
+    async def test_whitelisted_rich_urls_ignored(
+        self, mock_update, mock_context, group_config
+    ):
+        mock_update.message.rich_message = RichMessage(
+            blocks=[
+                RichBlockButtons(
+                    buttons=[
+                        RichMessageButton(
+                            text="Docs", url="https://docs.python.org/3/library/"
+                        )
+                    ]
+                )
+            ]
+        )
+
+        with patch(
+            "bot.handlers.anti_spam.get_group_config_for_update",
+            return_value=group_config,
+        ):
+            await handle_inline_keyboard_spam(mock_update, mock_context)
+
+        mock_update.message.delete.assert_not_called()
+        mock_context.bot.restrict_chat_member.assert_not_called()
+
+    async def test_no_rich_message_ignored(self, mock_update, mock_context, group_config):
+        mock_update.message.rich_message = None
+
+        with patch(
+            "bot.handlers.anti_spam.get_group_config_for_update",
+            return_value=group_config,
+        ):
+            await handle_inline_keyboard_spam(mock_update, mock_context)
+
+        mock_update.message.delete.assert_not_called()

From d9f92efbb017045c6400d2121a25b3acc6610257 Mon Sep 17 00:00:00 2001
From: Tayo <contact+tayo@rezhajulio.id>
Date: Sat, 17 Oct 2026 05:19:32 +0700
Subject: [PATCH 2/8] Refine: trigger rich auto-delete on buttons, not URLs

Per review: no stock Telegram client can compose rich messages or their
buttons, so button presence is the automation signature (same rationale
as inline keyboard spam). URL-only rich content without buttons now goes
to the AI monitor, mirroring how plain text links are handled, instead
of being auto-deleted.

- services.rich_message: new has_rich_buttons() covering block buttons
  and inline RichTextButton nodes.
- anti_spam: has_rich_message_buttons() replaces the URL-based rich
  detector; notification wording updated to mention buttons.
- Tests updated: whitelisted-URL buttons are still deleted; URL-only
  rich content is ignored by group 1.
---
 src/bot/constants.py             |  6 +-
 src/bot/handlers/anti_spam.py    | 40 +++++++-------
 src/bot/services/rich_message.py | 47 ++++++++++++++++
 tests/test_rich_message.py       | 94 ++++++++++++++++++++++++++++----
 4 files changed, 154 insertions(+), 33 deletions(-)

diff --git a/src/bot/constants.py b/src/bot/constants.py
index 8693934..3e01bb6 100644
--- a/src/bot/constants.py
+++ b/src/bot/constants.py
@@ -365,7 +365,8 @@ def format_hours_display(hours: int) -> str:
 RICH_MESSAGE_SPAM_NOTIFICATION = (
     "🚫 *Spam Terdeteksi*\n\n"
     "Pesan dari {user_mention} telah dihapus karena mengandung "
-    "pesan rich text dengan tautan mencurigakan.\n\n"
+    "tombol pada pesan rich text (tombol tidak dapat dibuat dari "
+    "aplikasi Telegram biasa).\n\n"
     "Pengguna telah dibatasi.\n\n"
     "📌 [Peraturan Grup]({rules_link})"
 )
@@ -373,7 +374,8 @@ def format_hours_display(hours: int) -> str:
 RICH_MESSAGE_SPAM_NOTIFICATION_NO_RESTRICT = (
     "🚫 *Spam Terdeteksi*\n\n"
     "Pesan dari {user_mention} telah dihapus karena mengandung "
-    "pesan rich text dengan tautan mencurigakan.\n\n"
+    "tombol pada pesan rich text (tombol tidak dapat dibuat dari "
+    "aplikasi Telegram biasa).\n\n"
     "📌 [Peraturan Grup]({rules_link})"
 )
 
diff --git a/src/bot/handlers/anti_spam.py b/src/bot/handlers/anti_spam.py
index 9f25b22..451624f 100644
--- a/src/bot/handlers/anti_spam.py
+++ b/src/bot/handlers/anti_spam.py
@@ -4,7 +4,7 @@
 This module enforces anti-spam rules including:
 - Contact card spam detection (all members)
 - Inline keyboard URL spam detection (all members)
-- Rich message URL spam detection (all members)
+- Rich message button spam detection (all members)
 - Probation enforcement for new users (forwards, links, external replies, stories, media)
 """
 
@@ -30,7 +30,7 @@
 )
 from bot.database.service import get_database
 from bot.group_config import get_group_config_for_update
-from bot.services.rich_message import extract_rich_urls
+from bot.services.rich_message import has_rich_buttons
 from bot.services.telegram_utils import (
     get_user_mention,
     is_url_whitelisted,
@@ -225,22 +225,24 @@ def has_non_whitelisted_inline_keyboard_urls(message: Message) -> bool:
     return False
 
 
-def has_non_whitelisted_rich_message_urls(message: Message) -> bool:
-    """Check if a message's rich content carries non-whitelisted URLs.
+def has_rich_message_buttons(message: Message) -> bool:
+    """Check if a message's rich content contains any buttons.
 
-    Rich messages (``Message.rich_message``) cannot be composed from
-    regular Telegram clients either, so the same policy as inline
-    keyboard spam applies: URL buttons and text links pointing to
-    non-whitelisted domains are treated as spam.
+    No stock Telegram client can compose rich messages or their buttons —
+    they are only sendable via the Bot API (bots) or crafted over MTProto
+    (spammers), or forwarded. The same "not client-craftable" rationale as
+    inline keyboard spam applies: buttons in a group rich message from a
+    human sender are treated as spam, regardless of where the buttons
+    point. URL-only rich content without buttons is left to the AI
+    monitor, mirroring how plain text links are handled.
 
     Args:
         message: Telegram message to check.
 
     Returns:
-        bool: True if any rich block URL is not whitelisted.
+        bool: True if any rich block or inline text carries a button.
     """
-    urls = extract_rich_urls(getattr(message, "rich_message", None))
-    return any(not is_url_whitelisted(url) for url in urls)
+    return has_rich_buttons(getattr(message, "rich_message", None))
 
 
 def has_contact(message: Message) -> bool:
@@ -386,24 +388,24 @@ async def handle_inline_keyboard_spam(
     update: Update, context: HandlerContext
 ) -> None:
     """
-    Handle spam messages containing non-whitelisted URLs in buttons.
+    Handle spam messages with non-client-craftable buttons.
 
-    Regular users cannot create inline keyboards — or rich messages —
-    from the Telegram client. Any group message with inline keyboard
-    URL buttons or rich message blocks pointing to non-whitelisted
-    domains is treated as spam: the message is deleted and the sender
-    restricted.
+    Regular users cannot create inline keyboards — or rich message buttons —
+    from the Telegram client. Any group message with inline keyboard URL
+    buttons pointing to non-whitelisted domains, or with rich message
+    buttons at all, is treated as spam: the message is deleted and the
+    sender restricted.
 
     Args:
         update: Telegram update containing the message.
         context: Bot context with helper methods.
     """
     message = update.message or update.edited_message
-    if message is not None and has_non_whitelisted_rich_message_urls(message):
+    if message is not None and has_rich_message_buttons(message):
         await _handle_group_spam(
             update,
             context,
-            detector=has_non_whitelisted_rich_message_urls,
+            detector=has_rich_message_buttons,
             label="rich message spam",
             should_restrict=lambda cfg: True,
             template_restricted=RICH_MESSAGE_SPAM_NOTIFICATION,
diff --git a/src/bot/services/rich_message.py b/src/bot/services/rich_message.py
index 770f7ea..5ab7ff1 100644
--- a/src/bot/services/rich_message.py
+++ b/src/bot/services/rich_message.py
@@ -87,3 +87,50 @@ def extract_rich_urls(rich_message: Any | None) -> list[str]:
         logger.warning("extract_rich_urls: failed to walk rich message", exc_info=True)
         return []
     return list(dict.fromkeys(urls))
+
+
+def has_rich_buttons(rich_message: Any | None) -> bool:
+    """Return True if a rich message contains any buttons.
+
+    Buttons appear as ``RichBlockButtons`` blocks or as inline
+    ``RichTextButton`` nodes in rich text. No stock Telegram client can
+    compose buttons, so their presence marks the message as
+    automation-crafted — the same rationale as the inline keyboard spam
+    rule.
+
+    The check is duck-typed (block containers plus the inline button
+    class name) so new button-carrying variants do not silently slip
+    through.
+    """
+    if rich_message is None:
+        return False
+    found = False
+
+    def _visit(node: Any) -> None:
+        nonlocal found
+        if found or node is None or isinstance(node, str):
+            return
+        if isinstance(node, (list, tuple)):
+            for item in node:
+                _visit(item)
+            return
+        if getattr(node, "buttons", None):
+            found = True
+            return
+        if type(node).__name__ == "RichTextButton":
+            found = True
+            return
+        for attr in ("text", "caption", "summary", "alternative_text",
+                     "blocks", "cells", "items"):
+            value = getattr(node, attr, None)
+            if value:
+                _visit(value)
+                if found:
+                    return
+
+    try:
+        _visit(rich_message)
+    except Exception:
+        logger.warning("has_rich_buttons: failed to walk rich message", exc_info=True)
+        return False
+    return found
diff --git a/tests/test_rich_message.py b/tests/test_rich_message.py
index 98fca64..4fa7b80 100644
--- a/tests/test_rich_message.py
+++ b/tests/test_rich_message.py
@@ -15,6 +15,7 @@
     RichMessage,
     RichMessageButton,
     RichTextBold,
+    RichTextButton,
     RichTextCustomEmoji,
     RichTextUrl,
     User,
@@ -25,9 +26,9 @@
 from bot.handlers.ai_spam_monitor import _classifiable_text
 from bot.handlers.anti_spam import (
     handle_inline_keyboard_spam,
-    has_non_whitelisted_rich_message_urls,
+    has_rich_message_buttons,
 )
-from bot.services.rich_message import extract_rich_text, extract_rich_urls
+from bot.services.rich_message import extract_rich_text, extract_rich_urls, has_rich_buttons
 
 SPAM_URL = "https://t.me/base6d1ef16_bot?startapp=138762"
 
@@ -138,11 +139,50 @@ def url(self):
         assert extract_rich_urls(_Boom()) == []
 
 
-class TestHasNonWhitelistedRichMessageUrls:
+class TestHasRichButtons:
+    def test_block_buttons_detected(self):
+        assert has_rich_buttons(RichMessage(blocks=_spam_blocks())) is True
+
+    def test_inline_text_button_detected(self):
+        blocks = [
+            RichBlockParagraph(
+                text=[
+                    "click ",
+                    RichTextButton(
+                        button=RichMessageButton(text="here", url="https://example.com")
+                    ),
+                ]
+            )
+        ]
+        assert has_rich_buttons(RichMessage(blocks=blocks)) is True
+
+    def test_no_buttons_returns_false(self):
+        blocks = [
+            RichBlockParagraph(
+                text=RichTextUrl(text="just a link", url="https://spam.example/x")
+            )
+        ]
+        assert has_rich_buttons(RichMessage(blocks=blocks)) is False
+
+    def test_none_returns_false(self):
+        assert has_rich_buttons(None) is False
+
+    def test_walk_failure_returns_false(self):
+        class _Boom:
+            @property
+            def buttons(self):
+                raise RuntimeError("boom")
+
+        assert has_rich_buttons(_Boom()) is False
+
+
+class TestHasRichMessageButtons:
     def test_spam_rich_message_detected(self):
-        assert has_non_whitelisted_rich_message_urls(_make_message(_spam_blocks())) is True
+        assert has_rich_message_buttons(_make_message(_spam_blocks())) is True
 
-    def test_whitelisted_rich_urls_ignored(self):
+    def test_buttons_with_whitelisted_urls_still_detected(self):
+        # Buttons are the trigger, not the URL: no stock client can
+        # compose them, so even whitelisted targets mean automation.
         blocks = [
             RichBlockButtons(
                 buttons=[
@@ -152,14 +192,20 @@ def test_whitelisted_rich_urls_ignored(self):
                 ]
             )
         ]
-        assert has_non_whitelisted_rich_message_urls(_make_message(blocks)) is False
+        assert has_rich_message_buttons(_make_message(blocks)) is True
 
     def test_no_rich_message_ignored(self):
-        assert has_non_whitelisted_rich_message_urls(_make_message()) is False
+        assert has_rich_message_buttons(_make_message()) is False
 
-    def test_rich_without_urls_ignored(self):
-        blocks = [RichBlockParagraph(text="just text, no links")]
-        assert has_non_whitelisted_rich_message_urls(_make_message(blocks)) is False
+    def test_rich_without_buttons_ignored(self):
+        # URL-only rich content goes to the AI monitor, mirroring how
+        # plain text links are handled.
+        blocks = [
+            RichBlockParagraph(
+                text=RichTextUrl(text="shady link", url="https://spam.example/x")
+            )
+        ]
+        assert has_rich_message_buttons(_make_message(blocks)) is False
 
 
 class TestClassifiableTextRich:
@@ -241,11 +287,13 @@ async def test_rich_spam_deleted_restricted_notified(
         mock_context.bot.restrict_chat_member.assert_called_once()
         mock_context.bot.send_message.assert_called_once()
         sent_text = mock_context.bot.send_message.call_args.kwargs["text"]
-        assert "rich text" in sent_text
+        assert "tombol pada pesan rich text" in sent_text
 
-    async def test_whitelisted_rich_urls_ignored(
+    async def test_buttons_with_whitelisted_urls_still_deleted(
         self, mock_update, mock_context, group_config
     ):
+        # Buttons are the trigger: no stock client can compose them, so the
+        # URL whitelist does not apply to rich message buttons.
         mock_update.message.rich_message = RichMessage(
             blocks=[
                 RichBlockButtons(
@@ -258,6 +306,28 @@ async def test_whitelisted_rich_urls_ignored(
             ]
         )
 
+        with patch(
+            "bot.handlers.anti_spam.get_group_config_for_update",
+            return_value=group_config,
+        ):
+            with pytest.raises(StopPropagation):
+                await handle_inline_keyboard_spam(mock_update, mock_context)
+
+        mock_update.message.delete.assert_called_once()
+        mock_context.bot.restrict_chat_member.assert_called_once()
+
+    async def test_rich_without_buttons_ignored(
+        self, mock_update, mock_context, group_config
+    ):
+        # URL-only rich content is left to the AI monitor, like plain text.
+        mock_update.message.rich_message = RichMessage(
+            blocks=[
+                RichBlockParagraph(
+                    text=RichTextUrl(text="shady link", url="https://spam.example/x")
+                )
+            ]
+        )
+
         with patch(
             "bot.handlers.anti_spam.get_group_config_for_update",
             return_value=group_config,

From 1816ab91807cf56b495fb78a61524ee3a0be620b Mon Sep 17 00:00:00 2001
From: Tayo <contact+tayo@rezhajulio.id>
Date: Sat, 17 Oct 2026 10:55:07 +0700
Subject: [PATCH 3/8] Fix review findings: credit/button traversal,
 auto-forward guard

- Walker now traverses RichBlockCaption.credit and RichTextButton.button,
  so text/URLs hidden in quotation credits and inline text buttons reach
  both the AI monitor and the button detector.
- Group 1 skips is_automatic_forward messages: linked-channel
  auto-forwards are admin-configured and must never trigger
  delete+restrict against the channel/service sender.
- Tests: credit extraction, inline button URL/text extraction,
  button-in-credit detection, auto-forward skip.
---
 src/bot/handlers/anti_spam.py    |  8 ++++-
 src/bot/services/rich_message.py | 10 +++---
 tests/test_rich_message.py       | 62 ++++++++++++++++++++++++++++++++
 3 files changed, 75 insertions(+), 5 deletions(-)

diff --git a/src/bot/handlers/anti_spam.py b/src/bot/handlers/anti_spam.py
index 451624f..1e64ea4 100644
--- a/src/bot/handlers/anti_spam.py
+++ b/src/bot/handlers/anti_spam.py
@@ -401,7 +401,13 @@ async def handle_inline_keyboard_spam(
         context: Bot context with helper methods.
     """
     message = update.message or update.edited_message
-    if message is not None and has_rich_message_buttons(message):
+    if message is None:
+        return
+    if getattr(message, "is_automatic_forward", False):
+        # Linked-channel auto-forwards are admin-configured; never punish
+        # the channel (or the 777000 service sender) for its content.
+        return
+    if has_rich_message_buttons(message):
         await _handle_group_spam(
             update,
             context,
diff --git a/src/bot/services/rich_message.py b/src/bot/services/rich_message.py
index 5ab7ff1..9164ab7 100644
--- a/src/bot/services/rich_message.py
+++ b/src/bot/services/rich_message.py
@@ -20,9 +20,11 @@
 logger = logging.getLogger(__name__)
 
 # Attributes that hold nested blocks, cells, buttons, or list items.
-_CONTAINER_ATTRS = ("blocks", "cells", "buttons", "items")
+# ``button`` is singular (``RichTextButton.button``); ``credit`` carries
+# rich text inside quotation/media captions (``RichBlockCaption.credit``).
+_CONTAINER_ATTRS = ("blocks", "cells", "buttons", "items", "button")
 # Text-bearing attributes besides the generic ``text``.
-_TEXT_ATTRS = ("caption", "summary", "alternative_text")
+_TEXT_ATTRS = ("caption", "summary", "alternative_text", "credit")
 
 
 def _walk(node: Any, texts: list[str], urls: list[str]) -> None:
@@ -120,8 +122,8 @@ def _visit(node: Any) -> None:
         if type(node).__name__ == "RichTextButton":
             found = True
             return
-        for attr in ("text", "caption", "summary", "alternative_text",
-                     "blocks", "cells", "items"):
+        for attr in ("text", "caption", "summary", "alternative_text", "credit",
+                     "blocks", "cells", "items", "button"):
             value = getattr(node, attr, None)
             if value:
                 _visit(value)
diff --git a/tests/test_rich_message.py b/tests/test_rich_message.py
index 4fa7b80..a2afcc2 100644
--- a/tests/test_rich_message.py
+++ b/tests/test_rich_message.py
@@ -9,6 +9,7 @@
     Message,
     RichBlockButtons,
     RichBlockParagraph,
+    RichBlockPullQuotation,
     RichBlockSectionHeading,
     RichBlockTable,
     RichBlockTableCell,
@@ -105,6 +106,29 @@ def test_plain_string_blocks(self):
         )
         assert text == "hello world"
 
+    def test_quotation_credit_included(self):
+        blocks = [
+            RichBlockPullQuotation(text="a wise quote", credit="— cited source")
+        ]
+        text = extract_rich_text(RichMessage(blocks=blocks))
+        assert "a wise quote" in text
+        assert "cited source" in text
+
+    def test_inline_text_button_content_included(self):
+        blocks = [
+            RichBlockParagraph(
+                text=[
+                    "tap ",
+                    RichTextButton(
+                        button=RichMessageButton(text="here", url=SPAM_URL)
+                    ),
+                ]
+            )
+        ]
+        text = extract_rich_text(RichMessage(blocks=blocks))
+        assert "tap" in text
+        assert "here" in text
+
     def test_walk_failure_returns_empty(self):
         class _Boom:
             @property
@@ -119,6 +143,16 @@ def test_extracts_button_and_text_link_urls_deduped(self):
         urls = extract_rich_urls(RichMessage(blocks=_spam_blocks()))
         assert urls == [SPAM_URL]
 
+    def test_inline_text_button_url_extracted(self):
+        blocks = [
+            RichBlockParagraph(
+                text=RichTextButton(
+                    button=RichMessageButton(text="here", url=SPAM_URL)
+                )
+            )
+        ]
+        assert extract_rich_urls(RichMessage(blocks=blocks)) == [SPAM_URL]
+
     def test_none_returns_empty(self):
         assert extract_rich_urls(None) == []
 
@@ -164,6 +198,17 @@ def test_no_buttons_returns_false(self):
         ]
         assert has_rich_buttons(RichMessage(blocks=blocks)) is False
 
+    def test_button_inside_credit_detected(self):
+        blocks = [
+            RichBlockPullQuotation(
+                text="a wise quote",
+                credit=RichTextButton(
+                    button=RichMessageButton(text="source", url=SPAM_URL)
+                ),
+            )
+        ]
+        assert has_rich_buttons(RichMessage(blocks=blocks)) is True
+
     def test_none_returns_false(self):
         assert has_rich_buttons(None) is False
 
@@ -347,3 +392,20 @@ async def test_no_rich_message_ignored(self, mock_update, mock_context, group_co
             await handle_inline_keyboard_spam(mock_update, mock_context)
 
         mock_update.message.delete.assert_not_called()
+
+    async def test_automatic_forward_skipped(
+        self, mock_update, mock_context, group_config
+    ):
+        # Linked-channel auto-forwards are admin-configured: never punish
+        # the channel for rich buttons in its posts.
+        mock_update.message.rich_message = RichMessage(blocks=_spam_blocks())
+        mock_update.message.is_automatic_forward = True
+
+        with patch(
+            "bot.handlers.anti_spam.get_group_config_for_update",
+            return_value=group_config,
+        ):
+            await handle_inline_keyboard_spam(mock_update, mock_context)
+
+        mock_update.message.delete.assert_not_called()
+        mock_context.bot.restrict_chat_member.assert_not_called()

From cc4fbdc59a450c6c200d677aff94026d76859c6c Mon Sep 17 00:00:00 2001
From: Tayo <contact+tayo@rezhajulio.id>
Date: Sat, 17 Oct 2026 11:01:22 +0700
Subject: [PATCH 4/8] Close probation gap: rich block URLs count as links

has_non_whitelisted_link() now also scans extract_rich_urls(), so
probation users cannot smuggle non-whitelisted links past the filter
inside rich text blocks. No behavior change for duplicate_spam: rich
messages without text return before the link check there.
---
 src/bot/handlers/anti_spam.py |  7 ++++++-
 tests/test_rich_message.py    | 35 +++++++++++++++++++++++++++++++++++
 2 files changed, 41 insertions(+), 1 deletion(-)

diff --git a/src/bot/handlers/anti_spam.py b/src/bot/handlers/anti_spam.py
index 1e64ea4..5343441 100644
--- a/src/bot/handlers/anti_spam.py
+++ b/src/bot/handlers/anti_spam.py
@@ -30,7 +30,7 @@
 )
 from bot.database.service import get_database
 from bot.group_config import get_group_config_for_update
-from bot.services.rich_message import has_rich_buttons
+from bot.services.rich_message import extract_rich_urls, has_rich_buttons
 from bot.services.telegram_utils import (
     get_user_mention,
     is_url_whitelisted,
@@ -163,6 +163,10 @@ def has_non_whitelisted_link(message: Message) -> bool:
     """
     Check if a message contains non-whitelisted URLs.
 
+    Covers URLs in text/caption entities as well as URLs hidden in rich
+    message blocks, so probation users cannot smuggle links past the
+    filter inside rich text.
+
     Args:
         message: Telegram message to check.
 
@@ -170,6 +174,7 @@ def has_non_whitelisted_link(message: Message) -> bool:
         bool: True if message contains non-whitelisted links.
     """
     urls = extract_urls(message)
+    urls.extend(extract_rich_urls(getattr(message, "rich_message", None)))
     if not urls:
         return False
 
diff --git a/tests/test_rich_message.py b/tests/test_rich_message.py
index a2afcc2..70afdaf 100644
--- a/tests/test_rich_message.py
+++ b/tests/test_rich_message.py
@@ -27,6 +27,7 @@
 from bot.handlers.ai_spam_monitor import _classifiable_text
 from bot.handlers.anti_spam import (
     handle_inline_keyboard_spam,
+    has_non_whitelisted_link,
     has_rich_message_buttons,
 )
 from bot.services.rich_message import extract_rich_text, extract_rich_urls, has_rich_buttons
@@ -253,6 +254,40 @@ def test_rich_without_buttons_ignored(self):
         assert has_rich_message_buttons(_make_message(blocks)) is False
 
 
+class TestHasNonWhitelistedLinkRich:
+    """Probation link check also sees URLs hidden in rich blocks."""
+
+    def test_rich_text_link_detected(self):
+        blocks = [
+            RichBlockParagraph(
+                text=RichTextUrl(text="shady link", url="https://spam.example/x")
+            )
+        ]
+        assert has_non_whitelisted_link(_make_message(blocks)) is True
+
+    def test_rich_button_url_detected(self):
+        blocks = [
+            RichBlockButtons(
+                buttons=[RichMessageButton(text="OPEN", url=SPAM_URL)]
+            )
+        ]
+        assert has_non_whitelisted_link(_make_message(blocks)) is True
+
+    def test_rich_whitelisted_url_ignored(self):
+        blocks = [
+            RichBlockParagraph(
+                text=RichTextUrl(
+                    text="docs", url="https://docs.python.org/3/library/"
+                )
+            )
+        ]
+        assert has_non_whitelisted_link(_make_message(blocks)) is False
+
+    def test_rich_without_urls_ignored(self):
+        blocks = [RichBlockParagraph(text="just text, no links")]
+        assert has_non_whitelisted_link(_make_message(blocks)) is False
+
+
 class TestClassifiableTextRich:
     def test_rich_content_included(self):
         text = _classifiable_text(_make_message(_spam_blocks()))

From 78149c32ab2db6836d6b052549c05f1c2e444f47 Mon Sep 17 00:00:00 2001
From: Tayo <contact+tayo@rezhajulio.id>
Date: Sat, 17 Oct 2026 11:45:06 +0700
Subject: [PATCH 5/8] Address independent review findings (B1/B2 blockers,
 M1-M5/M7-M9)

Blockers:
- B1: feed rich text into bio-bait matching (short text-only rich spam
  was invisible to every layer)
- B2: duplicate detection falls back to extracted rich text as the
  tracking key (rich-only floods were never tracked)

Walker hardening:
- M1: extract RichBlockListItem.label
- M2: descend into copy_text/login_url/web_app/chat-picker payloads;
  extract_rich_urls also detects URL-shaped strings in plain text
  (rich blocks carry no entity metadata; dotted-hostname required to
  avoid flagging fragments like 'https://' alone)
- M3: depth budget (64) on walkers; over-budget nesting is fail-closed
  for has_rich_buttons (only craftable, never client-composed)

Monitor:
- M5: truncate classifier input to 8000 chars
- M9: light normalization (NFKC + zero-width/bidi strip) before
  classification

Tests: e2e ai_spam_filter on rich-only updates, classify truncation,
is_automatic_forward on the legacy keyboard path, login_url/web_app
extraction, credit-nested button URLs, deep-nesting behavior,
NO_RESTRICT template path, admin exemption, bio-bait on rich text,
rich duplicate tracking. L4/L5/L15/L16 addressed.
---
 src/bot/handlers/ai_spam_monitor.py |  20 +-
 src/bot/handlers/bio_bait.py        |   7 +-
 src/bot/handlers/duplicate_spam.py  |   7 +-
 src/bot/services/rich_message.py    | 109 ++++++++---
 tests/test_bio_bait.py              |  18 +-
 tests/test_duplicate_spam.py        |  63 ++++++
 tests/test_rich_message.py          | 290 +++++++++++++++++++++++++++-
 7 files changed, 484 insertions(+), 30 deletions(-)

diff --git a/src/bot/handlers/ai_spam_monitor.py b/src/bot/handlers/ai_spam_monitor.py
index ec1f161..9eb8ba1 100644
--- a/src/bot/handlers/ai_spam_monitor.py
+++ b/src/bot/handlers/ai_spam_monitor.py
@@ -16,7 +16,9 @@
 """
 
 import logging
+import re
 import time
+import unicodedata
 
 from aiogram.types import InlineKeyboardButton, InlineKeyboardMarkup, Message, Update, User
 
@@ -66,6 +68,19 @@
 
 logger = logging.getLogger(__name__)
 
+_ZERO_WIDTH_RE = re.compile(r"[\u200b-\u200d\ufeff\u202a-\u202e\u2060-\u2064]")
+
+
+def _normalize_for_classification(text: str) -> str:
+    """Light normalization for classifier input.
+
+    NFKC + strip zero-width and bidi-override characters so obfuscated
+    (especially rich-block) text reaches the classifier clean. Unlike the
+    bio-bait normalizer this keeps case and punctuation intact.
+    """
+    return _ZERO_WIDTH_RE.sub("", unicodedata.normalize("NFKC", text))
+
+
 def _classifiable_text(message: Message) -> str:
     """Return the text available for classification.
 
@@ -80,7 +95,7 @@ def _classifiable_text(message: Message) -> str:
     if rich_text:
         parts.append(rich_text)
     parts.extend(extract_rich_urls(rich_message))
-    return " ".join(part for part in parts if part)
+    return _normalize_for_classification(" ".join(part for part in parts if part))
 
 
 def ai_spam_filter(update: Update) -> bool:
@@ -96,6 +111,7 @@ def ai_spam_filter(update: Update) -> bool:
 
 AI_SPAM_MIN_LENGTH = 20
 AI_SPAM_ALERT_MESSAGE_MAX_LEN = 500
+AI_SPAM_CLASSIFY_MAX_LEN = 8000
 
 ALERTS_KEY = "ai_spam_alerts"
 ALERTS_MAX_SIZE = 500
@@ -230,7 +246,7 @@ async def _classify_and_alert(
         return
 
     result = await classify_text(
-        message_text,
+        message_text[:AI_SPAM_CLASSIFY_MAX_LEN],
         labels=list(AI_SPAM_LABELS),
         instructions=AI_SPAM_INSTRUCTIONS,
         timeout=settings.classifier_timeout_seconds,
diff --git a/src/bot/handlers/bio_bait.py b/src/bot/handlers/bio_bait.py
index 19ae0d8..a14becd 100644
--- a/src/bot/handlers/bio_bait.py
+++ b/src/bot/handlers/bio_bait.py
@@ -43,6 +43,7 @@
     WHITELISTED_TELEGRAM_PATHS,
 )
 from bot.group_config import get_group_config_for_update
+from bot.services.rich_message import extract_rich_text
 from bot.services.telegram_utils import (
     get_user_mention,
     is_url_whitelisted,
@@ -425,7 +426,11 @@ async def handle_bio_bait_spam(
     if is_user_admin_or_trusted(context, group_config.group_id, user.id):
         return
 
-    text = message.text or message.caption or ""
+    text = (
+        message.text
+        or message.caption
+        or extract_rich_text(getattr(message, "rich_message", None))
+    )
 
     detection_reason: str | None = None
     user_bio: str | None = None
diff --git a/src/bot/handlers/duplicate_spam.py b/src/bot/handlers/duplicate_spam.py
index bc8eb24..475d9a7 100644
--- a/src/bot/handlers/duplicate_spam.py
+++ b/src/bot/handlers/duplicate_spam.py
@@ -33,6 +33,7 @@
 )
 from bot.group_config import GroupConfig, get_group_config_for_update
 from bot.handlers.anti_spam import has_non_whitelisted_link
+from bot.services.rich_message import extract_rich_text
 from bot.services.telegram_utils import (
     get_user_mention,
     is_user_admin_or_trusted,
@@ -172,7 +173,11 @@ async def handle_duplicate_spam(
     if is_user_admin_or_trusted(context, group_config.group_id, user.id):
         return
 
-    text = message.text or message.caption
+    text = (
+        message.text
+        or message.caption
+        or extract_rich_text(getattr(message, "rich_message", None))
+    )
     is_media = False
     if text:
         normalized = normalize_text(text)
diff --git a/src/bot/services/rich_message.py b/src/bot/services/rich_message.py
index 9164ab7..d8783d3 100644
--- a/src/bot/services/rich_message.py
+++ b/src/bot/services/rich_message.py
@@ -15,21 +15,67 @@
 from __future__ import annotations
 
 import logging
+import re
 from typing import Any
+from urllib.parse import urlparse
 
 logger = logging.getLogger(__name__)
 
+_URL_SHAPE_RE = re.compile(r"https?://[^\s<>\"]+", re.IGNORECASE)
+
+
+def _url_shaped_candidates(text: str) -> list[str]:
+    """Find URL-shaped strings inside plain text.
+
+    Bare URLs in rich blocks have no entity metadata (unlike text
+    messages, where Telegram tags them), so shape-check them here.
+    Requires a dotted hostname to avoid flagging fragments like
+    ``https://`` on their own.
+    """
+    out: list[str] = []
+    for match in _URL_SHAPE_RE.finditer(text):
+        candidate = match.group(0).rstrip(".,;:!?)")
+        try:
+            hostname = urlparse(candidate).hostname
+        except Exception:
+            continue
+        if hostname and "." in hostname:
+            out.append(candidate)
+    return out
+
 # Attributes that hold nested blocks, cells, buttons, or list items.
 # ``button`` is singular (``RichTextButton.button``); ``credit`` carries
 # rich text inside quotation/media captions (``RichBlockCaption.credit``).
 _CONTAINER_ATTRS = ("blocks", "cells", "buttons", "items", "button")
-# Text-bearing attributes besides the generic ``text``.
-_TEXT_ATTRS = ("caption", "summary", "alternative_text", "credit")
-
-
-def _walk(node: Any, texts: list[str], urls: list[str]) -> None:
+# Text-bearing attributes besides the generic ``text``. ``name`` (anchor /
+# reference ids) is deliberately skipped: internal link targets, not content.
+_TEXT_ATTRS = (
+    "caption",
+    "summary",
+    "alternative_text",
+    "credit",
+    "label",
+    "forward_text",
+)
+# Sub-objects whose own fields are walked for text/URLs: copy-text clipboard
+# payloads, login URLs, web apps, inline-query chat pickers.
+_PAYLOAD_ATTRS = (
+    "login_url",
+    "web_app",
+    "copy_text",
+    "switch_inline_query_chosen_chat",
+)
+
+# Nesting deeper than this cannot occur in genuine payloads (Telegram
+# documents depth <= 16 for sending); deeper trees are treated as hostile.
+_MAX_WALK_DEPTH = 64
+
+
+def _walk(
+    node: Any, texts: list[str], urls: list[str], depth: int = 0
+) -> None:
     """Recursively collect text fragments and URLs from a rich node."""
-    if node is None:
+    if node is None or depth > _MAX_WALK_DEPTH:
         return
     if isinstance(node, str):
         stripped = node.strip()
@@ -38,26 +84,26 @@ def _walk(node: Any, texts: list[str], urls: list[str]) -> None:
         return
     if isinstance(node, (list, tuple)):
         for item in node:
-            _walk(item, texts, urls)
+            _walk(item, texts, urls, depth + 1)
         return
 
     url = getattr(node, "url", None)
     if isinstance(url, str) and url.strip():
         urls.append(url.strip())
-    for nested in ("login_url", "web_app"):
-        nested_url = getattr(getattr(node, nested, None), "url", None)
-        if isinstance(nested_url, str) and nested_url.strip():
-            urls.append(nested_url.strip())
+    for attr in _PAYLOAD_ATTRS:
+        value = getattr(node, attr, None)
+        if value is not None:
+            _walk(value, texts, urls, depth + 1)
 
     for attr in ("text",) + _TEXT_ATTRS:
         value = getattr(node, attr, None)
         if value is not None:
-            _walk(value, texts, urls)
+            _walk(value, texts, urls, depth + 1)
 
     for attr in _CONTAINER_ATTRS:
         value = getattr(node, attr, None)
         if value:
-            _walk(value, texts, urls)
+            _walk(value, texts, urls, depth + 1)
 
 
 def extract_rich_text(rich_message: Any | None) -> str:
@@ -73,21 +119,30 @@ def extract_rich_text(rich_message: Any | None) -> str:
     try:
         _walk(rich_message, texts, [])
     except Exception:
-        logger.warning("extract_rich_text: failed to walk rich message", exc_info=True)
+        logger.info("extract_rich_text: failed to walk rich message")
         return ""
     return " ".join(texts)
 
 
 def extract_rich_urls(rich_message: Any | None) -> list[str]:
-    """Return deduplicated URLs from a rich message's buttons and text links."""
+    """Return deduplicated URLs from a rich message.
+
+    Covers explicit URL fields (buttons, text links, login/web-app URLs)
+    plus URL-shaped strings in plain text: rich blocks carry no entity
+    metadata, so a bare ``https://…`` in a paragraph or a copy-text
+    payload would otherwise evade link checks.
+    """
     if rich_message is None:
         return []
     urls: list[str] = []
+    texts: list[str] = []
     try:
-        _walk(rich_message, [], urls)
+        _walk(rich_message, texts, urls)
     except Exception:
-        logger.warning("extract_rich_urls: failed to walk rich message", exc_info=True)
+        logger.info("extract_rich_urls: failed to walk rich message")
         return []
+    for text in texts:
+        urls.extend(_url_shaped_candidates(text))
     return list(dict.fromkeys(urls))
 
 
@@ -108,31 +163,37 @@ class name) so new button-carrying variants do not silently slip
         return False
     found = False
 
-    def _visit(node: Any) -> None:
+    def _visit(node: Any, depth: int = 0) -> None:
         nonlocal found
         if found or node is None or isinstance(node, str):
             return
+        if depth > _MAX_WALK_DEPTH:
+            # Absurd nesting is only craftable, never client-composed:
+            # treat as suspicious rather than failing open.
+            found = True
+            return
         if isinstance(node, (list, tuple)):
             for item in node:
-                _visit(item)
+                _visit(item, depth + 1)
             return
         if getattr(node, "buttons", None):
             found = True
             return
-        if type(node).__name__ == "RichTextButton":
+        if getattr(node, "button", None) is not None:
+            # Inline text button (``RichTextButton``); the only node type
+            # with a singular ``button`` attribute.
             found = True
             return
-        for attr in ("text", "caption", "summary", "alternative_text", "credit",
-                     "blocks", "cells", "items", "button"):
+        for attr in ("text",) + _TEXT_ATTRS + _CONTAINER_ATTRS:
             value = getattr(node, attr, None)
             if value:
-                _visit(value)
+                _visit(value, depth + 1)
                 if found:
                     return
 
     try:
         _visit(rich_message)
     except Exception:
-        logger.warning("has_rich_buttons: failed to walk rich message", exc_info=True)
+        logger.info("has_rich_buttons: failed to walk rich message")
         return False
     return found
diff --git a/tests/test_bio_bait.py b/tests/test_bio_bait.py
index 86e6137..ef36284 100644
--- a/tests/test_bio_bait.py
+++ b/tests/test_bio_bait.py
@@ -3,7 +3,12 @@
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
-from aiogram.types import Message, User
+from aiogram.types import (
+    Message,
+    RichBlockParagraph,
+    RichMessage,
+    User,
+)
 
 from bot.dispatch import AppState, HandlerContext, StopPropagation
 
@@ -375,6 +380,17 @@ async def test_uses_caption_when_no_text(self, mock_update, mock_context, group_
                 await handle_bio_bait_spam(mock_update, mock_context)
         mock_update.message.delete.assert_called_once()
 
+    async def test_detects_bait_in_rich_text(self, mock_update, mock_context, group_config):
+        mock_update.message.text = None
+        mock_update.message.caption = None
+        mock_update.message.rich_message = RichMessage(
+            blocks=[RichBlockParagraph(text="cek bio aku dong")]
+        )
+        with patch("bot.handlers.bio_bait.get_group_config_for_update", return_value=group_config):
+            with pytest.raises(StopPropagation):
+                await handle_bio_bait_spam(mock_update, mock_context)
+        mock_update.message.delete.assert_called_once()
+
     async def test_detects_via_bio_links_with_innocuous_message(
         self, mock_update, mock_context, group_config
     ):
diff --git a/tests/test_duplicate_spam.py b/tests/test_duplicate_spam.py
index 8a51c4e..b57af4f 100644
--- a/tests/test_duplicate_spam.py
+++ b/tests/test_duplicate_spam.py
@@ -11,11 +11,14 @@
     Message,
     MessageEntity,
     PhotoSize,
+    RichBlockParagraph,
+    RichMessage,
     Sticker,
     Update,
     User,
 )
 
+
 from bot.dispatch import AppState, HandlerContext, StopPropagation
 
 from bot.group_config import GroupConfig
@@ -32,6 +35,22 @@
 DEFAULT_SIMILARITY = 0.95
 
 
+def _make_rich_update(message_id, rich_text):
+    """Build a real aiogram Update carrying a text-only rich message."""
+    user = User(id=42, is_bot=False, first_name="Test", username="testuser")
+    chat = Chat(id=-100, type="group", title="Test")
+    msg = Message(
+        message_id=message_id,
+        date=datetime.now(UTC),
+        chat=chat,
+        from_user=user,
+        rich_message=RichMessage(
+            blocks=[RichBlockParagraph(text=rich_text)]
+        ),
+    )
+    return Update(update_id=message_id, message=msg)
+
+
 def _make_sticker(file_unique_id):
     """Build a real aiogram Sticker with the given file_unique_id."""
     return Sticker(
@@ -326,6 +345,50 @@ async def test_identical_stickers_trigger_restriction(self, mock_context, group_
         assert mock_context.bot.delete_message.call_count == 2
         mock_context.bot.restrict_chat_member.assert_called_once()
 
+    async def test_identical_rich_text_triggers_restriction(
+        self, mock_context, group_config
+    ):
+        """Rich-only messages are tracked by their extracted rich text."""
+        rich_text = "Barangkali di sini ada yang sedang mencari kerja bisa menghubungi saya"
+        with patch(
+            "bot.handlers.duplicate_spam.get_group_config_for_update",
+            return_value=group_config,
+        ):
+            with patch(
+                "bot.handlers.duplicate_spam.get_user_mention", return_value="@testuser"
+            ):
+                await handle_duplicate_spam(
+                    _make_rich_update(100, rich_text), mock_context
+                )
+                with pytest.raises(StopPropagation):
+                    await handle_duplicate_spam(
+                        _make_rich_update(101, rich_text), mock_context
+                    )
+
+        assert mock_context.bot.delete_message.call_count == 2
+        mock_context.bot.restrict_chat_member.assert_called_once()
+
+    async def test_different_rich_text_not_counted(
+        self, mock_context, group_config
+    ):
+        with patch(
+            "bot.handlers.duplicate_spam.get_group_config_for_update",
+            return_value=group_config,
+        ):
+            for i, text in enumerate(
+                (
+                    "the quick brown fox jumps over the lazy dog today",
+                    "python packaging with uv is fast and reproducible",
+                    "jakarta traffic during rush hour is unbearable",
+                )
+            ):
+                await handle_duplicate_spam(
+                    _make_rich_update(100 + i, text), mock_context
+                )
+
+        mock_context.bot.delete_message.assert_not_called()
+        mock_context.bot.restrict_chat_member.assert_not_called()
+
     async def test_different_stickers_not_counted(self, mock_context, group_config):
         with patch("bot.handlers.duplicate_spam.get_group_config_for_update", return_value=group_config):
             for i in range(3):
diff --git a/tests/test_rich_message.py b/tests/test_rich_message.py
index 70afdaf..dcdcad2 100644
--- a/tests/test_rich_message.py
+++ b/tests/test_rich_message.py
@@ -6,8 +6,14 @@
 import pytest
 from aiogram.types import (
     Chat,
+    CopyTextButton,
+    InlineKeyboardButton,
+    InlineKeyboardMarkup,
+    LoginUrl,
     Message,
     RichBlockButtons,
+    RichBlockList,
+    RichBlockListItem,
     RichBlockParagraph,
     RichBlockPullQuotation,
     RichBlockSectionHeading,
@@ -19,12 +25,18 @@
     RichTextButton,
     RichTextCustomEmoji,
     RichTextUrl,
+    Update,
     User,
+    WebAppInfo,
 )
 
 from bot.dispatch import AppState, HandlerContext, StopPropagation
 from bot.group_config import GroupConfig
-from bot.handlers.ai_spam_monitor import _classifiable_text
+from bot.handlers import ai_spam_monitor
+from bot.handlers.ai_spam_monitor import (
+    AI_SPAM_CLASSIFY_MAX_LEN,
+    _classifiable_text,
+)
 from bot.handlers.anti_spam import (
     handle_inline_keyboard_spam,
     has_non_whitelisted_link,
@@ -154,6 +166,16 @@ def test_inline_text_button_url_extracted(self):
         ]
         assert extract_rich_urls(RichMessage(blocks=blocks)) == [SPAM_URL]
 
+    def test_bare_url_in_paragraph_detected(self):
+        blocks = [RichBlockParagraph(text="visit https://spam.example/deal now")]
+        assert extract_rich_urls(RichMessage(blocks=blocks)) == [
+            "https://spam.example/deal"
+        ]
+
+    def test_url_fragment_without_host_ignored(self):
+        blocks = [RichBlockParagraph(text="learn about the https:// protocol")]
+        assert extract_rich_urls(RichMessage(blocks=blocks)) == []
+
     def test_none_returns_empty(self):
         assert extract_rich_urls(None) == []
 
@@ -444,3 +466,269 @@ async def test_automatic_forward_skipped(
 
         mock_update.message.delete.assert_not_called()
         mock_context.bot.restrict_chat_member.assert_not_called()
+
+
+def _make_update(blocks=None, text=None):
+    return Update(
+        update_id=1,
+        message=Message(
+            message_id=1,
+            date=datetime.now(UTC),
+            chat=Chat(id=-100, type="supergroup"),
+            from_user=User(id=7, is_bot=False, first_name="U"),
+            text=text,
+            rich_message=RichMessage(blocks=blocks) if blocks is not None else None,
+        ),
+    )
+
+
+class TestRichWalkerGaps:
+    """Walker coverage for list labels, payload sub-objects, depth."""
+
+    def test_list_item_label_extracted(self):
+        blocks = [
+            RichBlockList(
+                items=[
+                    RichBlockListItem(
+                        label="PROMO SPESIAL HARI INI",
+                        blocks=[RichBlockParagraph(text="detail")],
+                    )
+                ]
+            )
+        ]
+        text = extract_rich_text(RichMessage(blocks=blocks))
+        assert "PROMO SPESIAL HARI INI" in text
+
+    def test_copy_text_payload_extracted(self):
+        blocks = [
+            RichBlockButtons(
+                buttons=[
+                    RichMessageButton(
+                        text="copy",
+                        copy_text=CopyTextButton(text="https://spam.example/steal"),
+                    )
+                ]
+            )
+        ]
+        assert extract_rich_urls(RichMessage(blocks=blocks)) == [
+            "https://spam.example/steal"
+        ]
+        assert has_rich_buttons(RichMessage(blocks=blocks)) is True
+
+    def test_login_url_and_web_app_extracted(self):
+        blocks = [
+            RichBlockButtons(
+                buttons=[
+                    RichMessageButton(
+                        text="login",
+                        login_url=LoginUrl(
+                            url="https://spam.example/login",
+                            forward_text="open login",
+                        ),
+                    ),
+                    RichMessageButton(
+                        text="app",
+                        web_app=WebAppInfo(url="https://spam.example/app"),
+                    ),
+                ]
+            )
+        ]
+        urls = extract_rich_urls(RichMessage(blocks=blocks))
+        assert urls == ["https://spam.example/login", "https://spam.example/app"]
+        assert "open login" in extract_rich_text(RichMessage(blocks=blocks))
+
+    def test_credit_nested_button_url_extracted(self):
+        blocks = [
+            RichBlockPullQuotation(
+                text="a wise quote",
+                credit=RichTextButton(
+                    button=RichMessageButton(text="source", url=SPAM_URL)
+                ),
+            )
+        ]
+        assert extract_rich_urls(RichMessage(blocks=blocks)) == [SPAM_URL]
+
+    def test_deep_nesting_buttons_fail_closed(self):
+        node = ["deep"]
+        for _ in range(3000):
+            node = [node]
+        assert has_rich_buttons(node) is True
+
+    def test_deep_nesting_extract_terminates(self):
+        node = ["deep"]
+        for _ in range(3000):
+            node = [node]
+        assert extract_rich_text(node) == ""
+        assert extract_rich_urls(node) == []
+
+
+class TestAiSpamFilterRich:
+    def test_rich_only_message_matches(self):
+        update = _make_update(
+            [RichBlockParagraph(text="this is a rich-only message with buttons")]
+        )
+        assert ai_spam_monitor.AI_SPAM_FILTER(update) is True
+
+    def test_rich_empty_no_match(self):
+        update = _make_update([])
+        assert ai_spam_monitor.AI_SPAM_FILTER(update) is False
+
+    def test_no_rich_no_match(self):
+        assert ai_spam_monitor.AI_SPAM_FILTER(_make_update()) is False
+
+
+class TestClassifyTruncation:
+    async def test_long_input_truncated_before_classify(self):
+        captured = {}
+
+        async def fake_classify(text, **kwargs):
+            captured["text"] = text
+            return None
+
+        settings = MagicMock()
+        settings.classifier_cooldown_seconds = 600
+        settings.classifier_timeout_seconds = 10
+        settings.ai_spam_daily_budget = 15000
+        context = HandlerContext(bot=MagicMock(), state=AppState(), args=[])
+        user = User(id=1, is_bot=False, first_name="T")
+        with (
+            patch("bot.handlers.ai_spam_monitor.classify_text", new=fake_classify),
+            patch(
+                "bot.handlers.ai_spam_monitor.get_settings", return_value=settings
+            ),
+            patch(
+                "bot.handlers.ai_spam_monitor.breaker_is_open", return_value=False
+            ),
+            patch(
+                "bot.handlers.ai_spam_monitor.try_spend_budget", return_value=True
+            ),
+        ):
+            await ai_spam_monitor._classify_and_alert(
+                context,
+                group_id=-100,
+                user=user,
+                message_id=1,
+                message_text="x" * 20000,
+            )
+        assert len(captured["text"]) == AI_SPAM_CLASSIFY_MAX_LEN
+
+
+class TestInlineKeyboardAutoForward:
+    """The is_automatic_forward guard covers the pre-existing keyboard path too."""
+
+    @pytest.fixture
+    def mock_update(self):
+        update = MagicMock()
+        update.message = MagicMock(spec=Message)
+        update.message.from_user = MagicMock(spec=User)
+        update.message.from_user.id = 12345
+        update.message.from_user.is_bot = False
+        update.message.from_user.full_name = "Spam User"
+        update.message.from_user.username = "spamuser"
+        update.message.delete = AsyncMock()
+        update.message.reply_markup = InlineKeyboardMarkup(
+            inline_keyboard=[
+                [InlineKeyboardButton(text="x", url="https://evil.example/")]
+            ]
+        )
+        update.message.rich_message = None
+        update.message.is_automatic_forward = True
+        update.edited_message = None
+        return update
+
+    @pytest.fixture
+    def mock_context(self):
+        bot = MagicMock()
+        bot.send_message = AsyncMock()
+        bot.restrict_chat_member = AsyncMock()
+        state = AppState()
+        state.group_admin_ids = {}
+        state.trusted_user_ids = set()
+        return HandlerContext(bot=bot, state=state, args=[])
+
+    @pytest.fixture
+    def group_config(self):
+        return GroupConfig(
+            group_id=-100123456,
+            warning_topic_id=123,
+            rules_link="https://example.com/rules",
+        )
+
+    async def test_keyboard_auto_forward_skipped(
+        self, mock_update, mock_context, group_config
+    ):
+        with patch(
+            "bot.handlers.anti_spam.get_group_config_for_update",
+            return_value=group_config,
+        ):
+            await handle_inline_keyboard_spam(mock_update, mock_context)
+
+        mock_update.message.delete.assert_not_called()
+        mock_context.bot.restrict_chat_member.assert_not_called()
+
+
+class TestRichNoRestrictAndExemptions:
+    @pytest.fixture
+    def mock_update(self):
+        update = MagicMock()
+        update.message = MagicMock(spec=Message)
+        update.message.from_user = MagicMock(spec=User)
+        update.message.from_user.id = 12345
+        update.message.from_user.is_bot = False
+        update.message.from_user.full_name = "Spam User"
+        update.message.from_user.username = "spamuser"
+        update.message.delete = AsyncMock()
+        update.message.reply_markup = None
+        update.message.rich_message = RichMessage(blocks=_spam_blocks())
+        update.message.is_automatic_forward = False
+        update.edited_message = None
+        return update
+
+    @pytest.fixture
+    def mock_context(self):
+        bot = MagicMock()
+        bot.send_message = AsyncMock()
+        bot.restrict_chat_member = AsyncMock()
+        state = AppState()
+        state.group_admin_ids = {}
+        state.trusted_user_ids = set()
+        return HandlerContext(bot=bot, state=state, args=[])
+
+    @pytest.fixture
+    def group_config(self):
+        return GroupConfig(
+            group_id=-100123456,
+            warning_topic_id=123,
+            rules_link="https://example.com/rules",
+        )
+
+    async def test_no_restrict_template_on_failure(
+        self, mock_update, mock_context, group_config
+    ):
+        mock_context.bot.restrict_chat_member = AsyncMock(
+            side_effect=Exception("Restrict failed")
+        )
+        with patch(
+            "bot.handlers.anti_spam.get_group_config_for_update",
+            return_value=group_config,
+        ):
+            with pytest.raises(StopPropagation):
+                await handle_inline_keyboard_spam(mock_update, mock_context)
+
+        text = mock_context.bot.send_message.call_args.kwargs["text"]
+        assert "dibatasi" not in text
+        assert "tombol pada pesan rich text" in text
+
+    async def test_admin_exempt_from_rich_delete(
+        self, mock_update, mock_context, group_config
+    ):
+        mock_context.state.group_admin_ids = {
+            group_config.group_id: [mock_update.message.from_user.id]
+        }
+        with patch(
+            "bot.handlers.anti_spam.get_group_config_for_update",
+            return_value=group_config,
+        ):
+            await handle_inline_keyboard_spam(mock_update, mock_context)
+
+        mock_update.message.delete.assert_not_called()

From cea87151cb3d3a32500666a4e4bb4d18e6f08670 Mon Sep 17 00:00:00 2001
From: Tayo <contact+tayo@rezhajulio.id>
Date: Sat, 17 Oct 2026 16:17:18 +0700
Subject: [PATCH 6/8] M6 follow-up: forward-aware wording + appeal line
 (forwarder stays delete+restrict)

Per policy decision: forwarders of rich-button messages keep
delete+restrict (same as the inline-keyboard rule; the incident
spammer was a forwarder). The notification now reads 'Pesan terusan'
for forwarded messages and carries an appeal line ('Hubungi admin
grup') in both restricted and non-restricted variants.

_handle_group_spam gains an optional extra_format mapping for
template placeholders; property tests for _url_shaped_candidates
invariants (no-http => empty, well-formed candidates, no dupes).
---
 src/bot/constants.py          |  7 ++++---
 src/bot/handlers/anti_spam.py |  6 ++++++
 tests/test_properties.py      | 30 ++++++++++++++++++++++++++++++
 tests/test_rich_message.py    | 34 ++++++++++++++++++++++++++++++++++
 4 files changed, 74 insertions(+), 3 deletions(-)

diff --git a/src/bot/constants.py b/src/bot/constants.py
index 3e01bb6..5263555 100644
--- a/src/bot/constants.py
+++ b/src/bot/constants.py
@@ -364,18 +364,19 @@ def format_hours_display(hours: int) -> str:
 # Rich message spam notification
 RICH_MESSAGE_SPAM_NOTIFICATION = (
     "🚫 *Spam Terdeteksi*\n\n"
-    "Pesan dari {user_mention} telah dihapus karena mengandung "
+    "Pesan{forward_note} dari {user_mention} telah dihapus karena mengandung "
     "tombol pada pesan rich text (tombol tidak dapat dibuat dari "
     "aplikasi Telegram biasa).\n\n"
-    "Pengguna telah dibatasi.\n\n"
+    "Pengguna telah dibatasi. Merasa ini kesalahan? Hubungi admin grup.\n\n"
     "📌 [Peraturan Grup]({rules_link})"
 )
 
 RICH_MESSAGE_SPAM_NOTIFICATION_NO_RESTRICT = (
     "🚫 *Spam Terdeteksi*\n\n"
-    "Pesan dari {user_mention} telah dihapus karena mengandung "
+    "Pesan{forward_note} dari {user_mention} telah dihapus karena mengandung "
     "tombol pada pesan rich text (tombol tidak dapat dibuat dari "
     "aplikasi Telegram biasa).\n\n"
+    "Merasa ini kesalahan? Hubungi admin grup.\n\n"
     "📌 [Peraturan Grup]({rules_link})"
 )
 
diff --git a/src/bot/handlers/anti_spam.py b/src/bot/handlers/anti_spam.py
index 5343441..dec6322 100644
--- a/src/bot/handlers/anti_spam.py
+++ b/src/bot/handlers/anti_spam.py
@@ -272,6 +272,7 @@ async def _handle_group_spam(
     should_restrict,
     template_restricted: str,
     template_no_restrict: str,
+    extra_format: dict[str, str] | None = None,
 ) -> None:
     """
     Shared helper for handling group spam detection and enforcement.
@@ -287,6 +288,7 @@ async def _handle_group_spam(
         should_restrict: Callable taking group_config and returning bool.
         template_restricted: Notification template when user is restricted.
         template_no_restrict: Notification template when user is not restricted.
+        extra_format: Additional template placeholders (e.g. forward_note).
     """
     message = update.message or update.edited_message
     if not message or not message.from_user:
@@ -348,6 +350,7 @@ async def _handle_group_spam(
         notification_text = template.format(
             user_mention=user_mention,
             rules_link=group_config.rules_link,
+            **(extra_format or {}),
         )
         await context.bot.send_message(
             chat_id=group_config.group_id,
@@ -421,6 +424,9 @@ async def handle_inline_keyboard_spam(
             should_restrict=lambda cfg: True,
             template_restricted=RICH_MESSAGE_SPAM_NOTIFICATION,
             template_no_restrict=RICH_MESSAGE_SPAM_NOTIFICATION_NO_RESTRICT,
+            extra_format={
+                "forward_note": " terusan" if is_forwarded(message) else ""
+            },
         )
         return
     await _handle_group_spam(
diff --git a/tests/test_properties.py b/tests/test_properties.py
index 0550b72..1aa4362 100644
--- a/tests/test_properties.py
+++ b/tests/test_properties.py
@@ -9,6 +9,8 @@
 """
 
 
+from urllib.parse import urlparse
+
 import pytest
 from hypothesis import HealthCheck, assume, given, settings
 from hypothesis import strategies as st
@@ -17,6 +19,7 @@
 from bot.handlers.ai_spam_monitor import flagged_aspects
 from bot.handlers.trust import _format_person, _format_person_with_username
 from bot.services.classifier_client import ClassificationResult
+from bot.services.rich_message import _url_shaped_candidates
 from bot.services.telegram_utils import is_url_whitelisted
 
 # Strategies -----------------------------------------------------------------
@@ -300,3 +303,30 @@ def test_legacy_single_label_path_follows_whitelist(self, threshold: float) -> N
 @given(st.integers(min_value=0, max_value=1000))
 def test_sanity_integer_strategy(x: int) -> None:
     assert 0 <= x <= 1000
+
+
+class TestUrlShapedCandidates:
+    """Invariants for rich-block bare-URL detection."""
+
+    @given(st.text(max_size=200))
+    @settings(max_examples=200)
+    def test_no_http_means_no_candidates(self, text: str) -> None:
+        """Without an http(s) marker there is nothing URL-shaped."""
+        assume("http" not in text.lower())
+        assert _url_shaped_candidates(text) == []
+
+    @given(st.text(max_size=200))
+    @settings(max_examples=200)
+    def test_candidates_are_well_formed_urls(self, text: str) -> None:
+        """Every candidate has an http(s) scheme and a dotted hostname."""
+        for candidate in _url_shaped_candidates(text):
+            lowered = candidate.lower()
+            assert lowered.startswith(("http://", "https://"))
+            hostname = urlparse(candidate).hostname or ""
+            assert "." in hostname
+
+    @given(st.text(max_size=200))
+    @settings(max_examples=200)
+    def test_no_duplicates(self, text: str) -> None:
+        candidates = _url_shaped_candidates(text)
+        assert len(candidates) == len(set(candidates))
diff --git a/tests/test_rich_message.py b/tests/test_rich_message.py
index dcdcad2..580c1d3 100644
--- a/tests/test_rich_message.py
+++ b/tests/test_rich_message.py
@@ -352,6 +352,7 @@ def mock_update(self):
         update.message.from_user.username = "spamuser"
         update.message.delete = AsyncMock()
         update.message.reply_markup = None
+        update.message.forward_origin = None
         update.edited_message = None
         return update
 
@@ -633,6 +634,7 @@ def mock_update(self):
         )
         update.message.rich_message = None
         update.message.is_automatic_forward = True
+        update.message.forward_origin = None
         update.edited_message = None
         return update
 
@@ -681,6 +683,7 @@ def mock_update(self):
         update.message.reply_markup = None
         update.message.rich_message = RichMessage(blocks=_spam_blocks())
         update.message.is_automatic_forward = False
+        update.message.forward_origin = None
         update.edited_message = None
         return update
 
@@ -732,3 +735,34 @@ async def test_admin_exempt_from_rich_delete(
             await handle_inline_keyboard_spam(mock_update, mock_context)
 
         mock_update.message.delete.assert_not_called()
+
+    async def _notification_text(
+        self, mock_update, mock_context, group_config, forwarded: bool
+    ) -> str:
+        mock_update.message.forward_origin = MagicMock() if forwarded else None
+        with patch(
+            "bot.handlers.anti_spam.get_group_config_for_update",
+            return_value=group_config,
+        ):
+            with pytest.raises(StopPropagation):
+                await handle_inline_keyboard_spam(mock_update, mock_context)
+        return mock_context.bot.send_message.call_args.kwargs["text"]
+
+    async def test_forwarded_wording_and_appeal_line(
+        self, mock_update, mock_context, group_config
+    ):
+        text = await self._notification_text(
+            mock_update, mock_context, group_config, forwarded=True
+        )
+        assert "Pesan terusan dari" in text
+        assert "Hubungi admin grup" in text
+
+    async def test_non_forwarded_wording_and_appeal_line(
+        self, mock_update, mock_context, group_config
+    ):
+        text = await self._notification_text(
+            mock_update, mock_context, group_config, forwarded=False
+        )
+        assert "Pesan dari" in text
+        assert "Pesan terusan" not in text
+        assert "Hubungi admin grup" in text

From 799482fd1ae8faa863bfd8ef4da9600a3688caa5 Mon Sep 17 00:00:00 2001
From: Tayo <contact+tayo@rezhajulio.id>
Date: Sat, 17 Oct 2026 16:49:03 +0700
Subject: [PATCH 7/8] Fix exponential walker blowup on non-aiogram objects

The M3 depth budget removed the RecursionError fail-fast that used
to terminate walks over infinitely-generative objects (mocks):
with every attribute truthy, the walker branched ~15^64 and never
finished, OOMing the suite and hanging CI.

_walk/_visit now only descend into str/list/tuple or aiogram-module
objects (_is_rich_node); the depth + node budgets remain as
defense-in-depth for hostile aiogram-shaped payloads.
---
 src/bot/services/rich_message.py | 51 +++++++++++++++++++++++++-------
 1 file changed, 41 insertions(+), 10 deletions(-)

diff --git a/src/bot/services/rich_message.py b/src/bot/services/rich_message.py
index d8783d3..10c0f84 100644
--- a/src/bot/services/rich_message.py
+++ b/src/bot/services/rich_message.py
@@ -69,14 +69,38 @@ def _url_shaped_candidates(text: str) -> list[str]:
 # Nesting deeper than this cannot occur in genuine payloads (Telegram
 # documents depth <= 16 for sending); deeper trees are treated as hostile.
 _MAX_WALK_DEPTH = 64
+# Total nodes visited per walk. Depth alone cannot bound the walk: an
+# object with every attribute truthy (e.g. unittest mocks, or hostile
+# crafted objects) branches exponentially, so the visited-node count
+# is capped as well. Genuine payloads stay far below this.
+_MAX_WALK_NODES = 10_000
+
+
+def _is_rich_node(value: Any) -> bool:
+    """Check whether a value is a traversable rich object.
+
+    Rich nodes always come from aiogram's parser. Anything else —
+    mocks, foreign objects, hostile lookalikes — is not content and
+    must not be descended into (mocks in particular generate infinite
+    attribute trees).
+    """
+    return type(value).__module__.split(".")[0] == "aiogram"
 
 
 def _walk(
-    node: Any, texts: list[str], urls: list[str], depth: int = 0
+    node: Any,
+    texts: list[str],
+    urls: list[str],
+    depth: int = 0,
+    budget: list[int] | None = None,
 ) -> None:
     """Recursively collect text fragments and URLs from a rich node."""
     if node is None or depth > _MAX_WALK_DEPTH:
         return
+    if budget is not None:
+        if budget[0] <= 0:
+            return
+        budget[0] -= 1
     if isinstance(node, str):
         stripped = node.strip()
         if stripped:
@@ -84,7 +108,9 @@ def _walk(
         return
     if isinstance(node, (list, tuple)):
         for item in node:
-            _walk(item, texts, urls, depth + 1)
+            _walk(item, texts, urls, depth + 1, budget)
+        return
+    if not _is_rich_node(node):
         return
 
     url = getattr(node, "url", None)
@@ -93,17 +119,17 @@ def _walk(
     for attr in _PAYLOAD_ATTRS:
         value = getattr(node, attr, None)
         if value is not None:
-            _walk(value, texts, urls, depth + 1)
+            _walk(value, texts, urls, depth + 1, budget)
 
     for attr in ("text",) + _TEXT_ATTRS:
         value = getattr(node, attr, None)
         if value is not None:
-            _walk(value, texts, urls, depth + 1)
+            _walk(value, texts, urls, depth + 1, budget)
 
     for attr in _CONTAINER_ATTRS:
         value = getattr(node, attr, None)
         if value:
-            _walk(value, texts, urls, depth + 1)
+            _walk(value, texts, urls, depth + 1, budget)
 
 
 def extract_rich_text(rich_message: Any | None) -> str:
@@ -117,7 +143,7 @@ def extract_rich_text(rich_message: Any | None) -> str:
         return ""
     texts: list[str] = []
     try:
-        _walk(rich_message, texts, [])
+        _walk(rich_message, texts, [], 0, [_MAX_WALK_NODES])
     except Exception:
         logger.info("extract_rich_text: failed to walk rich message")
         return ""
@@ -137,7 +163,7 @@ def extract_rich_urls(rich_message: Any | None) -> list[str]:
     urls: list[str] = []
     texts: list[str] = []
     try:
-        _walk(rich_message, texts, urls)
+        _walk(rich_message, texts, urls, 0, [_MAX_WALK_NODES])
     except Exception:
         logger.info("extract_rich_urls: failed to walk rich message")
         return []
@@ -162,20 +188,25 @@ class name) so new button-carrying variants do not silently slip
     if rich_message is None:
         return False
     found = False
+    budget = [_MAX_WALK_NODES]
 
     def _visit(node: Any, depth: int = 0) -> None:
         nonlocal found
         if found or node is None or isinstance(node, str):
             return
-        if depth > _MAX_WALK_DEPTH:
-            # Absurd nesting is only craftable, never client-composed:
-            # treat as suspicious rather than failing open.
+        if depth > _MAX_WALK_DEPTH or budget[0] <= 0:
+            # Absurd nesting or an exploding attribute tree is only
+            # craftable, never client-composed: treat as suspicious
+            # rather than failing open.
             found = True
             return
+        budget[0] -= 1
         if isinstance(node, (list, tuple)):
             for item in node:
                 _visit(item, depth + 1)
             return
+        if not _is_rich_node(node):
+            return
         if getattr(node, "buttons", None):
             found = True
             return

From 8424ac4c1590e9886b1f2474450658f5689c0ad8 Mon Sep 17 00:00:00 2001
From: Tayo <contact+tayo@rezhajulio.id>
Date: Sat, 17 Oct 2026 17:59:26 +0700
Subject: [PATCH 8/8] Dedup _url_shaped_candidates output (matches tested
 invariant)

---
 src/bot/services/rich_message.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/bot/services/rich_message.py b/src/bot/services/rich_message.py
index 10c0f84..2fcaf64 100644
--- a/src/bot/services/rich_message.py
+++ b/src/bot/services/rich_message.py
@@ -41,7 +41,7 @@ def _url_shaped_candidates(text: str) -> list[str]:
             continue
         if hostname and "." in hostname:
             out.append(candidate)
-    return out
+    return list(dict.fromkeys(out))
 
 # Attributes that hold nested blocks, cells, buttons, or list items.
 # ``button`` is singular (``RichTextButton.button``); ``credit`` carries