butler/gateway/bot_loop_guard.py

Rama main · réplica de lectura del repositorio original.

"""Suppress bot-to-bot reply loops in group chats (OpenClaw pair-loop-guard subset)."""

from __future__ import annotations

import logging
import os
import re
import threading
import time
from collections import deque

from butler.utilities.env_parse import env_truthy, int_env
from butler.defaults.env_defaults import (
    BOT_LOOP_WHITELIST_DEFAULT,
    SAFETY_BOT_LOOP_GUARD_DEFAULT,
)

logger = logging.getLogger(__name__)

_WINDOW_SEC = 120.0
_LOCK = threading.RLock()
_PAIR_COUNTS: dict[str, deque[float]] = {}


def bot_loop_guard_enabled() -> bool:
    return bool(env_truthy("BUTLER_BOT_LOOP_GUARD", default=SAFETY_BOT_LOOP_GUARD_DEFAULT))


def pair_threshold() -> int:
    try:
        return int(int_env("BUTLER_BOT_LOOP_PAIR_THRESHOLD", 6, min=2))
    except ValueError:
        return 6


def _pair_key(chat_id: str, sender_id: str) -> str:
    return f"{chat_id}::{sender_id}"


def _is_whitelisted(chat_id: str) -> bool:
    raw = os.getenv("BUTLER_BOT_LOOP_WHITELIST", BOT_LOOP_WHITELIST_DEFAULT).strip()
    if not raw:
        return False
    allowed = {x.strip() for x in raw.split(",") if x.strip()}
    return chat_id in allowed


def looks_like_bot_sender(sender_id: str, text: str) -> bool:
    """Heuristic: bot accounts or messages that @ other bots."""
    sid = str(sender_id or "").lower()
    if "bot" in sid or sid.endswith("@openim"):
        return True
    if re.search(r"@\S*bot\S*", text or "", re.I):
        return True
    return False


def record_and_should_suppress(
    *,
    chat_id: str,
    sender_id: str,
    text: str = "",
) -> tuple[bool, str]:
    """Return (suppress, reason)."""
    if not bot_loop_guard_enabled():
        return False, ""
    cid = str(chat_id or "").strip() or "unknown"
    if _is_whitelisted(cid):
        return False, ""
    if not looks_like_bot_sender(sender_id, text):
        return False, ""

    key = _pair_key(cid, sender_id)
    now = time.monotonic()
    with _LOCK:
        bucket = _PAIR_COUNTS.setdefault(key, deque())
        while bucket and (now - bucket[0]) > _WINDOW_SEC:
            bucket.popleft()
        bucket.append(now)
        count = len(bucket)

    if count >= pair_threshold():
        logger.info(
            "Bot loop guard suppress chat=%s sender=%s count=%d",
            cid,
            sender_id,
            count,
        )
        from butler.gateway.bot_loop_guard_ops import record_bot_loop_suppressed_event_safe

        record_bot_loop_suppressed_event_safe(
            chat_id=cid,
            sender_id=sender_id,
            count=count,
        )
        return True, f"bot_loop_guard count={count}"
    return False, ""