fix: /warn ID/username priority over reply + forum topic anchor skip

Fix two bugs in the /warn command:

1. ID mode warned the replied-to user instead of the specified user ID.
   In forum-topic groups, Telegram auto-sets reply_to_message to the
   topic anchor message, so reply mode always won over ID mode even
   when a user ID was explicitly provided. Now args are checked first:
   numeric → ID mode, @username → username mode, then reply fallback.

2. Forum topic anchor messages (forum_topic_created set) are now
   detected via _is_real_reply() and skipped, so they are never
   mistaken for a real user reply.

Also adds @username support as a third invocation mode:
  /warn @username [reason]
Since the Bot API cannot resolve usernames to user IDs, this mode
mentions @username directly without membership/bot/self checks.

Updated usage message to mention @username option.

Author: rezhajulio

src/bot/constants.py

@@ -234,8 +234,8 @@ def format_hours_display(hours: int) -> str:
 
 # Generic /warn command templates (admin-issued, in-group)
 WARN_COMMAND_USAGE = (
-    "❌ Penggunaan: balas pesan member dengan /warn [alasan] "
-    "atau gunakan /warn USER_ID [alasan]"
+    "❌ Penggunaan: balas pesan member dengan /warn [alasan], "
+    "/warn USER_ID [alasan], atau /warn @username [alasan]"
 )
 WARN_COMMAND_NOT_FOUND = "❌ Tidak dapat menemukan user dengan ID {user_id}."
 WARN_COMMAND_NOT_MEMBER = "❌ User {user_id} bukan member grup ini."

src/bot/handlers/warn.py

@@ -2,10 +2,15 @@
 Admin /warn command handler for the PythonID bot.
 
 Lets an admin make the bot send a generic warning to a group member.
-Two invocation modes:
+Three invocation modes:
 
-1. Reply mode: admin replies to the member's message with ``/warn [reason]``
-2. ID mode:    admin sends ``/warn USER_ID [reason]`` in the group
+1. Reply mode:   admin replies to the member's message with ``/warn [reason]``
+2. ID mode:      admin sends ``/warn USER_ID [reason]`` in the group
+3. Username mode: admin sends ``/warn @username [reason]`` in the group
+
+In forum-topic groups, Telegram auto-sets ``reply_to_message`` to the topic
+anchor message.  To avoid mistaking the anchor for a real reply, we check
+``forum_topic_created`` and skip it.
 
 The warning is sent to the moderation topic when ``moderation_topic_id`` is
 configured (per-group), otherwise to the main group chat.
@@ -44,18 +49,29 @@ async def _delete_command_message(update: Update) -> None:
         )
 
 
+def _is_real_reply(message: object) -> bool:
+    """Check if reply_to_message is a real user reply, not a forum topic anchor."""
+    reply = getattr(message, "reply_to_message", None)
+    if reply is None:
+        return False
+    if getattr(reply, "forum_topic_created", None) is not None:
+        return False
+    if getattr(reply, "from_user", None) is None:
+        return False
+    return True
+
+
 async def handle_warn_command(
     update: Update, context: ContextTypes.DEFAULT_TYPE
 ) -> None:
     """
     Handle /warn command in a monitored group.
 
-    Admin replies to a member's message with ``/warn [reason]``, or
-    provides a user ID: ``/warn USER_ID [reason]``.
-
-    Sends a warning message mentioning the target member. The admin's
-    command message is deleted before any network lookups to protect
-    their identity.
+    Resolution priority:
+    1. ``args[0]`` is numeric   → ID mode (get_chat_member, membership check)
+    2. ``args[0]`` starts with @ → username mode (mention only, no membership check)
+    3. Real reply_to_message     → reply mode
+    4. None of the above         → usage error
     """
     if not update.message or not update.message.from_user:
         return
@@ -67,32 +83,23 @@ async def handle_warn_command(
     if group_config is None:
         return
 
-    # Per-group admin check (not global union)
     if not is_user_admin_in_group(context, group_config.group_id, admin.id):
         return
 
-    # Delete command message early to protect admin identity on all paths
     await _delete_command_message(update)
 
-    # Resolve target user and reason
-    reply_user = (
-        message.reply_to_message.from_user
-        if message.reply_to_message
-        else None
-    )
+    args = context.args or []
+    has_real_reply = _is_real_reply(message)
 
-    if reply_user is not None:
-        target_user = reply_user
-        reason = " ".join(context.args) if context.args else ""
-    elif context.args:
-        try:
-            target_user_id = int(context.args[0])
-        except ValueError:
-            try:
-                await message.reply_text(WARN_COMMAND_USAGE, do_quote=False)
-            except Exception:
-                logger.error("Failed to send usage message", exc_info=True)
-            return
+    # --- Determine target and reason ---
+    target_user: object | None = None
+    target_username: str | None = None
+    reason = ""
+
+    if args and args[0].lstrip("-").isdigit():
+        # --- ID mode ---
+        target_user_id = int(args[0])
+        reason = " ".join(args[1:]) if len(args) > 1 else ""
         try:
             member = await context.bot.get_chat_member(
                 chat_id=group_config.group_id,
@@ -123,24 +130,37 @@ async def handle_warn_command(
         target_user = member.user
         if target_user is None:
             return
-        reason = " ".join(context.args[1:]) if len(context.args) > 1 else ""
+    elif args and args[0].startswith("@"):
+        # --- Username mode ---
+        target_username = args[0].lstrip("@")
+        reason = " ".join(args[1:]) if len(args) > 1 else ""
+    elif has_real_reply:
+        # --- Reply mode ---
+        target_user = message.reply_to_message.from_user  # type: ignore[union-attr]
+        reason = " ".join(args) if args else ""
     else:
         try:
             await message.reply_text(WARN_COMMAND_USAGE, do_quote=False)
         except Exception:
             logger.error("Failed to send usage message", exc_info=True)
         return
 
-    if target_user.is_bot:
-        return
-
-    if target_user.id == admin.id:
+    # --- Build warning text ---
+    if target_user is not None:
+        if target_user.is_bot:
+            return
+        if target_user.id == admin.id:
+            return
+        user_mention = get_user_mention_by_id(
+            target_user.id,
+            target_user.full_name,
+            getattr(target_user, "username", None),
+        )
+    elif target_username is not None:
+        user_mention = f"@{escape_markdown(target_username, version=1)}"
+    else:
         return
 
-    user_mention = get_user_mention_by_id(
-        target_user.id, target_user.full_name, getattr(target_user, "username", None)
-    )
-
     if reason:
         warn_text = WARN_COMMAND_WITH_REASON.format(
             user_mention=user_mention,
@@ -160,11 +180,13 @@ async def handle_warn_command(
         await context.bot.send_message(**send_kwargs)
     except Exception:
         logger.error(
-            f"Failed to send warning to group {group_config.group_id} for user {target_user.id}",
+            f"Failed to send warning to group {group_config.group_id}",
             exc_info=True,
         )
         return
 
     logger.info(
-        f"Admin {admin.id} warned user {target_user.id} in group {group_config.group_id}"
+        f"Admin {admin.id} warned "
+        f"{'@' + target_username if target_username else target_user.id} "  # type: ignore[union-attr]
+        f"in group {group_config.group_id}"
     )

tests/test_warn.py

@@ -37,6 +37,7 @@ def mock_update():
     update.message.chat_id = -1001234567890
     update.message.message_id = 999
     update.message.reply_to_message = None
+    update.message.forum_topic_created = None
     update.effective_chat = MagicMock()
     update.effective_chat.id = -1001234567890
     update.effective_chat.type = "supergroup"
@@ -69,6 +70,7 @@ def _make_target_user(user_id=67890, full_name="Bad Member", username="badmember
 def _make_reply_message(target_user):
     msg = MagicMock()
     msg.from_user = target_user
+    msg.forum_topic_created = None
     return msg
 
 
@@ -322,11 +324,12 @@ async def test_delete_failure_does_not_break(
     async def test_reply_without_from_user_falls_to_id_mode(
         self, mock_update, mock_context, mock_registry
     ):
-        """Reply to a channel message (no from_user) falls through to ID mode."""
+        """Reply to a channel message (no from_user) with ID arg uses ID mode."""
         target = _make_target_user()
         mock_context.bot.get_chat_member.return_value = _make_chat_member(target)
         mock_update.message.reply_to_message = MagicMock()
         mock_update.message.reply_to_message.from_user = None
+        mock_update.message.reply_to_message.forum_topic_created = None
         mock_context.args = ["67890", "spam"]
 
         with patch("bot.handlers.warn.get_group_config_for_update", return_value=mock_registry.get(-1001234567890)):
@@ -338,7 +341,7 @@ async def test_reply_without_from_user_falls_to_id_mode(
         mock_context.bot.send_message.assert_called_once()
         call_kwargs = mock_context.bot.send_message.call_args.kwargs
         assert "spam" in call_kwargs["text"]
-        assert "67890" not in call_kwargs["text"] or "badmember" in call_kwargs["text"]
+        assert "badmember" in call_kwargs["text"]
 
     async def test_reason_with_markdown_is_escaped(
         self, mock_update, mock_context, mock_registry
@@ -464,3 +467,123 @@ async def mock_get_member(**kwargs):
             await handle_warn_command(mock_update, mock_context)
 
         assert call_order[0] == "delete"
+
+    async def test_id_mode_takes_priority_over_reply(
+        self, mock_update, mock_context, mock_registry
+    ):
+        """Bug fix: /warn USER_ID reason as a reply warns the ID target, not the replied-to user."""
+        reply_target = _make_target_user(user_id=11111, full_name="Replied User", username="replieduser")
+        id_target = _make_target_user(user_id=67890, full_name="ID Target", username="idtarget")
+        mock_update.message.reply_to_message = _make_reply_message(reply_target)
+        mock_context.bot.get_chat_member.return_value = _make_chat_member(id_target)
+        mock_context.args = ["67890", "spamming"]
+
+        with patch("bot.handlers.warn.get_group_config_for_update", return_value=mock_registry.get(-1001234567890)):
+            await handle_warn_command(mock_update, mock_context)
+
+        mock_context.bot.get_chat_member.assert_called_once_with(
+            chat_id=-1001234567890, user_id=67890
+        )
+        mock_context.bot.send_message.assert_called_once()
+        call_kwargs = mock_context.bot.send_message.call_args.kwargs
+        assert "idtarget" in call_kwargs["text"]
+        assert "replieduser" not in call_kwargs["text"]
+        assert "spamming" in call_kwargs["text"]
+
+    async def test_forum_topic_anchor_not_treated_as_reply(
+        self, mock_update, mock_context, mock_registry
+    ):
+        """Bug fix: forum topic anchor (auto reply_to_message) is not treated as a real reply."""
+        target = _make_target_user()
+        mock_context.bot.get_chat_member.return_value = _make_chat_member(target)
+        # Simulate a forum topic anchor: reply_to_message exists with forum_topic_created set
+        mock_update.message.reply_to_message = MagicMock()
+        mock_update.message.reply_to_message.from_user = _make_target_user(user_id=11111, full_name="Topic Creator")
+        mock_update.message.reply_to_message.forum_topic_created = MagicMock()
+        mock_context.args = ["67890", "reason"]
+
+        with patch("bot.handlers.warn.get_group_config_for_update", return_value=mock_registry.get(-1001234567890)):
+            await handle_warn_command(mock_update, mock_context)
+
+        mock_context.bot.get_chat_member.assert_called_once_with(
+            chat_id=-1001234567890, user_id=67890
+        )
+        mock_context.bot.send_message.assert_called_once()
+        call_kwargs = mock_context.bot.send_message.call_args.kwargs
+        assert "badmember" in call_kwargs["text"]
+        assert "11111" not in call_kwargs["text"]
+
+    async def test_forum_topic_anchor_no_args_shows_usage(
+        self, mock_update, mock_context, mock_registry
+    ):
+        """Forum topic anchor with no args shows usage (not treated as reply)."""
+        mock_update.message.reply_to_message = MagicMock()
+        mock_update.message.reply_to_message.from_user = _make_target_user(user_id=11111)
+        mock_update.message.reply_to_message.forum_topic_created = MagicMock()
+        mock_context.args = []
+
+        with patch("bot.handlers.warn.get_group_config_for_update", return_value=mock_registry.get(-1001234567890)):
+            await handle_warn_command(mock_update, mock_context)
+
+        mock_update.message.reply_text.assert_called_once()
+        assert "Penggunaan" in mock_update.message.reply_text.call_args.args[0]
+        mock_context.bot.send_message.assert_not_called()
+
+    async def test_username_mode_with_reason(
+        self, mock_update, mock_context, mock_registry
+    ):
+        """Admin uses /warn @username <reason>."""
+        mock_context.args = ["@badmember", "stop", "spamming"]
+
+        with patch("bot.handlers.warn.get_group_config_for_update", return_value=mock_registry.get(-1001234567890)):
+            await handle_warn_command(mock_update, mock_context)
+
+        mock_context.bot.get_chat_member.assert_not_called()
+        mock_context.bot.send_message.assert_called_once()
+        call_kwargs = mock_context.bot.send_message.call_args.kwargs
+        assert "@badmember" in call_kwargs["text"]
+        assert "stop spamming" in call_kwargs["text"]
+
+    async def test_username_mode_without_reason(
+        self, mock_update, mock_context, mock_registry
+    ):
+        """Admin uses /warn @username with no reason."""
+        mock_context.args = ["@badmember"]
+
+        with patch("bot.handlers.warn.get_group_config_for_update", return_value=mock_registry.get(-1001234567890)):
+            await handle_warn_command(mock_update, mock_context)
+
+        mock_context.bot.get_chat_member.assert_not_called()
+        mock_context.bot.send_message.assert_called_once()
+        call_kwargs = mock_context.bot.send_message.call_args.kwargs
+        assert "@badmember" in call_kwargs["text"]
+        assert "patuhi aturan grup" in call_kwargs["text"]
+
+    async def test_username_mode_takes_priority_over_reply(
+        self, mock_update, mock_context, mock_registry
+    ):
+        """@username takes priority over a real reply."""
+        reply_target = _make_target_user(user_id=11111, full_name="Replied User", username="replieduser")
+        mock_update.message.reply_to_message = _make_reply_message(reply_target)
+        mock_context.args = ["@badmember", "reason"]
+
+        with patch("bot.handlers.warn.get_group_config_for_update", return_value=mock_registry.get(-1001234567890)):
+            await handle_warn_command(mock_update, mock_context)
+
+        mock_context.bot.get_chat_member.assert_not_called()
+        mock_context.bot.send_message.assert_called_once()
+        call_kwargs = mock_context.bot.send_message.call_args.kwargs
+        assert "@badmember" in call_kwargs["text"]
+        assert "replieduser" not in call_kwargs["text"]
+        assert "reason" in call_kwargs["text"]
+
+    async def test_username_mode_skips_bot_and_self_checks(
+        self, mock_update, mock_context, mock_registry
+    ):
+        """Username mode cannot check is_bot/self since we don't have the User object."""
+        mock_context.args = ["@somebot"]
+
+        with patch("bot.handlers.warn.get_group_config_for_update", return_value=mock_registry.get(-1001234567890)):
+            await handle_warn_command(mock_update, mock_context)
+
+        mock_context.bot.send_message.assert_called_once()