feat(awooop): expose incident reconciliation state

apps/api/src/services/awooop_truth_chain_service.py

@@ -97,6 +97,127 @@ def _operation_ids(automation_ops: list[dict[str, Any]]) -> list[str]:
     return [str(row["op_id"]) for row in automation_ops if row.get("op_id")]
 
 
+def _build_reconciliation(
+    *,
+    incident: dict[str, Any] | None,
+    approvals: list[dict[str, Any]],
+    evidence_rows: list[dict[str, Any]],
+    automation_ops: list[dict[str, Any]],
+    timeline_events: list[dict[str, Any]],
+) -> dict[str, Any]:
+    """Build a read-only consistency report across incident lifecycle tables."""
+    if incident is None:
+        return {
+            "schema_version": "incident_reconciliation_v1",
+            "applicable": False,
+            "consistency_status": "not_applicable",
+            "operator_next_state": "not_applicable",
+            "facts": {},
+            "mismatches": [],
+        }
+
+    incident_status = str(incident.get("status") or "unknown").upper()
+    incident_closed = incident_status in {"RESOLVED", "CLOSED"}
+    latest_approval = approvals[0] if approvals else None
+    approval_status = str((latest_approval or {}).get("status") or "none").upper()
+    approval_action = str((latest_approval or {}).get("action") or "")
+    approval_resolved = bool((latest_approval or {}).get("resolved_at"))
+    attempted = sum(int(row.get("sensors_attempted") or 0) for row in evidence_rows)
+    succeeded = sum(int(row.get("sensors_succeeded") or 0) for row in evidence_rows)
+    executed_ops = [
+        row
+        for row in automation_ops
+        if str(row.get("status") or "").lower()
+        in {"success", "completed", "executed"}
+    ]
+    mismatches: list[dict[str, Any]] = []
+
+    def add(code: str, severity: str, message: str) -> None:
+        mismatches.append({
+            "code": code,
+            "severity": severity,
+            "message": message,
+        })
+
+    if (
+        latest_approval
+        and not incident_closed
+        and (approval_resolved or approval_status in {"APPROVED", "REJECTED"})
+    ):
+        add(
+            "incident_open_after_approval_resolved",
+            "high",
+            "Approval reached a terminal state while the incident is still open.",
+        )
+
+    if approval_status == "APPROVED" and not automation_ops:
+        add(
+            "approval_approved_without_execution_record",
+            "high",
+            "Approval is approved but automation_operation_log has no linked execution record.",
+        )
+
+    if (
+        approval_status == "APPROVED"
+        and "NO_ACTION" in approval_action.upper()
+        and not executed_ops
+    ):
+        add(
+            "approval_no_action_without_execution",
+            "high",
+            "Approval resolved to NO_ACTION and no executor produced a successful operation.",
+        )
+
+    if attempted > 0 and succeeded == 0:
+        add(
+            "evidence_all_sensors_failed",
+            "medium",
+            "Evidence collection attempted sensors but none succeeded.",
+        )
+
+    if latest_approval and not timeline_events:
+        add(
+            "timeline_missing_for_approval",
+            "medium",
+            "Approval exists but timeline_events has no linked lifecycle entries.",
+        )
+
+    high_count = sum(1 for row in mismatches if row["severity"] == "high")
+    medium_count = sum(1 for row in mismatches if row["severity"] == "medium")
+    if high_count:
+        consistency_status = "blocked"
+        operator_next_state = "manual_required"
+    elif medium_count:
+        consistency_status = "degraded"
+        operator_next_state = "investigate"
+    else:
+        consistency_status = "consistent"
+        operator_next_state = "continue"
+
+    return {
+        "schema_version": "incident_reconciliation_v1",
+        "applicable": True,
+        "consistency_status": consistency_status,
+        "operator_next_state": operator_next_state,
+        "facts": {
+            "incident_id": incident.get("incident_id"),
+            "incident_status": incident_status,
+            "incident_closed": incident_closed,
+            "latest_approval_id": (latest_approval or {}).get("id"),
+            "latest_approval_status": approval_status,
+            "latest_approval_action": approval_action,
+            "approval_resolved": approval_resolved,
+            "evidence_records": len(evidence_rows),
+            "sensors_attempted": attempted,
+            "sensors_succeeded": succeeded,
+            "automation_operation_records": len(automation_ops),
+            "executed_operation_records": len(executed_ops),
+            "timeline_events": len(timeline_events),
+        },
+        "mismatches": mismatches,
+    }
+
+
 def _truth_status(
     *,
     incident: dict[str, Any] | None,
@@ -573,6 +694,13 @@ async def fetch_truth_chain(source_id: str, project_id: str = "awoooi") -> dict[
         legacy_mcp_total=legacy_mcp_summary["total"],
         outbound_visible_total=len(outbound_rows),
     )
+    reconciliation = _build_reconciliation(
+        incident=incident,
+        approvals=approvals,
+        evidence_rows=evidence_rows,
+        automation_ops=automation_ops,
+        timeline_events=timeline_events,
+    )
 
     evidence_totals = {
         "records": len(evidence_rows),
@@ -618,6 +746,7 @@ async def fetch_truth_chain(source_id: str, project_id: str = "awoooi") -> dict[
             "automation_operation_log": automation_ops,
             "ansible": build_ansible_truth(automation_ops, incident=incident, drift=drift),
         },
+        "reconciliation": reconciliation,
         "learning": {
             "knowledge_entries": km_entries,
         },

apps/api/tests/test_awooop_truth_chain_service.py

@@ -7,7 +7,11 @@ from src.services.awooop_ansible_audit_service import (
     build_ansible_decision_audit_payload,
     build_ansible_truth,
 )
-from src.services.awooop_truth_chain_service import _clean_row, _truth_status
+from src.services.awooop_truth_chain_service import (
+    _build_reconciliation,
+    _clean_row,
+    _truth_status,
+)
 from src.services.drift_repeat_state import (
     build_drift_fingerprint,
     build_drift_repeat_state,
@@ -157,6 +161,56 @@ def test_drift_repeat_state_counts_matching_fingerprint_only() -> None:
     ]
 
 
+def test_reconciliation_blocks_open_incident_after_no_action_approval() -> None:
+    reconciliation = _build_reconciliation(
+        incident={"incident_id": "INC-1", "status": "INVESTIGATING"},
+        approvals=[
+            {
+                "id": "approval-1",
+                "status": "APPROVED",
+                "action": "未知操作 | NO_ACTION",
+                "resolved_at": "2026-05-13T01:00:00+00:00",
+            }
+        ],
+        evidence_rows=[{"sensors_attempted": 8, "sensors_succeeded": 0}],
+        automation_ops=[],
+        timeline_events=[],
+    )
+
+    codes = {row["code"] for row in reconciliation["mismatches"]}
+    assert reconciliation["schema_version"] == "incident_reconciliation_v1"
+    assert reconciliation["consistency_status"] == "blocked"
+    assert reconciliation["operator_next_state"] == "manual_required"
+    assert reconciliation["facts"]["incident_closed"] is False
+    assert reconciliation["facts"]["automation_operation_records"] == 0
+    assert "incident_open_after_approval_resolved" in codes
+    assert "approval_approved_without_execution_record" in codes
+    assert "approval_no_action_without_execution" in codes
+    assert "evidence_all_sensors_failed" in codes
+    assert "timeline_missing_for_approval" in codes
+
+
+def test_reconciliation_marks_consistent_resolved_execution() -> None:
+    reconciliation = _build_reconciliation(
+        incident={"incident_id": "INC-2", "status": "RESOLVED"},
+        approvals=[
+            {
+                "id": "approval-2",
+                "status": "APPROVED",
+                "action": "restart service",
+                "resolved_at": "2026-05-13T01:00:00+00:00",
+            }
+        ],
+        evidence_rows=[{"sensors_attempted": 8, "sensors_succeeded": 7}],
+        automation_ops=[{"status": "success"}],
+        timeline_events=[{"event_type": "executor", "status": "success"}],
+    )
+
+    assert reconciliation["consistency_status"] == "consistent"
+    assert reconciliation["operator_next_state"] == "continue"
+    assert reconciliation["mismatches"] == []
+
+
 def test_ansible_truth_surfaces_audited_check_mode_record() -> None:
     truth = build_ansible_truth(
         [