LuxyAI Flawless SRE 控制面

147ACTION_OPERATOR_GUIDANCE: dict[str, dict[str, str]] = {
148    "create_workload": {"label": "创建新 Workload", "when_to_use": "发布治理已校验完整 YAML，需要创建新的 Deployment、StatefulSet 或 DaemonSet。", "operator_note": "高风险；创建前必须检查命名空间、镜像、资源、探针、ServiceAccount 和回滚方式。"},
149    "patch_workload": {"label": "修改 Workload 配置", "when_to_use": "证据确认 Deployment/StatefulSet/DaemonSet 的镜像、探针、资源、副本、环境变量或安全上下文配置有误。", "operator_note": "只修改受控字段，执行前展示差异；可通过恢复原模板回滚。"},
150    "restart": {"label": "滚动重启组件", "when_to_use": "配置已正确但进程卡死、连接未刷新或需要让 Workload 重新拉起 Pod。", "operator_note": "不会修复错误配置；必须先确认有足够副本和 PDB 允许滚动。"},
151    "scale_out": {"label": "增加副本", "when_to_use": "CPU、并发或流量证据显示容量不足，且应用支持水平扩展。", "operator_note": "只在副本上限内扩容，并观察下游依赖和资源配额。"},
152    "recreate_pod": {"label": "重建异常 Pod", "when_to_use": "单个 controller 管理的 Pod 状态损坏，而 Workload 模板和其他副本正常。", "operator_note": "删除后由控制器按原模板重建；不适合模板级、存储级或全副本故障。"},
153    "patch_hpa": {"label": "调整 HPA 范围", "when_to_use": "HPA 上下限阻止合理扩缩容，且指标语义和数据源正常。", "operator_note": "不会修改指标算法，只调整 min/max replicas。"},
154    "expand_pvc": {"label": "扩容 PVC", "when_to_use": "卷使用率接近上限，StorageClass 明确支持扩容。", "operator_note": "通常不可逆；必须核对文件系统扩容支持和业务备份。"},
155    "create_pvc": {"label": "创建缺失 PVC", "when_to_use": "Workload 明确引用不存在的 PVC，且存储策略、容量和访问模式已确认。", "operator_note": "高风险；只能按批准的 StorageClass 或模板创建。"},
156    "create_pv": {"label": "创建静态 PV", "when_to_use": "动态供卷不可用，且存储管理员已提供批准的后端路径和绑定信息。", "operator_note": "高风险；严禁由 LLM 编造 NFS、LUN 或目录路径。"},
157    "patch_workload_volume": {"label": "修正卷引用", "when_to_use": "证据证明 Workload 引用了错误的 PVC、volume 或 mount 配置。", "operator_note": "高风险；需要完整存储链证据和原配置回滚点。"},
158    "patch_workload_runtime_security": {"label": "修复运行时权限", "when_to_use": "日志/事件证明挂载目录所有权导致容器无法写入，且 fsGroup 方案不足，需要受控 initContainer 或安全上下文调整。", "operator_note": "高风险；必须逐步确认，只允许有界的 chown/chmod/mkdir 权限修复命令。"},
159    "cordon_node": {"label": "隔离节点", "when_to_use": "节点明确存在压力、NotReady、硬件或运行时故障，需要停止新 Pod 调度。", "operator_note": "只停止新调度，不会自动迁移现有 Pod；后续通常配合受控驱逐。"},
160    "evict_pod": {"label": "受控驱逐 Pod", "when_to_use": "节点维护或隔离后需要迁移 Pod，并且 PDB 允许中断。", "operator_note": "通过 Eviction API 执行，遵守 PDB；高风险且需人工确认。"},
161    "uncordon_node": {"label": "恢复节点调度", "when_to_use": "节点 Ready、压力条件恢复、容量和系统组件验证通过。", "operator_note": "恢复前必须确认故障已解除，避免工作负载重新落到问题节点。"},
162    "rollback_workload": {"label": "回滚 Workload", "when_to_use": "最近发布与故障时间线一致，且存在已验证的稳定镜像或模板 revision。", "operator_note": "高风险；只回滚到真实观测过的稳定版本。"},
163    "patch_service": {"label": "修正 Service", "when_to_use": "selector、端口或 targetPort 与 Ready Pod 明确不匹配。", "operator_note": "高风险；错误修改会造成流量黑洞，必须先保存原 selector 和端口映射。"},
164    "patch_service_account": {"label": "修正 ServiceAccount", "when_to_use": "镜像拉取失败且证据确认缺少企业批准的 imagePullSecret 绑定。", "operator_note": "只允许添加批准 Secret 的引用，不读取或修改 Secret 明文。"},
165    "create_configmap": {"label": "恢复 ConfigMap", "when_to_use": "Workload 引用的 ConfigMap 缺失，且平台已有运维人员批准的配置模板。", "operator_note": "高风险；不能让 LLM 自行生成生产配置值。"},
166    "patch_pdb": {"label": "修正 PDB", "when_to_use": "PDB 与副本数配置形成发布或驱逐死锁，并且可用性证据充分。", "operator_note": "高风险；修改期间必须持续观察可用副本和业务 SLO。"},
167    "db_restart_instance": {"label": "重启数据库实例", "when_to_use": "连接、角色、备份和业务影响证据齐全，且故障明确需要实例级恢复。", "operator_note": "高风险；必须通过 DBA 执行器，不允许 LLM 直接执行 SQL 或系统命令。"},
168    "db_kill_session": {"label": "终止数据库会话", "when_to_use": "证据确认某个会话是锁等待、长事务或资源耗尽的直接根因。", "operator_note": "高风险；必须展示会话来源、SQL 摘要、阻塞链和业务影响。"},
169    "db_expand_storage": {"label": "扩容数据库存储", "when_to_use": "表空间或磁盘容量接近上限，备份和存储策略已确认。", "operator_note": "通常不可逆；需要 DBA/存储管理员确认。"},
170    "db_failover": {"label": "数据库主备切换", "when_to_use": "主库不可用或延迟/错误已达到 HA 预案阈值，且从库状态满足接管条件。", "operator_note": "极高风险；必须二次确认和业务窗口/回切方案。"},
171    "db_apply_parameter": {"label": "调整数据库参数", "when_to_use": "参数配置被证明确认为锁、连接、内存或复制问题的根因。", "operator_note": "必须保存原参数快照并限制修改范围。"},
172    "vm_restart_service": {"label": "重启主机服务", "when_to_use": "单个 OS 服务异常，系统资源和配置证据支持服务级恢复。", "operator_note": "通过受控执行器重启指定服务，不接受任意 shell。"},
173    "vm_reboot": {"label": "重启虚拟机", "when_to_use": "主机不可达、内核/驱动/系统级异常，且有业务冗余或维护窗口。", "operator_note": "高风险；需确认快照、控制台、HA 和影响范围。"},
174    "vm_expand_disk": {"label": "扩容虚拟机磁盘", "when_to_use": "文件系统容量风险明确，快照、磁盘和分区扩容步骤已确认。", "operator_note": "通常不可逆；必须确认文件系统扩容命令由批准执行器完成。"},
175    "vm_run_approved_script": {"label": "执行批准主机脚本", "when_to_use": "企业脚本目录已有对应脚本，证据和触发条件完全满足。", "operator_note": "Skill 只引用脚本 ID，不保存脚本正文。"},
176    "vm_snapshot": {"label": "创建虚拟机快照", "when_to_use": "高风险主机变更前需要可回退点。", "operator_note": "快照不是长期备份，需设置清理窗口。"},
177    "middleware_rebalance": {"label": "中间件重平衡", "when_to_use": "Kafka/队列/缓存分片存在倾斜、积压或节点异常。", "operator_note": "必须确认客户端影响和回滚策略。"},
178    "storage_expand_volume": {"label": "扩容企业存储卷", "when_to_use": "存储池容量、卷容量和业务使用率证据支持扩容。", "operator_note": "需要存储平台执行器和容量审批。"},
179    "infra_run_approved_action": {"label": "执行基础设施批准动作", "when_to_use": "资源类型已接入外部执行器，但尚未细分到专用动作。", "operator_note": "必须由执行器返回审计号、结果和回滚提示。"},
180}
181
182
183EXPERT_PROBES = {
184    "current_logs", "previous_logs", "events", "workload_spec", "pod_metrics", "node_pressure",
185    "node_conditions", "node_capacity", "system_pods", "service_endpoints", "dns", "network_policy",
186    "mesh_routes", "dependency_topology", "storage_chain", "node_storage", "csi_status",
187    "pod_security_context", "image_pull_secrets", "registry_connectivity", "scheduler_constraints",
188    "node_labels", "quota", "pvc_binding", "hpa", "traffic_baseline", "dependency_latency",
189    "cni_events", "recent_changes", "pdb_state", "certificate_chain", "webhook_status",
190    "config_ref_exists",
191}
192
193
194def _infer_expert_probe(text: str) -> str:
195    """把 AI 的自然语言步骤绑定到平台真实可执行的只读探针。"""
196    lowered = str(text or "").lower()
197    mappings = [
198        (("previous", "上一次", "退出日志", "laststate"), "previous_logs"),
199        (("event", "事件", "failedscheduling", "failedmount"), "events"),
200        (("pvc", "pv", "storageclass", "csi", "存储", "挂载"), "storage_chain"),
201        (("service", "endpoint", "selector", "流量入口"), "service_endpoints"),
202        (("networkpolicy", "网络策略"), "network_policy"),
203        (("dns", "域名解析"), "dns"),
204        (("node", "节点", "diskpressure", "memorypressure"), "node_conditions"),
205        (("pdb", "disruption"), "pdb_state"),
206        (("quota", "limitrange", "配额"), "quota"),
207        (("rollout", "revision", "近期变更", "镜像版本", "回滚"), "recent_changes"),
208        (("hpa", "扩缩容"), "hpa"),
209        (("cmdb", "依赖", "调用链", "kafka"), "dependency_topology"),
210        (("registry", "imagepull", "镜像仓库", "拉取"), "registry_connectivity"),
211        (("workload", "deployment", "statefulset", "daemonset", "模板", "配置"), "workload_spec"),
212        (("log", "日志", "错误栈"), "current_logs"),
213    ]
214    for terms, probe in mappings:
215        if any(term in lowered for term in terms):
216            return probe
217    return "workload_spec"
218
219
220def expert_steps_from_diagnosis(diagnosis: dict) -> list[dict[str, Any]]:
221    """将本次 LLM 诊断转成可执行、可审计且绑定真实探针的差异化步骤。"""
222    steps: list[dict[str, Any]] = []
223    for index, raw in enumerate((diagnosis.get("immediate_actions") or [])[:10], start=1):
224        if isinstance(raw, dict):
225            title = str(raw.get("title") or raw.get("step") or raw.get("action") or f"专家步骤 {index}").strip()
226            description = str(raw.get("description") or raw.get("detail") or raw.get("purpose") or title).strip()
227            requested_probe = str(raw.get("probe") or "").strip()
228            probe = requested_probe if requested_probe in EXPERT_PROBES else _infer_expert_probe(f"{title} {description}")
229            decision_rule = str(raw.get("decision_rule") or raw.get("decision") or "根据该步骤返回的真实证据决定下一分支。").strip()
230            on_match = str(raw.get("on_match") or raw.get("action_if_matched") or raw.get("next_if_true") or "进入证据支持的最小变更候选。").strip()
231            on_miss = str(raw.get("on_miss") or raw.get("action_if_not_matched") or raw.get("next_if_false") or "排除该分支并检查下一候选根因。").strip()
232            expected = raw.get("expected_evidence") or raw.get("evidence") or []
233        else:
234            text = str(raw or "").strip()
235            if not text:
236                continue
237            title, _, detail = text.partition("：")
238            if not detail:
239                title, _, detail = text.partition(":")
240            title = title.strip() or f"专家步骤 {index}"
241            description = detail.strip() or text
242            probe = _infer_expert_probe(text)
243            decision_rule = "根据该步骤返回的真实证据决定下一分支。"
244            on_match = "进入证据支持的最小变更候选。"
245            on_miss = "排除该分支并检查下一候选根因。"
246            expected = []
247        if not title:
248            continue
249        steps.append({
250            "id": probe, "sequence": index, "title": title[:120], "description": description[:500],
251            "probe": probe, "expected_evidence": expected if isinstance(expected, list) else [str(expected)],
252            "decision_rule": decision_rule[:500], "on_match": on_match[:500], "on_miss": on_miss[:500],
253            "source": "llm_evidence_expert", "status": "pending",
254        })
255    return steps

415def score_root_causes(alert: dict, diagnosis: dict, context: dict) -> list[dict[str, Any]]:
416    """Rank runbooks with a transparent weighted evidence score.
417
418    Alert/user text is weak evidence, status/Events are medium evidence, and
419    logs/last termination details are strong evidence. The sigmoid keeps the
420    reported confidence stable as the amount of evidence grows.
421    """
422    sources = [
423        ("alert", _flatten_text(alert), 0.7),
424        ("diagnosis", _flatten_text(diagnosis.get("root_cause"), diagnosis.get("signals")), 0.9),
425        ("pod_status", _flatten_text(context.get("pods"), context.get("pod")), 1.25),
426        ("events", _flatten_text(context.get("events")), 1.45),
427        ("logs", _flatten_text(context.get("logs"), context.get("previous_logs"), context.get("diagnostics")), 1.8),
428    ]
429    decisive_text = _flatten_text(
430        alert,
431        diagnosis,
432        context.get("events"),
433        context.get("storage"),
434        context.get("pod"),
435        context.get("pods"),
436        context.get("logs"),
437        context.get("previous_logs"),
438    )
439    storage_binding_proven = any(term in decisive_text for term in (
440        "unbound immediate persistentvolumeclaims",
441        "no persistent volumes available",
442        "persistentvolumeclaim is not bound",
443        "persistentvolumeclaim pending",
444    ))
445    oom_proven = any(term in decisive_text for term in (
446        "oomkilled",
447        "exit code 137",
448        "exitcode': 137",
449        '"exitcode": 137',
450        "last_terminated_reason': 'oomkilled",
451        '"last_terminated_reason": "oomkilled',
452    ))
453    image_arch_proven = any(term in decisive_text for term in (
454        "exec format error",
455        "no matching manifest for linux/amd64",
456        "no matching manifest for linux/arm64",
457        "image architecture",
458        "platform mismatch",
459        "cannot execute binary file",
460        "standard_init_linux.go",
461    ))
462    storage_permission_proven = any(term in decisive_text for term in (
463        "permission denied",
464        "operation not permitted",
465        "read-only file system",
466        "can't create directory",
467        "cannot create directory",
468        "mkdir:",
469    ))
470    log_unavailable_proven = any(term in decisive_text for term in (
471        "current_error",
472        "previous_error",
473        "previous log unavailable",
474        "container not found",
475        "pod does not exist",
476        "waiting to start",
477        "logs unavailable",
478    ))
479    ranked = []
480    for runbook_id, runbook in RUNBOOKS.items():
481        matches = []
482        raw_score = 0.0
483        for source, text, weight in sources:
484            hit = [term for term in runbook["terms"] if term in text]
485            if hit:
486                contribution = weight * min(2.2, 1.0 + 0.28 * (len(hit) - 1))
487                raw_score += contribution
488                matches.append({"source": source, "terms": hit[:4], "weight": weight})
489        # FailedScheduling is only the symptom here. An unbound PVC is a
490        # deterministic storage root cause and must outrank generic scheduling.
491        if runbook_id == "storage_mount" and storage_binding_proven:
492            raw_score += 3.2
493            matches.append({
494                "source": "kubernetes_storage_state",
495                "terms": ["pvc_unbound"],
496                "weight": 3.2,
497            })
498        if runbook_id == "oom" and oom_proven:
499            raw_score += 2.8
500            matches.append({
501                "source": "container_last_state",
502                "terms": ["oomkilled_or_exit_137"],
503                "weight": 2.8,
504            })
505        if runbook_id == "crash_unknown" and oom_proven:
506            raw_score = max(0.0, raw_score - 1.4)
507            matches.append({
508                "source": "container_last_state",
509                "terms": ["demoted_by_oom_evidence"],
510                "weight": -1.4,
511            })
512        if runbook_id == "image_architecture" and image_arch_proven:
513            raw_score += 2.8
514            matches.append({
515                "source": "runtime_platform_state",
516                "terms": ["exec_format_or_manifest_platform_mismatch"],
517                "weight": 2.8,
518            })
519        if runbook_id == "crash_unknown" and image_arch_proven:
520            raw_score = max(0.0, raw_score - 1.2)
521            matches.append({
522                "source": "runtime_platform_state",
523                "terms": ["demoted_by_architecture_evidence"],
524                "weight": -1.2,
525            })
526        if runbook_id == "storage_permission" and storage_permission_proven:
527            raw_score += 2.6
528            matches.append({
529                "source": "container_logs",
530                "terms": ["write_permission_denied"],
531                "weight": 2.6,
532            })
533        if runbook_id == "crash_unknown" and storage_permission_proven:
534            raw_score = max(0.0, raw_score - 1.2)
535            matches.append({
536                "source": "container_logs",
537                "terms": ["demoted_by_storage_permission_evidence"],
538                "weight": -1.2,
539            })
540        if runbook_id == "crash_unknown" and log_unavailable_proven and not storage_binding_proven:
541            raw_score += 1.4
542            matches.append({
543                "source": "log_probe",
544                "terms": ["logs_unavailable_need_diagnostic_recreate"],
545                "weight": 1.4,
546            })
547        if matches:
548            confidence = 1.0 / (1.0 + math.exp(-(raw_score - 1.65)))
549            ranked.append({
550                "id": runbook_id,
551                "title": runbook["title"],
552                "score": round(raw_score, 3),
553                "confidence": round(min(0.99, confidence), 3),
554                "matched_evidence": matches,
555                "diagnostics": list(runbook["diagnostics"]),
556                "success_criteria": list(runbook["success"]),
557            })
558    return sorted(ranked, key=lambda item: (-item["score"], item["id"]))

961def build_remediation_plan(alert: dict, diagnosis: dict, context: dict) -> dict[str, Any]:
962    hypotheses = score_root_causes(alert, diagnosis, context)
963    namespace, workload_type, workload_name, pod_name, pod = _target(alert, diagnosis, context)
964    primary = hypotheses[0] if hypotheses else None
965    runbook_id = primary["id"] if primary else "unknown"
966    container = _first_container(pod)
967    container_name = container.get("name", "")
968    changes: list[dict[str, Any]] = []
969    patchable = workload_name and str(workload_type).lower() in {"deployment", "statefulset", "daemonset"}
970    confidence = float((primary or {}).get("confidence") or 0.0)
971
972    def workload_patch(patch: dict, reason: str):
973        changes.append({
974            "type": "patch_workload", "namespace": namespace, "workload_type": workload_type,
975            "workload_name": workload_name, "patch": patch, "reason": reason,
976            "runbook_id": runbook_id, **ACTION_CATALOG["patch_workload"],
977        })
978
979    if runbook_id == "storage_mount":
980        issue = _first_storage_issue(context)
981        issue_text = _flatten_text(issue)
982        pvc_name = str(issue.get("pvc") or issue.get("pvc_name") or issue.get("claim") or "").strip()
983        missing_pvc = bool(
984            pvc_name
985            and (
986                issue.get("missing") is True
987                or "not found" in issue_text
988                or "404" in issue_text
989                or "does not exist" in issue_text
990            )
991        )
992        phase = str(issue.get("pvc_phase") or issue.get("phase") or "").lower()
993        if missing_pvc:
994            changes.append({
995                "type": "create_pvc", "namespace": namespace, "pvc_name": pvc_name,
996                "manifest": _pvc_manifest(namespace, pvc_name, issue),
997                "reason": "Kubernetes API 已确认 Pod 引用的 PVC 不存在；创建受存储策略约束的 PVC，并等待动态供卷后验证 Pod。",
998                "runbook_id": runbook_id, **ACTION_CATALOG["create_pvc"],
999            })
1000        elif pvc_name and phase in {"pending", "lost"}:
1001            pv_manifest = _static_pv_manifest(namespace, pvc_name, issue)
1002            if pv_manifest:
1003                changes.append({
1004                    "type": "create_pv", "namespace": namespace, "pvc_name": pvc_name,
1005                    "manifest": pv_manifest,
1006                    "reason": "Kubernetes API 已确认 PVC 未绑定，且平台配置了批准的静态存储模板；创建预绑定 PV 并验证 PVC=Bound。",
1007                    "runbook_id": runbook_id, **ACTION_CATALOG["create_pv"],
1008                })
1009
1010    if primary and confidence >= 0.62 and patchable:
1011        if runbook_id == "oom" and container_name:
1012            resources = container.get("resources") or {}
1013            requests, limits = resources.get("requests") or {}, resources.get("limits") or {}
1014            workload_patch({"spec": {"template": {"spec": {"containers": [{
1015                **_container_patch_base(container_name, container),
1016                "resources": {
1017                    "requests": {"cpu": requests.get("cpu") or "100m", "memory": requests.get("memory") or "256Mi"},
1018                    "limits": {"cpu": limits.get("cpu") or "1", "memory": _memory_growth(limits.get("memory"))},
1019                },
1020            }]}}}}, "OOM 证据达到执行阈值：提高内存上限并保留 CPU 约束，滚动后验证 OOM 是否消失。")
1021        elif runbook_id == "probe" and container_name:
1022            workload_patch({"spec": {"template": {"spec": {"containers": [{
1023                **_container_patch_base(container_name, container), **_startup_probe_patch(container)
1024            }]}}}}, "探针/慢启动证据达到执行阈值：增加 startupProbe 容错窗口并验证 Endpoint Ready。")
1025        elif runbook_id == "storage_permission":
1026            fs_group = _security_group_from_pod(pod)
1027            workload_patch({"spec": {"template": {"spec": {"securityContext": {
1028                "fsGroup": fs_group, "fsGroupChangePolicy": "OnRootMismatch"
1029            }}}}}, f"卷写入权限证据达到执行阈值：按容器运行用户/组选择 fsGroup={fs_group}，并在 rollout 后复查挂载与写入错误。")
1030        elif runbook_id == "storage_mount" and not changes:
1031            issue = _first_storage_issue(context)
1032            issue_text = _flatten_text(issue)
1033            pvc_name = str(issue.get("pvc") or issue.get("pvc_name") or issue.get("claim") or "").strip()
1034            if pvc_name and ("not found" in issue_text or "404" in issue_text or issue.get("missing") is True):
1035                changes.append({
1036                    "type": "create_pvc", "namespace": namespace, "pvc_name": pvc_name,
1037                    "manifest": _pvc_manifest(namespace, pvc_name, issue),
1038                    "reason": "Pod 引用了不存在的 PVC；先创建受策略约束的 PVC，再等待存储插件动态供卷并验证 Pod 挂载。",
1039                    "runbook_id": runbook_id, **ACTION_CATALOG["create_pvc"],
1040                })
1041            elif pvc_name and str(issue.get("pvc_phase") or issue.get("phase") or "").lower() in {"pending", "lost"}:
1042                pv_manifest = _static_pv_manifest(namespace, pvc_name, issue)
1043                if pv_manifest:
1044                    changes.append({
1045                        "type": "create_pv", "namespace": namespace, "pvc_name": pvc_name,
1046                        "manifest": pv_manifest,
1047                        "reason": "PVC 已存在但未绑定 PV，且平台已配置静态 PV 模板；创建预绑定 PV 让 PVC 收敛到 Bound。",
1048                        "runbook_id": runbook_id, **ACTION_CATALOG["create_pv"],

110    def test_oom_plan_grows_memory_without_shell(self):
111        plan = build_remediation_plan(
112            self.alert,
113            {"root_cause": "OOMKilled out of memory"},
114            {
115                "pod": self.pod,
116                "pods": [self.pod],
117                "events": {"events": [{"reason": "OOMKilled", "message": "container exceeded memory limit; exit code 137"}]},
118            },
119        )
120        change = plan["changes"][0]
121        self.assertEqual(change["type"], "patch_workload")
122        container = change["patch"]["spec"]["template"]["spec"]["containers"][0]
123        self.assertEqual(container["image"], "docker.io/library/api:current")
124        self.assertEqual(container["resources"]["limits"]["memory"], "768Mi")
125        self.assertNotIn("command", change)
126
127    def test_oom_last_state_outranks_generic_crashloop(self):
128        pod = {
129            **self.pod,
130            "containers": [{
131                **self.pod["containers"][0],
132                "reason": "CrashLoopBackOff",
133                "state_detail": {
134                    "reason": "CrashLoopBackOff",
135                    "last_terminated_reason": "OOMKilled",
136                    "exit_code": 137,
137                },
138            }],
139        }
140        plan = build_remediation_plan(
141            self.alert,
142            {"root_cause": "container back-off restarting failed"},
143            {
144                "pod": pod,
145                "pods": [pod],
146                "events": {"events": [{"reason": "BackOff", "message": "Back-off restarting failed container"}]},
147            },
148        )
149        self.assertEqual(plan["runbook_id"], "oom")
150        self.assertEqual(plan["changes"][0]["type"], "patch_workload")
151
152    def test_exec_format_error_outranks_generic_crashloop(self):
153        old_mapping = os.environ.get("AUTO_OPS_IMAGE_REPLACEMENTS_JSON")
154        os.environ["AUTO_OPS_IMAGE_REPLACEMENTS_JSON"] = json.dumps({
155            "registry.example.com/api:bad-arch": {"amd64": "registry.example.com/api:stable-amd64"}
156        })
157        try:
158            pod = {
159                **self.pod,
160                "containers": [{
161                    **self.pod["containers"][0],
162                    "image": "registry.example.com/api:bad-arch",
163                    "reason": "CrashLoopBackOff",
164                    "state_detail": {"reason": "CrashLoopBackOff", "last_terminated_reason": "Error", "exit_code": 1},
165                }],
166            }
167            plan = build_remediation_plan(
168                self.alert,
169                {"root_cause": "container back-off restarting failed"},
170                {
171                    "pod": pod,
172                    "pods": [pod],
173                    "node": {"labels": {"kubernetes.io/arch": "amd64"}},
174                    "logs": {"app": "exec /app: exec format error"},
175                    "events": {"events": [{"reason": "BackOff", "message": "Back-off restarting failed container"}]},
176                },
177            )
178            self.assertEqual(plan["runbook_id"], "image_architecture")
179            self.assertEqual(plan["changes"][0]["type"], "rollback_workload")
180        finally:
181            if old_mapping is None:
182                os.environ.pop("AUTO_OPS_IMAGE_REPLACEMENTS_JSON", None)
183            else:
184                os.environ["AUTO_OPS_IMAGE_REPLACEMENTS_JSON"] = old_mapping
185
186    def test_high_risk_action_requires_second_confirmation(self):
187        result = asyncio.run(server._execute_change(
188            {"type": "cordon_node", "node_name": "worker-1", "risk": "high"},
189            {"cluster_id": "local", "namespace": "default", "high_risk_confirmed": False},
190        ))
191        self.assertEqual(result["status"], "blocked")
192        self.assertTrue(result["result"]["requires_high_risk_confirmation"])
193
194    def test_permission_guidance_is_action_specific(self):
195        guidance = server._permission_guidance(
196            {"error": "HTTP Error 403: Forbidden"},
197            {"cluster": "nonprod", "namespace": "k8s-agent", "source": "rancher"},
198            {"type": "recreate_pod"},
199        )
200        self.assertEqual(guidance["minimal_verbs"], ["get", "delete"])
201        self.assertEqual(guidance["minimal_resources"], ["pods"])
202        self.assertIn("Rancher Token", guidance["summary"])
203