"""
core/tool_dispatcher.py - 统一工具分发器
=========================================
[v1.22.0] 将 _execute_v2_tool() 中的所有 if/elif 硬编码分支
迁移为独立的 ToolDispatcher 类，实现:
  1. 统一入口 dispatch()，替代 ~350 行 if/elif 链
  2. 修复 _safe_sse bug（原代码在类方法中调用局部闭包，运行时会 NameError）
  3. 移除 code/code_* 和 command_run 工具（LLM 本身知道如何写代码/执行命令）
  4. [v1.22.0] 新增 image_ocr / image_analyze / audio_transcribe 内置工具
"""
from __future__ import annotations

import json
import re
from pathlib import Path as _P
from typing import Any, Callable, Dict, List, Optional

from core.logger import get_logger

logger = get_logger("myagent.tool_dispatcher")


class ToolDispatcher:
    """
    统一工具分发器。

    将原本分散在 MainAgent._execute_v2_tool() 中的 7 个 if/elif 分支
    整合为统一的 dispatch() 入口，每个工具对应一个独立的处理方法。

    用法:
        dispatcher = ToolDispatcher(
            executor=executor,
            memory_agent=memory_agent,
            skill_registry=skill_registry,
            permission_checker=(agent.check_permission, agent.name),
        )
        result = await dispatcher.dispatch(
            tool_name="file_read",
            params={"path": "/tmp/test.txt"},
            timeout=30,
            task_id="task_xxx",
            stream_callback=stream_cb,
            sent_files=[],
        )
    """

    def __init__(
        self,
        executor: Any = None,
        memory_agent: Any = None,
        skill_registry: Any = None,
        permission_checker: Optional[tuple] = None,
        llm_client: Any = None,
        memory_manager: Any = None,
        knowledge_base_dir: Optional[str] = None,
    ):
        """
        Args:
            executor: ExecutionEngine 实例（用于 command 工具）
            memory_agent: MemoryAgent 实例（用于 recall_memory 工具）
            skill_registry: SkillRegistry 实例（用于兜底工具分发）
            permission_checker: (checker_fn, agent_name) 元组，用于权限检查
            llm_client: LLMClient 实例（用于 image_analyze VLM 调用）
            memory_manager: MemoryManager 实例（用于 save_memory / recall_memory）
            knowledge_base_dir: 知识库目录（用于 save_knowledge / search_knowledge）
        """
        self.executor = executor
        self.memory_agent = memory_agent
        self.skills = skill_registry
        self._permission_checker = permission_checker
        self.llm = llm_client
        self.memory_manager = memory_manager
        self.knowledge_base_dir = knowledge_base_dir

    async def _emit_sse(
        self,
        event_type: str,
        data: Dict,
        stream_callback: Optional[Callable] = None,
    ) -> None:
        """发送 SSE 事件（替代原 _safe_sse 局部闭包，修复 NameError bug）"""
        if stream_callback is None:
            return
        import asyncio
        # [v1.23.25] 用 "data" 键包裹 payload，与 file_send.py 的格式一致
        # 前端 flow_engine.js 通过 evt.data 访问 payload
        event = {"type": event_type, "data": data}
        try:
            if asyncio.iscoroutinefunction(stream_callback):
                await stream_callback(event)
            else:
                stream_callback(event)
        except Exception as e:
            logger.debug(f"SSE 事件发送失败 ({event_type}): {e}")

    async def dispatch(
        self,
        tool_name: str,
        params: Dict[str, Any],
        timeout: int,
        task_id: str,
        stream_callback: Optional[Callable] = None,
        sent_files: Optional[List[Dict[str, Any]]] = None,
        agent_path: Optional[str] = None,
        agent_id: Optional[str] = None,
    ) -> Dict[str, Any]:
        """
        统一工具分发入口。

        Args:
            tool_name: 工具名称
            params: 已解析的参数字典
            timeout: 超时时间（秒）
            task_id: 任务 ID
            stream_callback: SSE 事件回调
            sent_files: 已发送文件追踪列表（file_send 时追加）
            agent_path: Agent 路径（传递给技能用于浏览器锁标识）

        Returns:
            {"success": bool, "output": str, "error": str, ...}
        """
        # ── 内置平台工具 (LLM 直接调用) ──
        if tool_name == "command":
            return await self._exec_command(params, timeout, task_id, stream_callback, sent_files, agent_id)
        elif tool_name == "web_control":
            return await self._exec_web_control(params, task_id, stream_callback)

        # ── [v1.38] 记忆与知识库工具（原 XML 标签转为原生工具调用） ──
        elif tool_name == "save_memory":
            return await self._exec_save_memory(params, task_id)
        elif tool_name == "recall_memory":
            return await self._exec_recall_memory(params, task_id)
        elif tool_name == "save_knowledge":
            return await self._exec_save_knowledge(params, task_id)
        elif tool_name == "search_knowledge":
            return await self._exec_search_knowledge(params, task_id)
        elif tool_name == "update_conversation_title":
            return await self._exec_update_title(params, task_id)

        # ── [v1.23.25] 恢复 playaudio/playvideo/file_send 直接工具调用 ──
        # 原因: CLI 迁移后 LLM 不稳定地使用 command 包裹，导致工具经常失败
        elif tool_name in ("playaudio", "playvideo"):
            return await self._exec_media(tool_name, params, task_id, stream_callback, sent_files)
        elif tool_name == "file_send":
            return await self._exec_file_send(params, task_id, stream_callback, sent_files)

        # ── [v1.23.0] 已迁移为 CLI 子命令的工具 — 提示使用 command 调用 ──
        elif tool_name in ("image_ocr", "ocr"):
            return {"success": False, "error": f"'{tool_name}' 已迁移为 CLI 命令，请使用: command {{\"command\": \"myagent-ai ocr <image_path> [ch|en]\"}}"}
        elif tool_name in ("image_analyze", "analyze_image"):
            return {"success": False, "error": f"'{tool_name}' 已迁移为 CLI 命令，请使用: command {{\"command\": \"myagent-ai analyze-image <image_path>\"}}"}
        elif tool_name in ("audio_transcribe", "transcribe"):
            return {"success": False, "error": f"'{tool_name}' 已迁移为 CLI 命令，请使用: command {{\"command\": \"myagent-ai transcribe <audio_path>\"}}"}

        # ── 兜底: SkillRegistry ──
        if self.skills:
            try:
                skill_result = await self.skills.execute(tool_name, _agent_path=agent_path or "", **params)
                result = {
                    "success": skill_result.success,
                    "output": skill_result.output or "",
                    "error": skill_result.error or "",
                    "message": skill_result.message or "",
                    "data": skill_result.data,
                    "files": skill_result.files,
                }
                # 自动 file_send: 如果 skill 产生了文件，尝试发送给用户
                if (skill_result.success
                        and skill_result.files
                        and stream_callback
                        and sent_files is not None):
                    await self._auto_send_skill_files(
                        skill_result.files, task_id, stream_callback, sent_files
                    )
                return result
            except Exception as e:
                return {"success": False, "error": f"技能执行异常: {tool_name} - {e}"}

        return {"success": False, "error": f"未知工具: {tool_name}"}

    # =========================================================================
    # 内置工具实现
    # =========================================================================

    async def _exec_command(self, params: Dict, timeout: int, task_id: str,
                              stream_callback: Optional[Callable] = None,
                              sent_files: Optional[List[Dict[str, Any]]] = None,
                              agent_id: Optional[str] = None) -> Dict:
        """执行 shell 命令"""
        code_text = params.get("command", "")
        if not code_text:
            return {"success": False, "error": "缺少 command 参数"}
        if not self.executor:
            return {"success": False, "error": "执行引擎未初始化"}
        
        # [v1.23.55] 检测 __CHAT_AGENT__ 标记并提取私聊信息
        import re as _re
        _chat_markers = []
        
        # 检查 sent_files 中是否有 chat_agent 类型的文件
        if sent_files:
            for sf in sent_files:
                if sf.get("_type") == "chat_agent":
                    pass  # 文件处理
        
        # 新格式(4段): __CHAT_AGENT__agent_path|agent_name|message|files__END__
        _chat_markers = _re.findall(r'__CHAT_AGENT__(.+?)\|(.+?)\|(.+?)\|(.+?)__END__', code_text)
        if _chat_markers:
            # 清理标记文本
            code_text = _re.sub(r'__CHAT_AGENT__.+?__END__\n?', '', code_text).strip()
        else:
            # 兼容旧格式(3段): __CHAT_AGENT__agent_path|agent_name|message__END__
            _chat_markers_old = _re.findall(r'__CHAT_AGENT__(.+?)\|(.+?)\|(.+?)__END__', code_text)
            if _chat_markers_old:
                _chat_markers = _chat_markers_old
                code_text = _re.sub(r'__CHAT_AGENT__.+?__END__\n?', '', code_text).strip()

        # 如果有私聊标记，保存到数据库
        if _chat_markers:
            try:
                logger.info(f"[{task_id}] [私聊保存] 开始处理，找到 {len(_chat_markers)} 条私聊标记")
                # 延迟导入避免循环依赖
                from groups.manager import GroupManager
                # 获取数据目录
                _data_dir = _P.home() / ".myagent" / "data"
                logger.info(f"[{task_id}] [私聊保存] 数据目录: {_data_dir}")
                _gm = GroupManager(_data_dir)
                logger.info(f"[{task_id}] [私聊保存] GroupManager 初始化完成")
                
                for _idx, _match in enumerate(_chat_markers):
                    _to_path = _match[0].strip()
                    _to_name = _match[1].strip()
                    _content = _match[2].strip()
                    
                    # 获取当前 agent 的 ID
                    _from_id = agent_id if agent_id else "unknown"
                    _from_name = _from_id if _from_id else "未知"
                    
                    logger.info(f"[{task_id}] [私聊保存] 处理第 {_idx+1} 条: from_id={_from_id}, from_name={_from_name}, to_path={_to_path}, to_name={_to_name}")
                    
                    _result = _gm.add_agent_chat(
                        group_id="",
                        from_agent=_from_id,
                        from_name=_from_name,
                        to_agent=_to_path,
                        to_name=_to_name,
                        content=_content,
                    )
                    logger.info(f"[{task_id}] [私聊保存] 第 {_idx+1} 条保存结果: msg_id={_result}")
                    logger.info(f"[{task_id}] 私聊记录已保存: {_from_name} → {_to_name}: {_content[:50]}...")
                
                _gm.close()
                logger.info(f"[{task_id}] [私聊保存] GroupManager 已关闭，处理完成")
            except Exception as _ce:
                logger.warning(f"[{task_id}] [私聊保存] 保存私聊记录失败: {_ce}")
                logger.exception(f"[{task_id}] [私聊保存] 异常详情")

        # 注入权限检查器
        if self._permission_checker:
            self.executor.set_permission_checker(*self._permission_checker)
        exec_result = await self.executor.execute(
            language="shell", code=code_text, timeout=timeout,
        )
        result = exec_result.to_dict()

        output = result.get("stdout", "") or result.get("output", "")
        import re as _re

        # [v1.23.55] 检测命令输出中的 __CHAT_AGENT__ 标记（CLI chat 命令输出此标记）
        # 新格式(4段): __CHAT_AGENT__agent_path|agent_name|message|files__END__
        _chat_markers_out = _re.findall(r'__CHAT_AGENT__(.+?)\|(.+?)\|(.+?)\|(.+?)__END__', output)
        if _chat_markers_out:
            # 清理标记文本
            output = _re.sub(r'__CHAT_AGENT__.+?__END__\n?', '', output).strip()
            result["output"] = output
        else:
            # 兼容旧格式(3段): __CHAT_AGENT__agent_path|agent_name|message__END__
            _chat_markers_old = _re.findall(r'__CHAT_AGENT__(.+?)\|(.+?)\|(.+?)__END__', output)
            if _chat_markers_old:
                _chat_markers_out = _chat_markers_old
                output = _re.sub(r'__CHAT_AGENT__.+?__END__\n?', '', output).strip()
                result["output"] = output

        # 如果输出中有私聊标记，保存到数据库
        if _chat_markers_out:
            try:
                # 延迟导入避免循环依赖
                from groups.manager import GroupManager
                # 获取数据目录
                _data_dir = _P.home() / ".myagent" / "data"
                _gm = GroupManager(_data_dir)
                
                for _match in _chat_markers_out:
                    _to_path = _match[0].strip()
                    _to_name = _match[1].strip()
                    _content = _match[2].strip()
                    
                    # 获取当前 agent 的 ID
                    _from_id = agent_id if agent_id else "unknown"
                    _from_name = _from_id if _from_id else "未知"
                    
                    _gm.add_agent_chat(
                        group_id="",
                        from_agent=_from_id,
                        from_name=_from_name,
                        to_agent=_to_path,
                        to_name=_to_name,
                        content=_content,
                    )
                    logger.info(f"[{task_id}] 私聊记录已保存: {_from_name} → {_to_name}: {_content[:50]}...")
                
                _gm.close()
            except Exception as _ce:
                logger.warning(f"[{task_id}] 保存私聊记录失败: {_ce}")

        # [v1.23.0] 检测 __SEND_FILE__ 标记 — CLI send-file 命令输出此标记
        # 格式: __SEND_FILE__绝对路径|描述__END__
        send_markers = _re.findall(r'__SEND_FILE__(.+?)\|(.+?)__END__', output)
        if send_markers:
            clean_output = _re.sub(r'__SEND_FILE__.+?__END__\n?', '', output).strip()
            result["output"] = clean_output
            for send_path, send_desc in send_markers:
                send_path = send_path.strip()
                send_desc = send_desc.strip()
                try:
                    p = _P(send_path)
                    if p.exists():
                        # [v1.23.29] 直接通过 _exec_file_send 发送（统一入口，确保 v2_file 推送）
                        file_result = await self._exec_file_send(
                            {"file_path": send_path, "description": send_desc},
                            task_id, stream_callback, sent_files,
                        )
                        if not file_result.get("success"):
                            result["output"] += f"\n[文件发送失败: {file_result.get('error', '')}]"
                except Exception as e:
                    logger.warning(f"[{task_id}] CLI 文件发送异常: {e}")
                    result["output"] += f"\n[文件发送异常: {e}]"
        else:
            clean_output = output

        # [v1.23.0] 检测 __EMBED_AUDIO__ / __EMBED_VIDEO__ 标记 — CLI playaudio/playvideo 输出
        # 格式: __EMBED_AUDIO__URL|标题__END__ 或 __EMBED_VIDEO__URL|标题__END__
        audio_markers = _re.findall(r'__EMBED_AUDIO__(.+?)\|(.+?)__END__', clean_output)
        video_markers = _re.findall(r'__EMBED_VIDEO__(.+?)\|(.+?)__END__', clean_output)
        if audio_markers:
            clean_output = _re.sub(r'__EMBED_AUDIO__.+?__END__\n?', '', clean_output).strip()
            result["output"] = clean_output
            for media_url, media_title in audio_markers:
                media_result = await self._exec_media(
                    "playaudio", {"url": media_url.strip(), "title": media_title.strip()},
                    task_id, stream_callback, sent_files,
                )
                if not media_result.get("success"):
                    result["output"] += f"\n[音频播放失败: {media_result.get('error', '')}]"
        if video_markers:
            # [v1.23.32] 修复: result 可能没有 output key（来自 to_dict 的 stdout）
            clean_output = result.get("output", "") or clean_output
            clean_output = _re.sub(r'__EMBED_VIDEO__.+?__END__\n?', '', clean_output).strip()
            result["output"] = clean_output
            for media_url, media_title in video_markers:
                media_result = await self._exec_media(
                    "playvideo", {"url": media_url.strip(), "title": media_title.strip()},
                    task_id, stream_callback, sent_files,
                )
                if not media_result.get("success"):
                    result["output"] += f"\n[视频播放失败: {media_result.get('error', '')}]"
        # [v1.23.55] 检测 __CHAT_AGENT__ 标记 — CLI chat 命令输出此标记
        # 新格式(4段): __CHAT_AGENT__agent_path|agent_name|message|files__END__
        # 旧格式(3段): __CHAT_AGENT__agent_path|agent_name|message__END__
        # files 为空时表示纯文本消息，非空时为 | 分隔的文件绝对路径列表
        chat_markers = _re.findall(r'__CHAT_AGENT__(.+?)\|(.+?)\|(.+?)\|(.+?)__END__', clean_output)
        if chat_markers:
            clean_output = _re.sub(r'__CHAT_AGENT__.+?__END__\n?', '', clean_output).strip()
            result["output"] = clean_output
            for chat_agent_path, chat_agent_name, chat_msg, chat_files in chat_markers:
                chat_files = chat_files.strip()
                file_list = [f.strip() for f in chat_files.split("|") if f.strip()] if chat_files else []
                logger.info(f"[{task_id}] Agent私聊: → {chat_agent_name.strip()} ({chat_agent_path.strip()}), 内容: {chat_msg.strip()[:100]}")
                # Store as a chat agent event in sent_files for persistence
                if sent_files is not None:
                    chat_entry = {
                        "_type": "chat_agent",
                        "target_agent": chat_agent_path.strip(),
                        "target_name": chat_agent_name.strip(),
                        "message": chat_msg.strip(),
                    }
                    if file_list:
                        chat_entry["files"] = file_list
                    sent_files.append(chat_entry)
                # Emit SSE event for frontend display
                try:
                    event_data = {
                        "target_agent": chat_agent_path.strip(),
                        "target_name": chat_agent_name.strip(),
                        "message": chat_msg.strip(),
                    }
                    if file_list:
                        event_data["files"] = file_list
                    await self._emit_sse("v2_chat_agent", event_data, stream_callback)
                except Exception:
                    pass
            # 返回人类可读的确认信息
            result["output"] = (result.get("output", "") + "\n[已向 " + chat_markers[0][1].strip() + " 发送私聊消息]").strip()
        else:
            # 兼容旧格式(3段): __CHAT_AGENT__agent_path|agent_name|message__END__
            chat_markers_old = _re.findall(r'__CHAT_AGENT__(.+?)\|(.+?)\|(.+?)__END__', clean_output)
            if chat_markers_old:
                clean_output = _re.sub(r'__CHAT_AGENT__.+?__END__\n?', '', clean_output).strip()
                result["output"] = clean_output
                for chat_agent_path, chat_agent_name, chat_msg in chat_markers_old:
                    logger.info(f"[{task_id}] Agent私聊: → {chat_agent_name.strip()} ({chat_agent_path.strip()}), 内容: {chat_msg.strip()[:100]}")
                    if sent_files is not None:
                        sent_files.append({
                            "_type": "chat_agent",
                            "target_agent": chat_agent_path.strip(),
                            "target_name": chat_agent_name.strip(),
                            "message": chat_msg.strip(),
                        })
                    try:
                        await self._emit_sse("v2_chat_agent", {
                            "target_agent": chat_agent_path.strip(),
                            "target_name": chat_agent_name.strip(),
                            "message": chat_msg.strip(),
                        }, stream_callback)
                    except Exception:
                        pass
                # 返回人类可读的确认信息
                result["output"] = (result.get("output", "") + "\n[已向 " + chat_markers_old[0][1].strip() + " 发送私聊消息]").strip()

        # [v1.23.32] 确保 result 始终有 output 字段（to_dict 返回 stdout，V2 循环依赖 output）
        if "output" not in result:
            result["output"] = clean_output

        return result

    async def _exec_recall_memory(self, params: Dict, task_id: str) -> Dict:
        """主动召回记忆"""
        if not self.memory_agent:
            return {"success": False, "error": "记忆系统未初始化"}
        try:
            recall_results = await self.memory_agent.recall_memory(
                keyword=params.get("keyword", ""),
                time_point=params.get("time_point", ""),
                session_id=params.get("session_id", ""),
                limit=params.get("limit", 5),
            )
            if recall_results:
                output_lines = [f"找到 {len(recall_results)} 条相关记忆:"]
                for i, mem in enumerate(recall_results, 1):
                    output_lines.append(
                        f"{i}. [{mem.get('created_at', '')}] "
                        f"[{mem.get('category', '')}] "
                        f"{mem.get('content', '')}"
                    )
                return {"success": True, "output": "\n".join(output_lines), "data": recall_results}
            else:
                return {"success": True, "output": "未找到相关记忆", "data": []}
        except Exception as e:
            return {"success": False, "error": f"记忆召回失败: {e}"}

    async def _exec_file_send(
        self, params: Dict, task_id: str,
        stream_callback: Optional[Callable] = None,
        sent_files: Optional[List[Dict]] = None,
    ) -> Dict:
        """发送文件给用户 — 后端推送 v2_file SSE 事件 + 持久化到聊天记录

        包含去重机制：同一会话中相同路径的文件不会重复发送。
        """
        try:
            from aiskills.file_send import UPLOADS_DIR
            fpath = params.get("file_path", "")
            fdesc = params.get("description", "")
            if not fpath:
                logger.warning(f"[{task_id}] file_send: 缺少 file_path 参数")
                return {"success": False, "error": "缺少 file_path 参数，请提供要发送的文件路径"}

            # [v1.38.1] 去重检查：同一文件路径不重复发送
            # 场景：技能自动发送 + LLM 再次调用 file_send → 避免文件卡片出现两次
            fpath_normalized = _P(fpath.strip().strip("'\"")).resolve()
            if sent_files is not None:
                for _sent in sent_files:
                    # 通过原始路径比对（sent_files 中记录的是 resolved 路径）
                    try:
                        sent_name = _sent.get("name", "")
                        sent_path = _sent.get("_source_path", "")
                        # 方法1: 比较 source_path
                        if sent_path and _P(sent_path).resolve() == fpath_normalized:
                            logger.info(f"[{task_id}] file_send: 文件已发送过，跳过: {fpath}")
                            return {
                                "success": True,
                                "output": f"文件已发送过，跳过重复发送: {_sent.get('name', fpath)}",
                                "data": _sent,
                                "skipped": True,
                            }
                    except Exception:
                        pass

            logger.info(f"[{task_id}] file_send: 发送文件 {fpath}")

            # [v1.23.35] 先复制文件（不依赖 file_send.execute 的 SSE 发送）
            # _P 已在模块顶部导入 (from pathlib import Path as _P)，此处不再重复导入
            # 以避免 Python 将 _P 视为局部变量导致去重代码处 UnboundLocalError
            import shutil, uuid as _uuid, time as _time
            fpath_resolved = _P(fpath.strip().strip("'\"")).expanduser()
            if not fpath_resolved.exists():
                return {"success": False, "error": f"文件不存在: {fpath}"}
            if not fpath_resolved.is_file():
                return {"success": False, "error": f"不是文件: {fpath}"}

            file_id = str(_uuid.uuid4())[:12]
            # [v1.23.35] 直接使用模块级 UPLOADS_DIR，不依赖 FileSendSkill 实例属性
            date_dir = UPLOADS_DIR / _time.strftime("%Y-%m")
            date_dir.mkdir(parents=True, exist_ok=True)
            stored_name = f"{file_id}_{fpath_resolved.name}"
            stored_path = date_dir / stored_name
            shutil.copy2(str(fpath_resolved), str(stored_path))

            mime_map = {
                ".pdf": "application/pdf", ".png": "image/png", ".jpg": "image/jpeg",
                ".jpeg": "image/jpeg", ".gif": "image/gif", ".webp": "image/webp",
                ".docx": "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
                ".xlsx": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
                ".pptx": "application/vnd.openxmlformats-officedocument.presentationml.presentation",
                ".txt": "text/plain", ".csv": "text/csv", ".md": "text/markdown",
                ".json": "application/json", ".html": "text/html",
                ".mp3": "audio/mpeg", ".mp4": "video/mp4", ".wav": "audio/wav",
                ".zip": "application/zip", ".tar.gz": "application/gzip",
            }
            mime = mime_map.get(fpath_resolved.suffix.lower(), "application/octet-stream")
            size = stored_path.stat().st_size

            file_data = {
                "id": file_id,
                "file_id": file_id,
                "name": fpath_resolved.name,
                "type": mime,
                "size": size,
                "description": fdesc or f"文件: {fpath_resolved.name}",
                "url": f"/api/file/{file_id}?name={fpath_resolved.name}",
                "download_url": f"/api/file/{file_id}/download?name={fpath_resolved.name}",
            }

            # [v1.23.29] 关键：通过 _emit_sse 后端推送 v2_file 事件到前端
            # 这是文件卡片显示的核心机制 — 不依赖 file_send.execute 内部的 SSE 发送
            await self._emit_sse("v2_file", file_data, stream_callback)
            logger.info(f"[{task_id}] file_send: v2_file 已推送 → {file_id} ({fpath_resolved.name})")

            # 持久化到 sent_files（写入聊天记录数据库）
            # [v1.38.1] 同时记录 _source_path 用于去重比对
            if sent_files is not None:
                sent_files.append({
                    "id": file_id,
                    "file_id": file_id,
                    "name": fpath_resolved.name,
                    "type": mime,
                    "size": size,
                    "description": fdesc or f"文件: {fpath_resolved.name}",
                    "url": file_data["url"],
                    "download_url": file_data["download_url"],
                    "_source_path": str(fpath_resolved),  # 原始路径，用于去重
                })

            return {
                "success": True,
                "output": f"文件已发送: {fpath_resolved.name} (ID: {file_id}, 大小: {size} bytes)",
                "data": file_data,
            }
        except Exception as e:
            logger.error(f"[{task_id}] file_send: 异常 - {e}", exc_info=True)
            return {"success": False, "error": f"文件发送失败: {e}"}

    async def _exec_media(
        self, tool_name: str, params: Dict, task_id: str,
        stream_callback: Optional[Callable] = None,
        sent_files: Optional[List[Dict]] = None,
    ) -> Dict:
        """播放音频/视频 — 内嵌播放器"""
        media_url = params.get("url", "").strip()
        media_file = params.get("file_path", "").strip()
        media_type = "audio" if tool_name == "playaudio" else "video"
        embed_url = None
        embed_title = params.get("title", "")
        fallback_link = None

        if not media_url and not media_file:
            logger.warning(f"[{task_id}] {tool_name}: 缺少 url 和 file_path 参数")
            return {"success": False, "error": f"请提供在线链接(url)或本地文件路径(file_path)"}

        if media_url:
            url_lower = media_url.lower()
            # YouTube
            yt_match = re.search(r'(?:youtube\.com/watch\?v=|youtu\.be/|youtube\.com/embed/)([\w-]+)', media_url)
            if yt_match:
                embed_url = f"https://www.youtube.com/embed/{yt_match.group(1)}"
                embed_title = embed_title or "YouTube 视频"
            # YouTube Music
            elif 'music.youtube.com' in url_lower:
                ym_match = re.search(r'list=([\w-]+)', media_url)
                if ym_match:
                    embed_url = f"https://music.youtube.com/embed?list={ym_match.group(1)}&layout=full"
                    embed_title = embed_title or "YouTube Music"
                else:
                    ymv_match = re.search(r'watch\?v=([\w-]+)', media_url)
                    if ymv_match:
                        embed_url = f"https://music.youtube.com/embed/{ymv_match.group(1)}"
                        embed_title = embed_title or "YouTube Music"
                    else:
                        fallback_link = media_url
                        embed_title = embed_title or "YouTube Music"
            # Bilibili
            elif 'bilibili.com' in url_lower or 'b23.tv' in url_lower:
                bv_match = re.search(r'bilibili\.com/video/(BV[\w]+)', media_url)
                if bv_match:
                    embed_url = f"https://player.bilibili.com/player.html?bvid={bv_match.group(1)}&autoplay=0"
                    embed_title = embed_title or "B站视频"
                else:
                    embed_url = media_url
                    embed_title = embed_title or "B站视频"
            # QQ音乐
            elif 'y.qq.com' in url_lower:
                qq_match = re.search(r'songDetail/(\w+)', media_url)
                if qq_match:
                    embed_url = f"https://y.qq.com/n/ryqq/songDetail/{qq_match.group(1)}"
                    embed_title = embed_title or "QQ音乐"
                else:
                    embed_url = media_url
                    embed_title = embed_title or "QQ音乐"
            # 网易云音乐
            elif 'music.163.com' in url_lower:
                song_match = re.search(r'music\.163\.com.*[?&]id=(\d+)', media_url)
                if song_match:
                    embed_url = f"https://music.163.com/outchain/player?type=2&id={song_match.group(1)}&auto=0&height=66"
                    embed_title = embed_title or "网易云音乐"
                else:
                    embed_url = media_url
                    embed_title = embed_title or "网易云音乐"
            # 抖音
            elif 'douyin.com' in url_lower:
                embed_url = media_url
                embed_title = embed_title or "抖音视频"
            else:
                embed_url = media_url
                embed_title = embed_title or ("在线音乐" if media_type == "audio" else "在线视频")

        # [v1.23.19] 构建媒体元数据，持久化到 sent_files 以支持历史消息恢复
        _media_meta = {
            "_type": "media",  # 标记为媒体类型（区别于文件）
            "media_type": media_type,
            "embed_url": embed_url or "",
            "title": embed_title,
            "original_url": media_url or fallback_link or "",
        }
        if sent_files is not None:
            sent_files.append(_media_meta)

        if embed_url and stream_callback:
            logger.info(f"[{task_id}] {tool_name}: 嵌入播放器 {embed_title} → {embed_url[:80]}")
            await self._emit_sse("v2_media", {
                "media_type": media_type,
                "embed_url": embed_url,
                "title": embed_title,
                "original_url": media_url,
            }, stream_callback)
            return {"success": True, "output": f"已嵌入{embed_title}播放器: {media_url}"}

        elif embed_url and not stream_callback:
            # [v1.23.26] stream_callback 缺失时仍返回成功（非 Web 环境降级）
            logger.warning(f"[{task_id}] {tool_name}: stream_callback 缺失，无法发送播放器到前端")
            return {"success": True, "output": f"已生成{embed_title}播放器（非 Web 环境无法显示）: {embed_url}"}

        elif fallback_link and stream_callback:
            await self._emit_sse("v2_media", {
                "media_type": media_type,
                "embed_url": "",
                "title": embed_title,
                "original_url": fallback_link,
            }, stream_callback)
            return {"success": True, "output": f"已提供{embed_title}链接（不支持嵌入播放）: {fallback_link}"}

        elif media_file:
            fpath = _P(media_file).expanduser().resolve()
            if not fpath.exists():
                return {"success": False, "error": f"文件不存在: {media_file}"}
            # [v1.23.32] 使用统一入口 _exec_file_send（正确的 MIME + v2_file 推送 + sent_files 持久化）
            desc = f"{'音频' if media_type == 'audio' else '视频'}播放: {fpath.name}"
            fresult = await self._exec_file_send(
                {"file_path": str(fpath), "description": desc},
                task_id, stream_callback, sent_files,
            )
            if fresult.get("success"):
                return {"success": True, "output": f"已发送{media_type}文件: {fpath.name}", "data": fresult.get("data", {})}
            else:
                return {"success": False, "error": fresult.get("error", "文件发送失败")}
        else:
            return {"success": False, "error": "请提供 url（在线链接）或 file_path（本地文件路径）参数"}

    async def _exec_web_control(
        self, params: Dict, task_id: str,
        stream_callback: Optional[Callable] = None,
    ) -> Dict:
        """网页控制器 — 浏览器面板"""
        try:
            from core.web_control import get_web_control_manager, LOGIN_URLS
            wc_mgr = get_web_control_manager()
            action = params.get("action", "open")
            session_id = params.get("session_id", "").strip()

            # 自动获取或创建会话
            session = None
            if session_id:
                session = wc_mgr.get_session(session_id)
            if not session:
                session = wc_mgr.create_session()
                session_id = session.session_id

            if action == "open":
                url = params.get("url", "").strip()
                if url:
                    session.current_url = url
                if stream_callback:
                    await self._emit_sse("v2_web_control", {
                        "action": "open",
                        "sid": session_id,
                        "url": url,
                        "panel_url": f"/api/web_control/panel?sid={session_id}",
                    }, stream_callback)
                return {
                    "success": True,
                    "output": f"已打开网页控制面板 (session: {session_id})" + (f"，URL: {url}" if url else ""),
                    "sid": session_id,
                }

            elif action == "close":
                wc_mgr.close_session(session_id)
                if stream_callback:
                    await self._emit_sse("v2_web_control", {
                        "action": "close", "sid": session_id
                    }, stream_callback)
                return {"success": True, "output": f"已关闭网页控制面板 (session: {session_id})"}

            elif action == "navigate":
                url = params.get("url", "").strip()
                if not url:
                    return {"success": False, "error": "请提供 url 参数"}
                session.current_url = url
                if stream_callback:
                    await self._emit_sse("v2_web_control", {
                        "action": "navigate", "sid": session_id, "url": url
                    }, stream_callback)
                return {"success": True, "output": f"正在导航到: {url}"}

            elif action in ("set_cookies", "get_cookies"):
                if action == "set_cookies":
                    cookies = params.get("cookies", [])
                    if isinstance(cookies, str):
                        try:
                            cookies = json.loads(cookies)
                        except Exception:
                            cookies = []
                    session.set_cookies(cookies)
                    return {"success": True, "output": f"已设置 {len(cookies)} 个 Cookie"}
                else:
                    cookies = session.get_cookies()
                    return {"success": True, "output": json.dumps(cookies, ensure_ascii=False, indent=2), "data": cookies}

            elif action in ("get_content", "click", "fill", "scroll", "evaluate", "wait", "screenshot"):
                # 这些 action 通过命令队列由浏览器面板执行
                cmd_result = await wc_mgr.queue_command(session_id, action, params, timeout=30)
                return {
                    "success": cmd_result.get("success", False),
                    "output": cmd_result.get("output", cmd_result.get("result", "")),
                    "error": cmd_result.get("error", ""),
                    "data": cmd_result.get("data"),
                }

            elif action == "human_interact":
                # [v1.25.3] 人机交互模式 — Agent 暂停, 用户手动操作
                prompt = params.get("prompt", "请在上方网页中完成登录或验证操作")
                platform = params.get("platform", "")
                timeout = int(params.get("timeout", 0)) or 300  # 默认5分钟
                auto_save = params.get("auto_save_cookies", True)

                # 1. 向面板发送 human_interact 命令（切换到人机模式）
                session.human_mode = True
                cmd_result = await wc_mgr.queue_command(
                    session_id, "human_interact",
                    {"prompt": prompt, "platform": platform},
                    timeout=15
                )

                # 2. 通知前端显示人机交互 UI
                if stream_callback:
                    await self._emit_sse("v2_web_control", {
                        "action": "human_interact",
                        "sid": session_id,
                        "prompt": prompt,
                        "platform": platform,
                        "timeout": timeout,
                    }, stream_callback)

                # 3. 创建 human_event 并等待用户完成
                loop = asyncio.get_event_loop()
                session.human_event = asyncio.Event()
                session.human_result = None

                try:
                    await asyncio.wait_for(session.human_event.wait(), timeout=timeout)
                except asyncio.TimeoutError:
                    session.human_mode = False
                    # 恢复 agent 模式
                    try:
                        await wc_mgr.queue_command(session_id, "agent_mode", {}, timeout=5)
                    except Exception:
                        pass
                    if stream_callback:
                        await self._emit_sse("v2_web_control", {
                            "action": "human_done",
                            "sid": session_id,
                            "timed_out": True,
                        }, stream_callback)
                    return {"success": False, "error": f"人机交互超时 ({timeout}s)"}

                # 4. 用户完成, 恢复 agent 模式
                session.human_mode = False
                try:
                    await wc_mgr.queue_command(session_id, "agent_mode", {}, timeout=5)
                except Exception:
                    pass

                # 5. 自动保存 cookies
                cookie_file = ""
                if auto_save:
                    try:
                        cookie_file = session.save_cookies_to_file()
                    except Exception as e:
                        cookie_file = f"(保存失败: {e})"

                # 6. 获取当前页面信息
                page_info = ""
                try:
                    info_result = await wc_mgr.queue_command(session_id, "get_content", {"what": "url,title"}, timeout=10)
                    if info_result.get("success"):
                        page_info = info_result.get("result", "")
                except Exception:
                    pass

                result_data = session.human_result or {}
                if stream_callback:
                    await self._emit_sse("v2_web_control", {
                        "action": "human_done",
                        "sid": session_id,
                    }, stream_callback)

                output_parts = [f"用户已完成人机交互操作"]
                if page_info:
                    output_parts.append(f"当前页面: {page_info}")
                if cookie_file:
                    output_parts.append(f"Cookies 已保存: {cookie_file}")
                if result_data.get("note"):
                    output_parts.append(f"用户备注: {result_data['note']}")

                return {
                    "success": True,
                    "output": "\n".join(output_parts),
                    "sid": session_id,
                    "cookies_saved": bool(cookie_file) and not cookie_file.startswith("("),
                    "cookie_file": cookie_file,
                    "data": result_data,
                }

            elif action == "login":
                # [v1.25.4] 一站式登录流程: 导航 → 人机交互 → 自动保存凭证
                platform = params.get("platform", "").strip().lower()
                login_url = params.get("url", "").strip()
                prompt = params.get("prompt", "")
                timeout = int(params.get("timeout", 0)) or 300
                label = params.get("label", "")

                # 查找平台登录 URL
                if not login_url and platform:
                    login_url = LOGIN_URLS.get(platform, "")
                    if not login_url:
                        return {"success": False, "error": f"未知平台 '{platform}'，未找到预置登录 URL，请通过 url 参数指定登录页地址"}

                if not login_url:
                    return {"success": False, "error": "请提供 platform (平台名称) 或 url (登录页地址) 参数"}

                # 构建默认提示
                platform_names = {
                    "qq": "QQ", "qq_mail": "QQ邮箱", "wechat_work": "企业微信",
                    "wechat_mp": "微信公众号", "telegram": "Telegram", "telegram_web": "Telegram Web",
                    "discord": "Discord", "feishu": "飞书", "feishu_admin": "飞书管理后台",
                    "lark": "Lark", "dingtalk": "钉钉", "github": "GitHub",
                    "google": "Google", "bilibili": "B站", "taobao": "淘宝",
                    "zhihu": "知乎", "weibo": "微博",
                }
                display_name = platform_names.get(platform, platform) if platform else ""
                if not prompt:
                    prompt = f"请在上方页面完成 {display_name} 登录操作" if display_name else "请在上方页面完成登录操作"
                    # 添加平台特定提示
                    if platform in ("qq", "wechat_mp", "telegram", "discord", "dingtalk"):
                        prompt += "（可能需要扫码或输入账号密码）"
                    elif platform in ("wechat_work", "feishu", "feishu_admin", "lark"):
                        prompt += "（请使用管理员账号扫码登录）"

                # Step 1: 打开面板
                session.current_url = login_url
                if stream_callback:
                    await self._emit_sse("v2_web_control", {
                        "action": "login",
                        "sid": session_id,
                        "url": login_url,
                        "panel_url": f"/api/web_control/panel?sid={session_id}",
                        "platform": platform,
                        "prompt": prompt,
                        "timeout": timeout,
                    }, stream_callback)

                # Step 2: 等待面板打开
                if not session.is_panel_open:
                    waited = 0
                    while waited < 15:
                        await asyncio.sleep(1)
                        waited += 1
                        if session.is_panel_open:
                            break
                        if session._closed:
                            return {"success": False, "error": "会话已关闭"}

                # Step 3: 切换人机交互模式
                session.human_mode = True
                try:
                    await wc_mgr.queue_command(
                        session_id, "human_interact",
                        {"prompt": prompt, "platform": display_name},
                        timeout=15
                    )
                except Exception as e:
                    logger.warning(f"[WebControl] human_interact 命令发送失败: {e}")

                # Step 4: 等待用户完成
                loop = asyncio.get_event_loop()
                session.human_event = asyncio.Event()
                session.human_result = None

                try:
                    await asyncio.wait_for(session.human_event.wait(), timeout=timeout)
                except asyncio.TimeoutError:
                    session.human_mode = False
                    try:
                        await wc_mgr.queue_command(session_id, "agent_mode", {}, timeout=5)
                    except Exception:
                        pass
                    if stream_callback:
                        await self._emit_sse("v2_web_control", {
                            "action": "human_done", "sid": session_id, "timed_out": True,
                        }, stream_callback)
                    return {"success": False, "error": f"登录超时 ({timeout}s)，请重试"}

                # Step 5: 用户完成，恢复 agent 模式
                session.human_mode = False
                try:
                    await wc_mgr.queue_command(session_id, "agent_mode", {}, timeout=5)
                except Exception:
                    pass

                # Step 6: 获取当前页面信息
                page_info = ""
                try:
                    info_result = await wc_mgr.queue_command(session_id, "get_content", {"what": "url,title"}, timeout=10)
                    if info_result.get("success"):
                        page_info = info_result.get("result", "")
                except Exception:
                    pass

                # Step 7: 自动保存 cookies + 凭证
                cookie_file = ""
                cred_file = ""
                result_data = session.human_result or {}

                try:
                    cookie_file = session.save_cookies_to_file()
                except Exception as e:
                    cookie_file = f"(保存失败: {e})"

                # 保存到凭证库
                if platform:
                    try:
                        extra = {"note": result_data.get("note", ""), "login_url": login_url}
                        cred_file = session.save_credentials_to_file(platform, label=label, extra=extra)
                    except Exception as e:
                        cred_file = f"(保存失败: {e})"

                # 检测是否登录成功 (URL 发生了变化)
                login_success = False
                if page_info and login_url:
                    current_url = page_info.split(",")[0].strip() if "," in page_info else page_info
                    login_success = (current_url != login_url and "login" not in current_url.lower())

                if stream_callback:
                    await self._emit_sse("v2_web_control", {
                        "action": "login_done",
                        "sid": session_id,
                        "platform": platform,
                        "login_success": login_success,
                    }, stream_callback)

                output_parts = []
                if login_success:
                    output_parts.append(f"{display_name} 登录成功")
                else:
                    output_parts.append(f"用户已完成 {display_name} 登录操作")
                if page_info:
                    output_parts.append(f"当前页面: {page_info}")
                if cookie_file and not cookie_file.startswith("("):
                    output_parts.append(f"Cookies 已保存: {cookie_file}")
                if cred_file and not cred_file.startswith("("):
                    output_parts.append(f"凭证已保存: {cred_file}")
                if result_data.get("note"):
                    output_parts.append(f"用户备注: {result_data['note']}")

                return {
                    "success": True,
                    "output": "\n".join(output_parts),
                    "sid": session_id,
                    "login_success": login_success,
                    "platform": platform,
                    "cookie_file": cookie_file,
                    "credential_file": cred_file,
                    "data": result_data,
                }

            elif action == "save_credentials":
                # [v1.25.4] 保存凭证到凭证库
                platform = params.get("platform", "").strip()
                label = params.get("label", "").strip()
                if not platform:
                    return {"success": False, "error": "请提供 platform 参数 (平台名称)"}
                try:
                    filepath = session.save_credentials_to_file(platform, label=label)
                    return {"success": True, "output": f"凭证已保存到: {filepath}", "credential_file": filepath}
                except Exception as e:
                    return {"success": False, "error": f"保存凭证失败: {e}"}

            elif action == "list_credentials":
                # [v1.25.4] 列出所有已保存的凭证
                creds = WebControlSession.list_credentials()
                if not creds:
                    return {"success": True, "output": "暂无已保存的登录凭证", "data": []}
                lines = [f"已保存 {len(creds)} 个凭证:"]
                for c in creds:
                    lines.append(f"  - {c['platform']} ({c.get('label', '无标签')}) | 保存时间: {c['saved_at']} | Cookies: {c['cookie_count']}个")
                return {"success": True, "output": "\n".join(lines), "data": creds}

            elif action == "delete_credentials":
                # [v1.25.4] 删除凭证
                platform = params.get("platform", "").strip()
                if not platform:
                    return {"success": False, "error": "请提供 platform 参数"}
                deleted = WebControlSession.delete_credentials(platform)
                if deleted:
                    return {"success": True, "output": f"已删除 {platform} 的凭证"}
                else:
                    return {"success": False, "error": f"未找到 {platform} 的凭证"}

            elif action == "save_cookies":
                # [v1.25.3] 保存 cookies 到文件
                label = params.get("label", "")
                try:
                    filepath = session.save_cookies_to_file(label)
                    return {"success": True, "output": f"Cookies 已保存到: {filepath} ({len(session.cookies)} 个)", "cookie_file": filepath}
                except Exception as e:
                    return {"success": False, "error": f"保存 cookies 失败: {e}"}

            elif action == "load_cookies":
                # [v1.25.3] 从文件加载 cookies
                label = params.get("label", "")
                count = session.load_cookies_from_file(label)
                if count > 0:
                    return {"success": True, "output": f"已加载 {count} 个 Cookies"}
                else:
                    return {"success": False, "error": f"未找到匹配的 Cookie 文件 (label: {label or '自动'})"}

            else:
                return {"success": False, "error": f"未知 web_control action: {action}"}

        except Exception as e:
            return {"success": False, "error": f"网页控制异常: {e}"}

    # =========================================================================
    # 内置感知工具 (v1.22.0)
    # =========================================================================

    async def _exec_image_ocr(self, params: Dict, task_id: str) -> Dict:
        """OCR 光学字符识别 — 从图片中提取文字"""
        image_path = params.get("image_path", "").strip()
        if not image_path:
            return {"success": False, "error": "缺少 image_path 参数"}
        p = _P(image_path).expanduser().resolve()
        if not p.exists():
            return {"success": False, "error": f"文件不存在: {image_path}"}
        if not p.suffix.lower().lstrip(".") in (
            "png", "jpg", "jpeg", "bmp", "tiff", "tif", "webp", "gif",
        ):
            return {"success": False, "error": f"不支持的图片格式: {p.suffix}"}

        # 优先使用 paddleocr
        try:
            from paddleocr import PaddleOCR
            lang = params.get("lang", "ch")  # ch=中英混合, en=英文
            ocr = PaddleOCR(use_angle_cls=True, lang=lang, show_log=False)
            result = ocr.ocr(str(p), cls=True)
            if result and result[0]:
                lines = []
                for line in result[0]:
                    text = line[1][0]
                    conf = line[1][1]
                    lines.append(f"{text}  (置信度: {conf:.0%})")
                return {
                    "success": True,
                    "output": "\n".join(lines),
                    "data": {"text_lines": len(lines), "engine": "paddleocr"},
                }
            else:
                return {"success": True, "output": "未检测到文字", "data": {"text_lines": 0}}
        except ImportError:
            return {
                "success": False,
                "error": "paddleocr 未安装。安装: pip install paddleocr paddlepaddle",
            }
        except Exception as e:
            return {"success": False, "error": f"OCR 识别失败: {e}"}

    async def _exec_image_analyze(self, params: Dict, task_id: str) -> Dict:
        """图片内容分析 — 使用 VLM (视觉语言模型) 分析图片"""
        image_path = params.get("image_path", "").strip()
        prompt = params.get("prompt", "请详细描述这张图片的内容，包括文字、布局、颜色、物体等信息。")
        if not image_path:
            return {"success": False, "error": "缺少 image_path 参数"}
        p = _P(image_path).expanduser().resolve()
        if not p.exists():
            return {"success": False, "error": f"文件不存在: {image_path}"}

        # 检查图片大小 (限制 20MB)
        file_size = p.stat().st_size
        if file_size > 20 * 1024 * 1024:
            return {"success": False, "error": f"图片过大 ({file_size / 1024 / 1024:.1f}MB)，限制 20MB"}

        # 编码为 base64
        import base64
        import mimetypes
        with open(p, "rb") as f:
            b64 = base64.b64encode(f.read()).decode("utf-8")
        mime = mimetypes.guess_type(str(p))[0] or "image/png"

        # 调用 VLM
        if not self.llm:
            return {"success": False, "error": "LLM 客户端未初始化，无法进行图片分析"}

        try:
            from core.llm import Message
            messages = [Message(
                role="user",
                content=[
                    {"type": "text", "text": prompt},
                    {
                        "type": "image_url",
                        "image_url": {"url": f"data:{mime};base64,{b64}"},
                    },
                ],
            )]
            response = await self.llm.chat(messages, temperature=0.1)
            if response.success and response.content:
                return {"success": True, "output": response.content}
            else:
                return {"success": False, "error": response.error or "VLM 未返回有效结果"}
        except Exception as e:
            return {"success": False, "error": f"图片分析失败: {e}"}

    async def _exec_audio_transcribe(self, params: Dict, task_id: str) -> Dict:
        """语音转文字 — 将音频文件转录为文本"""
        audio_path = params.get("audio_path", "").strip()
        language = params.get("language", "zh")
        if not audio_path:
            return {"success": False, "error": "缺少 audio_path 参数"}
        p = _P(audio_path).expanduser().resolve()
        if not p.exists():
            return {"success": False, "error": f"文件不存在: {audio_path}"}

        try:
            from core.stt import transcribe
            result = await transcribe(
                audio_path=str(p),
                language=language,
            )
            return result
        except Exception as e:
            return {"success": False, "error": f"语音转文字失败: {e}"}

    async def _auto_send_skill_files(
        self, files: List[str], task_id: str,
        stream_callback: Optional[Callable] = None,
        sent_files: Optional[List[Dict]] = None,
    ) -> None:
        """Skill 产生文件后，自动发送给用户 — 使用统一的 _exec_file_send 避免重复发送"""
        try:
            # 使用统一的 _exec_file_send 方法，避免重复的文件复制和SSE发送
            for fpath in files:
                try:
                    p = _P(fpath)
                    if p.exists():
                        # 直接调用 _exec_file_send，统一处理文件复制、SSE推送和持久化
                        file_result = await self._exec_file_send(
                            {"file_path": str(p), "description": f"生成的文件: {p.name}"},
                            task_id, stream_callback, sent_files
                        )
                        if not file_result.get("success"):
                            logger.warning(f"[{task_id}] 自动发送文件失败 ({fpath}): {file_result.get('error', '')}")
                except Exception as e:
                    logger.warning(f"[{task_id}] 自动发送文件异常 ({fpath}): {e}")
        except Exception as e:
            logger.warning(f"[{task_id}] 自动发送文件异常: {e}")

    # =========================================================================
    # [v1.38] 记忆与知识库工具（原 XML 标签转为原生工具调用）
    # =========================================================================

    async def _exec_save_memory(self, params: Dict, task_id: str) -> Dict:
        """保存记忆到记忆库（替代原 <remember> XML 标签）"""
        content = params.get("content", "").strip()
        if not content:
            return {"success": False, "error": "缺少 content 参数"}
        memory_type = params.get("type", "session")  # global 或 session
        session_id = params.get("session_id", "")

        try:
            if not self.memory_manager:
                return {"success": False, "error": "记忆管理器未初始化"}

            if memory_type == "global":
                # 全局记忆：查重后存储
                dup = self.memory_manager.find_duplicate_memory(
                    content=content, session_id=session_id,
                    key="conversation_insight",
                ) if hasattr(self.memory_manager, 'find_duplicate_memory') else None
                if dup:
                    self.memory_manager.update_memory(
                        memory_id=dup.id, content=content,
                    ) if hasattr(self.memory_manager, 'update_memory') else None
                    logger.info(f"[{task_id}] 全局记忆已更新: {dup.id}")
                    return {"success": True, "output": f"全局记忆已更新 (ID: {dup.id})", "type": "global"}
                else:
                    self.memory_manager.add_global(
                        session_id=session_id,
                        key="conversation_insight",
                        content=content,
                        summary=content[:200],
                        importance=0.7,
                    ) if hasattr(self.memory_manager, 'add_global') else None
                    logger.info(f"[{task_id}] 全局记忆已保存")
                    return {"success": True, "output": "全局记忆已保存", "type": "global"}
            else:
                # 会话记忆：查重后存储
                self.memory_manager.add_session(
                    session_id=session_id,
                    key="conversation_insight",
                    content=content,
                    importance=0.6,
                ) if hasattr(self.memory_manager, 'add_session') else None
                logger.info(f"[{task_id}] 会话记忆已保存")
                return {"success": True, "output": "会话记忆已保存", "type": "session"}
        except Exception as e:
            logger.warning(f"[{task_id}] 保存记忆失败: {e}")
            return {"success": False, "error": f"保存记忆失败: {e}"}

    async def _exec_recall_memory(self, params: Dict, task_id: str) -> Dict:
        """即时搜索记忆库（替代原 <recall> XML 标签，现在是即时返回结果而非下一轮注入）"""
        query = params.get("query", "").strip()
        if not query:
            return {"success": False, "error": "缺少 query 参数"}
        session_id = params.get("session_id", "")
        limit = params.get("limit", 5)

        try:
            if not self.memory_manager:
                return {"success": False, "error": "记忆管理器未初始化"}

            # 搜索相关记忆
            results = []
            if hasattr(self.memory_manager, 'search'):
                results = self.memory_manager.search(query, limit=limit) or []
            elif hasattr(self.memory_manager, 'recall'):
                results = self.memory_manager.recall(query, limit=limit) or []

            if not results:
                return {"success": True, "output": "未找到相关记忆", "results": []}

            # 格式化记忆结果
            output_parts = []
            for i, mem in enumerate(results[:limit], 1):
                if isinstance(mem, dict):
                    text = mem.get("content", "") or mem.get("text", "")
                    ts = mem.get("created_at", "") or mem.get("timestamp", "")
                    output_parts.append(f"{i}. {text}" + (f" ({ts})" if ts else ""))
                else:
                    output_parts.append(f"{i}. {getattr(mem, 'content', str(mem))}")

            output = "\n".join(output_parts)
            logger.info(f"[{task_id}] 记忆搜索返回 {len(results)} 条结果")
            return {"success": True, "output": output, "count": len(results)}
        except Exception as e:
            logger.warning(f"[{task_id}] 搜索记忆失败: {e}")
            return {"success": False, "error": f"搜索记忆失败: {e}"}

    async def _exec_save_knowledge(self, params: Dict, task_id: str) -> Dict:
        """保存知识到知识库（替代原 <knowledge> XML 标签）"""
        content = params.get("content", "").strip()
        if not content:
            return {"success": False, "error": "缺少 content 参数"}
        category = params.get("category", "general")

        try:
            if not self.knowledge_base_dir:
                return {"success": False, "error": "知识库目录未配置"}

            import os
            os.makedirs(self.knowledge_base_dir, exist_ok=True)

            # 保存为 markdown 文件（与原 _save_knowledge_to_base 逻辑一致）
            import time
            filename = f"kb_{int(time.time())}_{category}.md"
            filepath = os.path.join(self.knowledge_base_dir, filename)

            with open(filepath, "w", encoding="utf-8") as f:
                f.write(f"# {category}\n\n{content}\n")

            logger.info(f"[{task_id}] 知识已保存: {filepath}")
            return {"success": True, "output": f"知识已保存到知识库 ({filename})"}
        except Exception as e:
            logger.warning(f"[{task_id}] 保存知识失败: {e}")
            return {"success": False, "error": f"保存知识失败: {e}"}

    async def _exec_search_knowledge(self, params: Dict, task_id: str) -> Dict:
        """搜索知识库（替代原 <get_knowledge> XML 标签，现在是即时返回结果）"""
        query = params.get("query", "").strip()
        if not query:
            return {"success": False, "error": "缺少 query 参数"}
        limit = params.get("limit", 5)

        try:
            if not self.knowledge_base_dir:
                return {"success": False, "error": "知识库目录未配置"}

            import os
            if not os.path.isdir(self.knowledge_base_dir):
                return {"success": True, "output": "知识库为空", "results": []}

            # 简单关键词搜索（与原 ContextBuilder 的知识库搜索一致）
            results = []
            query_lower = query.lower()
            for fname in os.listdir(self.knowledge_base_dir):
                if not fname.endswith(".md"):
                    continue
                fpath = os.path.join(self.knowledge_base_dir, fname)
                try:
                    with open(fpath, "r", encoding="utf-8") as f:
                        text = f.read()
                    if query_lower in text.lower():
                        results.append({"file": fname, "content": text[:500]})
                except Exception:
                    continue

            if not results:
                return {"success": True, "output": f"未找到与 '{query}' 相关的知识", "results": []}

            output_parts = []
            for i, r in enumerate(results[:limit], 1):
                output_parts.append(f"{i}. [{r['file']}]\n{r['content'][:300]}")

            output = "\n\n".join(output_parts)
            logger.info(f"[{task_id}] 知识搜索返回 {len(results)} 条结果")
            return {"success": True, "output": output, "count": len(results)}
        except Exception as e:
            logger.warning(f"[{task_id}] 搜索知识失败: {e}")
            return {"success": False, "error": f"搜索知识失败: {e}"}

    async def _exec_update_title(self, params: Dict, task_id: str) -> Dict:
        """更新会话标题（替代原 <mainsubject> XML 标签）"""
        title = params.get("title", "").strip()
        if not title:
            return {"success": False, "error": "缺少 title 参数"}
        session_id = params.get("session_id", "")

        try:
            if self.memory_manager and session_id:
                self.memory_manager.rename_session(session_id, title[:20])
                logger.info(f"[{task_id}] 会话标题已更新: {title[:20]}")
                return {"success": True, "output": f"会话标题已更新为: {title[:20]}"}
            return {"success": False, "error": "记忆管理器未初始化或缺少 session_id"}
        except Exception as e:
            logger.warning(f"[{task_id}] 更新会话标题失败: {e}")
            return {"success": False, "error": f"更新会话标题失败: {e}"}