arise.py

import asyncio
import json
import logging
import os
import random
import re
import sys
import time
from datetime import datetime, timedelta, timezone
from typing import Any, Dict, List, Optional, Union

import certifi
from watchdog.events import FileSystemEventHandler
from watchdog.observers import Observer

from interfaces.arise_interface import AriseInterface
from interfaces.autonomous import AutonomousInterface
from modules.memory_manager import MemoryManager
from providers.arise_provider import (
    AriseProvider,
    ModelSpecificParams,
    LLMResponse,
    ToolCall,
)
from providers.bigmodel import BigModelProvider
from providers.mistral import MistralProvider
from subagents.conversation_scheduler import ConversationScheduler
from subagents.idle import IdleSubAgent
from tools.definitions import TOOLS, get_tools_description
from tools.executor import ToolExecutor, ToolResult
from utils.config import Config
from utils.i18n import get_string
from utils.prompt import get_prompt
from utils.user_mapping import UserMapping

# Fix SSL certificate verification on macOS
os.environ["SSL_CERT_FILE"] = certifi.where()


# Configure logging
is_verbose = Config.VERBOSE_LOG
handlers = [logging.StreamHandler(sys.stdout)]

if Config.SAVE_LOGS:
    from logging.handlers import TimedRotatingFileHandler

    # Ensure data directory exists
    log_dir = os.path.join(os.getcwd(), "data")
    os.makedirs(log_dir, exist_ok=True)
    log_file = os.path.join(log_dir, "latest.log")

    # Daily rotation, keep 7 days for safety, though "window of 1 day" might imply we don't care about old ones.
    # Requirement: "save a rolling window log of length 1 day".
    # This implies we want to see the last 24h. TimedRotatingFileHandler rotates at intervals.
    # If we rotate every 'D' (day), we get daily logs.
    file_handler = TimedRotatingFileHandler(
        log_file, when="D", interval=1, backupCount=1, encoding="utf-8"
    )
    handlers.append(file_handler)

logging.basicConfig(
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
    level=logging.DEBUG if is_verbose else logging.INFO,
    handlers=handlers,
    force=True,  # Force reconfiguration
)

# Reduce noise from underlying libraries even in verbose mode
logging.getLogger("httpcore").setLevel(logging.WARNING)
logging.getLogger("httpx").setLevel(logging.WARNING)
logging.getLogger("telegram").setLevel(logging.INFO)
logging.getLogger("watchdog").setLevel(logging.WARNING)
logging.getLogger("fsevents").setLevel(logging.WARNING)

logger = logging.getLogger("ARISE")


class HotReloadHandler(FileSystemEventHandler):
    def __init__(self, debounce_delay=2.0):
        self.debounce_delay = debounce_delay
        self.last_reload = 0

    def on_modified(self, event):
        # Ignore .git directory changes
        src_path = event.src_path
        if isinstance(src_path, (bytes, bytearray)):
            src_path = src_path.decode("utf-8")
        elif isinstance(src_path, memoryview):
            src_path = bytes(src_path).decode("utf-8")
        else:
            src_path = str(src_path)
        if "/.git/" in src_path or src_path.endswith("/.git"):
            return
        if "arise.db" in src_path or "__pycache__" in src_path:
            return
        if (
            src_path.endswith(".py") or os.path.basename(src_path) == ".env"
        ) and time.time() - self.last_reload > self.debounce_delay:
            logger.info(f"Detected change in {src_path}, restarting...")
            self.last_reload = time.time()
            # 保存状态如果需要（这里简化，实际可序列化状态）
            os.execv(sys.executable, [sys.executable] + sys.argv)


class Arise:
    def __init__(self):
        # Configuration attributes from Config
        self.verbose = Config.VERBOSE_LOG
        self.debounce_delay = Config.DEBOUNCE_DELAY
        self.start_time = time.time()

        self.providers: Dict[str, AriseProvider] = {}
        self.interfaces: List[AriseInterface] = []
        self.modules = []
        self.subagents = []

        self.user_mapping = UserMapping()
        self.memory_manager = MemoryManager()

        self.user_states: Dict[str, Dict[str, Any]] = {}
        self.context: List[Dict[str, Any]] = []
        self.main_provider: AriseProvider | None = None

        # Debounce management
        self.pending_messages: Dict[str, List[Dict[str, Any]]] = {}
        self.debounce_tasks: Dict[str, asyncio.Task] = {}
        self._background_tasks = set()

    def _create_background_task(self, coro):
        """Helper to track background tasks for clean shutdown."""
        task = asyncio.create_task(coro)
        self._background_tasks.add(task)
        task.add_done_callback(self._background_tasks.discard)
        return task

    def load_components(self):
        logger.info("Loading ARISE components...")
        self.modules.append(self.memory_manager)
        if self.verbose:
            logger.debug("VERBOSE LOGGING ENABLED")

        from interfaces.telegram import TelegramInterface

        # Initialize Providers
        bigmodel = BigModelProvider()
        self.providers["bigmodel"] = bigmodel

        mistral = MistralProvider()
        self.providers["mistral"] = mistral

        # Set default provider
        self.main_provider = self.providers.get(Config.DEFAULT_PROVIDER, mistral)

        logger.info(f"Default provider set to: {Config.DEFAULT_PROVIDER}")

        # Inject provider into memory manager for embeddings
        self.memory_manager.set_provider(self.main_provider)

        # Initialize Interfaces
        # Initialize Telegram Interface if enabled and token is configured
        if Config.TELEGRAM_ENABLED and Config.TELEGRAM_BOT_TOKEN:
            telegram = TelegramInterface()
            telegram.set_callback(self.handle_message)
            telegram.set_python_approval_callback(self.handle_python_approval)
            self.interfaces.append(telegram)
            logger.info("Telegram interface initialized.")
        elif Config.TELEGRAM_ENABLED and not Config.TELEGRAM_BOT_TOKEN:
            logger.warning("Telegram enabled but TELEGRAM_BOT_TOKEN not set. Skipping.")

        # Initialize Discord Interface if enabled and token is configured
        if Config.DISCORD_ENABLED and Config.DISCORD_BOT_TOKEN:
            from interfaces.discord import DiscordInterface

            discord_intf = DiscordInterface()
            discord_intf.set_callback(self.handle_message)
            discord_intf.set_python_approval_callback(self.handle_python_approval)
            self.interfaces.append(discord_intf)
            logger.info("Discord interface initialized.")
        elif Config.DISCORD_ENABLED and not Config.DISCORD_BOT_TOKEN:
            logger.warning("Discord enabled but DISCORD_BOT_TOKEN not set. Skipping.")

        # Initialize WeChat Interface if enabled
        if Config.WECHAT_ENABLED:
            from interfaces.wechat import WeChatInterface

            wechat_intf = WeChatInterface()
            wechat_intf.set_callback(self.handle_message)
            wechat_intf.set_python_approval_callback(self.handle_python_approval)
            self.interfaces.append(wechat_intf)
            logger.info("WeChat interface initialized.")

        self.tool_executor = ToolExecutor(
            memory_manager=self.memory_manager,
            user_mapping=self.user_mapping,
            python_approval_callback=self._python_approval_request,
        )

        self.idle_agent = IdleSubAgent()
        self.subagents.append(self.idle_agent)

        self.autonomous_interface = AutonomousInterface(
            user_mapping=self.user_mapping,
            distill_callback=None,
        )
        self.autonomous_interface.set_callback(self.handle_message)
        self.interfaces.append(self.autonomous_interface)

        self.conversation_scheduler = ConversationScheduler(
            user_mapping=self.user_mapping, trigger_callback=self._schedule_trigger
        )
        self.subagents.append(self.conversation_scheduler)

    async def _schedule_trigger(
        self, sender_id: str, metadata: Any, result_content: str, **kwargs
    ):
        logger.info(f"Schedule triggered response for {sender_id}")

        if "interface" not in metadata and "interface_name" in metadata:
            target_name = metadata["interface_name"]
            for intf in self.interfaces:
                if (
                    "telegram" in target_name.lower()
                    and "TelegramInterface" in intf.__class__.__name__
                ):
                    metadata["interface"] = intf
                    break
                elif (
                    "discord" in target_name.lower()
                    and "DiscordInterface" in intf.__class__.__name__
                ):
                    metadata["interface"] = intf
                    break

        await self._process_message_logic(
            sender_id,
            f"[--- SCHEDULED TASK ---]\n{result_content}",
            [],
            metadata,
            **kwargs,
        )

    async def _python_approval_request(
        self,
        sender_id: str,
        metadata: Any,
        code: str,
        approval_id: str,
        purpose: str = "",
    ):
        interface = metadata.get("interface")
        lang = metadata.get("lang", "zh")

        code_preview = code[:500] + "..." if len(code) > 500 else code

        if interface and hasattr(interface, "send_message"):
            approve_msg = get_string("python_approval_regular", lang).format(
                code_preview=code_preview
            )

            buttons = [
                [
                    {
                        "text": get_string("python_approve_btn", lang),
                        "callback_data": f"python_approve:{approval_id}",
                    },
                    {
                        "text": get_string("python_reject_btn", lang),
                        "callback_data": f"python_reject:{approval_id}",
                    },
                ]
            ]

            interface.send_message(sender_id, approve_msg, buttons=buttons)
            logger.info(
                f"Python code approval buttons sent to user {sender_id} (ID: {approval_id})"
            )

    async def handle_python_approval(
        self, sender_id: str, approval_id: str, approved: bool, metadata: Any
    ):
        interface = metadata.get("interface")
        lang = metadata.get("lang", "zh")

        if approved:
            logger.info(f"Python code approved by user {sender_id} (ID: {approval_id})")
            result = await self.tool_executor.execute_approved_python(approval_id)
            if interface:
                interface.send_message(sender_id, result)
        else:
            logger.info(f"Python code rejected by user {sender_id} (ID: {approval_id})")
            result = await self.tool_executor.reject_python(approval_id)
            if interface:
                interface.send_message(sender_id, result)

    async def _send_qrcode_to_superuser(
        self,
        interface: Any,
        superuser_id: str,
        qrcode_data: Dict[str, Any],
        wechat_interface: Any,
    ):
        """Send QR code to superuser and wait for login."""
        import qrcode as qr_module
        from io import BytesIO

        qrcode_url = qrcode_data["qrcode_url"]
        qrcode_id = qrcode_data["qrcode_id"]

        try:
            qr = qr_module.QRCode(
                version=1,
                error_correction=qr_module.constants.ERROR_CORRECT_L,
                box_size=10,
                border=2,
            )
            qr.add_data(qrcode_url)
            qr.make(fit=True)

            img = qr.make_image(fill_color="black", back_color="white")
            bio = BytesIO()
            img.save(bio, format="PNG")
            image_data = bio.getvalue()

            # Try to send image directly via WeChat
            if hasattr(wechat_interface, "send_image"):
                sent = await wechat_interface.send_image(superuser_id, image_data)
                if sent:
                    interface.send_message(
                        superuser_id,
                        "⬆️ 请用微信扫描上方二维码登录新账号\n"
                        "扫描完成后系统将自动检测登录状态",
                        save_to_history=Config.INCLUDE_COMMAND_CONTEXT,
                    )
                else:
                    # Fallback to text
                    interface.send_message(
                        superuser_id,
                        f"请用微信扫描以下二维码登录新账号：\n\n{qrcode_url}\n\n"
                        "扫描完成后系统将自动检测登录状态。",
                        save_to_history=Config.INCLUDE_COMMAND_CONTEXT,
                    )
            else:
                interface.send_message(
                    superuser_id,
                    f"请用微信扫描以下二维码登录新账号：\n\n{qrcode_url}\n\n"
                    "扫描完成后系统将自动检测登录状态。",
                    save_to_history=Config.INCLUDE_COMMAND_CONTEXT,
                )

            logger.info(
                f"QR code sent to superuser {superuser_id}, waiting for scan..."
            )

            async def _wait_for_login():
                account_id = await wechat_interface.wait_for_qrcode_login(qrcode_id)
                if account_id:
                    interface.send_message(
                        superuser_id,
                        f"✅ 新微信账号登录成功！\n账号ID: {account_id}\n\n"
                        "新用户现在可以开始对话了。",
                        save_to_history=Config.INCLUDE_COMMAND_CONTEXT,
                    )
                else:
                    interface.send_message(
                        superuser_id,
                        "❌ 微信登录超时或失败，请重试 /new_wechat",
                        save_to_history=Config.INCLUDE_COMMAND_CONTEXT,
                    )

            self._create_background_task(_wait_for_login())

        except ImportError:
            interface.send_message(
                superuser_id,
                f"请扫描以下二维码登录新微信账号：\n\n{qrcode_url}\n\n"
                "扫描完成后系统将自动检测登录状态。",
                save_to_history=Config.INCLUDE_COMMAND_CONTEXT,
            )

            async def _wait_for_login():
                account_id = await wechat_interface.wait_for_qrcode_login(
                    qrcode_data["qrcode_id"]
                )
                if account_id:
                    interface.send_message(
                        superuser_id,
                        f"✅ 新微信账号登录成功！账号ID: {account_id}",
                        save_to_history=Config.INCLUDE_COMMAND_CONTEXT,
                    )
                else:
                    interface.send_message(
                        superuser_id,
                        "❌ 微信登录超时或失败，请重试 /new_wechat",
                        save_to_history=Config.INCLUDE_COMMAND_CONTEXT,
                    )

            self._create_background_task(_wait_for_login())
        except Exception as e:
            logger.error(f"Failed to send QR code: {e}")
            interface.send_message(
                superuser_id,
                f"生成二维码时出错：{str(e)}",
                save_to_history=Config.INCLUDE_COMMAND_CONTEXT,
            )

    async def handle_message(self, sender_id: str, text: str, metadata: Any):
        """
        Entry point for messages from interfaces. Implements debouncing.
        """
        is_command = text.strip().startswith("/")

        # --- MUTE LOGIC START ---
        # Allow /mute command to pass through even if muted
        if text.strip() != "/mute" and self.user_states.get(sender_id, {}).get(
            "mute_mode", False
        ):
            logger.info(f"User {sender_id} is MUTED. Accumulating message.")
            if "accumulated_messages" not in self.user_states[sender_id]:
                self.user_states[sender_id]["accumulated_messages"] = []

            self.user_states[sender_id]["accumulated_messages"].append(
                {"text": text, "metadata": metadata, "timestamp": time.time()}
            )
            return
        # --- MUTE LOGIC END ---

        # Notify autonomous interface of activity only for non-command messages
        if hasattr(self, "autonomous_interface") and not is_command:
            self.autonomous_interface.notify_activity()

        # 1. Process commands immediately (e.g. /new, /start, /su)
        is_in_state = self.user_states.get(sender_id, {}).get("state")

        if is_command or is_in_state:
            logger.info(
                f"Processing {'command' if is_command else 'state-based message'} immediately for {sender_id}: {text}"
            )
            await self._process_message_logic(sender_id, text, [], metadata)
            return

        logger.info(f"Queuing message from {sender_id}: {text}")

        # Check if interface is currently sending a reply - if so, interrupt it
        # This mimics human behavior when interrupted: keep what was said, discard the rest
        interface = metadata.get("interface")
        if (
            interface
            and hasattr(interface, "is_sending")
            and hasattr(interface, "interrupt_sending")
        ):
            if interface.is_sending(sender_id):
                logger.info(
                    f"User {sender_id} sent new message while AI is replying. Interrupting current reply..."
                )
                interface.interrupt_sending(sender_id)

        if sender_id not in self.pending_messages:
            self.pending_messages[sender_id] = []

        self.pending_messages[sender_id].append({"text": text, "metadata": metadata})

        if sender_id in self.debounce_tasks:
            self.debounce_tasks[sender_id].cancel()

        self.debounce_tasks[sender_id] = self._create_background_task(
            self._process_debounced_messages(sender_id)
        )

    async def _process_debounced_messages(self, sender_id: str):
        try:
            # Add randomness to debounce: +/- 25% jitter
            jitter = self.debounce_delay * 0.25
            actual_delay = self.debounce_delay + random.uniform(-jitter, jitter)
            await asyncio.sleep(max(0, actual_delay))
        except asyncio.CancelledError:
            return

        messages = self.pending_messages.pop(sender_id, [])
        self.debounce_tasks.pop(sender_id, None)

        if not messages:
            return

        # Combine text and images
        combined_text = "\n".join([m["text"] for m in messages if m["text"]])
        all_images = []
        for m in messages:
            msg_images = m["metadata"].get("images", [])
            all_images.extend(msg_images)

        # Use metadata from the last message for context (lang, interface, etc.)
        last_metadata = messages[-1]["metadata"]

        await self._process_message_logic(
            sender_id,
            combined_text,
            all_images,
            last_metadata,
            original_messages=messages,
        )

    async def _process_message_logic(
        self,
        sender_id: str,
        text: str,
        images: List[str],
        metadata: Any,
        original_messages: List[Dict] | None = None,
        **kwargs,
    ):
        logger.info(f"Processing consolidated message for {sender_id}: {text[:50]}...")
        if self.verbose:
            logger.debug(f"FULL MESSAGE CONTENT FROM {sender_id}: {text}")
            logger.debug(f"METADATA: {metadata}")

        interface = metadata.get("interface")
        interface_name = metadata.get("interface_name", "unknown")
        is_autonomous = metadata.get("autonomous", False)

        # Check for SU mode
        if sender_id not in self.user_states:
            self.user_states[sender_id] = {}
        is_su_mode = self.user_states[sender_id].get("su_mode", False)

        # 1. Resolve User UUID (Skip if autonomous)
        user_uuid = None
        if not is_autonomous:
            if is_su_mode:
                user_uuid = "__sudo__"
            else:
                user_uuid = self.user_mapping.get_uuid(interface_name, sender_id)

            if self.verbose:
                logger.debug(
                    f"RESOLVED UUID FOR {sender_id} ({interface_name}): {user_uuid} (SU: {is_su_mode})"
                )
        else:
            logger.info("Handling autonomous exploration trigger.")
            user_uuid = "__autonomous__"  # Placeholder or special id

        # Inject user_uuid into metadata so subagents can access it
        metadata["user_uuid"] = user_uuid

        # Determine language early for command responses
        text_is_zh = bool(re.search(r"[\u4e00-\u9fff]", text))
        lang = "zh" if text_is_zh else metadata.get("lang", "zh")

        is_command = text.strip().startswith("/")

        # Save user command to history if enabled
        if Config.INCLUDE_COMMAND_CONTEXT and is_command:
            # List of handled commands (approximated by root)
            cmd_root = text.strip().split()[0]
            is_handled = False

            # Direct matches or prefixes
            if cmd_root in [
                "/start",
                "/new",
                "/su",
                "/uuid",
                "/status",
                "/create_user",
                "/bind",
                "/memory",
                "/schedule",
                "/profile",
                "/mute",
                "/new_wechat",
            ]:
                is_handled = True
            elif cmd_root.startswith("/memory_btn_") or cmd_root.startswith(
                "/schedule_"
            ):
                is_handled = True

            if is_handled and interface and hasattr(interface, "save_history"):
                u_msg = {
                    "role": "user",
                    "content": text,
                    "sender_id": sender_id,
                    "interface": interface_name,
                    "lang": lang,
                    "timestamp": datetime.now().isoformat(),
                }
                interface.save_history(sender_id, u_msg)

        if text.strip() == "/start":
            if user_uuid:
                interface.send_message(
                    sender_id,
                    get_string("welcome", lang),
                    save_to_history=Config.INCLUDE_COMMAND_CONTEXT,
                )
            else:
                interface.send_message(
                    sender_id,
                    get_string("welcome_new", lang),
                    save_to_history=Config.INCLUDE_COMMAND_CONTEXT,
                )
            return

        # Handle explicit commands for binding/creating
        if text.startswith("/bind "):
            target_uuid = text.split(" ")[1].strip()
            history_path = (
                interface.get_history_path(sender_id)
                if hasattr(interface, "get_history_path")
                else ""
            )
            if self.user_mapping.bind_user(
                target_uuid, interface_name, sender_id, history_path
            ):
                self.user_mapping.update_paths(
                    target_uuid,
                    interface_name,
                    history_path,
                    self.memory_manager.db.db_path,
                )
                interface.send_message(
                    sender_id,
                    get_string("bind_success", lang),
                    save_to_history=Config.INCLUDE_COMMAND_CONTEXT,
                )
            else:
                interface.send_message(
                    sender_id,
                    get_string("bind_fail", lang),
                    save_to_history=Config.INCLUDE_COMMAND_CONTEXT,
                )
            return
        elif text.strip() == "/create_user":
            history_path = (
                interface.get_history_path(sender_id)
                if hasattr(interface, "get_history_path")
                else ""
            )
            new_uuid = self.user_mapping.create_user(
                interface_name, sender_id, history_path, self.memory_manager.db.db_path
            )
            interface.send_message(
                sender_id,
                f"{get_string('new_user_success', lang)}\n`{new_uuid}`",
                save_to_history=Config.INCLUDE_COMMAND_CONTEXT,
            )
            return
        elif text.strip() == "/su":
            if sender_id not in self.user_states:
                self.user_states[sender_id] = {}

            # Since the interface already verified SU status before sending this command,
            # we can just toggle it here.
            current_status = self.user_states[sender_id].get("su_mode", False)
            self.user_states[sender_id]["su_mode"] = not current_status

            msg = get_string("su_on" if not current_status else "su_off", lang)
            interface.send_message(
                sender_id, msg, save_to_history=Config.INCLUDE_COMMAND_CONTEXT
            )
            return
        elif text.strip() == "/uuid":
            if user_uuid:
                interface.send_message(
                    sender_id,
                    f"{get_string('current_uuid', lang)}\n`{user_uuid}`",
                    save_to_history=Config.INCLUDE_COMMAND_CONTEXT,
                )
            else:
                interface.send_message(
                    sender_id,
                    get_string("no_uuid", lang),
                    save_to_history=Config.INCLUDE_COMMAND_CONTEXT,
                )
            return
        elif text.strip() == "/new_wechat":
            # Check if user is superuser (either in SUPERUSERS list or su_mode enabled)
            is_superuser = sender_id in Config.SUPERUSERS or self.user_states.get(
                sender_id, {}
            ).get("su_mode", False)
            if not is_superuser:
                interface.send_message(
                    sender_id,
                    "此命令仅限 superuser 使用 / This command is superuser only",
                    save_to_history=Config.INCLUDE_COMMAND_CONTEXT,
                )
                return

            # Find WeChat interface
            wechat_interface = None
            for iface in self.interfaces:
                if hasattr(iface, "generate_new_qrcode"):
                    wechat_interface = iface
                    break

            if not wechat_interface:
                interface.send_message(
                    sender_id,
                    "WeChat interface not available",
                    save_to_history=Config.INCLUDE_COMMAND_CONTEXT,
                )
                return

            # Generate QR code
            interface.send_message(
                sender_id,
                "正在生成新的微信登录二维码...",
                save_to_history=Config.INCLUDE_COMMAND_CONTEXT,
            )

            async def _generate_and_send_qrcode():
                qrcode_data = await wechat_interface.generate_new_qrcode()

                if qrcode_data:
                    await self._send_qrcode_to_superuser(
                        interface,
                        sender_id,
                        qrcode_data,
                        wechat_interface,
                    )
                else:
                    interface.send_message(
                        sender_id,
                        "生成二维码失败，请检查微信接口状态",
                        save_to_history=Config.INCLUDE_COMMAND_CONTEXT,
                    )

            self._create_background_task(_generate_and_send_qrcode())
            return
        elif text.strip() == "/mute":
            # Toggle mute
            if sender_id not in self.user_states:
                self.user_states[sender_id] = {}

            current_mute = self.user_states[sender_id].get("mute_mode", False)
            new_mute = not current_mute
            self.user_states[sender_id]["mute_mode"] = new_mute

            # Notify scheduler
            if getattr(self, "conversation_scheduler", None):
                self.conversation_scheduler.set_mute_status(sender_id, new_mute)

            state_key = "mute_on" if new_mute else "mute_off"
            msg = get_string(state_key, lang)
            interface.send_message(
                sender_id, msg, save_to_history=Config.INCLUDE_COMMAND_CONTEXT
            )

            if not new_mute:
                # Unmuted: Process accumulated messages
                accumulated = self.user_states[sender_id].pop(
                    "accumulated_messages", []
                )
                if accumulated:
                    logger.info(
                        f"Processing {len(accumulated)} accumulated messages for {sender_id}"
                    )
                    # Combine text
                    combined_text = "\n".join(
                        [m["text"] for m in accumulated if m["text"]]
                    )
                    all_images = []
                    for m in accumulated:
                        all_images.extend(m["metadata"].get("images", []))

                    if combined_text or all_images:
                        # Use metadata from last message
                        last_meta = accumulated[-1]["metadata"]
                        # Trigger processing
                        await self._process_message_logic(
                            sender_id,
                            combined_text,
                            all_images,
                            last_meta,
                            original_messages=accumulated,
                        )
            return
        elif text.strip().startswith("/quit"):
            if (
                not user_uuid
                or user_uuid == "__sudo__"
                or user_uuid == "__autonomous__"
            ):
                interface.send_message(
                    sender_id,
                    get_string("no_uuid", lang),
                    save_to_history=Config.INCLUDE_COMMAND_CONTEXT,
                )
                return

            if "confirm" not in text.lower():
                interface.send_message(
                    sender_id,
                    get_string("quit_confirm", lang),
                    save_to_history=Config.INCLUDE_COMMAND_CONTEXT,
                )
                return

            # Execute thorough deletion
            logger.info(f"User {sender_id} (UUID: {user_uuid}) requested DATA WIPE.")

            # 1. Memory
            self.memory_manager.delete_user_data(user_uuid)

            # 2. Schedules
            if hasattr(self, "conversation_scheduler"):
                self.conversation_scheduler.delete_user_schedules(user_uuid)
                # Also clear mute status in scheduler memory
                if hasattr(self.conversation_scheduler, "set_mute_status"):
                    self.conversation_scheduler.set_mute_status(sender_id, False)

            # 3. Chat History (Messages Table)
            self.memory_manager.db.execute_update(
                "DELETE FROM messages WHERE user_uuid = ?", (user_uuid,)
            )

            # 4. Physical files (Legacy support)
            history_path = (
                interface.get_history_path(sender_id)
                if hasattr(interface, "get_history_path")
                else ""
            )
            if history_path and os.path.exists(history_path):
                try:
                    os.remove(history_path)
                    if os.path.exists(history_path + ".bak"):
                        os.remove(history_path + ".bak")
                except:
                    pass

            # 5. User Profile and Mappings
            self.user_mapping.delete_user(user_uuid)

            # 6. Clear local state
            if sender_id in self.user_states:
                del self.user_states[sender_id]
            if sender_id in self.pending_messages:
                del self.pending_messages[sender_id]

            interface.send_message(
                sender_id, get_string("quit_success", lang), save_to_history=False
            )
            return
        elif text.strip() == "/status":
            uptime = int(time.time() - self.start_time)
            modules_list = ", ".join([m.__class__.__name__ for m in self.modules])
            subagents_list = ", ".join([s.name for s in self.subagents])
            interfaces_list = ", ".join([i.__class__.__name__ for i in self.interfaces])

            idle_msg = "N/A"
            if hasattr(self, "autonomous_interface"):
                idle_info = self.autonomous_interface.get_status()
                next_exp = idle_info.get("next_exploration_in", -1)
                if next_exp >= 0:
                    idle_msg = str(timedelta(seconds=next_exp))
                else:
                    idle_msg = "Disabled"

            status_msg = (
                f"ARISE System Status\n"
                f"-------------------\n"
                f"Uptime: {timedelta(seconds=uptime)}\n"
                f"Modules: {modules_list}\n"
                f"Subagents: {subagents_list}\n"
                f"Interfaces: {interfaces_list}\n"
                f"Idle Countdown: {idle_msg}\n"
                f"User UUID: {user_uuid}\n"
                f"SU Mode: {'ON' if is_su_mode else 'OFF'}\n\n"
                f"Registered Commands:\n"
                f"/start, /new, /status, /su, /uuid, /bind, /create_user\n"
                f"/memory, /schedule, /profile, /mute"
            )
            interface.send_message(
                sender_id, status_msg, save_to_history=Config.INCLUDE_COMMAND_CONTEXT
            )
            return
        elif text.startswith("/memory"):
            if not user_uuid:
                interface.send_message(
                    sender_id,
                    get_string("no_uuid", lang),
                    save_to_history=Config.INCLUDE_COMMAND_CONTEXT,
                )
                return

            if is_su_mode:
                # Default to core if no specific space is selected, or if we just typed /memory
                # If command is the base /memory (not a button action), reset space to core
                if text.strip() == "/memory":
                    self.user_states[sender_id]["memory_space"] = "core"

                space = self.user_states[sender_id].get("memory_space", "core")
                mems = self.memory_manager.get_memories(space)
            else:
                mems = self.memory_manager.get_memories("user", user_uuid)

            # Sub-commands handling (mostly triggered via buttons redirecting to these)
            parts = text.split(" ")
            cmd = parts[0]  # /memory or /memory_btn_...

            # Check for subcommands like /memory remove 0 first
            if cmd == "/memory" and len(parts) > 1:
                sub = parts[1].lower()
                if sub == "remove" and len(parts) > 2:
                    try:
                        idx = int(parts[2])
                        mem_type = self.user_states[sender_id].get(
                            "memory_space", "core" if is_su_mode else "user"
                        )
                        self.memory_manager.delete_memory(mem_type, user_uuid, idx)
                        interface.send_message(
                            sender_id,
                            get_string("memory_removed", lang),
                            save_to_history=Config.INCLUDE_COMMAND_CONTEXT,
                        )
                    except (ValueError, IndexError):
                        interface.send_message(
                            sender_id,
                            get_string("memory_invalid_index", lang),
                            save_to_history=Config.INCLUDE_COMMAND_CONTEXT,
                        )
                return

            if cmd == "/memory" or cmd == "/memory_btn_personal":
                if cmd == "/memory_btn_personal":
                    self.user_states[sender_id]["memory_space"] = "personal"
                    # Re-fetch mems for personal space
                    mems = self.memory_manager.get_memories("personal")

                space = self.user_states[sender_id].get(
                    "memory_space", "core" if is_su_mode else "user"
                )

                if not mems:
                    msg = get_string("memory_list_empty", lang)
                else:
                    title_key = (
                        "memory_list_title_personal"
                        if space == "personal"
                        else "memory_list_title"
                    )
                    msg = get_string(title_key, lang) + "\n"
                    # Only display the content field, handle both old string format and new dict format
                    msg += "\n".join(
                        [
                            f"{i}: {m['content'] if isinstance(m, dict) else m}"
                            for i, m in enumerate(mems)
                        ]
                    )

                main_buttons = [
                    {
                        "text": get_string("memory_btn_add", lang),
                        "callback_data": "memory_add",
                    },
                    {
                        "text": get_string("memory_btn_edit", lang),
                        "callback_data": "memory_edit_index",
                    },
                    {
                        "text": get_string("memory_btn_delete", lang),
                        "callback_data": "memory_delete_index",
                    },
                ]

                buttons = [main_buttons]

                if is_su_mode:
                    space = self.user_states[sender_id].get("memory_space", "core")
                    if space == "core":
                        buttons.append(
                            [
                                {
                                    "text": get_string("memory_btn_personal", lang),
                                    "callback_data": "memory_personal",
                                }
                            ]
                        )
                    else:
                        buttons.append(
                            [
                                {
                                    "text": get_string("memory_btn_distill", lang),
                                    "callback_data": "memory_distill",
                                },
                                {
                                    "text": get_string("memory_btn_back", lang),
                                    "callback_data": "memory_main",
                                },
                            ]
                        )

                interface.send_message(
                    sender_id,
                    msg,
                    buttons=buttons,
                    save_to_history=Config.INCLUDE_COMMAND_CONTEXT,
                )

            elif cmd == "/memory_btn_add":
                self.user_states[sender_id]["state"] = "memory_adding"
                interface.send_message(
                    sender_id,
                    get_string("memory_prompt_add", lang),
                    save_to_history=Config.INCLUDE_COMMAND_CONTEXT,
                )

            elif cmd == "/memory_btn_delete":
                if not mems:
                    interface.send_message(
                        sender_id,
                        get_string("memory_list_empty", lang),
                        save_to_history=Config.INCLUDE_COMMAND_CONTEXT,
                    )
                    return
                # Create a grid of index buttons
                buttons = []
                row = []
                for i in range(len(mems)):
                    row.append(
                        {"text": str(i), "callback_data": f"memory_do_delete:{i}"}
                    )
                    if len(row) == 5:
                        buttons.append(row)
                        row = []
                if row:
                    buttons.append(row)
                buttons.append(
                    [
                        {
                            "text": get_string("memory_btn_back", lang),
                            "callback_data": "memory_main",
                        }
                    ]
                )
                interface.send_message(
                    sender_id,
                    get_string("memory_select_delete", lang),
                    buttons=buttons,
                    save_to_history=Config.INCLUDE_COMMAND_CONTEXT,
                )

            elif cmd == "/memory_btn_edit":
                if not mems:
                    interface.send_message(
                        sender_id,
                        get_string("memory_list_empty", lang),
                        save_to_history=Config.INCLUDE_COMMAND_CONTEXT,
                    )
                    return
                buttons = []
                row = []
                for i in range(len(mems)):
                    row.append({"text": str(i), "callback_data": f"memory_do_edit:{i}"})
                    if len(row) == 5:
                        buttons.append(row)
                        row = []
                if row:
                    buttons.append(row)
                buttons.append(
                    [
                        {
                            "text": get_string("memory_btn_back", lang),
                            "callback_data": "memory_main",
                        }
                    ]
                )
                interface.send_message(
                    sender_id,
                    get_string("memory_select_edit", lang),
                    buttons=buttons,
                    save_to_history=Config.INCLUDE_COMMAND_CONTEXT,
                )

            elif cmd == "/memory_btn_do_edit":
                idx = int(parts[1])
                self.user_states[sender_id].update(
                    {"state": "memory_editing", "index": idx}
                )
                interface.send_message(
                    sender_id,
                    get_string("memory_prompt_edit", lang),
                    save_to_history=Config.INCLUDE_COMMAND_CONTEXT,
                )

            return

        if text.strip().startswith("/schedule"):
            if not getattr(self, "conversation_scheduler", None):
                interface.send_message(
                    sender_id,
                    "Conversation Scheduler is not enabled.",
                    save_to_history=Config.INCLUDE_COMMAND_CONTEXT,
                )
                return

            # /schedule list
            # /schedule remove <id>
            parts = text.strip().split(" ")
            cmd_type = parts[1] if len(parts) > 1 else "list"

            if cmd_type == "list":
                schedules = self.conversation_scheduler.get_user_schedules(sender_id)
                buttons = []

                if not schedules:
                    msg = (
                        "暂无计划中的消息。"
                        if lang == "zh"
                        else "No scheduled messages."
                    )
                else:
                    msg = "Scheduled Messages:\n------------------\n\n"
                    schedules = sorted(
                        schedules, key=lambda x: x.get("created_at", ""), reverse=True
                    )

                    for i, s in enumerate(schedules, 1):
                        trigger = s.get(
                            "trigger_at", "Condition: " + str(s.get("condition"))
                        )
                        if s.get("type") == "worker":
                            trigger = f"Worker: `{s.get('worker_code')}`"

                        # Calculate time remaining if it's a timestamp
                        if s.get("trigger_at"):
                            try:
                                trigger_dt = datetime.fromisoformat(s["trigger_at"])
                                remaining = trigger_dt - datetime.now()
                                if remaining.total_seconds() > 0:
                                    trigger += f" (in {str(remaining).split('.')[0]})"
                                else:
                                    trigger += " (Processing...)"
                            except:
                                pass

                        msg += f"{i}. Trigger: {trigger}\n   Message: {s['message'][:50]}...\n\n"

                        # Add a delete button for this item
                        btn_text = f"删除 #{i}" if lang == "zh" else f"Delete #{i}"
                        buttons.append(
                            [
                                {
                                    "text": btn_text,
                                    "callback_data": f"schedule_do_delete:{s['id']}",
                                }
                            ]
                        )

                # Global Refresh Button
                buttons.append(
                    [
                        {
                            "text": "刷新" if lang == "zh" else "Refresh",
                            "callback_data": "schedule_refresh",
                        }
                    ]
                )

                interface.send_message(
                    sender_id,
                    msg,
                    buttons=buttons,
                    save_to_history=Config.INCLUDE_COMMAND_CONTEXT,
                )

            elif cmd_type == "remove" and len(parts) > 2:
                arg = parts[2]
                sched_id = arg

                # Support removing by index (1-based)
                if arg.isdigit() and len(arg) < 5:
                    try:
                        idx = int(arg)
                        schedules = self.conversation_scheduler.get_user_schedules(
                            sender_id
                        )
                        # Must match the sort order in 'list'
                        schedules = sorted(
                            schedules,
                            key=lambda x: x.get("created_at", ""),
                            reverse=True,
                        )
                        if 1 <= idx <= len(schedules):
                            sched_id = schedules[idx - 1]["id"]
                        else:
                            # Invalid index will result in remove_schedule returning false, which is handled below
                            pass
                    except:
                        pass

                success = self.conversation_scheduler.remove_schedule(sched_id)

                # If command came from a callback (implied by context usually, but here we just send a msg)
                # Ideally we would edit the message, but current architecture sends new message.
                # Let's send a brief toast and then the updated list.

                status_msg = "删除成功" if success else "未找到该任务"
                if lang != "zh":
                    status_msg = "Schedule removed" if success else "Schedule not found"

                interface.send_message(
                    sender_id,
                    status_msg,
                    save_to_history=Config.INCLUDE_COMMAND_CONTEXT,
                )

                # Auto-refresh list
                await self._process_message_logic(
                    sender_id, "/schedule list", [], metadata
                )
                return
            else:
                interface.send_message(
                    sender_id,
                    "用法:\n/schedule list\n/schedule remove <id>"
                    if lang == "zh"
                    else "Usage:\n/schedule list\n/schedule remove <id>",
                    save_to_history=Config.INCLUDE_COMMAND_CONTEXT,
                )
            return

        elif text.startswith("/explore "):
            # Verify superuser for autonomous trigger
            update = metadata.get("update")
            username = (
                update.effective_user.username
                if update and update.effective_user
                else None
            )

            if not username or username not in Config.SUPERUSERS:
                logger.warning(
                    f"Unauthorized /explore attempt by {username if username else 'unknown'}"
                )
                if interface:
                    interface.send_message(
                        sender_id, get_string("explore_unauthorized", lang)
                    )
                return

            if not hasattr(self, "autonomous_interface"):
                if interface:
                    interface.send_message(
                        sender_id, get_string("explore_no_interface", lang)
                    )
                return

            parts = text.split(" ", 2)
            topic = parts[1] if len(parts) > 1 else ""
            additional_msg = parts[2] if len(parts) > 2 else ""

            if not topic:
                if interface:
                    interface.send_message(sender_id, get_string("explore_usage", lang))
                return

            # Inform user
            launch_msg = get_string("explore_started", lang)
            if interface:
                interface.send_message(sender_id, launch_msg)

            # Trigger in background
            self._create_background_task(
                self.autonomous_interface.trigger_exploration(topic, additional_msg)
            )
            return

        elif text.strip().startswith("/profile"):
            if user_uuid and user_uuid != "__autonomous__" and user_uuid != "__sudo__":
                parts = text.strip().split(" ")
                if len(parts) >= 3:
                    nickname = parts[1].strip()
                    timezone_str = parts[2].strip().upper()
                    if self.user_mapping.set_profile(user_uuid, nickname, timezone_str):
                        msg = get_string("profile_set_success", lang).format(
                            nickname=nickname, timezone=timezone_str
                        )
                    else:
                        msg = get_string("profile_set_fail", lang)
                else:
                    udata = self.user_mapping.get_user_data(user_uuid)
                    current_nick = udata.get("nickname") if udata else None
                    current_tz = udata.get("timezone") if udata else None
                    if current_nick and current_tz:
                        msg = get_string("profile_current", lang).format(
                            nickname=current_nick, timezone=current_tz
                        )
                    else:
                        msg = get_string("profile_usage", lang)
                interface.send_message(
                    sender_id, msg, save_to_history=Config.INCLUDE_COMMAND_CONTEXT
                )
            else:
                interface.send_message(
                    sender_id,
                    get_string("no_uuid", lang),
                    save_to_history=Config.INCLUDE_COMMAND_CONTEXT,
                )
            return

        # Check for incomplete profile (Startup Reminder / Migration)
        if (
            user_uuid
            and user_uuid not in ["__autonomous__", "__sudo__"]
            and not is_command
        ):
            udata = self.user_mapping.get_user_data(user_uuid)
            nickname = udata.get("nickname") if udata else None
            timezone_str = udata.get("timezone") if udata else None
            if not nickname or not timezone_str:
                if not self.user_states.get(sender_id, {}).get(
                    "profile_reminded", False
                ):
                    reminder = get_string("profile_reminder", lang)
                    interface.send_message(sender_id, reminder)
                    if sender_id not in self.user_states:
                        self.user_states[sender_id] = {}
                    self.user_states[sender_id]["profile_reminded"] = True

        # For existing users, ensure paths are up to date
        if user_uuid:
            history_path = (
                interface.get_history_path(sender_id)
                if hasattr(interface, "get_history_path")
                else ""
            )
            self.user_mapping.update_paths(
                user_uuid, interface_name, history_path, self.memory_manager.db.db_path
            )

        # 1.1 State Handling (Interactions from buttons)
        if sender_id in self.user_states and user_uuid:
            if text.startswith("/"):
                # If a new command is sent, cancel the pending state
                self.user_states[sender_id].pop("state", None)
                self.user_states[sender_id].pop("index", None)
            else:
                state = self.user_states[sender_id].pop("state", None)

                if state == "memory_adding":
                    mem_type = self.user_states[sender_id].get(
                        "memory_space", "core" if is_su_mode else "user"
                    )
                    self.memory_manager.add_memory(mem_type, user_uuid, text[:500])
                    interface.send_message(
                        sender_id,
                        get_string("memory_added", lang),
                        save_to_history=Config.INCLUDE_COMMAND_CONTEXT,
                    )
                    return
                elif state == "memory_editing":
                    idx = self.user_states[sender_id].pop("index", None)
                    if isinstance(idx, int):
                        mem_type = self.user_states[sender_id].get(
                            "memory_space", "core" if is_su_mode else "user"
                        )
                        self.memory_manager.update_memory(
                            mem_type, user_uuid, idx, text[:500]
                        )
                        interface.send_message(
                            sender_id,
                            get_string("memory_updated", lang),
                            save_to_history=Config.INCLUDE_COMMAND_CONTEXT,
                        )
                    return

        # 1.2 Command handling (Early Exit)
        if text.strip().lower() == "/new":
            if interface and hasattr(interface, "clear_history"):
                interface.clear_history(sender_id)
            elif interface and hasattr(interface, "history_dir"):
                # Legacy support
                file_path = os.path.join(interface.history_dir, f"{sender_id}.json")
                if os.path.exists(file_path):
                    os.remove(file_path)

            if sender_id in self.pending_messages:
                self.pending_messages.pop(sender_id)
            if sender_id in self.debounce_tasks:
                self.debounce_tasks[sender_id].cancel()
                self.debounce_tasks.pop(sender_id)

            for sa in self.subagents:
                if hasattr(sa, "clear_tasks"):
                    sa.clear_tasks(sender_id)

            # Clear session summary from DB as well
            if user_uuid:
                self.memory_manager.delete_session_summary(user_uuid)

            if interface:
                interface.send_message(
                    sender_id,
                    get_string("memory_cleared", lang),
                    save_to_history=Config.INCLUDE_COMMAND_CONTEXT,
                )
            return

        # 1.3 Check for missing UUID (guide new users)
        if not user_uuid and not is_command:
            interface.send_message(
                sender_id,
                get_string("no_uuid", lang),
                save_to_history=False,
            )
            return

        # 1.4 Save message to persistent history (Pre-history fetch)
        if (
            not is_autonomous
            and not is_su_mode
            and interface
            and hasattr(interface, "save_history")
        ):
            try:
                # If we have original individual messages (from debouncing), save each one
                if original_messages:
                    for m in original_messages:
                        m_text = m.get("text")
                        m_meta = m.get("metadata", {})
                        if not m_text:
                            continue

                        m_msg_id_meta = None
                        if interface_name == "telegram" and m_meta.get("update"):
                            upd = m_meta["update"]
                            if hasattr(upd, "message") and upd.message:
                                m_msg_id_meta = upd.message.message_id

                        m_id = interface.save_history(
                            sender_id,
                            {
                                "role": "user",
                                "content": m_text,
                                "timestamp": datetime.now().isoformat(),
                                "images": m_meta.get("images"),
                                "reply_to": m_meta.get("reply_to"),
                                "telegram_id": m_msg_id_meta,
                                "search_summary": m_meta.get("search_summary"),
                            },
                        )
                        # Store the ID of the LAST message for legacy reference if needed
                        if m_id:
                            metadata["current_msg_id"] = m_id
                else:
                    # Fallback for direct calls
                    msg_id_meta = None
                    if interface_name == "telegram" and metadata.get("update"):
                        upd = metadata["update"]
                        if hasattr(upd, "message") and upd.message:
                            msg_id_meta = upd.message.message_id

                    msg_id = interface.save_history(
                        sender_id,
                        {
                            "role": "user",
                            "content": text,
                            "timestamp": datetime.now().isoformat(),
                            "images": images if images else None,
                            "reply_to": metadata.get("reply_to"),
                            "telegram_id": msg_id_meta,
                            "search_summary": kwargs.get("search_summary"),
                        },
                    )
                    if msg_id:
                        metadata["current_msg_id"] = msg_id
            except Exception as e:
                logger.error(f"Failed to save user message to history: {e}")

        # 2. Fetch persistent history
        chat_history = []
        if is_autonomous:
            # Autonomous sessions use history stored in metadata to persist across search loops
            chat_history = metadata.get("auto_history", [])
        elif is_su_mode:
            # Gather all histories from all interfaces
            for intf in self.interfaces:
                if hasattr(intf, "get_all_histories"):
                    chat_history.extend(intf.get_all_histories())
            # Re-sort combined histories by timestamp
            chat_history.sort(key=lambda x: x.get("timestamp", ""))
            # Global history can be larger
            chat_history = chat_history[-100:]
        elif interface and hasattr(interface, "get_history"):
            chat_history = interface.get_history(sender_id)

        if self.verbose:
            logger.debug(
                f"CONVERSATION HISTORY ROUNDS FOR {sender_id}: {len(chat_history)} (SU: {is_su_mode})"
            )

        # Update detected language based on history if needed
        if not text_is_zh and chat_history:
            lang = chat_history[-1].get("lang", lang)

        # Prepare User Message Obj for background analysis
        user_msg_obj = {
            "role": "user",
            "content": text,
            "sender_id": sender_id,
            "interface": interface_name,
            "lang": lang,
            "images": images,
            "image_description": None,
            "search_summary": kwargs.get("search_summary"),
            "reply_to": metadata.get("reply_to"),
            "id": metadata.get("current_msg_id"),
            "timestamp": datetime.now().isoformat(),
        }

        if not self.main_provider:
            logger.error("No main provider available.")
            return

        provider = self.main_provider

        # Background: Analyze images if present to create "textual memory"
        if images:
            self._create_background_task(
                self._analyze_images_background(user_msg_obj, provider, lang, interface)
            )

        # Start "typing" status if interface supports it
        typing_event = asyncio.Event()
        typing_task = None
        if interface:

            async def typing_loop():
                while not typing_event.is_set():
                    interface.send_chat_action(sender_id, "typing")
                    try:
                        await asyncio.wait_for(typing_event.wait(), timeout=4)
                    except asyncio.TimeoutError:
                        continue

            typing_task = self._create_background_task(typing_loop())

        try:
            # Combine interface chat history with current user message
            # chat_history already includes current text if it was persistent
            if not is_autonomous and not is_su_mode:
                current_context = chat_history
            else:
                # For autonomous/sudo, manually handle context
                current_context = chat_history + [user_msg_obj]

            # Get localized system prompt
            current_system_prompt = get_prompt("system_prompt", lang)

            # Inject Long-term Memory
            if is_su_mode:
                # Exclude personal memory for objectivity in SU mode
                memory_context = self.memory_manager.get_all_memories_context(
                    include_personal=False
                )
                if memory_context:
                    current_system_prompt += f"\n\n{memory_context}"
                    su_note = get_prompt("main.sudo", lang)
                    current_system_prompt += f"\n\n{su_note}"
            elif user_uuid:
                # When not in sudo mode, include personal memory (for personality) and user memory. Core memory is now always included.
                # Use current chat history for vector context
                memory_context = await self.memory_manager.get_memory_context(
                    user_uuid, conversation=chat_history, include_personal=True
                )
                if memory_context:
                    current_system_prompt += f"\n\n{memory_context}"

                # Inject Session Summary (if available)
                # This helps bridge the gap for scrolled-out messages
                # We only inject if the context history actually exceeds the limit
                inject_summary = len(chat_history) >= Config.CONTEXT_HISTORY_LIMIT
                session_summary_context = (
                    self.memory_manager.get_session_summary(user_uuid)
                    if inject_summary
                    else None
                )
                if session_summary_context:
                    logger.info(f"Injecting Session Summary for {user_uuid}")
                    current_system_prompt += f"\n\n{session_summary_context}"

            # Inject dynamic time info (localized if possible)
            utc_now = datetime.now(timezone.utc)
            time_info = f"\n\n[UTC Time: {utc_now.strftime('%Y-%m-%d %H:%M:%S')}]"

            if user_uuid and user_uuid not in ["__autonomous__", "__sudo__"]:
                timezone_str = self.user_mapping.get_timezone(user_uuid)
                if timezone_str:
                    local_time = self._get_local_time(utc_now, timezone_str)
                    if local_time:
                        time_info += f"\n[User Local Time ({timezone_str}): {local_time.strftime('%Y-%m-%d %H:%M:%S')}]"

            current_system_prompt += time_info

            for sa in self.subagents:
                try:
                    sa_ctx = await sa.get_context(sender_id)
                    if sa_ctx:
                        ctx_json = json.dumps(sa_ctx, ensure_ascii=False, indent=2)
                        current_system_prompt += (
                            f"\n\n[Sub-agent Context: {sa.name}]\n{ctx_json}"
                        )
                except Exception as e:
                    logger.error(
                        f"Failed to process subagent {sa.name} for prompt: {e}"
                    )

            # Append interface-specific instructions if they exist, AND not in autonomous mode
            if not is_autonomous:
                interface_instruction = get_prompt(
                    f"interface_instruction_{interface_name}", lang
                )
                if (
                    interface_instruction
                    and interface_instruction
                    != f"interface_instruction_{interface_name}"
                ):
                    current_system_prompt += f"\n\n{interface_instruction}"

            # Prepare messages list for the provider
            timezone_str = (
                self.user_mapping.get_timezone(user_uuid)
                if user_uuid and user_uuid not in ["__autonomous__", "__sudo__"]
                else None
            )

            context_to_send = (
                current_context[-Config.CONTEXT_HISTORY_LIMIT :]
                if is_su_mode
                else current_context[-Config.CONTEXT_HISTORY_LIMIT :]
            )
            messages_to_send = self.build_messages(
                context_to_send, images if images else None, timezone_str=timezone_str
            )

            # Validate message order: LLM APIs require last message to be from 'user' role
            # This can happen when history is fetched after saving assistant response but before current user message is saved
            if messages_to_send and messages_to_send[-1].get("role") != "user":
                logger.warning(
                    f"Message sequence does not end with user role. Appending current user message."
                )
                # Build the current user message
                current_user_content = user_msg_obj.get("content", text)
                if images:
                    content_list = [{"type": "text", "text": current_user_content}]
                    for img in images:
                        content_list.append(
                            {
                                "type": "image_url",
                                "image_url": {"url": f"data:image/jpeg;base64,{img}"},
                            }
                        )
                    messages_to_send.append({"role": "user", "content": content_list})
                else:
                    messages_to_send.append(
                        {"role": "user", "content": current_user_content}
                    )

            model_to_use = (
                provider.default_multimodal_model
                if images
                else provider.default_text_model
            )

            if self.verbose:
                logger.debug(f"REQUESTING MODEL: {model_to_use}")
                logger.debug(f"SYSTEM PROMPT: {current_system_prompt}")
                logger.debug(
                    f"MESSAGES: {json.dumps(messages_to_send, ensure_ascii=False, indent=2)}"
                )

            request_params = {
                "temperature": 1.0,
                "max_tokens": 4096,
                "include_thinking": False,
            }

            loop = asyncio.get_running_loop()

            llm_response: LLMResponse = await loop.run_in_executor(
                None,
                lambda: provider.request(
                    model=model_to_use,
                    system_prompt=current_system_prompt,
                    messages=messages_to_send,
                    params=ModelSpecificParams(**request_params),
                    tools=TOOLS,
                ),
            )

            if self.verbose:
                logger.debug(f"MODEL RESPONSE: {llm_response.content}")
                if llm_response.tool_calls:
                    logger.debug(
                        f"TOOL CALLS: {[tc.name for tc in llm_response.tool_calls]}"
                    )

            max_tool_rounds = 5
            tool_round = 0
            tool_messages_for_history = []

            while llm_response.tool_calls and tool_round < max_tool_rounds:
                tool_round += 1
                logger.info(
                    f"Tool calling round {tool_round}: {len(llm_response.tool_calls)} tools"
                )

                tool_results = await self.tool_executor.execute_tools_parallel(
                    llm_response.tool_calls,
                    sender_id,
                    user_uuid or "",
                    metadata,
                )

                assistant_tool_msg = {
                    "role": "assistant",
                    "content": llm_response.content or "",
                    "tool_calls": [
                        {
                            "id": tc.id,
                            "type": "function",
                            "function": {
                                "name": tc.name,
                                "arguments": json.dumps(tc.arguments),
                            },
                        }
                        for tc in llm_response.tool_calls
                    ],
                    "timestamp": datetime.now().isoformat(),
                }
                messages_to_send.append(assistant_tool_msg)
                tool_messages_for_history.append(assistant_tool_msg)

                for tr in tool_results:
                    tool_msg = {
                        "role": "tool",
                        "tool_call_id": tr.tool_call_id,
                        "name": tr.name,
                        "content": tr.result,
                        "timestamp": datetime.now().isoformat(),
                    }
                    messages_to_send.append(tool_msg)
                    tool_messages_for_history.append(tool_msg)

                llm_response = await loop.run_in_executor(
                    None,
                    lambda: provider.request(
                        model=model_to_use,
                        system_prompt=current_system_prompt,
                        messages=messages_to_send,
                        params=ModelSpecificParams(**request_params),
                        tools=TOOLS,
                    ),
                )

                if self.verbose:
                    logger.debug(
                        f"MODEL RESPONSE (after tools): {llm_response.content}"
                    )

            response_text = llm_response.content or ""

            if not response_text or not response_text.strip():
                logger.error(f"Received empty response from provider for {sender_id}")
                return

            response_text = response_text.strip()

            # Deduplicate: LLM sometimes outputs text with [SPLIT] markers
            # then repeats the same text without markers. Remove the duplicate.
            # Pattern: "text1[SPLIT]\ntext2[SPLIT]\ntext3\n\ntext1\ntext2\ntext3"
            split_parts = re.split(r"\[SPLIT\]", response_text)
            split_parts = [p.strip() for p in split_parts if p.strip()]
            if len(split_parts) >= 2:
                # Try to detect if the text after all [SPLIT]s is a repetition
                split_content = "\n".join(split_parts)
                remaining = response_text
                for part in split_parts:
                    remaining = remaining.replace(part, "", 1)
                remaining = remaining.replace("[SPLIT]", "").strip()

                if remaining and len(remaining) > 20:
                    # Check if remaining is a subset/duplicate of split_content
                    remaining_lines = [
                        l.strip() for l in remaining.split("\n") if l.strip()
                    ]
                    split_lines = [
                        l.strip() for l in split_content.split("\n") if l.strip()
                    ]
                    match_count = sum(1 for l in remaining_lines if l in split_lines)
                    if (
                        len(remaining_lines) > 0
                        and match_count / len(remaining_lines) > 0.7
                    ):
                        # More than 70% of remaining lines are duplicates
                        response_text = split_content

            if response_text.upper().startswith("EMPTY"):
                response_text = response_text[5:].strip()
                if not response_text:
                    response_text = "[EMPTY]"
            elif response_text.upper().startswith("[EMPTY]"):
                response_text = response_text[7:].strip()
                if not response_text:
                    response_text = "[EMPTY]"

            if response_text != "[EMPTY]":
                if response_text.endswith("[EMPTY]"):
                    response_text = response_text[:-7].strip()
                elif response_text.endswith("EMPTY"):
                    response_text = response_text[:-5].strip()

            user_facing_response = response_text
            if not user_facing_response or user_facing_response.upper() == "[EMPTY]":
                user_facing_response = "[EMPTY]"

            actual_response_context = current_context + tool_messages_for_history
            if response_text.strip() != "[EMPTY]":
                assistant_msg = {
                    "role": "assistant",
                    "content": response_text,
                    "timestamp": datetime.now().isoformat(),
                }
                actual_response_context = actual_response_context + [assistant_msg]

                if is_autonomous:
                    if "auto_history" not in metadata:
                        metadata["auto_history"] = []
                    metadata["auto_history"].append(user_msg_obj)
                    metadata["auto_history"].extend(tool_messages_for_history)
                    metadata["auto_history"].append(assistant_msg)
                    metadata["auto_history"] = metadata["auto_history"][-10:]

            if interface and not is_autonomous:
                if hasattr(interface, "save_history"):
                    interface.save_history(sender_id, user_msg_obj)
                    for tool_msg in tool_messages_for_history:
                        interface.save_history(sender_id, tool_msg)

                if user_facing_response.strip() == "[EMPTY]":
                    logger.info(f"Model chose to not respond for {sender_id}")
                else:
                    if response_text.strip() != "[EMPTY]":
                        interface.save_history(sender_id, assistant_msg)
                    interface.send_message(sender_id, user_facing_response)

            if is_autonomous:
                # Check for termination
                if "[DONE]" in response_text or "[ABANDON]" in response_text:
                    logger.info(
                        f"Autonomous session ended by AI ({'[DONE]' if '[DONE]' in response_text else '[ABANDON]'})."
                    )
                    return

                # Handle proactive contact response
                if metadata.get("proactive_contact"):
                    target_id = metadata.get("proactive_target_id")
                    target_uuid = metadata.get("proactive_target_uuid")

                    if "[SKIP]" in response_text:
                        logger.info(
                            f"Proactive contact to {target_uuid} skipped by AI: {response_text[:100]}"
                        )
                        return

                    # AI decided to send a message - deliver it to the target user
                    if target_id and user_facing_response.strip():
                        logger.info(
                            f"Proactive contact: Sending message to {target_uuid} ({target_id})"
                        )

                        # Find the Telegram interface
                        tg_interface = None
                        for iface in self.interfaces:
                            if (
                                hasattr(iface, "__class__")
                                and iface.__class__.__name__ == "TelegramInterface"
                            ):
                                tg_interface = iface
                                break

                        if tg_interface:
                            try:
                                # Clean up the response (remove any system markers)
                                clean_message = user_facing_response.strip()
                                # Remove [SKIP] if it somehow ended up in the message
                                clean_message = clean_message.replace(
                                    "[SKIP]", ""
                                ).strip()

                                if clean_message:
                                    tg_interface.send_message(
                                        target_id, clean_message, save_to_history=True
                                    )
                                    logger.info(
                                        f"Proactive message sent to {target_uuid}"
                                    )
                            except Exception as e:
                                logger.error(f"Failed to send proactive message: {e}")
                        else:
                            logger.warning(
                                "Telegram interface not found for proactive contact delivery"
                            )
                    return

                # If it's autonomous and didn't use a search tag, it's just "thinking".
                # We check if a search task was added to skip the nudge.
                if "<search>" not in response_text:
                    nudge_msg = None
                    for sa in self.subagents:
                        custom_nudge = await sa.get_nudge_message(lang, metadata)
                        if custom_nudge:
                            nudge_msg = custom_nudge
                            break

                    if nudge_msg:
                        logger.info(
                            f"Nudging autonomous session ({sender_id}) to continue..."
                        )
                        # This trigger is awaited, but let's be careful. subagent_trigger is async.
                        # It's called from _process_message_logic which is also async.
                        # Wait, subagent_trigger is awaited below:
                        await self.subagent_trigger(sender_id, metadata, nudge_msg)
        finally:
            # Stop "typing" status
            typing_event.set()
            if typing_task:
                await typing_task

    async def _analyze_images_background(
        self,
        message_obj: Dict[str, Any],
        provider: AriseProvider,
        lang: str,
        interface: AriseInterface | None,
    ):
        images = message_obj.get("images", [])
        if not images:
            return

        sender_id = message_obj.get("sender_id")
        logger.info(f"Analyzing {len(images)} images in background for {sender_id}...")

        analysis_prompt = get_prompt("image_analysis_prompt", lang)

        content: List[Dict[str, Any]] = [{"type": "text", "text": analysis_prompt}]
        for img in images:
            content.append(
                {
                    "type": "image_url",
                    "image_url": {"url": f"data:image/jpeg;base64,{img}"},
                }
            )

        try:
            loop = asyncio.get_running_loop()
            description = await loop.run_in_executor(
                None,
                lambda: provider.request(
                    model=provider.default_multimodal_model,
                    system_prompt="You are a vision assistant. Provide a detailed but concise textual description of images.",
                    messages=[{"role": "user", "content": content}],
                    params=ModelSpecificParams(
                        temperature=0.3, max_tokens=300, include_thinking=False
                    ),
                ),
            )

            if description:
                message_obj["image_description"] = description.strip()
                logger.info(f"Image analysis complete for {sender_id}.")

                # Update persistent history if possible
                if (
                    interface
                    and sender_id
                    and hasattr(interface, "update_last_message_content")
                ):
                    # We need to find the user message. Usually it's the second to last one
                    # because the assistant message was just appended.
                    # Let's adjust TelegramInterface to find the specific message or just update a specific one.
                    # For simplicity, we'll try to update the second-to-last one if the last is assistant.

                    if hasattr(interface, "update_last_message_content"):
                        try:
                            interface.update_last_message_content(
                                sender_id,
                                {"image_description": description.strip()},
                                match_content=message_obj["content"],
                            )
                        except Exception as e:
                            logger.error(
                                f"Failed to update history with image description: {e}"
                            )
        except Exception as e:
            logger.error(f"Background image analysis failed: {e}")

    def build_messages(
        self,
        history: List[Dict[str, Any]],
        images: Union[List[str], None] = None,
        timezone_str: Optional[str] = None,
    ) -> List[Dict[str, Any]]:
        messages = []
        for i, msg in enumerate(history):
            role = msg["role"]

            if role == "tool":
                messages.append(
                    {
                        "role": "tool",
                        "tool_call_id": msg.get("tool_call_id", ""),
                        "content": msg.get("content", ""),
                    }
                )
                continue

            if role == "assistant" and msg.get("tool_calls"):
                messages.append(
                    {
                        "role": "assistant",
                        "content": msg.get("content", ""),
                        "tool_calls": msg.get("tool_calls", []),
                    }
                )
                continue

            content = msg["content"]

            conv_id = msg.get("conv_id")
            if conv_id and role == "user":
                content = f"[Conv:{conv_id}] {content}"

            reply_to = msg.get("reply_to")
            if reply_to:
                rep_sender = (
                    "you" if reply_to.get("is_bot") else reply_to.get("sender", "User")
                )
                rep_text = reply_to.get("text", "")
                content = f'(Replying to {rep_sender}: "{rep_text}")\n{content}'

            if role == "user" and msg.get("image_description"):
                content += f"\n[Image Description: {msg['image_description']}]"

            if role == "user" and msg.get("search_summary"):
                content += f"\n[Search Summary: {msg['search_summary']}]"

            if i == len(history) - 1 and role == "user" and images:
                content_list = [{"type": "text", "text": content}]
                for img in images:
                    content_list.append(
                        {
                            "type": "image_url",
                            "image_url": {"url": f"data:image/jpeg;base64,{img}"},
                        }
                    )
                messages.append({"role": role, "content": content_list})
            else:
                messages.append({"role": role, "content": content})

        return messages

    def _get_local_time(self, utc_now: datetime, tz_str: str) -> Optional[datetime]:
        """Helper to calculate local time based on common 3-letter timezone abbreviations."""
        # Common mapping. For CST, we default to UTC+8 if it's the most common use case in this project?
        # Actually, let's support some standard ones.
        tz_map = {
            "UTC": 0,
            "GMT": 0,
            "EST": -5,
            "EDT": -4,
            "CST": 8,  # Defaulting to China Standard Time as per project primary locale
            "MST": -7,
            "MDT": -6,
            "PST": -8,
            "PDT": -7,
            "JST": 9,
            "KST": 9,
            "AEST": 10,
            "AEDT": 11,
            "BJT": 8,
            "HKT": 8,
        }

        # Check if it's a numeric offset first (e.g. +8, -5)
        try:
            if tz_str.startswith("+") or tz_str.startswith("-"):
                offset = int(tz_str)
                return utc_now + timedelta(hours=offset)

            if tz_str in tz_map:
                offset = tz_map[tz_str]
                return utc_now + timedelta(hours=offset)
        except:
            pass
        return None

    def build_prompt_with_history(
        self, history: List[Dict[str, Any]], timezone_str: Optional[str] = None
    ) -> str:
        prompt = ""

        # Incorporate messages for context window
        for msg in history:
            role = "User" if msg["role"] == "user" else "Assistant"

            # If we have a conversation ID (from SU mode), include it to help the model distinguish users
            conv_id = msg.get("conv_id")
            if conv_id:
                role = f"{role} [Conv:{conv_id}]"

            content = msg["content"]

            # Include Message ID and Time for LLM referencing
            msg_id = msg.get("id")
            msg_ts = msg.get("timestamp", "")

            # Localize timestamp if timezone is available
            raw_ts = msg.get("raw_timestamp")
            if timezone_str and raw_ts:
                try:
                    dt_utc = datetime.fromtimestamp(raw_ts, tz=timezone.utc)
                    local_dt = self._get_local_time(dt_utc, timezone_str)
                    if local_dt:
                        msg_ts = local_dt.strftime("%Y-%m-%d %H:%M:%S")
                except:
                    pass

            if msg_id:
                content = f"[time:{msg_ts};id:{msg_id}] {content}"

            # Handle Reply Context
            reply_to = msg.get("reply_to")
            if reply_to:
                rep_sender = (
                    "you" if reply_to.get("is_bot") else reply_to.get("sender", "User")
                )
                rep_text = reply_to.get("text", "")
                content = f'(Replying to {rep_sender}: "{rep_text}")\n{content}'

            # If this was a user message with an image that has been analyzed, augment the content
            if msg["role"] == "user" and msg.get("image_description"):
                content += f"\n[Image Description: {msg['image_description']}]"

            # If this was a search result message with a summary, augment the content
            if msg["role"] == "user" and msg.get("search_summary"):
                content += f"\n[Search Summary: {msg['search_summary']}]"

            prompt += f"{role}: {content}\n"

        prompt += "\nAssistant:"
        return prompt

    async def run(self):
        self.load_components()

        # 热重载逻辑（仅在DEBUG模式下）
        observer = None
        if Config.DEBUG:
            event_handler = HotReloadHandler(debounce_delay=2.0)
            observer = Observer()
            observer.schedule(event_handler, path=".", recursive=True)
            observer.start()
            logger.info("Hot reload enabled in DEBUG mode.")

        tasks = [asyncio.create_task(i.start()) for i in self.interfaces]
        # Start subagent background loops
        for sa in self.subagents:
            tasks.append(asyncio.create_task(sa.run()))

        # Calculate load time and show ready message (always shown regardless of verbose setting)
        load_time_ms = (time.time() - self.start_time) * 1000
        # Use ANSI color codes for terminal highlighting (cyan for the message)
        logger.info(
            f"\033[96m✨ A.R.I.S.E. is ready! (loaded in {load_time_ms:.0f}ms)\033[0m"
        )
        try:
            await asyncio.gather(*tasks)
        except (asyncio.CancelledError, KeyboardInterrupt):
            logger.info("Stopping all ARISE components...")

            # Cancel all tracked background tasks
            if self._background_tasks:
                logger.info(
                    f"Cancelling {len(self._background_tasks)} background tasks..."
                )
                for task in self._background_tasks:
                    task.cancel()
                await asyncio.gather(*self._background_tasks, return_exceptions=True)

            for task in tasks:
                task.cancel()
            await asyncio.gather(*tasks, return_exceptions=True)

            # Stop MemoryManager tasks
            await self.memory_manager.stop()

            logger.info("All components stopped.")
        finally:
            if observer:
                observer.stop()
                observer.join()


if __name__ == "__main__":
    arise = Arise()
    try:
        asyncio.run(arise.run())
    except KeyboardInterrupt:
        logger.info("ARISE stopped by user.")