feat(awooop): link recurring alerts to repair work

apps/api/src/services/channel_event_dossier_service.py

@@ -7,6 +7,7 @@ automation state.
 
 from __future__ import annotations
 
+import re
 from typing import Any
 from uuid import UUID
 
@@ -18,6 +19,8 @@ from src.db.base import get_db_context
 _MAX_DOSSIER_EVENTS = 50
 _MAX_COVERAGE_EVENTS = 200
 _MAX_RECURRENCE_EVENTS = 300
+_MAX_REPAIR_INCIDENTS = 200
+_INCIDENT_ID_RE = re.compile(r"\bINC-\d{8}-[A-Z0-9]{4,}\b")
 
 
 def _as_dict(value: Any) -> dict[str, Any]:
@@ -41,6 +44,39 @@ def _ref_count(source_refs: dict[str, Any], key: str) -> int:
     return 1 if value else 0
 
 
+def _append_unique(values: list[str], candidate: Any) -> None:
+    text_value = str(candidate or "").strip()
+    if text_value and text_value not in values:
+        values.append(text_value)
+
+
+def _append_incident_ids_from_text(values: list[str], text_value: Any) -> None:
+    if not text_value:
+        return
+    for incident_id in _INCIDENT_ID_RE.findall(str(text_value)):
+        _append_unique(values, incident_id)
+
+
+def _append_incident_ids_from_refs(
+    values: list[str], source_refs: dict[str, Any]
+) -> None:
+    incident_ids = source_refs.get("incident_ids")
+    if isinstance(incident_ids, list):
+        for incident_id in incident_ids:
+            _append_unique(values, incident_id)
+    else:
+        _append_unique(values, incident_ids)
+
+
+def _event_incident_ids(event: dict[str, Any]) -> list[str]:
+    incident_ids: list[str] = []
+    _append_incident_ids_from_refs(incident_ids, _as_dict(event.get("source_refs")))
+    _append_incident_ids_from_text(incident_ids, event.get("content_preview"))
+    _append_incident_ids_from_text(incident_ids, event.get("content_redacted"))
+    _append_incident_ids_from_text(incident_ids, event.get("provider_event_id"))
+    return incident_ids
+
+
 def _recurrence_key(event: dict[str, Any]) -> str:
     fingerprint = str(event.get("fingerprint") or "").strip()
     if fingerprint:
@@ -61,15 +97,18 @@ def build_dossier_recurrence(
     *,
     project_id: str,
     limit: int,
+    repair_summaries_by_incident: dict[str, dict[str, Any]] | None = None,
 ) -> dict[str, Any]:
     """Group recent source events into recurrence buckets with linked run state."""
     groups: dict[str, dict[str, Any]] = {}
+    repair_summaries = repair_summaries_by_incident or {}
 
     for row in rows:
         event = build_dossier_event(row)
         key = _recurrence_key(event)
         source_ref_count = int(event.get("source_ref_count") or 0)
         source_refs = _as_dict(event.get("source_refs"))
+        incident_ids = _event_incident_ids(event)
         run_id = row.get("run_id")
         run_state = row.get("run_state")
         received_at = event.get("received_at")
@@ -90,6 +129,8 @@ def build_dossier_recurrence(
                 "latest_run_id": run_id,
                 "latest_run_state": run_state,
                 "latest_agent_id": row.get("run_agent_id"),
+                "latest_incident_id": incident_ids[0] if incident_ids else None,
+                "incident_ids": [],
                 "occurrence_total": 0,
                 "duplicate_total": 0,
                 "linked_run_total": 0,
@@ -112,6 +153,9 @@ def build_dossier_recurrence(
         if event.get("is_duplicate"):
             group["duplicate_total"] += 1
 
+        for incident_id in incident_ids:
+            _append_unique(group["incident_ids"], incident_id)
+
         group["sentry_ref_total"] += _ref_count(source_refs, "sentry_issue_ids")
         group["signoz_ref_total"] += _ref_count(source_refs, "signoz_alerts")
         group["alert_ref_total"] += _ref_count(source_refs, "alert_ids")
@@ -136,19 +180,27 @@ def build_dossier_recurrence(
             or str(received_at) > str(group.get("latest_received_at"))
         ):
             group["latest_received_at"] = received_at
+            group["latest_incident_id"] = (
+                incident_ids[0] if incident_ids else group.get("latest_incident_id")
+            )
 
     items = []
     linked_run_total = 0
     for group in groups.values():
         run_ids = group.pop("_run_ids")
         group["linked_run_total"] = len(run_ids)
+        _attach_work_item_summary(group, repair_summaries)
         linked_run_total += len(run_ids)
         items.append(group)
 
     items.sort(key=lambda item: str(item.get("latest_received_at") or ""), reverse=True)
     items.sort(key=lambda item: int(item.get("occurrence_total") or 0), reverse=True)
     latest_received_at = max(
-        (item.get("latest_received_at") for item in items if item.get("latest_received_at")),
+        (
+            item.get("latest_received_at")
+            for item in items
+            if item.get("latest_received_at")
+        ),
         default=None,
     )
 
@@ -161,15 +213,133 @@ def build_dossier_recurrence(
             "recurrent_group_total": sum(
                 1 for item in items if int(item.get("occurrence_total") or 0) > 1
             ),
-            "duplicate_event_total": sum(int(item.get("duplicate_total") or 0) for item in items),
+            "duplicate_event_total": sum(
+                int(item.get("duplicate_total") or 0) for item in items
+            ),
             "linked_run_total": linked_run_total,
             "unlinked_event_total": sum(1 for row in rows if not row.get("run_id")),
+            "auto_repair_linked_total": sum(
+                1
+                for item in items
+                if _as_dict(item.get("repair_summary")).get("latest_auto_repair_id")
+            ),
+            "verified_repair_group_total": sum(
+                1
+                for item in items
+                if _as_dict(item.get("repair_summary")).get("status")
+                == "auto_repair_verified"
+            ),
+            "open_work_item_group_total": sum(
+                1
+                for item in items
+                if _as_dict(item.get("work_item")).get("status") == "open"
+            ),
+            "manual_gate_group_total": sum(
+                1
+                for item in items
+                if _as_dict(item.get("repair_summary")).get("status") == "manual_gate"
+            ),
             "latest_received_at": latest_received_at,
         },
         "items": items,
     }
 
 
+def _repair_status(
+    *,
+    incident_id: str | None,
+    latest_run_state: str | None,
+    repair_summary: dict[str, Any] | None,
+) -> str:
+    if not incident_id:
+        return "no_incident_link"
+    if repair_summary:
+        latest_success = repair_summary.get("latest_success")
+        verification = str(
+            repair_summary.get("latest_verification_result") or ""
+        ).lower()
+        if latest_success is True and verification == "success":
+            return "auto_repair_verified"
+        if latest_success is True:
+            return "auto_repair_succeeded_unverified"
+        if latest_success is False:
+            return "auto_repair_failed"
+        return "auto_repair_recorded"
+    if latest_run_state == "waiting_approval":
+        return "manual_gate"
+    if latest_run_state in {"pending", "running", "waiting_tool"}:
+        return "investigating"
+    if latest_run_state == "completed":
+        return "run_completed_no_repair"
+    return "no_repair_record"
+
+
+def _work_item_status(repair_status: str) -> str:
+    if repair_status in {"no_incident_link", "run_completed_no_repair"}:
+        return "none"
+    if repair_status == "auto_repair_verified":
+        return "closed"
+    return "open"
+
+
+def _attach_work_item_summary(
+    group: dict[str, Any],
+    repair_summaries_by_incident: dict[str, dict[str, Any]],
+) -> None:
+    incident_ids = [
+        str(incident_id) for incident_id in group.get("incident_ids", []) if incident_id
+    ]
+    latest_incident_id = str(group.get("latest_incident_id") or "") or (
+        incident_ids[0] if incident_ids else None
+    )
+    repair_summary = (
+        repair_summaries_by_incident.get(latest_incident_id)
+        if latest_incident_id
+        else None
+    )
+    status_value = _repair_status(
+        incident_id=latest_incident_id,
+        latest_run_state=group.get("latest_run_state"),
+        repair_summary=repair_summary,
+    )
+    if repair_summary:
+        repair_payload = dict(repair_summary)
+        repair_payload["status"] = status_value
+    else:
+        repair_payload = {
+            "schema_version": "awooop_recurrence_repair_summary_v1",
+            "status": status_value,
+            "incident_id": latest_incident_id,
+            "latest_auto_repair_id": None,
+            "latest_verification_result": None,
+            "auto_repair_total": 0,
+            "success_total": 0,
+            "failed_total": 0,
+        }
+
+    work_status = _work_item_status(status_value)
+    auto_repair_id = repair_payload.get("latest_auto_repair_id")
+    work_item_id = None
+    if latest_incident_id and work_status != "none":
+        work_item_id = (
+            f"verification:{latest_incident_id}:{auto_repair_id}"
+            if auto_repair_id
+            else f"incident:{latest_incident_id}"
+        )
+
+    group["latest_incident_id"] = latest_incident_id
+    group["repair_summary"] = repair_payload
+    group["work_item"] = {
+        "schema_version": "awooop_recurrence_work_item_link_v1",
+        "work_item_id": work_item_id,
+        "incident_id": latest_incident_id,
+        "auto_repair_id": auto_repair_id,
+        "status": work_status,
+        "kind": "verification" if auto_repair_id else "incident_followup",
+        "needs_human": work_status == "open",
+    }
+
+
 def build_dossier_coverage(
     rows: list[dict[str, Any]],
     *,
@@ -229,9 +399,9 @@ def build_dossier_coverage(
         provider_item["sentry_ref_total"] += event_sentry_refs
         provider_item["signoz_ref_total"] += event_signoz_refs
         provider_item["alert_ref_total"] += event_alert_refs
-        provider_item["latest_received_at"] = (
-            provider_item["latest_received_at"] or event.get("received_at")
-        )
+        provider_item["latest_received_at"] = provider_item[
+            "latest_received_at"
+        ] or event.get("received_at")
 
     duplicate_total = sum(1 for event in events if event.get("is_duplicate"))
     redacted_total = sum(1 for event in events if event.get("has_redacted_content"))
@@ -258,7 +428,10 @@ def build_dossier_coverage(
         },
         "providers": sorted(
             provider_map.values(),
-            key=lambda item: (-int(item.get("total") or 0), str(item.get("provider") or "")),
+            key=lambda item: (
+                -int(item.get("total") or 0),
+                str(item.get("provider") or ""),
+            ),
         ),
     }
 
@@ -299,6 +472,104 @@ def build_dossier_event(row: dict[str, Any]) -> dict[str, Any]:
     }
 
 
+def _collect_incident_ids_from_rows(rows: list[dict[str, Any]]) -> list[str]:
+    incident_ids: list[str] = []
+    for row in rows:
+        event = build_dossier_event(row)
+        for incident_id in _event_incident_ids(event):
+            _append_unique(incident_ids, incident_id)
+    return incident_ids
+
+
+async def _fetch_auto_repair_summaries_by_incident(
+    db: Any,
+    incident_ids: list[str],
+) -> dict[str, dict[str, Any]]:
+    """Fetch latest auto-repair and verifier evidence for recurrence groups."""
+    visible_incident_ids = incident_ids[:_MAX_REPAIR_INCIDENTS]
+    if not visible_incident_ids:
+        return {}
+
+    placeholders: list[str] = []
+    params: dict[str, Any] = {}
+    for index, incident_id in enumerate(visible_incident_ids):
+        key = f"incident_id_{index}"
+        placeholders.append(f":{key}")
+        params[key] = incident_id
+
+    result = await db.execute(
+        text(
+            f"""
+            WITH ranked AS (
+                SELECT
+                    are.id AS latest_auto_repair_id,
+                    are.incident_id,
+                    are.playbook_id AS latest_playbook_id,
+                    are.playbook_name AS latest_playbook_name,
+                    are.success AS latest_success,
+                    left(coalesce(are.error_message, ''), 240) AS latest_error_message_preview,
+                    are.triggered_by AS latest_triggered_by,
+                    are.risk_level AS latest_risk_level,
+                    are.execution_time_ms AS latest_execution_time_ms,
+                    are.created_at AS latest_auto_repair_at,
+                    latest_evidence.verification_result AS latest_verification_result,
+                    latest_evidence.collected_at AS latest_verification_at,
+                    COUNT(*) OVER (PARTITION BY are.incident_id) AS auto_repair_total,
+                    COUNT(*) FILTER (WHERE are.success IS TRUE)
+                        OVER (PARTITION BY are.incident_id) AS success_total,
+                    COUNT(*) FILTER (WHERE are.success IS FALSE)
+                        OVER (PARTITION BY are.incident_id) AS failed_total,
+                    ROW_NUMBER() OVER (
+                        PARTITION BY are.incident_id
+                        ORDER BY are.created_at DESC
+                    ) AS rn
+                FROM auto_repair_executions are
+                LEFT JOIN LATERAL (
+                    SELECT
+                        ev.verification_result,
+                        ev.collected_at
+                    FROM incident_evidence ev
+                    WHERE ev.incident_id = are.incident_id
+                      AND ev.verification_result IS NOT NULL
+                    ORDER BY ev.collected_at DESC
+                    LIMIT 1
+                ) latest_evidence ON TRUE
+                WHERE are.incident_id IN ({", ".join(placeholders)})
+            )
+            SELECT
+                latest_auto_repair_id,
+                incident_id,
+                latest_playbook_id,
+                latest_playbook_name,
+                latest_success,
+                latest_error_message_preview,
+                latest_triggered_by,
+                latest_risk_level,
+                latest_execution_time_ms,
+                latest_auto_repair_at,
+                latest_verification_result,
+                latest_verification_at,
+                auto_repair_total,
+                success_total,
+                failed_total
+            FROM ranked
+            WHERE rn = 1
+        """
+        ),
+        params,
+    )
+
+    summaries: dict[str, dict[str, Any]] = {}
+    for row in result.mappings().all():
+        item = dict(row)
+        incident_id = str(item.get("incident_id") or "")
+        if not incident_id:
+            continue
+        item["schema_version"] = "awooop_recurrence_repair_summary_v1"
+        summaries[incident_id] = item
+    return summaries
+
+
 async def fetch_channel_event_dossier(
     *,
     project_id: str | None,
@@ -329,7 +600,8 @@ async def fetch_channel_event_dossier(
 
     async with get_db_context(effective_project_id) as db:
         result = await db.execute(
-            text(f"""
+            text(
+                f"""
                 SELECT
                     event_id,
                     project_id,
@@ -347,7 +619,8 @@ async def fetch_channel_event_dossier(
                 WHERE {" AND ".join(where_clauses)}
                 ORDER BY received_at ASC
                 LIMIT :limit
-            """),
+            """
+            ),
             params,
         )
         rows = [dict(row) for row in result.mappings().all()]
@@ -364,7 +637,9 @@ async def fetch_channel_event_dossier(
             "source_count": len(events),
             "duplicate_total": duplicate_total,
             "redacted_total": redacted_total,
-            "source_ref_total": sum(int(event.get("source_ref_count") or 0) for event in events),
+            "source_ref_total": sum(
+                int(event.get("source_ref_count") or 0) for event in events
+            ),
         },
     }
 
@@ -392,7 +667,8 @@ async def fetch_channel_event_dossier_coverage(
 
     async with get_db_context(effective_project_id) as db:
         result = await db.execute(
-            text(f"""
+            text(
+                f"""
                 SELECT
                     event_id,
                     project_id,
@@ -410,7 +686,8 @@ async def fetch_channel_event_dossier_coverage(
                 WHERE {" AND ".join(where_clauses)}
                 ORDER BY received_at DESC
                 LIMIT :limit
-            """),
+            """
+            ),
             params,
         )
         rows = [dict(row) for row in result.mappings().all()]
@@ -445,7 +722,8 @@ async def fetch_channel_event_dossier_recurrence(
 
     async with get_db_context(effective_project_id) as db:
         result = await db.execute(
-            text(f"""
+            text(
+                f"""
                 SELECT
                     e.event_id,
                     e.project_id,
@@ -469,13 +747,19 @@ async def fetch_channel_event_dossier_recurrence(
                 WHERE {" AND ".join(where_clauses)}
                 ORDER BY e.received_at DESC
                 LIMIT :limit
-            """),
+            """
+            ),
             params,
         )
         rows = [dict(row) for row in result.mappings().all()]
+        repair_summaries = await _fetch_auto_repair_summaries_by_incident(
+            db,
+            _collect_incident_ids_from_rows(rows),
+        )
 
     return build_dossier_recurrence(
         rows,
         project_id=effective_project_id,
         limit=safe_limit,
+        repair_summaries_by_incident=repair_summaries,
     )