"""AwoooP read-only truth chain aggregation.

T0 only: this service does not mutate incident, approval, execution, or channel
state. It stitches existing durable records into one operator-facing status so
Telegram cards can be audited without guessing which subsystem owns the truth.
"""

from __future__ import annotations

import json
import os
import shutil
from datetime import UTC, date, datetime, timedelta
from decimal import Decimal
from pathlib import Path
from time import perf_counter, time
from typing import Any
from uuid import UUID

import structlog
from sqlalchemy import text

from src.core.config import settings
from src.db.base import get_db_context
from src.services.operator_outcome import build_operator_outcome
from src.services.awooop_ansible_check_mode_service import detect_ansible_transport_blockers
from src.services.awooop_ansible_audit_service import build_ansible_truth
from src.services.drift_repeat_state import build_drift_repeat_state
from src.services.operator_summary_cache import (
    get_cached_operator_summary_async,
    store_operator_summary_async,
)

logger = structlog.get_logger(__name__)

_MAX_ROWS = 100
_JSON_TEXT_FIELDS = {"gate_result", "source_envelope"}
_QUALITY_SUMMARY_CACHE_TTL_SECONDS = int(
    os.getenv("AWOOOP_QUALITY_SUMMARY_CACHE_TTL_SECONDS", "30")
)
_QUALITY_SUMMARY_OBSERVATIONS: dict[str, dict[str, Any]] = {}


def record_quality_summary_observation(
    *,
    project_id: str,
    hours: int,
    limit: int,
    cache_status: str,
    success: bool,
    duration_seconds: float,
    error: str | None = None,
) -> None:
    normalized_project_id = project_id or "awoooi"
    normalized_cache_status = cache_status or "unknown"
    key = "|".join([
        normalized_project_id,
        str(int(hours)),
        str(int(limit)),
        normalized_cache_status,
        "success" if success else "failed",
    ])
    _QUALITY_SUMMARY_OBSERVATIONS[key] = {
        "project_id": normalized_project_id,
        "hours": int(hours),
        "limit": int(limit),
        "cache_status": normalized_cache_status,
        "success": bool(success),
        "duration_seconds": max(0.0, float(duration_seconds)),
        "observed_at": time(),
        "error": str(error)[:160] if error else None,
    }


def get_quality_summary_observations() -> list[dict[str, Any]]:
    return [
        dict(observation)
        for observation in sorted(
            _QUALITY_SUMMARY_OBSERVATIONS.values(),
            key=lambda item: (
                str(item.get("project_id") or ""),
                int(item.get("hours") or 0),
                int(item.get("limit") or 0),
                str(item.get("cache_status") or ""),
                bool(item.get("success")),
            ),
        )
    ]


def _clean(value: Any) -> Any:
    """Convert DB values into JSON-friendly primitives."""
    if isinstance(value, UUID):
        return str(value)
    if isinstance(value, (datetime, date)):
        return value.isoformat()
    if isinstance(value, Decimal):
        return float(value)
    if isinstance(value, dict):
        return {str(k): _clean(v) for k, v in value.items()}
    if isinstance(value, list):
        return [_clean(v) for v in value]
    return value


def _clean_row(row: Any) -> dict[str, Any]:
    cleaned: dict[str, Any] = {}
    for key, value in dict(row).items():
        if key in _JSON_TEXT_FIELDS and isinstance(value, str):
            try:
                value = json.loads(value)
            except json.JSONDecodeError:
                pass
        cleaned[key] = _clean(value)
    return cleaned


async def _fetch_all(db: Any, sql: str, params: dict[str, Any]) -> list[dict[str, Any]]:
    result = await db.execute(text(sql), params)
    return [_clean_row(row) for row in result.mappings().all()]


async def _fetch_one(
    db: Any,
    sql: str,
    params: dict[str, Any],
) -> dict[str, Any] | None:
    result = await db.execute(text(sql), params)
    row = result.mappings().first()
    return _clean_row(row) if row else None


def _source_type(source_id: str, incident: dict[str, Any] | None, drift: dict[str, Any] | None) -> str:
    if incident is not None:
        return "incident"
    if drift is not None:
        return "drift_report"
    try:
        UUID(source_id)
    except ValueError:
        return "unknown"
    return "run"


def _failed_mcp_tools(evidence_rows: list[dict[str, Any]]) -> list[str]:
    failed: set[str] = set()
    for evidence in evidence_rows:
        mcp_health = evidence.get("mcp_health")
        if isinstance(mcp_health, dict):
            failed.update(str(tool) for tool, ok in mcp_health.items() if ok is False)
    return sorted(failed)


def _approval_ids(approvals: list[dict[str, Any]]) -> list[str]:
    return [str(row["id"]) for row in approvals if row.get("id")]


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 _auto_repair_ids(auto_repair_executions: list[dict[str, Any]]) -> list[str]:
    return [str(row["id"]) for row in auto_repair_executions if row.get("id")]


def _incident_fingerprints(incident: dict[str, Any] | None) -> list[str]:
    """Extract durable alert fingerprints from incident signals, when present."""
    if not incident:
        return []
    signals = incident.get("signals")
    if not isinstance(signals, list):
        return []

    fingerprints: set[str] = set()
    for signal in signals:
        if not isinstance(signal, dict):
            continue
        direct = signal.get("fingerprint")
        if direct:
            fingerprints.add(str(direct))
        labels = signal.get("labels")
        if isinstance(labels, dict):
            value = labels.get("fingerprint")
            if value:
                fingerprints.add(str(value))
    return sorted(fingerprints)


def _looks_like_no_action(value: Any) -> bool:
    text = str(value or "").upper()
    return (
        "NO_ACTION" in text
        or "NO-ACTION" in text
        or "NOACTION" in text
        or text.startswith("OBSERVE")
        or text.startswith("INVESTIGATE")
    )


def _approval_has_no_action(approvals: list[dict[str, Any]]) -> bool:
    return any(_looks_like_no_action(row.get("action")) for row in approvals)


def _approval_suppresses_repair_execution(approvals: list[dict[str, Any]]) -> bool:
    for row in approvals:
        metadata = row.get("extra_metadata")
        if not isinstance(metadata, dict):
            continue
        execution_kind = str(metadata.get("execution_kind") or "").lower()
        if execution_kind in {"no_action", "diagnostic", "parse_failed", "unsupported_action"}:
            return True
        if metadata.get("repair_executed") is False and not execution_kind:
            return True
    return False


def _is_no_action_operation(row: dict[str, Any]) -> bool:
    """Return true for durable audit rows that represent observation, not repair."""
    if str(row.get("operation_type") or "") != "playbook_executed":
        return False
    return any(
        _looks_like_no_action(row.get(key))
        for key in (
            "input_action",
            "output_action",
            "output_reason",
            "output_not_used_reason",
        )
    )


def _is_audit_only_operation(row: dict[str, Any]) -> bool:
    operation_type = str(row.get("operation_type") or "")
    status = str(row.get("status") or "").lower()
    if status == "dry_run":
        return True
    return operation_type in {
        "ansible_candidate_matched",
        "ansible_execution_skipped",
    }


def _effective_execution_ops(automation_ops: list[dict[str, Any]]) -> list[dict[str, Any]]:
    return [
        row
        for row in automation_ops
        if not _is_no_action_operation(row) and not _is_audit_only_operation(row)
    ]


def build_incident_reconciliation(
    *,
    incident: dict[str, Any] | None,
    approvals: list[dict[str, Any]],
    evidence_rows: list[dict[str, Any]],
    automation_ops: list[dict[str, Any]],
    auto_repair_executions: list[dict[str, Any]] | None = None,
    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)
    repair_rows = auto_repair_executions or []
    approval_suppressed = _approval_suppresses_repair_execution(approvals)
    effective_ops = [] if approval_suppressed else _effective_execution_ops(automation_ops)
    executed_ops = [
        row
        for row in effective_ops
        if str(row.get("status") or "").lower()
        in {"success", "completed", "executed"}
    ]
    successful_repairs = [row for row in repair_rows if row.get("success") is True]
    effective_execution_count = len(executed_ops) + len(successful_repairs)
    latest_verification = _latest_verification_result(incident, evidence_rows)
    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"})
    ):
        if effective_execution_count:
            add(
                "incident_open_after_successful_execution",
                "high",
                "Execution evidence exists while the incident is still open.",
            )
        else:
            add(
                "incident_open_after_approval_resolved",
                "high",
                "Approval reached a terminal state while the incident is still open.",
            )

    if approval_status == "APPROVED" and not effective_execution_count:
        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 effective_execution_count
    ):
        add(
            "approval_no_action_without_execution",
            "high",
            "Approval resolved to NO_ACTION and no executor produced a successful operation.",
        )

    if successful_repairs and str(latest_verification or "").lower() == "degraded":
        add(
            "verification_degraded_after_auto_repair",
            "medium",
            "Auto-repair succeeded, but post-execution verification reported degraded evidence.",
        )

    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),
            "auto_repair_execution_records": len(repair_rows),
            "successful_auto_repair_records": len(successful_repairs),
            "effective_execution_records": effective_execution_count,
            "latest_verification_result": latest_verification,
            "timeline_events": len(timeline_events),
        },
        "mismatches": mismatches,
    }


def _truth_status(
    *,
    incident: dict[str, Any] | None,
    approvals: list[dict[str, Any]],
    evidence_rows: list[dict[str, Any]],
    automation_ops: list[dict[str, Any]],
    drift: dict[str, Any] | None,
    drift_repeat_count: int,
    gateway_mcp_total: int,
    legacy_mcp_total: int,
    outbound_visible_total: int,
    inbound_visible_total: int = 0,
    auto_repair_executions: list[dict[str, Any]] | None = None,
) -> dict[str, Any]:
    """Derive the current operator-visible truth-chain stage."""
    blockers: list[str] = []
    needs_human = False
    stage = "not_found"
    stage_status = "missing"

    if drift is not None:
        stage = "dedup_or_repeat_updated" if drift_repeat_count > 1 else "received"
        stage_status = str(drift.get("status") or "unknown")
        interpretation = drift.get("interpretation") or {}
        confidence = interpretation.get("confidence") if isinstance(interpretation, dict) else None
        if stage_status == "pending":
            needs_human = True
            blockers.append("drift_report_pending_without_resolution")
        if confidence in (0, 0.0):
            blockers.append("drift_ai_confidence_zero")

    if incident is None and drift is None and inbound_visible_total > 0:
        stage = "inbound_received"
        stage_status = "observed"

    if incident is not None:
        incident_status = str(incident.get("status") or "unknown")
        repair_rows = auto_repair_executions or []
        approval_suppressed = _approval_suppresses_repair_execution(approvals)
        effective_ops = [] if approval_suppressed else _effective_execution_ops(automation_ops)
        has_execution_records = bool(effective_ops or repair_rows)
        stage = "received"
        stage_status = incident_status.lower()
        if incident_status in {"RESOLVED", "CLOSED"}:
            stage = "resolved"
            stage_status = "success"
        elif evidence_rows:
            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)
            if attempted > 0 and succeeded == 0:
                stage = "evidence_degraded"
                stage_status = "warning"
                needs_human = True
                blockers.append("all_evidence_sensors_failed")
            else:
                stage = "evidence_collected"
                stage_status = "completed"

        approval_statuses = {str(row.get("status") or "").upper() for row in approvals}
        approval_actions = " ".join(str(row.get("action") or "") for row in approvals).upper()
        approval_no_action = _approval_has_no_action(approvals)
        if any(status in {"PENDING", "WAITING_APPROVAL"} for status in approval_statuses):
            stage = "approval_required"
            stage_status = "waiting"
            needs_human = True
            blockers.append("pending_human_approval")
        elif "EXPIRED" in approval_statuses and not has_execution_records:
            stage = "approval_expired"
            stage_status = "expired"
            needs_human = True
            blockers.append("approval_expired_without_operator_decision")
        elif "REJECTED" in approval_statuses and not has_execution_records:
            stage = "approval_rejected"
            stage_status = "closed"
        elif "EXECUTION_FAILED" in approval_statuses and not has_execution_records:
            stage = "execution_failed"
            stage_status = "error"
            needs_human = True
            blockers.append("approval_execution_failed_without_execution_record")
        elif not has_execution_records and (approval_no_action or "NO_ACTION" in approval_actions):
            if approval_statuses:
                stage = "manual_required"
                stage_status = "blocked"
                needs_human = True
                blockers.append("approval_resolved_no_action_without_execution")
        elif "APPROVED" in approval_statuses and not has_execution_records:
            stage = "execution_missing"
            stage_status = "blocked"
            needs_human = True
            blockers.append("approved_without_execution_record")

        op_statuses = {str(row.get("status") or "").lower() for row in effective_ops}
        repair_successes = {row.get("success") for row in repair_rows}
        execution_succeeded = (op_statuses & {"success", "completed"}) or True in repair_successes
        execution_failed = (op_statuses & {"failed", "error"}) or False in repair_successes
        if op_statuses or repair_successes:
            if execution_succeeded:
                stage = "execution_succeeded"
                stage_status = "success"
            elif execution_failed:
                stage = "execution_failed"
                stage_status = "error"
                needs_human = True
            else:
                stage = "execution_started"
                stage_status = "running"

        if incident_status == "INVESTIGATING" and approvals:
            if execution_succeeded:
                blockers.append("incident_open_after_successful_execution")
                needs_human = True
            elif not has_execution_records:
                blockers.append("incident_still_investigating_after_approval")

    if gateway_mcp_total == 0:
        blockers.append("awooop_mcp_gateway_audit_empty")
    if legacy_mcp_total == 0 and incident is not None:
        blockers.append("legacy_mcp_audit_missing")
    if outbound_visible_total == 0:
        blockers.append("outbound_mirror_not_visible_for_source")

    return {
        "current_stage": stage,
        "stage_status": stage_status,
        "needs_human": needs_human,
        "blockers": blockers,
    }


def _latest_verification_result(
    incident: dict[str, Any] | None,
    evidence_rows: list[dict[str, Any]],
) -> str | None:
    if incident and incident.get("verification_result"):
        return str(incident["verification_result"])
    for row in evidence_rows:
        if row.get("verification_result"):
            return str(row["verification_result"])
    return None


def build_automation_quality(
    *,
    incident: dict[str, Any] | None,
    approvals: list[dict[str, Any]],
    evidence_rows: list[dict[str, Any]],
    automation_ops: list[dict[str, Any]],
    auto_repair_executions: list[dict[str, Any]],
    gateway_mcp_summary: dict[str, Any],
    legacy_mcp_summary: dict[str, Any],
    outbound_rows: list[dict[str, Any]],
    km_entries: list[dict[str, Any]],
    timeline_events: list[dict[str, Any]],
) -> dict[str, Any]:
    """Summarize whether a card reached the real automation flywheel."""
    if incident is None:
        return {
            "schema_version": "automation_quality_v1",
            "applicable": False,
            "verdict": "not_applicable",
            "score": 0,
            "gates": [],
            "facts": {},
            "blockers": [],
        }

    blockers: list[str] = []
    gates: list[dict[str, Any]] = []

    def gate(name: str, status: str, detail: str | None = None) -> None:
        gates.append({"name": name, "status": status, "detail": detail})
        if status in {"failed", "missing"}:
            blockers.append(name)

    evidence_attempted = sum(int(row.get("sensors_attempted") or 0) for row in evidence_rows)
    evidence_succeeded = sum(int(row.get("sensors_succeeded") or 0) for row in evidence_rows)
    gateway_total = int(gateway_mcp_summary.get("total") or 0)
    legacy_total = int(legacy_mcp_summary.get("total") or 0)
    approval_statuses = {str(row.get("status") or "").upper() for row in approvals}
    approval_actions = " ".join(str(row.get("action") or "") for row in approvals).upper()
    approval_no_action = _approval_has_no_action(approvals)
    approval_suppressed = _approval_suppresses_repair_execution(approvals)
    effective_ops = (
        []
        if approval_suppressed
        else _effective_execution_ops(automation_ops)
    )
    noop_ops = [row for row in automation_ops if _is_no_action_operation(row)]
    audit_only_ops = [row for row in automation_ops if _is_audit_only_operation(row)]
    automation_statuses = {str(row.get("status") or "").lower() for row in effective_ops}
    auto_repair_successes = {row.get("success") for row in auto_repair_executions}
    has_execution = bool(effective_ops or auto_repair_executions)
    verification_result = _latest_verification_result(incident, evidence_rows)

    gate("source_persisted", "passed", str(incident.get("incident_id")))
    gate("outbound_recorded", "passed" if outbound_rows else "missing", str(len(outbound_rows)))
    if not evidence_rows:
        gate("evidence_collected", "missing", "no incident_evidence rows")
    elif evidence_attempted > 0 and evidence_succeeded == 0:
        gate("evidence_collected", "failed", f"{evidence_succeeded}/{evidence_attempted}")
    elif evidence_succeeded > 0:
        gate("evidence_collected", "passed", f"{evidence_succeeded}/{evidence_attempted}")
    else:
        gate("evidence_collected", "warning", f"{evidence_succeeded}/{evidence_attempted}")

    if gateway_total > 0:
        gate("mcp_gateway_observed", "passed", str(gateway_total))
    elif legacy_total > 0:
        gate("mcp_gateway_observed", "warning", f"legacy_only={legacy_total}")
    else:
        gate("mcp_gateway_observed", "missing", "no mcp audit")

    if any(status in {"PENDING", "WAITING_APPROVAL"} for status in approval_statuses):
        gate("approval_state", "warning", "waiting_approval")
    elif "EXPIRED" in approval_statuses and not has_execution:
        gate("approval_state", "warning", "expired_without_execution")
    elif "REJECTED" in approval_statuses and not has_execution:
        gate("approval_state", "passed", "rejected_no_execution")
    elif approval_statuses and approval_suppressed and not has_execution:
        gate("approval_state", "warning", "approved_diagnostic_or_observe_only")
    elif approval_statuses and (approval_no_action or "NO_ACTION" in approval_actions) and not has_execution:
        gate("approval_state", "failed", "approved_no_action_without_execution")
    elif approvals:
        gate("approval_state", "passed", ",".join(sorted(approval_statuses)))
    else:
        gate("approval_state", "not_applicable", "no approval")

    gate("execution_recorded", "passed" if has_execution else "missing", str(len(effective_ops) + len(auto_repair_executions)))
    gate("auto_repair_recorded", "passed" if auto_repair_executions else "missing", str(len(auto_repair_executions)))

    verification_status = str(verification_result or "").lower()

    if not has_execution:
        gate("verification_recorded", "not_applicable", "no execution")
    elif verification_status == "success":
        gate("verification_recorded", "passed", verification_result)
    elif verification_result:
        gate("verification_recorded", "warning", verification_result)
    else:
        gate("verification_recorded", "missing", "execution without verification_result")

    if not has_execution:
        gate("learning_recorded", "not_applicable", "no execution")
    elif km_entries:
        gate("learning_recorded", "passed", str(len(km_entries)))
    else:
        gate("learning_recorded", "missing", "execution without KM entry")

    gate("timeline_recorded", "passed" if timeline_events else "missing", str(len(timeline_events)))

    if has_execution and (
        False in auto_repair_successes or automation_statuses & {"failed", "error"}
    ):
        verdict = "execution_failed"
    elif has_execution and verification_status == "success":
        verdict = "auto_repaired_verified"
    elif has_execution and verification_result:
        verdict = "auto_repaired_verification_degraded"
    elif has_execution:
        verdict = "execution_unverified"
    elif "EXPIRED" in approval_statuses:
        verdict = "approval_expired_manual_review"
    elif "REJECTED" in approval_statuses:
        verdict = "approval_rejected_no_execution"
    elif approval_suppressed:
        verdict = "manual_required_diagnostic_only"
    elif approval_statuses and (approval_no_action or "NO_ACTION" in approval_actions):
        verdict = "manual_required_no_action"
    elif any(status in {"PENDING", "WAITING_APPROVAL"} for status in approval_statuses):
        verdict = "approval_required"
    elif evidence_rows or gateway_total or legacy_total:
        verdict = "observed_not_executed"
    else:
        verdict = "received_only"

    score_weights = {
        "source_persisted": 10,
        "outbound_recorded": 10,
        "evidence_collected": 15,
        "mcp_gateway_observed": 15,
        "approval_state": 10,
        "execution_recorded": 15,
        "auto_repair_recorded": 10,
        "verification_recorded": 10,
        "learning_recorded": 3,
        "timeline_recorded": 2,
    }
    score = 0
    for row in gates:
        weight = score_weights.get(str(row["name"]), 0)
        if row["status"] == "passed":
            score += weight
        elif row["status"] == "not_applicable" and row["name"] == "approval_state":
            score += weight
        elif row["status"] == "warning":
            score += weight // 2

    return {
        "schema_version": "automation_quality_v1",
        "applicable": True,
        "verdict": verdict,
        "score": score,
        "gates": gates,
        "facts": {
            "incident_id": incident.get("incident_id"),
            "evidence_records": len(evidence_rows),
            "sensors_attempted": evidence_attempted,
            "sensors_succeeded": evidence_succeeded,
            "mcp_gateway_total": gateway_total,
            "legacy_mcp_total": legacy_total,
            "approvals": len(approvals),
            "automation_operation_records": len(automation_ops),
            "effective_execution_records": len(effective_ops),
            "noop_operation_records": len(noop_ops),
            "audit_only_operation_records": len(audit_only_ops),
            "approval_repair_suppressed": approval_suppressed,
            "auto_repair_execution_records": len(auto_repair_executions),
            "verification_result": verification_result,
            "knowledge_entries": len(km_entries),
            "timeline_events": len(timeline_events),
            "outbound_messages": len(outbound_rows),
        },
        "blockers": blockers,
    }


def _automation_quality_score_bucket(score: int) -> str:
    if score >= 85:
        return "green"
    if score >= 60:
        return "yellow"
    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)
    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)
        terminal_check_mode_parent_ids = {
            str(row.get("parent_op_id"))
            for row in ansible_records
            if isinstance(row, dict)
            and str(row.get("operation_type") or "") in {
                "ansible_check_mode_executed",
                "ansible_execution_skipped",
            }
            and row.get("parent_op_id")
        }

        for row in ansible_records:
            if not isinstance(row, dict):
                continue
            operation_type = str(row.get("operation_type") or "")
            status = str(row.get("status") or "").lower()
            if operation_type == "ansible_check_mode_executed" and status != "pending":
                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"
                and str(row.get("op_id")) not in terminal_check_mode_parent_ids
            ):
                summary["ansible_pending_check_mode_total"] += 1

    return summary


def _ansible_observed_runtime_blockers(records: list[dict[str, Any]]) -> list[str]:
    blockers: set[str] = set()
    for record in records:
        execution = record.get("execution") if isinstance(record.get("execution"), dict) else {}
        ansible = execution.get("ansible") if isinstance(execution.get("ansible"), dict) else {}
        ansible_records = ansible.get("records") if isinstance(ansible.get("records"), list) else []
        for row in ansible_records:
            if not isinstance(row, dict):
                continue
            if str(row.get("operation_type") or "") != "ansible_check_mode_executed":
                continue
            dry_run_result = row.get("dry_run_result") if isinstance(row.get("dry_run_result"), dict) else {}
            blockers.update(
                detect_ansible_transport_blockers(
                    row.get("error"),
                    dry_run_result.get("stdout_tail"),
                    dry_run_result.get("stderr_tail"),
                )
            )
    return sorted(blockers)


def _ansible_playbook_roots(module_path: Path | None = None) -> list[Path]:
    resolved_module_path = (module_path or Path(__file__)).resolve()
    return [
        Path("/app/infra/ansible"),
        Path.cwd() / "infra" / "ansible",
        *(parent / "infra" / "ansible" for parent in resolved_module_path.parents),
    ]


def _path_readable(path: Path) -> bool:
    return path.exists() and path.is_file() and os.access(path, os.R_OK)


def _check_mode_uses_repair_forced_command_transport() -> bool:
    return Path(settings.AWOOOP_ANSIBLE_CHECK_MODE_SSH_KEY_PATH) == Path("/etc/repair-ssh/id_ed25519")


def _ansible_runtime_readiness(
    *,
    check_mode_ssh_key_path: Path | None = None,
    check_mode_known_hosts_path: Path | None = None,
    repair_ssh_key_path: Path = Path("/etc/repair-ssh/id_ed25519"),
    repair_known_hosts_path: Path = Path("/etc/repair-known-hosts/known_hosts"),
) -> dict[str, Any]:
    playbook_roots = _ansible_playbook_roots()
    playbook_root = next((path for path in playbook_roots if path.exists()), None)
    playbook_paths = (
        sorted((playbook_root / "playbooks").glob("*.yml"))
        if playbook_root is not None and (playbook_root / "playbooks").exists()
        else []
    )
    inventory_path = playbook_root / "inventory" / "hosts.yml" if playbook_root is not None else None
    binary_path = shutil.which("ansible-playbook")
    selected_ssh_key_path = check_mode_ssh_key_path or Path(settings.AWOOOP_ANSIBLE_CHECK_MODE_SSH_KEY_PATH)
    selected_known_hosts_path = check_mode_known_hosts_path or Path(
        settings.AWOOOP_ANSIBLE_CHECK_MODE_KNOWN_HOSTS_PATH
    )
    check_mode_ssh_key_readable = _path_readable(selected_ssh_key_path)
    check_mode_known_hosts_readable = _path_readable(selected_known_hosts_path)
    repair_ssh_key_readable = _path_readable(repair_ssh_key_path)
    repair_known_hosts_readable = _path_readable(repair_known_hosts_path)
    blockers: list[str] = []
    if not binary_path:
        blockers.append("ansible_playbook_binary_missing")
    if playbook_root is None:
        blockers.append("ansible_playbook_catalog_missing")
    if inventory_path is None or not inventory_path.exists():
        blockers.append("ansible_inventory_missing")
    if not playbook_paths:
        blockers.append("ansible_playbooks_missing")
    if not check_mode_ssh_key_readable:
        blockers.append("ansible_check_mode_ssh_key_missing")
    if not check_mode_known_hosts_readable:
        blockers.append("ansible_check_mode_known_hosts_missing")

    return {
        "ansible_playbook_binary_present": bool(binary_path),
        "ansible_playbook_binary_path": binary_path,
        "playbook_root_present": playbook_root is not None,
        "playbook_root": str(playbook_root) if playbook_root is not None else None,
        "inventory_present": bool(inventory_path and inventory_path.exists()),
        "playbook_count": len(playbook_paths),
        "check_mode_transport_profile": settings.AWOOOP_ANSIBLE_CHECK_MODE_TRANSPORT_PROFILE,
        "check_mode_ssh_key_present": selected_ssh_key_path.exists(),
        "check_mode_ssh_key_readable": check_mode_ssh_key_readable,
        "check_mode_ssh_key_path": str(selected_ssh_key_path),
        "check_mode_known_hosts_present": selected_known_hosts_path.exists(),
        "check_mode_known_hosts_readable": check_mode_known_hosts_readable,
        "check_mode_known_hosts_path": str(selected_known_hosts_path),
        "check_mode_candidate_max_age_hours": settings.AWOOOP_ANSIBLE_CHECK_MODE_CANDIDATE_MAX_AGE_HOURS,
        "repair_ssh_key_present": repair_ssh_key_path.exists(),
        "repair_ssh_key_readable": repair_ssh_key_readable,
        "repair_ssh_key_path": str(repair_ssh_key_path),
        "repair_known_hosts_present": repair_known_hosts_path.exists(),
        "repair_known_hosts_readable": repair_known_hosts_readable,
        "repair_known_hosts_path": str(repair_known_hosts_path),
        "can_run_check_mode": not blockers,
        "blockers": blockers,
    }


def summarize_automation_quality_records(
    *,
    project_id: str,
    window_hours: int,
    records: list[dict[str, Any]],
    limit: int,
) -> dict[str, Any]:
    """Aggregate per-incident automation quality into an operator summary."""
    verdicts: dict[str, dict[str, Any]] = {}
    gate_failures: dict[str, dict[str, Any]] = {}
    score_buckets: dict[str, int] = {"green": 0, "yellow": 0, "red": 0}
    examples: list[dict[str, Any]] = []
    total_score = 0
    evaluated_total = 0
    verified_total = 0

    for record in records:
        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 {}
        quality = record.get("automation_quality") if isinstance(record.get("automation_quality"), dict) else {}
        if quality.get("applicable") is not True:
            continue

        evaluated_total += 1
        score = int(quality.get("score") or 0)
        total_score += score
        bucket = _automation_quality_score_bucket(score)
        score_buckets[bucket] += 1

        verdict = str(quality.get("verdict") or "unknown")
        if verdict == "auto_repaired_verified":
            verified_total += 1
        verdict_row = verdicts.setdefault(
            verdict,
            {
                "verdict": verdict,
                "total": 0,
                "score_sum": 0,
                "min_score": score,
                "max_score": score,
                "needs_human": False,
            },
        )
        verdict_row["total"] += 1
        verdict_row["score_sum"] += score
        verdict_row["min_score"] = min(int(verdict_row["min_score"]), score)
        verdict_row["max_score"] = max(int(verdict_row["max_score"]), score)
        verdict_row["needs_human"] = bool(
            verdict_row["needs_human"] or truth_status.get("needs_human")
        )

        for gate in quality.get("gates") or []:
            if not isinstance(gate, dict):
                continue
            gate_status = str(gate.get("status") or "")
            if gate_status not in {"failed", "missing"}:
                continue
            gate_name = str(gate.get("name") or "unknown")
            gate_row = gate_failures.setdefault(
                gate_name,
                {"gate": gate_name, "total": 0, "statuses": {}},
            )
            gate_row["total"] += 1
            gate_row["statuses"][gate_status] = int(gate_row["statuses"].get(gate_status, 0)) + 1

        examples.append({
            "incident_id": incident.get("incident_id"),
            "alertname": incident.get("alertname"),
            "severity": incident.get("severity"),
            "status": incident.get("status"),
            "created_at": incident.get("created_at"),
            "truth_stage": truth_status.get("current_stage"),
            "truth_stage_status": truth_status.get("stage_status"),
            "needs_human": bool(truth_status.get("needs_human")),
            "verdict": verdict,
            "score": score,
            "score_bucket": bucket,
            "blockers": list(quality.get("blockers") or [])[:8],
        })

    by_verdict = []
    for row in verdicts.values():
        total = int(row["total"])
        row["avg_score"] = round(float(row.pop("score_sum")) / total, 1) if total else 0.0
        by_verdict.append(row)
    by_verdict.sort(key=lambda row: (-int(row["total"]), str(row["verdict"])))

    failing_gates = sorted(
        gate_failures.values(),
        key=lambda row: (-int(row["total"]), str(row["gate"])),
    )

    ansible_runtime = _ansible_runtime_readiness()
    observed_ansible_blockers = _ansible_observed_runtime_blockers(records)
    if observed_ansible_blockers:
        ansible_runtime["observed_transport_blockers"] = observed_ansible_blockers
        if _check_mode_uses_repair_forced_command_transport():
            ansible_runtime["blockers"] = sorted(
                set(ansible_runtime.get("blockers") or []) | set(observed_ansible_blockers)
            )
            ansible_runtime["can_run_check_mode"] = False
        else:
            ansible_runtime["historical_transport_blockers"] = observed_ansible_blockers

    return {
        "schema_version": "automation_quality_summary_v1",
        "project_id": project_id,
        "window_hours": window_hours,
        "limit": limit,
        "incident_total": len(records),
        "evaluated_total": evaluated_total,
        "verified_auto_repair_total": verified_total,
        "average_score": round(total_score / evaluated_total, 1) if evaluated_total else 0.0,
        "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],
        "production_claim": {
            "can_claim_full_auto_repair": evaluated_total > 0 and verified_total == evaluated_total,
            "reason": (
                "all_evaluated_incidents_auto_repaired_verified"
                if evaluated_total > 0 and verified_total == evaluated_total
                else "some_incidents_are_not_auto_repaired_verified"
            ),
        },
    }


def _summarize_mcp(rows: list[dict[str, Any]]) -> dict[str, Any]:
    by_tool: dict[str, dict[str, Any]] = {}
    success_count = 0
    failure_count = 0
    for row in rows:
        key = f"{row.get('mcp_server') or 'unknown'}:{row.get('tool_name') or 'unknown'}"
        item = by_tool.setdefault(
            key,
            {
                "mcp_server": row.get("mcp_server"),
                "tool_name": row.get("tool_name"),
                "success": 0,
                "failed": 0,
                "last_error": None,
            },
        )
        if row.get("success") is True:
            item["success"] += 1
            success_count += 1
        else:
            item["failed"] += 1
            failure_count += 1
            item["last_error"] = row.get("error_message")
    return {
        "total": len(rows),
        "success": success_count,
        "failed": failure_count,
        "by_tool": list(by_tool.values()),
    }


def _as_bool(value: Any) -> bool:
    if isinstance(value, bool):
        return value
    return str(value).lower() == "true"


def _counter_bucket(
    buckets: dict[str, dict[str, Any]],
    key: str,
    *,
    label: str,
) -> dict[str, Any]:
    return buckets.setdefault(
        key,
        {
            label: key,
            "total": 0,
            "success": 0,
            "failed": 0,
            "blocked": 0,
        },
    )


def _summarize_gateway_mcp(rows: list[dict[str, Any]]) -> dict[str, Any]:
    by_agent: dict[str, dict[str, Any]] = {}
    by_tool: dict[str, dict[str, Any]] = {}
    by_scope: dict[str, dict[str, Any]] = {}
    success_count = 0
    failed_count = 0
    blocked_count = 0
    first_class_count = 0
    bridge_count = 0
    policy_enforced_count = 0
    approval_executor_count = 0

    for row in rows:
        gate_result = row.get("gate_result") if isinstance(row.get("gate_result"), dict) else {}
        gateway_path = str(gate_result.get("gateway_path") or "")
        schema_version = str(gate_result.get("schema_version") or "")
        policy_enforced = _as_bool(gate_result.get("policy_enforced"))
        required_scope = str(gate_result.get("required_scope") or "unknown")
        status = str(row.get("result_status") or "unknown").lower()
        is_blocked = status == "blocked" or row.get("block_gate") is not None
        is_success = status == "success"
        is_failed = status == "failed"
        agent_id = str(row.get("agent_id") or "unknown")
        tool_name = str(row.get("tool_name") or "unknown")

        if gateway_path == "awooop_mcp_gateway" and policy_enforced:
            first_class_count += 1
        if schema_version == "legacy_mcp_bridge_v1" or policy_enforced is False:
            bridge_count += 1
        if policy_enforced:
            policy_enforced_count += 1
        if agent_id == "approval_executor":
            approval_executor_count += 1
        if is_success:
            success_count += 1
        if is_failed:
            failed_count += 1
        if is_blocked:
            blocked_count += 1

        for bucket in (
            _counter_bucket(by_agent, agent_id, label="agent_id"),
            _counter_bucket(by_tool, tool_name, label="tool_name"),
            _counter_bucket(by_scope, required_scope, label="required_scope"),
        ):
            bucket["total"] += 1
            if is_success:
                bucket["success"] += 1
            if is_failed:
                bucket["failed"] += 1
            if is_blocked:
                bucket["blocked"] += 1

    blockers: list[str] = []
    if blocked_count:
        stage = "gateway_blocked"
        stage_status = "blocked"
        blockers.append("mcp_gateway_blocked")
    elif first_class_count and failed_count:
        stage = "provider_failed_after_gateway"
        stage_status = "failed"
        blockers.append("provider_failed_after_gateway")
    elif success_count:
        stage = "gateway_execution_succeeded"
        stage_status = "success"
    elif bridge_count and not first_class_count:
        stage = "legacy_bridge_only"
        stage_status = "observed"
        blockers.append("legacy_bridge_not_policy_enforced")
    else:
        stage = "no_gateway_records"
        stage_status = "missing"

    return {
        "total": len(rows),
        "success": success_count,
        "failed": failed_count,
        "blocked": blocked_count,
        "first_class_total": first_class_count,
        "legacy_bridge_total": bridge_count,
        "policy_enforced_total": policy_enforced_count,
        "approval_executor_total": approval_executor_count,
        "stage": stage,
        "stage_status": stage_status,
        "needs_human": bool(blocked_count or (first_class_count and failed_count)),
        "blockers": blockers,
        "by_agent": list(by_agent.values()),
        "by_tool": list(by_tool.values()),
        "by_scope": list(by_scope.values()),
    }


async def fetch_truth_chain(source_id: str, project_id: str = "awoooi") -> dict[str, Any]:
    """Return a read-only truth chain for an incident, drift report, or run id."""
    async with get_db_context(project_id) as db:
        incident = await _fetch_one(
            db,
            """
            SELECT
                incident_id,
                project_id,
                status::text AS status,
                severity::text AS severity,
                alertname,
                alert_category,
                notification_type,
                created_at,
                updated_at,
                resolved_at,
                verification_result,
                frequency_snapshot,
                signals,
                decision_chain
            FROM incidents
            WHERE incident_id = :source_id
              AND (project_id = :project_id OR project_id IS NULL)
            """,
            {"source_id": source_id, "project_id": project_id},
        )

        drift = await _fetch_one(
            db,
            """
            SELECT
                report_id,
                namespace,
                status,
                triggered_by,
                high_count,
                medium_count,
                info_count,
                scanned_at,
                created_at,
                resolved_at,
                interpretation,
                items,
                narrative_text
            FROM drift_reports
            WHERE report_id = :source_id
            """,
            {"source_id": source_id},
        )

        runs = await _fetch_all(
            db,
            """
            SELECT
                run_id,
                project_id,
                agent_id,
                state,
                trigger_type,
                trigger_ref,
                is_shadow,
                step_count,
                created_at,
                started_at,
                completed_at,
                error_code,
                error_detail
            FROM awooop_run_state
            WHERE project_id = :project_id
              AND (run_id::text = :source_id OR trigger_ref = :source_id)
            ORDER BY created_at DESC
            LIMIT :limit
            """,
            {"source_id": source_id, "project_id": project_id, "limit": _MAX_ROWS},
        )

        approvals: list[dict[str, Any]] = []
        evidence_rows: list[dict[str, Any]] = []
        timeline_events: list[dict[str, Any]] = []
        legacy_mcp_rows: list[dict[str, Any]] = []
        automation_ops: list[dict[str, Any]] = []
        auto_repair_executions: list[dict[str, Any]] = []
        km_entries: list[dict[str, Any]] = []
        inbound_rows: list[dict[str, Any]] = []
        gateway_mcp_rows: list[dict[str, Any]] = []
        outbound_rows: list[dict[str, Any]] = []

        if incident is not None:
            incident_id = str(incident["incident_id"])
            approvals = await _fetch_all(
                db,
                """
                SELECT
                    id,
                    incident_id,
                    status::text AS status,
                    risk_level::text AS risk_level,
                    action,
                    description,
                    hit_count,
                    created_at,
                    updated_at,
                    resolved_at,
                    matched_playbook_id,
                    extra_metadata,
                    decision_fusion_details
                FROM approval_records
                WHERE incident_id = :incident_id
                ORDER BY created_at DESC
                LIMIT :limit
                """,
                {"incident_id": incident_id, "limit": _MAX_ROWS},
            )
            approval_ids = _approval_ids(approvals)
            evidence_rows = await _fetch_all(
                db,
                """
                SELECT
                    id,
                    incident_id,
                    matched_playbook_id,
                    collected_at,
                    collection_duration_ms,
                    sensors_attempted,
                    sensors_succeeded,
                    evidence_summary,
                    metrics_snapshot,
                    mcp_health,
                    verification_result,
                    pre_execution_state,
                    post_execution_state,
                    self_healing_score,
                    self_healing_detail
                FROM incident_evidence
                WHERE incident_id = :incident_id
                ORDER BY collected_at DESC
                LIMIT :limit
                """,
                {"incident_id": incident_id, "limit": _MAX_ROWS},
            )
            timeline_events = await _fetch_all(
                db,
                """
                SELECT
                    id,
                    event_type,
                    status,
                    title,
                    description,
                    actor,
                    actor_role,
                    risk_level,
                    approval_id,
                    incident_id,
                    created_at
                FROM timeline_events
                WHERE incident_id = :incident_id
                   OR approval_id = ANY(:approval_ids)
                ORDER BY created_at ASC
                LIMIT :limit
                """,
                {
                    "incident_id": incident_id,
                    "approval_ids": approval_ids or ["__none__"],
                    "limit": _MAX_ROWS,
                },
            )
            legacy_mcp_rows = await _fetch_all(
                db,
                """
                SELECT
                    id,
                    session_id,
                    flywheel_node,
                    mcp_server,
                    tool_name,
                    duration_ms,
                    success,
                    error_message,
                    incident_id,
                    agent_role,
                    created_at
                FROM mcp_audit_log
                WHERE incident_id = :incident_id
                ORDER BY created_at ASC
                LIMIT :limit
                """,
                {"incident_id": incident_id, "limit": _MAX_ROWS},
            )
            automation_ops = await _fetch_all(
                db,
                """
                SELECT
                    op_id,
                    operation_type,
                    status,
                    incident_id,
                    run_id,
                    parent_op_id,
                    actor,
                    dry_run_result,
                    error,
                    duration_ms,
                    tags,
                    input ->> 'action' AS input_action,
                    input ->> 'executor' AS input_executor,
                    input ->> 'execution_backend' AS input_execution_backend,
                    input ->> 'catalog_id' AS input_catalog_id,
                    input ->> 'execution_mode' AS input_execution_mode,
                    input ->> 'approval_source' AS input_approval_source,
                    input ->> 'apply_enabled' AS input_apply_enabled,
                    input ->> 'apply_executed' AS input_apply_executed,
                    input ->> 'check_mode_executed' AS input_check_mode_executed,
                    input ->> 'returncode' AS input_returncode,
                    input ->> 'playbook_id' AS input_playbook_id,
                    input ->> 'playbook_path' AS input_playbook_path,
                    input ->> 'ansible_playbook_path' AS input_ansible_playbook_path,
                    input ->> 'check_mode' AS input_check_mode,
                    input ->> 'not_used_reason' AS input_not_used_reason,
                    output ->> 'action' AS output_action,
                    output ->> 'reason' AS output_reason,
                    output ->> 'executor' AS output_executor,
                    output ->> 'execution_backend' AS output_execution_backend,
                    output ->> 'catalog_id' AS output_catalog_id,
                    output ->> 'execution_mode' AS output_execution_mode,
                    output ->> 'approval_source' AS output_approval_source,
                    output ->> 'apply_enabled' AS output_apply_enabled,
                    output ->> 'apply_executed' AS output_apply_executed,
                    output ->> 'check_mode_executed' AS output_check_mode_executed,
                    output ->> 'returncode' AS output_returncode,
                    output ->> 'playbook_id' AS output_playbook_id,
                    output ->> 'playbook_path' AS output_playbook_path,
                    output ->> 'ansible_playbook_path' AS output_ansible_playbook_path,
                    output ->> 'check_mode' AS output_check_mode,
                    output ->> 'not_used_reason' AS output_not_used_reason,
                    dry_run_result ->> 'returncode' AS dry_run_returncode,
                    dry_run_result ->> 'apply_executed' AS dry_run_apply_executed,
                    dry_run_result ->> 'check_mode_executed' AS dry_run_check_mode_executed,
                    created_at
                FROM automation_operation_log
                WHERE incident_id::text = :incident_id
                   OR coalesce(input::text, '') LIKE :needle
                   OR coalesce(output::text, '') LIKE :needle
                   OR coalesce(array_to_string(tags, ','), '') LIKE :needle
                ORDER BY created_at DESC
                LIMIT :limit
                """,
                {"incident_id": incident_id, "needle": f"%{incident_id}%", "limit": _MAX_ROWS},
            )
            auto_repair_executions = await _fetch_all(
                db,
                """
                SELECT
                    id,
                    incident_id,
                    playbook_id,
                    playbook_name,
                    success,
                    executed_steps,
                    error_message,
                    triggered_by,
                    similarity_score,
                    risk_level,
                    execution_time_ms,
                    created_at
                FROM auto_repair_executions
                WHERE incident_id = :incident_id
                ORDER BY created_at DESC
                LIMIT :limit
                """,
                {"incident_id": incident_id, "limit": _MAX_ROWS},
            )
            km_entries = await _fetch_all(
                db,
                """
                SELECT
                    id,
                    title,
                    entry_type::text AS entry_type,
                    status::text AS status,
                    related_incident_id,
                    related_approval_id,
                    created_at
                FROM knowledge_entries
                WHERE related_incident_id = :incident_id
                ORDER BY created_at DESC
                LIMIT :limit
                """,
                {"incident_id": incident_id, "limit": _MAX_ROWS},
            )

        drift_repeats: dict[str, Any] = {
            "occurrences_12h": 0,
            "first_scanned_at": None,
            "last_scanned_at": None,
            "reports": [],
        }
        if drift is not None:
            recent_drift_reports = await _fetch_all(
                db,
                """
                SELECT
                    report_id,
                    namespace,
                    status,
                    scanned_at,
                    created_at,
                    items,
                    interpretation,
                    narrative_text
                FROM drift_reports
                WHERE created_at > now() - interval '24 hours'
                  AND namespace = :namespace
                ORDER BY scanned_at DESC
                LIMIT 200
                """,
                {"namespace": drift["namespace"]},
            )
            drift_repeats = build_drift_repeat_state(drift, recent_drift_reports)

        gateway_mcp_rows = await _fetch_all(
            db,
            """
            SELECT
                call_id,
                project_id,
                run_id,
                trace_id,
                agent_id,
                tool_name,
                gate_result,
                result_status,
                block_gate,
                block_reason,
                latency_ms,
                created_at
            FROM awooop_mcp_gateway_audit
            WHERE project_id = :project_id
              AND (
                trace_id = :source_id
                OR run_id::text = :source_id
              )
            ORDER BY created_at ASC
            LIMIT :limit
            """,
            {"source_id": source_id, "project_id": project_id, "limit": _MAX_ROWS},
        )
        incident_fingerprints = _incident_fingerprints(incident)
        fingerprint_needle = (
            f"%{incident_fingerprints[0]}%"
            if incident_fingerprints
            else ""
        )
        fingerprint_value = incident_fingerprints[0] if incident_fingerprints else ""
        inbound_rows = await _fetch_all(
            db,
            """
            SELECT
                event_id,
                project_id,
                channel_type,
                provider_event_id,
                platform_subject_id,
                channel_user_id,
                channel_chat_id,
                run_id,
                content_type,
                content_hash,
                content_preview,
                content_redacted,
                redaction_version,
                source_envelope,
                attachment_sha256,
                is_duplicate,
                provider_ts,
                received_at
            FROM awooop_conversation_event
            WHERE project_id = :project_id
              AND (
                run_id::text = :source_id
                OR provider_event_id = :source_id
                OR content_preview ILIKE :source_needle
                OR coalesce(source_envelope #> '{source_refs,event_ids}', '[]'::jsonb) ? :source_id
                OR coalesce(source_envelope #> '{source_refs,incident_ids}', '[]'::jsonb) ? :source_id
                OR coalesce(source_envelope #> '{source_refs,approval_ids}', '[]'::jsonb) ? :source_id
                OR coalesce(source_envelope #> '{source_refs,alert_ids}', '[]'::jsonb) ? :source_id
                OR coalesce(source_envelope #> '{source_refs,sentry_issue_ids}', '[]'::jsonb) ? :source_id
                OR coalesce(source_envelope #> '{source_refs,signoz_alerts}', '[]'::jsonb) ? :source_id
                OR (
                  :fingerprint_needle != ''
                  AND (
                    provider_event_id ILIKE :fingerprint_needle
                    OR content_preview ILIKE :fingerprint_needle
                    OR coalesce(source_envelope #> '{source_refs,fingerprints}', '[]'::jsonb) ? :fingerprint_value
                  )
                )
              )
            ORDER BY received_at DESC
            LIMIT :limit
            """,
            {
                "source_id": source_id,
                "project_id": project_id,
                "source_needle": f"%{source_id}%",
                "fingerprint_needle": fingerprint_needle,
                "fingerprint_value": fingerprint_value,
                "limit": _MAX_ROWS,
            },
        )
        outbound_rows = await _fetch_all(
            db,
            """
            SELECT
                message_id,
                project_id,
                run_id,
                channel_type,
                message_type,
                content_hash,
                content_preview,
                content_redacted,
                redaction_version,
                source_envelope,
                provider_message_id,
                send_status,
                queued_at,
                sent_at,
                triggered_by_state
            FROM awooop_outbound_message
            WHERE project_id = :project_id
              AND (
                run_id::text = :source_id
                OR content_preview ILIKE :needle
                OR coalesce(source_envelope #> '{source_refs,incident_ids}', '[]'::jsonb) ? :source_id
                OR coalesce(source_envelope #> '{source_refs,code_refs}', '[]'::jsonb) ? :source_id
              )
            ORDER BY queued_at DESC
            LIMIT :limit
            """,
            {
                "source_id": source_id,
                "project_id": project_id,
                "needle": f"%{source_id}%",
                "limit": _MAX_ROWS,
            },
        )

    source_type = _source_type(source_id, incident, drift)
    legacy_mcp_summary = _summarize_mcp(legacy_mcp_rows)
    gateway_mcp_summary = _summarize_gateway_mcp(gateway_mcp_rows)
    truth_status = _truth_status(
        incident=incident,
        approvals=approvals,
        evidence_rows=evidence_rows,
        automation_ops=automation_ops,
        drift=drift,
        drift_repeat_count=int(drift_repeats.get("occurrences_12h") or 0),
        gateway_mcp_total=len(gateway_mcp_rows),
        legacy_mcp_total=legacy_mcp_summary["total"],
        outbound_visible_total=len(outbound_rows),
        inbound_visible_total=len(inbound_rows),
        auto_repair_executions=auto_repair_executions,
    )
    if incident is None and drift is None and not runs and gateway_mcp_rows:
        truth_status = {
            "current_stage": gateway_mcp_summary["stage"],
            "stage_status": gateway_mcp_summary["stage_status"],
            "needs_human": gateway_mcp_summary["needs_human"],
            "blockers": gateway_mcp_summary["blockers"],
        }
    reconciliation = build_incident_reconciliation(
        incident=incident,
        approvals=approvals,
        evidence_rows=evidence_rows,
        automation_ops=automation_ops,
        auto_repair_executions=auto_repair_executions,
        timeline_events=timeline_events,
    )

    evidence_totals = {
        "records": len(evidence_rows),
        "sensors_attempted": sum(int(row.get("sensors_attempted") or 0) for row in evidence_rows),
        "sensors_succeeded": sum(int(row.get("sensors_succeeded") or 0) for row in evidence_rows),
        "failed_tools": _failed_mcp_tools(evidence_rows),
    }
    automation_quality = build_automation_quality(
        incident=incident,
        approvals=approvals,
        evidence_rows=evidence_rows,
        automation_ops=automation_ops,
        auto_repair_executions=auto_repair_executions,
        gateway_mcp_summary=gateway_mcp_summary,
        legacy_mcp_summary=legacy_mcp_summary,
        outbound_rows=outbound_rows,
        km_entries=km_entries,
        timeline_events=timeline_events,
    )
    automation_quality["operator_outcome"] = build_operator_outcome(
        truth_status=truth_status,
        automation_quality=automation_quality,
        source_id=source_id,
    )

    result = {
        "project_id": project_id,
        "source_id": source_id,
        "source_type": source_type,
        "found": (
            incident is not None
            or drift is not None
            or bool(runs)
            or bool(gateway_mcp_rows)
            or bool(inbound_rows)
            or bool(outbound_rows)
        ),
        "truth_status": truth_status,
        "linked_ids": {
            "incident_id": incident.get("incident_id") if incident else None,
            "approval_ids": _approval_ids(approvals),
            "run_ids": [row["run_id"] for row in runs],
            "drift_report_id": drift.get("report_id") if drift else None,
            "operation_ids": _operation_ids(automation_ops),
            "auto_repair_execution_ids": _auto_repair_ids(auto_repair_executions),
            "conversation_event_ids": [row["event_id"] for row in inbound_rows],
        },
        "incident": incident,
        "drift": {
            "report": drift,
            "repeat_state": drift_repeats,
        },
        "approvals": approvals,
        "evidence": {
            "summary": evidence_totals,
            "records": evidence_rows,
        },
        "mcp": {
            "awooop_gateway": {
                **gateway_mcp_summary,
                "records": gateway_mcp_rows,
            },
            "legacy": {
                **legacy_mcp_summary,
                "records": legacy_mcp_rows,
            },
        },
        "execution": {
            "automation_operation_log": automation_ops,
            "auto_repair_executions": auto_repair_executions,
            "ansible": build_ansible_truth(automation_ops, incident=incident, drift=drift),
        },
        "automation_quality": automation_quality,
        "operator_outcome": automation_quality["operator_outcome"],
        "reconciliation": reconciliation,
        "learning": {
            "knowledge_entries": km_entries,
        },
        "timeline_events": timeline_events,
        "channel": {
            "inbound_events_visible": len(inbound_rows),
            "inbound_events": inbound_rows,
            "outbound_messages_visible": len(outbound_rows),
            "outbound_messages": outbound_rows,
            "visibility_note": (
                "If inbound is zero while Alertmanager fired, or outbound is zero while "
                "Telegram delivered a card, channel mirrors or RLS project context are "
                "not reliable sources of truth yet."
            ),
        },
    }
    logger.info(
        "awooop_truth_chain_fetched",
        project_id=project_id,
        source_id=source_id,
        source_type=source_type,
        current_stage=truth_status["current_stage"],
    )
    return result


def _empty_incident_groups(incident_ids: list[str]) -> dict[str, list[dict[str, Any]]]:
    return {incident_id: [] for incident_id in incident_ids}


def _append_unique_row(group: list[dict[str, Any]], row: dict[str, Any]) -> None:
    if row not in group:
        group.append(row)


def _row_source_refs(row: dict[str, Any]) -> dict[str, Any]:
    envelope = row.get("source_envelope")
    if not isinstance(envelope, dict):
        return {}
    refs = envelope.get("source_refs")
    return refs if isinstance(refs, dict) else {}


def _source_ref_contains(row: dict[str, Any], keys: tuple[str, ...], incident_id: str) -> bool:
    refs = _row_source_refs(row)
    for key in keys:
        value = refs.get(key)
        if isinstance(value, list) and incident_id in {str(item) for item in value}:
            return True
        if str(value or "") == incident_id:
            return True
    return False


def _group_rows_by_incident_reference(
    *,
    rows: list[dict[str, Any]],
    incident_ids: list[str],
    direct_fields: tuple[str, ...],
    text_fields: tuple[str, ...] = (),
    source_ref_keys: tuple[str, ...] = (),
) -> dict[str, list[dict[str, Any]]]:
    """Group batch-fetched rows back to incidents using durable direct/source refs."""
    incident_set = set(incident_ids)
    groups = _empty_incident_groups(incident_ids)
    for row in rows:
        direct_matches = {
            str(row.get(field) or "")
            for field in direct_fields
            if str(row.get(field) or "") in incident_set
        }
        for incident_id in direct_matches:
            groups[incident_id].append(row)
        if direct_matches:
            continue

        text_blob = "\n".join(str(row.get(field) or "") for field in text_fields)
        for incident_id in incident_ids:
            if (
                (text_blob and incident_id in text_blob)
                or (
                    source_ref_keys
                    and _source_ref_contains(row, source_ref_keys, incident_id)
                )
            ):
                _append_unique_row(groups[incident_id], row)
    return groups


def _group_timeline_events_by_incident(
    *,
    rows: list[dict[str, Any]],
    incident_ids: list[str],
    approval_to_incident: dict[str, str],
) -> dict[str, list[dict[str, Any]]]:
    groups = _empty_incident_groups(incident_ids)
    incident_set = set(incident_ids)
    for row in rows:
        incident_id = str(row.get("incident_id") or "")
        if incident_id in incident_set:
            groups[incident_id].append(row)
            continue
        approval_incident_id = approval_to_incident.get(str(row.get("approval_id") or ""))
        if approval_incident_id:
            groups[approval_incident_id].append(row)
    return groups


def _build_summary_quality_records(
    *,
    incidents: list[dict[str, Any]],
    approvals_by_incident: dict[str, list[dict[str, Any]]],
    evidence_by_incident: dict[str, list[dict[str, Any]]],
    timeline_by_incident: dict[str, list[dict[str, Any]]],
    legacy_mcp_by_incident: dict[str, list[dict[str, Any]]],
    automation_ops_by_incident: dict[str, list[dict[str, Any]]],
    auto_repair_by_incident: dict[str, list[dict[str, Any]]],
    km_by_incident: dict[str, list[dict[str, Any]]],
    gateway_mcp_by_incident: dict[str, list[dict[str, Any]]],
    outbound_by_incident: dict[str, list[dict[str, Any]]],
) -> list[dict[str, Any]]:
    records: list[dict[str, Any]] = []
    for incident in incidents:
        incident_id = str(incident.get("incident_id") or "")
        if not incident_id:
            continue
        approvals = approvals_by_incident.get(incident_id, [])
        evidence_rows = evidence_by_incident.get(incident_id, [])
        timeline_events = timeline_by_incident.get(incident_id, [])
        legacy_mcp_rows = legacy_mcp_by_incident.get(incident_id, [])
        automation_ops = automation_ops_by_incident.get(incident_id, [])
        auto_repair_executions = auto_repair_by_incident.get(incident_id, [])
        km_entries = km_by_incident.get(incident_id, [])
        gateway_mcp_rows = gateway_mcp_by_incident.get(incident_id, [])
        outbound_rows = outbound_by_incident.get(incident_id, [])
        legacy_mcp_summary = _summarize_mcp(legacy_mcp_rows)
        gateway_mcp_summary = _summarize_gateway_mcp(gateway_mcp_rows)
        truth_status = _truth_status(
            incident=incident,
            approvals=approvals,
            evidence_rows=evidence_rows,
            automation_ops=automation_ops,
            drift=None,
            drift_repeat_count=0,
            gateway_mcp_total=len(gateway_mcp_rows),
            legacy_mcp_total=legacy_mcp_summary["total"],
            outbound_visible_total=len(outbound_rows),
            inbound_visible_total=0,
            auto_repair_executions=auto_repair_executions,
        )
        automation_quality = build_automation_quality(
            incident=incident,
            approvals=approvals,
            evidence_rows=evidence_rows,
            automation_ops=automation_ops,
            auto_repair_executions=auto_repair_executions,
            gateway_mcp_summary=gateway_mcp_summary,
            legacy_mcp_summary=legacy_mcp_summary,
            outbound_rows=outbound_rows,
            km_entries=km_entries,
            timeline_events=timeline_events,
        )
        automation_quality["operator_outcome"] = build_operator_outcome(
            truth_status=truth_status,
            automation_quality=automation_quality,
            source_id=incident_id,
        )
        records.append({
            "incident": incident,
            "truth_status": truth_status,
            "automation_quality": automation_quality,
            "execution": {
                "automation_operation_log": automation_ops,
                "auto_repair_executions": auto_repair_executions,
                "ansible": build_ansible_truth(automation_ops, incident=incident, drift=None),
            },
        })
    return records


async def fetch_automation_quality_summary(
    *,
    project_id: str = "awoooi",
    hours: int = 24,
    limit: int = 200,
    refresh: bool = False,
) -> dict[str, Any]:
    """Return a recent incident-level quality summary for the automation flywheel."""
    started_at = perf_counter()
    bounded_hours = max(1, min(int(hours), 168))
    bounded_limit = max(1, min(int(limit), 500))
    normalized_project_id = project_id or "awoooi"
    cache_key = {
        "project_id": normalized_project_id,
        "hours": bounded_hours,
        "limit": bounded_limit,
    }
    if not refresh:
        cached_summary = await get_cached_operator_summary_async(
            "truth_chain_quality_summary",
            cache_key,
            ttl_seconds=_QUALITY_SUMMARY_CACHE_TTL_SECONDS,
        )
        if cached_summary is not None:
            duration_seconds = perf_counter() - started_at
            record_quality_summary_observation(
                project_id=normalized_project_id,
                hours=bounded_hours,
                limit=bounded_limit,
                cache_status="hit",
                success=True,
                duration_seconds=duration_seconds,
            )
            logger.info(
                "awooop_automation_quality_summary_cache_hit",
                project_id=normalized_project_id,
                window_hours=bounded_hours,
                limit=bounded_limit,
                ttl_seconds=_QUALITY_SUMMARY_CACHE_TTL_SECONDS,
                duration_seconds=round(duration_seconds, 3),
            )
            cached_summary = dict(cached_summary)
            cached_summary["cache_status"] = "hit"
            cached_summary["aggregation_duration_seconds"] = round(duration_seconds, 3)
            return cached_summary

    cutoff = datetime.now(UTC) - timedelta(hours=bounded_hours)

    async with get_db_context(normalized_project_id) as db:
        incidents = await _fetch_all(
            db,
            """
            WITH source_candidates AS (
                SELECT
                    incident_id::text AS incident_id,
                    created_at AS seen_at
                FROM incidents
                WHERE (project_id = :project_id OR project_id IS NULL)
                  AND created_at >= :cutoff

                UNION ALL

                SELECT
                    coalesce(incident_id::text, input ->> 'incident_id') AS incident_id,
                    created_at AS seen_at
                FROM automation_operation_log
                WHERE created_at >= :cutoff
                  AND operation_type IN (
                    'ansible_candidate_matched',
                    'ansible_check_mode_executed',
                    'ansible_execution_skipped',
                    'ansible_apply_executed',
                    'ansible_rollback_executed'
                  )
                  AND coalesce(incident_id::text, input ->> 'incident_id', '') <> ''
            ),
            source_ids AS (
                SELECT
                    incident_id,
                    max(seen_at) AS recent_evidence_at
                FROM source_candidates
                WHERE incident_id IS NOT NULL AND incident_id <> ''
                GROUP BY incident_id
            )
            SELECT
                incidents.incident_id,
                incidents.project_id,
                incidents.status::text AS status,
                incidents.severity::text AS severity,
                incidents.alertname,
                incidents.alert_category,
                incidents.notification_type,
                incidents.created_at,
                incidents.updated_at,
                incidents.resolved_at,
                incidents.verification_result,
                source_ids.recent_evidence_at
            FROM source_ids
            JOIN incidents ON incidents.incident_id = source_ids.incident_id
            WHERE (incidents.project_id = :project_id OR incidents.project_id IS NULL)
            ORDER BY source_ids.recent_evidence_at DESC
            LIMIT :limit
            """,
            {
                "project_id": normalized_project_id,
                "cutoff": cutoff,
                "limit": bounded_limit,
            },
        )

        incident_ids = [
            str(incident.get("incident_id") or "")
            for incident in incidents
            if incident.get("incident_id")
        ]
        if not incident_ids:
            records = []
        else:
            batch_params = {"incident_ids": incident_ids, "limit": _MAX_ROWS * len(incident_ids)}
            approvals = await _fetch_all(
                db,
                """
                SELECT
                    id,
                    incident_id,
                    status::text AS status,
                    risk_level::text AS risk_level,
                    action,
                    description,
                    hit_count,
                    created_at,
                    updated_at,
                    resolved_at,
                    matched_playbook_id,
                    extra_metadata,
                    decision_fusion_details
                FROM approval_records
                WHERE incident_id = ANY(CAST(:incident_ids AS text[]))
                ORDER BY created_at DESC
                LIMIT :limit
                """,
                batch_params,
            )
            approval_to_incident = {
                str(row.get("id")): str(row.get("incident_id"))
                for row in approvals
                if row.get("id") and row.get("incident_id")
            }
            approval_ids = list(approval_to_incident.keys()) or ["__none__"]
            evidence_rows = await _fetch_all(
                db,
                """
                SELECT
                    id,
                    incident_id,
                    matched_playbook_id,
                    collected_at,
                    collection_duration_ms,
                    sensors_attempted,
                    sensors_succeeded,
                    evidence_summary,
                    metrics_snapshot,
                    mcp_health,
                    verification_result,
                    pre_execution_state,
                    post_execution_state,
                    self_healing_score,
                    self_healing_detail
                FROM incident_evidence
                WHERE incident_id = ANY(CAST(:incident_ids AS text[]))
                ORDER BY collected_at DESC
                LIMIT :limit
                """,
                batch_params,
            )
            timeline_events = await _fetch_all(
                db,
                """
                SELECT
                    id,
                    event_type,
                    status,
                    title,
                    description,
                    actor,
                    actor_role,
                    risk_level,
                    approval_id,
                    incident_id,
                    created_at
                FROM timeline_events
                WHERE incident_id = ANY(CAST(:incident_ids AS text[]))
                   OR approval_id::text = ANY(CAST(:approval_ids AS text[]))
                ORDER BY created_at ASC
                LIMIT :limit
                """,
                {**batch_params, "approval_ids": approval_ids},
            )
            legacy_mcp_rows = await _fetch_all(
                db,
                """
                SELECT
                    id,
                    session_id,
                    flywheel_node,
                    mcp_server,
                    tool_name,
                    duration_ms,
                    success,
                    error_message,
                    incident_id,
                    agent_role,
                    created_at
                FROM mcp_audit_log
                WHERE incident_id = ANY(CAST(:incident_ids AS text[]))
                ORDER BY created_at ASC
                LIMIT :limit
                """,
                batch_params,
            )
            automation_ops = await _fetch_all(
                db,
                """
                SELECT
                    op_id,
                    operation_type,
                    status,
                    incident_id,
                    run_id,
                    parent_op_id,
                    actor,
                    dry_run_result,
                    error,
                    duration_ms,
                    tags,
                    input ->> 'action' AS input_action,
                    input ->> 'executor' AS input_executor,
                    input ->> 'execution_backend' AS input_execution_backend,
                    input ->> 'catalog_id' AS input_catalog_id,
                    input ->> 'execution_mode' AS input_execution_mode,
                    input ->> 'approval_source' AS input_approval_source,
                    input ->> 'apply_enabled' AS input_apply_enabled,
                    input ->> 'apply_executed' AS input_apply_executed,
                    input ->> 'check_mode_executed' AS input_check_mode_executed,
                    input ->> 'returncode' AS input_returncode,
                    input ->> 'playbook_id' AS input_playbook_id,
                    input ->> 'playbook_path' AS input_playbook_path,
                    input ->> 'ansible_playbook_path' AS input_ansible_playbook_path,
                    input ->> 'check_mode' AS input_check_mode,
                    input ->> 'not_used_reason' AS input_not_used_reason,
                    output ->> 'action' AS output_action,
                    output ->> 'reason' AS output_reason,
                    output ->> 'executor' AS output_executor,
                    output ->> 'execution_backend' AS output_execution_backend,
                    output ->> 'catalog_id' AS output_catalog_id,
                    output ->> 'execution_mode' AS output_execution_mode,
                    output ->> 'approval_source' AS output_approval_source,
                    output ->> 'apply_enabled' AS output_apply_enabled,
                    output ->> 'apply_executed' AS output_apply_executed,
                    output ->> 'check_mode_executed' AS output_check_mode_executed,
                    output ->> 'returncode' AS output_returncode,
                    output ->> 'playbook_id' AS output_playbook_id,
                    output ->> 'playbook_path' AS output_playbook_path,
                    output ->> 'ansible_playbook_path' AS output_ansible_playbook_path,
                    output ->> 'check_mode' AS output_check_mode,
                    output ->> 'not_used_reason' AS output_not_used_reason,
                    dry_run_result ->> 'returncode' AS dry_run_returncode,
                    dry_run_result ->> 'apply_executed' AS dry_run_apply_executed,
                    dry_run_result ->> 'check_mode_executed' AS dry_run_check_mode_executed,
                    coalesce(input::text, '') AS input_text,
                    coalesce(output::text, '') AS output_text,
                    coalesce(array_to_string(tags, ','), '') AS tags_text,
                    created_at
                FROM automation_operation_log
                WHERE incident_id::text = ANY(CAST(:incident_ids AS text[]))
                   OR EXISTS (
                     SELECT 1
                     FROM unnest(CAST(:incident_ids AS text[])) AS source_ids(incident_id)
                     WHERE coalesce(input::text, '') LIKE '%' || source_ids.incident_id || '%'
                        OR coalesce(output::text, '') LIKE '%' || source_ids.incident_id || '%'
                        OR coalesce(array_to_string(tags, ','), '') LIKE '%' || source_ids.incident_id || '%'
                   )
                ORDER BY created_at DESC
                LIMIT :limit
                """,
                batch_params,
            )
            auto_repair_executions = await _fetch_all(
                db,
                """
                SELECT
                    id,
                    incident_id,
                    playbook_id,
                    playbook_name,
                    success,
                    executed_steps,
                    error_message,
                    triggered_by,
                    similarity_score,
                    risk_level,
                    execution_time_ms,
                    created_at
                FROM auto_repair_executions
                WHERE incident_id = ANY(CAST(:incident_ids AS text[]))
                ORDER BY created_at DESC
                LIMIT :limit
                """,
                batch_params,
            )
            km_entries = await _fetch_all(
                db,
                """
                SELECT
                    id,
                    title,
                    entry_type::text AS entry_type,
                    status::text AS status,
                    related_incident_id,
                    related_approval_id,
                    created_at
                FROM knowledge_entries
                WHERE related_incident_id = ANY(CAST(:incident_ids AS text[]))
                ORDER BY created_at DESC
                LIMIT :limit
                """,
                batch_params,
            )
            gateway_mcp_rows = await _fetch_all(
                db,
                """
                SELECT
                    call_id,
                    project_id,
                    run_id,
                    trace_id,
                    agent_id,
                    tool_name,
                    gate_result,
                    result_status,
                    block_gate,
                    block_reason,
                    latency_ms,
                    created_at
                FROM awooop_mcp_gateway_audit
                WHERE project_id = :project_id
                  AND (
                    trace_id = ANY(CAST(:incident_ids AS text[]))
                    OR run_id::text = ANY(CAST(:incident_ids AS text[]))
                  )
                ORDER BY created_at ASC
                LIMIT :limit
                """,
                {**batch_params, "project_id": normalized_project_id},
            )
            outbound_rows = await _fetch_all(
                db,
                """
                SELECT
                    message_id,
                    project_id,
                    run_id,
                    channel_type,
                    message_type,
                    content_hash,
                    content_preview,
                    content_redacted,
                    redaction_version,
                    source_envelope,
                    provider_message_id,
                    send_status,
                    queued_at,
                    sent_at,
                    triggered_by_state
                FROM awooop_outbound_message
                WHERE project_id = :project_id
                  AND (
                    run_id::text = ANY(CAST(:incident_ids AS text[]))
                    OR EXISTS (
                      SELECT 1
                      FROM unnest(CAST(:incident_ids AS text[])) AS source_ids(incident_id)
                      WHERE coalesce(content_preview, '') ILIKE '%' || source_ids.incident_id || '%'
                         OR coalesce(source_envelope::text, '') LIKE '%' || source_ids.incident_id || '%'
                    )
                  )
                ORDER BY queued_at DESC
                LIMIT :limit
                """,
                {**batch_params, "project_id": normalized_project_id},
            )

            approvals_by_incident = _group_rows_by_incident_reference(
                rows=approvals,
                incident_ids=incident_ids,
                direct_fields=("incident_id",),
            )
            evidence_by_incident = _group_rows_by_incident_reference(
                rows=evidence_rows,
                incident_ids=incident_ids,
                direct_fields=("incident_id",),
            )
            timeline_by_incident = _group_timeline_events_by_incident(
                rows=timeline_events,
                incident_ids=incident_ids,
                approval_to_incident=approval_to_incident,
            )
            legacy_mcp_by_incident = _group_rows_by_incident_reference(
                rows=legacy_mcp_rows,
                incident_ids=incident_ids,
                direct_fields=("incident_id",),
            )
            automation_ops_by_incident = _group_rows_by_incident_reference(
                rows=automation_ops,
                incident_ids=incident_ids,
                direct_fields=("incident_id",),
                text_fields=("input_text", "output_text", "tags_text"),
            )
            auto_repair_by_incident = _group_rows_by_incident_reference(
                rows=auto_repair_executions,
                incident_ids=incident_ids,
                direct_fields=("incident_id",),
            )
            km_by_incident = _group_rows_by_incident_reference(
                rows=km_entries,
                incident_ids=incident_ids,
                direct_fields=("related_incident_id",),
            )
            gateway_mcp_by_incident = _group_rows_by_incident_reference(
                rows=gateway_mcp_rows,
                incident_ids=incident_ids,
                direct_fields=("trace_id", "run_id"),
            )
            outbound_by_incident = _group_rows_by_incident_reference(
                rows=outbound_rows,
                incident_ids=incident_ids,
                direct_fields=("run_id",),
                text_fields=("content_preview",),
                source_ref_keys=("incident_ids", "code_refs"),
            )
            records = _build_summary_quality_records(
                incidents=incidents,
                approvals_by_incident=approvals_by_incident,
                evidence_by_incident=evidence_by_incident,
                timeline_by_incident=timeline_by_incident,
                legacy_mcp_by_incident=legacy_mcp_by_incident,
                automation_ops_by_incident=automation_ops_by_incident,
                auto_repair_by_incident=auto_repair_by_incident,
                km_by_incident=km_by_incident,
                gateway_mcp_by_incident=gateway_mcp_by_incident,
                outbound_by_incident=outbound_by_incident,
            )

    summary = summarize_automation_quality_records(
        project_id=normalized_project_id,
        window_hours=bounded_hours,
        records=records,
        limit=bounded_limit,
    )
    logger.info(
        "awooop_automation_quality_summary_fetched",
        project_id=normalized_project_id,
        window_hours=bounded_hours,
        incident_total=summary["incident_total"],
        evaluated_total=summary["evaluated_total"],
        can_claim_full_auto_repair=summary["production_claim"]["can_claim_full_auto_repair"],
        cache_status="miss",
        cache_ttl_seconds=_QUALITY_SUMMARY_CACHE_TTL_SECONDS,
    )
    stored_summary = await store_operator_summary_async(
        "truth_chain_quality_summary",
        cache_key,
        summary,
        ttl_seconds=_QUALITY_SUMMARY_CACHE_TTL_SECONDS,
    )
    duration_seconds = perf_counter() - started_at
    record_quality_summary_observation(
        project_id=normalized_project_id,
        hours=bounded_hours,
        limit=bounded_limit,
        cache_status="miss",
        success=True,
        duration_seconds=duration_seconds,
    )
    stored_summary = dict(stored_summary)
    stored_summary["cache_status"] = "miss"
    stored_summary["aggregation_duration_seconds"] = round(duration_seconds, 3)
    return stored_summary
