diff --git a/README.md b/README.md index 54a5079..c97ebec 100644 --- a/README.md +++ b/README.md @@ -13,6 +13,12 @@ - **健康评分**：100 分制量化集群健康度（严重问题 -15 分，警告 -5 分） - **AI 智能分析**：接入 OpenAI 兼容接口，自动分析诊断报告 - 根因定位、影响评估、修复命令、优先级排序、预防建议 +- **AI 自主诊断（双执行模式）**：启动前可选择命令执行方式 + - **手动执行命令**：Agent 只生成命令，用户复制到自己的终端执行，服务端不会运行 + - **AI 执行命令**：自动执行 `get`、`describe`、`logs`、`top` 等只读 kubectl 命令 + - 实时展示思考步骤、执行命令和真实输出 + - 即使选择 AI 执行，`apply`、`delete`、`patch`、`scale`、`rollout`、`exec` 等命令仍必须人工确认 + - 禁止 shell、管道、重定向和多命令拼接 - **AI 追问对话**：基于诊断上下文多轮追问，深入排查问题 - **Web 仪表盘**：Vue3 + Element Plus，实时展示诊断结果 @@ -55,6 +61,8 @@ 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}` | | POST | `/api/chat` | AI 多轮对话 `{"messages":[...]}` | | GET | `/api/report` | 获取诊断报告文本 | diff --git a/backend/agent_tools.py b/backend/agent_tools.py new file mode 100644 index 0000000..48aef4a --- /dev/null +++ b/backend/agent_tools.py @@ -0,0 +1,109 @@ +"""AI Agent 可用的 kubectl 命令工具和安全策略。""" +import asyncio +import shlex +from dataclasses import dataclass + +from backend.diagnosis import KUBECTL + + +READ_ONLY_VERBS = { + "api-resources", "api-versions", "auth", "cluster-info", "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 = {";", "|", "||", "&&", ">", ">>", "<", "<<", "`", "$"} + + +@dataclass(frozen=True) +class CommandDecision: + mode: str + requires_approval: bool + verb: str + args: list[str] + + +def parse_kubectl_command(command: str) -> list[str]: + """仅解析单条 kubectl 命令，不允许 shell、重定向或命令组合。""" + if not isinstance(command, str) or not command.strip(): + raise ValueError("命令不能为空") + if any(token in command for token in FORBIDDEN_TOKENS) or "\n" in command or "\r" in command: + raise ValueError("不允许 shell 操作符、重定向或多条命令") + try: + parts = shlex.split(command) + except ValueError as exc: + raise ValueError(f"命令格式错误: {exc}") from exc + 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 操作类型") + + +def classify_kubectl_command(command: str) -> CommandDecision: + args = parse_kubectl_command(command) + verb = _find_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) + + +async def execute_kubectl(command: str, timeout: int = 30) -> dict: + """无 shell 执行 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)", + } diff --git a/backend/ai_agent.py b/backend/ai_agent.py index 9107004..2f1b879 100644 --- a/backend/ai_agent.py +++ b/backend/ai_agent.py @@ -1,7 +1,9 @@ """AI Agent 分析模块 - 调用 OpenAI 兼容接口分析诊断结果 (支持流式输出)""" import json +import uuid import httpx from backend.config import AI_API_BASE, AI_API_KEY, AI_MODEL +from backend.agent_tools import classify_kubectl_command, execute_kubectl SYSTEM_PROMPT = """你是一位资深的 Kubernetes 运维专家和 SRE 工程师。你的任务是分析 K8S 集群诊断报告，找出根本原因并给出修复建议。 @@ -20,6 +22,143 @@ _HEADERS = { } +AGENT_SYSTEM_PROMPT = """你是 Kubernetes 自主诊断 Agent。你可以根据诊断报告决定执行 kubectl 命令获取更多证据，并在证据充分后给出结论。 + +安全规则： +1. 只允许提出单条 kubectl 命令，不得使用 shell、管道、重定向或命令连接符。 +2. get、describe、logs、top 等只读命令可以自动执行。 +3. apply、delete、patch、scale、rollout、exec 等修改或交互命令必须暂停并请求人工批准。 +4. 不要猜测命令输出；必须依据实际工具结果继续判断。 +5. 最多执行有限轮次，优先使用 namespace 和资源名缩小范围。 + +每轮必须只返回一个 JSON 对象，不要使用 Markdown 代码块： +- 继续调查：{"action":"command","command":"kubectl ...","reason":"为什么执行"} +- 完成分析：{"action":"final","analysis":"中文 Markdown 分析结论"} +""" + +_PENDING_APPROVALS: dict[str, dict] = {} + + +def _parse_agent_action(content: str) -> dict: + try: + action = json.loads(content.strip()) + except json.JSONDecodeError as exc: + raise ValueError("AI 未返回合法 JSON 动作") from exc + 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") + 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 = 6, + execution_mode: str = "ai", +): + """自主诊断循环：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}, + ] + + for step in range(1, max_steps + 1): + yield {"type": "thinking", "step": step, "message": f"AI 正在规划第 {step} 轮诊断"} + try: + content = await _complete_chat(messages) + action = _parse_agent_action(content) + except Exception as exc: + yield {"type": "error", "message": str(exc)} + return + + if action["action"] == "final": + yield {"type": "final", "content": action["analysis"], "model": AI_MODEL} + return + + command = action["command"] + reason = action.get("reason", "AI 需要更多集群证据") + try: + decision = classify_kubectl_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": "请在终端手动执行命令，并根据结果继续排查", + } + return + + if decision.requires_approval: + approval_id = uuid.uuid4().hex + _PENDING_APPROVALS[approval_id] = { + "command": command, + "reason": reason, + "created_by": AI_MODEL, + } + yield { + "type": "approval_required", "approval_id": approval_id, + "command": command, "reason": reason, + } + return + + result = await execute_kubectl(command) + yield {"type": "result", **result} + messages.extend([ + {"role": "assistant", "content": content}, + {"role": "user", "content": "命令实际执行结果：\n" + json.dumps(result, ensure_ascii=False)}, + ]) + + yield { + "type": "final", + "content": f"已达到自主诊断最大轮次（{max_steps}），请根据上方命令输出继续人工排查。", + "model": AI_MODEL, + } + + +def take_pending_approval(approval_id: str) -> dict | None: + return _PENDING_APPROVALS.pop(approval_id, None) + + async def _stream_chat(messages: list[dict]): """通用流式请求 - yield 文本 chunk""" # connect=10s 快速失败; read=90s 防止 AI 服务无响应时前端无限卡死 diff --git a/backend/main.py b/backend/main.py index c06f3e0..4e49ee9 100644 --- a/backend/main.py +++ b/backend/main.py @@ -22,7 +22,9 @@ 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, take_pending_approval, ) +from backend.agent_tools import execute_kubectl app = FastAPI(title="K8S 智能诊断平台", version="1.0.0") @@ -64,6 +66,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()} @@ -250,6 +257,48 @@ async def analyze_stream(question: str = ""): ) +@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: + _last_diagnosis = await run_full_diagnosis() + _last_report = build_diagnosis_report(_last_diagnosis) + + async def event_generator(): + yield f"data: {json.dumps({'type': 'start', 'model': AI_MODEL, 'mode': execution_mode}, ensure_ascii=False)}\n\n" + try: + async for event in run_diagnostic_agent( + _last_report, + question=question, + execution_mode=execution_mode, + ): + yield f"data: {json.dumps(event, ensure_ascii=False)}\n\n" + yield f"data: {json.dumps({'type': 'done'}, ensure_ascii=False)}\n\n" + except Exception as exc: + yield f"data: {json.dumps({'type': 'error', 'message': f'自主诊断失败: {exc}'}, 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: + return {"approved": False, "command": pending["command"], "message": "用户已拒绝执行"} + result = await execute_kubectl(pending["command"]) + return {"approved": True, "reason": pending["reason"], **result} + + @app.post("/api/chat") async def chat(req: ChatRequest): """AI 多轮对话 (非流式，兼容旧接口)""" diff --git a/frontend/src/App.vue b/frontend/src/App.vue index 2ed61fd..7ef6d39 100644 --- a/frontend/src/App.vue +++ b/frontend/src/App.vue @@ -14,9 +14,16 @@ {{ diagnosing ? '诊断中...' : '一键诊断' }} - + {{ analyzing ? 'AI 分析中...' : 'AI 分析' }} + + 手动执行命令 + AI 执行命令 + + + {{ agentRunning ? 'Agent 诊断中...' : '自主诊断' }} + @@ -257,6 +264,62 @@
+ + + + +
+
+
+ + + + + {{ agentEventTitle(event) }} + + {{ event.mode === 'read' ? '只读' : '需审批' }} + +
+

{{ event.message }}

+

目的：{{ event.reason }}

+ 
{{ event.command }}
+ 
{{ event.stdout || event.stderr }}
+
+
+ 复制命令 + 请在你的终端中执行，Agent 不会自动运行该命令。 +
+
+ 确认执行 + 拒绝 +
+ +
+
+ + Agent 正在分析命令结果并规划下一步... +
+
+ + @@ -288,6 +351,10 @@ 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('') @@ -462,6 +529,89 @@ async function runAnalysis() { } } +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: '命令已被安全策略拒绝', + 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 }), + }) + const result = await resp.json() + if (!resp.ok) throw new Error(result.detail || `HTTP ${resp.status}`) + event.resolved = true + event.approvalSuccess = approved && result.returncode === 0 + if (!approved) { + event.approvalResult = '已拒绝执行该命令' + } else { + event.approvalResult = result.returncode === 0 ? '命令执行成功' : `命令执行失败：${result.stderr || result.returncode}` + agentEvents.value.push({ type: 'result', ...result }) + } + ElMessage[approved && result.returncode === 0 ? 'success' : 'info'](event.approvalResult) + } catch (e) { + ElMessage.error('审批处理失败: ' + e.message) + } finally { + event.executing = false + } +} + async function sendChat() { const q = chatInput.value.trim() if (!q || chatLoading.value) return @@ -611,6 +761,26 @@ 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; } + .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; } diff --git a/tests/test_agent_api.py b/tests/test_agent_api.py new file mode 100644 index 0000000..ebc14f6 --- /dev/null +++ b/tests/test_agent_api.py @@ -0,0 +1,48 @@ +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_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 run(): + with patch.object(main, "take_pending_approval", return_value=approval), patch.object(main, "execute_kubectl", fake_execute): + return await main.execute_approved_agent_command(main.AgentApprovalRequest(approval_id="abc", approved=True)) + + result = asyncio.run(run()) + self.assertEqual(result["returncode"], 0) + self.assertEqual(result["stdout"], "scaled") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_agent_tools.py b/tests/test_agent_tools.py new file mode 100644 index 0000000..032b0db --- /dev/null +++ b/tests/test_agent_tools.py @@ -0,0 +1,38 @@ +import unittest + +from backend.agent_tools import classify_kubectl_command, parse_kubectl_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): + for command in ( + "kubectl get pods; rm -rf /", + "kubectl get pods | sh", + "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) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_diagnostic_agent.py b/tests/test_diagnostic_agent.py new file mode 100644 index 0000000..c8bd020 --- /dev/null +++ b/tests/test_diagnostic_agent.py @@ -0,0 +1,97 @@ +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"}', + ]) + + 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_kubectl", 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], ["thinking", "command", "result", "thinking", "final"]) + self.assertEqual(events[1]["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_kubectl") 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"}' + + async def run(): + with patch("backend.ai_agent._complete_chat", fake_completion), patch("backend.ai_agent.execute_kubectl") as execute: + events = await collect_events(run_diagnostic_agent("report", max_steps=3, execution_mode="ai")) + return events, execute + + events, execute = asyncio.run(run()) + execute.assert_not_called() + 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_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"]) + + +if __name__ == "__main__": + unittest.main()