Implement plans/01: cache cap + RetryAfter helper + handler refactor + /status command (#22)

* refactor: extract helpers from 5 complex handlers (Phase 3)

Behavior-preserving readability refactor with zero test changes:

1. bio_bait.py: extract enforce delete+restrict+notify into
   _enforce_bio_bait_restriction() — parent raises ApplicationHandlerStop
2. dm.py: extract unrestrict loop into _unrestrict_in_groups()
   returning success count — parent handles response messages
3. anti_spam.py: extract guard clauses into
   _should_skip_new_user_spam_check() predicate
4. message.py: extract is_bot guard into
   _should_skip_profile_check() predicate
5. check.py: extract callback data parsing into
   _parse_warn_callback_data() returning (user_id, missing_code) or None

All 948 tests pass, coverage 99% (22 missed, unchanged).
Ruff + mypy clean.

* fix(duplicate_spam): cap in-memory recent-messages dict to prevent unbounded growth

* add retry-after-helper: send_message_with_retry + restrict_chat_member_with_retry

- Two helpers in telegram_utils.py: RetryAfter-aware wrappers that sleep
  + retry once, return False on second RetryAfter, re-raise other errors
- unrestrict_user uses restrict_chat_member_with_retry internally
- 4 call sites updated: check.py handle_warn_callback, scheduler.py
  auto-restrict loop, dm.py unrestrict loop, bio_bait.py monitor alert
- 8 unit tests for helpers, 2 integration tests for bio_bait error paths
- 958 tests pass, 99% coverage, ruff clean, mypy clean

ponytail: scheduler.py RetryAfter-fallback continue (lines 95-98) not
covered by tests; add scheduler-level RetryAfter test if it becomes a
real problem.

* feat(status): add /status DM admin command + job timestamp tracking

* fix(retry-after): handle int | timedelta in RetryAfter.retry_after

* chore: gitignore .pi-subagents artifacts (consistent with reviews/)

* fix(review): inline dead _should_skip_profile_check, gate auto-restrict stamp on real work

Applies multi-model review feedback:
- Remove premature-abstraction _should_skip_profile_check predicate (M1)
- Track processed_any in auto_restrict to skip stamp on no-op cycles (R3)
- Add logger.warning for missing message/from_user in status handler (C3)
- Drop group_titles read; use str(gid) directly (M2, M6)

* chore: gitignore plans/ (consistent with docs/, reviews/)

Plan docs are working notes for a session, not source. Untrack the one
that snuck into the improvement branch's history.

* fix(review): apply additional reviewer feedback

- verify.py: catch RuntimeError from unrestrict_user retry exhaustion;
  wrap clearance notification with send_message_with_retry
- bio_bait.py: wrap enforcement path (restrict + notify) with retry helpers
  (the monitor-only alert path was already wrapped)
- scheduler.py: stamp last_auto_restrict unconditionally on every successful
  run (matches admin_cache behavior)
- telegram_utils.py: cap retry sleep at 30s to prevent one bad RetryAfter
  from stalling per-group loop for a minute-plus
- status.py: drop dead title variable (was duplicating group_id)

* fix(status): drop duplicate group_id display in /status group line

title = str(gid) just re-stringified the same group_id with nothing else
feeding it (GroupConfig has no name/title field), so each group line
printed the ID twice: `-100123` — -100123. Show it once.

* style(telegram_utils): use f-strings for retry warning/error logs

Author: rezhajulio

.gitignore

@@ -14,9 +14,11 @@ data/
 
 # Agent/planning docs
 docs/
+plans/
 .pi/
 
 # Local review artifacts (parallel-reviewer outputs)
 reviews/
+.pi-subagents/
 
 .tokensave/

src/bot/database/service.py

@@ -634,6 +634,17 @@ def get_all_pending_captchas(self) -> list[PendingCaptchaValidation]:
             statement = select(PendingCaptchaValidation)
             return list(session.exec(statement).all())
 
+    def get_all_new_user_probations(self) -> list[NewUserProbation]:
+        """
+        Get all new-user probation records.
+
+        Returns:
+            list[NewUserProbation]: All probation records.
+        """
+        with Session(self._engine) as session:
+            statement = select(NewUserProbation)
+            return list(session.exec(statement).all())
+
     def start_new_user_probation(self, user_id: int, group_id: int) -> NewUserProbation:
         """
         Start or refresh probation for a new user.

src/bot/handlers/anti_spam.py

@@ -348,6 +348,18 @@ async def handle_inline_keyboard_spam(
     )
 
 
+def _should_skip_new_user_spam_check(update, context, group_config) -> bool:
+    """Check if new user spam handler should skip this message."""
+    if group_config is None:
+        return True
+    user = update.message.from_user
+    if user.is_bot:
+        return True
+    if is_user_admin_or_trusted(context, group_config.group_id, user.id):
+        return True
+    return False
+
+
 async def handle_new_user_spam(
     update: Update, context: ContextTypes.DEFAULT_TYPE
 ) -> None:
@@ -371,15 +383,7 @@ async def handle_new_user_spam(
     group_config = get_group_config_for_update(update)
     user = update.message.from_user
 
-    # Only process messages from monitored groups
-    if group_config is None:
-        return
-
-    # Ignore bots
-    if user.is_bot:
-        return
-
-    if is_user_admin_or_trusted(context, group_config.group_id, user.id):
+    if _should_skip_new_user_spam_check(update, context, group_config):
         return
 
     db = get_database()

src/bot/handlers/bio_bait.py

@@ -37,7 +37,13 @@
     WHITELISTED_TELEGRAM_PATHS,
 )
 from bot.group_config import get_group_config_for_update
-from bot.services.telegram_utils import get_user_mention, is_user_admin_or_trusted, is_url_whitelisted
+from bot.services.telegram_utils import (
+    get_user_mention,
+    is_user_admin_or_trusted,
+    is_url_whitelisted,
+    restrict_chat_member_with_retry,
+    send_message_with_retry,
+)
 
 # Filter for bio-bait handler registration in main.py.
 # Must NOT restrict to TEXT|CAPTION so non-text messages (e.g. photos
@@ -269,7 +275,14 @@ async def send_monitor_alert_to_owner(
 
     try:
         for chunk in _chunk_telegram_text(alert_text):
-            await context.bot.send_message(chat_id=alert_chat_id, text=chunk)
+            ok = await send_message_with_retry(
+                context.bot, chat_id=alert_chat_id, text=chunk
+            )
+            if not ok:
+                logger.error(
+                    f"Failed to send bio bait monitor alert chunk: user_id={user_id}, group_id={group_id}"
+                )
+                return False
         return True
     except Exception:
         logger.error(f"Failed to send bio bait monitor alert: user_id={user_id}, group_id={group_id}")
@@ -312,6 +325,64 @@ async def get_cached_user_bio(
         cache[user_id] = (now + USER_BIO_FAILURE_CACHE_TTL_SECONDS, _BIO_CACHE_FAILURE)
         return None
 
+
+async def _enforce_bio_bait_restriction(
+    update: Update,
+    context: ContextTypes.DEFAULT_TYPE,
+    group_config,
+    user,
+    detection_reason: str,
+) -> None:
+    """Delete, restrict, notify for confirmed bio bait spam. Caller raises ApplicationHandlerStop."""
+    user_mention = get_user_mention(user)
+
+    try:
+        await update.message.delete()
+        logger.info(f"Deleted bio bait spam from user_id={user.id}")
+    except Exception:
+        logger.error(f"Failed to delete bio bait spam: user_id={user.id}", exc_info=True)
+
+    restricted = False
+    try:
+        await restrict_chat_member_with_retry(
+            context.bot,
+            chat_id=group_config.group_id,
+            user_id=user.id,
+            permissions=RESTRICTED_PERMISSIONS,
+        )
+        restricted = True
+        clear_cached_user_bio(context, user.id)
+        logger.info(f"Restricted user_id={user.id} for bio bait spam")
+    except Exception:
+        logger.error(f"Failed to restrict user for bio bait spam: user_id={user.id}", exc_info=True)
+
+    try:
+        if detection_reason == "bio_links":
+            template = (
+                BIO_LINK_SPAM_NOTIFICATION if restricted
+                else BIO_LINK_SPAM_NOTIFICATION_NO_RESTRICT
+            )
+        else:
+            template = (
+                BIO_BAIT_SPAM_NOTIFICATION if restricted
+                else BIO_BAIT_SPAM_NOTIFICATION_NO_RESTRICT
+            )
+        notification_text = template.format(
+            user_mention=user_mention,
+            rules_link=group_config.rules_link,
+        )
+        await send_message_with_retry(
+            context.bot,
+            chat_id=group_config.group_id,
+            message_thread_id=group_config.warning_topic_id,
+            text=notification_text,
+            parse_mode="Markdown",
+        )
+        logger.info(f"Sent bio bait spam notification for user_id={user.id}")
+    except Exception:
+        logger.error(f"Failed to send bio bait spam notification: user_id={user.id}", exc_info=True)
+
+
 async def handle_bio_bait_spam(
     update: Update, context: ContextTypes.DEFAULT_TYPE
 ) -> None:
@@ -392,50 +463,5 @@ async def handle_bio_bait_spam(
         )
         return
 
-    user_mention = get_user_mention(user)
-
-    try:
-        await update.message.delete()
-        logger.info(f"Deleted bio bait spam from user_id={user.id}")
-    except Exception:
-        logger.error(f"Failed to delete bio bait spam: user_id={user.id}", exc_info=True)
-
-    restricted = False
-    try:
-        await context.bot.restrict_chat_member(
-            chat_id=group_config.group_id,
-            user_id=user.id,
-            permissions=RESTRICTED_PERMISSIONS,
-        )
-        restricted = True
-        clear_cached_user_bio(context, user.id)
-        logger.info(f"Restricted user_id={user.id} for bio bait spam")
-    except Exception:
-        logger.error(f"Failed to restrict user for bio bait spam: user_id={user.id}", exc_info=True)
-
-    try:
-        if detection_reason == "bio_links":
-            template = (
-                BIO_LINK_SPAM_NOTIFICATION if restricted
-                else BIO_LINK_SPAM_NOTIFICATION_NO_RESTRICT
-            )
-        else:
-            template = (
-                BIO_BAIT_SPAM_NOTIFICATION if restricted
-                else BIO_BAIT_SPAM_NOTIFICATION_NO_RESTRICT
-            )
-        notification_text = template.format(
-            user_mention=user_mention,
-            rules_link=group_config.rules_link,
-        )
-        await context.bot.send_message(
-            chat_id=group_config.group_id,
-            message_thread_id=group_config.warning_topic_id,
-            text=notification_text,
-            parse_mode="Markdown",
-        )
-        logger.info(f"Sent bio bait spam notification for user_id={user.id}")
-    except Exception:
-        logger.error(f"Failed to send bio bait spam notification: user_id={user.id}", exc_info=True)
-
+    await _enforce_bio_bait_restriction(update, context, group_config, user, detection_reason)
     raise ApplicationHandlerStop

src/bot/handlers/check.py

@@ -30,6 +30,7 @@
     get_user_mention,
     get_user_mention_by_id,
     require_admin_dm_target,
+    send_message_with_retry,
 )
 from bot.services.user_checker import check_user_profile
 
@@ -205,6 +206,17 @@ async def handle_check_forwarded_message(
         logger.error(f"Error checking forwarded user {user_id}: {e}", exc_info=True)
 
 
+def _parse_warn_callback_data(data: str) -> tuple[int, str] | None:
+    """Parse warn callback data (warn:<user_id>:<missing_code>). Returns (user_id, missing_code) or None."""
+    try:
+        parts = data.split(":")
+        user_id = int(parts[1])
+        missing_code = parts[2] if len(parts) > 2 else ""
+        return (user_id, missing_code)
+    except (IndexError, ValueError):
+        return None
+
+
 async def handle_warn_callback(
     update: Update, context: ContextTypes.DEFAULT_TYPE
 ) -> None:
@@ -231,14 +243,12 @@ async def handle_warn_callback(
         return
 
     # Parse callback data: warn:<user_id>:<missing_code>
-    try:
-        parts = query.data.split(":")
-        target_user_id = int(parts[1])
-        missing_code = parts[2] if len(parts) > 2 else ""
-    except (IndexError, ValueError):
+    parsed = _parse_warn_callback_data(query.data)
+    if parsed is None:
         await query.edit_message_text("❌ Data callback tidak valid.")
         logger.error(f"Invalid callback_data format: {query.data}")
         return
+    target_user_id, missing_code = parsed
 
     # Build missing items text
     missing_items = []
@@ -264,16 +274,18 @@ async def handle_warn_callback(
                 rules_link=group_config.rules_link,
             )
             try:
-                await context.bot.send_message(
+                ok = await send_message_with_retry(
+                    context.bot,
                     chat_id=group_config.group_id,
                     message_thread_id=group_config.warning_topic_id,
                     text=warn_message,
                     parse_mode="Markdown",
                 )
-                sent_to_any = True
-                logger.info(
-                    f"Admin {admin_user_id} sent warning to user {target_user_id} in group {group_config.group_id}"
-                )
+                if ok:
+                    sent_to_any = True
+                    logger.info(
+                        f"Admin {admin_user_id} sent warning to user {target_user_id} in group {group_config.group_id}"
+                    )
             except Exception as e:
                 logger.error(f"Failed to send warning to group {group_config.group_id}: {e}")