hendrik/services/xmpp_client.py

import asyncio
import logging
import random
import signal
import threading
import traceback
from datetime import datetime

from slixmpp import ClientXMPP
from services.session_manager import SessionManager
from lib.agent_loop import run_agent_loop

import config
from tools.roleplayer import should_respond
from lib import personality
from lib import ragroleplay

# Anti-ban: delay constants for MUC rejoin behavior
MUC_REJOIN_INITIAL_DELAY   = 30.0  # detik, delay awal sebelum rejoin (lebih gentle)
MUC_REJOIN_BACKOFF_MULT    = 1.5   # multiplier exponential backoff (lebih gentle)
MUC_REJOIN_MAX_DELAY       = 600.0 # detik, batas max backoff (10 menit)
MUC_REJOIN_COOLDOWN        = 60.0  # detik, cooldown minimum antar rejoin attempt
MUC_NICK_SUFFIX_MAX        = 3     # max coba nick alternatif (anti-ban: jangan terlalu banyak)
MUC_JOIN_JITTER            = 0.3   # jitter 30% untuk randomize delay

# Anti-ban: Message queue rate limiting
MSG_QUEUE_DELAY_MIN        = 3.0   # detik, delay minimum antar pesan keluar
MSG_QUEUE_DELAY_MAX        = 8.0   # detik, delay maksimum antar pesan keluar
MSG_QUEUE_JITTER           = 0.25  # jitter 25% untuk randomize delay
MSG_TYPING_SPEED_MIN       = 8.0   # chars/sec, minimum typing speed
MSG_TYPING_SPEED_MAX       = 20.0  # chars/sec, maksimum typing speed
MSG_READ_DELAY_MIN         = 2.0   # detik, minimum read delay
MSG_READ_DELAY_MAX         = 5.0   # detik, maksimum read delay

# Anti-ban: Adaptive rate limiting (deteksi throttle dari server)
ADAPTIVE_WINDOW_SIZE       = 20    # jumlah pesan terakhir untuk tracking
ADAPTIVE_ERROR_THRESHOLD   = 0.15  # 15% error rate = server mulai throttle
ADAPTIVE_SLOW_THRESHOLD    = 5.0   # detik, response time > ini = server lambat
ADAPTIVE_BACKOFF_MULT      = 2.0   # multiplier delay saat throttle detected
ADAPTIVE_RECOVERY_TIME     = 300   # detik, waktu tunggu sebelum recovery ke normal

# Anti-ban: Connection behavior (sangat penting untuk conversations.im)
CONNECTION_KEEPALIVE       = True   # enable keepalive
CONNECTION_KEEPALIVE_INTERVAL = 300 # detik, ping interval (5 menit, jangan terlalu sering)
AUTO_RECONNECT_ENABLED     = False  # JANGAN auto reconnect instant (bot-like)
RECONNECT_DELAY_MIN        = 30.0   # detik, delay minimum sebelum reconnect manual
RECONNECT_DELAY_MAX        = 120.0  # detik, delay maksimum sebelum reconnect


def _ts():
    return datetime.now().strftime('%H:%M:%S')


def _setup_debug_logging():
    """Debug: aktifkan logging DEBUG slixmpp untuk menampilkan SEMUA stanza XML keluar/masuk.

    Sumber utama diagnosa:
      - [slixmpp.xmlstream.xmlstream] SEND: <xml>  -> stanza yang benar2 dikirim ke server
      - [slixmpp.xmlstream.xmlstream] RECV: <xml>  -> stanza yang diterima dari server
    """
    if not config.XMPP_DEBUG:
        return
    logging.basicConfig(
        level=logging.DEBUG,
        format='[%(asctime)s][%(name)s] %(message)s',
        datefmt='%H:%M:%S',
        force=True,
    )
    logging.getLogger('slixmpp').setLevel(logging.DEBUG)
    print(f'[{_ts()}] DEBUG: XMPP debug logging ENABLED (semua stanza XML akan dicetak)', flush=True)


def _dbg(msg):
    """Debug: cetak pesan debug hanya jika config.XMPP_DEBUG aktif."""
    if config.XMPP_DEBUG:
        print(f'[{_ts()}] DEBUG: {msg}', flush=True)


def _add_jitter(delay: float, jitter_factor: float = MSG_QUEUE_JITTER) -> float:
    """Tambahkan jitter acak ke delay untuk menghindari pola terdeteksi."""
    jitter = delay * jitter_factor
    return delay + random.uniform(-jitter, jitter)


def _typing_delay(text: str) -> float:
    """Hitung delay mengetik (detik) proporsional dengan panjang teks."""
    char_count = len(text) if text else 0
    # Random typing speed untuk lebih human-like
    typing_speed = random.uniform(MSG_TYPING_SPEED_MIN, MSG_TYPING_SPEED_MAX)
    delay = char_count / typing_speed
    # Clamp antara 1-15 detik, tambah jitter
    return _add_jitter(max(1.0, min(delay, 15.0)))


async def _read_delay():
    """Delay simulasi membaca pesan user."""
    delay = random.uniform(MSG_READ_DELAY_MIN, MSG_READ_DELAY_MAX)
    await asyncio.sleep(delay)


class AdaptiveRateLimiter:
    """Anti-ban: adaptive rate limiting yang mendeteksi throttle dari server."""
    
    def __init__(self):
        self._send_times: list[float] = []      # timestamp pengiriman pesan
        self._errors: list[bool] = []            # apakah pengiriman error
        self._response_times: list[float] = []   # waktu response server (jika ada)
        self._throttle_detected = False
        self._throttle_start: float | None = None
        self._backoff_multiplier = 1.0
        
    def record_send(self, success: bool, response_time: float | None = None):
        """Catat hasil pengiriman pesan."""
        now = asyncio.get_event_loop().time()
        self._send_times.append(now)
        self._errors.append(not success)
        
        if response_time is not None:
            self._response_times.append(response_time)
        
        # Keep only recent window
        if len(self._send_times) > ADAPTIVE_WINDOW_SIZE:
            self._send_times.pop(0)
            self._errors.pop(0)
        if len(self._response_times) > ADAPTIVE_WINDOW_SIZE:
            self._response_times.pop(0)
        
        self._check_throttle()
    
    def _check_throttle(self):
        """Cek apakah server mulai throttle."""
        if len(self._errors) < 5:  # butuh minimal sample
            return
        
        # Hitung error rate
        error_rate = sum(self._errors) / len(self._errors)
        
        # Hitung avg response time
        avg_response = 0
        if self._response_times:
            avg_response = sum(self._response_times) / len(self._response_times)
        
        # Deteksi throttle: error rate tinggi ATAU response time lambat
        was_throttled = self._throttle_detected
        
        if error_rate >= ADAPTIVE_ERROR_THRESHOLD or avg_response >= ADAPTIVE_SLOW_THRESHOLD:
            if not self._throttle_detected:
                self._throttle_detected = True
                self._throttle_start = asyncio.get_event_loop().time()
                self._backoff_multiplier = ADAPTIVE_BACKOFF_MULT
                print(f'[{_ts()}] ADAPTIVE: Throttle detected! '
                      f'error_rate={error_rate:.1%}, avg_response={avg_response:.1f}s, '
                      f'backing off x{self._backoff_multiplier}', flush=True)
        else:
            # Recovery: cek apakah sudah cukup waktu untuk kembali normal
            if self._throttle_detected and self._throttle_start:
                elapsed = asyncio.get_event_loop().time() - self._throttle_start
                if elapsed >= ADAPTIVE_RECOVERY_TIME:
                    self._throttle_detected = False
                    self._backoff_multiplier = 1.0
                    print(f'[{_ts()}] ADAPTIVE: Recovered to normal rate', flush=True)
    
    def get_delay_multiplier(self) -> float:
        """Dapatkan multiplier untuk delay saat ini."""
        return self._backoff_multiplier
    
    def is_throttled(self) -> bool:
        """Apakah sedang dalam kondisi throttle."""
        return self._throttle_detected


class XMPPClient(ClientXMPP):
    def __init__(self, jid, password, llm_client, tools_definition, TOOLS,
                 TOOL_HANDLERS, build_system_prompt, agent_max_iterations,
                 muc_rooms=None):
        super().__init__(jid, password)

        self._llm = llm_client
        self._tools_def = tools_definition
        self._TOOLS = TOOLS
        self._TOOL_HANDLERS = TOOL_HANDLERS
        self._build_system_prompt = build_system_prompt
        self._max_iterations = agent_max_iterations
        self._skill = config.AGENT_SKILL
        self._muc_rooms = muc_rooms or []
        # Custom nick dari config, fallback ke username JID
        self._muc_nick = config.XMPP_NICKNAME.strip() or jid.split('@')[0]
        self._muc_nick_suffix = 0  # counter untuk nick alternatif saat 409
        self._muc_ready: set[str] = set()

        self._session_mgr = SessionManager()
        self._loop = None
        self._stopped: asyncio.Event | None = None

        # Anti-ban: MUC rejoin tracking per room
        self._muc_rejoin_attempts: dict[str, int] = {}   # room -> jumlah attempt
        self._muc_rejoin_tasks: dict[str, asyncio.Task] = {}  # room -> pending rejoin task
        self._muc_last_join: dict[str, datetime] = {}    # room -> terakhir join (cooldown)

        # Anti-ban: Message queue untuk rate limiting
        self._msg_queue: asyncio.Queue | None = None
        self._msg_worker_task: asyncio.Task | None = None
        self._msg_worker_running = False
        
        # Anti-ban: Adaptive rate limiter
        self._rate_limiter = AdaptiveRateLimiter()
        
        # Anti-ban: reconnect tracking (jangan instant reconnect)
        self._reconnect_scheduled = False
        self._last_disconnect: datetime | None = None

        # Anti-ban: JANGAN auto reconnect instant (bot-like behavior)
        self.auto_reconnect = AUTO_RECONNECT_ENABLED

        # ── Anti-ban (conversations.im): ringkasan solusi yang membuat akun
        #    tidak dianggap spam/di-block (penyebab: klien default slixmpp
        #    mengiklankan identity `client/bot` + caps node slixmpp) ────────────
        #
        #  1. JANGAN kirim 'subscribe' balik saat ada yang subscribe (contact
        #     farming = red flag). auto_authorize=True utk tetap menerima,
        #     auto_subscribe=False utk tidak membalas subscribe.
        #  2. Caps node netral (bukan http://slixmpp.com/ver/X.Y.Z) supaya klien
        #     tidak mudah dikenali sebagai slixmpp.
        #  3. Disco identity `client/console` di-set di _on_session_start (setelah
        #     bind). CATATAN: slixmpp menyimpan identity per key `boundjid.full` —
        #     di __init__ boundjid masih bare-JID, lookup runtime (setelah bind)
        #     memakai JID ber-resource, sehingga identity di __init__ TIDAK pernah
        #     ditemukan dan server melihat fallback `client/bot`. Harus di session_start.
        self.roster.auto_authorize = True    # terima subscribe masuk (sopan)
        self.roster.auto_subscribe = False   # JANGAN kirim subscribe balik

        # Anti-ban: register plugin dengan konfigurasi gentle
        self.register_plugin('xep_0030')  # Service Discovery
        self.register_plugin('xep_0045')  # MUC
        self.register_plugin('xep_0199', {  # XMPP Ping
            'keepalive': CONNECTION_KEEPALIVE,
            'interval': CONNECTION_KEEPALIVE_INTERVAL,
            'timeout': 30,
        })
        # Muat plugin sekarang agar bisa set caps node sebelum konek
        self.init_plugins()

        if self.plugin.get('xep_0115', None) is not None:
            self.plugin['xep_0115'].caps_node = 'https://hendrik.local/caps'

        self.add_event_handler('session_start', self._on_session_start)
        self.add_event_handler('message', self._on_message)
        self.add_event_handler('groupchat_message', self._on_groupchat_message)
        self.add_event_handler('disconnected', self._on_disconnected)
        self.add_event_handler('connected', self._on_connected)
        self.add_event_handler('groupchat_presence', self._on_muc_presence)
        # Anti-ban: handler untuk melihat error dari server (perlu selalu aktif)
        self.add_event_handler('message_error', self._on_message_error)
        self.add_event_handler('stream_error', self._on_stream_error)
        # Anti-ban: kontrol manual atas subscription request (jangan auto-subscribe balik)
        self.add_event_handler('roster_subscription_request', self._on_roster_subscription_request)

    def _get_muc_nick(self, room: str) -> str:
        """Anti-ban: resolve nick untuk room, coba nick alternatif kalau conflict."""
        base = config.XMPP_NICKNAME.strip() or self._muc_nick
        suffix = self._muc_rejoin_attempts.get("_nick_" + room, 0)
        if suffix == 0:
            return base
        # Anti-ban: append suffix untuk menghindari 409 Conflict
        return f"{base}_{suffix}"

    def _calc_rejoin_delay(self, room: str) -> float:
        """Anti-ban: hitung delay rejoin dengan exponential backoff + jitter."""
        attempts = self._muc_rejoin_attempts.get(room, 0)
        delay = MUC_REJOIN_INITIAL_DELAY * (MUC_REJOIN_BACKOFF_MULT ** attempts)
        delay = min(delay, MUC_REJOIN_MAX_DELAY)
        # Tambah jitter untuk menghindari pola terdeteksi
        return _add_jitter(delay, MUC_JOIN_JITTER)

    def _schedule_muc_rejoin(self, room: str):
        """Anti-ban: schedule rejoin room dengan backoff & cooldown."""
        # Cancel pending rejoin task untuk room yang sama (anti-ban: avoid duplicate rejoin)
        pending = self._muc_rejoin_tasks.get(room)
        if pending and not pending.done():
            pending.cancel()
            print(f'[{_ts()}] MUC [{room}] Cancelled pending rejoin (new trigger)', flush=True)

        # Check cooldown: jangan rejoin terlalu cepat berturut-turut
        now = datetime.now()
        last_join = self._muc_last_join.get(room)
        if last_join:
            elapsed = (now - last_join).total_seconds()
            if elapsed < MUC_REJOIN_COOLDOWN:
                # Anti-ban: too soon, schedule delayed rejoin instead of immediate
                cooldown_left = MUC_REJOIN_COOLDOWN - elapsed
                print(f'[{_ts()}] MUC [{room}] Cooldown active ({cooldown_left:.0f}s left), delaying rejoin', flush=True)
                delay = cooldown_left + self._calc_rejoin_delay(room)
            else:
                delay = self._calc_rejoin_delay(room)
        else:
            delay = self._calc_rejoin_delay(room)

        # Increment attempt counter (anti-ban: track for exponential backoff)
        attempts = self._muc_rejoin_attempts.get(room, 0) + 1
        self._muc_rejoin_attempts[room] = attempts

        print(f'[{_ts()}] MUC [{room}] Rejoin scheduled in {delay:.0f}s (attempt #{attempts})', flush=True)

        if self._loop and not self._loop.is_closed():
            task = asyncio.run_coroutine_threadsafe(
                self._muc_rejoin_coro(room, delay), self._loop
            )
            self._muc_rejoin_tasks[room] = task

    async def _muc_rejoin_coro(self, room: str, delay: float):
        """Anti-ban: coroutine untuk rejoin room setelah delay."""
        try:
            await asyncio.sleep(delay)
            # Double-check: jangan rejoin kalau sudah di _muc_ready
            if room in self._muc_ready:
                print(f'[{_ts()}] MUC [{room}] Already ready, skip rejoin', flush=True)
                return
            nick = self._get_muc_nick(room)
            print(f'[{_ts()}] MUC [{room}] Rejoining as {nick}...', flush=True)
            await self.plugin['xep_0045'].join_muc_wait(room, nick, maxstanzas=0)
            self._muc_last_join[room] = datetime.now()
            # _muc_ready akan di-set oleh _on_muc_presence saat join berhasil
            self._muc_rejoin_attempts.pop(room, None)
            self._muc_rejoin_attempts.pop("_nick_" + room, None)
            # Catat sukses untuk adaptive rate limiting
            self._rate_limiter.record_send(success=True)
            print(f'[{_ts()}] MUC [{room}] Rejoin successful as {nick}', flush=True)
        except asyncio.CancelledError:
            print(f'[{_ts()}] MUC [{room}] Rejoin cancelled', flush=True)
        except Exception as e:
            print(f'[{_ts()}] MUC [{room}] Rejoin failed: {e}', flush=True)
            # Catat error untuk adaptive rate limiting
            self._rate_limiter.record_send(success=False)
            # Anti-ban: handle 409 Conflict - nick sudah dipakai orang lain
            if '409' in str(e) or 'conflict' in str(e).lower():
                nick_attempts = self._muc_rejoin_attempts.get("_nick_" + room, 0)
                if nick_attempts < MUC_NICK_SUFFIX_MAX:
                    # Anti-ban: coba nick alternatif (lily_, lily__)
                    self._muc_rejoin_attempts["_nick_" + room] = nick_attempts + 1
                    new_nick = self._get_muc_nick(room)
                    print(f'[{_ts()}] MUC [{room}] Nick conflict, trying alternative: {new_nick}', flush=True)
                    # Retry segera dengan nick baru (tanpa backoff rejoin, tapi tetap ada delay biasa)
                    self._schedule_muc_rejoin(room)
                else:
                    # Anti-ban: semua nick alternativehabis, stop retry untuk avoid ban
                    print(f'[{_ts()}] MUC [{room}] All nick variations exhausted, skipping room', flush=True)
                    print(f'[{_ts()}] MUC [{room}] Set XMPP_NICKNAME in .env to a unique nick', flush=True)
            else:
                # Anti-ban: error biasa (network, dll), retry with backoff
                self._schedule_muc_rejoin(room)

    async def _on_connected(self, event):
        print(f'[{_ts()}] XMPP connected', flush=True)
        _dbg(f'connected state: authenticated={self.authenticated}, bound={self.bound}, '
             f'sessionstarted={self.sessionstarted}, '
             f'_session_started={getattr(self, "_session_started", None)}')
        _dbg(f'connected via: {self.transport}')

    async def _on_disconnected(self, event):
        print(f'[{_ts()}] XMPP disconnected', flush=True)
        # Anti-ban: cancel all pending rejoin tasks on disconnect
        for room, task in list(self._muc_rejoin_tasks.items()):
            if not task.done():
                task.cancel()
                print(f'[{_ts()}] MUC [{room}] Cancelled pending rejoin (disconnected)', flush=True)
        self._muc_rejoin_tasks.clear()
        
        # Anti-ban: schedule manual reconnect dengan delay (jangan instant)
        if not self._stopped.is_set() and not self._reconnect_scheduled:
            self._reconnect_scheduled = True
            self._last_disconnect = datetime.now()
            delay = random.uniform(RECONNECT_DELAY_MIN, RECONNECT_DELAY_MAX)
            print(f'[{_ts()}] Will attempt reconnect in {delay:.0f}s...', flush=True)
            asyncio.create_task(self._delayed_reconnect(delay))

    async def _delayed_reconnect(self, delay: float):
        """Anti-ban: reconnect dengan delay, bukan instant."""
        await asyncio.sleep(delay)
        
        if self._stopped.is_set():
            return
            
        print(f'[{_ts()}] Attempting manual reconnect...', flush=True)
        try:
            await self.connect()
        except Exception as e:
            print(f'[{_ts()}] Reconnect failed: {e}', flush=True)
            # Schedule another reconnect with longer delay
            self._reconnect_scheduled = False
            self._on_disconnected(None)
        else:
            self._reconnect_scheduled = False

    async def _on_session_start(self, event):
        _dbg('--- SESSION START ---')
        _dbg(f'session state: authenticated={self.authenticated}, bound={self.bound}, '
             f'sessionstarted={self.sessionstarted}, '
             f'_session_started={getattr(self, "_session_started", None)}')
        _dbg(f'boundjid: full={self.boundjid.full}, bare={self.boundjid.bare}, '
             f'resource={self.boundjid.resource}, host={self.boundjid.host}')
        print(f'[{_ts()}] XMPP online as {self.boundjid.full}', flush=True)

        # Anti-ban: set disco identity SEKARANG (setelah bind, boundjid ber-resource)
        # supaya identity masuk ke caps ver (presence) DAN respon disco#info server.
        # Sebelumnya di __init__ -> tersimpan di key bare-JID -> tak pernah ditemukan
        # saat runtime -> server melihat identity fallback `client/bot`.
        self.plugin['xep_0030'].add_identity(
            category='client', itype='console', name=config.AGENT_CHARACTER.title() or 'Hendrik',
        )
        try:
            local_info = await self.plugin['xep_0030'].get_info(local=True)
            if 'disco_info' in local_info:
                identities = local_info['disco_info']['identities']
            else:
                identities = local_info['identities']
            _dbg(f'disco identity (local) after set: {identities}')
        except Exception as e:
            _dbg(f'disco identity check failed: {e}')

        # Anti-ban: send presence + request roster (wajib sebelum kirim pesan DM)
        print(f'[{_ts()}] Sending initial presence...', flush=True)
        self.send_presence()
        print(f'[{_ts()}] Requesting roster...', flush=True)
        self.get_roster()

        # Anti-ban: delay sebelum join MUC pertama agar startup tidak terlihat bot
        if self._muc_rooms:
            pre_delay = random.uniform(5.0, 15.0)
            print(f'[{_ts()}] MUC pre-join delay {pre_delay:.1f}s (anti-ban)...', flush=True)
            await asyncio.sleep(pre_delay)

        # Anti-ban: delay sebelum join MUC untuk menghindari koneksi yang terlalu agresif
        for i, room in enumerate(self._muc_rooms):
            # Delay antar room join (3-8 detik per room)
            if i > 0:
                join_delay = random.uniform(3.0, 8.0)
                print(f'[{_ts()}] MUC [{room}] Waiting {join_delay:.1f}s before join...', flush=True)
                await asyncio.sleep(join_delay)
            
            # Anti-ban: retry join dengan incremental delay & nick fallback
            success = False
            for attempt in range(1, 4):
                nick = self._get_muc_nick(room)
                try:
                    await self.plugin['xep_0045'].join_muc_wait(room, nick, maxstanzas=0)
                    print(f'[{_ts()}] Joined MUC room: {room} as {nick}', flush=True)
                    self._muc_last_join[room] = datetime.now()
                    self._muc_rejoin_attempts.pop(room, None)
                    self._muc_rejoin_attempts.pop("_nick_" + room, None)
                    success = True
                    break
                except Exception as e:
                    print(f'[{_ts()}] MUC join attempt #{attempt} failed ({room}): {e}', flush=True)
                    # Anti-ban: handle 409 Conflict - coba nick alternatif
                    if '409' in str(e) or 'conflict' in str(e).lower():
                        nick_attempts = self._muc_rejoin_attempts.get("_nick_" + room, 0)
                        if nick_attempts < MUC_NICK_SUFFIX_MAX:
                            nick_attempts += 1
                            self._muc_rejoin_attempts["_nick_" + room] = nick_attempts
                            print(f'[{_ts()}] MUC [{room}] Nick conflict, switching to: {self._get_muc_nick(room)}', flush=True)
                            # Retry segera dengan nick baru (jangan wait)
                            continue
                        else:
                            # Anti-ban: semua nick alternatif habis
                            print(f'[{_ts()}] MUC [{room}] All nick variations exhausted', flush=True)
                            break
                    elif attempt < 3:
                        # Anti-ban: error biasa, wait before retry (5s, 10s, 15s)
                        retry_delay = 5.0 * attempt
                        print(f'[{_ts()}] MUC [{room}] Retrying in {retry_delay:.0f}s...', flush=True)
                        await asyncio.sleep(retry_delay)
            if not success:
                # Anti-ban: semua attempt gagal, schedule background rejoin
                print(f'[{_ts()}] MUC [{room}] All join attempts failed, scheduling background rejoin', flush=True)
                self._schedule_muc_rejoin(room)

    def _on_message(self, msg):
        if msg['type'] not in ('chat', 'normal'):
            return
        jid = msg['from'].bare
        body = msg['body'].strip()
        if not body:
            return
        print(f'[{_ts()}] DM from {jid}: {body[:60]}', flush=True)
        # Anti-ban: proses langsung di event loop, bukan thread baru
        # Ini memastikan msg.send() dipanggil dari thread yang benar
        asyncio.create_task(self._process_dm_async(jid, body))

    def _on_message_error(self, msg):
        """Anti-ban: handle error message dari server."""
        print(f'[{_ts()}] MESSAGE ERROR from server:', flush=True)
        print(f'[{_ts()}]   Type: {msg.get("type", "unknown")}', flush=True)
        print(f'[{_ts()}]   From: {msg.get("from", "unknown")}', flush=True)
        print(f'[{_ts()}]   To: {msg.get("to", "unknown")}', flush=True)
        print(f'[{_ts()}]   Error: {msg.get("error", "unknown")}', flush=True)
        print(f'[{_ts()}]   Full stanza: {msg}', flush=True)

        # Catat error untuk adaptive rate limiting
        self._rate_limiter.record_send(success=False)

    def _on_stream_error(self, error):
        """Anti-ban: handle stream error dari server."""
        print(f'[{_ts()}] STREAM ERROR from server:', flush=True)
        print(f'[{_ts()}]   Condition: {error.get("condition", "unknown")}', flush=True)
        print(f'[{_ts()}]   Text: {error.get("text", "unknown")}', flush=True)
        print(f'[{_ts()}]   Full error: {error}', flush=True)

    async def _on_roster_subscription_request(self, presence):
        print(f'[{_ts()}] Subscription request from {presence["from"]}', flush=True)
        # Anti-ban: TIDAK merespons subscribe otomatis (auto_subscribe=False).
        # Balasan 'subscribe' balik adalah red flag (contact farming) di conversations.im.
        _dbg(f'NOT auto-responding subscribe from {presence["from"]} '
             f'(auto_subscribe={self.roster.auto_subscribe}, '
             f'auto_authorize={self.roster.auto_authorize})')

    def _on_groupchat_message(self, msg):
        if msg['type'] != 'groupchat':
            return
        room = msg['from'].bare
        nick = msg['from'].resource
        if self._is_my_nick(room, nick):
            return
        room = msg['from'].bare
        if room not in self._muc_ready:
            return
        body = msg['body'].strip()
        if not body:
            return
        print(f'[{_ts()}] MUC [{room}] <{nick}>: {body[:60]}', flush=True)
        # Anti-ban: proses langsung di event loop, bukan thread baru
        asyncio.create_task(self._process_muc_async(room, nick, body))

    def _is_my_nick(self, room: str, nick: str) -> bool:
        """Anti-ban: cek apakah nick yang dimasukan sesuai dengan nick bot di room."""
        expected = self._get_muc_nick(room)
        # Bandingkan dengan nick yang diharapkan, plus base nick tanpa suffix
        base = config.XMPP_NICKNAME.strip() or self._muc_nick
        return nick == expected or nick == base

    def _on_muc_presence(self, presence):
        room = presence['from'].bare
        nick = presence['from'].resource
        ptype = presence['type']
        if self._is_my_nick(room, nick) and ptype not in ('unavailable', 'error'):
            self._muc_ready.add(room)
            # Reset rejoin counter on successful join (anti-ban: avoid accumulating backoff)
            self._muc_rejoin_attempts.pop(room, None)
            self._muc_rejoin_attempts.pop("_nick_" + room, None)
        if ptype == 'unavailable':
            print(f'[{_ts()}] MUC [{room}] <{nick}> left', flush=True)
            # Anti-ban: remove from ready set on unavailable to keep state consistent
            self._muc_ready.discard(room)
            # Anti-ban: trigger auto-rejoin with exponential backoff
            if self._is_my_nick(room, nick):
                self._schedule_muc_rejoin(room)
        elif ptype == 'error':
            print(f'[{_ts()}] MUC [{room}] error: {presence}', flush=True)
            # Anti-ban: also rejoin on error (e.g. temporary failure)
            if self._is_my_nick(room, nick):
                self._muc_ready.discard(room)
                self._schedule_muc_rejoin(room)
        else:
            print(f'[{_ts()}] MUC [{room}] <{nick}> joined (type={ptype})', flush=True)

    async def _process_dm_async(self, jid, body):
        """Anti-ban: versi async dari _process_dm, dipanggil dari event loop."""
        session = self._session_mgr.get_or_create(
            jid, self._build_system_prompt(
                tools_definition=self._tools_def,
                character=config.AGENT_CHARACTER or None,
                skills=config.AGENT_SKILLS.split(",") if config.AGENT_SKILLS else None,
            )
        )
        session.cancel_timer()

        # Anti-ban: JANGAN kirim presence subscription otomatis
        # Ini adalah red flag untuk server (contact farming behavior)
        # Hanya kirim jika user explicitly request
        # self.send_presence_subscription(pto=jid, ptype='subscribed')

        if body == ':new':
            self._session_mgr.reset(jid)
            print(f'[{_ts()}] Session reset for {jid}', flush=True)
            self._schedule_send(jid, 'Memulai sesi baru. Ada yang bisa di bantu?')
            return

        if 'roleplayer' in self._skill:
            try:
                results = ragroleplay.user_load(config.ragroleplay_db_path, unique_id=jid, character=personality.PERSONALITY.name)
                if results:
                    u = results[0]
                    context = (
                        f'[User Context]\n'
                        f'ID: {u["id"]}\n'
                        f'Nama: {u["fullname"]} ({u["nickname"]})\n'
                        f'Family: {u.get("familyname", "-") or "-"}\n'
                        f'Character: {u.get("character", "-") or "-"}\n'
                        f'Alias: {u.get("alias", "-") or "-"}\n'
                        f'Salutation: {u.get("salutation", "-") or "-"}\n'
                        f'Persona: {u.get("persona", "-") or "-"}\n'
                        f'Telegram: {u.get("telegram_id", "-") or "-"} / @{u.get("telegram_username", "-") or "-"}\n'
                        f'XMPP: {u.get("xmpp_username", "-") or "-"}\n'
                        f'[/User Context]\n'
                        f'[PENTING: Kamu SUDAH mengenal user ini. WAJIB panggil memories_latest(character="{personality.PERSONALITY.name}", user_id="{u["id"]}", limit=10) untuk mengambil riwayat percakapan terbaru. GUNAKAN data kondisi emotional/physical dari memori terbaru untuk melanjutkan state character secara natural sebelum merespon.]'
                    )
                    session.add_message('system', context)
                else:
                    session.add_message('system',
                        f'[User Context: Pengguna baru — belum ada di database]\n'
                        f'[PENTING: Kamu BELUM mengenal user ini. WAJIB tanya nama sebagai pembuka. '
                        f'Setelah nama diketahui, simpan via users_store. '
                        f'Lanjutkan percakapan natural dan proaktif melengkapi data user lainnya di pesan berikutnya.]\n'
                        f'[Platform: XMPP ({jid})]')
            except Exception:
                pass

        session.add_message('user', body)

        is_roleplay = 'roleplayer' in self._skill
        if not is_roleplay:
            self._schedule_send(jid, f'> {body}\nThinking...')

        # Delay 1: simulasi membaca pesan user
        await _read_delay()

        my_name = personality.PERSONALITY.name
        quote = body

        def on_tool_calls(content):
            if content and content.strip():
                self._schedule_send(jid, content, 'chat')

        tool_reminder = None

        final_content, should_close = run_agent_loop(
            session, self._llm, self._TOOLS, self._TOOL_HANDLERS,
            self._max_iterations, on_tool_calls=on_tool_calls, tool_reminder=tool_reminder
        )

        if final_content is not None:
            if is_roleplay:
                if config.XMPP_SELECTIVE_RESPONSE:
                    recent_msgs = []
                    for msg in session.messages[-6:]:
                        if msg.get('role') == 'user':
                            recent_msgs.append(f"User: {msg.get('content', '')}")
                        elif msg.get('role') == 'assistant' and msg.get('content'):
                            recent_msgs.append(f"{my_name}: {msg.get('content', '')}")
                    recent_history = "\n".join(recent_msgs)

                    if should_respond(
                        message=quote,
                        sender_nickname=jid,
                        recent_history=recent_history,
                        my_name=my_name,
                    ):
                        print(f'[{_ts()}] need_response=True → sending response', flush=True)
                        self._schedule_send(jid, final_content, 'chat')
                    else:
                        print(f'[{_ts()}] need_response=False → staying silent', flush=True)
                else:
                    from tools.roleplayer import _name_mentioned
                    if _name_mentioned(my_name, quote):
                        print(f'[{_ts()}] Name mentioned → sending response', flush=True)
                        self._schedule_send(jid, final_content, 'chat')
                    else:
                        print(f'[{_ts()}] Name not mentioned → staying silent', flush=True)
            else:
                self._schedule_send(jid, f'> {quote}\n{final_content}', 'chat')
        else:
            msg = 'Max iterations reached without final answer.'
            if is_roleplay:
                self._schedule_send(jid, msg, 'chat')
            else:
                self._schedule_send(jid, f'> {quote}\n{msg}', 'chat')

        # Natural close: DM only, roleplayer only
        if should_close and 'roleplayer' in self._skill:
            print(f'[{_ts()}] Natural close triggered for {jid}', flush=True)
            self._session_mgr.reset(jid)
        else:
            # DM: timeout 24 jam (efektif tidak auto-close), MUC tetap 5 menit
            session.start_timer(86400, self._timeout_session, jid, 'chat')

    async def _process_muc_async(self, room, nick, body):
        """Anti-ban: versi async dari _process_muc, dipanggil dari event loop."""
        session = self._session_mgr.get_or_create(
            room, self._build_system_prompt(
                tools_definition=self._tools_def,
                character=config.AGENT_CHARACTER or None,
                skills=config.AGENT_SKILLS.split(",") if config.AGENT_SKILLS else None,
            )
        )
        session.cancel_timer()

        if body == ':new':
            self._session_mgr.reset(room)
            print(f'[{_ts()}] Session reset for MUC room {room}', flush=True)
            self._schedule_send(room, 'Memulai sesi baru. Ada yang bisa di bantu?', mtype='groupchat')
            return

        prefixed = f'[{nick}] {body}'
        session.add_message('user', prefixed)

        if self._skill != 'roleplayer':
            self._schedule_send(room, f'> [{nick}] {body}\nThinking...', mtype='groupchat')

        # Delay 1: simulasi membaca pesan user
        await _read_delay()

        my_name = personality.PERSONALITY.name
        quote = f'[{nick}] {body}'

        _is_roleplay = 'roleplayer' in self._skill

        def on_tool_calls(content):
            if content and content.strip():
                self._schedule_send(room, content, 'groupchat')

        tool_reminder = None

        final_content, _should_close = run_agent_loop(
            session, self._llm, self._TOOLS, self._TOOL_HANDLERS,
            self._max_iterations, on_tool_calls=on_tool_calls, tool_reminder=tool_reminder
        )

        if final_content is not None:
            if _is_roleplay:
                if config.XMPP_SELECTIVE_RESPONSE:
                    recent_msgs = []
                    for msg in session.messages[-6:]:
                        if msg.get('role') == 'user':
                            recent_msgs.append(f"User: {msg.get('content', '')}")
                        elif msg.get('role') == 'assistant' and msg.get('content'):
                            recent_msgs.append(f"{my_name}: {msg.get('content', '')}")
                    recent_history = "\n".join(recent_msgs)

                    if should_respond(
                        message=quote,
                        sender_nickname=nick,
                        recent_history=recent_history,
                        my_name=my_name,
                    ):
                        print(f'[{_ts()}] need_response=True → sending response', flush=True)
                        self._schedule_send(room, final_content, 'groupchat')
                    else:
                        print(f'[{_ts()}] need_response=False → staying silent', flush=True)
                else:
                    from tools.roleplayer import _name_mentioned
                    if _name_mentioned(my_name, quote):
                        print(f'[{_ts()}] Name mentioned → sending response', flush=True)
                        self._schedule_send(room, final_content, 'groupchat')
                    else:
                        print(f'[{_ts()}] Name not mentioned → staying silent', flush=True)
            else:
                self._schedule_send(room, f'> {quote}\n{final_content}', 'groupchat')
        else:
            msg = 'Max iterations reached without final answer.'
            if _is_roleplay:
                self._schedule_send(room, msg, 'groupchat')
            else:
                self._schedule_send(room, f'> {quote}\n{msg}', 'groupchat')

        session.start_timer(300, self._timeout_session, room, 'groupchat')

    def _execute_tool(self, tool_call):
        from lib.agent_loop import execute_tool
        return execute_tool(tool_call, self._TOOL_HANDLERS)

    def _schedule_send(self, to, body, mtype='chat'):
        """Anti-ban: enqueue pesan ke queue, bukan langsung kirim."""
        if self._msg_queue and self._loop and not self._loop.is_closed():
            try:
                self._msg_queue.put_nowait((to, body, mtype))
                print(f'[{_ts()}] Queued message to {to} ({len(body)} chars)', flush=True)
            except asyncio.QueueFull:
                print(f'[{_ts()}] WARNING: Message queue full, dropping message to {to}', flush=True)
        else:
            print(f'[{_ts()}] WARNING: cannot queue message to {to} — queue unavailable', flush=True)

    async def _msg_worker(self):
        """Anti-ban: worker coroutine yang mengirim pesan dari queue dengan rate limiting."""
        print(f'[{_ts()}] Message queue worker started', flush=True)
        self._msg_worker_running = True
        
        while self._msg_worker_running:
            try:
                # Ambil pesan dari queue dengan timeout
                try:
                    to, body, mtype = await asyncio.wait_for(
                        self._msg_queue.get(), timeout=1.0
                    )
                except asyncio.TimeoutError:
                    continue
                
                # Adaptive: apply backoff multiplier jika throttle detected
                base_delay = random.uniform(MSG_QUEUE_DELAY_MIN, MSG_QUEUE_DELAY_MAX)
                multiplier = self._rate_limiter.get_delay_multiplier()
                delay = _add_jitter(base_delay) * multiplier
                
                if multiplier > 1.0:
                    print(f'[{_ts()}] ADAPTIVE: Throttled, delay {delay:.1f}s (x{multiplier}) to {to}', flush=True)
                else:
                    print(f'[{_ts()}] Pre-send delay: {delay:.1f}s to {to}', flush=True)
                
                await asyncio.sleep(delay)
                
                # Kirim pesan dan catat hasilnya
                send_start = asyncio.get_event_loop().time()
                send_success = False
                
                try:
                    # Anti-ban: gunakan full JID untuk mfrom (server strict tentang ini)
                    mfrom = self.boundjid.full if self.boundjid else None

                    msg = self.make_message(mto=to, mbody=body, mtype=mtype, mfrom=mfrom)

                    # Debug: snapshot state koneksi + stanza XML lengkap sebelum kirim.
                    _dbg(f'SEND STATE: to={to}, mtype={mtype}, '
                         f'_session_started={getattr(self, "_session_started", None)}, '
                         f'authenticated={self.authenticated}, bound={self.bound}')
                    _dbg(f'Stanza to send (full): {msg}')

                    msg.send()
                    send_success = True
                    print(f'[{_ts()}] Sent to {to} ({len(body)} chars)', flush=True)
                except Exception as e:
                    print(f'[{_ts()}] SEND ERROR to {to}: {e}', flush=True)
                    print(traceback.format_exc(), flush=True)
                    # Catat error untuk adaptive rate limiting
                    self._rate_limiter.record_send(success=False)
                
                if send_success:
                    # Catat sukses (tanpa response time karena XMPP async)
                    self._rate_limiter.record_send(success=True)
                
                # Mark task done
                self._msg_queue.task_done()
                
                # Small delay antara pesan untuk menghindari burst
                # Adaptive: lebih lama jika throttled
                inter_delay = random.uniform(0.5, 1.5) * multiplier
                await asyncio.sleep(inter_delay)
                
            except asyncio.CancelledError:
                print(f'[{_ts()}] Message queue worker cancelled', flush=True)
                break
            except Exception as e:
                print(f'[{_ts()}] Message queue worker error: {e}', flush=True)
                await asyncio.sleep(1)
        
        self._msg_worker_running = False
        print(f'[{_ts()}] Message queue worker stopped', flush=True)

    async def _send_coro(self, to, body, mtype):
        try:
            # Delay 2: simulasi mengetik (proporsional dengan panjang pesan)
            delay = _typing_delay(body)
            print(f'[{_ts()}] Typing delay: {delay:.1f}s ({len(body)} chars)', flush=True)
            await asyncio.sleep(delay)

            # Anti-ban: gunakan full JID untuk mfrom (server strict tentang ini)
            mfrom = self.boundjid.full if self.boundjid else None
            msg = self.make_message(mto=to, mbody=body, mtype=mtype, mfrom=mfrom)
            msg.send()
        except Exception as e:
            print(f'[{_ts()}] SEND ERROR: {e}', flush=True)

    def _timeout_session(self, session_id, mtype):
        print(f'[{_ts()}] Session timeout: {session_id}', flush=True)
        self._schedule_send(session_id, 'Sesi ditutup. Sampai jumpa', mtype)
        self._session_mgr.reset(session_id)

    def start(self):
        print(f'[{_ts()}] Starting XMPP service...', flush=True)
        _setup_debug_logging()
        asyncio.run(self._run())

    async def _run(self):
        self._stopped = asyncio.Event()
        self._loop = asyncio.get_running_loop()

        # Anti-ban: inisialisasi message queue dan worker
        self._msg_queue = asyncio.Queue(maxsize=100)
        self._msg_worker_task = asyncio.create_task(self._msg_worker())

        # Hanya tangani SIGTERM untuk shutdown.
        # SENGATKAN SIGHUP: nohup kirim SIGHUP saat terminal close,
        # dan kita tidak mau proses mati karena itu.
        try:
            self._loop.add_signal_handler(signal.SIGTERM, self._stopped.set)
        except (NotImplementedError, RuntimeError):
            pass

        print(f'[{_ts()}] Connecting to server (jid={self.jid})...', flush=True)
        try:
            await self.connect()
        except Exception as e:
            print(f'[{_ts()}] CONNECT ERROR: {e}', flush=True)
            print(traceback.format_exc(), flush=True)
            raise
        print(f'[{_ts()}] Connected, waiting for stream events...', flush=True)
        try:
            await self._stopped.wait()
        except (asyncio.CancelledError, KeyboardInterrupt):
            pass
        print(f'[{_ts()}] Shutting down...', flush=True)
        await self.disconnect()

    def stop(self):
        if self._loop and not self._loop.is_closed():
            asyncio.run_coroutine_threadsafe(self._async_stop(), self._loop)

    async def _async_stop(self):
        # Anti-ban: stop message queue worker dulu
        if self._msg_worker_task and not self._msg_worker_task.done():
            self._msg_worker_running = False
            self._msg_worker_task.cancel()
            try:
                await self._msg_worker_task
            except asyncio.CancelledError:
                pass
        
        # Cancel semua pending rejoin tasks
        for room, task in list(self._muc_rejoin_tasks.items()):
            if not task.done():
                task.cancel()
        self._muc_rejoin_tasks.clear()
        
        self._stopped.set()