fix(ai): improve docker repair verification signals

apps/api/src/services/adr100_slo_status_service.py

@@ -574,6 +574,8 @@ def _classify_non_success_failure(row: dict[str, Any]) -> str:
         return "verifier_target_missing_pod"
     if not bool(row.get("auto_success")):
         return "auto_repair_execution_failed"
+    if "mcp:ssh_diagnose" in combined or "ssh_diagnose" in combined:
+        return "observe_only_playbook"
 
     result = str(row.get("verification_result") or "").lower()
     if result in {"failed", "timeout"}:
@@ -615,6 +617,13 @@ def _remediation_for_failure_class(failure_class: str) -> dict[str, str]:
             "owner": "solver_or_operator",
             "reason": "execution_failed_after_route_normalization",
         }
+    if failure_class == "observe_only_playbook":
+        return {
+            "status": "needs_playbook_ticket",
+            "action": "promote_diagnostic_to_repair_playbook",
+            "owner": "solver_or_operator",
+            "reason": "auto_repair_only_collected_evidence",
+        }
     if failure_class in {"verification_failed", "verification_timeout"}:
         return {
             "status": "manual_review",
@@ -639,6 +648,8 @@ def _next_step_for_failure_class(failure_class: str) -> str:
         return "map_verifier_target"
     if failure_class == "auto_repair_execution_failed":
         return "review_auto_repair_execution"
+    if failure_class == "observe_only_playbook":
+        return "author_mutating_repair_step"
     if failure_class in {"verification_failed", "verification_timeout"}:
         return "escalate_verification_failure"
     return "review_degraded_verification"

apps/api/src/services/auto_repair_service.py

@@ -149,6 +149,19 @@ _EXTERNAL_SITE_ALERTNAMES = {
 _K3S_NODE_ALERTNAMES = {"K3sNodeDown", "K3sVIPDown"}
 
 
+def _build_verification_action_taken(playbook_id: str, executed_steps: list[str]) -> str:
+    """Build a compact verifier action string with the real executed step shape."""
+    base = f"auto_repair:{playbook_id}"
+    if not executed_steps:
+        return base
+
+    joined = " | ".join(step.strip() for step in executed_steps if step.strip())
+    joined = re.sub(r"\s+", " ", joined)
+    if not joined:
+        return base
+    return f"{base} steps={joined[:900]}"
+
+
 # =============================================================================
 # Auto Repair Service Interface
 # =============================================================================
@@ -600,7 +613,10 @@ class AutoRepairService:
                 from src.services.post_execution_verifier import get_post_execution_verifier
                 from src.services.learning_service import get_learning_service
 
-                _action_taken = f"auto_repair:{playbook.playbook_id}"
+                _action_taken = _build_verification_action_taken(
+                    playbook.playbook_id,
+                    executed_steps,
+                )
                 _verifier = get_post_execution_verifier()
                 _learning = get_learning_service()
 

apps/api/src/services/mcp_tool_registry.py

@@ -175,10 +175,7 @@ class MCPToolRegistry:
         # workload/node locator prevents host alerts such as HostErrorLogFlood
         # from being misrouted into Kubernetes tools just because an upstream
         # bridge added namespace="infra".
-        has_k8s_locator = any(
-            labels.get(key)
-            for key in ("deployment", "pod", "node", "container")
-        )
+        has_k8s_locator = _has_k8s_locator(alertname, labels)
 
         # 依優先度排序後篩選
         sorted_tools = sorted(self._tools, key=lambda t: t.priority)
@@ -256,6 +253,23 @@ def _select_provider_balanced_tools(
     return selected
 
 
+def _has_k8s_locator(alertname: str, labels: dict[str, Any]) -> bool:
+    """Return true only when labels can identify a Kubernetes workload/node.
+
+    Docker exporters also use a generic ``container`` label.  Treating that
+    label as a pod locator made DockerContainer* alerts call Kubernetes pod
+    tools with an empty pod name, which polluted post-repair verification.
+    """
+    if any(labels.get(key) for key in ("deployment", "pod", "node")):
+        return True
+
+    alert = alertname or ""
+    if alert.startswith(("Kube", "Pod", "Deploy", "Node", "Velero", "ArgoCD")):
+        return bool(labels.get("namespace") and labels.get("container"))
+
+    return False
+
+
 # ─────────────────────────────────────────────────────────────────────────────
 # 工具自動分類（根據 tool name 推斷感官維度）
 # ─────────────────────────────────────────────────────────────────────────────

apps/api/src/services/post_execution_verifier.py

@@ -255,6 +255,8 @@ class PostExecutionVerifier:
             "pod_name": labels.get("pod", labels.get("name", "")),
             "deployment": labels.get("deployment", ""),
             "host": labels.get("instance", "").split(":")[0] or labels.get("host", ""),
+            "container_name": _extract_container_name(labels),
+            "filter_name": _extract_container_name(labels),
             "query": _build_prometheus_query(alertname, labels),
         }
 
@@ -495,7 +497,13 @@ def _assess_recovery(
         "'success': true",
         '"success": true',
     ]
-    if any(sig in post_str for sig in success_signals):
+    has_success_signal = any(sig in post_str for sig in success_signals)
+    if not has_success_signal and _docker_state_indicates_running(post_str):
+        has_success_signal = True
+
+    if has_success_signal:
+        if _is_observe_only_action(action_taken):
+            return "degraded"
         # 但如果 pre_state 已經是 running，可能是無效操作
         if pre_str and any(sig in pre_str for sig in success_signals):
             # 如果執行的是 restart，即使 pre/post 都 Running 也算 success
@@ -533,6 +541,53 @@ def _get_labels(incident: "Incident") -> dict[str, Any]:
     return {}
 
 
+def _extract_container_name(labels: dict[str, Any]) -> str:
+    """Resolve Docker container target labels for post-execution SSH sensors."""
+    for key in ("filter_name", "container_name", "container", "resource", "name"):
+        value = str(labels.get(key) or "").strip()
+        if value and "{" not in value and "}" not in value:
+            return value
+    return ""
+
+
+def _is_observe_only_action(action_taken: str) -> bool:
+    """Return true when the executed step collected evidence but did not mutate state."""
+    lowered = (action_taken or "").lower()
+    observe_tokens = (
+        "mcp:ssh_diagnose",
+        "ssh_diagnose",
+        "docker stats",
+        "ps aux",
+        "free -h",
+        "df -h",
+    )
+    mutation_tokens = (
+        "restart",
+        "delete",
+        "rollout",
+        "scale",
+        "patch",
+        "apply",
+        "prune",
+        "truncate",
+        "clear",
+    )
+    return any(token in lowered for token in observe_tokens) and not any(
+        token in lowered for token in mutation_tokens
+    )
+
+
+def _docker_state_indicates_running(post_str: str) -> bool:
+    """Recognize Docker-specific healthy/running output without matching uptime."""
+    if not any(token in post_str for token in ("docker ps", "docker inspect", "docker stats")):
+        return False
+    return bool(
+        re.search(r'\bup\s+\d', post_str)
+        or re.search(r'["\']status["\']\s*:\s*["\']running["\']', post_str)
+        or re.search(r'["\']running["\']\s*:\s*true', post_str)
+    )
+
+
 def _build_prometheus_query(alertname: str, labels: dict[str, Any]) -> str:
     """Build a non-empty PromQL probe for post-execution metric sensors."""