fix(api): 產生 MCP PlayBook 修復候選

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

@@ -81,6 +81,7 @@ from src.services.incident_service import (
 # Phase 5: OpenClaw AI Engine
 from src.services.openclaw import get_openclaw
 from src.services.playbook_match_resolver import resolve_playbook_id_for_alert
+from src.services.repair_candidate_service import get_repair_candidate_service
 from src.services.security_interceptor import check_webhook_nonce  # P0-06: nonce dedup via Service 層
 from src.services.signal_producer import SignalData, get_signal_producer
 
@@ -2239,64 +2240,18 @@ async def _process_new_alert_background(
             record_alert_chain_success("alertmanager")
 
         else:
-            # LLM 失敗 - 使用預設值
-            # 2026-04-27 Claude Sonnet 4.6: shadow-run Step1 — 補 metadata kwarg，讓 extra_metadata 可觀測
+            # LLM 失敗時，不再把 NO_ACTION 當成終點。
+            # 先用預配置 approval id 建立 incident，讓後續 MCP evidence、
+            # PlayBook trust、approval 與 Telegram 都指向同一條真相鏈。
+            preallocated_approval_id = str(uuid.uuid4())
             _matched_playbook_id_cs4 = await resolve_playbook_id_for_alert(
                 rule_id=str(rule_response.get("rule_id", "")),
                 alertname=alertname,
                 affected_services=[target_resource] if target_resource else [],
                 severity="medium",
             )
-            _approval_metadata_cs4 = {
-                "source": "fallback",
-                "confidence_score": None,
-                "is_rule_based": False,
-                "playbook_id": _matched_playbook_id_cs4,
-            }
-            fallback_create = ApprovalRequestCreate(
-                action="NO_ACTION - REPAIR_CANDIDATE_MISSING: LLM 分析失敗，尚未產生可安全執行的修復指令",
-                description=f"[LLM Failed] {message}",
-                risk_level=RiskLevel.LOW,
-                blast_radius=BlastRadius(
-                    affected_pods=1,
-                    estimated_downtime="unknown",
-                    related_services=[],
-                    data_impact=DataImpact.NONE,
-                ),
-                dry_run_checks=[],
-                requested_by="OpenClaw (fallback)",
-                metadata=_approval_metadata_cs4,
-                matched_playbook_id=_matched_playbook_id_cs4,
-            )
-
-            approval = await service.create_approval_with_fingerprint(
-                request=fallback_create,
-                fingerprint=fingerprint,
-            )
-
-            # 2026-04-27 Claude Sonnet 4.6: shadow-run Step2 — 只記 log，不改執行決策
-            try:
-                _shadow_proposal_cs4 = {
-                    "risk_level": "low",
-                    "confidence": 0.0,
-                    "action": fallback_create.action,
-                    "kubectl_command": "",
-                    "is_rule_based": False,
-                    "source": "fallback",
-                }
-                _shadow_result_cs4 = get_auto_approve_policy().evaluate(_shadow_proposal_cs4)
-                logger.info(
-                    "shadow_auto_approve_result",
-                    approval_id=str(approval.id),
-                    should_auto=_shadow_result_cs4.should_auto_approve,
-                    reason=_shadow_result_cs4.reason.value,
-                    source="fallback",
-                )
-            except Exception as _shadow_err_cs4:
-                logger.warning("shadow_auto_approve_failed", error=str(_shadow_err_cs4))
-
             fallback_incident_id = await create_incident_for_approval(
-                approval_id=str(approval.id),
+                approval_id=preallocated_approval_id,
                 risk_level="medium",
                 target_resource=target_resource,
                 namespace=namespace,
@@ -2309,6 +2264,106 @@ async def _process_new_alert_background(
                 alert_category=alert_category,
             )
 
+            fallback_action_text = (
+                "NO_ACTION - REPAIR_CANDIDATE_MISSING: "
+                "LLM 分析失敗，MCP evidence / PlayBook trust 尚未產生可安全執行的修復指令"
+            )
+            repair_candidate_result = await get_repair_candidate_service().build_from_incident_id(
+                incident_id=fallback_incident_id,
+                alertname=alertname,
+                target_resource=target_resource,
+                namespace=namespace,
+                message=message,
+                fallback_action=fallback_action_text,
+                matched_playbook_id=_matched_playbook_id_cs4,
+                rule_id=str(rule_response.get("rule_id", "")),
+                severity="medium",
+            )
+
+            _approval_metadata_cs4 = {
+                "source": "llm_fallback_mcp_playbook_candidate",
+                "confidence_score": None,
+                "is_rule_based": False,
+                "playbook_id": _matched_playbook_id_cs4,
+                "preallocated_approval_id": preallocated_approval_id,
+            }
+            _approval_metadata_cs4.update(repair_candidate_result.metadata)
+            _approval_metadata_cs4["preallocated_approval_id"] = preallocated_approval_id
+
+            candidate_confidence = 0.0
+            if repair_candidate_result.candidate_found and repair_candidate_result.approval_request:
+                evidence = repair_candidate_result.evidence
+                playbook = repair_candidate_result.playbook
+                evidence_ratio = 0.0
+                if evidence and evidence.sensors_attempted:
+                    evidence_ratio = evidence.sensors_succeeded / max(evidence.sensors_attempted, 1)
+                trust_score = float(playbook.trust_score) if playbook else 0.0
+                candidate_confidence = min(0.82, 0.45 + evidence_ratio * 0.2 + trust_score * 0.2)
+                fallback_create = repair_candidate_result.approval_request.model_copy(
+                    update={
+                        "incident_id": fallback_incident_id,
+                        "metadata": _approval_metadata_cs4,
+                    }
+                )
+                telegram_root_cause = (
+                    "LLM fallback 後已由 MCP evidence + PlayBook trust 產生修復候選；"
+                    "等待人工批准後進入 execution / verifier / KM 回寫。"
+                )
+                primary_responsibility = "OPENCLAW_PLAYBOOK"
+            else:
+                blockers = repair_candidate_result.blockers or ["repair_candidate_missing"]
+                blocker_text = ", ".join(blockers)
+                fallback_create = ApprovalRequestCreate(
+                    action=f"NO_ACTION - REPAIR_CANDIDATE_MISSING: {blocker_text}",
+                    description=f"[LLM Failed] {message}\n修復候選阻擋：{blocker_text}",
+                    risk_level=RiskLevel.LOW,
+                    blast_radius=BlastRadius(
+                        affected_pods=1,
+                        estimated_downtime="unknown",
+                        related_services=[target_resource] if target_resource else [],
+                        data_impact=DataImpact.NONE,
+                    ),
+                    dry_run_checks=[
+                        DryRunCheck(
+                            name="MCP/PlayBook candidate gate",
+                            passed=False,
+                            message=blocker_text[:240],
+                        )
+                    ],
+                    requested_by="OpenClaw (fallback candidate gate)",
+                    incident_id=fallback_incident_id,
+                    metadata=_approval_metadata_cs4,
+                    matched_playbook_id=_matched_playbook_id_cs4,
+                )
+                telegram_root_cause = f"LLM fallback 後未產生修復候選；阻擋：{blocker_text}"
+                primary_responsibility = "HUMAN"
+
+            approval = await service.create_approval_with_fingerprint(
+                request=fallback_create,
+                fingerprint=fingerprint,
+            )
+
+            # 2026-04-27 Claude Sonnet 4.6: shadow-run Step2 — 只記 log，不改執行決策
+            try:
+                _shadow_proposal_cs4 = {
+                    "risk_level": fallback_create.risk_level.value,
+                    "confidence": candidate_confidence,
+                    "action": fallback_create.action,
+                    "kubectl_command": fallback_create.action if fallback_create.action.startswith("kubectl") else "",
+                    "is_rule_based": False,
+                    "source": _approval_metadata_cs4.get("source", "fallback"),
+                }
+                _shadow_result_cs4 = get_auto_approve_policy().evaluate(_shadow_proposal_cs4)
+                logger.info(
+                    "shadow_auto_approve_result",
+                    approval_id=str(approval.id),
+                    should_auto=_shadow_result_cs4.should_auto_approve,
+                    reason=_shadow_result_cs4.reason.value,
+                    source="fallback_candidate",
+                )
+            except Exception as _shadow_err_cs4:
+                logger.warning("shadow_auto_approve_failed", error=str(_shadow_err_cs4))
+
             try:
                 await service.update_incident_id(approval.id, fallback_incident_id)
                 approval.incident_id = fallback_incident_id
@@ -2339,46 +2394,61 @@ async def _process_new_alert_background(
             )
 
             _is_heartbeat = is_heartbeat_alertname(alertname)
-            if can_auto_repair and not _is_heartbeat:
-                await _try_auto_repair_background(
-                    incident_id=fallback_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:
+            if not _is_heartbeat:
                 from src.repositories.alert_operation_log_repository import get_alert_operation_log_repository
                 _op_log_fallback = get_alert_operation_log_repository()
-                await _op_log_fallback.append(
-                    "GUARDRAIL_BLOCKED",
-                    incident_id=fallback_incident_id,
-                    approval_id=str(approval.id),
-                    actor="prometheus-rule",
-                    action_detail=f"Prometheus rule 設定 auto_repair=false，fallback 轉人工: {alertname}",
-                    success=False,
-                    context={"alertname": alertname, "auto_repair_flag": False},
-                )
-                await _escalate_auto_repair_unavailable(
-                    incident_id=fallback_incident_id,
-                    approval_id=str(approval.id),
-                    alert_type=alert_type,
-                    target_resource=target_resource,
-                    namespace=namespace,
-                    failure_reason="Prometheus rule auto_repair=false，fallback 未進入自動修復評估",
-                    attempted_actions="llm_fallback -> guardrail:auto_repair_false -> emergency_intervention",
-                )
+                if repair_candidate_result.candidate_found:
+                    await _op_log_fallback.append(
+                        "REPAIR_CANDIDATE_READY",
+                        incident_id=fallback_incident_id,
+                        approval_id=str(approval.id),
+                        actor="openclaw-repair-candidate",
+                        action_detail=f"MCP evidence + PlayBook trust 產生候選，等待批准: {fallback_create.action[:220]}",
+                        success=True,
+                        context={
+                            "alertname": alertname,
+                            "auto_repair_flag": bool(can_auto_repair),
+                            "playbook_id": fallback_create.matched_playbook_id,
+                            "candidate_status": "ready_for_approval",
+                        },
+                    )
+                else:
+                    await _op_log_fallback.append(
+                        "REPAIR_CANDIDATE_BLOCKED",
+                        incident_id=fallback_incident_id,
+                        approval_id=str(approval.id),
+                        actor="openclaw-repair-candidate",
+                        action_detail=f"fallback 未產生候選: {fallback_create.action[:220]}",
+                        success=False,
+                        context={
+                            "alertname": alertname,
+                            "auto_repair_flag": bool(can_auto_repair),
+                            "blockers": repair_candidate_result.blockers,
+                        },
+                    )
+                    await _escalate_auto_repair_unavailable(
+                        incident_id=fallback_incident_id,
+                        approval_id=str(approval.id),
+                        alert_type=alert_type,
+                        target_resource=target_resource,
+                        namespace=namespace,
+                        failure_reason=telegram_root_cause,
+                        attempted_actions=(
+                            "llm_fallback -> mcp_evidence -> playbook_trust -> "
+                            f"candidate_blocked:{','.join(repair_candidate_result.blockers or ['unknown'])}"
+                        ),
+                    )
 
             await _push_to_telegram_background(
                 approval_id=str(approval.id),
-                risk_level="low",
+                risk_level=fallback_create.risk_level.value,
                 resource_name=target_resource,
-                root_cause=message,
+                root_cause=telegram_root_cause,
                 suggested_action=fallback_create.action,
                 estimated_downtime="unknown",
                 hit_count=1,
-                primary_responsibility="HUMAN",
-                confidence=0.0,
+                primary_responsibility=primary_responsibility,
+                confidence=candidate_confidence,
                 namespace=namespace,
                 incident_id=fallback_incident_id,
                 notification_type=notification_type,

apps/api/src/services/approval_db.py

@@ -130,6 +130,11 @@ def approval_request_to_record_data(
     """
     Convert ApprovalRequestCreate to dict for ApprovalRecord creation
     """
+    metadata = dict(request.metadata or {})
+    preallocated_approval_id = str(metadata.pop("preallocated_approval_id", "") or "").strip()
+    if preallocated_approval_id:
+        UUID(preallocated_approval_id)
+
     blast_radius_dict = None
     if request.blast_radius:
         blast_radius_dict = {
@@ -151,7 +156,7 @@ def approval_request_to_record_data(
             })
 
     now = datetime.now(UTC)
-    return {
+    record_data = {
         "action": request.action,
         "description": request.description,
         "status": ApprovalStatus.APPROVED if risk_level == RiskLevel.LOW else ApprovalStatus.PENDING,
@@ -163,7 +168,7 @@ def approval_request_to_record_data(
         "dry_run_checks": dry_run_checks_list,
         "requested_by": request.requested_by,
         "expires_at": request.expires_at,
-        "extra_metadata": request.metadata,
+        "extra_metadata": metadata or None,
         "resolved_at": now if risk_level == RiskLevel.LOW else None,
         # 戰略 B: 告警風暴收斂
         "fingerprint": fingerprint,
@@ -177,6 +182,9 @@ def approval_request_to_record_data(
         # 原本缺失 → Playbook 人工審核後 trust score 永遠不更新（學習閉環斷鏈）
         "matched_playbook_id": getattr(request, "matched_playbook_id", None),
     }
+    if preallocated_approval_id:
+        record_data["id"] = preallocated_approval_id
+    return record_data
 
 
 def _record_value(value: Any) -> str:

apps/api/src/services/repair_candidate_service.py

@@ -0,0 +1,427 @@
+"""Repair candidate generation from MCP evidence and PlayBook trust.
+
+This service is intentionally candidate-only. It never executes a repair; it
+builds an ApprovalRequestCreate only when evidence, PlayBook status, trust and
+command safety gates are all explicit enough for a human approval flow.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any
+
+import structlog
+
+from src.models.approval import (
+    ApprovalRequestCreate,
+    BlastRadius,
+    DataImpact,
+    DryRunCheck,
+)
+from src.models.approval import RiskLevel as ApprovalRiskLevel
+from src.models.incident import Incident
+from src.models.playbook import (
+    ActionType,
+    Playbook,
+    PlaybookStatus,
+)
+from src.models.playbook import RiskLevel as PlaybookRiskLevel
+from src.repositories.playbook_repository import get_playbook_repository
+from src.services.action_parser import ActionKind, parse_kubectl_action
+from src.services.auto_repair_service import AutoRepairService
+from src.services.evidence_snapshot import EvidenceSnapshot
+from src.services.incident_service import get_incident_service
+from src.services.playbook_match_resolver import resolve_playbook_id_for_alert
+from src.services.pre_decision_investigator import get_pre_decision_investigator
+
+logger = structlog.get_logger(__name__)
+
+MIN_REPAIR_CANDIDATE_TRUST = 0.30
+
+
+@dataclass
+class RepairCandidateResult:
+    """Candidate generation result used by webhook and Telegram handoff paths."""
+
+    approval_request: ApprovalRequestCreate | None = None
+    evidence: EvidenceSnapshot | None = None
+    playbook: Playbook | None = None
+    blockers: list[str] = field(default_factory=list)
+    metadata: dict[str, Any] = field(default_factory=dict)
+
+    @property
+    def candidate_found(self) -> bool:
+        return self.approval_request is not None
+
+
+class RepairCandidateService:
+    """Build a repair approval candidate without executing it."""
+
+    def __init__(
+        self,
+        *,
+        incident_service: Any | None = None,
+        investigator: Any | None = None,
+        playbook_repository: Any | None = None,
+        auto_repair_service: AutoRepairService | None = None,
+    ) -> None:
+        self._incident_service = incident_service or get_incident_service()
+        self._investigator = investigator or get_pre_decision_investigator()
+        self._playbook_repository = playbook_repository or get_playbook_repository()
+        self._auto_repair = auto_repair_service or AutoRepairService()
+
+    async def build_from_incident_id(
+        self,
+        *,
+        incident_id: str,
+        alertname: str,
+        target_resource: str,
+        namespace: str,
+        message: str,
+        fallback_action: str,
+        matched_playbook_id: str | None = None,
+        rule_id: str | None = None,
+        severity: str | None = None,
+    ) -> RepairCandidateResult:
+        """Load an incident and produce a repair candidate or explicit blockers."""
+
+        incident = await self._incident_service.get_from_working_memory(incident_id)
+        if incident is None:
+            return RepairCandidateResult(
+                blockers=["incident_not_found"],
+                metadata={"repair_candidate_status": "blocked", "blockers": ["incident_not_found"]},
+            )
+
+        return await self.build_from_incident(
+            incident=incident,
+            alertname=alertname,
+            target_resource=target_resource,
+            namespace=namespace,
+            message=message,
+            fallback_action=fallback_action,
+            matched_playbook_id=matched_playbook_id,
+            rule_id=rule_id,
+            severity=severity,
+        )
+
+    async def build_from_incident(
+        self,
+        *,
+        incident: Incident,
+        alertname: str,
+        target_resource: str,
+        namespace: str,
+        message: str,
+        fallback_action: str,
+        matched_playbook_id: str | None = None,
+        rule_id: str | None = None,
+        severity: str | None = None,
+    ) -> RepairCandidateResult:
+        evidence = await self._collect_evidence(incident)
+        metadata = self._base_metadata(evidence=evidence)
+        blockers: list[str] = []
+
+        if not evidence or evidence.sensors_succeeded <= 0:
+            blockers.append("mcp_evidence_missing")
+
+        playbook_id = matched_playbook_id or await resolve_playbook_id_for_alert(
+            rule_id=rule_id,
+            alertname=alertname,
+            affected_services=[target_resource] if target_resource else [],
+            severity=severity,
+        )
+        if not playbook_id:
+            blockers.append("playbook_not_matched")
+            return self._blocked_result(
+                blockers=blockers,
+                metadata=metadata,
+                evidence=evidence,
+                fallback_action=fallback_action,
+            )
+
+        playbook = await self._playbook_repository.get_by_id(playbook_id)
+        if not playbook:
+            blockers.append("playbook_not_found")
+            return self._blocked_result(
+                blockers=blockers,
+                metadata=metadata,
+                evidence=evidence,
+                fallback_action=fallback_action,
+            )
+
+        metadata["playbook_trust"] = {
+            "playbook_id": playbook.playbook_id,
+            "name": playbook.name,
+            "status": playbook.status.value,
+            "trust_score": float(playbook.trust_score),
+        }
+
+        if playbook.status != PlaybookStatus.APPROVED:
+            blockers.append("playbook_not_approved")
+        if float(playbook.trust_score) < MIN_REPAIR_CANDIDATE_TRUST:
+            blockers.append("playbook_trust_below_gate")
+
+        step, step_blockers = self._select_executable_step(incident, playbook)
+        blockers.extend(step_blockers)
+        if blockers or step is None:
+            return self._blocked_result(
+                blockers=blockers,
+                metadata=metadata,
+                evidence=evidence,
+                playbook=playbook,
+                fallback_action=fallback_action,
+            )
+
+        metadata["repair_candidate"] = {
+            "source": "mcp_evidence_playbook_trust",
+            "playbook_id": playbook.playbook_id,
+            "playbook_name": playbook.name,
+            "step_number": step.step_number,
+            "command": step.command,
+            "action_type": step.action_type.value,
+            "risk_level": step.risk_level.value,
+            "expected_result": step.expected_result,
+            "rollback_command": step.rollback_command,
+            "requires_approval": True,
+        }
+        metadata["verifier_plan"] = self._build_verifier_plan(
+            command=step.command,
+            namespace=namespace,
+            target_resource=target_resource,
+            evidence=evidence,
+        )
+        metadata["repair_candidate_status"] = "candidate_ready_for_approval"
+        metadata["fallback_replaced"] = True
+
+        approval_request = ApprovalRequestCreate(
+            action=step.command,
+            description=self._build_description(
+                message=message,
+                playbook=playbook,
+                step_command=step.command,
+                evidence=evidence,
+                verifier_plan=metadata["verifier_plan"],
+            ),
+            risk_level=self._approval_risk(step.risk_level),
+            blast_radius=BlastRadius(
+                affected_pods=1,
+                estimated_downtime=f"{playbook.estimated_duration_minutes} min",
+                related_services=[target_resource] if target_resource else [],
+                data_impact=DataImpact.WRITE,
+            ),
+            dry_run_checks=self._build_dry_run_checks(
+                evidence=evidence,
+                playbook=playbook,
+                command=step.command,
+            ),
+            requested_by="OpenClaw (MCP evidence + PlayBook trust)",
+            incident_id=incident.incident_id,
+            metadata=metadata,
+            matched_playbook_id=playbook.playbook_id,
+        )
+
+        logger.info(
+            "repair_candidate_generated",
+            incident_id=incident.incident_id,
+            playbook_id=playbook.playbook_id,
+            sensors_succeeded=evidence.sensors_succeeded if evidence else 0,
+            command=step.command[:160],
+        )
+        return RepairCandidateResult(
+            approval_request=approval_request,
+            evidence=evidence,
+            playbook=playbook,
+            metadata=metadata,
+        )
+
+    async def _collect_evidence(self, incident: Incident) -> EvidenceSnapshot | None:
+        try:
+            return await self._investigator.investigate(incident)
+        except Exception as exc:
+            logger.warning(
+                "repair_candidate_evidence_collect_failed",
+                incident_id=incident.incident_id,
+                error=str(exc),
+            )
+            return None
+
+    def _select_executable_step(
+        self,
+        incident: Incident,
+        playbook: Playbook,
+    ) -> tuple[Any | None, list[str]]:
+        blockers: list[str] = []
+        if not playbook.repair_steps:
+            return None, ["playbook_has_no_repair_steps"]
+
+        rejected_readonly = False
+        rejected_unsafe = False
+        for step in playbook.repair_steps:
+            command = (step.command or "").strip()
+            if not command:
+                continue
+            if step.action_type == ActionType.MANUAL:
+                rejected_readonly = True
+                continue
+            if step.action_type == ActionType.KUBECTL or command.startswith("kubectl"):
+                parsed = parse_kubectl_action(command)
+                if not parsed.ok:
+                    rejected_unsafe = True
+                    continue
+                if parsed.kind == ActionKind.READONLY:
+                    rejected_readonly = True
+                    continue
+                return step, []
+            if step.action_type == ActionType.SSH_COMMAND or command.startswith("ssh "):
+                if self._auto_repair.preview_write_ssh_mcp_route(incident, command):
+                    return step, []
+                rejected_unsafe = True
+
+        if rejected_unsafe:
+            blockers.append("playbook_command_not_safely_routable")
+        if rejected_readonly:
+            blockers.append("playbook_observe_only")
+        if not blockers:
+            blockers.append("playbook_has_no_executable_step")
+        return None, blockers
+
+    def _base_metadata(self, *, evidence: EvidenceSnapshot | None) -> dict[str, Any]:
+        if not evidence:
+            return {
+                "source": "mcp_playbook_repair_candidate",
+                "mcp_evidence": {"status": "missing"},
+            }
+        return {
+            "source": "mcp_playbook_repair_candidate",
+            "mcp_evidence": {
+                "status": "collected" if evidence.sensors_succeeded > 0 else "degraded",
+                "snapshot_id": evidence.snapshot_id,
+                "sensors_attempted": evidence.sensors_attempted,
+                "sensors_succeeded": evidence.sensors_succeeded,
+                "mcp_health": evidence.mcp_health,
+                "summary_excerpt": (evidence.evidence_summary or "")[:800],
+            },
+        }
+
+    def _blocked_result(
+        self,
+        *,
+        blockers: list[str],
+        metadata: dict[str, Any],
+        fallback_action: str,
+        evidence: EvidenceSnapshot | None = None,
+        playbook: Playbook | None = None,
+    ) -> RepairCandidateResult:
+        metadata["repair_candidate_status"] = "blocked"
+        metadata["repair_candidate_blockers"] = list(dict.fromkeys(blockers))
+        metadata["fallback_action"] = fallback_action
+        return RepairCandidateResult(
+            evidence=evidence,
+            playbook=playbook,
+            blockers=metadata["repair_candidate_blockers"],
+            metadata=metadata,
+        )
+
+    def _build_description(
+        self,
+        *,
+        message: str,
+        playbook: Playbook,
+        step_command: str,
+        evidence: EvidenceSnapshot | None,
+        verifier_plan: list[str],
+    ) -> str:
+        evidence_line = "MCP evidence missing"
+        if evidence:
+            evidence_line = (
+                f"MCP evidence {evidence.sensors_succeeded}/{evidence.sensors_attempted}; "
+                f"snapshot={evidence.snapshot_id}"
+            )
+        verifier_text = "; ".join(verifier_plan)
+        return (
+            "LLM fallback 後由 MCP evidence + PlayBook trust 產生修復候選。\n"
+            f"原始告警：{message[:500]}\n"
+            f"PlayBook：{playbook.playbook_id} / {playbook.name} / trust={playbook.trust_score:.2f}\n"
+            f"證據：{evidence_line}\n"
+            f"候選命令：{step_command}\n"
+            f"Verifier plan：{verifier_text}\n"
+            "注意：這只是 approval candidate；未經批准不得執行。"
+        )
+
+    def _build_dry_run_checks(
+        self,
+        *,
+        evidence: EvidenceSnapshot | None,
+        playbook: Playbook,
+        command: str,
+    ) -> list[DryRunCheck]:
+        sensors_attempted = evidence.sensors_attempted if evidence else 0
+        sensors_succeeded = evidence.sensors_succeeded if evidence else 0
+        return [
+            DryRunCheck(
+                name="MCP evidence collected",
+                passed=sensors_succeeded > 0,
+                message=f"{sensors_succeeded}/{sensors_attempted} sensors succeeded",
+            ),
+            DryRunCheck(
+                name="PlayBook approved and trusted",
+                passed=(
+                    playbook.status == PlaybookStatus.APPROVED
+                    and float(playbook.trust_score) >= MIN_REPAIR_CANDIDATE_TRUST
+                ),
+                message=f"{playbook.playbook_id} trust={playbook.trust_score:.2f}",
+            ),
+            DryRunCheck(
+                name="Command safety gate",
+                passed=True,
+                message=command[:240],
+            ),
+            DryRunCheck(
+                name="Post execution verifier planned",
+                passed=True,
+                message="verify health/evidence after execution before closing incident",
+            ),
+        ]
+
+    def _build_verifier_plan(
+        self,
+        *,
+        command: str,
+        namespace: str,
+        target_resource: str,
+        evidence: EvidenceSnapshot | None,
+    ) -> list[str]:
+        plan = [
+            "rerun_pre_decision_evidence_after_execution",
+            "compare_mcp_sensor_success_before_after",
+        ]
+        if command.startswith("kubectl"):
+            plan.append(f"kubectl rollout status/get events in {namespace or 'awoooi-prod'}")
+        if command.startswith("ssh "):
+            plan.append("ssh_mcp_readonly_health_check_after_write_route")
+        if target_resource:
+            plan.append(f"confirm target_resource={target_resource} alert stops recurring")
+        if evidence and evidence.snapshot_id:
+            plan.append(f"baseline_snapshot={evidence.snapshot_id}")
+        return plan
+
+    def _approval_risk(self, risk: PlaybookRiskLevel) -> ApprovalRiskLevel:
+        try:
+            mapped = ApprovalRiskLevel(risk.value.lower())
+        except Exception:
+            return ApprovalRiskLevel.MEDIUM
+        if mapped == ApprovalRiskLevel.LOW:
+            return ApprovalRiskLevel.MEDIUM
+        return mapped
+
+
+_repair_candidate_service: RepairCandidateService | None = None
+
+
+def get_repair_candidate_service() -> RepairCandidateService:
+    """Return singleton candidate builder."""
+
+    global _repair_candidate_service
+    if _repair_candidate_service is None:
+        _repair_candidate_service = RepairCandidateService()
+    return _repair_candidate_service

apps/api/tests/test_repair_candidate_service.py

@@ -0,0 +1,202 @@
+from datetime import datetime, timezone
+from uuid import uuid4
+
+import pytest
+
+from src.models.approval import ApprovalRequestCreate, BlastRadius, DataImpact, RiskLevel
+from src.models.incident import Incident, Severity, Signal
+from src.models.playbook import (
+    ActionType,
+    Playbook,
+    PlaybookSource,
+    PlaybookStatus,
+    RepairStep,
+)
+from src.models.playbook import RiskLevel as PlaybookRiskLevel
+from src.services.approval_db import approval_request_to_record_data
+from src.services.evidence_snapshot import EvidenceSnapshot
+from src.services.repair_candidate_service import RepairCandidateService
+
+
+class FakeInvestigator:
+    def __init__(self, evidence: EvidenceSnapshot | None) -> None:
+        self.evidence = evidence
+
+    async def investigate(self, incident: Incident) -> EvidenceSnapshot | None:
+        return self.evidence
+
+
+class FakePlaybookRepository:
+    def __init__(self, playbook: Playbook | None) -> None:
+        self.playbook = playbook
+
+    async def get_by_id(self, playbook_id: str) -> Playbook | None:
+        if self.playbook and self.playbook.playbook_id == playbook_id:
+            return self.playbook
+        return None
+
+
+class FakeIncidentService:
+    async def get_from_working_memory(self, incident_id: str) -> None:
+        return None
+
+
+class FakeAutoRepairService:
+    def preview_write_ssh_mcp_route(self, incident: Incident, command: str) -> bool:
+        return True
+
+
+def _incident() -> Incident:
+    return Incident(
+        incident_id="INC-TEST-REPAIR",
+        severity=Severity.P2,
+        signals=[
+            Signal(
+                alert_name="NodeExporterDown",
+                severity=Severity.P2,
+                source="alertmanager",
+                fired_at=datetime.now(timezone.utc),
+                labels={"namespace": "awoooi-prod", "deployment": "awoooi-api"},
+                annotations={"summary": "node exporter down"},
+            )
+        ],
+        affected_services=["awoooi-api"],
+    )
+
+
+def _evidence(incident_id: str, *, sensors_succeeded: int = 2) -> EvidenceSnapshot:
+    return EvidenceSnapshot(
+        incident_id=incident_id,
+        sensors_attempted=3,
+        sensors_succeeded=sensors_succeeded,
+        mcp_health={"k8s": sensors_succeeded > 0, "prometheus": sensors_succeeded > 1},
+        evidence_summary="k8s events and metrics collected",
+    )
+
+
+def _playbook(command: str, *, risk_level: PlaybookRiskLevel = PlaybookRiskLevel.LOW) -> Playbook:
+    return Playbook(
+        playbook_id="PB-REPAIR-001",
+        name="重啟 API deployment",
+        description="以 approved PlayBook 修復 API 工作負載",
+        status=PlaybookStatus.APPROVED,
+        source=PlaybookSource.MANUAL,
+        trust_score=0.72,
+        estimated_duration_minutes=4,
+        repair_steps=[
+            RepairStep(
+                step_number=1,
+                action_type=ActionType.KUBECTL,
+                command=command,
+                expected_result="deployment restarted and rollout verified",
+                risk_level=risk_level,
+                requires_approval=True,
+            )
+        ],
+    )
+
+
+@pytest.mark.asyncio
+async def test_build_candidate_from_mcp_evidence_and_approved_playbook() -> None:
+    incident = _incident()
+    service = RepairCandidateService(
+        incident_service=FakeIncidentService(),
+        investigator=FakeInvestigator(_evidence(incident.incident_id)),
+        playbook_repository=FakePlaybookRepository(
+            _playbook("kubectl rollout restart deployment/awoooi-api -n awoooi-prod")
+        ),
+        auto_repair_service=FakeAutoRepairService(),
+    )
+
+    result = await service.build_from_incident(
+        incident=incident,
+        alertname="NodeExporterDown",
+        target_resource="awoooi-api",
+        namespace="awoooi-prod",
+        message="node exporter is down",
+        fallback_action="NO_ACTION - REPAIR_CANDIDATE_MISSING",
+        matched_playbook_id="PB-REPAIR-001",
+        severity="medium",
+    )
+
+    assert result.candidate_found is True
+    assert result.approval_request is not None
+    assert result.approval_request.action == "kubectl rollout restart deployment/awoooi-api -n awoooi-prod"
+    assert result.approval_request.risk_level == RiskLevel.MEDIUM
+    assert result.approval_request.matched_playbook_id == "PB-REPAIR-001"
+    assert result.metadata["repair_candidate_status"] == "candidate_ready_for_approval"
+    assert result.metadata["mcp_evidence"]["sensors_succeeded"] == 2
+    assert result.metadata["playbook_trust"]["trust_score"] == 0.72
+    assert result.metadata["verifier_plan"]
+
+
+@pytest.mark.asyncio
+async def test_candidate_blocked_when_mcp_evidence_missing() -> None:
+    incident = _incident()
+    service = RepairCandidateService(
+        incident_service=FakeIncidentService(),
+        investigator=FakeInvestigator(_evidence(incident.incident_id, sensors_succeeded=0)),
+        playbook_repository=FakePlaybookRepository(
+            _playbook("kubectl rollout restart deployment/awoooi-api -n awoooi-prod")
+        ),
+        auto_repair_service=FakeAutoRepairService(),
+    )
+
+    result = await service.build_from_incident(
+        incident=incident,
+        alertname="NodeExporterDown",
+        target_resource="awoooi-api",
+        namespace="awoooi-prod",
+        message="node exporter is down",
+        fallback_action="NO_ACTION - REPAIR_CANDIDATE_MISSING",
+        matched_playbook_id="PB-REPAIR-001",
+        severity="medium",
+    )
+
+    assert result.candidate_found is False
+    assert "mcp_evidence_missing" in result.blockers
+    assert result.metadata["repair_candidate_status"] == "blocked"
+
+
+@pytest.mark.asyncio
+async def test_candidate_blocked_when_playbook_is_observe_only() -> None:
+    incident = _incident()
+    service = RepairCandidateService(
+        incident_service=FakeIncidentService(),
+        investigator=FakeInvestigator(_evidence(incident.incident_id)),
+        playbook_repository=FakePlaybookRepository(
+            _playbook("kubectl get pods -n awoooi-prod")
+        ),
+        auto_repair_service=FakeAutoRepairService(),
+    )
+
+    result = await service.build_from_incident(
+        incident=incident,
+        alertname="NodeExporterDown",
+        target_resource="awoooi-api",
+        namespace="awoooi-prod",
+        message="node exporter is down",
+        fallback_action="NO_ACTION - REPAIR_CANDIDATE_MISSING",
+        matched_playbook_id="PB-REPAIR-001",
+        severity="medium",
+    )
+
+    assert result.candidate_found is False
+    assert "playbook_observe_only" in result.blockers
+
+
+def test_approval_record_data_uses_preallocated_id_without_leaking_metadata() -> None:
+    approval_id = str(uuid4())
+    request = ApprovalRequestCreate(
+        action="kubectl rollout restart deployment/awoooi-api -n awoooi-prod",
+        description="candidate",
+        risk_level=RiskLevel.MEDIUM,
+        requested_by="repair-candidate-test",
+        blast_radius=BlastRadius(data_impact=DataImpact.WRITE),
+        metadata={"preallocated_approval_id": approval_id, "source": "test"},
+    )
+
+    data = approval_request_to_record_data(request, RiskLevel.MEDIUM, required_sigs=1)
+
+    assert data["id"] == approval_id
+    assert data["extra_metadata"] == {"source": "test"}