feat(awooop): surface automation flow gates

apps/api/src/services/awooop_truth_chain_service.py

@@ -681,6 +681,184 @@ def _automation_quality_score_bucket(score: int) -> str:
     return "red"
 
 
+_AUTOMATION_FLOW_GATE_DEFINITIONS: tuple[dict[str, Any], ...] = (
+    {
+        "gate": "alert_intake",
+        "quality_gates": ("source_persisted", "outbound_recorded"),
+        "allow_not_applicable": False,
+        "next_action": "repair_alert_intake_or_outbound_mirror",
+    },
+    {
+        "gate": "mcp_investigation",
+        "quality_gates": ("evidence_collected", "mcp_gateway_observed"),
+        "allow_not_applicable": False,
+        "next_action": "route_incident_to_mcp_gateway_and_evidence_collectors",
+    },
+    {
+        "gate": "approval_policy",
+        "quality_gates": ("approval_state",),
+        "allow_not_applicable": True,
+        "next_action": "resolve_pending_or_expired_human_gate",
+    },
+    {
+        "gate": "execution_recorded",
+        "quality_gates": ("execution_recorded",),
+        "allow_not_applicable": False,
+        "next_action": "record_effective_execution_or_mark_manual_no_action",
+    },
+    {
+        "gate": "repair_recorded",
+        "quality_gates": ("auto_repair_recorded",),
+        "allow_not_applicable": False,
+        "next_action": "write_auto_repair_execution_or_blocker_reason",
+    },
+    {
+        "gate": "verification_recorded",
+        "quality_gates": ("verification_recorded",),
+        "allow_not_applicable": False,
+        "next_action": "run_post_execution_verification",
+    },
+    {
+        "gate": "knowledge_recorded",
+        "quality_gates": ("learning_recorded",),
+        "allow_not_applicable": False,
+        "next_action": "write_km_or_learning_evidence",
+    },
+    {
+        "gate": "operator_visible",
+        "quality_gates": ("outbound_recorded", "timeline_recorded"),
+        "allow_not_applicable": False,
+        "next_action": "repair_timeline_or_operator_notification_visibility",
+    },
+)
+
+
+def _quality_gate_status_map(quality: dict[str, Any]) -> dict[str, str]:
+    statuses: dict[str, str] = {}
+    for row in quality.get("gates") or []:
+        if not isinstance(row, dict):
+            continue
+        name = str(row.get("name") or "")
+        if not name:
+            continue
+        statuses[name] = str(row.get("status") or "missing")
+    return statuses
+
+
+def _automation_flow_gate_record_status(
+    definition: dict[str, Any],
+    gate_statuses: dict[str, str],
+) -> tuple[str, dict[str, str]]:
+    aliases = definition.get("source_gate_aliases") or {}
+    source_statuses: dict[str, str] = {}
+    normalized: list[str] = []
+    for source_gate in definition["quality_gates"]:
+        mapped_gate = str(aliases.get(source_gate, source_gate))
+        raw_status = gate_statuses.get(mapped_gate, "missing")
+        status = str(raw_status or "missing")
+        source_statuses[mapped_gate] = status
+        if status == "not_applicable" and not definition.get("allow_not_applicable"):
+            status = "missing"
+        normalized.append(status)
+
+    if any(status == "failed" for status in normalized):
+        return "failed", source_statuses
+    if any(status == "missing" for status in normalized):
+        return "missing", source_statuses
+    if any(status == "warning" for status in normalized):
+        return "warning", source_statuses
+    return "passed", source_statuses
+
+
+def _automation_flow_gate_summary(records: list[dict[str, Any]]) -> dict[str, Any]:
+    applicable_records = [
+        record
+        for record in records
+        if isinstance(record.get("automation_quality"), dict)
+        and record["automation_quality"].get("applicable") is True
+    ]
+    evaluated_total = len(applicable_records)
+    gate_rows: list[dict[str, Any]] = []
+
+    for definition in _AUTOMATION_FLOW_GATE_DEFINITIONS:
+        counts = {"passed": 0, "warning": 0, "missing": 0, "failed": 0}
+        examples: list[dict[str, Any]] = []
+        for record in applicable_records:
+            quality = record["automation_quality"]
+            incident = record.get("incident") if isinstance(record.get("incident"), dict) else {}
+            truth_status = (
+                record.get("truth_status")
+                if isinstance(record.get("truth_status"), dict)
+                else {}
+            )
+            status, source_statuses = _automation_flow_gate_record_status(
+                definition,
+                _quality_gate_status_map(quality),
+            )
+            counts[status] += 1
+            if status != "passed" and len(examples) < 5:
+                examples.append({
+                    "incident_id": incident.get("incident_id"),
+                    "alertname": incident.get("alertname"),
+                    "verdict": quality.get("verdict"),
+                    "truth_stage": truth_status.get("current_stage"),
+                    "source_statuses": source_statuses,
+                    "blockers": list(quality.get("blockers") or [])[:6],
+                })
+
+        blocked_total = counts["failed"] + counts["missing"]
+        if evaluated_total == 0:
+            status = "no_data"
+        elif blocked_total > 0:
+            status = "blocked"
+        elif counts["warning"] > 0:
+            status = "warning"
+        else:
+            status = "passed"
+        passed_percent = (
+            round((counts["passed"] / evaluated_total) * 100, 1)
+            if evaluated_total
+            else 0.0
+        )
+        gate_rows.append({
+            "gate": definition["gate"],
+            "status": status,
+            "passed_total": counts["passed"],
+            "warning_total": counts["warning"],
+            "missing_total": counts["missing"],
+            "failed_total": counts["failed"],
+            "evaluated_total": evaluated_total,
+            "passed_percent": passed_percent,
+            "quality_gates": list(definition["quality_gates"]),
+            "next_action": definition["next_action"],
+            "examples": examples,
+        })
+
+    blocked_gates = [
+        row["gate"]
+        for row in gate_rows
+        if row["status"] in {"blocked", "no_data"}
+    ]
+    warning_gates = [row["gate"] for row in gate_rows if row["status"] == "warning"]
+    if evaluated_total == 0:
+        overall_status = "no_data"
+    elif blocked_gates:
+        overall_status = "blocked"
+    elif warning_gates:
+        overall_status = "warning"
+    else:
+        overall_status = "passed"
+
+    return {
+        "schema_version": "automation_flow_gate_summary_v1",
+        "evaluated_total": evaluated_total,
+        "overall_status": overall_status,
+        "blocked_gates": blocked_gates,
+        "warning_gates": warning_gates,
+        "gates": gate_rows,
+    }
+
+
 def _int_value(value: Any, fallback: int = 0) -> int:
     try:
         return int(value)
@@ -999,6 +1177,7 @@ def summarize_automation_quality_records(
         "score_buckets": score_buckets,
         "by_verdict": by_verdict,
         "gate_failures": failing_gates,
+        "automation_flow_gates": _automation_flow_gate_summary(records),
         "execution_backend_summary": _execution_backend_summary(records),
         "ansible_runtime": ansible_runtime,
         "examples": examples[:25],