main.py

"""LLM 驱动的 ReAct 智能体的 CLI 入口点。"""

from __future__ import annotations

import argparse
import json
import os
import re
import shutil
import subprocess
import sys
import textwrap
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List

from dotenv import load_dotenv

from dm_agent import (
    CriticAgent,
    LLMError,
    ReactAgent,
    Tool,
    create_llm_client,
    default_tools,
    PROVIDER_DEFAULTS,
)
from dm_agent.mcp import MCPManager, load_mcp_config
from dm_agent.skills import SkillManager
from dm_agent.tracing import TraceWriter

try:
    from rich import box
    from rich.console import Console, Group
    from rich.padding import Padding
    from rich.panel import Panel
    from rich.prompt import Prompt
    from rich.table import Table
    from rich.text import Text

    RICH_AVAILABLE = True
    RICH_CONSOLE = Console(highlight=False, soft_wrap=True)
except ImportError:
    RICH_AVAILABLE = False
    RICH_CONSOLE = None

# 尝试导入 colorama 用于彩色输出
try:
    from colorama import Fore, Style, init as colorama_init

    colorama_init(autoreset=True)
    COLORS_AVAILABLE = True
except ImportError:
    COLORS_AVAILABLE = False

    # 如果没有 colorama，定义空的颜色常量
    class Fore:
        GREEN = ""
        YELLOW = ""
        RED = ""
        CYAN = ""
        MAGENTA = ""
        BLUE = ""
        WHITE = ""

    class Style:
        BRIGHT = ""
        DIM = ""
        RESET_ALL = ""


@dataclass
class Config:
    """运行时配置"""

    api_key: str
    provider: str = "deepseek"
    model: str = "deepseek-chat"
    base_url: str = "https://api.deepseek.com"
    max_steps: int = 100
    temperature: float = 0.7
    show_steps: bool = False
    enable_reflexion: bool = False
    max_trials: int = 3
    enable_critic: bool = False
    enable_adaptive_replanning: bool = False
    max_replans: int = -1
    enable_repeated_failure_policy_experiment: bool = False
    enable_evolution: bool = False


# 配置文件路径
CONFIG_FILE = os.path.join(os.path.dirname(__file__), "config.json")


class UI:
    """Terminal UI layer with Rich rendering and a colorama fallback."""

    WIDTH = 88
    RICH_STYLES = {
        "ok": "bold white on green",
        "error": "bold white on red",
        "warn": "bold black on yellow",
        "info": "bold white on blue",
        "run": "bold white on magenta",
    }
    RICH_LABELS = {
        "ok": "DONE",
        "error": "ERROR",
        "warn": "WARN",
        "info": "INFO",
        "run": "RUN",
    }

    @staticmethod
    def rich_enabled() -> bool:
        return RICH_AVAILABLE and RICH_CONSOLE is not None

    @staticmethod
    def clear() -> None:
        if UI.rich_enabled():
            RICH_CONSOLE.clear()
            return
        os.system("cls" if os.name == "nt" else "clear")

    @staticmethod
    def width() -> int:
        return max(72, min(UI.WIDTH, shutil.get_terminal_size((UI.WIDTH, 20)).columns))

    @staticmethod
    def paint(text: str, color: str = "", *, bright: bool = False, dim: bool = False) -> str:
        style = ""
        if bright:
            style += Style.BRIGHT
        if dim:
            style += getattr(Style, "DIM", "")
        return f"{style}{color}{text}{Style.RESET_ALL}"

    @staticmethod
    def rule(label: str = "", *, color: str = Fore.CYAN) -> None:
        if UI.rich_enabled():
            title = Text(f" {label} ", style="bright_black") if label else ""
            RICH_CONSOLE.rule(title, style="bright_black")
            return
        width = UI.width()
        if label:
            prefix = f" {label} "
            line = prefix + "-" * max(width - len(prefix), 0)
        else:
            line = "-" * width
        print(UI.paint(line, color, dim=not label))

    @staticmethod
    def banner(title: str, subtitle: str = "") -> None:
        if UI.rich_enabled():
            heading = Text()
            if title == "DM-Code-Agent":
                heading.append("DM", style="bold cyan")
                heading.append("-Code-Agent", style="bold white")
            else:
                heading.append(title, style="bold white")

            body: Group | Text
            if subtitle:
                caption = Text(subtitle, style="bright_black")
                body = Group(heading, Text(""), caption)
            else:
                body = heading
            RICH_CONSOLE.print(
                Panel(
                    body,
                    border_style="bright_black",
                    box=box.ROUNDED,
                    padding=(1, 2),
                    expand=True,
                    subtitle=(
                        Text("trace  tools  skills  memory", style="bright_black")
                        if title == "DM-Code-Agent"
                        else None
                    ),
                )
            )
            return
        print()
        print(UI.paint(title, Fore.GREEN, bright=True))
        if subtitle:
            print("  " + UI.paint(subtitle, Fore.WHITE, dim=True))
        print()

    @staticmethod
    def section(title: str, subtitle: str = "") -> None:
        if UI.rich_enabled():
            text = Text()
            text.append(title, style="bold white")
            if subtitle:
                text.append("\n")
                text.append(subtitle, style="bright_black")
            RICH_CONSOLE.print(Padding(text, (1, 0, 0, 0)))
            return
        print()
        print(UI.paint(title, Fore.CYAN, bright=True))
        if subtitle:
            for line in UI.wrap(subtitle, width=UI.width() - 4):
                print("  " + UI.paint(line, Fore.WHITE, dim=True))

    @staticmethod
    def panel(title: str, body: str = "", *, color: str = Fore.CYAN) -> None:
        if UI.rich_enabled():
            style = UI._rich_color(color)
            RICH_CONSOLE.print(
                Panel(
                    UI._rich_wrapped_text(str(body), width=UI.width() - 10),
                    title=Text(f" {title} ", style=f"bold {style}"),
                    title_align="left",
                    border_style="bright_black",
                    box=box.ROUNDED,
                    padding=(1, 2),
                    expand=True,
                )
            )
            return
        print()
        print(UI.paint(title, color, bright=True))
        if body:
            for raw_line in str(body).splitlines() or [""]:
                wrapped = UI.wrap(raw_line, width=UI.width() - 4) or [""]
                for line in wrapped:
                    print("  " + line)

    @staticmethod
    def wrap(text: str, *, width: int | None = None) -> List[str]:
        return textwrap.wrap(
            str(text),
            width=width or UI.width() - 4,
            replace_whitespace=False,
            drop_whitespace=False,
        )

    @staticmethod
    def _rich_wrapped_text(text: str, *, width: int) -> Text:
        wrapped = Text()
        target_width = max(36, width)
        lines: List[str] = []
        for raw_line in str(text).splitlines() or [""]:
            if raw_line.strip():
                lines.extend(UI.wrap(raw_line, width=target_width) or [""])
            else:
                lines.append("")
        for index, line in enumerate(lines):
            if index:
                wrapped.append("\n")
            wrapped.append(line)
        return wrapped

    @staticmethod
    def status(kind: str, message: str, detail: str = "") -> None:
        if UI.rich_enabled():
            style = UI.RICH_STYLES.get(kind, UI.RICH_STYLES["info"])
            label = UI.RICH_LABELS.get(kind, kind.upper())
            text = Text()
            text.append(f" {label:<5} ", style=style)
            text.append(" ")
            text.append(message, style="white")
            if detail:
                text.append("  ")
                text.append(detail, style="bright_black")
            RICH_CONSOLE.print(Padding(text, (0, 0, 0, 1)))
            return
        palette = {
            "ok": (Fore.GREEN, "ok"),
            "error": (Fore.RED, "err"),
            "warn": (Fore.YELLOW, "warn"),
            "info": (Fore.CYAN, "info"),
            "run": (Fore.MAGENTA, "run"),
        }
        color, icon = palette.get(kind, palette["info"])
        line = f"{UI.paint(icon.ljust(5), color, bright=True)} {message}"
        if detail:
            line += UI.paint(f"  {detail}", Fore.WHITE, dim=True)
        print(line)

    @staticmethod
    def key_values(title: str, rows: List[tuple[str, Any]]) -> None:
        if UI.rich_enabled():
            table = Table.grid(padding=(0, 3))
            table.add_column(style="bright_black", no_wrap=True)
            table.add_column(style="bold white")
            for key, value in rows:
                table.add_row(str(key), str(value))
            RICH_CONSOLE.print(
                Panel(
                    table,
                    title=Text(f" {title} ", style="bold cyan"),
                    title_align="left",
                    border_style="bright_black",
                    box=box.ROUNDED,
                    padding=(1, 2),
                )
            )
            return
        UI.section(title)
        key_width = max((len(key) for key, _ in rows), default=0)
        for key, value in rows:
            print(
                f"  {UI.paint(key.ljust(key_width), Fore.WHITE, dim=True)}  "
                f"{UI.paint(str(value), Fore.YELLOW)}"
            )

    @staticmethod
    def menu(items: List[tuple[str, str]]) -> None:
        if UI.rich_enabled():
            table = Table.grid(expand=True, padding=(0, 2))
            table.add_column(justify="right", no_wrap=True, width=5)
            table.add_column(style="bold white", no_wrap=True, width=18)
            table.add_column(style="bright_black", ratio=1)
            for index, (title, description) in enumerate(items, start=1):
                badge = Text(f" {index} ", style="bold black on cyan")
                table.add_row(badge, title, description)
            RICH_CONSOLE.print(
                Panel(
                    table,
                    title=Text(" 主菜单 ", style="bold white"),
                    title_align="left",
                    subtitle=Text("输入编号选择操作", style="bright_black"),
                    border_style="bright_black",
                    box=box.ROUNDED,
                    padding=(1, 2),
                )
            )
            return
        UI.section("主菜单", "输入编号选择一个操作")
        for index, (title, description) in enumerate(items, start=1):
            badge = UI.paint(f"[{index}]", Fore.GREEN, bright=True)
            name = UI.paint(title.ljust(14), Fore.WHITE, bright=True)
            print(f"  {badge} {name} {UI.paint(description, Fore.WHITE, dim=True)}")
        print()

    @staticmethod
    def truncate(value: Any, limit: int = 220) -> str:
        text = str(value)
        if len(text) <= limit:
            return text
        return text[: max(limit - 3, 0)].rstrip() + "..."

    @staticmethod
    def ask(
        prompt: str,
        *,
        choices: List[str] | None = None,
        default: str | None = None,
        show_choices: bool = True,
    ) -> str:
        if UI.rich_enabled():
            prompt_kwargs = {
                "choices": choices,
                "console": RICH_CONSOLE,
                "show_choices": show_choices,
            }
            if default is not None:
                prompt_kwargs["default"] = default
            return Prompt.ask(Text(prompt, style="bold cyan"), **prompt_kwargs)
        suffix = f" [{default}]" if default is not None else ""
        return input(f"{UI.paint(prompt, Fore.CYAN)}{suffix}: ")

    @staticmethod
    def prompt_line(prompt: str, *, default: str = "") -> str:
        if UI.rich_enabled():
            return Prompt.ask(
                Text(prompt, style="bold cyan"),
                console=RICH_CONSOLE,
                default=default,
                show_default=False,
            )
        suffix = f" [{default}]" if default else ""
        return input(f"{UI.paint(prompt, Fore.CYAN)}{suffix}: ")

    @staticmethod
    def pause(prompt: str = "按 Enter 返回") -> None:
        UI.prompt_line(prompt, default="")

    @staticmethod
    def _rich_color(color: str) -> str:
        if color == Fore.GREEN:
            return "green"
        if color == Fore.YELLOW:
            return "yellow"
        if color == Fore.RED:
            return "red"
        if color == Fore.MAGENTA:
            return "magenta"
        if color == Fore.BLUE:
            return "blue"
        return "cyan"


def configure_console_encoding() -> None:
    """Avoid crashes when Windows terminals cannot encode Unicode status symbols."""
    for stream in (sys.stdout, sys.stderr):
        if hasattr(stream, "reconfigure"):
            try:
                stream.reconfigure(errors="replace")
            except Exception:
                pass


def load_config_from_file() -> Dict[str, Any]:
    """从配置文件加载设置"""
    if os.path.exists(CONFIG_FILE):
        try:
            with open(CONFIG_FILE, "r", encoding="utf-8") as f:
                return json.load(f)
        except Exception as e:
            UI.status("warn", "配置文件加载失败，使用默认设置", str(e))
    return {}


def save_config_to_file(config: Config) -> None:
    """保存配置到文件"""
    try:
        config_data = {
            "provider": config.provider,
            "model": config.model,
            "base_url": config.base_url,
            "max_steps": config.max_steps,
            "temperature": config.temperature,
            "show_steps": config.show_steps,
            "enable_reflexion": config.enable_reflexion,
            "max_trials": config.max_trials,
            "enable_critic": config.enable_critic,
            "enable_adaptive_replanning": config.enable_adaptive_replanning,
            "max_replans": config.max_replans,
            "enable_repeated_failure_policy_experiment": (
                config.enable_repeated_failure_policy_experiment
            ),
            "enable_evolution": config.enable_evolution,
        }
        with open(CONFIG_FILE, "w", encoding="utf-8") as f:
            json.dump(config_data, f, indent=2, ensure_ascii=False)
        UI.status("ok", "配置已保存")
    except Exception as e:
        UI.status("error", "配置保存失败", str(e))


def get_api_key_for_provider(provider: str) -> str | None:
    """根据提供商获取对应的 API 密钥"""
    provider_env_map = {
        "deepseek": "DEEPSEEK_API_KEY",
        "openai": "OPENAI_API_KEY",
        "claude": "CLAUDE_API_KEY",
        "gemini": "GEMINI_API_KEY",
    }
    env_var = provider_env_map.get(provider.lower())
    return os.getenv(env_var) if env_var else None


def resolve_advanced_features(config: Config) -> Dict[str, bool]:
    """Return effective advanced feature switches for one agent run."""
    adaptive_replanning = config.enable_adaptive_replanning or config.enable_evolution
    repeated_failure_policy = (
        config.enable_repeated_failure_policy_experiment or config.enable_evolution
    )
    return {
        "reflexion": config.enable_reflexion,
        "critic": config.enable_critic,
        "adaptive_replanning": adaptive_replanning,
        "repeated_failure_policy_experiment": repeated_failure_policy,
        "evolution": config.enable_evolution,
    }


def format_advanced_feature_status(config: Config) -> str:
    """Compact human-readable summary of advanced feature switches."""
    advanced = resolve_advanced_features(config)
    enabled = [
        label
        for key, label in [
            ("reflexion", "reflexion"),
            ("critic", "critic"),
            ("adaptive_replanning", "adaptive-replan"),
            ("repeated_failure_policy_experiment", "loop-break"),
            ("evolution", "evolution"),
        ]
        if advanced[key]
    ]
    return ", ".join(enabled) if enabled else "off"


def validate_feature_args(args: argparse.Namespace) -> str:
    """Validate default-off advanced feature CLI arguments."""
    if args.max_trials < 1:
        return "--max-trials must be at least 1."
    if args.max_replans < -1:
        return "--max-replans must be -1 or greater."
    if (
        args.enable_repeated_failure_policy_experiment
        and not args.enable_adaptive_replanning
        and not args.enable_evolution
    ):
        return (
            "--enable-repeated-failure-policy-experiment requires "
            "--enable-adaptive-replanning or --enable-evolution."
        )
    return ""


def parse_args(argv: Any) -> argparse.Namespace:
    # 先加载配置文件中的默认值
    saved_config = load_config_from_file()

    parser = argparse.ArgumentParser(description="运行基于 LLM 的 ReAct 智能体来完成任务描述。")
    parser.add_argument("task", nargs="?", help="智能体要完成的自然语言任务。")

    # 获取配置中的提供商或默认值
    default_provider = saved_config.get("provider", "deepseek")

    # 根据提供商获取对应的 API 密钥
    default_api_key = get_api_key_for_provider(default_provider)

    parser.add_argument(
        "--api-key",
        dest="api_key",
        default=default_api_key,
        help="API 密钥（默认使用环境变量）。",
    )
    parser.add_argument(
        "--provider",
        default=saved_config.get("provider", "deepseek"),
        help="LLM 提供商 (deepseek/openai/claude/gemini，默认：deepseek)。",
    )
    parser.add_argument(
        "--model",
        default=saved_config.get("model", "deepseek-chat"),
        help="模型标识符（默认根据提供商选择）。",
    )
    parser.add_argument(
        "--base-url",
        dest="base_url",
        default=saved_config.get("base_url"),
        help="API 基础 URL（可选，使用提供商默认值）。",
    )
    parser.add_argument(
        "--max-steps",
        type=int,
        default=saved_config.get("max_steps", 100),
        help="放弃前的最大推理/工具步骤数（默认：100）。",
    )
    parser.add_argument(
        "--temperature",
        type=float,
        default=saved_config.get("temperature", 0.7),
        help="模型的采样温度（默认：0.7）。",
    )
    parser.add_argument(
        "--show-steps",
        action="store_true",
        default=saved_config.get("show_steps", False),
        help="打印智能体执行的中间 ReAct 步骤。",
    )
    parser.add_argument(
        "--enable-reflexion",
        action="store_true",
        default=saved_config.get("enable_reflexion", False),
        help="启用失败后的 Reflexion 反思重试。默认关闭。",
    )
    parser.add_argument(
        "--max-trials",
        type=int,
        default=saved_config.get("max_trials", 3),
        help="启用 Reflexion 时最多尝试的轮数（默认：3）。",
    )
    parser.add_argument(
        "--enable-critic",
        action="store_true",
        default=saved_config.get("enable_critic", False),
        help="启用完成前 Critic 审查门禁。默认关闭。",
    )
    parser.add_argument(
        "--enable-adaptive-replanning",
        action="store_true",
        default=saved_config.get("enable_adaptive_replanning", False),
        help="启用基于失败信号的自适应重规划。默认关闭。",
    )
    parser.add_argument(
        "--max-replans",
        type=int,
        default=saved_config.get("max_replans", -1),
        help="自适应重规划最多触发次数；-1 表示不限（默认：-1）。",
    )
    parser.add_argument(
        "--enable-repeated-failure-policy-experiment",
        action="store_true",
        default=saved_config.get("enable_repeated_failure_policy_experiment", False),
        help="启用重复失败时的实验性跳出策略。默认关闭。",
    )
    parser.add_argument(
        "--enable-evolution",
        action="store_true",
        default=saved_config.get("enable_evolution", False),
        help="启用实验性进化恢复模式：自动打开自适应重规划和重复失败跳出策略。默认关闭。",
    )
    parser.add_argument(
        "--interactive",
        action="store_true",
        help="启动交互式菜单模式。",
    )
    parser.add_argument(
        "--trace",
        type=Path,
        help="将本次任务的结构化执行轨迹写入 JSONL 文件。",
    )
    parser.add_argument(
        "--trace-llm-io",
        action="store_true",
        help="在 trace 中包含完整 LLM 输入/输出。仅建议在私有调试时启用。",
    )
    parser.add_argument(
        "--report",
        type=Path,
        help="将本次任务的人类可读运行报告写入 Markdown 文件。",
    )
    return parser.parse_args(argv)


def print_separator(char: str = "=", length: int = 70) -> None:
    """打印分隔线"""
    _ = (char, length)
    UI.rule()


def print_header(text: str) -> None:
    """打印标题"""
    UI.banner(text)


def print_welcome() -> None:
    """打印欢迎界面"""
    UI.banner("DM-Code-Agent", "Local-first code agent with trace, tools, skills, and memory.")

    # 显示配置文件状态
    if os.path.exists(CONFIG_FILE):
        UI.status("ok", "已加载配置文件", "config.json")
    else:
        UI.status("info", "使用默认配置", "max_steps=100 | temperature=0.7")
    print()


def print_menu() -> None:
    """打印主菜单"""
    UI.menu(
        [
            ("执行新任务", "一次性运行一个代码维护任务"),
            ("多轮对话模式", "复用当前 agent 的短期上下文和本地记忆"),
            ("查看工具列表", "浏览文件、Shell、测试、MCP 等可用工具"),
            ("配置设置", "切换模型、温度、最大步骤和显示选项"),
            ("查看可用技能列表", "查看内置和自定义技能"),
            ("退出程序", "关闭 MCP 并返回终端"),
        ]
    )


def show_tools(tools: List[Tool]) -> None:
    """显示可用工具列表"""
    if UI.rich_enabled():
        table = Table(
            show_header=True,
            header_style="bold white",
            box=box.SIMPLE_HEAD,
            border_style="bright_black",
            pad_edge=False,
        )
        table.add_column("#", justify="right", style="cyan", no_wrap=True)
        table.add_column("Tool", style="bold white", no_wrap=True)
        table.add_column("Description", style="bright_black")
        for idx, tool in enumerate(tools, start=1):
            table.add_row(str(idx), tool.name, tool.description)
        RICH_CONSOLE.print(
            Panel(
                table,
                title=Text(f" 可用工具 ({len(tools)}) ", style="bold cyan"),
                title_align="left",
                border_style="bright_black",
                box=box.ROUNDED,
                padding=(1, 2),
            )
        )
        return

    UI.section("可用工具", f"{len(tools)} 个工具已加载")

    for idx, tool in enumerate(tools, start=1):
        print(
            f"  {UI.paint(f'{idx:>2}', Fore.GREEN, bright=True)}  "
            f"{UI.paint(tool.name, Fore.WHITE, bright=True)}"
        )
        for line in UI.wrap(tool.description, width=UI.width() - 8):
            print(f"      {UI.paint(line, Fore.WHITE, dim=True)}")

    print()


def show_skills(skill_manager: SkillManager) -> None:
    """显示可用技能列表"""
    skills_info = skill_manager.get_all_skill_info()
    if UI.rich_enabled():
        table = Table(
            show_header=True,
            header_style="bold white",
            box=box.SIMPLE_HEAD,
            border_style="bright_black",
            pad_edge=False,
        )
        table.add_column("#", justify="right", style="cyan", no_wrap=True)
        table.add_column("Skill", no_wrap=True)
        table.add_column("Source", style="yellow", no_wrap=True)
        table.add_column("Tools", justify="right", style="magenta", no_wrap=True)
        table.add_column("Description", style="bright_black")
        for idx, info in enumerate(skills_info, start=1):
            source = "内置" if info["is_builtin"] else "自定义"
            skill_name = Text(str(info["display_name"]), style="bold white")
            if info["is_active"]:
                skill_name.append("  ACTIVE", style="bold green")
            table.add_row(
                str(idx),
                skill_name,
                source,
                str(info["tools_count"]),
                info["description"],
            )
        body = table if skills_info else Text("暂无可用技能", style="yellow")
        RICH_CONSOLE.print(
            Panel(
                body,
                title=Text(f" 可用技能 ({len(skills_info)}) ", style="bold cyan"),
                title_align="left",
                border_style="bright_black",
                box=box.ROUNDED,
                padding=(1, 2),
            )
        )
        return

    UI.section("可用技能", f"{len(skills_info)} 个技能已发现")
    if not skills_info:
        UI.status("warn", "暂无可用技能")
    else:
        for idx, info in enumerate(skills_info, start=1):
            status = UI.paint("active", Fore.GREEN, bright=True) if info["is_active"] else ""
            source = "内置" if info["is_builtin"] else "自定义"
            header = (
                f"  {UI.paint(f'{idx:>2}', Fore.GREEN, bright=True)}  "
                f"{UI.paint(info['display_name'], Fore.WHITE, bright=True)}"
            )
            print(f"{header}  {status}".rstrip())
            print(
                f"      {UI.paint(info['name'], Fore.YELLOW)} | {source} | "
                f"v{info['version']} | {info['tools_count']} tools"
            )
            for line in UI.wrap(info["description"], width=UI.width() - 8):
                print(f"      {UI.paint(line, Fore.WHITE, dim=True)}")
            print(
                f"      {UI.paint('关键词', Fore.WHITE, dim=True)}  "
                f"{', '.join(info['keywords'][:8])}"
                f"{'...' if len(info['keywords']) > 8 else ''}"
            )
            print()

    print()


def ask_bool_setting(label: str, current: bool) -> bool:
    value = UI.ask(label, choices=["y", "n"], default="y" if current else "n").strip().lower()
    if value in {"y", "yes", "是"}:
        return True
    if value in {"n", "no", "否"}:
        return False
    return current


def configure_settings(config: Config) -> None:
    """配置设置"""
    UI.key_values(
        "当前配置",
        [
            ("Provider", config.provider),
            ("Model", config.model),
            ("Base URL", config.base_url),
            ("Max steps", config.max_steps),
            ("Temperature", config.temperature),
            ("Show steps", "是" if config.show_steps else "否"),
            ("Advanced", format_advanced_feature_status(config)),
            ("Reflexion max trials", config.max_trials),
            ("Max replans", config.max_replans),
        ],
    )
    UI.status("info", "选择要修改的设置", "直接回车跳过")
    print()

    config_changed = False

    # 修改提供商
    provider_input = (
        UI.ask(
            "LLM 提供商",
            choices=["deepseek", "openai", "claude", "gemini"],
            default=config.provider,
        )
        .strip()
        .lower()
    )
    if provider_input and provider_input in ["deepseek", "openai", "claude", "gemini"]:
        if provider_input != config.provider:
            # 尝试获取新提供商的 API 密钥
            new_api_key = get_api_key_for_provider(provider_input)
            if not new_api_key:
                UI.status("error", f"未找到 {provider_input.upper()}_API_KEY 环境变量")
                UI.status("warn", f"请在 .env 文件中配置 {provider_input.upper()}_API_KEY")
            else:
                config.provider = provider_input
                config.api_key = new_api_key  # 更新 API 密钥
                # 自动更新默认模型和 base_url
                defaults = PROVIDER_DEFAULTS.get(provider_input, {})
                config.model = defaults.get("model", config.model)
                config.base_url = defaults.get("base_url", config.base_url)
                config_changed = True
                UI.status("ok", f"已更新提供商为 {provider_input}", "模型和 URL 已自动调整")
    elif provider_input and provider_input not in ["deepseek", "openai", "claude", "gemini"]:
        UI.status("error", "无效的提供商")

    # 修改模型
    model_input = UI.ask("模型名称", default=config.model).strip()
    if model_input and model_input != config.model:
        config.model = model_input
        config_changed = True
        UI.status("ok", f"已更新模型为 {model_input}")

    # 修改 Base URL
    base_url_input = UI.ask("Base URL", default=config.base_url).strip()
    if base_url_input and base_url_input != config.base_url:
        config.base_url = base_url_input
        config_changed = True
        UI.status("ok", f"已更新 Base URL 为 {base_url_input}")

    # 修改最大步骤数
    try:
        max_steps_input = UI.ask("最大步骤数", default=str(config.max_steps)).strip()
        if max_steps_input:
            new_max_steps = int(max_steps_input)
            if new_max_steps > 0:
                if new_max_steps != config.max_steps:
                    config.max_steps = new_max_steps
                    config_changed = True
                    UI.status("ok", f"已更新最大步骤数为 {new_max_steps}")
            else:
                UI.status("error", "最大步骤数必须大于 0")
    except ValueError:
        UI.status("error", "无效的数字")

    # 修改温度
    try:
        temp_input = UI.ask("温度 (0.0-2.0)", default=str(config.temperature)).strip()
        if temp_input:
            new_temp = float(temp_input)
            if 0.0 <= new_temp <= 2.0:
                if new_temp != config.temperature:
                    config.temperature = new_temp
                    config_changed = True
                    UI.status("ok", f"已更新温度为 {new_temp}")
            else:
                UI.status("error", "温度必须在 0.0 到 2.0 之间")
    except ValueError:
        UI.status("error", "无效的数字")

    # 修改显示步骤
    new_show_steps = ask_bool_setting("显示步骤", config.show_steps)
    if new_show_steps != config.show_steps:
        config.show_steps = new_show_steps
        config_changed = True
    UI.status("ok", "已启用显示步骤" if config.show_steps else "已禁用显示步骤")

    UI.section(
        "高级功能",
        "这些能力会增加模型调用或改变恢复策略；默认关闭，建议按任务显式启用。",
    )

    new_reflexion = ask_bool_setting("启用 Reflexion 反思重试", config.enable_reflexion)
    if new_reflexion != config.enable_reflexion:
        config.enable_reflexion = new_reflexion
        config_changed = True
    if config.enable_reflexion:
        try:
            max_trials_input = UI.ask("反思最大尝试轮数", default=str(config.max_trials)).strip()
            if max_trials_input:
                new_max_trials = int(max_trials_input)
                if new_max_trials >= 1:
                    if new_max_trials != config.max_trials:
                        config.max_trials = new_max_trials
                        config_changed = True
                else:
                    UI.status("error", "反思最大尝试轮数必须至少为 1")
        except ValueError:
            UI.status("error", "无效的数字")

    new_critic = ask_bool_setting("启用 Critic 完成审查", config.enable_critic)
    if new_critic != config.enable_critic:
        config.enable_critic = new_critic
        config_changed = True

    new_adaptive = ask_bool_setting("启用自适应重规划", config.enable_adaptive_replanning)
    if new_adaptive != config.enable_adaptive_replanning:
        config.enable_adaptive_replanning = new_adaptive
        config_changed = True
    if config.enable_adaptive_replanning:
        try:
            max_replans_input = UI.ask(
                "最大重规划次数 (-1 表示不限)", default=str(config.max_replans)
            ).strip()
            if max_replans_input:
                new_max_replans = int(max_replans_input)
                if new_max_replans >= -1:
                    if new_max_replans != config.max_replans:
                        config.max_replans = new_max_replans
                        config_changed = True
                else:
                    UI.status("error", "最大重规划次数必须为 -1 或更大")
        except ValueError:
            UI.status("error", "无效的数字")

    new_loop_break = ask_bool_setting(
        "启用重复失败跳出实验", config.enable_repeated_failure_policy_experiment
    )
    if new_loop_break != config.enable_repeated_failure_policy_experiment:
        config.enable_repeated_failure_policy_experiment = new_loop_break
        config_changed = True
    if config.enable_repeated_failure_policy_experiment and not config.enable_adaptive_replanning:
        config.enable_adaptive_replanning = True
        config_changed = True
        UI.status("info", "已同步启用自适应重规划", "重复失败跳出实验依赖重规划")

    new_evolution = ask_bool_setting("启用进化恢复模式", config.enable_evolution)
    if new_evolution != config.enable_evolution:
        config.enable_evolution = new_evolution
        config_changed = True
    if config.enable_evolution and not config.enable_adaptive_replanning:
        UI.status("info", "进化恢复会在运行时自动启用自适应重规划和重复失败跳出")

    # 保存配置
    if config_changed:
        print()
        save_choice = (
            UI.ask("是否保存为永久配置？", choices=["y", "n"], default="y").strip().lower()
        )
        if save_choice in ["", "y", "yes", "是"]:
            save_config_to_file(config)

    print()


def display_result(result: Dict[str, Any], show_steps: bool = False) -> None:
    """格式化显示任务结果"""
    if show_steps and result.get("steps"):
        if UI.rich_enabled():
            table = Table(
                show_header=True,
                header_style="bold white",
                box=box.SIMPLE_HEAD,
                border_style="bright_black",
                pad_edge=False,
            )
            table.add_column("#", justify="right", style="cyan", no_wrap=True)
            table.add_column("Action", style="bold white", no_wrap=True)
            table.add_column("Thought", style="bright_black")
            table.add_column("Observation", style="bright_black")
            for idx, step in enumerate(result.get("steps", []), start=1):
                table.add_row(
                    str(idx),
                    str(step.get("action", "")),
                    UI.truncate(step.get("thought", ""), 120),
                    UI.truncate(step.get("observation", ""), 140),
                )
            RICH_CONSOLE.print(
                Panel(
                    table,
                    title=Text(" 执行步骤 ", style="bold magenta"),
                    title_align="left",
                    border_style="bright_black",
                    box=box.ROUNDED,
                    padding=(1, 2),
                )
            )
        else:
            UI.section("执行步骤")
            for idx, step in enumerate(result.get("steps", []), start=1):
                print(
                    f"  {UI.paint(f'{idx:>2}', Fore.MAGENTA, bright=True)}  "
                    f"{UI.paint(str(step.get('action', '')), Fore.WHITE, bright=True)}"
                )
                print(f"      {UI.paint('thought', Fore.WHITE, dim=True)}  {step.get('thought')}")
                action_input = step.get("action_input")
                if action_input:
                    print(
                        f"      {UI.paint('input', Fore.WHITE, dim=True)}    "
                        f"{json.dumps(action_input, ensure_ascii=False)}"
                    )
                print(
                    f"      {UI.paint('observe', Fore.WHITE, dim=True)}  "
                    f"{UI.truncate(step.get('observation'))}"
                )
                print()

    final_answer = result.get("final_answer", "")
    UI.panel("最终答案", str(final_answer), color=Fore.GREEN)
    completion_summary = str(result.get("metadata", {}).get("completion_summary", "")).strip()
    if completion_summary and completion_summary != str(final_answer).strip():
        UI.panel("本轮总结", completion_summary, color=Fore.CYAN)


def collect_git_status() -> List[str]:
    """Return short git status lines for the current workspace, if available."""
    try:
        completed = subprocess.run(
            ["git", "status", "--short"],
            capture_output=True,
            text=True,
            timeout=10,
        )
    except Exception:
        return []
    if completed.returncode != 0:
        return []
    return [line for line in completed.stdout.splitlines() if line.strip()]


def write_run_report(
    path: Path,
    *,
    config: Config,
    task: str,
    result: Dict[str, Any],
    trace_path: Path | None = None,
    git_status_before: List[str] | None = None,
    git_status_after: List[str] | None = None,
) -> None:
    """Write a human-readable Markdown report for one agent run."""
    metadata = result.get("metadata", {})
    steps = result.get("steps", [])
    path.parent.mkdir(parents=True, exist_ok=True)

    lines = [
        "# DM-Code-Agent Run Report",
        "",
        "## Task",
        "",
        task,
        "",
        "## Runtime",
        "",
        f"- Provider: `{config.provider}`",
        f"- Model: `{config.model}`",
        f"- Status: `{metadata.get('status', 'unknown')}`",
        f"- Duration: `{float(metadata.get('duration_seconds', 0.0)):.2f}s`",
        f"- Steps: `{len(steps)}`",
        f"- Tool errors: `{metadata.get('tool_error_count', 0)}`",
        f"- Replans: `{metadata.get('replan_count', 0)}`",
        f"- Advanced features: `{format_advanced_feature_status(config)}`",
    ]
    if trace_path:
        lines.append(f"- Trace: `{trace_path}`")

    before = git_status_before or []
    after = git_status_after or []
    lines.extend(["", "## Workspace Status", ""])
    if not before and not after:
        lines.append("No git status information available.")
    else:
        lines.append(f"- Dirty entries before run: `{len(before)}`")
        lines.append(f"- Dirty entries after run: `{len(after)}`")
        if before:
            lines.extend(["", "Before:", ""])
            lines.extend(f"- `{line}`" for line in before)
        if after:
            lines.extend(["", "After:", ""])
            lines.extend(f"- `{line}`" for line in after)

    lines.extend(
        [
            "",
            "## Steps",
            "",
            "| # | Action | Observation |",
            "| ---: | --- | --- |",
        ]
    )
    for index, step in enumerate(steps, start=1):
        observation = str(step.get("observation", "")).replace("\n", " ")
        if len(observation) > 180:
            observation = observation[:177] + "..."
        lines.append(f"| {index} | `{step.get('action', '')}` | {observation} |")

    lines.extend(
        [
            "",
            "## Final Answer",
            "",
            str(result.get("final_answer", "")),
            "",
        ]
    )
    path.write_text("\n".join(lines), encoding="utf-8")


def format_duration(seconds: Any) -> str:
    try:
        value = float(seconds)
    except (TypeError, ValueError):
        return "-"
    if value < 60:
        return f"{value:.1f}s"
    minutes, remainder = divmod(int(value), 60)
    return f"{minutes}m {remainder}s"


def default_report_path(task: str) -> Path:
    timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
    slug = re.sub(r"[^a-z0-9]+", "-", task.lower()).strip("-")[:32] or "task"
    return Path(".dm_agent") / "run_reports" / f"{timestamp}-{slug}.md"


def format_run_status(status: Any) -> str:
    value = str(status or "unknown")
    labels = {
        "success": "完成",
        "running": "运行中",
        "max_steps_exceeded": "达到步骤上限",
    }
    return labels.get(value, value)


def format_step_input(value: Any, *, limit: int = 90) -> str:
    if value in (None, "", {}):
        return "-"
    try:
        text = json.dumps(value, ensure_ascii=False)
    except TypeError:
        text = str(value)
    return UI.truncate(text, limit)


def display_completion_screen(
    result: Dict[str, Any],
    *,
    task: str | None = None,
    context_status: str | None = None,
    trace_path: Path | None = None,
    report_path: Path | None = None,
    clear_screen: bool = False,
    review_hint: bool = False,
) -> None:
    if clear_screen:
        UI.clear()

    metadata = result.get("metadata", {})
    steps = result.get("steps", [])
    final_answer = str(result.get("final_answer", "")).strip()
    completion_summary = str(metadata.get("completion_summary", "")).strip()
    summary = completion_summary or final_answer or "任务已结束。"

    UI.banner("任务结束", "最终总结在这里；运行过程已收纳，可按需查看。")
    UI.panel("最终总结", summary, color=Fore.GREEN)
    if final_answer and final_answer != summary:
        UI.panel("最终答案", final_answer, color=Fore.CYAN)

    rows: List[tuple[str, Any]] = [
        ("状态", format_run_status(metadata.get("status"))),
        ("步骤", len(steps)),
        ("耗时", format_duration(metadata.get("duration_seconds"))),
        ("工具错误", metadata.get("tool_error_count", 0)),
        ("重规划", metadata.get("replan_count", 0)),
    ]
    if task:
        rows.insert(0, ("任务", UI.truncate(task, 96)))
    memory_count = metadata.get("memory_items")
    if memory_count is not None:
        rows.append(("记忆", f"{memory_count} items"))
    if context_status:
        rows.append(("会话", context_status))
    if trace_path:
        rows.append(("Trace", trace_path))
    if report_path:
        rows.append(("Report", report_path))
    elif not review_hint:
        rows.append(("过程", "使用 --report 保存 Markdown，或 --trace 保存 JSONL"))
    UI.key_values("运行概览", rows)

    if review_hint:
        UI.status("info", "查看过程", "Enter 继续 | v 分页查看步骤 | s 保存 Markdown 报告")


def display_step_page(result: Dict[str, Any], *, page: int, page_size: int) -> None:
    steps = result.get("steps", [])
    total = len(steps)
    page_count = max(1, (total + page_size - 1) // page_size)
    start = page * page_size
    page_steps = steps[start : start + page_size]

    UI.clear()
    UI.banner("运行过程", f"第 {page + 1}/{page_count} 页，共 {total} 步")
    if not page_steps:
        UI.status("info", "本轮没有记录到步骤")
        return

    if UI.rich_enabled():
        table = Table(
            show_header=True,
            header_style="bold white",
            box=box.SIMPLE_HEAD,
            border_style="bright_black",
            pad_edge=False,
        )
        table.add_column("#", justify="right", style="cyan", no_wrap=True, width=4)
        table.add_column("状态", no_wrap=True, width=6)
        table.add_column("动作", style="bold white", no_wrap=True, width=18)
        table.add_column("输入", style="bright_black", ratio=1, overflow="fold")
        table.add_column("观察", style="bright_black", ratio=2, overflow="fold")
        for offset, step in enumerate(page_steps, start=start + 1):
            observation = str(step.get("observation", ""))
            action = str(step.get("action", ""))
            failed = action == "error" or "failed" in observation.lower()
            status = Text("ERR", style=UI.RICH_STYLES["error"]) if failed else Text("OK", "green")
            table.add_row(
                str(offset),
                status,
                action,
                format_step_input(step.get("action_input")),
                UI.truncate(observation, 180),
            )
        RICH_CONSOLE.print(
            Panel(
                table,
                title=Text(" 步骤摘要 ", style="bold magenta"),
                title_align="left",
                border_style="bright_black",
                box=box.ROUNDED,
                padding=(1, 2),
            )
        )
    else:
        UI.section("步骤摘要")
        for offset, step in enumerate(page_steps, start=start + 1):
            action = str(step.get("action", ""))
            observation = UI.truncate(step.get("observation", ""), 180)
            print(
                f"  {UI.paint(f'{offset:>3}', Fore.CYAN, bright=True)}  "
                f"{UI.paint(action, Fore.WHITE, bright=True)}"
            )
            print(
                f"       {UI.paint('input', Fore.WHITE, dim=True)}  {format_step_input(step.get('action_input'))}"
            )
            print(f"       {UI.paint('observe', Fore.WHITE, dim=True)}  {observation}")
            print()

    UI.status("info", "导航", "n 下一页 | p 上一页 | q 返回总结")


def browse_run_steps(result: Dict[str, Any], *, page_size: int = 12) -> None:
    steps = result.get("steps", [])
    if not steps:
        UI.clear()
        UI.status("info", "本轮没有可查看的步骤")
        UI.pause()
        return

    page = 0
    page_count = max(1, (len(steps) + page_size - 1) // page_size)
    while True:
        display_step_page(result, page=page, page_size=page_size)
        choice = UI.prompt_line("操作", default="q").strip().lower()
        if choice in {"", "q", "quit", "back"}:
            return
        if choice in {"n", "next"}:
            page = min(page + 1, page_count - 1)
            continue
        if choice in {"p", "prev", "previous"}:
            page = max(page - 1, 0)
            continue
        UI.status("warn", "未知操作", "请输入 n、p 或 q")
        UI.pause()


def review_completed_run(
    result: Dict[str, Any],
    *,
    config: Config,
    task: str,
    trace_path: Path | None = None,
    report_path: Path | None = None,
    context_status: str | None = None,
    git_status_before: List[str] | None = None,
    git_status_after: List[str] | None = None,
    interactive: bool = True,
) -> Path | None:
    current_report_path = report_path
    while True:
        display_completion_screen(
            result,
            task=task,
            context_status=context_status,
            trace_path=trace_path,
            report_path=current_report_path,
            clear_screen=True,
            review_hint=interactive,
        )
        if not interactive:
            return current_report_path

        choice = UI.prompt_line("操作", default="").strip().lower()
        if choice in {"", "c", "continue"}:
            return current_report_path
        if choice in {"v", "view", "steps", "process", "过程"}:
            browse_run_steps(result)
            continue
        if choice in {"s", "save", "report", "保存"}:
            current_report_path = current_report_path or default_report_path(task)
            write_run_report(
                current_report_path,
                config=config,
                task=task,
                result=result,
                trace_path=trace_path,
                git_status_before=git_status_before,
                git_status_after=git_status_after,
            )
            UI.status("ok", "Report 已保存", str(current_report_path))
            UI.pause()
            continue
        UI.status("warn", "未知操作", "Enter 继续，v 查看步骤，s 保存报告")
        UI.pause()


def create_step_callback(show_steps: bool):
    """创建步骤回调函数，用于实时打印 agent 执行状态"""

    def callback(step_num: int, step: Any) -> None:
        if show_steps:
            UI.panel(
                f"Step {step_num:02d} | {step.action}",
                (
                    f"thought  {step.thought}\n"
                    f"input    "
                    f"{json.dumps(step.action_input, ensure_ascii=False) if step.action_input else '-'}\n"
                    f"observe  {UI.truncate(step.observation, 360)}"
                ),
                color=Fore.MAGENTA,
            )
        else:
            status = "error" if step.action == "error" else "ok"
            should_print = (
                step_num == 1
                or step_num % 10 == 0
                or status == "error"
                or step.action in {"finish", "task_complete"}
            )
            if not should_print:
                return
            if UI.rich_enabled():
                text = Text()
                text.append(f"{step_num:02d}", style="cyan")
                text.append("  ")
                if status == "error":
                    text.append(" ERR ", style=UI.RICH_STYLES["error"])
                else:
                    text.append(" OK  ", style=UI.RICH_STYLES["ok"])
                text.append(" ")
                text.append(step.action, style="bold white")
                if step.action_input:
                    text.append("  input", style="bright_black")
                RICH_CONSOLE.print(Padding(text, (0, 0, 0, 1)))
                return

            action = UI.paint(step.action, Fore.WHITE, bright=True)
            marker = UI.paint(f"{step_num:02d}", Fore.CYAN, bright=True)
            result = UI.paint(
                "err" if status == "error" else "ok", Fore.RED if status == "error" else Fore.GREEN
            )
            suffix = UI.paint(" input", Fore.WHITE, dim=True) if step.action_input else ""
            print(f"  {marker}  {result}  {action}{suffix}")

    return callback


def create_agent(
    config: Config,
    client: Any,
    tools: List[Tool],
    *,
    step_callback: Any = None,
    skill_manager: SkillManager | None = None,
    trace_writer: TraceWriter | None = None,
) -> ReactAgent:
    """Create a ReactAgent with the CLI's default-off advanced switches."""
    advanced = resolve_advanced_features(config)
    return ReactAgent(
        client,
        tools,
        max_steps=config.max_steps,
        temperature=config.temperature,
        step_callback=step_callback,
        skill_manager=skill_manager,
        trace_writer=trace_writer,
        enable_reflexion=advanced["reflexion"],
        max_trials=config.max_trials,
        critic=CriticAgent(client) if advanced["critic"] else None,
        enable_adaptive_replanning=advanced["adaptive_replanning"],
        max_replans=config.max_replans,
        enable_repeated_failure_policy_experiment=(advanced["repeated_failure_policy_experiment"]),
    )


def format_agent_context_status(agent: ReactAgent) -> str:
    stats = agent.get_context_stats()
    return (
        f"history={stats['conversation_messages']} messages | "
        f"memory={stats['memory_items']} items | "
        f"compression={'on' if stats['compression_enabled'] else 'off'}"
    )


def multi_turn_conversation(
    config: Config, tools: List[Tool], skill_manager: SkillManager | None = None
) -> None:
    """多轮对话模式"""
    UI.section(
        "多轮对话",
        "同一个 agent 实例会保留当前会话，并使用本地原子记忆整理旧上下文。"
        "这份记忆只在当前 CLI 进程内连续；输入 exit 退出，输入 reset 重置。",
    )

    try:
        # 创建客户端和智能体
        client = create_llm_client(
            provider=config.provider,
            api_key=config.api_key,
            model=config.model,
            base_url=config.base_url,
        )
        step_callback = create_step_callback(config.show_steps)

        agent = create_agent(
            config,
            client,
            tools,
            step_callback=step_callback,
            skill_manager=skill_manager,
        )

        conversation_count = 0

        while True:
            UI.section(f"对话 {conversation_count + 1}")
            UI.status("info", "当前会话记忆", format_agent_context_status(agent))
            task = UI.ask("请输入任务 (exit 退出，reset 重置历史)").strip()

            if not task:
                UI.status("error", "任务描述不能为空")
                continue

            if task.lower() == "exit":
                UI.status("info", "退出多轮对话模式")
                break

            if task.lower() == "reset":
                agent.reset_conversation()
                conversation_count = 0
                UI.status("ok", "对话历史和本地记忆已重置")
                continue

            try:
                UI.status("run", "正在执行任务")

                # 执行任务
                git_status_before = collect_git_status()
                result = agent.run(task)
                git_status_after = collect_git_status()
                conversation_count += 1

                metadata = result.get("metadata", {})
                context_status = (
                    f"{format_agent_context_status(agent)} | "
                    f"injections={metadata.get('memory_injection_count', 0)}"
                )
                review_completed_run(
                    result,
                    config=config,
                    task=task,
                    context_status=context_status,
                    git_status_before=git_status_before,
                    git_status_after=git_status_after,
                    interactive=True,
                )

            except LLMError as e:
                UI.status("error", "API 错误", str(e))
            except KeyboardInterrupt:
                UI.status("info", "退出多轮对话模式")
                break
            except Exception as e:
                UI.status("error", "发生错误", str(e))

    except Exception as e:
        UI.status("error", "初始化错误", str(e))


def execute_task(
    config: Config, tools: List[Tool], skill_manager: SkillManager | None = None
) -> None:
    """执行任务"""
    UI.section("执行新任务")
    task = UI.ask("请输入任务描述").strip()

    if not task:
        UI.status("error", "任务描述不能为空")
        return

    try:
        # 创建客户端和智能体
        client = create_llm_client(
            provider=config.provider,
            api_key=config.api_key,
            model=config.model,
            base_url=config.base_url,
        )

        # 创建步骤回调函数
        step_callback = create_step_callback(config.show_steps)

        agent = create_agent(
            config,
            client,
            tools,
            step_callback=step_callback,
            skill_manager=skill_manager,
        )

        UI.status("run", "正在执行任务")

        # 执行任务
        git_status_before = collect_git_status()
        result = agent.run(task)
        git_status_after = collect_git_status()

        review_completed_run(
            result,
            config=config,
            task=task,
            git_status_before=git_status_before,
            git_status_after=git_status_after,
            interactive=True,
        )

    except LLMError as e:
        UI.status("error", "API 错误", str(e))
    except KeyboardInterrupt:
        UI.status("warn", "任务已被用户中断")
    except Exception as e:
        UI.status("error", "发生错误", str(e))


def interactive_mode(config: Config) -> int:
    """交互式菜单模式"""
    print_welcome()

    # 初始化 MCP 管理器
    mcp_config = load_mcp_config()
    mcp_manager = MCPManager(mcp_config)

    # 启动所有启用的 MCP 服务器
    UI.status("run", "正在加载 MCP 服务器")
    started_count = mcp_manager.start_all()
    if started_count > 0:
        UI.status("ok", f"成功启动 {started_count} 个 MCP 服务器")
    else:
        UI.status("info", "未启用 MCP 服务器")

    # 获取包含 MCP 工具的工具列表
    mcp_tools = mcp_manager.get_tools()
    tools = default_tools(include_mcp=True, mcp_tools=mcp_tools)

    if mcp_tools:
        UI.status("ok", f"加载了 {len(mcp_tools)} 个 MCP 工具")

    # 初始化技能管理器
    skill_manager = SkillManager()
    skill_count = skill_manager.load_all()
    if skill_count > 0:
        UI.status("ok", f"加载了 {skill_count} 个技能")
    else:
        UI.status("info", "未加载任何技能")

    try:
        while True:
            try:
                print_menu()
                choice = UI.ask(
                    "选择操作 (1-6)",
                    choices=["1", "2", "3", "4", "5", "6"],
                    show_choices=False,
                ).strip()

                if choice == "1":
                    # 执行新任务
                    execute_task(config, tools, skill_manager)

                elif choice == "2":
                    # 多轮对话模式
                    multi_turn_conversation(config, tools, skill_manager)

                elif choice == "3":
                    # 查看工具列表
                    show_tools(tools)

                elif choice == "4":
                    # 配置设置
                    configure_settings(config)

                elif choice == "5":
                    # 查看技能列表
                    show_skills(skill_manager)

                elif choice == "6":
                    # 退出程序
                    UI.status("info", "感谢使用，再见")
                    return 0

                else:
                    UI.status("error", "无效的选择", "请输入 1-6")

            except KeyboardInterrupt:
                UI.status("info", "感谢使用，再见")
                return 0
            except EOFError:
                UI.status("info", "感谢使用，再见")
                return 0
            except Exception as e:
                UI.status("error", "发生错误", str(e))

    finally:
        # 清理 MCP 资源
        UI.status("run", "正在关闭 MCP 服务器")
        mcp_manager.stop_all()
        UI.status("ok", "MCP 服务器已关闭")


def run_single_task(
    config: Config,
    task: str,
    *,
    trace_path: Path | None = None,
    trace_llm_io: bool = False,
    report_path: Path | None = None,
) -> int:
    """运行单个任务（命令行模式）"""
    # 初始化 MCP
    mcp_config = load_mcp_config()
    mcp_manager = MCPManager(mcp_config)
    trace_writer: TraceWriter | None = None

    try:
        # 启动 MCP 服务器
        started_count = mcp_manager.start_all()
        if started_count > 0:
            UI.status("ok", f"启动了 {started_count} 个 MCP 服务器")

        # 获取工具
        mcp_tools = mcp_manager.get_tools()
        tools = default_tools(include_mcp=True, mcp_tools=mcp_tools)

        # 初始化技能管理器
        skill_manager = SkillManager()
        skill_count = skill_manager.load_all()

        client = create_llm_client(
            provider=config.provider,
            api_key=config.api_key,
            model=config.model,
            base_url=config.base_url,
        )
        advanced = resolve_advanced_features(config)
        if trace_path:
            trace_writer = TraceWriter(trace_path, capture_llm_io=trace_llm_io)
            trace_writer.record(
                "runtime",
                {
                    "provider": config.provider,
                    "model": config.model,
                    "base_url": config.base_url,
                    "max_steps": config.max_steps,
                    "temperature": config.temperature,
                    "show_steps": config.show_steps,
                    "mcp_started_count": started_count,
                    "mcp_tool_count": len(mcp_tools),
                    "skill_count": skill_count,
                    "trace_llm_io": trace_llm_io,
                    "reflexion_enabled": advanced["reflexion"],
                    "max_trials": config.max_trials,
                    "critic_enabled": advanced["critic"],
                    "adaptive_replanning_enabled": advanced["adaptive_replanning"],
                    "max_replans": config.max_replans,
                    "repeated_failure_policy_experiment_enabled": (
                        advanced["repeated_failure_policy_experiment"]
                    ),
                    "evolution_enabled": advanced["evolution"],
                },
            )

        # 创建步骤回调函数
        step_callback = create_step_callback(config.show_steps)

        agent = create_agent(
            config,
            client,
            tools,
            step_callback=step_callback,
            skill_manager=skill_manager,
            trace_writer=trace_writer,
        )

        UI.panel(
            "Run",
            (
                f"task       {task}\n"
                f"provider   {config.provider}\n"
                f"model      {config.model}\n"
                f"max steps  {config.max_steps}\n"
                f"advanced   {format_advanced_feature_status(config)}"
            ),
            color=Fore.CYAN,
        )

        git_status_before = collect_git_status()
        result = agent.run(task)
        git_status_after = collect_git_status()
        if report_path:
            write_run_report(
                report_path,
                config=config,
                task=task,
                result=result,
                trace_path=trace_path,
                git_status_before=git_status_before,
                git_status_after=git_status_after,
            )

        review_completed_run(
            result,
            config=config,
            task=task,
            trace_path=trace_path,
            report_path=report_path,
            git_status_before=git_status_before,
            git_status_after=git_status_after,
            interactive=False,
        )

        return 0

    except LLMError as e:
        if trace_writer:
            trace_writer.record("run_error", {"error_type": "LLMError", "message": str(e)})
        print(f"{UI.paint('[ERR] API 错误', Fore.RED, bright=True)} {e}", file=sys.stderr)
        return 1
    except Exception as e:
        if trace_writer:
            trace_writer.record(
                "run_error",
                {"error_type": type(e).__name__, "message": str(e)},
            )
        print(f"{UI.paint('[ERR] 发生错误', Fore.RED, bright=True)} {e}", file=sys.stderr)
        return 1
    finally:
        if trace_writer:
            trace_writer.close()
        # 清理 MCP 资源
        mcp_manager.stop_all()


def main(argv: Any = None) -> int:
    """主入口函数"""
    configure_console_encoding()
    load_dotenv()
    args = parse_args(argv if argv is not None else sys.argv[1:])

    feature_error = validate_feature_args(args)
    if feature_error:
        print(UI.paint("[ERR] 高级功能参数无效", Fore.RED, bright=True), file=sys.stderr)
        print(feature_error, file=sys.stderr)
        return 2

    # 如果没有提供 API 密钥，尝试根据提供商获取
    if not args.api_key:
        args.api_key = get_api_key_for_provider(args.provider)

    # 检查 API 密钥
    if not args.api_key:
        print(UI.paint("[ERR] 缺少 API 密钥", Fore.RED, bright=True), file=sys.stderr)
        print(f"请提供 --api-key 或设置环境变量 {args.provider.upper()}_API_KEY。", file=sys.stderr)
        return 2

    # 获取提供商的默认配置
    provider_defaults = PROVIDER_DEFAULTS.get(args.provider, {})

    # 如果没有指定 base_url，使用提供商默认值
    if not args.base_url:
        args.base_url = provider_defaults.get("base_url", "https://api.deepseek.com")

    # 如果模型是默认的 deepseek-chat 但提供商不是 deepseek，更新模型
    if args.model == "deepseek-chat" and args.provider != "deepseek":
        args.model = provider_defaults.get("model", args.model)

    # 创建配置
    config = Config(
        api_key=args.api_key,
        provider=args.provider,
        model=args.model,
        base_url=args.base_url,
        max_steps=args.max_steps,
        temperature=args.temperature,
        show_steps=args.show_steps,
        enable_reflexion=args.enable_reflexion,
        max_trials=args.max_trials,
        enable_critic=args.enable_critic,
        enable_adaptive_replanning=args.enable_adaptive_replanning,
        max_replans=args.max_replans,
        enable_repeated_failure_policy_experiment=(args.enable_repeated_failure_policy_experiment),
        enable_evolution=args.enable_evolution,
    )

    # 如果提供了任务参数，直接执行任务
    if args.task:
        return run_single_task(
            config,
            args.task,
            trace_path=args.trace,
            trace_llm_io=args.trace_llm_io,
            report_path=args.report,
        )

    # 如果指定了交互模式或没有提供任务，进入交互式菜单
    if args.interactive or not args.task:
        return interactive_mode(config)

    return 0


if __name__ == "__main__":
    raise SystemExit(main())