"""Agent loop: the core processing engine."""

from __future__ import annotations

import asyncio
import json
import re
import time
import uuid
from contextlib import AsyncExitStack
from datetime import datetime
from pathlib import Path
from typing import TYPE_CHECKING, Any

from loguru import logger

from vikingbot.agent.context import ContextBuilder
from vikingbot.agent.memory import MemoryStore
from vikingbot.agent.subagent import SubagentManager
from vikingbot.agent.tools import register_default_tools
from vikingbot.agent.tools.registry import ToolRegistry
from vikingbot.bus.events import InboundMessage, OutboundEventType, OutboundMessage
from vikingbot.bus.queue import MessageBus
from vikingbot.config import load_config
from vikingbot.config.schema import BotMode, Config, SessionKey
from vikingbot.heartbeat.service import HEARTBEAT_METADATA_KEY, is_heartbeat_noop_response
from vikingbot.hooks import HookContext
from vikingbot.hooks.manager import hook_manager
from vikingbot.integrations.langfuse import LangfuseClient
from vikingbot.observability.outcome import evaluate_response_outcome, should_update_outcome
from vikingbot.providers.base import LLMProvider
from vikingbot.sandbox import SandboxManager
from vikingbot.session.manager import Session, SessionManager
from vikingbot.utils.helpers import cal_str_tokens
from vikingbot.utils.tracing import set_response_id, trace

if TYPE_CHECKING:
    from vikingbot.config.schema import ExecToolConfig
    from vikingbot.cron.service import CronService


class AgentLoop:
    """
    The agent loop is the core processing engine.

    It:
    1. Receives messages from the bus
    2. Builds context with history, memory, skills
    3. Calls the LLM
    4. Executes tool calls
    5. Sends responses back
    """

    def __init__(
        self,
        bus: MessageBus,
        provider: LLMProvider,
        workspace: Path,
        model: str | None = None,
        max_iterations: int = 50,
        memory_window: int = 50,
        brave_api_key: str | None = None,
        exa_api_key: str | None = None,
        gen_image_model: str | None = None,
        exec_config: "ExecToolConfig | None" = None,
        cron_service: "CronService | None" = None,
        session_manager: SessionManager | None = None,
        sandbox_manager: SandboxManager | None = None,
        config: Config = None,
        eval: bool = False,
        mcp_servers: dict | None = None,
    ):
        """
        Initialize the AgentLoop with all required dependencies and configuration.

        Args:
            bus: MessageBus instance for publishing and subscribing to messages.
            provider: LLMProvider instance for making LLM calls.
            workspace: Path to the workspace directory for file operations.
            model: Optional model identifier. If not provided, uses the provider's default.
            max_iterations: Maximum number of tool execution iterations per message (default: 50).
            memory_window: Maximum number of messages to keep in session memory (default: 50).
            brave_api_key: Optional API key for Brave search integration.
            exa_api_key: Optional API key for Exa search integration.
            gen_image_model: Optional model identifier for image generation (default: openai/doubao-seedream-4-5-251128).
            exec_config: Optional configuration for the exec tool (command execution).
            cron_service: Optional CronService for scheduled task management.
            session_manager: Optional SessionManager for session persistence. If not provided, a new one is created.
            sandbox_manager: Optional SandboxManager for sandboxed operations.
            config: Optional Config object with full configuration. Used if other parameters are not provided.

        Note:
            The AgentLoop creates its own ContextBuilder, SessionManager (if not provided),
            ToolRegistry, and SubagentManager during initialization.

        Example:
            >>> loop = AgentLoop(
            ...     bus=message_bus,
            ...     provider=llm_provider,
            ...     workspace=Path("/path/to/workspace"),
            ...     model="gpt-4",
            ...     max_iterations=30,
            ... )
        """
        from vikingbot.config.schema import ExecToolConfig  # noqa: F811

        self.bus = bus
        self.provider = provider
        self.workspace = workspace
        self.model = model or provider.get_default_model()
        self.max_iterations = max_iterations
        self.memory_window = memory_window
        self.brave_api_key = brave_api_key
        self.exa_api_key = exa_api_key
        self.gen_image_model = gen_image_model or "openai/doubao-seedream-4-5-251128"
        self.exec_config = exec_config or ExecToolConfig()
        self.cron_service = cron_service
        self.sandbox_manager = sandbox_manager
        self.config = config

        self.context = ContextBuilder(workspace, sandbox_manager=sandbox_manager)

        self._register_builtin_hooks()
        self.sessions = session_manager or SessionManager(
            self.config.bot_data_path, sandbox_manager=sandbox_manager
        )
        self.tools = ToolRegistry()
        self._eval = eval
        self.subagents = SubagentManager(
            provider=provider,
            workspace=workspace,
            bus=bus,
            config=self.config,
            model=self.model,
            sandbox_manager=sandbox_manager,
        )

        self._running = False
        self._mcp_servers = mcp_servers or {}
        self._mcp_stack: AsyncExitStack | None = None
        self._mcp_connected = False
        self._mcp_connecting = False
        self._register_default_tools()

    async def _connect_mcp(self) -> None:
        """Connect to configured MCP servers (one-time, lazy, retryable on failure).

        Ported from HKUDS/nanobot v0.1.5.
        """
        if self._mcp_connected or self._mcp_connecting or not self._mcp_servers:
            return
        self._mcp_connecting = True
        try:
            from vikingbot.agent.tools.mcp import connect_mcp_servers

            self._mcp_stack = AsyncExitStack()
            await self._mcp_stack.__aenter__()
            await connect_mcp_servers(self._mcp_servers, self.tools, self._mcp_stack)
            self._mcp_connected = True
        except Exception as e:
            logger.error(f"Failed to connect MCP servers (will retry next message): {e}")
            if self._mcp_stack:
                try:
                    await self._mcp_stack.aclose()
                except Exception:
                    pass
                self._mcp_stack = None
        finally:
            self._mcp_connecting = False

    async def close_mcp(self) -> None:
        """Close MCP server connections. Ported from HKUDS/nanobot v0.1.5."""
        if self._mcp_stack:
            try:
                await self._mcp_stack.aclose()
            except Exception:
                pass  # MCP SDK cancel scope cleanup is noisy but harmless
            self._mcp_stack = None
        self._mcp_connected = False

    async def _publish_thinking_event(
        self, session_key: SessionKey, event_type: OutboundEventType, content: str
    ) -> None:
        """
        Publish a thinking event to the message bus.

        Thinking events are used to communicate the agent's internal processing
        state to the user, such as when the agent is executing a tool or
        processing a complex request.

        Args:
            session_key: The session key identifying the conversation.
            event_type: The type of thinking event (e.g., THINKING, TOOL_START).
            content: The message content to display to the user.

        Note:
            This is an internal method used by the agent loop to communicate
            progress to users during long-running operations.

        Example:
            async def notify_tool_call() -> None:
                await self._publish_thinking_event(
                    session_key=SessionKey(
                        type="telegram",
                        channel_id="default",
                        chat_id="123",
                    ),
                    event_type=OutboundEventType.TOOL_CALL,
                    content="Executing web search...",
                )
        """
        await self.bus.publish_outbound(
            OutboundMessage(
                session_key=session_key,
                content=content,
                event_type=event_type,
            )
        )

    def _register_builtin_hooks(self):
        """Register built-in hooks."""
        hook_manager.register_path(self.config.hooks)

    def _register_default_tools(self) -> None:
        """Register default set of tools."""
        register_default_tools(
            registry=self.tools,
            config=self.config,
            send_callback=self.bus.publish_outbound,
            subagent_manager=self.subagents,
            cron_service=self.cron_service,
        )

    async def run(self) -> None:
        """Run the agent loop, processing messages from the bus."""
        self._running = True
        await self._connect_mcp()
        logger.info("Agent loop started")

        while self._running:
            try:
                # Wait for next message
                msg = await asyncio.wait_for(self.bus.consume_inbound(), timeout=1.0)

                # Process it
                try:
                    response = await self._process_message(msg)
                    if response:
                        await self.bus.publish_outbound(response)
                except Exception as e:
                    logger.exception(f"Error processing message: {e}")
                    # Send error response
                    await self.bus.publish_outbound(
                        OutboundMessage(
                            session_key=msg.session_key,
                            content=f"Sorry, I encountered an error: {str(e)}",
                            metadata=msg.metadata,
                        )
                    )
            except asyncio.TimeoutError:
                continue

    def stop(self) -> None:
        """Stop the agent loop."""
        self._running = False
        logger.info("Agent loop stopping")

    async def _run_agent_loop(
        self,
        messages: list[dict],
        session_key: SessionKey,
        publish_events: bool = True,
        sender_id: str | None = None,
        ov_tools_enable: bool = True,
        memory_user_ids: list[str] | None = None,
        disabled_tools: list[str] | None = None,
    ) -> tuple[str | None, str | None, list[dict], dict[str, int], int]:
        """
        Run the core agent loop: call LLM, execute tools, repeat until done.

        Args:
            messages: Initial message list
            session_key: Session key for tool execution context
            publish_events: Whether to publish ITERATION/REASONING/TOOL_CALL events to the bus
            ov_tools_enable: Whether to enable OpenViking tools for this session
            memory_user_ids: List of user IDs for memory retrieval
            disabled_tools: Tool names to hide from the model for this request

        Returns:
            tuple of (final_content, final_reasoning_content, tools_used, token_usage, iteration)
        """
        iteration = 0
        final_content = None
        final_reasoning_content = None
        tools_used: list[dict] = []
        token_usage = {
            "prompt_tokens": 0,
            "completion_tokens": 0,
            "total_tokens": 0,
        }
        write_exp_injected = False

        while iteration < self.max_iterations:
            iteration += 1

            if publish_events:
                await self.bus.publish_outbound(
                    OutboundMessage(
                        session_key=session_key,
                        content=f"Iteration {iteration}/{self.max_iterations}",
                        event_type=OutboundEventType.ITERATION,
                    )
                )

            response = await self.provider.chat(
                messages=messages,
                tools=self.tools.get_definitions(
                    ov_tools_enable=ov_tools_enable,
                    disabled_tools=disabled_tools,
                ),
                model=self.model,
                session_id=session_key.safe_name(),
            )
            if response.usage:
                cur_token = response.usage
                token_usage["prompt_tokens"] += cur_token["prompt_tokens"]
                token_usage["completion_tokens"] += cur_token["completion_tokens"]
                token_usage["total_tokens"] += cur_token["total_tokens"]

            if publish_events and response.reasoning_content:
                await self.bus.publish_outbound(
                    OutboundMessage(
                        session_key=session_key,
                        content=response.reasoning_content,
                        event_type=OutboundEventType.REASONING,
                    )
                )

            if response.has_tool_calls:
                # Inject experience memory before write-related tool calls (once per session)
                if not write_exp_injected:
                    _ov_cfg = load_config().ov_server
                    _write_tools = set(_ov_cfg.exp_write_tools)
                    if any(tc.name in _write_tools for tc in response.tool_calls):
                        write_exp_injected = True
                        try:
                            # Build query from last 3 user messages
                            _user_msgs = [
                                m["content"] for m in messages
                                if m.get("role") == "user" and isinstance(m.get("content"), str)
                            ]
                            _query = "\n".join(_user_msgs[-3:])
                            workspace_id = self.sandbox_manager.to_workspace_id(session_key) if self.sandbox_manager else "shared"
                            _exp = await self.context.memory.get_viking_experience_context(
                                query=_query, workspace_id=workspace_id
                            )
                            logger.info(f"[WRITE_EXP]: write tool detected, exp_found={bool(_exp)}, query={_query[:50]}")
                            if _exp:
                                messages.append({
                                    "role": "user",
                                    "content": f"## Relevant Agent Experience\n{_exp}",
                                })
                                continue
                        except Exception as _e:
                            logger.warning(f"[WRITE_EXP]: failed to load experience: {_e}")

                final_reasoning_content = response.reasoning_content
                args_list = [tc.arguments for tc in response.tool_calls]
                tool_call_dicts = [
                    {
                        "id": tc.id,
                        "type": "function",
                        "function": {
                            "name": tc.name,
                            "arguments": json.dumps(args),
                        },
                    }
                    for tc, args in zip(response.tool_calls, args_list, strict=False)
                ]
                messages = self.context.add_assistant_message(
                    messages,
                    response.content,
                    tool_call_dicts,
                    reasoning_content=response.reasoning_content,
                )

                # Stage 2: Execute all tools in parallel
                async def execute_single_tool(idx: int, tool_call):
                    """Execute a single tool and track execution time."""
                    tool_execute_start_time = time.time()
                    result = await self.tools.execute(
                        tool_call.name,
                        tool_call.arguments,
                        session_key=session_key,
                        sandbox_manager=self.sandbox_manager,
                        sender_id=sender_id,
                        memory_user_ids=memory_user_ids,
                    )
                    tool_execute_duration = (time.time() - tool_execute_start_time) * 1000
                    return idx, tool_call, result, tool_execute_duration

                # Run all tool executions in parallel
                tool_tasks = [
                    execute_single_tool(idx, tool_call)
                    for idx, tool_call in enumerate(response.tool_calls)
                ]
                results = await asyncio.gather(*tool_tasks)

                # Stage 3: Process results sequentially in original order
                for _idx, tool_call, result, tool_execute_duration in results:
                    args_str = json.dumps(tool_call.arguments, ensure_ascii=False)
                    logger.info(f"[TOOL_CALL]: {tool_call.name}({args_str[:200]})")
                    logger.info(f"[RESULT]: {str(result)[:600]}")

                    if publish_events:
                        await self.bus.publish_outbound(
                            OutboundMessage(
                                session_key=session_key,
                                content=f"{tool_call.name}({args_str})",
                                event_type=OutboundEventType.TOOL_CALL,
                            )
                        )
                        await self.bus.publish_outbound(
                            OutboundMessage(
                                session_key=session_key,
                                content=str(result),
                                event_type=OutboundEventType.TOOL_RESULT,
                            )
                        )
                    messages = self.context.add_tool_result(
                        messages, tool_call.id, tool_call.name, result
                    )

                    tool_used_dict = {
                        "tool_name": tool_call.name,
                        "args": args_str,
                        "result": result,
                        "duration": tool_execute_duration,
                        "execute_success": True
                        if result and "Error executing" not in result
                        else False,
                        "input_token": tool_call.tokens,
                        "output_token": cal_str_tokens(result, text_type="mixed"),
                    }
                    tools_used.append(tool_used_dict)

                messages.append(
                    {"role": "user", "content": "Reflect on the results and decide next steps."}
                )
            else:
                final_content = response.content
                final_reasoning_content = response.reasoning_content
                break

        if final_content is None or (isinstance(final_content, str) and not final_content.strip()):
            if iteration >= self.max_iterations:
                final_content = f"Reached {self.max_iterations} iterations without completion."
            else:
                final_content = "I've completed processing but have no response to give."

        return final_content, final_reasoning_content, tools_used, token_usage, iteration

    @trace(
        name="process_message",
        extract_session_id=lambda msg: msg.session_key.safe_name(),
        extract_user_id=lambda msg: msg.sender_id,
    )
    async def _process_message(self, msg: InboundMessage) -> OutboundMessage | None:
        """
        Process a single inbound message.

        Args:
            msg: The inbound message to process.
            session_key: Override session key (used by process_direct).

        Returns:
            The response message, or None if no response needed.
        """
        # Handle system messages (subagent announces)
        # The chat_id contains the original "channel:chat_id" to route back to
        start_time = time.time()
        long_running_notified = False

        # 监控处理时长，每50秒发送处理中提示事件
        async def check_long_running():
            nonlocal long_running_notified
            tick_count = 0
            # 最多发送7次提示
            max_ticks = 7

            while not long_running_notified and tick_count < max_ticks:
                await asyncio.sleep(60)
                if long_running_notified:
                    break
                if msg.metadata:
                    message_id = msg.metadata.get("message_id")
                    if message_id:
                        try:
                            # 发送处理中tick事件，对应channel会自行处理展示逻辑
                            await self.bus.publish_outbound(
                                OutboundMessage(
                                    session_key=msg.session_key,
                                    content="",
                                    metadata={
                                        "action": "processing_tick",
                                        "tick_count": tick_count,
                                        "message_id": message_id,
                                    },
                                )
                            )
                            tick_count += 1
                        except Exception as e:
                            logger.debug(f"Failed to send processing tick: {e}")

        monitor_task = asyncio.create_task(check_long_running())

        try:
            if msg.session_key.type == "system":
                return await self._process_system_message(msg)

            preview = msg.content[:80] + "..." if len(msg.content) > 80 else msg.content
            logger.info(f"Processing message from {msg.session_key}:{msg.sender_id}: {preview}")

            session_key = msg.session_key
            # For CLI/direct sessions, skip heartbeat by default
            skip_heartbeat = session_key.type == "cli"
            session = self.sessions.get_or_create(session_key, skip_heartbeat=skip_heartbeat)

            ov_tools_enable = self._get_ov_tools_enable(session_key)
            disabled_tools = msg.metadata.get("disabled_tools", []) if msg.metadata else []
            if not isinstance(disabled_tools, list):
                disabled_tools = []
            # Get profile_user_list from channel config
            profile_user_list = []
            # Try to get memory_users from message metadata first (CLI mode), then from channel config
            memory_user = msg.metadata.get("memory_users", []) if msg.metadata else []
            channel_config = self._get_channel_config(session_key)

            if channel_config and ov_tools_enable:
                profile_user_list = getattr(channel_config, "profile_user_list", [])
                # Only override if not already set from metadata
                if not memory_user:
                    memory_user = getattr(channel_config, "memory_user", None) or []

            # Handle slash commands
            is_group_chat = msg.metadata.get("chat_type") == "group" if msg.metadata else False
            if is_group_chat:
                cmd = msg.content
                cmd = re.sub(r"^\[[^\]]+\]:\s*", "", cmd)
                cmd = cmd.replace(f"@{msg.sender_id}", "").strip().lower()
            else:
                cmd = msg.content.strip().lower()
            if cmd == "/new":
                # Clone session for async consolidation, then immediately clear original
                if not self._check_cmd_auth(msg):
                    return OutboundMessage(
                        session_key=msg.session_key,
                        content="🐈 Sorry, you are not authorized to use this command.",
                        metadata=msg.metadata,
                    )
                session.clear()
                await self.sessions.save(session)
                return OutboundMessage(
                    session_key=msg.session_key,
                    content="🐈 New session started. Session history droped.",
                    metadata=msg.metadata,
                )
            elif cmd == "/compact":
                # Clone session for async consolidation, then immediately clear original
                if not self._check_cmd_auth(msg):
                    return OutboundMessage(
                        session_key=msg.session_key,
                        content="🐈 Sorry, you are not authorized to use this command.",
                        metadata=msg.metadata,
                    )
                session_clone = session.clone()
                session.clear()
                await self.sessions.save(session)
                # Run consolidation in background
                await self._safe_consolidate_memory(session_clone, archive_all=True)
                return OutboundMessage(
                    session_key=msg.session_key,
                    content="🐈 New session started. Memory consolidated.",
                    metadata=msg.metadata,
                )
            if cmd == "/remember":
                if not self._check_cmd_auth(msg):
                    return OutboundMessage(
                        session_key=msg.session_key,
                        content="🐈 Sorry, you are not authorized to use this command.",
                        metadata=msg.metadata,
                    )
                if ov_tools_enable:
                    session_clone = session.clone()
                    await self._consolidate_viking_memory(session_clone)
                return OutboundMessage(
                    session_key=msg.session_key,
                    content="This conversation has been submitted to memory storage.",
                    metadata=msg.metadata,
                )
            if cmd == "/help":
                return OutboundMessage(
                    session_key=msg.session_key,
                    content="🐈 vikingbot commands:\n/new — Start a new conversation\n/remember — Submit current session to memories and start new session\n/help — Show available commands",
                    metadata=msg.metadata,
                )

            # Debug mode handling
            if self.config.mode == BotMode.DEBUG:
                # In debug mode, only record message to session, no processing or reply
                await self._evaluate_previous_response_outcome(session, msg)
                session.add_message("user", msg.content, sender_id=msg.sender_id)
                await self.sessions.save(session)
                return None

            if not msg.need_reply:
                await self._evaluate_previous_response_outcome(session, msg)
                session.add_message("user", msg.content, sender_id=msg.sender_id)
                await self.sessions.save(session)
                return OutboundMessage(
                    session_key=msg.session_key,
                    content="",
                    metadata=msg.metadata,
                    event_type=OutboundEventType.NO_REPLY,
                )

            # Consolidate memory before processing if session is too large
            if len(session.messages) > self.memory_window:
                # Clone session for async consolidation, then immediately trim original
                session_clone = session.clone()
                keep_count = min(10, max(2, self.memory_window // 2))
                session.messages = session.messages[-keep_count:] if keep_count else []
                await self.sessions.save(session)
                # Run consolidation in background
                await self._safe_consolidate_memory(session_clone, archive_all=False)

            await self._evaluate_previous_response_outcome(session, msg)

            if self.sandbox_manager:
                message_workspace = self.sandbox_manager.get_workspace_path(session_key)
            else:
                message_workspace = self.workspace

            from vikingbot.agent.context import ContextBuilder

            message_context = ContextBuilder(
                message_workspace,
                sandbox_manager=self.sandbox_manager,
                sender_id=msg.sender_id,
                sender_name=msg.sender_name,
                is_group_chat=is_group_chat,
                eval=self._eval,
            )

            # Build initial messages (use get_history for LLM-formatted messages)
            provider_name = self.config.get_provider_name(self.model) if self.config else None
            messages = await message_context.build_messages(
                history=session.get_history(provider_name=provider_name),
                current_message=msg.content,
                media=msg.media if msg.media else None,
                session_key=msg.session_key,
                ov_tools_enable=ov_tools_enable,
                profile_user_list=profile_user_list,
                memory_users=memory_user,
            )
            relevant_memories = message_context.latest_relevant_memories
            # logger.info(f"New messages: {json.dumps(messages, indent=4)}")

            # Run agent loop within a stable response identity for tracing/tool spans.
            response_id = uuid.uuid4().hex
            with set_response_id(response_id):
                (
                    final_content,
                    final_reasoning_content,
                    tools_used,
                    token_usage,
                    iteration,
                ) = await self._run_agent_loop(
                    messages=messages,
                    session_key=session_key,
                    publish_events=True,
                    sender_id=msg.sender_id,
                    ov_tools_enable=ov_tools_enable,
                    memory_user_ids=memory_user,
                    disabled_tools=disabled_tools,
                )

            # Log response preview
            preview = final_content[:300] + "..." if len(final_content) > 300 else final_content
            logger.info(f"Response to {msg.session_key}: {preview}")

            response_completed = self._build_response_completed_payload(
                msg=msg,
                response_id=response_id,
                final_content=final_content,
                final_reasoning_content=final_reasoning_content,
                token_usage=token_usage,
                time_cost_seconds=time.time() - start_time,
                iteration=iteration,
                tools_used=tools_used,
            )

            is_heartbeat = bool(msg.metadata.get(HEARTBEAT_METADATA_KEY))
            if not (is_heartbeat and is_heartbeat_noop_response(final_content)):
                session.add_message("user", msg.content, sender_id=msg.sender_id)
                session.add_message(
                    "assistant",
                    final_content,
                    response_id=response_id,
                    tools_used=tools_used if tools_used else None,
                    token_usage=token_usage,
                    sender_id=msg.sender_id,
                    reasoning_content=final_reasoning_content,
                )
                session.metadata.setdefault("response_facts", {})[response_id] = response_completed
                await self.sessions.save(session)
            LangfuseClient.get_instance().update_generation_metadata(
                response_id, response_completed
            )
            response_metadata = dict(msg.metadata or {})
            if relevant_memories is not None:
                response_metadata["relevant_memories"] = relevant_memories
            await self.bus.publish_outbound(
                OutboundMessage(
                    session_key=msg.session_key,
                    content="",
                    event_type=OutboundEventType.RESPONSE_COMPLETED,
                    response_id=response_id,
                    metadata={"response_completed": response_completed},
                )
            )
            return OutboundMessage(
                session_key=msg.session_key,
                content=final_content,
                metadata=response_metadata,
                response_id=response_id,
                token_usage=token_usage,
                time_cost=response_completed["time_cost_ms"] / 1000,
                iteration=response_completed["iteration_count"],
                tools_used_names=response_completed["tools_used_names"],
            )
        finally:
            long_running_notified = True
            monitor_task.cancel()
            try:
                await monitor_task
            except asyncio.CancelledError:
                pass

    @staticmethod
    def _build_response_completed_payload(
        msg: InboundMessage,
        response_id: str,
        final_content: str,
        final_reasoning_content: str | None,
        token_usage: dict[str, Any],
        time_cost_seconds: float,
        iteration: int,
        tools_used: list[dict[str, Any]] | None,
    ) -> dict[str, Any]:
        """Build a stable response fact shared by analytics sinks."""
        prompt_tokens = int(token_usage.get("prompt_tokens", 0) or 0)
        completion_tokens = int(token_usage.get("completion_tokens", 0) or 0)
        total_tokens = int(token_usage.get("total_tokens", prompt_tokens + completion_tokens) or 0)
        tools_used_names = [
            str(tool_name)
            for tool in (tools_used or [])
            if (tool_name := tool.get("tool_name")) is not None
        ]
        return {
            "response_id": response_id,
            "session_id": msg.session_key.safe_name(),
            "user_id": msg.sender_id,
            "channel": msg.session_key.channel_key(),
            "session_type": msg.session_key.type,
            "time_cost_ms": round(time_cost_seconds * 1000),
            "prompt_tokens": prompt_tokens,
            "completion_tokens": completion_tokens,
            "total_tokens": total_tokens,
            "iteration_count": iteration,
            "tool_count": len(tools_used_names),
            "tools_used_names": tools_used_names,
            "response_length": len(final_content),
            "created_at": datetime.now().isoformat(),
            "has_reasoning": bool(final_reasoning_content),
        }

    async def _evaluate_previous_response_outcome(
        self, session: Session, msg: InboundMessage
    ) -> None:
        """Evaluate the latest assistant response before appending a new user turn."""
        if msg.metadata.get(HEARTBEAT_METADATA_KEY):
            return

        last_response = None
        for message in reversed(session.messages):
            if message.get("role") == "assistant" and message.get("response_id"):
                last_response = message
                break
            if message.get("role") == "user":
                break

        if last_response is None:
            return

        response_id = last_response["response_id"]
        evaluation = evaluate_response_outcome(
            session.messages
            + [{"role": "user", "content": msg.content, "timestamp": msg.timestamp.isoformat()}],
            response_id,
            feedback_events=session.metadata.get("feedback_events", []),
            now=msg.timestamp,
        )
        if evaluation is None:
            return

        outcomes = session.metadata.setdefault("response_outcomes", {})
        previous = outcomes.get(response_id)
        if not should_update_outcome(previous, evaluation):
            return

        outcome_payload = evaluation.to_dict()
        outcomes[response_id] = outcome_payload
        LangfuseClient.get_instance().update_response_outcome(
            response_id,
            outcome_payload["outcome_label"],
            outcome_payload,
        )
        await self.bus.publish_outbound(
            OutboundMessage(
                session_key=msg.session_key,
                content="",
                event_type=OutboundEventType.RESPONSE_OUTCOME_EVALUATED,
                response_id=response_id,
                metadata={"response_outcome_evaluated": outcome_payload},
            )
        )

    def _get_channel_config(self, session_key: SessionKey):
        """Get channel config for a session key.

        Args:
            session_key: Session key to get channel config for

        Returns:
            Channel config object if found, None otherwise
        """
        return self.config.channels_config.get_channel_by_key(session_key.channel_key())

    def _get_ov_tools_enable(self, session_key: SessionKey) -> bool:
        """Get ov_tools_enable setting from channel config.

        Args:
            session_key: Session key to get channel config for

        Returns:
            True if ov tools should be enabled, False otherwise
        """
        channel_config = self._get_channel_config(session_key)
        return getattr(channel_config, "ov_tools_enable", True) if channel_config else True

    async def _process_system_message(self, msg: InboundMessage) -> OutboundMessage | None:
        """
        Process a system message (e.g., subagent announce).

        The chat_id field contains "original_channel:original_chat_id" to route
        the response back to the correct destination.
        """
        logger.info(f"Processing system message from {msg.sender_id}")

        session = self.sessions.get_or_create(msg.session_key)

        # Get channel config
        ov_tools_enable = self._get_ov_tools_enable(msg.session_key)
        profile_user_list = []
        channel_config = self._get_channel_config(msg.session_key)
        if channel_config and ov_tools_enable:
            profile_user_list = getattr(channel_config, "profile_user_list", [])

        # Build messages with the announce content
        provider_name = self.config.get_provider_name(self.model) if self.config else None
        messages = await self.context.build_messages(
            history=session.get_history(provider_name=provider_name),
            current_message=msg.content,
            session_key=msg.session_key,
            ov_tools_enable=ov_tools_enable,
            profile_user_list=profile_user_list,
        )

        # Run agent loop (no events published)
        (
            final_content,
            final_reasoning_content,
            tools_used,
            token_usage,
            iteration,
        ) = await self._run_agent_loop(
            messages=messages,
            session_key=msg.session_key,
            publish_events=False,
            ov_tools_enable=ov_tools_enable,
            memory_user_ids=None,
        )

        if final_content is None or (isinstance(final_content, str) and not final_content.strip()):
            final_content = "Background task completed."

        # Save to session (mark as system message in history)
        session.add_message("user", f"[System: {msg.sender_id}] {msg.content}")
        session.add_message(
            "assistant",
            final_content,
            tools_used=tools_used if tools_used else None,
            reasoning_content=final_reasoning_content,
        )
        await self.sessions.save(session)

        return OutboundMessage(session_key=msg.session_key, content=final_content)

    async def _consolidate_memory(self, session, archive_all: bool = False) -> None:
        """Consolidate old messages into MEMORY.md + HISTORY.md. Works on a cloned session."""
        try:
            if not session.messages:
                return

            # use openviking tools to extract memory
            config = self.config
            if config.mode == BotMode.READONLY:
                if not config.channels_config or not config.channels_config.get_all_channels():
                    return
                allow_from = [config.ov_server.admin_user_id]
                for channel_config in config.channels_config.get_all_channels():
                    if channel_config and channel_config.type.value == session.key.type:
                        if hasattr(channel_config, "allow_from"):
                            allow_from.extend(channel_config.allow_from)
                messages = [msg for msg in session.messages if msg.get("sender_id") in allow_from]
                session.messages = messages
            await self._consolidate_viking_memory(session)

            if self.sandbox_manager:
                memory_workspace = self.sandbox_manager.get_workspace_path(session.key)
            else:
                memory_workspace = self.workspace

            memory = MemoryStore(memory_workspace)
            if archive_all:
                old_messages = session.messages
                keep_count = 0
            else:
                keep_count = min(10, max(2, self.memory_window // 2))
                old_messages = session.messages[:-keep_count]
            if not old_messages:
                return
            logger.info(
                f"Memory consolidation started: {len(session.messages)} messages, archiving {len(old_messages)}, keeping {keep_count}"
            )

            # Format messages for LLM (include tool names when available)
            lines = []
            for m in old_messages:
                if not m.get("content"):
                    continue
                tools_used = m.get("tools_used", [])
                if tools_used and isinstance(tools_used, list):
                    tool_names = [
                        tc.get("tool_name", "unknown") for tc in tools_used if isinstance(tc, dict)
                    ]
                    tools_str = f" [tools: {', '.join(tool_names)}]" if tool_names else ""
                else:
                    tools_str = ""
                lines.append(
                    f"[{m.get('timestamp', '?')[:16]}] {m['role'].upper()}{tools_str}: {m['content']}"
                )
            conversation = "\n".join(lines)
            current_memory = memory.read_long_term()

            prompt = f"""You are a memory consolidation agent. Process this conversation and return a JSON object with exactly two keys:

1. "history_entry": A paragraph (2-5 sentences) summarizing the key events/decisions/topics. Start with a timestamp like [YYYY-MM-DD HH:MM]. Include enough detail to be useful when found by grep search later.

2. "memory_update": The updated long-term memory content. Add any new facts: user location, preferences, personal info, habits, project context, technical decisions, tools/services used. If nothing new, return the existing content unchanged.

## Current Long-term Memory
{current_memory or "(empty)"}

## Conversation to Process
{conversation}

Respond with ONLY valid JSON, no markdown fences."""

            response = await self.provider.chat(
                messages=[
                    {
                        "role": "system",
                        "content": "You are a memory consolidation agent. Respond only with valid JSON.",
                    },
                    {"role": "user", "content": prompt},
                ],
                model=self.model,
                session_id=session.key.safe_name(),
            )
            text = (response.content or "").strip()
            if text.startswith("```"):
                text = text.split("\n", 1)[-1].rsplit("```", 1)[0].strip()
            result = json.loads(text)

            if entry := result.get("history_entry"):
                memory.append_history(entry)
            if update := result.get("memory_update"):
                if load_config().use_local_memory and update != current_memory:
                    memory.write_long_term(update)

            # Session trimming and saving is handled by the caller before calling _consolidate_memory
            # This method works on a cloned session, so no need to save it
            logger.info("Memory consolidation done")
        except Exception as e:
            logger.exception(f"Memory consolidation failed: {e}")

    async def _consolidate_viking_memory(self, session) -> None:
        """Consolidate old messages into MEMORY.md + HISTORY.md. Works on a cloned session."""
        try:
            if not session.messages:
                logger.info(
                    f"No messages to commit openviking for session {session.key.safe_name()} (allow_from filter applied)"
                )
                return

            # use openviking tools to extract memory
            await hook_manager.execute_hooks(
                context=HookContext(
                    event_type="message.compact",
                    session_id=session.key.safe_name(),
                    workspace_id=self.sandbox_manager.to_workspace_id(session.key),
                    session_key=session.key,
                ),
                session=session,
            )
        except Exception as e:
            logger.exception(f"Memory consolidation failed: {e}")

    async def _safe_consolidate_memory(self, session, archive_all: bool = False) -> None:
        """Safe wrapper for _consolidate_memory that ensures all exceptions are caught."""
        try:
            await self._consolidate_memory(session, archive_all)
        except Exception as e:
            logger.exception(f"Background memory consolidation task failed: {e}")

    def _check_cmd_auth(self, msg: InboundMessage) -> bool:
        """Check if the session key is authorized for command execution.

        Returns:
            True if authorized, False otherwise.
        Args:
            session_key: Session key to check.
        """
        if self.config.mode == BotMode.NORMAL:
            return True
        allow_from = []
        if self.config.ov_server and self.config.ov_server.admin_user_id:
            allow_from.append(self.config.ov_server.admin_user_id)
        channel_config = self._get_channel_config(msg.session_key)
        if channel_config:
            allow_cmd = getattr(channel_config, "allow_cmd_from", [])
            if allow_cmd:
                allow_from.extend(allow_cmd)

        # If channel not found or sender not in allow_from list, ignore message
        if msg.sender_id not in allow_from:
            logger.debug(
                f"Sender {msg.sender_id} not allowed in channel {msg.session_key.channel_key()}"
            )
            return False
        return True

    async def process_direct(
        self,
        content: str,
        session_key: SessionKey = SessionKey(type="cli", channel_id="default", chat_id="direct"),
        metadata: dict[str, object] | None = None,
    ) -> str:
        """
        Process a message directly (for CLI or cron usage).

        Args:
            content: The message content.
            session_key: Session identifier (overrides channel:chat_id for session lookup).

        Returns:
            The agent's response.
        """
        await self._connect_mcp()
        msg = InboundMessage(
            session_key=session_key,
            sender_id="user",
            content=content,
            metadata=metadata or {},
        )

        response = await self._process_message(msg)
        return response.content if response else ""