fix(aiops): escalate blocked auto repair

apps/api/src/api/v1/webhooks.py

@@ -111,6 +111,26 @@ def _should_bypass_alertmanager_llm(
     )
 
 
+def _should_use_alertmanager_rule_first(
+    rule_response: dict | None,
+    alert_category: str,
+) -> bool:
+    """Host 類告警命中權威 YAML 規則時，避免再被 LLM 污染成 K8s 動作。"""
+
+    if not rule_response or alert_category != "host_resource":
+        return False
+    if rule_response.get("rule_id", "") in ("generic_fallback", ""):
+        return False
+
+    suggested_action = str(rule_response.get("suggested_action", "")).upper()
+    command = str(rule_response.get("kubectl_command", "")).strip().lower()
+    return (
+        suggested_action == "NO_ACTION"
+        or suggested_action.startswith("SSH_")
+        or command.startswith("ssh ")
+    )
+
+
 async def _escalate_auto_repair_unavailable(
     *,
     incident_id: str,
@@ -1334,7 +1354,7 @@ async def _process_new_alert_background(
         openclaw = get_openclaw()
 
         rule_response = match_rule(alert_context)
-        should_bypass_llm = _should_bypass_alertmanager_llm(rule_response, alert_category)
+        should_bypass_llm = _should_use_alertmanager_rule_first(rule_response, alert_category)
 
         if should_bypass_llm:
             logger.info(
@@ -1343,7 +1363,7 @@ async def _process_new_alert_background(
                 alertname=alertname,
                 rule_id=rule_response.get("rule_id", ""),
                 alert_category=alert_category,
-                reason="host_resource YAML NO_ACTION 規則命中，跳過 LLM 產生人工排查卡片",
+                reason="host_resource YAML 權威規則命中，跳過 LLM 避免產生錯誤 K8s 動作",
             )
             risk_mapping = {
                 "low": RiskLevel.LOW,
@@ -1517,6 +1537,37 @@ async def _process_new_alert_background(
                     error=str(_meta_err),
                 )
 
+            _is_heartbeat = is_heartbeat_alertname(alertname)
+            if can_auto_repair and not _is_heartbeat:
+                await _try_auto_repair_background(
+                    incident_id=incident_id,
+                    approval_id=str(approval.id),
+                    alert_type=alert_type,
+                    target_resource=target_resource,
+                    namespace=namespace,
+                )
+            elif not can_auto_repair and not _is_heartbeat:
+                from src.repositories.alert_operation_log_repository import get_alert_operation_log_repository
+                _op_log_rule = get_alert_operation_log_repository()
+                await _op_log_rule.append(
+                    "GUARDRAIL_BLOCKED",
+                    incident_id=incident_id,
+                    approval_id=str(approval.id),
+                    actor="prometheus-rule",
+                    action_detail=f"Prometheus rule 設定 auto_repair=false，轉入緊急介入: {alertname}",
+                    success=False,
+                    context={"alertname": alertname, "auto_repair_flag": False},
+                )
+                await _escalate_auto_repair_unavailable(
+                    incident_id=incident_id,
+                    approval_id=str(approval.id),
+                    alert_type=alert_type,
+                    target_resource=target_resource,
+                    namespace=namespace,
+                    failure_reason="Prometheus rule auto_repair=false，未進入自動修復評估",
+                    attempted_actions="rule_engine -> rule_first -> emergency_intervention",
+                )
+
             await _push_to_telegram_background(
                 approval_id=str(approval.id),
                 risk_level=rule_risk.value,
@@ -1710,6 +1761,16 @@ async def _process_new_alert_background(
                     success=False,
                     context={"alertname": alertname, "auto_repair_flag": False},
                 )
+                if not _is_heartbeat:
+                    await _escalate_auto_repair_unavailable(
+                        incident_id=incident_id,
+                        approval_id=str(approval.id),
+                        alert_type=alert_type,
+                        target_resource=target_resource,
+                        namespace=namespace,
+                        failure_reason="Prometheus rule auto_repair=false，未進入自動修復評估",
+                        attempted_actions="llm_analysis -> guardrail:auto_repair_false -> emergency_intervention",
+                    )
 
             await _push_to_telegram_background(
                 approval_id=str(approval.id),

apps/api/src/services/decision_manager.py

@@ -74,6 +74,61 @@ def _phase2_fallback_reason(package: Any) -> str | None:
     return None
 
 
+def _should_escalate_auto_approve_rejection(reason: Any) -> bool:
+    """Return True for manual gates that mean the automation path went blind."""
+
+    value = getattr(reason, "value", str(reason or "")).lower()
+    return value in {
+        "no_playbook",
+        "no_executable_action",
+        "low_trust",
+    }
+
+
+async def _escalate_decision_auto_repair_unavailable(
+    *,
+    incident: Incident,
+    token: Any,
+    failure_reason: str,
+    attempted_actions: str,
+) -> None:
+    """Send the same emergency channel used by auto-repair when decision gates block."""
+
+    try:
+        from src.services.emergency_escalation_service import (
+            escalate_auto_repair_unavailable,
+        )
+
+        labels = incident.signals[0].labels if incident.signals else {}
+        alertname = labels.get("alertname") or getattr(incident, "alertname", "") or "DecisionAutoRepairBlocked"
+        target = (
+            (incident.affected_services or [None])[0]
+            or labels.get("deployment")
+            or labels.get("pod")
+            or labels.get("service")
+            or labels.get("component")
+            or labels.get("instance")
+            or "unknown"
+        )
+        namespace = labels.get("namespace") or "awoooi-prod"
+
+        await escalate_auto_repair_unavailable(
+            incident_id=incident.incident_id,
+            approval_id=getattr(token, "proposal_id", None) or getattr(token, "token", ""),
+            alert_type=str(alertname),
+            target_resource=str(target),
+            namespace=str(namespace),
+            failure_reason=failure_reason,
+            attempted_actions=attempted_actions,
+        )
+    except Exception as exc:
+        logger.warning(
+            "decision_auto_repair_escalation_failed",
+            incident_id=getattr(incident, "incident_id", ""),
+            error=str(exc),
+        )
+
+
 # =============================================================================
 # Phase 31 (ADR-067 2026-04-10): Log 異常摘要 — NemoTron deepseek-r1:14b
 # =============================================================================
@@ -1634,6 +1689,20 @@ class DecisionManager:
                 # ADR-071: 注入 decision_state + auto_executed 供 classify_notification 使用
                 token.proposal_data["decision_state"] = token.state.value if token.state else ""
                 token.proposal_data["auto_executed"] = False
+                token.proposal_data["auto_approve_reason"] = auto_decision.reason.value
+                token.proposal_data["blocked_reason"] = auto_decision.reason_detail
+                if _should_escalate_auto_approve_rejection(auto_decision.reason):
+                    _fire_and_forget(
+                        _escalate_decision_auto_repair_unavailable(
+                            incident=incident,
+                            token=token,
+                            failure_reason=auto_decision.reason_detail,
+                            attempted_actions=(
+                                "decision_manager -> auto_approve_policy:"
+                                f"{auto_decision.reason.value} -> emergency_intervention"
+                            ),
+                        )
+                    )
                 _fire_and_forget(
                     _push_decision_to_telegram(incident, token.proposal_data)
                 )
@@ -1925,7 +1994,7 @@ class DecisionManager:
         # alert_category = "infrastructure" 表示 Docker 告警，非 kubectl action → SSH
         # P1-1 fix 2026-04-12: 必須在 kubectl safety guard 之前 routing，否則 docker 指令被 _action_safe=False 攔截
         _alert_category = getattr(incident, "alert_category", None) or ""
-        if _alert_category == "infrastructure" and action and not action.startswith("kubectl"):
+        if _alert_category in {"infrastructure", "host_resource"} and action and not action.startswith("kubectl"):
             await self._ssh_execute(incident, token, action, _target)
             return
 
@@ -1945,6 +2014,14 @@ class DecisionManager:
             token.proposal_data["mcp_all_failed"] = True
             token.proposal_data["blocked_reason"] = "host_resource 告警禁止 K8s kubectl，請人工排查主機"
             await self._save_token(token)
+            _fire_and_forget(
+                _escalate_decision_auto_repair_unavailable(
+                    incident=incident,
+                    token=token,
+                    failure_reason=token.proposal_data["blocked_reason"],
+                    attempted_actions="auto_execute -> host_resource_k8s_block -> emergency_intervention",
+                )
+            )
             _fire_and_forget(_push_decision_to_telegram(incident, token.proposal_data))
             return
 
@@ -1973,6 +2050,17 @@ class DecisionManager:
             token.proposal_data["auto_executed"] = False
             token.proposal_data["mcp_all_failed"] = True  # 標記讓 classify_notification → TYPE-4
             await self._save_token(token)
+            _fire_and_forget(
+                _escalate_decision_auto_repair_unavailable(
+                    incident=incident,
+                    token=token,
+                    failure_reason=(
+                        "Auto-execute safety guard blocked action: "
+                        f"{_action_parse.reason}"
+                    ),
+                    attempted_actions="auto_execute -> action_parser_guard -> emergency_intervention",
+                )
+            )
             _fire_and_forget(
                 _push_decision_to_telegram(incident, token.proposal_data)
             )
@@ -1999,6 +2087,16 @@ class DecisionManager:
                 token.proposal_data["auto_executed"] = False
                 token.proposal_data["mcp_all_failed"] = True
                 await self._save_token(token)
+                _fire_and_forget(
+                    _escalate_decision_auto_repair_unavailable(
+                        incident=incident,
+                        token=token,
+                        failure_reason=(
+                            f"K8s target '{_target}' not found in namespace '{_ns}'"
+                        ),
+                        attempted_actions="auto_execute -> k8s_existence_check -> emergency_intervention",
+                    )
+                )
                 _fire_and_forget(
                     _push_decision_to_telegram(incident, token.proposal_data)
                 )
@@ -3159,6 +3257,17 @@ class DecisionManager:
         # 2026-04-27 Claude Sonnet 4.6: 加入主機診斷路徑
         # 根因：只支援 docker/systemctl restart，主機告警 ssh {host} '...' 格式全降級人工
         # 修復：識別 ssh_diagnose 模式，路由到 ssh_get_top_processes / ssh_get_disk_usage
+        _labels = incident.signals[0].labels if incident.signals else {}
+        if not target or target == "unknown":
+            target = (
+                _labels.get("container")
+                or _labels.get("name")
+                or _labels.get("component")
+                or _labels.get("service")
+                or _labels.get("job")
+                or "unknown"
+            )
+
         _action_lower = action.lower().strip()
         if _action_lower.startswith("docker restart"):
             _tool = "docker_restart"
@@ -3166,6 +3275,14 @@ class DecisionManager:
         elif _action_lower.startswith("systemctl restart"):
             _tool = "service_restart"
             _service = target
+        elif _action_lower.startswith("ssh ") and "systemctl restart" in _action_lower:
+            import re as _ssh_re
+            _tool = "service_restart"
+            _svc_match = _ssh_re.search(r"systemctl\s+restart\s+([a-z0-9_.-]+)", _action_lower)
+            _service = _svc_match.group(1) if _svc_match else target
+        elif _action_lower.startswith("ssh ") and "docker restart" in _action_lower and target != "unknown":
+            _tool = "docker_restart"
+            _container = target
         elif _action_lower.startswith("ssh ") and ("ps aux" in _action_lower or "top" in _action_lower or "free" in _action_lower or "df -h" in _action_lower or "uptime" in _action_lower):
             # 主機診斷指令：自動收集 CPU/記憶體/磁碟，不修改任何狀態
             _tool = "ssh_diagnose"

apps/api/src/services/emergency_escalation_service.py

@@ -41,6 +41,7 @@ async def escalate_auto_repair_unavailable(
         from src.repositories.alert_operation_log_repository import (
             get_alert_operation_log_repository,
         )
+        from src.services.approval_db import get_timeline_service
         from src.services.telegram_gateway import get_telegram_gateway
 
         await get_telegram_gateway().send_escalation_card(
@@ -69,6 +70,26 @@ async def escalate_auto_repair_unavailable(
                 "attempted_actions": attempted_actions,
             },
         )
+        try:
+            await get_timeline_service().add_event(
+                event_type="agent",
+                status="warning",
+                title="AI emergency intervention requested",
+                description=(
+                    f"{failure_reason} | attempted={attempted_actions}"
+                )[:500],
+                actor="auto_repair",
+                actor_role="emergency_intervention",
+                approval_id=approval_id,
+                incident_id=incident_id,
+            )
+        except Exception as timeline_exc:
+            logger.warning(
+                "auto_repair_emergency_timeline_failed",
+                incident_id=incident_id,
+                approval_id=approval_id,
+                error=str(timeline_exc),
+            )
         logger.warning(
             "auto_repair_emergency_escalated",
             incident_id=incident_id,