928 lines
44 KiB
Python
928 lines
44 KiB
Python
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
|
|
# Thinking message removed for natural feel
|
|
|
|
# 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)
|
|
|
|
# Thinking message removed for natural feel
|
|
_is_roleplay = 'roleplayer' in self._skill
|
|
|
|
# Delay 1: simulasi membaca pesan user
|
|
await _read_delay()
|
|
|
|
my_name = personality.PERSONALITY.name
|
|
quote = f'[{nick}] {body}'
|
|
|
|
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()
|