feat(auto-repair): route ssh diagnostics through mcp gateway

apps/api/src/services/auto_repair_service.py

@@ -22,9 +22,10 @@ Phase 8: 自動化層實作
 - P0/P1 嚴重度 Incident 需要人工確認
 """
 
-from dataclasses import dataclass
 from collections.abc import Callable
-from typing import Protocol
+from dataclasses import dataclass
+import re
+from typing import Any, Protocol
 
 import structlog
 
@@ -81,6 +82,55 @@ class AutoRepairResult:
     execution_time_ms: int = 0
 
 
+@dataclass(frozen=True)
+class _SshMcpRoute:
+    """Route a legacy SSH playbook command to a governed MCP tool."""
+
+    tool_name: str
+    params: dict[str, Any]
+
+
+_SHORT_HOST_MAP: dict[str, str] = {
+    "110": "192.168.0.110",
+    "120": "192.168.0.120",
+    "121": "192.168.0.121",
+    "188": "192.168.0.188",
+}
+
+_SSH_DIAGNOSTIC_KEYWORDS = (
+    "ps aux",
+    "docker stats",
+    "docker inspect",
+    "docker logs",
+    "docker ps",
+    "docker top",
+    "df -h",
+    "du -",
+    "free -h",
+    "journalctl",
+    "systemctl show",
+    "tail ",
+    "top ",
+    "uptime",
+)
+
+_SSH_WRITE_KEYWORDS = (
+    "docker restart",
+    "docker start",
+    "docker stop",
+    "docker rm",
+    "docker prune",
+    "systemctl restart",
+    "systemctl stop",
+    "systemctl start",
+    "truncate ",
+    " rm ",
+    "rm -",
+    "certbot renew",
+    "bash ",
+)
+
+
 # =============================================================================
 # Auto Repair Service Interface
 # =============================================================================
@@ -918,6 +968,152 @@ class AutoRepairService:
             # 安全降級：檢查失敗 → 保守拒絕
             return False
 
+    def _route_legacy_ssh_command_to_mcp(
+        self,
+        incident: Incident,
+        command: str,
+    ) -> _SshMcpRoute | None:
+        """Map read-only legacy ``ssh {host} '...'`` steps to MCP Gateway.
+
+        YAML_RULE playbooks predate the URI executor and can contain compound
+        shell diagnostics.  Those commands should not bypass the newer
+        scheme-based HostRepairAgent or loosen its shell safety guard; read-only
+        diagnostics are instead routed to the governed SSH MCP provider.
+        """
+
+        raw_command = (command or "").strip()
+        lowered = raw_command.lower()
+        if not lowered.startswith("ssh "):
+            return None
+
+        if any(token in lowered for token in _SSH_WRITE_KEYWORDS):
+            return None
+
+        if not any(token in lowered for token in _SSH_DIAGNOSTIC_KEYWORDS):
+            return None
+
+        host = self._resolve_ssh_host_for_incident(incident, raw_command)
+        if not host:
+            return None
+
+        params: dict[str, Any] = {"host": host}
+        container_name = self._resolve_container_name_for_incident(incident, raw_command)
+        if container_name:
+            params["container_name"] = container_name
+
+        return _SshMcpRoute(tool_name="ssh_diagnose", params=params)
+
+    def _resolve_ssh_host_for_incident(self, incident: Incident, command: str) -> str:
+        """Resolve ``{host}``, short host labels, and exporter instance ports."""
+
+        labels = self._incident_labels(incident)
+        raw_host = ""
+        match = re.match(r"ssh\s+([^\s'\"]+)", command.strip(), flags=re.IGNORECASE)
+        if match:
+            raw_host = match.group(1)
+
+        if not raw_host or "{" in raw_host or "}" in raw_host:
+            raw_host = (
+                str(labels.get("host") or "")
+                or str(labels.get("instance") or "")
+                or str(labels.get("node") or "")
+                or str(labels.get("exported_instance") or "")
+            )
+
+        return self._normalize_ssh_host(raw_host)
+
+    @staticmethod
+    def _normalize_ssh_host(raw_host: str) -> str:
+        host = (raw_host or "").strip()
+        if host.startswith("ssh://"):
+            host = host.removeprefix("ssh://")
+        if "@" in host:
+            host = host.rsplit("@", 1)[1]
+        if host.startswith("[") and "]" in host:
+            host = host[1:host.index("]")]
+        if host.count(":") == 1:
+            maybe_host, maybe_port = host.rsplit(":", 1)
+            if maybe_port.isdigit():
+                host = maybe_host
+        if host in _SHORT_HOST_MAP:
+            return _SHORT_HOST_MAP[host]
+        match = re.fullmatch(r"(?:node-exporter-|host-)?(110|120|121|188)", host)
+        if match:
+            return _SHORT_HOST_MAP[match.group(1)]
+        return host
+
+    def _resolve_container_name_for_incident(
+        self,
+        incident: Incident,
+        command: str,
+    ) -> str:
+        labels = self._incident_labels(incident)
+        for key in ("container_name", "container", "name"):
+            value = str(labels.get(key) or "").strip()
+            if value and "{" not in value and "}" not in value:
+                return value
+
+        match = re.search(
+            r"docker\s+(?:stats\s+--no-stream|inspect|logs|top|ps\s+-a\s+--filter\s+name=)\s+([a-zA-Z0-9._-]+)",
+            command,
+        )
+        return match.group(1) if match else ""
+
+    @staticmethod
+    def _incident_labels(incident: Incident) -> dict[str, Any]:
+        for signal in incident.signals or []:
+            labels = getattr(signal, "labels", None)
+            if labels:
+                return labels
+        return {}
+
+    async def _execute_ssh_mcp_route(
+        self,
+        incident: Incident,
+        route: _SshMcpRoute,
+    ) -> str:
+        """Execute a routed SSH diagnostic through AwoooP MCP Gateway."""
+
+        try:
+            from src.db.base import get_db_context
+            from src.plugins.mcp.gateway import GatewayContext, McpGateway, McpGatewayError
+            from src.services.mcp_audit_context import with_mcp_audit_context
+
+            incident_id = incident.incident_id
+            params = with_mcp_audit_context(
+                route.params,
+                session_id=f"incident:{incident_id}:auto_repair_execute",
+                incident_id=incident_id,
+                flywheel_node="execute",
+                agent_role="auto_repair_executor",
+            )
+            async with get_db_context("awoooi") as db:
+                ctx = GatewayContext(
+                    project_id="awoooi",
+                    agent_id="auto_repair_executor",
+                    tool_name=route.tool_name,
+                    trace_id=incident_id,
+                    is_shadow=False,
+                    environment={"env": "prod"},
+                    required_scope="read",
+                )
+                result = await McpGateway(db).call(ctx, params)
+        except McpGatewayError as exc:
+            return f"FAILED: mcp:{route.tool_name} {exc.error_code}: {exc}"
+        except Exception as exc:
+            logger.warning(
+                "auto_repair_ssh_mcp_route_failed",
+                incident_id=incident.incident_id,
+                tool=route.tool_name,
+                error=str(exc),
+            )
+            return f"FAILED: mcp:{route.tool_name} {exc}"
+
+        if result.success:
+            preview = str(result.output or "")[:500]
+            return f"SUCCESS: mcp:{route.tool_name} {preview}".strip()
+        return f"FAILED: mcp:{route.tool_name} {result.error or 'execution failed'}"
+
     async def _execute_step(self, incident: Incident, step) -> str:
         """
         執行單一修復步驟
@@ -949,6 +1145,10 @@ class AutoRepairService:
 
         # 2026-04-06 Claude Code: Sprint 3 — repair_by_uri (URI scheme 路由)
         if step.action_type == ActionType.SSH_COMMAND:
+            route = self._route_legacy_ssh_command_to_mcp(incident, step.command)
+            if route is not None:
+                return await self._execute_ssh_mcp_route(incident, route)
+
             from src.services.host_repair_agent import HostRepairAgent
             agent = HostRepairAgent()
             approved = not getattr(step, "requires_approval", False)

apps/api/src/services/post_execution_verifier.py

@@ -34,6 +34,7 @@ MASTER §3.1 L6×D1
 from __future__ import annotations
 
 import asyncio
+import re
 from typing import TYPE_CHECKING, Any
 
 import structlog
@@ -236,6 +237,7 @@ class PostExecutionVerifier:
             "pod_name": labels.get("pod", labels.get("name", "")),
             "deployment": labels.get("deployment", ""),
             "host": labels.get("instance", "").split(":")[0] or labels.get("host", ""),
+            "query": _build_prometheus_query(alertname, labels),
         }
 
         async def _call_one(reg) -> tuple[str, Any]:
@@ -501,16 +503,95 @@ def _get_incident_id(incident: "Incident") -> str:
 
 def _get_alertname(incident: "Incident") -> str:
     if incident.signals:
-        return incident.signals[0].labels.get("alertname", "")
+        signal = incident.signals[0]
+        labels = signal.labels or {}
+        return labels.get("alertname", "") or getattr(signal, "alert_name", "")
     return ""
 
 
 def _get_labels(incident: "Incident") -> dict[str, Any]:
     if incident.signals:
-        return incident.signals[0].labels
+        return incident.signals[0].labels or {}
     return {}
 
 
+def _build_prometheus_query(alertname: str, labels: dict[str, Any]) -> str:
+    """Build a non-empty PromQL probe for post-execution metric sensors."""
+
+    alert = (alertname or "").lower()
+    namespace = _safe_label_value(str(labels.get("namespace") or "awoooi-prod"))
+    pod_name = _safe_label_value(str(labels.get("pod") or labels.get("name") or ""))
+    host = _safe_label_value(str(labels.get("host") or _short_instance(labels.get("instance")) or ""))
+    container = _safe_label_value(str(
+        labels.get("container_name")
+        or labels.get("container")
+        or labels.get("name")
+        or ""
+    ))
+
+    if alert.startswith("dockercontainer"):
+        selector = _selector({
+            "host": host,
+            "container_name": container,
+        })
+        if "memory" in alert:
+            return (
+                f"docker_container_memory_usage_bytes{selector} / "
+                f"docker_container_memory_limit_bytes{selector}"
+            )
+        if "restart" in alert:
+            return (
+                f"increase(docker_container_inspect_restart_count{selector}[15m]) "
+                f"or increase(docker_container_restart_count{selector}[15m])"
+            )
+        if "cpu" in alert:
+            return f"docker_container_cpu_cores{selector}"
+        return f"docker_container_info{selector}"
+
+    if any(key in alert for key in ("host", "node")):
+        instance_selector = _selector({"instance": str(labels.get("instance") or "")})
+        if "memory" in alert or "oom" in alert:
+            return f"node_memory_MemAvailable_bytes{instance_selector}"
+        if "disk" in alert or "storage" in alert:
+            return f"node_filesystem_avail_bytes{instance_selector}"
+        if "load" in alert or "cpu" in alert:
+            return f"node_load5{instance_selector}"
+        return f"up{instance_selector}"
+
+    pod_filter = f',pod=~"{pod_name}.*"' if pod_name else ""
+    if any(key in alert for key in ("memory", "mem", "oom")):
+        return (
+            f'avg(container_memory_working_set_bytes{{namespace="{namespace}"{pod_filter}}}) '
+            "/ 1048576"
+        )
+    if any(key in alert for key in ("cpu", "load", "throttl")):
+        return f'avg(rate(container_cpu_usage_seconds_total{{namespace="{namespace}"{pod_filter}}}[5m]))'
+    if any(key in alert for key in ("crash", "restart", "backoff")):
+        return f'sum(increase(kube_pod_container_status_restarts_total{{namespace="{namespace}"}}[15m]))'
+    if any(key in alert for key in ("http", "error", "5xx", "probe", "down", "unhealthy")):
+        return "1 - avg(probe_success)"
+    return f'up{{namespace="{namespace}"}}'
+
+
+def _selector(labels: dict[str, str]) -> str:
+    parts = [f'{key}="{value}"' for key, value in labels.items() if value]
+    return "{" + ",".join(parts) + "}" if parts else ""
+
+
+def _short_instance(instance: Any) -> str:
+    raw = str(instance or "")
+    if raw.count(":") == 1:
+        return raw.rsplit(":", 1)[0]
+    return raw
+
+
+def _safe_label_value(value: str) -> str:
+    cleaned = (value or "").strip()
+    if re.fullmatch(r"[a-zA-Z0-9_.:-]{1,128}", cleaned):
+        return cleaned
+    return ""
+
+
 async def _update_snapshot(
     snapshot: EvidenceSnapshot,
     post_state: dict[str, Any],