feat(governance): 顯示報表資料可信度 gate

apps/api/src/services/ai_agent_report_truth_actionability_review.py

@@ -33,6 +33,7 @@ def load_latest_ai_agent_report_truth_actionability_review(
     _require_schema(payload, str(latest))
     _require_runtime_boundaries(payload, str(latest))
     _require_findings(payload, str(latest))
+    _require_source_and_actionability_gates(payload, str(latest))
     _require_telegram_routing(payload, str(latest))
     _require_rollup_consistency(payload, str(latest))
     return payload
@@ -46,8 +47,8 @@ def _require_schema(payload: dict[str, Any], label: str) -> None:
         raise ValueError(f"{label}: program_status.read_only_mode must be true")
     if status.get("runtime_authority") != "report_truth_actionability_review_only_no_report_send_or_runtime_fix":
         raise ValueError(f"{label}: runtime_authority must remain report-truth review only")
-    if status.get("current_task_id") != "P2-403J" or status.get("next_task_id") != "P2-403K":
-        raise ValueError(f"{label}: current/next task must remain P2-403J -> P2-403K")
+    if status.get("current_task_id") != "P2-403K" or status.get("next_task_id") != "P2-403L":
+        raise ValueError(f"{label}: current/next task must remain P2-403K -> P2-403L")
 
 
 def _require_runtime_boundaries(payload: dict[str, Any], label: str) -> None:
@@ -58,6 +59,11 @@ def _require_runtime_boundaries(payload: dict[str, Any], label: str) -> None:
         "freshness_gate_implemented",
         "source_confidence_gate_implemented",
         "actionability_score_implemented",
+    ):
+        if truth.get(flag) is not True:
+            raise ValueError(f"{label}: {flag} must be true for P2-403K read-only gates")
+
+    for flag in (
         "ai_agent_runtime_control_allowed",
         "telegram_report_send_allowed",
         "cronjob_change_allowed",
@@ -98,6 +104,74 @@ def _require_findings(payload: dict[str, Any], label: str) -> None:
             raise ValueError(f"{label}: missing actionability lane {lane_id}")
 
 
+def _require_source_and_actionability_gates(payload: dict[str, Any], label: str) -> None:
+    freshness = payload.get("source_freshness_gates") or []
+    freshness_ids = {item.get("gate_id") for item in freshness}
+    required_freshness = {
+        "stats_api_incident_rollup",
+        "k3s_metrics_rollup",
+        "gitea_activity_rollup",
+        "ai_cost_ledger",
+        "telegram_delivery_receipt",
+        "heartbeat_watchdog",
+    }
+    missing_freshness = sorted(required_freshness - freshness_ids)
+    if missing_freshness:
+        raise ValueError(f"{label}: missing source freshness gates: {missing_freshness}")
+    for item in freshness:
+        if item.get("work_item_required") is not True:
+            raise ValueError(f"{label}: source freshness gate {item.get('gate_id')} must require a work item")
+        for field in ("freshness_state", "last_success_at", "required_ttl", "ai_next_action", "owner_agent"):
+            if not item.get(field):
+                raise ValueError(f"{label}: source freshness gate {item.get('gate_id')} missing {field}")
+
+    confidence = payload.get("source_confidence_gates") or []
+    confidence_ids = {item.get("confidence_id") for item in confidence}
+    required_confidence = {
+        "alerts_count_confidence",
+        "ai_performance_confidence",
+        "k3s_health_confidence",
+        "development_activity_confidence",
+        "ai_cost_confidence",
+    }
+    missing_confidence = sorted(required_confidence - confidence_ids)
+    if missing_confidence:
+        raise ValueError(f"{label}: missing source confidence gates: {missing_confidence}")
+    for item in confidence:
+        if item.get("confidence") != "low":
+            raise ValueError(f"{label}: source confidence gate {item.get('confidence_id')} must remain low until live source readback")
+        if item.get("blocked_green_badge") is not True:
+            raise ValueError(f"{label}: source confidence gate {item.get('confidence_id')} must block green badge")
+
+    scores = payload.get("report_actionability_scores") or []
+    score_ids = {item.get("report_id") for item in scores}
+    required_scores = {"daily_report", "weekly_report", "monthly_report", "heartbeat_digest"}
+    missing_scores = sorted(required_scores - score_ids)
+    if missing_scores:
+        raise ValueError(f"{label}: missing report actionability scores: {missing_scores}")
+    weekly = next((item for item in scores if item.get("report_id") == "weekly_report"), None)
+    if not weekly or weekly.get("score", 0) < 80 or weekly.get("lane") != "action_required_source_gap":
+        raise ValueError(f"{label}: weekly report must remain action-required source gap with score >= 80")
+
+    work_items = payload.get("zero_signal_work_items") or []
+    work_item_ids = {item.get("work_item_id") for item in work_items}
+    required_work_items = {
+        "report-source-gap:stats_api",
+        "report-source-gap:k3s_metrics",
+        "report-source-gap:gitea_activity",
+        "report-source-gap:ai_cost_ledger",
+        "report-source-gap:telegram_delivery",
+    }
+    missing_work_items = sorted(required_work_items - work_item_ids)
+    if missing_work_items:
+        raise ValueError(f"{label}: missing zero signal work items: {missing_work_items}")
+    for item in work_items:
+        if item.get("status") != "draft_required":
+            raise ValueError(f"{label}: zero signal work item {item.get('work_item_id')} must remain draft_required")
+        if not item.get("required_assets"):
+            raise ValueError(f"{label}: zero signal work item {item.get('work_item_id')} missing required assets")
+
+
 def _require_telegram_routing(payload: dict[str, Any], label: str) -> None:
     routing = payload.get("telegram_routing_consolidation") or {}
     if routing.get("canonical_room_name") != "AwoooI SRE 戰情室":
@@ -136,6 +210,10 @@ def _require_rollup_consistency(payload: dict[str, Any], label: str) -> None:
     lanes = payload.get("alert_actionability_lanes") or []
     routes = payload.get("telegram_route_findings") or []
     actions = payload.get("operator_actions") or []
+    freshness = payload.get("source_freshness_gates") or []
+    confidence = payload.get("source_confidence_gates") or []
+    scores = payload.get("report_actionability_scores") or []
+    work_items = payload.get("zero_signal_work_items") or []
     blocked = {
         *(item.get("blocked_runtime_action") for item in findings),
         *(item.get("blocked_runtime_action") for item in routes),
@@ -145,6 +223,16 @@ def _require_rollup_consistency(payload: dict[str, Any], label: str) -> None:
     blocked.discard(None)
     expected_counts = {
         "zero_signal_finding_count": len(findings),
+        "source_freshness_gate_count": len(freshness),
+        "stale_or_untrusted_source_count": sum(
+            1 for item in freshness if item.get("freshness_state") != "report_only_when_fresh"
+        ),
+        "source_confidence_gate_count": len(confidence),
+        "low_confidence_gate_count": sum(1 for item in confidence if item.get("confidence") == "low"),
+        "report_actionability_score_count": len(scores),
+        "actionability_score_ready_count": sum(1 for item in scores if isinstance(item.get("score"), int) and item.get("score") > 0),
+        "zero_signal_work_item_count": len(work_items),
+        "work_item_required_count": sum(1 for item in freshness if item.get("work_item_required") is True),
         "critical_finding_count": sum(1 for item in findings if item.get("severity") == "critical"),
         "high_finding_count": sum(1 for item in findings if item.get("severity") == "high"),
         "cadence_contract_count": len(cadences),
@@ -176,3 +264,6 @@ def _require_rollup_consistency(payload: dict[str, Any], label: str) -> None:
         raise ValueError(f"{label}: approval_required_action_ids mismatch")
     if rollups.get("all_zero_weekly_report_confidence") != "low_trust_actionable_anomaly":
         raise ValueError(f"{label}: all-zero weekly report confidence must stay low trust")
+    weekly = next((item for item in scores if item.get("report_id") == "weekly_report"), {})
+    if rollups.get("all_zero_weekly_report_actionability_score") != weekly.get("score"):
+        raise ValueError(f"{label}: all-zero weekly report actionability score mismatch")