feat(awooop): show incident source correlation evidence

apps/api/src/services/platform_operator_service.py

@@ -12,7 +12,7 @@ import re
 import uuid
 from collections import defaultdict
 from collections.abc import Mapping
-from datetime import UTC, datetime
+from datetime import UTC, datetime, timedelta
 from typing import Any, get_args
 from uuid import UUID
 
@@ -30,7 +30,7 @@ from src.db.awooop_models import (
     AwoooPRunStepJournal,
 )
 from src.db.base import get_db_context
-from src.db.models import MCPAuditLog
+from src.db.models import IncidentRecord, MCPAuditLog
 from src.services.audit_sink import write_audit
 from src.services.awooop_approval_token import issue_approval_token, record_approval
 from src.services.awooop_truth_chain_service import (
@@ -87,6 +87,11 @@ _CALLBACK_REPLY_RAW_STATUS_BY_FILTER = {
 _CALLBACK_REPLY_ACTION_RE = re.compile(r"^[a-z0-9_:-]{1,64}$", re.IGNORECASE)
 _AI_ROUTE_STATUS_SCHEMA_VERSION = "awooop_ai_route_status_v1"
 _AI_ROUTE_WORKLOADS = set(get_args(OllamaWorkloadType))
+_SOURCE_CORRELATION_SCHEMA_VERSION = "source_provider_correlation_v1"
+_SOURCE_CORRELATION_PROVIDERS = ("sentry", "signoz")
+_SOURCE_CORRELATION_EVENT_LIMIT = 200
+_SOURCE_CORRELATION_LOOKBACK_DAYS = 7
+_SOURCE_CORRELATION_PRE_WINDOW_HOURS = 2
 
 # =============================================================================
 # Tenants
@@ -1314,6 +1319,406 @@ def _source_ref_values(envelope: Any, key: str) -> list[str]:
     return []
 
 
+def _source_correlation_empty(
+    incident_ids: list[str],
+    *,
+    status_value: str,
+    missing_reason: str,
+) -> dict[str, Any]:
+    return {
+        "schema_version": _SOURCE_CORRELATION_SCHEMA_VERSION,
+        "status": status_value,
+        "missing_reason": missing_reason,
+        "incident_ids": incident_ids,
+        "direct_ref_total": 0,
+        "candidate_total": 0,
+        "provider_event_total": 0,
+        "providers": {
+            provider: {
+                "direct_ref_total": 0,
+                "candidate_total": 0,
+                "latest_event_at": None,
+                "latest_heartbeat_at": None,
+            }
+            for provider in _SOURCE_CORRELATION_PROVIDERS
+        },
+        "top_candidates": [],
+        "matching_criteria": [
+            "direct_source_ref",
+            "fingerprint_overlap",
+            "alertname_overlap",
+            "service_or_namespace_overlap",
+            "severity_overlap",
+        ],
+    }
+
+
+def _normalize_correlation_value(value: Any) -> str:
+    if hasattr(value, "value"):
+        value = value.value
+    return str(value or "").strip().lower()
+
+
+def _append_correlation_term(values: list[str], value: Any) -> None:
+    term = _normalize_correlation_value(value)
+    if term in {"", "--", "n/a", "none", "null", "unknown"}:
+        return
+    if len(term) < 2:
+        return
+    if term not in values:
+        values.append(term)
+
+
+def _intersection(left: list[str], right: list[str]) -> list[str]:
+    right_set = set(right)
+    return [item for item in left if item in right_set]
+
+
+def _as_utc_naive(value: Any) -> datetime | None:
+    if not isinstance(value, datetime):
+        return None
+    if value.tzinfo is not None:
+        return value.astimezone(UTC).replace(tzinfo=None)
+    return value
+
+
+def _iso_or_none(value: Any) -> str | None:
+    if hasattr(value, "isoformat"):
+        return value.isoformat()
+    if value in (None, ""):
+        return None
+    return str(value)
+
+
+def _incident_correlation_context(record: IncidentRecord) -> dict[str, list[str]]:
+    """Build compact incident terms used only for read-only source matching."""
+    alertnames: list[str] = []
+    severities: list[str] = []
+    fingerprints: list[str] = []
+    namespaces: list[str] = []
+    targets: list[str] = []
+
+    _append_correlation_term(alertnames, record.alertname)
+    _append_correlation_term(severities, record.severity)
+    for service in record.affected_services or []:
+        _append_correlation_term(targets, service)
+
+    for signal in record.signals or []:
+        if not isinstance(signal, dict):
+            continue
+        _append_correlation_term(alertnames, signal.get("alert_name"))
+        _append_correlation_term(severities, signal.get("severity"))
+        _append_correlation_term(fingerprints, signal.get("fingerprint"))
+        labels = _as_dict(signal.get("labels"))
+        annotations = _as_dict(signal.get("annotations"))
+        _append_correlation_term(alertnames, labels.get("alertname"))
+        _append_correlation_term(fingerprints, labels.get("fingerprint"))
+        for key in (
+            "namespace",
+            "kubernetes_namespace",
+        ):
+            _append_correlation_term(namespaces, labels.get(key))
+        for key in (
+            "service",
+            "service_name",
+            "pod",
+            "pod_name",
+            "deployment",
+            "deployment_name",
+            "container",
+            "job",
+            "instance",
+            "target",
+            "target_resource",
+            "workload",
+            "app",
+            "app.kubernetes.io/name",
+        ):
+            _append_correlation_term(targets, labels.get(key))
+        for key in ("summary", "description"):
+            _append_correlation_term(alertnames, annotations.get(key))
+
+    return {
+        "incident_ids": [record.incident_id],
+        "alertnames": alertnames,
+        "severities": severities,
+        "fingerprints": fingerprints,
+        "namespaces": namespaces,
+        "targets": targets,
+    }
+
+
+def _source_event_correlation_context(row: Mapping[str, Any]) -> dict[str, Any]:
+    envelope = _as_dict(row.get("source_envelope"))
+    source_refs = _as_dict(envelope.get("source_refs"))
+    log_correlation = _as_dict(envelope.get("log_correlation"))
+    labels = _as_dict(envelope.get("labels"))
+    annotations = _as_dict(envelope.get("annotations"))
+
+    alertnames: list[str] = []
+    severities: list[str] = []
+    fingerprints: list[str] = []
+    namespaces: list[str] = []
+    targets: list[str] = []
+
+    _append_correlation_term(alertnames, log_correlation.get("alertname"))
+    _append_correlation_term(alertnames, labels.get("alertname"))
+    for value in _source_ref_values(envelope, "signoz_alerts"):
+        _append_correlation_term(alertnames, value)
+    _append_correlation_term(severities, log_correlation.get("severity"))
+    _append_correlation_term(severities, labels.get("severity"))
+    _append_correlation_term(fingerprints, log_correlation.get("fingerprint"))
+    _append_correlation_term(fingerprints, labels.get("fingerprint"))
+    for value in _source_ref_values(envelope, "fingerprints"):
+        _append_correlation_term(fingerprints, value)
+    for key in ("namespace", "kubernetes_namespace"):
+        _append_correlation_term(namespaces, log_correlation.get(key))
+        _append_correlation_term(namespaces, labels.get(key))
+    for key in (
+        "target_resource",
+        "service",
+        "service_name",
+        "pod",
+        "pod_name",
+        "deployment",
+        "deployment_name",
+        "container",
+        "job",
+        "instance",
+        "target",
+        "workload",
+        "app",
+        "app.kubernetes.io/name",
+    ):
+        _append_correlation_term(targets, log_correlation.get(key))
+        _append_correlation_term(targets, labels.get(key))
+    for key in ("summary", "description"):
+        _append_correlation_term(alertnames, annotations.get(key))
+
+    return {
+        "provider": str(row.get("provider") or envelope.get("provider") or "").lower(),
+        "stage": str(row.get("stage") or envelope.get("stage") or ""),
+        "provider_event_id": row.get("provider_event_id") or envelope.get("provider_event_id"),
+        "received_at": row.get("received_at"),
+        "source_refs": source_refs,
+        "incident_ids": _source_ref_values(envelope, "incident_ids"),
+        "alertnames": alertnames,
+        "severities": severities,
+        "fingerprints": fingerprints,
+        "namespaces": namespaces,
+        "targets": targets,
+    }
+
+
+def _score_source_correlation_event(
+    incident_context: dict[str, list[str]],
+    event_context: dict[str, Any],
+) -> dict[str, Any]:
+    """Return a deterministic, read-only source-match score for UI evidence."""
+    reasons: list[str] = []
+    score = 0
+    is_direct = False
+
+    if _intersection(incident_context["incident_ids"], event_context["incident_ids"]):
+        is_direct = True
+        score += 100
+        reasons.append("direct_incident_ref")
+
+    fingerprint_hits = _intersection(
+        incident_context["fingerprints"],
+        event_context["fingerprints"],
+    )
+    if fingerprint_hits:
+        is_direct = True
+        score += 80
+        reasons.append("fingerprint_overlap")
+
+    if _intersection(incident_context["alertnames"], event_context["alertnames"]):
+        score += 35
+        reasons.append("alertname_overlap")
+
+    if _intersection(incident_context["targets"], event_context["targets"]):
+        score += 25
+        reasons.append("target_overlap")
+
+    if _intersection(incident_context["namespaces"], event_context["namespaces"]):
+        score += 10
+        reasons.append("namespace_overlap")
+
+    if _intersection(incident_context["severities"], event_context["severities"]):
+        score += 5
+        reasons.append("severity_overlap")
+
+    return {
+        "is_direct": is_direct,
+        "is_candidate": bool(is_direct or score >= 35),
+        "score": min(score, 100),
+        "reasons": reasons[:5],
+    }
+
+
+async def _fetch_source_correlation_summary(
+    *,
+    project_id: str,
+    incident_ids: list[str],
+) -> dict[str, Any]:
+    """Fetch read-only Sentry/SigNoz evidence candidates for incident status-chain."""
+    if not incident_ids:
+        return _source_correlation_empty(
+            incident_ids,
+            status_value="no_incident_context",
+            missing_reason="no_incident_ids",
+        )
+
+    safe_project_id = project_id or "awoooi"
+    async with get_db_context(safe_project_id) as db:
+        incident_result = await db.execute(
+            select(IncidentRecord)
+            .where(IncidentRecord.project_id == safe_project_id)
+            .where(IncidentRecord.incident_id.in_(incident_ids))
+        )
+        incident_rows = list(incident_result.scalars().all())
+
+        if not incident_rows:
+            heartbeat_rows = []
+            source_rows = []
+        else:
+            now = _utc_now_naive()
+            created_candidates = [
+                value
+                for value in (_as_utc_naive(row.created_at) for row in incident_rows)
+                if value is not None
+            ]
+            earliest_created = min(created_candidates) if created_candidates else now
+            window_start = max(
+                earliest_created - timedelta(hours=_SOURCE_CORRELATION_PRE_WINDOW_HOURS),
+                now - timedelta(days=_SOURCE_CORRELATION_LOOKBACK_DAYS),
+            )
+            provider_sql = (
+                "LOWER(COALESCE(NULLIF(source_envelope->>'provider', ''), "
+                "NULLIF(split_part(provider_event_id, ':', 1), ''), channel_type))"
+            )
+            source_result = await db.execute(
+                text(f"""
+                    SELECT
+                        event_id::text AS event_id,
+                        project_id,
+                        channel_type,
+                        provider_event_id,
+                        content_preview,
+                        source_envelope,
+                        received_at,
+                        {provider_sql} AS provider,
+                        LOWER(COALESCE(source_envelope->>'stage', '')) AS stage
+                    FROM awooop_conversation_event
+                    WHERE project_id = :project_id
+                      AND {provider_sql} IN ('sentry', 'signoz')
+                      AND LOWER(COALESCE(source_envelope->>'stage', '')) <> 'heartbeat'
+                      AND received_at >= :window_start
+                    ORDER BY received_at DESC
+                    LIMIT :limit
+                """),
+                {
+                    "project_id": safe_project_id,
+                    "window_start": window_start,
+                    "limit": _SOURCE_CORRELATION_EVENT_LIMIT,
+                },
+            )
+            source_rows = list(source_result.mappings().all())
+
+            heartbeat_result = await db.execute(
+                text(f"""
+                    SELECT
+                        {provider_sql} AS provider,
+                        MAX(received_at) AS latest_heartbeat_at
+                    FROM awooop_conversation_event
+                    WHERE project_id = :project_id
+                      AND {provider_sql} IN ('sentry', 'signoz')
+                      AND LOWER(COALESCE(source_envelope->>'stage', '')) = 'heartbeat'
+                    GROUP BY {provider_sql}
+                """),
+                {"project_id": safe_project_id},
+            )
+            heartbeat_rows = list(heartbeat_result.mappings().all())
+
+    if not incident_rows:
+        summary = _source_correlation_empty(
+            incident_ids,
+            status_value="no_incident_context",
+            missing_reason="incident_not_found",
+        )
+        return summary
+
+    contexts = [_incident_correlation_context(row) for row in incident_rows]
+    summary = _source_correlation_empty(
+        incident_ids,
+        status_value="missing",
+        missing_reason="no_matching_provider_source_event",
+    )
+    providers = summary["providers"]
+    for heartbeat in heartbeat_rows:
+        provider = str(heartbeat.get("provider") or "").lower()
+        if provider in providers:
+            providers[provider]["latest_heartbeat_at"] = _iso_or_none(
+                heartbeat.get("latest_heartbeat_at")
+            )
+
+    top_candidates: list[dict[str, Any]] = []
+    for row in source_rows:
+        event_context = _source_event_correlation_context(row)
+        provider = str(event_context.get("provider") or "").lower()
+        if provider not in providers:
+            continue
+        provider_item = providers[provider]
+        if provider_item.get("latest_event_at") is None:
+            provider_item["latest_event_at"] = _iso_or_none(row.get("received_at"))
+
+        best_match: dict[str, Any] | None = None
+        for context in contexts:
+            scored = _score_source_correlation_event(context, event_context)
+            if best_match is None or scored["score"] > best_match["score"]:
+                best_match = scored
+        if not best_match or not best_match["is_candidate"]:
+            continue
+
+        summary["provider_event_total"] += 1
+        if best_match["is_direct"]:
+            summary["direct_ref_total"] += 1
+            provider_item["direct_ref_total"] += 1
+        else:
+            summary["candidate_total"] += 1
+            provider_item["candidate_total"] += 1
+
+        top_candidates.append(
+            {
+                "provider": provider,
+                "provider_event_id": str(event_context.get("provider_event_id") or ""),
+                "stage": str(event_context.get("stage") or ""),
+                "score": best_match["score"],
+                "match_type": "direct" if best_match["is_direct"] else "candidate",
+                "reasons": best_match["reasons"],
+                "received_at": _iso_or_none(row.get("received_at")),
+            }
+        )
+
+    if summary["direct_ref_total"] > 0:
+        summary["status"] = "linked"
+        summary["missing_reason"] = None
+    elif summary["candidate_total"] > 0:
+        summary["status"] = "candidate_found"
+        summary["missing_reason"] = None
+    elif any(item.get("latest_heartbeat_at") for item in providers.values()):
+        summary["status"] = "provider_fresh_no_match"
+        summary["missing_reason"] = "provider_heartbeat_present_but_no_incident_match"
+
+    summary["top_candidates"] = sorted(
+        top_candidates,
+        key=lambda item: (item.get("score") or 0, item.get("received_at") or ""),
+        reverse=True,
+    )[:5]
+    return summary
+
+
 def _status_chain_source_section(truth_chain: dict[str, Any] | None) -> dict[str, Any]:
     channel = truth_chain.get("channel") if isinstance(truth_chain, dict) else {}
     if not isinstance(channel, dict):
@@ -1377,6 +1782,7 @@ def _build_awooop_status_chain(
     remediation_history: dict[str, Any] | None = None,
     source_id: str | None = None,
     fetch_error: str | None = None,
+    source_correlation: dict[str, Any] | None = None,
 ) -> dict[str, Any]:
     """Build the shared read-only status chain used by Telegram and Operator UI."""
     truth_status = (
@@ -1450,6 +1856,8 @@ def _build_awooop_status_chain(
     mcp_section = _status_chain_mcp_section(truth_chain)
     execution_section = _status_chain_execution_section(truth_chain)
     source_section = _status_chain_source_section(truth_chain)
+    if source_correlation is not None:
+        source_section["correlation"] = source_correlation
     blockers = [
         str(item)
         for item in [
@@ -1522,12 +1930,31 @@ async def _fetch_awooop_status_chain(
                 error=fetch_error,
             )
 
+    try:
+        source_correlation = await _fetch_source_correlation_summary(
+            incident_ids=incident_ids,
+            project_id=project_id or "awoooi",
+        )
+    except Exception as exc:
+        logger.warning(
+            "operator_source_correlation_fetch_failed",
+            incident_ids=incident_ids,
+            project_id=project_id,
+            error=str(exc),
+        )
+        source_correlation = _source_correlation_empty(
+            incident_ids,
+            status_value="fetch_failed",
+            missing_reason="source_correlation_fetch_failed",
+        )
+
     return _build_awooop_status_chain(
         incident_ids=incident_ids,
         truth_chain=truth_chain,
         remediation_history=remediation_history,
         source_id=source_id,
         fetch_error=fetch_error,
+        source_correlation=source_correlation,
     )