fix(api): 補修復候選 coverage gap 契約

apps/api/src/services/repair_candidate_service.py

@@ -478,6 +478,15 @@ class RepairCandidateService:
             "approve_no_action_as_repair",
             "generic_fallback_repair",
         ]
+        coverage_gap = self._build_coverage_gap(
+            blockers=blockers,
+            lane=lane,
+            alertname=alertname,
+            target_resource=target_resource,
+            namespace=namespace,
+            evidence=evidence,
+            playbook=playbook,
+        )
         incident_id = getattr(incident, "incident_id", "") if incident else ""
         project_id = str(getattr(incident, "project_id", "awoooi") or "awoooi")
         work_item: dict[str, Any] | None = None
@@ -507,6 +516,7 @@ class RepairCandidateService:
                 "reason": ",".join(blockers),
                 "next_step": next_step,
                 "required_fields": required_fields,
+                "coverage_gap": coverage_gap,
                 "blocked_operations": blocked_operations,
                 "target_href": f"/awooop/runs?{run_query}",
                 "work_item_href": f"/awooop/work-items?{work_item_query}",
@@ -531,12 +541,110 @@ class RepairCandidateService:
             "matched_playbook_name": playbook.name if playbook else None,
             "evidence_snapshot_id": evidence_ref,
             "required_fields": required_fields,
+            "coverage_gap": coverage_gap,
             "blocked_operations": blocked_operations,
         }
         if work_item:
             package["awooop_work_item"] = work_item
         return package
 
+    def _build_coverage_gap(
+        self,
+        *,
+        blockers: list[str],
+        lane: str,
+        alertname: str,
+        target_resource: str,
+        namespace: str,
+        evidence: EvidenceSnapshot | None,
+        playbook: Playbook | None,
+    ) -> dict[str, Any]:
+        """Describe why this alert is not yet covered by an executable PlayBook."""
+
+        blocker_set = set(blockers)
+        coverage_key = self._coverage_key(alertname=alertname, target_resource=target_resource)
+        target_kind = self._infer_target_kind(target_resource=target_resource, namespace=namespace)
+        if "mcp_evidence_missing" in blocker_set:
+            blocking_stage = "mcp_evidence"
+        elif {
+            "playbook_not_matched",
+            "playbook_not_found",
+            "playbook_generic_fallback_not_repair",
+            "playbook_observe_only",
+            "playbook_has_no_repair_steps",
+            "playbook_has_no_executable_step",
+        } & blocker_set:
+            blocking_stage = "service_playbook_coverage"
+        elif {"playbook_not_approved", "playbook_trust_below_gate"} & blocker_set:
+            blocking_stage = "playbook_trust_gate"
+        elif "playbook_command_not_safely_routable" in blocker_set:
+            blocking_stage = "safe_execution_route"
+        else:
+            blocking_stage = "owner_review"
+
+        evidence_refs = [
+            "alertmanager_event",
+            "incident_timeline",
+            "mcp_health_snapshot",
+            "k8s_or_host_status",
+            "metrics_or_logs_window",
+            "recurrence_fingerprint",
+        ]
+        if target_kind == "k8s_workload":
+            evidence_refs.extend(["k8s_events", "rollout_status"])
+        elif target_kind == "host_service":
+            evidence_refs.extend(["systemd_or_container_status", "host_resource_window"])
+
+        return {
+            "schema_version": "repair_candidate_coverage_gap_v1",
+            "coverage_key": coverage_key,
+            "target_kind": target_kind,
+            "blocking_stage": blocking_stage,
+            "next_owner_lane": lane,
+            "alertname": alertname or None,
+            "namespace": namespace or None,
+            "target_resource": target_resource or None,
+            "matched_playbook_id": playbook.playbook_id if playbook else None,
+            "evidence_snapshot_id": evidence.snapshot_id if evidence else None,
+            "mcp_evidence_ready": bool(evidence and evidence.sensors_succeeded > 0),
+            "required_mcp_evidence_refs": list(dict.fromkeys(evidence_refs)),
+            "playbook_template_required": True,
+            "playbook_template_fields": [
+                "symptom_pattern.alert_names",
+                "symptom_pattern.affected_services",
+                "mcp_evidence_refs",
+                "repair_steps.command_or_ansible_ref",
+                "repair_steps.rollback_command",
+                "verifier_plan",
+                "owner_review_record",
+                "trust_score_update_policy",
+            ],
+            "blocked_operations": [
+                "auto_execute",
+                "mark_repaired_without_execution",
+                "approve_no_action_as_repair",
+            ],
+            "runtime_execution_authorized": False,
+            "writes_runtime_state": False,
+        }
+
+    def _coverage_key(self, *, alertname: str, target_resource: str) -> str:
+        raw_alert = (alertname or "unknown-alert").strip().lower()
+        raw_target = (target_resource or "unknown-target").strip().lower()
+        safe_alert = "-".join(raw_alert.replace("_", "-").split())
+        safe_target = "-".join(raw_target.replace("_", "-").split())
+        return f"{safe_alert}:{safe_target}"
+
+    def _infer_target_kind(self, *, target_resource: str, namespace: str) -> str:
+        target = (target_resource or "").lower()
+        if any(marker in target for marker in ("node-exporter", "host", "188", "111", "168")):
+            return "host_service"
+        if namespace or any(marker in target for marker in ("deployment/", "pod/", "svc/", "api")):
+            return "k8s_workload"
+        if target:
+            return "service"
+        return "unknown"
+
     def _build_description(
         self,
         *,

apps/api/tests/test_repair_candidate_service.py

@@ -229,6 +229,15 @@ async def test_candidate_blocked_when_playbook_is_generic_fallback() -> None:
     )
     assert "建立專屬 PlayBook 草案" in result.metadata["repair_candidate_next_step"]
     assert "repair_command" in result.metadata["repair_candidate_draft_package"]["required_fields"]
+    coverage_gap = result.metadata["repair_candidate_draft_package"]["coverage_gap"]
+    assert coverage_gap["schema_version"] == "repair_candidate_coverage_gap_v1"
+    assert coverage_gap["coverage_key"] == "unknownalert:awoooi-api"
+    assert coverage_gap["blocking_stage"] == "service_playbook_coverage"
+    assert coverage_gap["next_owner_lane"] == "create_service_specific_repair_playbook"
+    assert coverage_gap["mcp_evidence_ready"] is True
+    assert coverage_gap["runtime_execution_authorized"] is False
+    assert "recurrence_fingerprint" in coverage_gap["required_mcp_evidence_refs"]
+    assert "repair_steps.command_or_ansible_ref" in coverage_gap["playbook_template_fields"]
     work_item = result.metadata["repair_candidate_draft_package"]["awooop_work_item"]
     assert work_item["schema_version"] == "awooop_repair_candidate_draft_work_item_v1"
     assert work_item["work_item_id"].startswith(
@@ -238,6 +247,7 @@ async def test_candidate_blocked_when_playbook_is_generic_fallback() -> None:
     assert work_item["needs_human"] is True
     assert work_item["decision_effect"] == "none"
     assert work_item["writes_runtime_state"] is False
+    assert work_item["coverage_gap"]["coverage_key"] == "unknownalert:awoooi-api"
     assert "/awooop/work-items?" in work_item["work_item_href"]
     assert "https://awoooi.wooo.work/zh-TW/awooop/work-items?" in work_item["work_item_url"]
 
@@ -278,11 +288,50 @@ async def test_candidate_blocked_observe_only_prompts_repair_playbook_draft() ->
     assert result.metadata["repair_candidate_draft_package"]["lane"] == (
         "promote_diagnostic_to_repair_playbook"
     )
+    coverage_gap = result.metadata["repair_candidate_draft_package"]["coverage_gap"]
+    assert coverage_gap["coverage_key"] == "nodeexporterdown:node-exporter-188"
+    assert coverage_gap["target_kind"] == "host_service"
+    assert coverage_gap["blocking_stage"] == "service_playbook_coverage"
+    assert coverage_gap["matched_playbook_id"] == "PB-REPAIR-001"
+    assert "systemd_or_container_status" in coverage_gap["required_mcp_evidence_refs"]
     assert "診斷命令保留為 MCP evidence collector" in result.metadata["repair_candidate_next_step"]
     work_item = result.metadata["repair_candidate_draft_package"]["awooop_work_item"]
     assert work_item["target_resource"] == "node-exporter-188"
     assert work_item["lane"] == "promote_diagnostic_to_repair_playbook"
     assert work_item["safety_level"] == "read_only_work_item_projection"
+    assert work_item["coverage_gap"]["next_owner_lane"] == "promote_diagnostic_to_repair_playbook"
+
+
+@pytest.mark.asyncio
+async def test_missing_mcp_evidence_records_collectable_coverage_gap() -> 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",
+    )
+
+    coverage_gap = result.metadata["repair_candidate_draft_package"]["coverage_gap"]
+    assert coverage_gap["coverage_key"] == "nodeexporterdown:awoooi-api"
+    assert coverage_gap["blocking_stage"] == "mcp_evidence"
+    assert coverage_gap["mcp_evidence_ready"] is False
+    assert coverage_gap["target_kind"] == "k8s_workload"
+    assert "mcp_health_snapshot" in coverage_gap["required_mcp_evidence_refs"]
+    assert coverage_gap["runtime_execution_authorized"] is False
 
 
 def test_approval_record_data_uses_preallocated_id_without_leaking_metadata() -> None: