feat: 支持 shell 变量定义（VAR=value command）、export、多行命令

完整 shell 能力支持：
- VAR=value command 语法（自动跳过赋值检测实际 binary）
- export VAR=value 支持（export 加入只读白名单）
- 多行命令（\n 分割后逐段检查）
- && || ; 等连接符的武器级命令检测
- _split_into_commands 重写：使用 while 索引遍历，正确处理 && / || / |
- _command_has_dangerous_cmds 跳过 VAR=value 和 export 前缀
- classify_agent_command 检测 VAR=value kubectl 语法路由到 kubectl 分类器
- execute_command 支持变量赋值后的 binary 定位
- 新增 shell 内置白名单：echo printf cd pwd export source . sleep timeout
  nohup unset shift return local readonly trap type command set shopt eval
  exec alias unalias declare typeset read mapfile
- 白名单只读命令新增：timeout、nohup、sleep 等

.gitignore

@@ -15,6 +15,12 @@ venv/
 # Environment
 .env
 
+# Runtime data
+data/
+*.db
+*.db-shm
+*.db-wal
+
 # IDE
 .vscode/
 .idea/

README.md

@@ -13,6 +13,21 @@
 - **健康评分**：100 分制量化集群健康度（严重问题 -15 分，警告 -5 分）
 - **AI 智能分析**：接入 OpenAI 兼容接口，自动分析诊断报告
   - 根因定位、影响评估、修复命令、优先级排序、预防建议
+- **AI 自主诊断（双执行模式）**：启动前可选择命令执行方式
+  - **手动执行命令**：Agent 只生成命令，用户复制到自己的终端执行，服务端不会运行
+  - **AI 执行命令**：自动执行只读诊断命令，覆盖 kubectl 与节点宿主机两个维度
+    - kubectl: `get`、`describe`、`logs`、`top` 等只读命令
+    - 系统诊断: `systemctl status`、`journalctl`、`crictl ps`、`df`、`free`、`ss`、`ip addr` 等只读命令
+  - 实时展示思考步骤、执行命令和真实输出
+  - 即使选择 AI 执行，修改命令人工确认：
+    - kubectl: `apply`、`delete`、`patch`、`scale`、`rollout`、`exec` 等
+    - 系统: `systemctl restart/stop/start` 等
+  - 命令白名单严格限制，禁止 shell、管道、重定向和多命令拼接
+- **历史记录持久化**：使用 SQLite 保存全部诊断与处理轨迹
+  - 保存完整诊断结果、文本报告、AI 分析结果
+  - 保存 Agent 思考、命令、真实输出、审批和执行结果
+  - 支持按状态、命名空间查询并查看完整详情
+  - 服务重启后仍可读取历史和恢复最近一次诊断
 - **AI 追问对话**：基于诊断上下文多轮追问，深入排查问题
 - **Web 仪表盘**：Vue3 + Element Plus，实时展示诊断结果
 
@@ -44,6 +59,9 @@ PORT=8900
 
 # Kubeconfig 路径 (留空使用默认 ~/.kube/config)
 KUBECONFIG_PATH=
+
+# 历史记录 SQLite 文件（K8S 部署时请挂载持久卷到该路径）
+HISTORY_DB_PATH=/app/data/history.db
 ```
 
 ## API 接口
@@ -55,6 +73,10 @@ KUBECONFIG_PATH=
 | POST | `/api/diagnose` | 执行诊断 `{"namespace":"", "checks":[]}` |
 | GET | `/api/diagnosis/latest` | 获取最近诊断结果 |
 | POST | `/api/analyze` | AI 分析诊断结果 `{"question":""}` |
+| GET | `/api/agent/diagnose/stream` | 自主诊断 SSE 流，`execution_mode=manual|ai` |
+| POST | `/api/agent/approve` | 批准或拒绝待执行命令 `{"approval_id":"...","approved":true}` |
+| GET | `/api/history/diagnoses` | 查询全部诊断历史，支持 `status`、`namespace` 过滤 |
+| GET | `/api/history/diagnoses/{id}` | 查询诊断详情及全部 AI/Agent 处理轨迹 |
 | POST | `/api/chat` | AI 多轮对话 `{"messages":[...]}` |
 | GET | `/api/report` | 获取诊断报告文本 |
 
@@ -72,6 +94,7 @@ k8smanager-cli/
 │   ├── main.py          # FastAPI 主应用
 │   ├── config.py        # 配置管理
 │   ├── diagnosis.py     # K8S 诊断引擎 (9 大检查项)
+│   ├── agent_tools.py   # Agent 命令工具与安全策略 (kubectl + 系统诊断白名单)
 │   └── ai_agent.py      # AI Agent 分析模块
 ├── frontend/
 │   ├── src/App.vue      # 仪表盘界面

backend/agent_tools.py

@@ -0,0 +1,576 @@
+"""AI Agent 可用的命令工具和安全策略。
+
+支持两类命令：
+1. kubectl 命令 - 通过 classify_kubectl_command 分类
+2. 系统诊断 shell 命令 - 通过 classify_shell_command 分类 (白名单)
+
+classify_agent_command 统一分发，execute_command 自动选择执行方式。
+所有命令均通过 subprocess_exec 执行 (无 shell)，禁止管道、重定向和命令组合。
+"""
+import asyncio
+import shlex
+from dataclasses import dataclass
+
+from backend.diagnosis import KUBECTL
+
+# ── kubectl 命令分类 ──────────────────────────────────────
+
+READ_ONLY_VERBS = {
+    "api-resources", "api-versions", "auth", "cluster-info", "config", "describe",
+    "diff", "explain", "get", "logs", "top", "version", "wait",
+}
+WRITE_OR_INTERACTIVE_VERBS = {
+    "annotate", "apply", "attach", "autoscale", "cordon", "cp", "create",
+    "debug", "delete", "drain", "edit", "exec", "expose", "label", "patch",
+    "port-forward", "replace", "rollout", "run", "scale", "set", "taint",
+    "uncordon",
+}
+GLOBAL_FLAGS_WITH_VALUE = {
+    "--as", "--as-group", "--as-uid", "--cache-dir", "--certificate-authority",
+    "--client-certificate", "--client-key", "--cluster", "--context", "--kubeconfig",
+    "--namespace", "-n", "--profile", "--request-timeout", "--server", "-s",
+    "--tls-server-name", "--token", "--user",
+}
+FORBIDDEN_TOKENS = set()  # 全部 shell 操作符均允许；安全靠白名单+危险命令检查
+# 永远拒绝的武器级危险命令（无论是否在白名单中）
+_ALWAYS_DENY_CMDS = {"sh", "bash", "dash", "zsh",
+                     "dd", "format", "mkfs", "fdisk", "parted"}
+
+
+
+@dataclass(frozen=True)
+class CommandDecision:
+    mode: str          # "read" (只读，自动执行) 或 "write" (修改，需审批)
+    requires_approval: bool
+    verb: str          # kubectl 子命令 或 shell 命令名
+    args: list[str]
+    needs_shell: bool = False  # True 时需用 shell=True 执行
+
+
+def _has_shell_operators(command: str) -> bool:
+    """检测命令是否包含 shell 操作符（管道、重定向、分号、逻辑连接符、多行等）。"""
+    stripped = command.strip()
+    if not stripped:
+        return False
+    if "\n" in stripped or "\r" in stripped:
+        return True  # 多行命令需要 shell=True
+    in_single = in_double = False
+    for i, ch in enumerate(stripped):
+        if ch == "'" and not in_double:
+            in_single = not in_single
+        elif ch == '"' and not in_single:
+            in_double = not in_double
+        if in_single or in_double:
+            continue
+        if ch in ('|', '>', ';', '&', '<'):
+            # 单纯 & 不算（后台任务标记）
+            if ch == '&':
+                if i + 1 < len(stripped) and stripped[i + 1] in ('&',):
+                    return True  # &&
+                continue
+            return True
+    return False
+
+
+def _split_into_commands(command: str) -> list[str]:
+    """按 shell 连接符（; && || | \n 等）分割命令块。
+
+    返回每个可执行段，忽略引号内的操作符。重定向符号 > < 不分割。
+    """
+    segments = []
+    buf = []
+    in_single = in_double = False
+    i = 0
+    while i < len(command):
+        ch = command[i]
+        if ch == "'" and not in_double:
+            in_single = not in_single
+        elif ch == '"' and not in_single:
+            in_double = not in_double
+        if not in_single and not in_double:
+            if ch in (';', '\n'):
+                if buf:
+                    segments.append(''.join(buf).strip())
+                    buf = []
+                i += 1
+                continue
+            if ch == '|':
+                if i + 1 < len(command) and command[i + 1] == '|':
+                    # || — 分割命令
+                    if buf:
+                        segments.append(''.join(buf).strip())
+                        buf = []
+                    i += 2
+                    continue
+                # 单 | — 分割命令
+                if buf:
+                    segments.append(''.join(buf).strip())
+                    buf = []
+                i += 1
+                continue
+            if ch == '&':
+                if i + 1 < len(command) and command[i + 1] == '&':
+                    # && — 分割命令
+                    if buf:
+                        segments.append(''.join(buf).strip())
+                        buf = []
+                    i += 2
+                    continue
+                # 单独 & — 保留
+        buf.append(ch)
+        i += 1
+    if buf:
+        segments.append(''.join(buf).strip())
+    return [s for s in segments if s]
+
+
+def _command_has_dangerous_cmds(command: str) -> bool:
+    """检查命令中任何可执行段是否包含武器级危险命令。"""
+    for segment in _split_into_commands(command):
+        try:
+            seg_args = shlex.split(segment)
+        except ValueError:
+            continue
+        if not seg_args:
+            continue
+        # 跳过 VAR=value 赋值
+        idx = 0
+        while idx < len(seg_args) and "=" in seg_args[idx] and not seg_args[idx].startswith("-"):
+            idx += 1
+        if idx < len(seg_args) and seg_args[idx] in _ALWAYS_DENY_CMDS:
+            return True
+        # 也检查 export VAR=value 后的命令
+        if idx < len(seg_args) and seg_args[idx] == "export":
+            idx2 = idx + 1
+            while idx2 < len(seg_args) and "=" in seg_args[idx2]:
+                idx2 += 1
+            if idx2 < len(seg_args) and seg_args[idx2] in _ALWAYS_DENY_CMDS:
+                return True
+    return False
+
+
+def _parse_command(command: str) -> list[str]:
+    """解析命令，检查合法性。
+
+    允许所有 shell 操作符（| > ; && || < 等）以及多行命令，安全靠白名单检查。
+    武器级危险命令 sh/bash/dd/mkfs 等在 classify 阶段处理。
+    """
+    if not isinstance(command, str) or not command.strip():
+        raise ValueError("命令不能为空")
+    if _command_has_dangerous_cmds(command):
+        raise ValueError("命令包含不允许的危险命令（sh/bash/dd/mkfs 等），请手动执行")
+    try:
+        return shlex.split(command)
+    except ValueError as exc:
+        raise ValueError(f"命令格式错误: {exc}") from exc
+
+
+def parse_kubectl_command(command: str) -> list[str]:
+    """仅解析单条 kubectl 命令，不允许 shell、重定向或命令组合。"""
+    parts = _parse_command(command)
+    if not parts or parts[0] != "kubectl":
+        raise ValueError("只允许执行 kubectl 命令")
+    if len(parts) < 2:
+        raise ValueError("kubectl 命令缺少操作类型")
+    return parts[1:]
+
+
+def _find_verb(args: list[str]) -> str:
+    index = 0
+    while index < len(args):
+        arg = args[index]
+        if arg == "--":
+            break
+        if arg in GLOBAL_FLAGS_WITH_VALUE:
+            index += 2
+            continue
+        if any(arg.startswith(flag + "=") for flag in GLOBAL_FLAGS_WITH_VALUE if flag.startswith("--")):
+            index += 1
+            continue
+        if arg.startswith("-"):
+            index += 1
+            continue
+        return arg.lower()
+    raise ValueError("无法识别 kubectl 操作类型")
+
+
+READ_ONLY_CONFIG_SUBCOMMANDS = {"current-context", "get-clusters", "get-contexts", "get-users", "view"}
+
+
+def classify_kubectl_command(command: str) -> CommandDecision:
+    args = parse_kubectl_command(command)
+    verb = _find_verb(args)
+    if verb == "config":
+        try:
+            verb_index = args.index(next(arg for arg in args if arg.lower() == "config"))
+            subcommand = args[verb_index + 1].lower()
+        except (StopIteration, ValueError, IndexError):
+            return CommandDecision("write", True, verb, args)
+        if subcommand in READ_ONLY_CONFIG_SUBCOMMANDS:
+            return CommandDecision("read", False, verb, args)
+        return CommandDecision("write", True, verb, args)
+    if verb in READ_ONLY_VERBS:
+        return CommandDecision("read", False, verb, args)
+    if verb in WRITE_OR_INTERACTIVE_VERBS:
+        return CommandDecision("write", True, verb, args)
+    return CommandDecision("write", True, verb, args)
+
+
+# ── 系统诊断 shell 命令分类 (白名单) ─────────────────────
+
+# 只读诊断命令白名单：value 为 None 表示该命令本身只读，任意参数均可；
+# 为集合时，第一个非 flag 参数必须属于该集合。
+SHELL_READ_ONLY_COMMANDS: dict[str, set[str] | None] = {
+    "systemctl": {
+        "status", "is-active", "is-enabled", "is-failed", "list-units",
+        "list-sockets", "list-jobs", "show", "cat", "list-unit-files",
+        "list-dependencies",
+    },
+    "journalctl": None,  # 只读，任意参数 (-u, -n, --no-pager, --since 等)
+    "crictl": {"ps", "inspect", "logs", "info", "version", "images", "stats"},
+    "ctr": None,
+    "nerdctl": {"ps", "logs", "inspect", "info", "version", "images", "stats", "top", "events", "system", "namespace", "network", "volume"},
+    "docker": {"ps", "logs", "inspect", "stats", "version", "info", "images", "top"},
+    "df": None, "du": None, "free": None, "uptime": None, "uname": None, "hostname": None,
+    "ip": {"addr", "address", "route", "link", "neighbor", "neigh", "rule", "maddr", "monitor", "tunnel", "tuntap"},
+    "ss": None, "netstat": None, "lsblk": None, "ls": None,
+    "cat": None, "head": None, "tail": None, "wc": None, "grep": None, "egrep": None, "fgrep": None,
+    "ps": None, "pstree": None, "mount": None, "lsmod": None, "lscpu": None, "lsof": None,
+    "dmesg": None, "ping": None, "nslookup": None, "dig": None, "getent": None,
+    "timedatectl": None, "date": None, "stat": None, "file": None, "blkid": None,
+    "top": None, "htop": None, "iostat": None, "vmstat": None, "mpstat": None,
+    "pidstat": None, "sar": None, "nproc": None, "env": None, "which": None,
+    "readlink": None, "realpath": None, "find": None, "sort": None, "uniq": None,
+    "awk": None, "sed": None, "cut": None, "tr": None, "tee": None,
+    "basename": None, "dirname": None, "xargs": None, "bc": None, "expr": None,
+    "curl": None, "wget": None, "traceroute": None, "tracepath": None, "mtr": None,
+    "nc": None, "nmap": None, "telnet": None, "tcpdump": None, "ethtool": None,
+    "lspci": None, "lsusb": None, "dmidecode": None, "sysctl": None, "udevadm": None,
+    "echo": None, "printf": None, "cd": None, "pwd": None, "export": None,
+    "test": None, "[": None, "sleep": None, "source": None, ".": None,
+    "timeout": None, "nohup": None, "unset": None, "shift": None, "return": None,
+    "local": None, "readonly": None, "trap": None, "type": None, "command": None,
+    "set": None, "shopt": None, "eval": None, "exec": None, "alias": None,
+    "unalias": None, "declare": None, "typeset": None, "read": None, "mapfile": None,
+}
+
+# etcdctl 由 classify_shell_command 中的专用块处理 (两段式命令)
+# 从通用白名单中移除，避免被单段式逻辑误判
+
+# etcdctl 单段式只读 verb
+_ETCDCTL_READ_VERBS = {"get", "watch", "version", "lock", "lease", "move-leader", "leader"}
+# etcdctl 单段式写 verb (需人工审批)
+_ETCDCTL_WRITE_VERBS = {"put", "del", "delete", "compact", "txn", "snapshot", "defrag"}
+# etcdctl 两段式 (verb, subverb) 只读组合
+_ETCDCTL_READ_SUBVERBS = {
+    "endpoint": {"status", "health", "hashkv"},
+    "member": {"list", "list-urls"},
+    "alarm": {"list"},
+    "auth": {"status"},
+    "user": {"list", "get"},
+    "role": {"list", "get"},
+    "check": {"perf", "datascale"},
+}
+# etcdctl 两段式 (verb, subverb) 写组合 (需人工审批)
+_ETCDCTL_WRITE_SUBVERBS = {
+    "member": {"add", "remove", "update"},
+    "alarm": {"disarm"},
+    "auth": {"enable", "disable"},
+    "user": {"add", "delete", "grant-role", "revoke-role", "passwd"},
+    "role": {"add", "delete", "grant-permission", "revoke-permission"},
+}
+# 系统修改命令白名单：需要人工审批
+SHELL_WRITE_COMMANDS: dict[str, set[str] | None] = {
+    "systemctl": {"restart", "start", "stop", "reload", "enable", "disable", "daemon-reload"},
+    # 文件操作命令：任意参数都需审批
+    "rm": None,
+    "cp": None,
+    "mv": None,
+    "chmod": None,
+    "chown": None,
+    "mkdir": None,
+    "touch": None,
+    "ln": None,
+}
+
+# ip 命令的写操作动词 (出现即拒绝自动执行)
+_IP_WRITE_ACTIONS = {"set", "add", "del", "delete", "change", "replace", "append", "prepend"}
+
+
+
+def _first_non_flag(args: list[str]) -> str | None:
+    for arg in args:
+        if not arg.startswith("-"):
+            return arg.lower()
+    return None
+
+
+def parse_shell_command(command: str) -> list[str]:
+    """解析单条 shell 诊断命令，禁止 shell 操作符和命令组合。"""
+    parts = _parse_command(command)
+    if not parts:
+        raise ValueError("命令不能为空")
+    return parts
+
+
+def classify_shell_command(command: str) -> CommandDecision:
+    """对白名单内的系统诊断命令进行分类。
+
+    systemctl 同时出现在只读和写白名单中：restart/stop 等走写表（需审批），
+    status/is-active 等走只读表（自动执行）。因此先检查写表，再检查只读表。
+
+    含管道 (|) 或重定向 (>) 的命令：只检查管道前的 binary 是否在白名单内，
+    并设置 needs_shell=True 以便用 shell 方式执行。
+    """
+    needs_shell = _has_shell_operators(command)
+
+    if needs_shell:
+        # 含 shell 操作符或为多行命令：先检查武器级危险命令
+        if _command_has_dangerous_cmds(command):
+            raise ValueError("命令包含不允许的危险命令（sh/bash/dd/mkfs 等），请手动执行")
+
+        # 多行或含操作符的命令：按行分割后，对每行逐段检查，取最高权限模式
+        lines = command.split("\n") if "\n" in command else [command]
+        overall_mode = "read"
+        overall_approval = False
+        first_binary = None
+        first_segment = ""
+
+        for line in lines:
+            line = line.strip()
+            if not line or line.startswith("#"):
+                continue
+            segments = _split_into_commands(line)
+            for segment in segments:
+                # 检测 kubectl 命令
+                if segment.startswith("kubectl ") or segment == "kubectl":
+                    try:
+                        kdec = classify_kubectl_command(segment)
+                    except (ValueError, IndexError) as exc:
+                        raise ValueError(str(exc))
+                    if first_binary is None:
+                        first_binary = "kubectl"
+                        first_segment = segment
+                    if kdec.mode == "write":
+                        overall_mode = "write"
+                        overall_approval = True
+                    continue
+
+                try:
+                    seg_parts = shlex.split(segment)
+                except ValueError:
+                    continue
+                if not seg_parts:
+                    continue
+                # 跳过 VAR=value 赋值前缀（含 export）
+                idx = 0
+                while idx < len(seg_parts):
+                    if "=" in seg_parts[idx] and not seg_parts[idx].startswith("-"):
+                        idx += 1
+                    elif seg_parts[idx] == "export":
+                        idx += 1
+                    else:
+                        break
+                if idx >= len(seg_parts):
+                    continue  # 纯赋值段，跳过
+                binary = seg_parts[idx]
+                check_parts = seg_parts[idx:]
+                if first_binary is None:
+                    first_binary = binary
+
+                if binary not in SHELL_READ_ONLY_COMMANDS and binary not in SHELL_WRITE_COMMANDS:
+                    raise ValueError(f"不允许执行该命令: {binary}")
+
+                if binary in SHELL_WRITE_COMMANDS:
+                    allowed = SHELL_WRITE_COMMANDS[binary]
+                    if allowed is None:
+                        overall_mode = "write"
+                        overall_approval = True
+                    else:
+                        verb = _first_non_flag(check_parts[1:])
+                        if verb is not None and verb in allowed:
+                            overall_mode = "write"
+                            overall_approval = True
+                        elif binary in SHELL_READ_ONLY_COMMANDS:
+                            ro_allowed = SHELL_READ_ONLY_COMMANDS[binary]
+                            if ro_allowed is None or (verb is not None and verb in ro_allowed):
+                                pass  # 该段是只读，不改变 overall
+                            else:
+                                raise ValueError(f"命令 {binary} 不支持该子命令: {verb}")
+                        else:
+                            raise ValueError(f"命令 {binary} 不支持该子命令: {verb}")
+
+        if first_binary is None:
+            raise ValueError("命令不能为空")
+
+        return CommandDecision(overall_mode, overall_approval, first_binary, [command], needs_shell=True)
+
+    parts = parse_shell_command(command)
+    binary = parts[0]
+    # 支持 VAR=value command 语法
+    start_idx = 0
+    while start_idx < len(parts) and "=" in parts[start_idx] and not parts[start_idx].startswith("-"):
+        start_idx += 1
+    if start_idx > 0:
+        if start_idx < len(parts):
+            binary = parts[start_idx]
+            effective_parts = parts[start_idx:]
+        else:
+            # 纯赋值行如 LOG_DIR=/tmp/log — 放行
+            return CommandDecision("read", False, parts[0], parts[1:])
+        verb = _first_non_flag(effective_parts[1:])
+    else:
+        effective_parts = parts
+        verb = _first_non_flag(parts[1:])
+
+    if binary in SHELL_WRITE_COMMANDS:
+        allowed = SHELL_WRITE_COMMANDS[binary]
+        if allowed is None or (verb is not None and verb in allowed):
+            return CommandDecision("write", True, binary, effective_parts[1:])
+
+    # etcdctl: 单段式或两段式命令，verb/subverb 均跳过全局 flag (--endpoints 等)
+    # member/alarm/auth 等同时有只读和写子命令，需按 (verb, subverb) 组合精确判断
+    if binary == "etcdctl":
+        non_flags = [a for a in effective_parts[1:] if not a.startswith("-")]
+        ev = non_flags[0].lower() if non_flags else None
+        esub = non_flags[1].lower() if len(non_flags) > 1 else None
+        # 单段式 verb
+        if ev in _ETCDCTL_READ_VERBS:
+            return CommandDecision("read", False, ev, effective_parts[1:])
+        if ev in _ETCDCTL_WRITE_VERBS:
+            return CommandDecision("write", True, ev, effective_parts[1:])
+        # 两段式: 先查写子命令 (精确匹配 subverb)，再查只读子命令
+        if ev in _ETCDCTL_WRITE_SUBVERBS and esub in _ETCDCTL_WRITE_SUBVERBS[ev]:
+            return CommandDecision("write", True, ev, effective_parts[1:])
+        if ev in _ETCDCTL_READ_SUBVERBS:
+            if esub is None or esub in _ETCDCTL_READ_SUBVERBS[ev]:
+                return CommandDecision("read", False, ev, effective_parts[1:])
+            raise ValueError(f"etcdctl {ev} 不支持子命令: {esub}")
+        raise ValueError(f"etcdctl 不支持该子命令: {ev}")
+
+    if binary in SHELL_READ_ONLY_COMMANDS:
+        allowed = SHELL_READ_ONLY_COMMANDS[binary]
+        if allowed is not None and (verb is None or verb not in allowed):
+            raise ValueError(f"命令 {binary} 不支持该子命令: {verb}")
+        if binary == "ip" and any(arg in _IP_WRITE_ACTIONS for arg in effective_parts[1:]):
+            raise ValueError("ip 命令包含写操作，请使用只读形式如 'ip addr show'")
+        return CommandDecision("read", False, binary, effective_parts[1:])
+
+    raise ValueError(f"不允许执行该命令: {binary}")
+
+
+def classify_agent_command(command: str) -> CommandDecision:
+    """统一分类 Agent 命令：kubectl 或白名单系统诊断命令。"""
+    stripped = (command or "").strip()
+    # 检测 VAR=value kubectl 语法（有变量赋值时）
+    rest = stripped
+    while True:
+        try:
+            first = rest.split(None, 1)[0]
+        except IndexError:
+            break
+        if "=" in first and not first.startswith("-"):
+            idx = rest.index("=")
+            # 必须是合法变量赋值 VAR=value
+            parts_split = rest.split(None, 1)
+            if len(parts_split) < 2:
+                break
+            rest = parts_split[1]
+        elif first == "export":
+            parts_split = rest.split(None, 1)
+            if len(parts_split) < 2:
+                break
+            rest = parts_split[1]
+        else:
+            break
+    if rest.startswith("kubectl ") or rest == "kubectl":
+        return classify_kubectl_command(rest)
+    if stripped.startswith("kubectl ") or stripped == "kubectl":
+        return classify_kubectl_command(stripped)
+    return classify_shell_command(stripped)
+
+
+# ── 命令执行 ──────────────────────────────────────────────
+
+async def execute_command(command: str, timeout: int = 30) -> dict:
+    """执行 Agent 命令 (kubectl 或白名单系统诊断命令)。
+
+    含管道/重定向的命令通过 shell=True 执行，其他用 subprocess_exec 直接执行。
+    """
+    decision = classify_agent_command(command)
+    parts = _parse_command(command)
+    # 支持 VAR=value command 语法，跳过赋值找实际 binary
+    cmd_idx = 0
+    while cmd_idx < len(parts) and "=" in parts[cmd_idx] and not parts[cmd_idx].startswith("-"):
+        cmd_idx += 1
+    if cmd_idx > 0 and cmd_idx < len(parts):
+        parts = parts[cmd_idx:]
+    if parts[0] == "kubectl":
+        executable = KUBECTL
+        args = decision.args
+    else:
+        executable = parts[0]
+        args = parts[1:]
+
+    if decision.needs_shell:
+        proc = await asyncio.create_subprocess_shell(
+            command,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+        )
+    else:
+        proc = await asyncio.create_subprocess_exec(
+            executable,
+            *args,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+        )
+    try:
+        stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout)
+        return {
+            "command": command,
+            "mode": decision.mode,
+            "returncode": proc.returncode,
+            "stdout": stdout.decode(errors="replace")[:20000].strip(),
+            "stderr": stderr.decode(errors="replace")[:10000].strip(),
+        }
+    except asyncio.TimeoutError:
+        proc.kill()
+        await proc.communicate()
+        return {
+            "command": command,
+            "mode": decision.mode,
+            "returncode": -1,
+            "stdout": "",
+            "stderr": f"命令执行超时 ({timeout}s)",
+        }
+
+
+async def execute_kubectl(command: str, timeout: int = 30) -> dict:
+    """执行单条 kubectl 命令 (兼容旧接口)。"""
+    decision = classify_kubectl_command(command)
+    proc = await asyncio.create_subprocess_exec(
+        KUBECTL,
+        *decision.args,
+        stdout=asyncio.subprocess.PIPE,
+        stderr=asyncio.subprocess.PIPE,
+    )
+    try:
+        stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout)
+        return {
+            "command": command,
+            "mode": decision.mode,
+            "returncode": proc.returncode,
+            "stdout": stdout.decode(errors="replace")[:20000].strip(),
+            "stderr": stderr.decode(errors="replace")[:10000].strip(),
+        }
+    except asyncio.TimeoutError:
+        proc.kill()
+        await proc.communicate()
+        return {
+            "command": command,
+            "mode": decision.mode,
+            "returncode": -1,
+            "stdout": "",
+            "stderr": f"命令执行超时 ({timeout}s)",
+        }

backend/ai_agent.py

@@ -1,7 +1,12 @@
 """AI Agent 分析模块 - 调用 OpenAI 兼容接口分析诊断结果 (支持流式输出)"""
 import json
+import re
+import uuid
+from collections import Counter
+from typing import Optional
 import httpx
 from backend.config import AI_API_BASE, AI_API_KEY, AI_MODEL
+from backend.agent_tools import classify_agent_command, execute_command
 
 SYSTEM_PROMPT = """你是一位资深的 Kubernetes 运维专家和 SRE 工程师。你的任务是分析 K8S 集群诊断报告，找出根本原因并给出修复建议。
 
@@ -20,6 +25,392 @@ _HEADERS = {
 }
 
 
+AGENT_SYSTEM_PROMPT = """你是 Kubernetes 自主诊断 Agent。你的首要职责是主动调用命令获取现场证据，而不是只给用户一段脚本或操作建议。
+
+可用工具：
+1. kubectl 命令 - 查询集群内资源状态。例如: kubectl get pods -A, kubectl describe node <name>, kubectl logs <pod> -n <ns>
+2. 系统诊断 shell 命令 - 查询节点宿主机状态。例如: systemctl status kubelet, journalctl -u kubelet -n 100 --no-pager, crictl ps, df -h, free -m, ss -tlnp, ip addr, top, du -sh, curl, tcpdump
+3. containerd 命令 - 查询容器运行时状态。例如: ctr containers list, ctr images list, ctr tasks ls, nerdctl ps -a, nerdctl logs <container>, nerdctl stats
+4. etcd 诊断命令 - 检查 etcd 集群健康与数据。例如: etcdctl endpoint status, etcdctl endpoint health, etcdctl member list, etcdctl alarm list, etcdctl get /kubernetes --prefix
+5. Linux 基础调优命令 - 例如: iostat -xz 1 3, vmstat 1 3, mpstat -P ALL 1 3, pidstat -d 1 3, lsof -i :<port>, ethtool <iface>, tcpdump -i any -c 10 -nn
+
+行为要求：
+1. 如果证据不足，必须返回 command 动作并逐条调查；至少检查与故障资源直接相关的状态、describe、events 或 logs。
+2. 禁止在 final 中仅提供"请执行以下命令"的脚本。只有在已经依据真实命令输出形成结论时才能返回 final。
+3. 每次只提出一条命令，服务端会执行并把真实输出返回给你，然后你继续下一轮判断。
+
+安全规则：
+1. 可以使用管道 (|)、重定向 (>/>>)、逻辑连接符 (&&/||) 或分号 (;) 组合多条命令进行精确排查。不得使用反引号 (`` ` ``) 执行命令替换。
+2. kubectl 的 get、describe、logs、top 等只读命令可以自动执行。
+3. 系统诊断只读命令 (systemctl status、journalctl、crictl ps、df、free、ss、ip addr 等) 可以自动执行。
+4. apply、delete、patch、scale、rollout、exec 等修改或交互命令必须暂停并请求人工批准。
+5. systemctl restart/stop/start 等系统修改命令必须暂停并请求人工批准。
+6. 不要猜测命令输出；必须依据实际工具结果继续判断。
+7. 只要问题没有完全解决（verified=true 且 remaining_issues=0），就要持续执行命令诊断，没有轮数上限。
+
+每轮必须只返回一个 JSON 对象，不要使用 Markdown 代码块：
+- 继续调查：{"action":"command","command":"kubectl ...","reason":"为什么执行"}
+- 完成分析：{"action":"final","analysis":"中文 Markdown 分析结论","verified":true,"remaining_issues":0}
+
+只有完成修复后的全量复检，并确认 remaining_issues 为 0 时才能返回 final。只定位根因、执行命令成功或部分资源恢复都不能结束。
+"""
+
+_PENDING_APPROVALS: dict[str, dict] = {}
+
+
+def _parse_agent_action(content: str) -> dict:
+    text = content.strip()
+    candidates = [text]
+    fenced = re.search(r"```(?:json)?\s*(\{.*?\})\s*```", text, re.DOTALL | re.IGNORECASE)
+    if fenced:
+        candidates.insert(0, fenced.group(1))
+    first_brace = text.find("{")
+    last_brace = text.rfind("}")
+    if first_brace >= 0 and last_brace > first_brace:
+        candidates.append(text[first_brace:last_brace + 1])
+
+    action = None
+    for candidate in candidates:
+        try:
+            action = json.loads(candidate)
+            break
+        except json.JSONDecodeError:
+            continue
+    if action is None:
+        raise ValueError("AI 未返回合法 JSON 动作")
+    if not isinstance(action, dict) or action.get("action") not in {"command", "final"}:
+        raise ValueError("AI 返回了不支持的动作")
+    if action["action"] == "command" and not action.get("command"):
+        raise ValueError("AI 命令动作缺少 command")
+    if action["action"] == "final" and not action.get("analysis"):
+        raise ValueError("AI 最终动作缺少 analysis")
+    if action["action"] == "final":
+        if action.get("verified") is not True or action.get("remaining_issues") != 0:
+            raise ValueError("AI 最终动作必须确认 verified=true 且 remaining_issues=0")
+    return action
+
+
+async def _complete_chat(messages: list[dict]) -> str:
+    timeout = httpx.Timeout(90, connect=10)
+    async with httpx.AsyncClient(timeout=timeout) as client:
+        response = await client.post(
+            f"{AI_API_BASE}/chat/completions",
+            headers=_HEADERS,
+            json={
+                "model": AI_MODEL,
+                "messages": messages,
+                "temperature": 0.1,
+                "max_tokens": 4096,
+                "stream": False,
+            },
+        )
+        response.raise_for_status()
+        payload = response.json()
+        return payload["choices"][0]["message"]["content"]
+
+
+async def run_diagnostic_agent(
+    report: str,
+    question: str = "",
+    max_steps: int = 30,
+    execution_mode: str = "ai",
+    run_id: str = "",
+    history_store=None,
+):
+    """自主诊断循环：AI 模式执行只读命令，手动模式等待用户执行。"""
+    if execution_mode not in {"ai", "manual"}:
+        yield {"type": "error", "message": "不支持的命令执行模式，仅支持 ai 或 manual"}
+        return
+
+    user_content = f"当前 K8S 诊断报告：\n\n{report}"
+    if question:
+        user_content += f"\n\n用户关注点：{question}"
+    messages = [
+        {"role": "system", "content": AGENT_SYSTEM_PROMPT},
+        {"role": "user", "content": user_content},
+    ]
+
+    command_counts = Counter()
+
+    if execution_mode == "ai":
+        initial_command = "kubectl get nodes -o wide"
+        command_counts[initial_command] += 1
+        yield {
+            "type": "command", "step": 0, "command": initial_command,
+            "reason": "建立集群节点状态基线，确保 AI 执行模式实际采集现场证据",
+            "mode": "read",
+        }
+        initial_result = await execute_command(initial_command)
+        yield {"type": "result", **initial_result}
+        messages.append({
+            "role": "user",
+            "content": "AI 执行模式已自动采集第一份现场证据：\n" + json.dumps(initial_result, ensure_ascii=False),
+        })
+
+    async for event in _run_agent_loop(
+        messages=messages,
+        execution_mode=execution_mode,
+        max_steps=max_steps,
+        start_step=1,
+        command_counts=command_counts,
+        report=report,
+        question=question,
+        run_id=run_id,
+        history_store=history_store,
+    ):
+        yield event
+
+
+async def _run_agent_loop(
+    messages: list[dict],
+    execution_mode: str,
+    max_steps: int,
+    start_step: int,
+    command_counts: Counter,
+    report: str,
+    question: str,
+    run_id: str = "",
+    history_store=None,
+):
+    invalid_response_count = 0
+    step = start_step
+    while True:
+        if max_steps > 0 and step > max_steps:
+            break
+        yield {"type": "thinking", "step": step, "message": f"AI 正在规划第 {step} 轮诊断"}
+        try:
+            content = await _complete_chat(messages)
+            action = _parse_agent_action(content)
+        except ValueError as exc:
+            if "最终动作必须确认" in str(exc):
+                invalid_response_count = 0
+                messages.extend([
+                    {"role": "assistant", "content": content},
+                    {
+                        "role": "user",
+                        "content": (
+                            "不能结束诊断：尚未通过修复后的全量复检确认所有问题清零。"
+                            "请继续返回 command，检查节点、Pod、Deployment、事件以及原故障资源；"
+                            "只有 verified=true 且 remaining_issues=0 时才可返回 final。"
+                        ),
+                    },
+                ])
+                yield {"type": "verification_required", "message": "尚未确认所有问题清零，Agent 继续执行诊断验证"}
+                continue
+            invalid_response_count += 1
+            if invalid_response_count >= 3:
+                yield {
+                    "type": "error",
+                    "message": "AI 连续 3 次未返回合法 JSON 动作，已停止本次诊断以避免无效循环",
+                }
+                return
+            messages.extend([
+                {"role": "assistant", "content": content},
+                {
+                    "role": "user",
+                    "content": (
+                        f"上一次响应无法解析：{exc}。请不要输出解释、脚本或多条命令，"
+                        "严格只返回一个 JSON 对象："
+                        '{"action":"command","command":"kubectl ...","reason":"..."} '
+                        "或在全量复检问题清零后返回 final。"
+                    ),
+                },
+            ])
+            yield {
+                "type": "response_rejected",
+                "message": f"AI 响应格式无效，正在自动纠正并重试（{invalid_response_count}/3）",
+                "reason": str(exc),
+            }
+            continue
+        except Exception as exc:
+            yield {"type": "error", "message": str(exc)}
+            return
+
+        if action["action"] == "final":
+            invalid_response_count = 0
+            yield {
+                "type": "final", "content": action["analysis"], "model": AI_MODEL,
+                "verified": True, "remaining_issues": 0,
+            }
+            return
+
+        command = action["command"]
+        invalid_response_count = 0
+        reason = action.get("reason", "AI 需要更多集群证据")
+        command_counts[command] += 1
+        if command_counts[command] > 2:
+            messages.extend([
+                {"role": "assistant", "content": content},
+                {
+                    "role": "user",
+                    "content": (
+                        f"命令 {command!r} 已重复执行两次，不能再次执行。"
+                        "请基于已有输出选择不同的只读诊断命令，或在证据充分时返回 final。"
+                    ),
+                },
+            ])
+            yield {
+                "type": "command_rejected", "command": command,
+                "reason": "相同命令已执行两次，请更换诊断方向",
+            }
+            continue
+        try:
+            decision = classify_agent_command(command)
+        except ValueError as exc:
+            messages.extend([
+                {"role": "assistant", "content": content},
+                {"role": "user", "content": f"命令被安全策略拒绝：{exc}。请改用单条合法 kubectl 或系统诊断命令。"},
+            ])
+            yield {"type": "command_rejected", "command": command, "reason": str(exc)}
+            continue
+
+        yield {
+            "type": "command", "step": step, "command": command,
+            "reason": reason, "mode": decision.mode,
+        }
+        if execution_mode == "manual":
+            yield {
+                "type": "manual_command", "command_id": uuid.uuid4().hex,
+                "command": command, "reason": reason, "mode": decision.mode,
+                "message": "请在终端手动执行命令，并根据结果继续排查",
+            }
+            # 手动模式也保存上下文，支持断点续传
+            self_save_context(run_id, history_store, messages, command_counts, step, step + 1,
+                              max_steps, report, question, execution_mode, content)
+            return
+
+        if decision.requires_approval:
+            approval_id = uuid.uuid4().hex
+            _PENDING_APPROVALS[approval_id] = {
+                "command": command,
+                "reason": reason,
+                "created_by": AI_MODEL,
+                "run_id": run_id,
+                "report": report,
+                "question": question,
+                "execution_mode": execution_mode,
+                "history_store": history_store,
+                "messages": messages + [{"role": "assistant", "content": content}],
+                "next_step": step + 1,
+                "max_steps": max_steps,
+                "command_counts": dict(command_counts),
+            }
+            # 保存上下文到 DB，避免浏览器关闭/刷新丢失进度
+            self_save_context(run_id, history_store, messages, command_counts, step, step + 1,
+                              max_steps, report, question, execution_mode, content)
+            yield {
+                "type": "approval_required", "approval_id": approval_id,
+                "command": command, "reason": reason,
+            }
+            return
+
+        result = await execute_command(command)
+        yield {"type": "result", **result}
+        messages.extend([
+            {"role": "assistant", "content": content},
+            {"role": "user", "content": "命令实际执行结果：\n" + json.dumps(result, ensure_ascii=False)},
+        ])
+
+        # 每次执行后保存上下文，支持断点续传
+        self_save_context(run_id, history_store, messages, command_counts, step, step + 1,
+                          max_steps, report, question, execution_mode)
+
+        step += 1
+
+    if max_steps > 0:
+        yield {
+            "type": "error",
+            "message": (
+                f"自主诊断已达到安全上限（{max_steps} 轮），但 Agent 仍未形成有效结论。"
+                "系统已停止继续执行以避免无限循环，请检查是否重复执行相同命令或 AI 服务响应异常。"
+            ),
+        }
+    # max_steps=0 表示不限轮数，不输出错误 — Agent 会持续执行直到完成或遇到致命错误
+
+
+async def resume_diagnostic_agent(pending: dict, result: dict):
+    """审批命令执行后，把真实结果交回原 Agent 上下文并继续诊断。"""
+    messages = list(pending.get("messages") or [])
+    messages.append({
+        "role": "user",
+        "content": (
+            "用户已批准命令，以下是实际执行结果。请继续自主诊断并验证问题是否解决：\n"
+            + json.dumps(result, ensure_ascii=False)
+        ),
+    })
+    run_id = pending.get("run_id", "")
+    history_store = pending.get("history_store")
+    async for event in _run_agent_loop(
+        messages=messages,
+        execution_mode=pending.get("execution_mode", "ai"),
+        max_steps=int(pending.get("max_steps", 30)),
+        start_step=int(pending.get("next_step", 1)),
+        command_counts=Counter(pending.get("command_counts") or {}),
+        report=pending.get("report", ""),
+        question=pending.get("question", ""),
+        run_id=run_id,
+        history_store=history_store,
+    ):
+        yield event
+
+
+def take_pending_approval(approval_id: str) -> dict | None:
+    return _PENDING_APPROVALS.pop(approval_id, None)
+
+
+def self_save_context(
+    run_id: str, history_store,
+    messages: list, command_counts: Counter,
+    step: int, next_step: int,
+    max_steps: int, report: str, question: str,
+    execution_mode: str,
+    last_response_content: str = "",
+):
+    """保存 Agent 执行上下文到 DB，支持断点续传。"""
+    if not run_id or not history_store:
+        return
+    context = {
+        "messages": messages,
+        "command_counts": dict(command_counts),
+        "step": step,
+        "next_step": next_step,
+        "max_steps": max_steps,
+        "report": report,
+        "question": question,
+        "execution_mode": execution_mode,
+        "last_response": last_response_content,
+    }
+    history_store.update_context(run_id, context)
+
+
+async def resume_from_context(
+    ctx: dict,
+    run_id: str = "",
+    history_store=None,
+):
+    """从 DB 保存的上下文恢复 Agent 诊断，不重跑历史命令，直接进入下一轮。"""
+    messages = list(ctx.get("messages") or [])
+    command_counts = Counter(ctx.get("command_counts") or {})
+    next_step = int(ctx.get("next_step", 1))
+    max_steps = int(ctx.get("max_steps", 0))
+    report = ctx.get("report", "")
+    question = ctx.get("question", "")
+    execution_mode = ctx.get("execution_mode", "ai")
+
+    async for event in _run_agent_loop(
+        messages=messages,
+        execution_mode=execution_mode,
+        max_steps=max_steps,
+        start_step=next_step,
+        command_counts=command_counts,
+        report=report,
+        question=question,
+        run_id=run_id,
+        history_store=history_store,
+    ):
+        yield event
+
+
 async def _stream_chat(messages: list[dict]):
     """通用流式请求 - yield 文本 chunk"""
     # connect=10s 快速失败; read=90s 防止 AI 服务无响应时前端无限卡死

backend/config.py

@@ -9,3 +9,4 @@ AI_API_KEY = os.getenv("AI_API_KEY", "hermes")
 AI_MODEL = os.getenv("AI_MODEL", "qwen3.8-max-preview")
 PORT = int(os.getenv("PORT", "8900"))
 KUBECONFIG_PATH = os.getenv("KUBECONFIG_PATH", "") or None
+HISTORY_DB_PATH = os.getenv("HISTORY_DB_PATH", os.path.join(os.path.dirname(os.path.dirname(__file__)), "data", "history.db"))

backend/history_store.py

@@ -0,0 +1,211 @@
+"""SQLite 持久化诊断历史、AI 分析和 Agent 处理记录。"""
+import json
+import os
+import sqlite3
+import uuid
+from datetime import datetime
+from typing import Optional
+
+
+def _now() -> str:
+    return datetime.now().isoformat(timespec="seconds")
+
+
+class HistoryStore:
+    def __init__(self, path: str):
+        self.path = path
+        parent = os.path.dirname(os.path.abspath(path))
+        os.makedirs(parent, exist_ok=True)
+        self._init_db()
+
+    def _connect(self):
+        connection = sqlite3.connect(self.path, timeout=30)
+        connection.row_factory = sqlite3.Row
+        connection.execute("PRAGMA journal_mode=WAL")
+        connection.execute("PRAGMA foreign_keys=ON")
+        return connection
+
+    def _init_db(self):
+        with self._connect() as db:
+            db.executescript("""
+                CREATE TABLE IF NOT EXISTS diagnoses (
+                    id TEXT PRIMARY KEY,
+                    created_at TEXT NOT NULL,
+                    namespace TEXT NOT NULL,
+                    score INTEGER NOT NULL,
+                    status TEXT NOT NULL,
+                    critical_count INTEGER NOT NULL DEFAULT 0,
+                    warning_count INTEGER NOT NULL DEFAULT 0,
+                    elapsed_seconds REAL NOT NULL DEFAULT 0,
+                    diagnosis_json TEXT NOT NULL,
+                    report TEXT NOT NULL
+                );
+                CREATE INDEX IF NOT EXISTS idx_diagnoses_created_at ON diagnoses(created_at DESC);
+                CREATE INDEX IF NOT EXISTS idx_diagnoses_status ON diagnoses(status);
+
+                CREATE TABLE IF NOT EXISTS processing_runs (
+                    id TEXT PRIMARY KEY,
+                    diagnosis_id TEXT NOT NULL,
+                    kind TEXT NOT NULL,
+                    mode TEXT NOT NULL DEFAULT '',
+                    question TEXT NOT NULL DEFAULT '',
+                    status TEXT NOT NULL,
+                    summary TEXT NOT NULL DEFAULT '',
+                    started_at TEXT NOT NULL,
+                    finished_at TEXT,
+                    context_json TEXT NOT NULL DEFAULT '',
+                    FOREIGN KEY(diagnosis_id) REFERENCES diagnoses(id) ON DELETE CASCADE
+                );
+                CREATE INDEX IF NOT EXISTS idx_runs_diagnosis ON processing_runs(diagnosis_id, started_at);
+
+                CREATE TABLE IF NOT EXISTS processing_events (
+                    id INTEGER PRIMARY KEY AUTOINCREMENT,
+                    run_id TEXT NOT NULL,
+                    sequence INTEGER NOT NULL,
+                    type TEXT NOT NULL,
+                    created_at TEXT NOT NULL,
+                    payload_json TEXT NOT NULL,
+                    FOREIGN KEY(run_id) REFERENCES processing_runs(id) ON DELETE CASCADE,
+                    UNIQUE(run_id, sequence)
+                );
+            """)
+            # 兼容旧表：确保 context_json 列存在
+            self._ensure_column(db, "processing_runs", "context_json", "TEXT NOT NULL DEFAULT ''")
+
+    def _ensure_column(self, db, table: str, column: str, col_def: str):
+        try:
+            db.execute(f"ALTER TABLE {table} ADD COLUMN {column} {col_def}")
+        except sqlite3.OperationalError as e:
+            if "duplicate column" not in str(e).lower():
+                raise
+
+    def save_diagnosis(self, diagnosis: dict, report: str) -> str:
+        record_id = uuid.uuid4().hex
+        created_at = diagnosis.get("timestamp") or _now()
+        with self._connect() as db:
+            db.execute(
+                """INSERT INTO diagnoses
+                   (id, created_at, namespace, score, status, critical_count, warning_count,
+                    elapsed_seconds, diagnosis_json, report)
+                   VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
+                (
+                    record_id, created_at, diagnosis.get("namespace", "all"),
+                    int(diagnosis.get("score", 0)), diagnosis.get("status", "unknown"),
+                    int(diagnosis.get("critical_count", 0)), int(diagnosis.get("warning_count", 0)),
+                    float(diagnosis.get("elapsed_seconds", 0)),
+                    json.dumps(diagnosis, ensure_ascii=False), report,
+                ),
+            )
+        return record_id
+
+    def list_diagnoses(self, status: str = "", namespace: str = "", limit: int = 100, offset: int = 0) -> list[dict]:
+        conditions = []
+        params: list = []
+        if status:
+            conditions.append("status = ?")
+            params.append(status)
+        if namespace:
+            conditions.append("namespace = ?")
+            params.append(namespace)
+        where = " WHERE " + " AND ".join(conditions) if conditions else ""
+        params.extend([max(1, min(limit, 500)), max(0, offset)])
+        with self._connect() as db:
+            rows = db.execute(
+                f"""SELECT id, created_at, namespace, score, status, critical_count,
+                           warning_count, elapsed_seconds,
+                           (SELECT COUNT(*) FROM processing_runs r WHERE r.diagnosis_id = diagnoses.id) AS run_count
+                    FROM diagnoses{where}
+                    ORDER BY created_at DESC, rowid DESC LIMIT ? OFFSET ?""",
+                params,
+            ).fetchall()
+        return [dict(row) for row in rows]
+
+    def get_diagnosis(self, record_id: str) -> Optional[dict]:
+        with self._connect() as db:
+            row = db.execute("SELECT * FROM diagnoses WHERE id = ?", (record_id,)).fetchone()
+            if row is None:
+                return None
+            run_rows = db.execute(
+                "SELECT * FROM processing_runs WHERE diagnosis_id = ? ORDER BY started_at, rowid",
+                (record_id,),
+            ).fetchall()
+            runs = []
+            for run_row in run_rows:
+                run = dict(run_row)
+                event_rows = db.execute(
+                    "SELECT type, created_at, payload_json FROM processing_events WHERE run_id = ? ORDER BY sequence",
+                    (run["id"],),
+                ).fetchall()
+                run["events"] = [
+                    {"type": event["type"], "created_at": event["created_at"], **json.loads(event["payload_json"])}
+                    for event in event_rows
+                ]
+                run["context_json"] = run.get("context_json") or ""
+                if run["context_json"]:
+                    run["context"] = json.loads(run["context_json"])
+                else:
+                    run["context"] = None
+                runs.append(run)
+        return {
+            "id": row["id"], "created_at": row["created_at"],
+            "diagnosis": json.loads(row["diagnosis_json"]),
+            "report": row["report"], "runs": runs,
+        }
+
+    def latest(self) -> Optional[dict]:
+        records = self.list_diagnoses(limit=1)
+        return self.get_diagnosis(records[0]["id"]) if records else None
+
+    def create_run(self, diagnosis_id: str, kind: str, mode: str = "", question: str = "") -> str:
+        run_id = uuid.uuid4().hex
+        with self._connect() as db:
+            db.execute(
+                """INSERT INTO processing_runs
+                   (id, diagnosis_id, kind, mode, question, status, started_at)
+                   VALUES (?, ?, ?, ?, ?, 'running', ?)""",
+                (run_id, diagnosis_id, kind, mode, question, _now()),
+            )
+        return run_id
+
+    def update_context(self, run_id: str, context: dict):
+        """保存/更新 Agent 执行上下文到 processing_runs，支持断点续传。"""
+        with self._connect() as db:
+            db.execute(
+                "UPDATE processing_runs SET context_json = ? WHERE id = ?",
+                (json.dumps(context, ensure_ascii=False), run_id),
+            )
+
+    def get_context(self, run_id: str) -> Optional[dict]:
+        """读取 Agent 执行上下文，如果不存在或为空则返回 None。"""
+        with self._connect() as db:
+            row = db.execute(
+                "SELECT context_json, diagnosis_id, report FROM processing_runs pr JOIN diagnoses d ON pr.diagnosis_id = d.id WHERE pr.id = ?",
+                (run_id,),
+            ).fetchone()
+            if row is None:
+                return None
+            ctx_str = row["context_json"] or ""
+            if not ctx_str:
+                return None
+            ctx = json.loads(ctx_str)
+            ctx["report"] = row["report"]
+            return ctx
+
+    def append_event(self, run_id: str, event_type: str, payload: dict):
+        with self._connect() as db:
+            sequence = db.execute(
+                "SELECT COALESCE(MAX(sequence), 0) + 1 FROM processing_events WHERE run_id = ?",
+                (run_id,),
+            ).fetchone()[0]
+            db.execute(
+                """INSERT INTO processing_events (run_id, sequence, type, created_at, payload_json)
+                   VALUES (?, ?, ?, ?, ?)""",
+                (run_id, sequence, event_type, _now(), json.dumps(payload, ensure_ascii=False)),
+            )
+
+    def finish_run(self, run_id: str, status: str, summary: str = ""):
+        with self._connect() as db:
+            db.execute(
+                "UPDATE processing_runs SET status = ?, summary = ?, finished_at = ? WHERE id = ?",
+                (status, summary, _now(), run_id),
+            )

backend/main.py

@@ -11,7 +11,7 @@ from fastapi.responses import StreamingResponse
 from pydantic import BaseModel
 from typing import Optional
 
-from backend.config import PORT, AI_API_BASE, AI_MODEL
+from backend.config import PORT, AI_API_BASE, AI_MODEL, HISTORY_DB_PATH
 from backend.diagnosis import (
     run_full_diagnosis, build_diagnosis_report, ALL_CHECKS,
     check_connectivity,
@@ -22,9 +22,14 @@ from backend.diagnosis import (
 from backend.ai_agent import (
     analyze_with_ai, chat_with_ai,
     analyze_with_ai_stream, chat_with_ai_stream,
+    run_diagnostic_agent, resume_diagnostic_agent, resume_from_context,
+    take_pending_approval,
 )
+from backend.agent_tools import execute_command
+from backend.history_store import HistoryStore
 
 app = FastAPI(title="K8S 智能诊断平台", version="1.0.0")
+history_store = HistoryStore(HISTORY_DB_PATH)
 
 app.add_middleware(
     CORSMiddleware,
@@ -36,6 +41,25 @@ app.add_middleware(
 # 存储最近诊断结果 (内存)
 _last_diagnosis: Optional[dict] = None
 _last_report: str = ""
+_last_diagnosis_id: Optional[str] = None
+
+
+def _save_current_diagnosis(diagnosis: dict, report: str) -> str:
+    global _last_diagnosis, _last_report, _last_diagnosis_id
+    _last_diagnosis = diagnosis
+    _last_report = report
+    _last_diagnosis_id = history_store.save_diagnosis(diagnosis, report)
+    return _last_diagnosis_id
+
+
+def _ensure_current_history() -> str:
+    global _last_diagnosis_id
+    if _last_diagnosis_id:
+        return _last_diagnosis_id
+    if _last_diagnosis is None:
+        raise RuntimeError("暂无诊断记录")
+    _last_diagnosis_id = history_store.save_diagnosis(_last_diagnosis, _last_report)
+    return _last_diagnosis_id
 
 # 诊断项执行顺序 (带中文标题)
 CHECK_ORDER = [
@@ -64,6 +88,11 @@ class ChatRequest(BaseModel):
     messages: list[dict]
 
 
+class AgentApprovalRequest(BaseModel):
+    approval_id: str
+    approved: bool
+
+
 @app.get("/api/health")
 async def health():
     return {"status": "ok", "time": datetime.now().isoformat()}
@@ -114,9 +143,9 @@ async def diagnose_stream(namespace: str = ""):
                 }],
                 "checks": {},
             }
-            _last_diagnosis = diagnosis
-            _last_report = build_diagnosis_report(diagnosis)
-            yield f"data: {json.dumps({'type': 'complete', 'diagnosis': diagnosis}, ensure_ascii=False)}\n\n"
+            report = build_diagnosis_report(diagnosis)
+            _save_current_diagnosis(diagnosis, report)
+            yield f"data: {json.dumps({'type': 'complete', 'diagnosis': diagnosis, 'history_id': _last_diagnosis_id}, ensure_ascii=False)}\n\n"
             return
 
         yield f"data: {json.dumps({'type': 'progress', 'step': 0, 'total': len(CHECK_ORDER) + 1, 'name': 'connectivity', 'title': '集群连通性检查', 'status': 'done', 'issues': 0}, ensure_ascii=False)}\n\n"
@@ -169,10 +198,10 @@ async def diagnose_stream(namespace: str = ""):
             "checks": results,
         }
 
-        _last_diagnosis = diagnosis
-        _last_report = build_diagnosis_report(diagnosis)
+        report = build_diagnosis_report(diagnosis)
+        _save_current_diagnosis(diagnosis, report)
 
-        yield f"data: {json.dumps({'type': 'complete', 'diagnosis': diagnosis}, ensure_ascii=False)}\n\n"
+        yield f"data: {json.dumps({'type': 'complete', 'diagnosis': diagnosis, 'history_id': _last_diagnosis_id}, ensure_ascii=False)}\n\n"
 
     return StreamingResponse(
         event_generator(),
@@ -185,17 +214,24 @@ async def diagnose_stream(namespace: str = ""):
 async def diagnose(req: DiagnoseRequest):
     """执行一键诊断 (非流式，兼容旧接口)"""
     global _last_diagnosis, _last_report
-    _last_diagnosis = await run_full_diagnosis(namespace=req.namespace, checks=req.checks)
-    _last_report = build_diagnosis_report(_last_diagnosis)
-    return _last_diagnosis
+    diagnosis = await run_full_diagnosis(namespace=req.namespace, checks=req.checks)
+    report = build_diagnosis_report(diagnosis)
+    _save_current_diagnosis(diagnosis, report)
+    return {**diagnosis, "history_id": _last_diagnosis_id}
 
 
 @app.get("/api/diagnosis/latest")
 async def get_latest():
-    """获取最近一次诊断结果"""
+    """获取最近一次诊断结果，服务重启后从 SQLite 恢复。"""
+    global _last_diagnosis, _last_report, _last_diagnosis_id
     if _last_diagnosis is None:
+        latest = history_store.latest()
+        if latest is None:
             raise HTTPException(status_code=404, detail="暂无诊断记录，请先执行诊断")
-    return _last_diagnosis
+        _last_diagnosis = latest["diagnosis"]
+        _last_report = latest["report"]
+        _last_diagnosis_id = latest["id"]
+    return {**_last_diagnosis, "history_id": _last_diagnosis_id}
 
 
 @app.post("/api/analyze")
@@ -203,19 +239,25 @@ async def analyze(req: AnalyzeRequest):
     """AI 分析诊断结果 (非流式，兼容旧接口)"""
     global _last_diagnosis, _last_report
     if _last_diagnosis is None:
-        _last_diagnosis = await run_full_diagnosis()
-        _last_report = build_diagnosis_report(_last_diagnosis)
+        diagnosis = await run_full_diagnosis()
+        report = build_diagnosis_report(diagnosis)
+        _save_current_diagnosis(diagnosis, report)
+    run_id = history_store.create_run(_ensure_current_history(), "analysis", "standard", req.question)
     result = await analyze_with_ai(_last_report, question=req.question)
-    return result
+    history_store.append_event(run_id, "analysis", result)
+    history_store.finish_run(run_id, "completed" if result.get("success") else "failed", result.get("analysis", ""))
+    return {**result, "run_id": run_id}
 
 
 @app.get("/api/analyze/stream")
 async def analyze_stream(question: str = ""):
-    """SSE 流式 AI 分析 - 逐字推送分析内容"""
+    """SSE 流式 AI 分析 - 逐字推送分析内容并持久化"""
     global _last_diagnosis, _last_report
     if _last_diagnosis is None:
-        _last_diagnosis = await run_full_diagnosis()
-        _last_report = build_diagnosis_report(_last_diagnosis)
+        diagnosis = await run_full_diagnosis()
+        report = build_diagnosis_report(diagnosis)
+        _save_current_diagnosis(diagnosis, report)
+    run_id = history_store.create_run(_ensure_current_history(), "analysis", "stream", question)
 
     async def event_generator():
         process_steps = [
@@ -225,23 +267,190 @@ async def analyze_stream(question: str = ""):
             ("summary", "整理分析结论", "正在组织根因、影响和预防建议"),
         ]
 
-        yield f"data: {json.dumps({'type': 'start', 'model': AI_MODEL}, ensure_ascii=False)}\n\n"
+        yield f"data: {json.dumps({'type': 'start', 'model': AI_MODEL, 'run_id': run_id}, ensure_ascii=False)}\n\n"
+        content = ""
         try:
             for step, title, detail in process_steps:
-                yield f"data: {json.dumps({'type': 'process', 'step': step, 'title': title, 'detail': detail, 'status': 'running'}, ensure_ascii=False)}\n\n"
+                process_event = {'step': step, 'title': title, 'detail': detail, 'status': 'running'}
+                history_store.append_event(run_id, "process", process_event)
+                yield f"data: {json.dumps({'type': 'process', **process_event}, ensure_ascii=False)}\n\n"
                 await asyncio.sleep(0)
-                yield f"data: {json.dumps({'type': 'process', 'step': step, 'title': title, 'detail': detail, 'status': 'done'}, ensure_ascii=False)}\n\n"
+                process_event = {'step': step, 'title': title, 'detail': detail, 'status': 'done'}
+                history_store.append_event(run_id, "process", process_event)
+                yield f"data: {json.dumps({'type': 'process', **process_event}, ensure_ascii=False)}\n\n"
 
             async for chunk in analyze_with_ai_stream(_last_report, question=question):
+                content += chunk
                 yield f"data: {json.dumps({'type': 'chunk', 'content': chunk}, ensure_ascii=False)}\n\n"
-            yield f"data: {json.dumps({'type': 'done'}, ensure_ascii=False)}\n\n"
+            history_store.append_event(run_id, "final", {"content": content, "model": AI_MODEL})
+            history_store.finish_run(run_id, "completed", content)
+            yield f"data: {json.dumps({'type': 'done', 'run_id': run_id}, ensure_ascii=False)}\n\n"
         except httpx.HTTPStatusError as e:
             err = f"AI 接口返回错误: {e.response.status_code} - {str(e)[:300]}"
+            history_store.append_event(run_id, "error", {"message": err})
+            history_store.finish_run(run_id, "failed", err)
             yield f"data: {json.dumps({'type': 'error', 'message': err}, ensure_ascii=False)}\n\n"
         except httpx.ConnectError:
-            yield f"data: {json.dumps({'type': 'error', 'message': f'无法连接 AI 服务 ({AI_API_BASE})，请检查 .env 中的 AI_API_BASE 配置'}, ensure_ascii=False)}\n\n"
+            err = f"无法连接 AI 服务 ({AI_API_BASE})，请检查 .env 中的 AI_API_BASE 配置"
+            history_store.append_event(run_id, "error", {"message": err})
+            history_store.finish_run(run_id, "failed", err)
+            yield f"data: {json.dumps({'type': 'error', 'message': err}, ensure_ascii=False)}\n\n"
         except Exception as e:
-            yield f"data: {json.dumps({'type': 'error', 'message': f'AI 分析失败: {str(e)}'}, ensure_ascii=False)}\n\n"
+            err = f"AI 分析失败: {str(e)}"
+            history_store.append_event(run_id, "error", {"message": err})
+            history_store.finish_run(run_id, "failed", err)
+            yield f"data: {json.dumps({'type': 'error', 'message': err}, ensure_ascii=False)}\n\n"
+
+    return StreamingResponse(
+        event_generator(),
+        media_type="text/event-stream",
+        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
+    )
+
+
+@app.get("/api/agent/diagnose/stream")
+async def agent_diagnose_stream(question: str = "", execution_mode: str = "ai"):
+    """自主诊断 Agent：支持手动执行和安全 AI 执行模式。"""
+    if execution_mode not in {"ai", "manual"}:
+        raise HTTPException(status_code=400, detail="execution_mode 仅支持 ai 或 manual")
+    global _last_diagnosis, _last_report
+    if _last_diagnosis is None:
+        diagnosis = await run_full_diagnosis()
+        report = build_diagnosis_report(diagnosis)
+        _save_current_diagnosis(diagnosis, report)
+    run_id = history_store.create_run(_ensure_current_history(), "agent", execution_mode, question)
+
+    async def event_generator():
+        yield f"data: {json.dumps({'type': 'start', 'model': AI_MODEL, 'mode': execution_mode, 'run_id': run_id}, ensure_ascii=False)}\n\n"
+        final_status = "completed"
+        summary = ""
+        try:
+            async for event in run_diagnostic_agent(
+                _last_report,
+                question=question,
+                max_steps=0,
+                execution_mode=execution_mode,
+                run_id=run_id,
+                history_store=history_store,
+            ):
+                history_store.append_event(run_id, event.get("type", "event"), {k: v for k, v in event.items() if k != "type"})
+                if event.get("type") == "final":
+                    summary = event.get("content", "")
+                elif event.get("type") == "error":
+                    final_status = "failed"
+                    summary = event.get("message", "")
+                elif event.get("type") in {"manual_command", "approval_required"}:
+                    final_status = "waiting"
+                    if event.get("type") == "approval_required":
+                        from backend.ai_agent import _PENDING_APPROVALS
+                        pending = _PENDING_APPROVALS.get(event.get("approval_id"))
+                        if pending is not None:
+                            pending["run_id"] = run_id
+                yield f"data: {json.dumps(event, ensure_ascii=False)}\n\n"
+            history_store.finish_run(run_id, final_status, summary)
+            yield f"data: {json.dumps({'type': 'done', 'run_id': run_id}, ensure_ascii=False)}\n\n"
+        except Exception as exc:
+            message = f"自主诊断失败: {exc}"
+            history_store.append_event(run_id, "error", {"message": message})
+            history_store.finish_run(run_id, "failed", message)
+            yield f"data: {json.dumps({'type': 'error', 'message': message}, ensure_ascii=False)}\n\n"
+
+    return StreamingResponse(
+        event_generator(),
+        media_type="text/event-stream",
+        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
+    )
+
+
+@app.post("/api/agent/approve")
+async def execute_approved_agent_command(req: AgentApprovalRequest):
+    """人工批准后执行一次待审批的修改命令。"""
+    pending = take_pending_approval(req.approval_id)
+    if pending is None:
+        raise HTTPException(status_code=404, detail="审批请求不存在或已经处理")
+    if not req.approved:
+        result = {"approved": False, "command": pending["command"], "message": "用户已拒绝执行"}
+        if pending.get("run_id"):
+            history_store.append_event(pending["run_id"], "approval_rejected", result)
+            history_store.finish_run(pending["run_id"], "rejected", result["message"])
+        return result
+    result = await execute_command(pending["command"])
+    approved_event = {"approved": True, "reason": pending["reason"], **result}
+
+    async def event_generator():
+        yield f"data: {json.dumps({'type': 'approved_result', **approved_event}, ensure_ascii=False)}\n\n"
+        run_id = pending.get("run_id", "")
+        if run_id:
+            history_store.append_event(run_id, "approved_result", approved_event)
+
+        final_status = "completed"
+        summary = result.get("stdout") or result.get("stderr", "")
+        async for event in resume_diagnostic_agent(pending, result):
+            if run_id:
+                history_store.append_event(run_id, event.get("type", "event"), {k: v for k, v in event.items() if k != "type"})
+            if event.get("type") == "final":
+                summary = event.get("content", "")
+            elif event.get("type") == "error":
+                final_status = "failed"
+                summary = event.get("message", "")
+            elif event.get("type") == "approval_required":
+                final_status = "waiting"
+                from backend.ai_agent import _PENDING_APPROVALS
+                next_pending = _PENDING_APPROVALS.get(event.get("approval_id"))
+                if next_pending is not None:
+                    next_pending["run_id"] = run_id
+            yield f"data: {json.dumps(event, ensure_ascii=False)}\n\n"
+
+        if run_id:
+            history_store.finish_run(run_id, final_status, summary)
+        yield f"data: {json.dumps({'type': 'done', 'run_id': run_id}, ensure_ascii=False)}\n\n"
+
+    return StreamingResponse(
+        event_generator(),
+        media_type="text/event-stream",
+        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
+    )
+
+
+@app.get("/api/agent/resume-stream/{run_id}")
+async def agent_resume_stream(run_id: str):
+    """从 DB 上下文恢复 Agent 诊断，不重跑历史命令，直接继续诊断。"""
+    ctx = history_store.get_context(run_id)
+    if ctx is None:
+        raise HTTPException(status_code=404, detail="该运行记录不存在或不支持断点续传")
+    ctx.pop("report", None)  # get_context 已经注入了 report
+
+    async def event_generator():
+        yield f"data: {json.dumps({'type': 'start', 'mode': ctx.get('execution_mode', 'ai'), 'run_id': run_id, 'resumed': True}, ensure_ascii=False)}\n\n"
+        final_status = "completed"
+        summary = ""
+        try:
+            async for event in resume_from_context(
+                ctx=ctx,
+                run_id=run_id,
+                history_store=history_store,
+            ):
+                history_store.append_event(run_id, event.get("type", "event"), {k: v for k, v in event.items() if k != "type"})
+                if event.get("type") == "final":
+                    summary = event.get("content", "")
+                elif event.get("type") == "error":
+                    final_status = "failed"
+                    summary = event.get("message", "")
+                elif event.get("type") in {"manual_command", "approval_required"}:
+                    final_status = "waiting"
+                    if event.get("type") == "approval_required":
+                        from backend.ai_agent import _PENDING_APPROVALS
+                        pending = _PENDING_APPROVALS.get(event.get("approval_id"))
+                        if pending is not None:
+                            pending["run_id"] = run_id
+                yield f"data: {json.dumps(event, ensure_ascii=False)}\n\n"
+            history_store.finish_run(run_id, final_status, summary)
+            yield f"data: {json.dumps({'type': 'done', 'run_id': run_id}, ensure_ascii=False)}\n\n"
+        except Exception as exc:
+            message = f"断点续传诊断失败: {exc}"
+            history_store.append_event(run_id, "error", {"message": message})
+            history_store.finish_run(run_id, "failed", message)
+            yield f"data: {json.dumps({'type': 'error', 'message': message}, ensure_ascii=False)}\n\n"
 
     return StreamingResponse(
         event_generator(),
@@ -276,6 +485,27 @@ async def chat_stream(req: ChatRequest):
     )
 
 
+@app.get("/api/history/diagnoses")
+async def list_diagnosis_history(
+    status: str = "",
+    namespace: str = "",
+    limit: int = 100,
+    offset: int = 0,
+):
+    """查询全部历史诊断记录，可按状态和命名空间过滤。"""
+    items = history_store.list_diagnoses(status=status, namespace=namespace, limit=limit, offset=offset)
+    return {"total": len(items), "items": items, "limit": limit, "offset": offset}
+
+
+@app.get("/api/history/diagnoses/{record_id}")
+async def get_diagnosis_history(record_id: str):
+    """查询单次诊断及其全部 AI 分析、Agent 命令和处理记录。"""
+    record = history_store.get_diagnosis(record_id)
+    if record is None:
+        raise HTTPException(status_code=404, detail="历史诊断记录不存在")
+    return record
+
+
 @app.get("/api/report")
 async def get_report():
     """获取诊断报告文本"""

frontend/src/App.vue

@@ -14,9 +14,17 @@
         <el-button type="primary" :loading="diagnosing" @click="runDiagnosis" :icon="Refresh" :disabled="diagnosing">
           {{ diagnosing ? '诊断中...' : '一键诊断' }}
         </el-button>
-        <el-button type="success" :loading="analyzing" @click="runAnalysis" :icon="MagicStick" :disabled="!diagnosis || diagnosing">
+        <el-button type="success" :loading="analyzing" @click="runAnalysis" :icon="MagicStick" :disabled="!diagnosis || diagnosing || agentRunning">
           {{ analyzing ? 'AI 分析中...' : 'AI 分析' }}
         </el-button>
+        <el-radio-group v-model="agentExecutionMode" :disabled="agentRunning" class="agent-mode-switch">
+          <el-radio-button value="manual">手动执行命令</el-radio-button>
+          <el-radio-button value="ai">AI 执行命令</el-radio-button>
+        </el-radio-group>
+        <el-button @click="openHistory" :disabled="diagnosing || analyzing || agentRunning">历史记录</el-button>
+        <el-button type="warning" :loading="agentRunning" @click="runAgentDiagnosis" :disabled="!diagnosis || diagnosing || analyzing">
+          {{ agentRunning ? 'Agent 诊断中...' : '自主诊断' }}
+        </el-button>
       </div>
     </el-header>
 
@@ -257,6 +265,62 @@
         <div class="ai-content" v-html="renderedAnalysis"></div>
       </el-card>
 
+      <!-- 自主诊断 Agent -->
+      <el-card v-if="diagnosis && !diagnosing && (agentRunning || agentEvents.length)" class="section-card agent-card" shadow="never">
+        <template #header>
+          <div class="card-header">
+            <span>🧠 AI Agent 自主诊断</span>
+            <div class="ai-header-meta">
+              <el-tag :type="activeAgentMode === 'ai' ? 'success' : 'info'" size="small">
+                {{ activeAgentMode === 'ai' ? 'AI 执行模式' : '手动执行模式' }}
+              </el-tag>
+              <el-tag v-if="activeAgentMode === 'ai'" type="warning" size="small">修改命令人工确认</el-tag>
+            </div>
+          </div>
+        </template>
+        <el-alert
+          :title="activeAgentMode === 'ai'
+            ? 'AI 执行模式：Agent 自动执行只读 kubectl 和系统诊断命令 (systemctl status、journalctl、crictl ps、df 等)，修改或交互命令仍需人工批准。'
+            : '手动执行模式：Agent 只生成安全的诊断命令，不会在服务器上执行，请复制命令到终端手动运行。'"
+          type="info"
+          :closable="false"
+          show-icon
+          class="agent-alert"
+        />
+        <div class="agent-timeline">
+          <div v-for="(event, index) in agentEvents" :key="index" class="agent-event" :class="'event-' + event.type">
+            <div class="agent-event-title">
+              <el-icon v-if="event.type === 'thinking'" class="is-loading" color="#409eff"><Loading /></el-icon>
+              <el-icon v-else-if="event.type === 'result' || event.type === 'final'" color="#67c23a"><CircleCheckFilled /></el-icon>
+              <el-icon v-else-if="event.type === 'approval_required'" color="#e6a23c"><WarningFilled /></el-icon>
+              <el-icon v-else color="#909399"><Monitor /></el-icon>
+              <strong>{{ agentEventTitle(event) }}</strong>
+              <el-tag v-if="event.mode" :type="event.mode === 'read' ? 'success' : 'warning'" size="small">
+                {{ event.mode === 'read' ? '只读' : '需审批' }}
+              </el-tag>
+            </div>
+            <p v-if="event.message" class="agent-event-message">{{ event.message }}</p>
+            <p v-if="event.reason" class="agent-event-message">目的：{{ event.reason }}</p>
+            <pre v-if="event.command" class="agent-command">{{ event.command }}</pre>
+            <pre v-if="event.stdout || event.stderr" class="agent-output">{{ event.stdout || event.stderr }}</pre>
+            <div v-if="event.type === 'final'" class="ai-content" v-html="renderMd(event.content)"></div>
+            <div v-if="event.type === 'manual_command'" class="manual-actions">
+              <el-button type="primary" plain @click="copyAgentCommand(event.command)">复制命令</el-button>
+              <span>请在你的终端中执行，Agent 不会自动运行该命令。</span>
+            </div>
+            <div v-if="event.type === 'approval_required' && !event.resolved" class="approval-actions">
+              <el-button type="danger" @click="resolveApproval(event, true)" :loading="event.executing">确认执行</el-button>
+              <el-button @click="resolveApproval(event, false)" :disabled="event.executing">拒绝</el-button>
+            </div>
+            <el-alert v-if="event.approvalResult" :title="event.approvalResult" :type="event.approvalSuccess ? 'success' : 'info'" :closable="false" />
+          </div>
+          <div v-if="agentRunning" class="agent-running">
+            <el-icon class="is-loading" color="#409eff"><Loading /></el-icon>
+            <span>Agent 正在分析命令结果并规划下一步...</span>
+          </div>
+        </div>
+      </el-card>
+
       <!-- AI 对话 (连接成功或失败都显示) -->
       <el-card v-if="diagnosis && !diagnosing" class="section-card" shadow="never">
         <template #header><span>💬 追问 AI 助手</span></template>
@@ -274,6 +338,61 @@
         </div>
       </el-card>
     </el-main>
+
+    <el-drawer v-model="historyVisible" title="历史诊断与处理记录" size="75%" @open="loadHistory">
+      <div class="history-toolbar">
+        <el-select v-model="historyStatus" placeholder="全部状态" clearable style="width: 150px" @change="loadHistory">
+          <el-option label="健康" value="healthy" />
+          <el-option label="警告" value="warning" />
+          <el-option label="严重" value="critical" />
+        </el-select>
+        <el-input v-model="historyNamespace" placeholder="命名空间" clearable style="width: 180px" @keyup.enter="loadHistory" />
+        <el-button type="primary" @click="loadHistory" :loading="historyLoading">查询</el-button>
+      </div>
+      <el-table :data="historyItems" stripe highlight-current-row @row-click="loadHistoryDetail" v-loading="historyLoading">
+        <el-table-column prop="created_at" label="诊断时间" min-width="180" />
+        <el-table-column prop="namespace" label="命名空间" min-width="110" />
+        <el-table-column prop="score" label="评分" width="80" />
+        <el-table-column label="状态" width="90">
+          <template #default="{ row }"><el-tag :type="historyStatusType(row.status)">{{ historyStatusText(row.status) }}</el-tag></template>
+        </el-table-column>
+        <el-table-column prop="critical_count" label="严重" width="75" />
+        <el-table-column prop="warning_count" label="警告" width="75" />
+        <el-table-column prop="run_count" label="处理记录" width="100" />
+        <el-table-column label="操作" width="100">
+          <template #default="{ row }"><el-button link type="primary" @click.stop="loadHistoryDetail(row)">查看详情</el-button></template>
+        </el-table-column>
+      </el-table>
+
+      <div v-if="historyDetail" class="history-detail">
+        <el-divider content-position="left">诊断详情</el-divider>
+        <el-descriptions :column="4" border>
+          <el-descriptions-item label="时间">{{ historyDetail.created_at }}</el-descriptions-item>
+          <el-descriptions-item label="命名空间">{{ historyDetail.diagnosis.namespace }}</el-descriptions-item>
+          <el-descriptions-item label="健康评分">{{ historyDetail.diagnosis.score }}</el-descriptions-item>
+          <el-descriptions-item label="状态">{{ historyStatusText(historyDetail.diagnosis.status) }}</el-descriptions-item>
+        </el-descriptions>
+        <el-collapse class="history-collapse">
+          <el-collapse-item title="完整诊断报告" name="report">
+            <pre class="history-report">{{ historyDetail.report }}</pre>
+          </el-collapse-item>
+          <el-collapse-item v-for="run in historyDetail.runs" :key="run.id" :name="run.id">
+            <template #title>
+              <span>{{ runKindText(run.kind) }} · {{ run.started_at }} · {{ runStatusText(run.status) }}</span>
+            </template>
+            <el-descriptions :column="3" size="small" border>
+              <el-descriptions-item label="模式">{{ run.mode || '-' }}</el-descriptions-item>
+              <el-descriptions-item label="问题">{{ run.question || '-' }}</el-descriptions-item>
+              <el-descriptions-item label="结束时间">{{ run.finished_at || '-' }}</el-descriptions-item>
+            </el-descriptions>
+            <div v-for="(event, index) in run.events" :key="index" class="history-event">
+              <div class="history-event-head"><el-tag size="small">{{ event.type }}</el-tag><span>{{ event.created_at }}</span></div>
+              <pre>{{ formatHistoryEvent(event) }}</pre>
+            </div>
+          </el-collapse-item>
+        </el-collapse>
+      </div>
+    </el-drawer>
   </div>
 </template>
 
@@ -288,10 +407,20 @@ const API = '/api'
 const namespace = ref('')
 const diagnosing = ref(false)
 const analyzing = ref(false)
+const agentRunning = ref(false)
+const agentExecutionMode = ref('manual')
+const activeAgentMode = ref('manual')
+const agentEvents = ref([])
 const diagnosis = ref(null)
 const aiAnalysis = ref('')
 const aiModel = ref('')
 const analysisProcess = ref([])
+const historyVisible = ref(false)
+const historyLoading = ref(false)
+const historyItems = ref([])
+const historyDetail = ref(null)
+const historyStatus = ref('')
+const historyNamespace = ref('')
 const chatMessages = ref([])
 const chatInput = ref('')
 const chatLoading = ref(false)
@@ -462,6 +591,176 @@ async function runAnalysis() {
   }
 }
 
+function openHistory() {
+  historyVisible.value = true
+}
+
+async function loadHistory() {
+  historyLoading.value = true
+  historyDetail.value = null
+  try {
+    const params = new URLSearchParams({ limit: '200' })
+    if (historyStatus.value) params.set('status', historyStatus.value)
+    if (historyNamespace.value) params.set('namespace', historyNamespace.value)
+    const resp = await fetch(`${API}/history/diagnoses?${params.toString()}`)
+    if (!resp.ok) throw new Error(`HTTP ${resp.status}`)
+    const data = await resp.json()
+    historyItems.value = data.items || []
+  } catch (e) {
+    ElMessage.error('加载历史记录失败: ' + e.message)
+  } finally {
+    historyLoading.value = false
+  }
+}
+
+async function loadHistoryDetail(row) {
+  historyLoading.value = true
+  try {
+    const resp = await fetch(`${API}/history/diagnoses/${row.id}`)
+    if (!resp.ok) throw new Error(`HTTP ${resp.status}`)
+    historyDetail.value = await resp.json()
+  } catch (e) {
+    ElMessage.error('加载历史详情失败: ' + e.message)
+  } finally {
+    historyLoading.value = false
+  }
+}
+
+function historyStatusType(status) {
+  return status === 'healthy' ? 'success' : status === 'warning' ? 'warning' : status === 'critical' ? 'danger' : 'info'
+}
+
+function historyStatusText(status) {
+  return { healthy: '健康', warning: '警告', critical: '严重' }[status] || status
+}
+
+function runKindText(kind) {
+  return { analysis: 'AI 分析', agent: 'Agent 自主诊断', chat: 'AI 对话' }[kind] || kind
+}
+
+function runStatusText(status) {
+  return { running: '运行中', completed: '已完成', waiting: '等待处理', failed: '失败', rejected: '已拒绝' }[status] || status
+}
+
+function formatHistoryEvent(event) {
+  const { type, created_at, ...payload } = event
+  return JSON.stringify(payload, null, 2)
+}
+
+async function runAgentDiagnosis() {
+  activeAgentMode.value = agentExecutionMode.value
+  agentRunning.value = true
+  agentEvents.value = []
+  try {
+    const params = new URLSearchParams()
+    params.set('execution_mode', activeAgentMode.value)
+    if (namespace.value) params.set('question', `重点检查命名空间 ${namespace.value}`)
+    const resp = await fetch(`${API}/agent/diagnose/stream?${params.toString()}`)
+    if (!resp.ok || !resp.body) throw new Error(`HTTP ${resp.status}`)
+    const reader = resp.body.getReader()
+    const decoder = new TextDecoder()
+    let buffer = ''
+
+    while (true) {
+      const { done, value } = await reader.read()
+      if (done) break
+      buffer += decoder.decode(value, { stream: true })
+      const lines = buffer.split('\n')
+      buffer = lines.pop() || ''
+      for (const line of lines) {
+        if (!line.startsWith('data: ')) continue
+        try {
+          const event = JSON.parse(line.slice(6))
+          if (!['start', 'done'].includes(event.type)) agentEvents.value.push(event)
+          if (event.type === 'error') ElMessage.error(event.message)
+        } catch (e) {
+          // ignore partial events
+        }
+      }
+    }
+  } catch (e) {
+    ElMessage.error('自主诊断失败: ' + e.message)
+  } finally {
+    agentRunning.value = false
+  }
+}
+
+function agentEventTitle(event) {
+  const titles = {
+    thinking: 'Agent 思考', command: '准备执行命令', result: '命令执行结果',
+    manual_command: '请手动执行命令',
+    approval_required: '发现修改命令，等待审批', command_rejected: '命令已被安全策略拒绝',
+    verification_required: '需要继续复检', response_rejected: 'AI 响应格式已自动纠正',
+    final: '自主诊断结论', error: 'Agent 执行异常',
+  }
+  return titles[event.type] || event.type
+}
+
+async function copyAgentCommand(command) {
+  try {
+    await navigator.clipboard.writeText(command)
+    ElMessage.success('命令已复制，请在终端中手动执行')
+  } catch (e) {
+    ElMessage.error('复制失败，请手动选择命令文本')
+  }
+}
+
+async function resolveApproval(event, approved) {
+  event.executing = true
+  try {
+    const resp = await fetch(`${API}/agent/approve`, {
+      method: 'POST',
+      headers: { 'Content-Type': 'application/json' },
+      body: JSON.stringify({ approval_id: event.approval_id, approved }),
+    })
+    if (!resp.ok) {
+      const error = await resp.json()
+      throw new Error(error.detail || `HTTP ${resp.status}`)
+    }
+
+    if (!approved) {
+      const result = await resp.json()
+      event.resolved = true
+      event.approvalSuccess = false
+      event.approvalResult = '已拒绝执行该命令'
+      ElMessage.info(event.approvalResult)
+      return
+    }
+
+    if (!resp.body) throw new Error('审批接口未返回流式响应')
+    event.resolved = true
+    agentRunning.value = true
+    const reader = resp.body.getReader()
+    const decoder = new TextDecoder()
+    let buffer = ''
+    while (true) {
+      const { done, value } = await reader.read()
+      if (done) break
+      buffer += decoder.decode(value, { stream: true })
+      const lines = buffer.split('\n')
+      buffer = lines.pop() || ''
+      for (const line of lines) {
+        if (!line.startsWith('data: ')) continue
+        const nextEvent = JSON.parse(line.slice(6))
+        if (nextEvent.type === 'approved_result') {
+          event.approvalSuccess = nextEvent.returncode === 0
+          event.approvalResult = nextEvent.returncode === 0 ? '命令执行成功，Agent 继续诊断中' : `命令执行失败：${nextEvent.stderr || nextEvent.returncode}`
+          agentEvents.value.push({ type: 'result', ...nextEvent })
+        } else if (!['start', 'done'].includes(nextEvent.type)) {
+          agentEvents.value.push(nextEvent)
+          if (nextEvent.type === 'error') ElMessage.error(nextEvent.message)
+        }
+      }
+    }
+    ElMessage.success('审批命令已执行，Agent 已继续诊断')
+  } catch (e) {
+    ElMessage.error('审批处理失败: ' + e.message)
+  } finally {
+    event.executing = false
+    agentRunning.value = false
+  }
+}
+
 async function sendChat() {
   const q = chatInput.value.trim()
   if (!q || chatLoading.value) return
@@ -611,6 +910,34 @@ body { font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', 'PingFang SC'
 .ai-content table { border-collapse: collapse; width: 100%; margin: 12px 0; }
 .ai-content th, .ai-content td { border: 1px solid #ebeef5; padding: 8px 12px; text-align: left; }
 
+.agent-card { border: 1px solid #f3d19e; }
+.agent-mode-switch { flex-shrink: 0; }
+.agent-alert { margin-bottom: 16px; }
+.agent-timeline { display: flex; flex-direction: column; gap: 12px; }
+.agent-event { padding: 14px; border: 1px solid #ebeef5; border-radius: 8px; background: #fafafa; }
+.agent-event.event-thinking { border-color: #a0cfff; background: #ecf5ff; }
+.agent-event.event-approval_required { border-color: #eebe77; background: #fdf6ec; }
+.agent-event.event-final { border-color: #b3e19d; background: #f0f9eb; }
+.agent-event-title { display: flex; align-items: center; gap: 8px; color: #303133; }
+.agent-event-message { margin-top: 8px; color: #606266; font-size: 13px; }
+.agent-command, .agent-output {
+  margin-top: 10px; padding: 12px; border-radius: 6px; overflow-x: auto;
+  white-space: pre-wrap; word-break: break-all; font-size: 13px;
+}
+.agent-command { color: #d4d4d4; background: #1e1e1e; }
+.agent-output { max-height: 320px; color: #303133; background: #f4f4f5; }
+.approval-actions { display: flex; gap: 8px; margin-top: 12px; }
+.manual-actions { display: flex; align-items: center; gap: 12px; margin-top: 12px; color: #606266; font-size: 13px; }
+.agent-running { display: flex; align-items: center; gap: 8px; color: #409eff; padding: 8px; }
+
+.history-toolbar { display: flex; gap: 10px; margin-bottom: 16px; }
+.history-detail { margin-top: 20px; }
+.history-collapse { margin-top: 16px; }
+.history-report { max-height: 480px; padding: 14px; overflow: auto; border-radius: 6px; background: #f4f4f5; white-space: pre-wrap; }
+.history-event { margin-top: 10px; padding: 12px; border: 1px solid #ebeef5; border-radius: 6px; }
+.history-event-head { display: flex; align-items: center; gap: 10px; color: #909399; font-size: 12px; }
+.history-event pre { max-height: 300px; margin-top: 8px; padding: 10px; overflow: auto; background: #f4f4f5; white-space: pre-wrap; }
+
 .chat-box { max-height: 400px; overflow-y: auto; padding: 12px 0; display: flex; flex-direction: column; gap: 12px; }
 .chat-msg { display: flex; }
 .chat-msg.user { justify-content: flex-end; }

tests/test_agent_api.py

@@ -0,0 +1,97 @@
+import asyncio
+import json
+import unittest
+from unittest.mock import patch
+
+import backend.main as main
+
+
+def parse_sse(event):
+    return json.loads(event[6:].strip())
+
+
+class AgentApiTest(unittest.TestCase):
+    def test_agent_stream_forwards_agent_events(self):
+        async def fake_agent(report, question="", max_steps=6, execution_mode="ai"):
+            self.assertEqual(execution_mode, "manual")
+            yield {"type": "thinking", "step": 1, "message": "规划"}
+            yield {"type": "final", "content": "完成"}
+
+        async def run():
+            main._last_diagnosis = {"score": 60}
+            main._last_report = "report"
+            with patch.object(main, "run_diagnostic_agent", fake_agent):
+                response = await main.agent_diagnose_stream(question="检查 Pod", execution_mode="manual")
+                return [parse_sse(event) async for event in response.body_iterator]
+
+        events = asyncio.run(run())
+        self.assertEqual(events[0]["type"], "start")
+        self.assertEqual(events[-1]["type"], "done")
+        self.assertEqual(events[-2]["type"], "final")
+
+    def test_config_view_is_read_only_and_does_not_require_approval(self):
+        from backend.agent_tools import classify_kubectl_command
+
+        decision = classify_kubectl_command("kubectl config view")
+
+        self.assertEqual(decision.mode, "read")
+        self.assertFalse(decision.requires_approval)
+
+    def test_execute_approved_command_continues_agent_with_command_result(self):
+        approval = {
+            "command": "kubectl rollout restart deployment/api -n prod",
+            "reason": "重启工作负载",
+            "run_id": "",
+            "report": "diagnosis report",
+            "question": "修复 API",
+            "execution_mode": "ai",
+            "messages": [{"role": "system", "content": "context"}],
+            "next_step": 2,
+            "max_steps": 30,
+            "command_counts": {"kubectl rollout restart deployment/api -n prod": 1},
+        }
+
+        async def fake_execute(command):
+            return {"command": command, "mode": "write", "returncode": 0, "stdout": "restarted", "stderr": ""}
+
+        async def fake_resume(pending, result):
+            self.assertEqual(result["stdout"], "restarted")
+            yield {"type": "thinking", "step": 2, "message": "继续验证"}
+            yield {"type": "command", "command": "kubectl get pods -n prod", "mode": "read", "reason": "验证恢复"}
+            yield {"type": "result", "command": "kubectl get pods -n prod", "mode": "read", "returncode": 0, "stdout": "Running", "stderr": ""}
+            yield {"type": "final", "content": "故障已恢复", "model": "test"}
+
+        async def run():
+            with patch.object(main, "take_pending_approval", return_value=approval), patch.object(main, "execute_command", fake_execute), patch.object(main, "resume_diagnostic_agent", fake_resume):
+                response = await main.execute_approved_agent_command(main.AgentApprovalRequest(approval_id="abc", approved=True))
+                return [parse_sse(event) async for event in response.body_iterator]
+
+        events = asyncio.run(run())
+        self.assertEqual(events[0]["type"], "approved_result")
+        self.assertEqual(events[1]["type"], "thinking")
+        self.assertEqual(events[-2]["type"], "final")
+        self.assertEqual(events[-1]["type"], "done")
+
+    def test_execute_approved_command_once(self):
+        approval = {"command": "kubectl scale deployment/api --replicas=2", "reason": "恢复副本"}
+
+        async def fake_execute(command):
+            return {"command": command, "mode": "write", "returncode": 0, "stdout": "scaled", "stderr": ""}
+
+        async def fake_resume(pending, result):
+            yield {"type": "final", "content": "验证完成", "model": "test"}
+
+        async def run():
+            with patch.object(main, "take_pending_approval", return_value=approval), patch.object(main, "execute_command", fake_execute), patch.object(main, "resume_diagnostic_agent", fake_resume):
+                response = await main.execute_approved_agent_command(main.AgentApprovalRequest(approval_id="abc", approved=True))
+                return [parse_sse(event) async for event in response.body_iterator]
+
+        events = asyncio.run(run())
+        self.assertEqual(events[0]["type"], "approved_result")
+        self.assertEqual(events[0]["returncode"], 0)
+        self.assertEqual(events[0]["stdout"], "scaled")
+        self.assertEqual(events[-1]["type"], "done")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_agent_tools.py

@@ -0,0 +1,214 @@
+import unittest
+
+from backend.agent_tools import (
+    classify_agent_command,
+    classify_kubectl_command,
+    classify_shell_command,
+    parse_kubectl_command,
+    parse_shell_command,
+)
+
+
+class KubectlCommandPolicyTest(unittest.TestCase):
+    def test_allows_read_only_kubectl_command_for_autonomous_execution(self):
+        decision = classify_kubectl_command("kubectl get pods -n production -o wide")
+
+        self.assertEqual(decision.mode, "read")
+        self.assertFalse(decision.requires_approval)
+
+    def test_requires_approval_for_mutating_kubectl_command(self):
+        decision = classify_kubectl_command("kubectl rollout restart deployment/api -n production")
+
+        self.assertEqual(decision.mode, "write")
+        self.assertTrue(decision.requires_approval)
+
+    def test_rejects_shell_operators_and_non_kubectl_commands(self):
+        # 武器级危险命令（sh/bash）仍被拒绝
+        with self.assertRaises(ValueError):
+            parse_kubectl_command("kubectl get pods | sh")
+        # 非 kubectl 命令仍被拒绝
+        for command in (
+            "bash -c kubectl get pods",
+            "sudo kubectl get pods",
+        ):
+            with self.subTest(command=command):
+                with self.assertRaises(ValueError):
+                    parse_kubectl_command(command)
+
+    def test_rejects_exec_even_when_command_looks_read_only(self):
+        decision = classify_kubectl_command("kubectl exec api-0 -- cat /etc/hosts")
+
+        self.assertEqual(decision.mode, "write")
+        self.assertTrue(decision.requires_approval)
+
+
+class ShellCommandPolicyTest(unittest.TestCase):
+    def test_allows_read_only_systemctl_status_for_autonomous_execution(self):
+        decision = classify_shell_command("systemctl status kubelet")
+
+        self.assertEqual(decision.mode, "read")
+        self.assertFalse(decision.requires_approval)
+
+    def test_allows_journalctl_with_flags(self):
+        decision = classify_shell_command("journalctl -u kubelet -n 100 --no-pager")
+
+        self.assertEqual(decision.mode, "read")
+        self.assertFalse(decision.requires_approval)
+
+    def test_allows_crictl_ps(self):
+        decision = classify_shell_command("crictl ps -a")
+
+        self.assertEqual(decision.mode, "read")
+        self.assertFalse(decision.requires_approval)
+
+    def test_allows_df_and_free(self):
+        for command in ("df -h", "free -m", "uptime"):
+            with self.subTest(command=command):
+                decision = classify_shell_command(command)
+                self.assertEqual(decision.mode, "read")
+                self.assertFalse(decision.requires_approval)
+
+    def test_allows_ip_addr(self):
+        decision = classify_shell_command("ip addr show")
+
+        self.assertEqual(decision.mode, "read")
+        self.assertFalse(decision.requires_approval)
+
+    def test_requires_approval_for_systemctl_restart(self):
+        decision = classify_shell_command("systemctl restart kubelet")
+
+        self.assertEqual(decision.mode, "write")
+        self.assertTrue(decision.requires_approval)
+
+    def test_rejects_unknown_binary(self):
+        # rm 现在在写白名单中（需审批），不会直接拒绝
+        decision = classify_shell_command("rm -rf /tmp/cache")
+        self.assertEqual(decision.mode, "write")
+        self.assertTrue(decision.requires_approval)
+        # 完全不在任何白名单中的命令才拒绝
+        with self.assertRaises(ValueError):
+            classify_shell_command("hacktool -x")
+
+    def test_rejects_shell_operators_in_shell_command(self):
+        # 所有 shell 操作符（; | > && ||）现在都允许通过分类器
+        # 分号 ;
+        decision = classify_shell_command("systemctl status kubelet; df -h")
+        self.assertEqual(decision.mode, "read")
+        self.assertTrue(decision.needs_shell)
+        # 管道 |
+        decision = classify_shell_command("df -h | grep vda")
+        self.assertEqual(decision.mode, "read")
+        self.assertFalse(decision.requires_approval)
+        self.assertTrue(decision.needs_shell)
+        # 重定向 >
+        decision = classify_shell_command("journalctl -u kubelet > /tmp/log")
+        self.assertEqual(decision.mode, "read")
+        self.assertFalse(decision.requires_approval)
+        self.assertTrue(decision.needs_shell)
+        # 武器级 sh/bash 仍拒绝
+        with self.assertRaises(ValueError):
+            classify_shell_command("crictl ps -a | bash")
+        with self.assertRaises(ValueError):
+            classify_shell_command("crictl ps -a && sh -c 'echo x'")
+
+    def test_cat_only_allows_safe_paths(self):
+        # cat 完全放行，任意路径均可
+        decision = classify_shell_command("cat /etc/os-release")
+        self.assertEqual(decision.mode, "read")
+        self.assertFalse(decision.requires_approval)
+
+    def test_cat_rejects_unsafe_path(self):
+        # cat 不再限制路径
+        decision = classify_shell_command("cat /etc/shadow")
+        self.assertEqual(decision.mode, "read")
+        self.assertFalse(decision.requires_approval)
+
+
+class EtcdCommandPolicyTest(unittest.TestCase):
+    def test_etcdctl_endpoint_status_is_read_only(self):
+        decision = classify_shell_command("etcdctl endpoint status")
+        self.assertEqual(decision.mode, "read")
+        self.assertFalse(decision.requires_approval)
+
+    def test_etcdctl_with_global_flags_still_classifies(self):
+        decision = classify_shell_command(
+            "etcdctl --endpoints=https://127.0.0.1:2379 --cacert=/etc/ssl/ca.pem endpoint health"
+        )
+        self.assertEqual(decision.mode, "read")
+        self.assertFalse(decision.requires_approval)
+
+    def test_etcdctl_member_list_is_read_only(self):
+        decision = classify_shell_command("etcdctl member list")
+        self.assertEqual(decision.mode, "read")
+        self.assertFalse(decision.requires_approval)
+
+    def test_etcdctl_get_and_version_are_read_only(self):
+        for command in ("etcdctl get /kubernetes", "etcdctl version", "etcdctl alarm list"):
+            with self.subTest(command=command):
+                decision = classify_shell_command(command)
+                self.assertEqual(decision.mode, "read")
+                self.assertFalse(decision.requires_approval)
+
+    def test_etcdctl_alarm_list_is_read_only(self):
+        decision = classify_shell_command("etcdctl alarm list")
+        self.assertEqual(decision.mode, "read")
+        self.assertFalse(decision.requires_approval)
+
+    def test_etcdctl_put_requires_approval(self):
+        decision = classify_shell_command("etcdctl put key value")
+        self.assertEqual(decision.mode, "write")
+        self.assertTrue(decision.requires_approval)
+
+    def test_etcdctl_member_remove_requires_approval(self):
+        decision = classify_shell_command("etcdctl member remove abc123")
+        self.assertEqual(decision.mode, "write")
+        self.assertTrue(decision.requires_approval)
+
+    def test_etcdctl_snapshot_save_requires_approval(self):
+        decision = classify_shell_command("etcdctl snapshot save /tmp/etcd.db")
+        self.assertEqual(decision.mode, "write")
+        self.assertTrue(decision.requires_approval)
+
+    def test_etcdctl_compact_requires_approval(self):
+        decision = classify_shell_command("etcdctl compact 12345")
+        self.assertEqual(decision.mode, "write")
+        self.assertTrue(decision.requires_approval)
+
+    def test_etcdctl_auth_enable_requires_approval(self):
+        decision = classify_shell_command("etcdctl auth enable")
+        self.assertEqual(decision.mode, "write")
+        self.assertTrue(decision.requires_approval)
+
+    def test_etcdctl_rejects_unknown_subcommand(self):
+        with self.assertRaises(ValueError):
+            classify_shell_command("etcdctl bogus command")
+
+
+class AgentCommandDispatchTest(unittest.TestCase):
+    def test_dispatches_kubectl_command_to_kubectl_classifier(self):
+        decision = classify_agent_command("kubectl get pods -A")
+
+        self.assertEqual(decision.mode, "read")
+        self.assertEqual(decision.verb, "get")
+
+    def test_dispatches_shell_command_to_shell_classifier(self):
+        decision = classify_agent_command("systemctl status kubelet")
+
+        self.assertEqual(decision.mode, "read")
+        self.assertEqual(decision.verb, "systemctl")
+
+    def test_dispatches_write_shell_command_to_shell_classifier(self):
+        decision = classify_agent_command("systemctl restart kubelet")
+
+        self.assertEqual(decision.mode, "write")
+        self.assertTrue(decision.requires_approval)
+
+    def test_dispatches_write_kubectl_command(self):
+        decision = classify_agent_command("kubectl delete pod api-0 -n prod")
+
+        self.assertEqual(decision.mode, "write")
+        self.assertTrue(decision.requires_approval)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_diagnostic_agent.py

@@ -0,0 +1,328 @@
+import asyncio
+import unittest
+from unittest.mock import patch
+
+from backend.ai_agent import run_diagnostic_agent
+
+
+async def collect_events(stream):
+    return [event async for event in stream]
+
+
+class DiagnosticAgentTest(unittest.TestCase):
+    def test_executes_read_only_tool_and_returns_final_answer(self):
+        responses = iter([
+            '{"action":"command","command":"kubectl get pods -A","reason":"查看异常 Pod"}',
+            '{"action":"final","analysis":"发现一个异常 Pod","verified":true,"remaining_issues":0}',
+        ])
+
+        async def fake_completion(messages):
+            return next(responses)
+
+        async def fake_execute(command):
+            return {"command": command, "mode": "read", "returncode": 0, "stdout": "pod-a Running", "stderr": ""}
+
+        async def run():
+            with patch("backend.ai_agent._complete_chat", fake_completion), patch("backend.ai_agent.execute_command", fake_execute):
+                return await collect_events(run_diagnostic_agent("report", max_steps=3))
+
+        events = asyncio.run(run())
+        self.assertEqual(
+            [event["type"] for event in events],
+            ["command", "result", "thinking", "command", "result", "thinking", "final"],
+        )
+        self.assertEqual(events[3]["mode"], "read")
+        self.assertEqual(events[-1]["content"], "发现一个异常 Pod")
+
+    def test_pauses_mutating_command_for_human_approval(self):
+        async def fake_completion(messages):
+            return '{"action":"command","command":"kubectl rollout restart deployment/api -n prod","reason":"重启故障工作负载"}'
+
+        async def run():
+            with patch("backend.ai_agent._complete_chat", fake_completion):
+                return await collect_events(run_diagnostic_agent("report", max_steps=3))
+
+        events = asyncio.run(run())
+        self.assertEqual(events[-1]["type"], "approval_required")
+        self.assertEqual(events[-1]["command"], "kubectl rollout restart deployment/api -n prod")
+        self.assertTrue(events[-1]["approval_id"])
+
+    def test_manual_mode_never_executes_command_and_waits_for_result(self):
+        async def fake_completion(messages):
+            return '{"action":"command","command":"kubectl get pods -A","reason":"查看 Pod"}'
+
+        async def run():
+            with patch("backend.ai_agent._complete_chat", fake_completion), patch("backend.ai_agent.execute_command") as execute:
+                events = await collect_events(run_diagnostic_agent("report", max_steps=3, execution_mode="manual"))
+                return events, execute
+
+        events, execute = asyncio.run(run())
+        execute.assert_not_called()
+        self.assertEqual(events[-1]["type"], "manual_command")
+        self.assertEqual(events[-1]["command"], "kubectl get pods -A")
+        self.assertTrue(events[-1]["command_id"])
+
+    def test_ai_mode_keeps_write_command_behind_approval(self):
+        async def fake_completion(messages):
+            return '{"action":"command","command":"kubectl delete pod api-0 -n prod","reason":"删除故障 Pod"}'
+
+        executed = []
+
+        async def run():
+            async def fake_execute(command):
+                executed.append(command)
+                return {"command": command, "mode": "read", "returncode": 0, "stdout": "nodes ready", "stderr": ""}
+
+            with patch("backend.ai_agent._complete_chat", fake_completion), patch("backend.ai_agent.execute_command", fake_execute):
+                return await collect_events(run_diagnostic_agent("report", max_steps=3, execution_mode="ai"))
+
+        events = asyncio.run(run())
+        self.assertEqual(executed, ["kubectl get nodes -o wide"])
+        self.assertEqual(events[-1]["type"], "approval_required")
+
+    def test_rejects_unknown_execution_mode(self):
+        async def run():
+            return await collect_events(run_diagnostic_agent("report", execution_mode="invalid"))
+
+        events = asyncio.run(run())
+        self.assertEqual(events[-1]["type"], "error")
+        self.assertIn("执行模式", events[-1]["message"])
+
+    def test_ai_mode_forces_initial_read_only_probe_before_model_planning(self):
+        responses = iter([
+            '{"action":"final","analysis":"根据采集结果完成分析","verified":true,"remaining_issues":0}',
+        ])
+
+        async def fake_completion(messages):
+            return next(responses)
+
+        executed = []
+
+        async def fake_execute(command):
+            executed.append(command)
+            return {"command": command, "mode": "read", "returncode": 0, "stdout": "node-a Ready", "stderr": ""}
+
+        async def run():
+            with patch("backend.ai_agent._complete_chat", fake_completion), patch("backend.ai_agent.execute_command", fake_execute):
+                return await collect_events(run_diagnostic_agent("report", max_steps=3, execution_mode="ai"))
+
+        events = asyncio.run(run())
+        self.assertEqual(executed, ["kubectl get nodes -o wide"])
+        self.assertEqual([event["type"] for event in events[:2]], ["command", "result"])
+        self.assertEqual(events[-1]["type"], "final")
+
+    def test_parse_agent_action_accepts_markdown_json_fence(self):
+        from backend.ai_agent import _parse_agent_action
+
+        action = _parse_agent_action(
+            '我将先检查 Pod。\n```json\n{"action":"command","command":"kubectl get pods -A","reason":"查看 Pod"}\n```'
+        )
+
+        self.assertEqual(action["action"], "command")
+        self.assertEqual(action["command"], "kubectl get pods -A")
+
+    def test_ai_mode_continues_beyond_six_commands_until_model_finishes(self):
+        command_count = 8
+        responses = iter([
+            *[
+                (
+                    '{"action":"command","command":"kubectl get pods -A '
+                    f'--field-selector metadata.name=pod-{index}","reason":"继续收集证据"}}'
+                )
+                for index in range(command_count)
+            ],
+            '{"action":"final","analysis":"已完成根因分析","verified":true,"remaining_issues":0}',
+        ])
+        executed = []
+
+        async def fake_completion(messages):
+            return next(responses)
+
+        async def fake_execute(command):
+            executed.append(command)
+            return {"command": command, "mode": "read", "returncode": 0, "stdout": "evidence", "stderr": ""}
+
+        async def run():
+            with patch("backend.ai_agent._complete_chat", fake_completion), patch("backend.ai_agent.execute_command", fake_execute):
+                return await collect_events(run_diagnostic_agent("report", execution_mode="ai", max_steps=20))
+
+        events = asyncio.run(run())
+        self.assertEqual(len(executed), command_count + 1)
+        self.assertEqual(events[-1]["type"], "final")
+        self.assertEqual(events[-1]["content"], "已完成根因分析")
+        self.assertNotIn("最大轮次", events[-1]["content"])
+
+    def test_safety_limit_reports_agent_error_instead_of_sending_user_to_manual_diagnosis(self):
+        async def fake_completion(messages):
+            return '{"action":"command","command":"kubectl get pods -A","reason":"继续调查"}'
+
+        async def fake_execute(command):
+            return {"command": command, "mode": "read", "returncode": 0, "stdout": "same result", "stderr": ""}
+
+        async def run():
+            with patch("backend.ai_agent._complete_chat", fake_completion), patch("backend.ai_agent.execute_command", fake_execute):
+                return await collect_events(run_diagnostic_agent("report", execution_mode="ai", max_steps=2))
+
+        events = asyncio.run(run())
+        self.assertEqual(events[-1]["type"], "error")
+        self.assertIn("安全上限", events[-1]["message"])
+        self.assertNotIn("人工排查", events[-1]["message"])
+
+    def test_agent_rejects_final_until_verification_reports_no_remaining_issues(self):
+        responses = iter([
+            '{"action":"final","analysis":"已经修复"}',
+            '{"action":"command","command":"kubectl get pods -A","reason":"执行全量复检"}',
+            '{"action":"final","analysis":"所有工作负载恢复正常","verified":true,"remaining_issues":0}',
+        ])
+        executed = []
+
+        async def fake_completion(messages):
+            return next(responses)
+
+        async def fake_execute(command):
+            executed.append(command)
+            stdout = "node-a Ready" if command == "kubectl get nodes -o wide" else "all pods Running"
+            return {"command": command, "mode": "read", "returncode": 0, "stdout": stdout, "stderr": ""}
+
+        async def run():
+            with patch("backend.ai_agent._complete_chat", fake_completion), patch("backend.ai_agent.execute_command", fake_execute):
+                return await collect_events(run_diagnostic_agent("report", execution_mode="ai", max_steps=5))
+
+        events = asyncio.run(run())
+        self.assertEqual(executed, ["kubectl get nodes -o wide", "kubectl get pods -A"])
+        self.assertEqual(events[-1]["type"], "final")
+        self.assertTrue(events[-1]["verified"])
+        self.assertEqual(events[-1]["remaining_issues"], 0)
+
+    def test_parse_final_requires_explicit_verification_fields(self):
+        from backend.ai_agent import _parse_agent_action
+
+        with self.assertRaises(ValueError):
+            _parse_agent_action('{"action":"final","analysis":"看起来正常"}')
+
+        action = _parse_agent_action(
+            '{"action":"final","analysis":"全部正常","verified":true,"remaining_issues":0}'
+        )
+        self.assertTrue(action["verified"])
+        self.assertEqual(action["remaining_issues"], 0)
+
+    def test_invalid_json_response_is_retried_instead_of_stopping_agent(self):
+        responses = iter([
+            "我需要继续检查，但没有按格式输出",
+            '{"action":"command","command":"kubectl get pods -A","reason":"继续检查"}',
+            '{"action":"final","analysis":"检查完成","verified":true,"remaining_issues":0}',
+        ])
+        executed = []
+
+        async def fake_completion(messages):
+            return next(responses)
+
+        async def fake_execute(command):
+            executed.append(command)
+            return {"command": command, "mode": "read", "returncode": 0, "stdout": "healthy", "stderr": ""}
+
+        async def run():
+            with patch("backend.ai_agent._complete_chat", fake_completion), patch("backend.ai_agent.execute_command", fake_execute):
+                return await collect_events(run_diagnostic_agent("report", max_steps=5))
+
+        events = asyncio.run(run())
+        self.assertTrue(any(event["type"] == "response_rejected" for event in events))
+        self.assertEqual(events[-1]["type"], "final")
+        self.assertEqual(executed, ["kubectl get nodes -o wide", "kubectl get pods -A"])
+
+    def test_stops_after_three_consecutive_invalid_json_responses(self):
+        async def fake_completion(messages):
+            return "still invalid"
+
+        async def fake_execute(command):
+            return {"command": command, "mode": "read", "returncode": 0, "stdout": "node Ready", "stderr": ""}
+
+        async def run():
+            with patch("backend.ai_agent._complete_chat", fake_completion), patch("backend.ai_agent.execute_command", fake_execute):
+                return await collect_events(run_diagnostic_agent("report", max_steps=10))
+
+        events = asyncio.run(run())
+        self.assertEqual(events[-1]["type"], "error")
+        self.assertIn("连续 3 次", events[-1]["message"])
+
+    def test_stops_when_model_returns_invalid_action(self):
+        async def fake_completion(messages):
+            return "not-json"
+
+        async def run():
+            with patch("backend.ai_agent._complete_chat", fake_completion):
+                return await collect_events(run_diagnostic_agent("report", max_steps=3))
+
+        events = asyncio.run(run())
+        self.assertEqual(events[-1]["type"], "error")
+        self.assertIn("JSON", events[-1]["message"])
+
+    def test_agent_can_mix_kubectl_and_shell_commands(self):
+        """Agent 应能在同一轮诊断中交替使用 kubectl 和系统诊断 shell 命令。"""
+        responses = iter([
+            '{"action":"command","command":"kubectl get nodes -o wide","reason":"查看节点状态"}',
+            '{"action":"command","command":"systemctl status kubelet","reason":"检查 kubelet 服务"}',
+            '{"action":"command","command":"journalctl -u kubelet -n 50 --no-pager","reason":"查看 kubelet 日志"}',
+            '{"action":"final","analysis":"kubelet 服务正常，节点就绪","verified":true,"remaining_issues":0}',
+        ])
+        executed = []
+
+        async def fake_completion(messages):
+            return next(responses)
+
+        async def fake_execute(command):
+            executed.append(command)
+            if command.startswith("kubectl"):
+                stdout = "node-a Ready"
+            elif command.startswith("systemctl"):
+                stdout = "active (running)"
+            else:
+                stdout = "no errors"
+            return {"command": command, "mode": "read", "returncode": 0, "stdout": stdout, "stderr": ""}
+
+        async def run():
+            with patch("backend.ai_agent._complete_chat", fake_completion), patch("backend.ai_agent.execute_command", fake_execute):
+                return await collect_events(run_diagnostic_agent("report", execution_mode="ai", max_steps=10))
+
+        events = asyncio.run(run())
+        # 初始探测 + 3 条 Agent 命令 = 4 次执行
+        self.assertEqual(
+            executed,
+            ["kubectl get nodes -o wide", "kubectl get nodes -o wide", "systemctl status kubelet", "journalctl -u kubelet -n 50 --no-pager"],
+        )
+        # 所有命令都是只读模式 (自动执行)
+        command_events = [e for e in events if e["type"] == "command"]
+        self.assertTrue(all(e["mode"] == "read" for e in command_events))
+        self.assertEqual(events[-1]["type"], "final")
+
+    def test_agent_rejects_non_whitelisted_shell_command(self):
+        """Agent 应拒绝不在白名单内的 shell 命令 (如 rm)。"""
+        responses = iter([
+            '{"action":"command","command":"rm -rf /tmp/test","reason":"清理测试文件"}',
+            '{"action":"final","analysis":"已完成","verified":true,"remaining_issues":0}',
+        ])
+        executed = []
+
+        async def fake_completion(messages):
+            return next(responses)
+
+        async def fake_execute(command):
+            executed.append(command)
+            return {"command": command, "mode": "read", "returncode": 0, "stdout": "", "stderr": ""}
+
+        async def run():
+            with patch("backend.ai_agent._complete_chat", fake_completion), patch("backend.ai_agent.execute_command", fake_execute):
+                return await collect_events(run_diagnostic_agent("report", execution_mode="ai", max_steps=5))
+
+        events = asyncio.run(run())
+        # rm 命令应被拒绝
+        rejected = [e for e in events if e["type"] == "command_rejected"]
+        self.assertTrue(rejected)
+        self.assertIn("不允许执行该命令", rejected[0]["reason"])
+        # rm 不应被执行
+        self.assertNotIn("rm -rf /tmp/test", executed)
+        self.assertEqual(events[-1]["type"], "final")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_history_api.py

@@ -0,0 +1,43 @@
+import asyncio
+import os
+import tempfile
+import unittest
+from unittest.mock import patch
+
+import backend.main as main
+from backend.history_store import HistoryStore
+
+
+class HistoryApiTest(unittest.TestCase):
+    def setUp(self):
+        fd, self.path = tempfile.mkstemp(suffix=".db")
+        os.close(fd)
+        self.store = HistoryStore(self.path)
+        self.original_store = main.history_store
+        main.history_store = self.store
+
+    def tearDown(self):
+        main.history_store = self.original_store
+        os.remove(self.path)
+
+    def test_lists_and_gets_full_diagnosis_history(self):
+        record_id = self.store.save_diagnosis(
+            {"timestamp": "2026-07-25T12:00:00", "namespace": "prod", "score": 50, "status": "critical"},
+            "report",
+        )
+
+        records = asyncio.run(main.list_diagnosis_history())
+        detail = asyncio.run(main.get_diagnosis_history(record_id))
+
+        self.assertEqual(records["total"], 1)
+        self.assertEqual(records["items"][0]["id"], record_id)
+        self.assertEqual(detail["report"], "report")
+
+    def test_get_history_returns_404_for_unknown_id(self):
+        with self.assertRaises(main.HTTPException) as context:
+            asyncio.run(main.get_diagnosis_history("missing"))
+        self.assertEqual(context.exception.status_code, 404)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_history_store.py

@@ -0,0 +1,69 @@
+import os
+import tempfile
+import unittest
+
+from backend.history_store import HistoryStore
+
+
+class HistoryStoreTest(unittest.TestCase):
+    def setUp(self):
+        fd, self.path = tempfile.mkstemp(suffix=".db")
+        os.close(fd)
+        self.store = HistoryStore(self.path)
+
+    def tearDown(self):
+        os.remove(self.path)
+
+    def test_persists_diagnosis_and_can_query_after_reopen(self):
+        diagnosis = {
+            "timestamp": "2026-07-25T12:00:00",
+            "namespace": "prod",
+            "score": 65,
+            "status": "warning",
+            "critical_count": 1,
+            "warning_count": 4,
+            "issues": [{"level": "critical", "message": "pod failed"}],
+            "checks": {},
+        }
+        record_id = self.store.save_diagnosis(diagnosis, "report text")
+
+        reopened = HistoryStore(self.path)
+        records = reopened.list_diagnoses()
+        detail = reopened.get_diagnosis(record_id)
+
+        self.assertEqual(len(records), 1)
+        self.assertEqual(records[0]["namespace"], "prod")
+        self.assertEqual(detail["diagnosis"]["score"], 65)
+        self.assertEqual(detail["report"], "report text")
+
+    def test_records_all_processing_events_in_order(self):
+        diagnosis_id = self.store.save_diagnosis(
+            {"timestamp": "2026-07-25T12:00:00", "namespace": "all", "score": 80, "status": "warning"},
+            "report",
+        )
+        run_id = self.store.create_run(diagnosis_id, "agent", "ai", "check pods")
+        self.store.append_event(run_id, "command", {"command": "kubectl get pods -A"})
+        self.store.append_event(run_id, "result", {"returncode": 0, "stdout": "ok"})
+        self.store.finish_run(run_id, "completed", "done")
+
+        detail = self.store.get_diagnosis(diagnosis_id)
+
+        self.assertEqual(len(detail["runs"]), 1)
+        self.assertEqual(detail["runs"][0]["status"], "completed")
+        self.assertEqual([event["type"] for event in detail["runs"][0]["events"]], ["command", "result"])
+
+    def test_lists_all_diagnoses_newest_first_and_filters_status(self):
+        first = {"timestamp": "2026-07-25T11:00:00", "namespace": "dev", "score": 100, "status": "healthy"}
+        second = {"timestamp": "2026-07-25T12:00:00", "namespace": "prod", "score": 20, "status": "critical"}
+        self.store.save_diagnosis(first, "first")
+        self.store.save_diagnosis(second, "second")
+
+        all_records = self.store.list_diagnoses()
+        critical = self.store.list_diagnoses(status="critical")
+
+        self.assertEqual([item["namespace"] for item in all_records], ["prod", "dev"])
+        self.assertEqual([item["namespace"] for item in critical], ["prod"])
+
+
+if __name__ == "__main__":
+    unittest.main()