feat(awooop): surface execution backend evidence

apps/api/src/services/awooop_truth_chain_service.py

@@ -620,6 +620,91 @@ def _automation_quality_score_bucket(score: int) -> str:
     return "red"
 
 
+def _int_value(value: Any, fallback: int = 0) -> int:
+    try:
+        return int(value)
+    except (TypeError, ValueError):
+        return fallback
+
+
+def _execution_backend_summary(records: list[dict[str, Any]]) -> dict[str, Any]:
+    summary = {
+        "operation_records_total": 0,
+        "effective_execution_records_total": 0,
+        "noop_operation_records_total": 0,
+        "audit_only_operation_records_total": 0,
+        "auto_repair_execution_records_total": 0,
+        "ansible_considered_total": 0,
+        "ansible_audit_record_total": 0,
+        "ansible_candidate_total": 0,
+        "ansible_check_mode_total": 0,
+        "ansible_apply_total": 0,
+        "ansible_rollback_total": 0,
+        "ansible_pending_check_mode_total": 0,
+    }
+
+    for record in records:
+        quality = record.get("automation_quality") if isinstance(record.get("automation_quality"), dict) else {}
+        facts = quality.get("facts") if isinstance(quality.get("facts"), dict) else {}
+        execution = record.get("execution") if isinstance(record.get("execution"), dict) else {}
+        ops = (
+            execution.get("automation_operation_log")
+            if isinstance(execution.get("automation_operation_log"), list)
+            else []
+        )
+        auto_repair_executions = (
+            execution.get("auto_repair_executions")
+            if isinstance(execution.get("auto_repair_executions"), list)
+            else []
+        )
+        ansible = execution.get("ansible") if isinstance(execution.get("ansible"), dict) else {}
+        ansible_records = ansible.get("records") if isinstance(ansible.get("records"), list) else []
+        catalog = (
+            ansible.get("candidate_catalog")
+            if isinstance(ansible.get("candidate_catalog"), dict)
+            else {}
+        )
+        candidates = catalog.get("candidates") if isinstance(catalog.get("candidates"), list) else []
+
+        summary["operation_records_total"] += _int_value(facts.get("automation_operation_records"), len(ops))
+        summary["effective_execution_records_total"] += _int_value(
+            facts.get("effective_execution_records"),
+            len(_effective_execution_ops([row for row in ops if isinstance(row, dict)])),
+        )
+        summary["noop_operation_records_total"] += _int_value(
+            facts.get("noop_operation_records"),
+            sum(1 for row in ops if isinstance(row, dict) and _is_no_action_operation(row)),
+        )
+        summary["audit_only_operation_records_total"] += _int_value(
+            facts.get("audit_only_operation_records"),
+            sum(1 for row in ops if isinstance(row, dict) and _is_audit_only_operation(row)),
+        )
+        summary["auto_repair_execution_records_total"] += _int_value(
+            facts.get("auto_repair_execution_records"),
+            len(auto_repair_executions),
+        )
+
+        if ansible.get("considered") is True:
+            summary["ansible_considered_total"] += 1
+        summary["ansible_audit_record_total"] += len(ansible_records)
+        summary["ansible_candidate_total"] += len(candidates)
+
+        for row in ansible_records:
+            if not isinstance(row, dict):
+                continue
+            operation_type = str(row.get("operation_type") or "")
+            if operation_type == "ansible_check_mode_executed":
+                summary["ansible_check_mode_total"] += 1
+            elif operation_type == "ansible_apply_executed":
+                summary["ansible_apply_total"] += 1
+            elif operation_type == "ansible_rollback_executed":
+                summary["ansible_rollback_total"] += 1
+            elif operation_type == "ansible_candidate_matched":
+                summary["ansible_pending_check_mode_total"] += 1
+
+    return summary
+
+
 def summarize_automation_quality_records(
     *,
     project_id: str,
@@ -724,6 +809,7 @@ def summarize_automation_quality_records(
         "score_buckets": score_buckets,
         "by_verdict": by_verdict,
         "gate_failures": failing_gates,
+        "execution_backend_summary": _execution_backend_summary(records),
         "examples": examples[:25],
         "production_claim": {
             "can_claim_full_auto_repair": evaluated_total > 0 and verified_total == evaluated_total,
@@ -1491,6 +1577,7 @@ async def fetch_automation_quality_summary(
             "incident": truth_chain.get("incident") or incident,
             "truth_status": truth_chain.get("truth_status") or {},
             "automation_quality": truth_chain.get("automation_quality") or {},
+            "execution": truth_chain.get("execution") or {},
         }
 
     records = [