feat(awooop): record source correlation review decisions

apps/api/src/services/channel_event_dossier_service.py

@@ -26,9 +26,18 @@ _MAX_REPAIR_INCIDENTS = 200
 _SOURCE_CORRELATION_REVIEW_PROVIDERS = {"sentry", "signoz"}
 _SOURCE_CORRELATION_REVIEW_EXCLUDED_STAGES = {"heartbeat"}
 _SOURCE_CORRELATION_WORK_ITEM_ID_MAX = 180
+_SOURCE_CORRELATION_DECISION_SCHEMA_VERSION = (
+    "awooop_source_correlation_review_decision_v1"
+)
+_SOURCE_CORRELATION_REVIEW_ACTOR = "awooop_source_correlation_review_service"
 _INCIDENT_ID_RE = re.compile(r"\bINC-\d{8}-[A-Z0-9]{4,}\b")
 RecurrenceWorkItemMode = Literal["auto", "ticket", "reverify", "approval_review", "observe"]
 RecurrenceWorkItemHandoffKind = Literal["ticket_proposal", "manual_review"]
+SourceCorrelationReviewDecision = Literal[
+    "accepted",
+    "rejected",
+    "needs_more_evidence",
+]
 
 
 class RecurrenceWorkItemNotFoundError(LookupError):
@@ -42,9 +51,9 @@ def _as_dict(value: Any) -> dict[str, Any]:
 def _json_safe(value: Any) -> Any:
     if isinstance(value, dict):
         return {str(key): _json_safe(item) for key, item in value.items()}
-    if isinstance(value, (list, tuple, set)):
+    if isinstance(value, list | tuple | set):
         return [_json_safe(item) for item in value]
-    if value is None or isinstance(value, (str, int, float, bool)):
+    if value is None or isinstance(value, str | int | float | bool):
         return value
     return str(value)
 
@@ -94,6 +103,58 @@ def _source_correlation_work_item_id(group: dict[str, Any]) -> str:
     return f"source-evidence:{source_id}"[:_SOURCE_CORRELATION_WORK_ITEM_ID_MAX]
 
 
+def _source_correlation_review_outcome(decision: str) -> dict[str, str]:
+    if decision == "accepted":
+        return {
+            "status": "source_correlation_accepted",
+            "next_step": "verify_source_match_in_status_chain",
+            "reason": "provider_native_evidence_accepted",
+            "review_status": "accepted",
+        }
+    if decision == "rejected":
+        return {
+            "status": "source_correlation_rejected",
+            "next_step": "monitor_for_new_provider_evidence",
+            "reason": "provider_native_evidence_rejected",
+            "review_status": "rejected",
+        }
+    return {
+        "status": "source_correlation_review",
+        "next_step": "collect_more_source_evidence",
+        "reason": "provider_native_evidence_needs_more_evidence",
+        "review_status": "needs_more_evidence",
+    }
+
+
+def _normalize_source_review_decision(row: dict[str, Any]) -> dict[str, Any] | None:
+    context = _as_dict(row.get("context"))
+    if context.get("schema_version") != _SOURCE_CORRELATION_DECISION_SCHEMA_VERSION:
+        return None
+    work_item_id = str(context.get("work_item_id") or "").strip()
+    decision = str(context.get("decision") or "").strip()
+    if not work_item_id or decision not in {
+        "accepted",
+        "rejected",
+        "needs_more_evidence",
+    }:
+        return None
+
+    return {
+        "schema_version": _SOURCE_CORRELATION_DECISION_SCHEMA_VERSION,
+        "review_id": row.get("id"),
+        "work_item_id": work_item_id,
+        "decision": decision,
+        "review_status": context.get("review_status") or decision,
+        "target_incident_id": context.get("target_incident_id"),
+        "reviewer_id": context.get("reviewer_id"),
+        "operator_note": context.get("operator_note"),
+        "latest_provider_event_id": context.get("latest_provider_event_id"),
+        "latest_stage": context.get("latest_stage"),
+        "recorded_at": row.get("created_at"),
+        "history": context.get("history") or {},
+    }
+
+
 def _append_unique(values: list[str], candidate: Any) -> None:
     text_value = str(candidate or "").strip()
     if text_value and text_value not in values:
@@ -148,10 +209,12 @@ def build_dossier_recurrence(
     project_id: str,
     limit: int,
     repair_summaries_by_incident: dict[str, dict[str, Any]] | None = None,
+    source_review_decisions_by_work_item: 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 {}
+    source_review_decisions = source_review_decisions_by_work_item or {}
 
     for row in rows:
         event = build_dossier_event(row)
@@ -249,7 +312,11 @@ def build_dossier_recurrence(
     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)
+        _attach_work_item_summary(
+            group,
+            repair_summaries,
+            source_review_decisions,
+        )
         linked_run_total += len(run_ids)
         items.append(group)
 
@@ -317,6 +384,9 @@ def build_dossier_recurrence(
                 if _as_dict(item.get("repair_summary")).get("status")
                 == "source_correlation_review"
             ),
+            "source_correlation_decision_recorded_group_total": sum(
+                1 for item in items if item.get("source_correlation_review")
+            ),
             "latest_received_at": latest_received_at,
         },
         "items": items,
@@ -355,13 +425,21 @@ def _repair_status(
 def _work_item_status(repair_status: str) -> str:
     if repair_status == "no_incident_link":
         return "none"
-    if repair_status == "auto_repair_verified":
+    if repair_status in {
+        "auto_repair_verified",
+        "source_correlation_accepted",
+        "source_correlation_rejected",
+    }:
         return "closed"
     return "open"
 
 
 def _work_item_kind(repair_status: str, auto_repair_id: Any) -> str:
-    if repair_status == "source_correlation_review":
+    if repair_status in {
+        "source_correlation_review",
+        "source_correlation_accepted",
+        "source_correlation_rejected",
+    }:
         return "source_correlation_review"
     if auto_repair_id:
         return "verification"
@@ -377,6 +455,8 @@ def _work_item_kind(repair_status: str, auto_repair_id: Any) -> str:
 def _work_item_next_step(repair_status: str) -> str:
     return {
         "source_correlation_review": "review_provider_source_match",
+        "source_correlation_accepted": "verify_source_match_in_status_chain",
+        "source_correlation_rejected": "monitor_for_new_provider_evidence",
         "auto_repair_succeeded_unverified": "run_post_verification",
         "auto_repair_failed": "triage_failed_repair",
         "auto_repair_recorded": "review_repair_record",
@@ -390,6 +470,8 @@ def _work_item_next_step(repair_status: str) -> str:
 def _work_item_reason(repair_status: str) -> str:
     return {
         "source_correlation_review": "provider_native_evidence_unlinked",
+        "source_correlation_accepted": "provider_native_evidence_accepted",
+        "source_correlation_rejected": "provider_native_evidence_rejected",
         "auto_repair_succeeded_unverified": "auto_repair_missing_verification",
         "auto_repair_failed": "auto_repair_failed",
         "auto_repair_recorded": "auto_repair_record_needs_review",
@@ -403,6 +485,7 @@ def _work_item_reason(repair_status: str) -> str:
 def _attach_work_item_summary(
     group: dict[str, Any],
     repair_summaries_by_incident: dict[str, dict[str, Any]],
+    source_review_decisions_by_work_item: dict[str, dict[str, Any]],
 ) -> None:
     incident_ids = [
         str(incident_id) for incident_id in group.get("incident_ids", []) if incident_id
@@ -450,17 +533,38 @@ def _attach_work_item_summary(
     elif status_value == "source_correlation_review" and work_status != "none":
         work_item_id = _source_correlation_work_item_id(group)
 
+    source_review_decision = (
+        source_review_decisions_by_work_item.get(work_item_id)
+        if work_item_id
+        else None
+    )
+    matched_incident_id = None
+    work_item_next_step = _work_item_next_step(status_value)
+    work_item_reason = _work_item_reason(status_value)
+    if source_review_decision:
+        outcome = _source_correlation_review_outcome(
+            str(source_review_decision.get("decision") or "")
+        )
+        status_value = outcome["status"]
+        repair_payload["status"] = status_value
+        work_status = _work_item_status(status_value)
+        matched_incident_id = source_review_decision.get("target_incident_id")
+        work_item_next_step = outcome["next_step"]
+        work_item_reason = outcome["reason"]
+        group["source_correlation_review"] = source_review_decision
+
     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,
+        "matched_incident_id": matched_incident_id,
         "auto_repair_id": auto_repair_id,
         "status": work_status,
         "kind": _work_item_kind(status_value, auto_repair_id),
-        "next_step": _work_item_next_step(status_value),
-        "reason": _work_item_reason(status_value),
+        "next_step": work_item_next_step,
+        "reason": work_item_reason,
         "needs_human": work_status == "open",
     }
 
@@ -481,6 +585,9 @@ def _recurrence_work_item_target(item: dict[str, Any]) -> dict[str, Any]:
         "latest_run_state": item.get("latest_run_state"),
         "latest_agent_id": item.get("latest_agent_id"),
         "latest_incident_id": item.get("latest_incident_id"),
+        "matched_incident_id": _as_dict(item.get("work_item")).get(
+            "matched_incident_id"
+        ),
     }
 
 
@@ -653,6 +760,8 @@ def _recurrence_current_state_summary(
         "latest_provider_event_id": item.get("latest_provider_event_id"),
         "latest_run_state": item.get("latest_run_state"),
         "latest_run_id": item.get("latest_run_id"),
+        "matched_incident_id": work_item.get("matched_incident_id"),
+        "source_correlation_review": item.get("source_correlation_review"),
         "repair_status": repair_summary.get("status"),
         "latest_auto_repair_id": repair_summary.get("latest_auto_repair_id"),
         "latest_verification_result": repair_summary.get("latest_verification_result"),
@@ -1063,6 +1172,243 @@ def _recurrence_handoff_history_description(context: dict[str, Any]) -> str:
     )[:500]
 
 
+def _source_correlation_review_checks(
+    item: dict[str, Any],
+    work_item: dict[str, Any],
+    *,
+    decision: str,
+    target_incident_id: str | None,
+) -> list[dict[str, Any]]:
+    source_ref_total = int(item.get("source_ref_total") or 0)
+    is_source_review = work_item.get("kind") == "source_correlation_review"
+    checks = [
+        {
+            "name": "source_review_work_item",
+            "passed": is_source_review,
+            "detail": str(work_item.get("kind") or "unknown"),
+        },
+        {
+            "name": "known_decision",
+            "passed": decision
+            in {"accepted", "rejected", "needs_more_evidence"},
+            "detail": decision or "unknown",
+        },
+        {
+            "name": "source_refs_present",
+            "passed": source_ref_total > 0,
+            "detail": str(source_ref_total),
+        },
+        {
+            "name": "provider_event_present",
+            "passed": bool(item.get("latest_provider_event_id")),
+            "detail": str(item.get("latest_provider_event_id") or "missing"),
+        },
+        {
+            "name": "review_record_only",
+            "passed": True,
+            "detail": "alert_operation_log_only",
+        },
+    ]
+    if decision == "accepted":
+        checks.append(
+            {
+                "name": "target_incident_required",
+                "passed": bool(
+                    target_incident_id and _INCIDENT_ID_RE.search(target_incident_id)
+                ),
+                "detail": target_incident_id or "missing target_incident_id",
+            }
+        )
+    return checks
+
+
+def _source_correlation_review_context(payload: dict[str, Any]) -> dict[str, Any]:
+    return {
+        "schema_version": _SOURCE_CORRELATION_DECISION_SCHEMA_VERSION,
+        "source": payload.get("source"),
+        "project_id": payload.get("project_id"),
+        "work_item_id": payload.get("work_item_id"),
+        "decision": payload.get("decision"),
+        "review_status": payload.get("review_status"),
+        "target_incident_id": payload.get("target_incident_id"),
+        "reviewer_id": payload.get("reviewer_id"),
+        "operator_note": payload.get("operator_note"),
+        "latest_provider_event_id": payload.get("latest_provider_event_id"),
+        "latest_stage": payload.get("latest_stage"),
+        "provider": payload.get("provider"),
+        "alertname": payload.get("alertname"),
+        "namespace": payload.get("namespace"),
+        "target_resource": payload.get("target_resource"),
+        "safety_level": payload.get("safety_level"),
+        "writes_incident_state": payload.get("writes_incident_state"),
+        "writes_source_event": payload.get("writes_source_event"),
+        "writes_auto_repair_result": payload.get("writes_auto_repair_result"),
+        "writes_ticket": payload.get("writes_ticket"),
+        "creates_external_ticket": payload.get("creates_external_ticket"),
+        "checks": payload.get("checks"),
+        "current_state_summary": payload.get("current_state_summary"),
+        "plan": payload.get("plan"),
+        "read_model_route": payload.get("read_model_route"),
+        "next_step": payload.get("next_step"),
+        "history": payload.get("history"),
+    }
+
+
+def _source_correlation_review_description(context: dict[str, Any]) -> str:
+    return (
+        f"decision={context.get('decision')} "
+        f"work_item={context.get('work_item_id')} "
+        f"provider_event={context.get('latest_provider_event_id')} "
+        f"target_incident={context.get('target_incident_id') or '--'} "
+        f"writes_incident={context.get('writes_incident_state')} "
+        f"writes_source_event={context.get('writes_source_event')}"
+    )[:500]
+
+
+def build_source_correlation_review_decision(
+    recurrence: dict[str, Any],
+    *,
+    work_item_id: str,
+    decision: SourceCorrelationReviewDecision,
+    target_incident_id: str | None = None,
+    reviewer_id: str = "operator_console",
+    operator_note: str | None = None,
+) -> dict[str, Any]:
+    """Build a record-only source-correlation review decision payload."""
+
+    item, work_item = _find_recurrence_work_item(recurrence, work_item_id)
+    safe_decision = str(decision)
+    target_id = str(target_incident_id or "").strip() or None
+    checks = _source_correlation_review_checks(
+        item,
+        work_item,
+        decision=safe_decision,
+        target_incident_id=target_id,
+    )
+    allowed = all(check["passed"] for check in checks)
+    outcome = _source_correlation_review_outcome(safe_decision)
+    writes_timeline = bool(target_id and safe_decision == "accepted")
+
+    payload = {
+        "schema_version": _SOURCE_CORRELATION_DECISION_SCHEMA_VERSION,
+        "source": "channel_event_dossier.recurrence",
+        "project_id": recurrence.get("project_id"),
+        "work_item_id": work_item.get("work_item_id"),
+        "incident_id": work_item.get("incident_id"),
+        "target_incident_id": target_id,
+        "decision": safe_decision,
+        "review_status": outcome["review_status"],
+        "reviewer_id": str(reviewer_id or "operator_console")[:100],
+        "operator_note": str(operator_note or "").strip()[:500] or None,
+        "allowed": allowed,
+        "executed": allowed,
+        "safety_level": "source_review_record_only",
+        "writes_incident_state": False,
+        "writes_source_event": False,
+        "writes_auto_repair_result": False,
+        "writes_ticket": False,
+        "creates_external_ticket": False,
+        "latest_provider_event_id": item.get("latest_provider_event_id"),
+        "latest_stage": item.get("latest_stage"),
+        "provider": item.get("provider"),
+        "alertname": item.get("alertname"),
+        "namespace": item.get("namespace"),
+        "target_resource": item.get("target_resource"),
+        "checks": checks,
+        "current_state_summary": _recurrence_current_state_summary(item, work_item),
+        "plan": {
+            "step": "record_source_correlation_review_decision",
+            "flywheel_node": "source_correlation_review",
+            "agent_id": "awooop_source_correlation_reviewer",
+            "required_scope": "record_history",
+            "writes": ["alert_operation_log", *(["timeline_events"] if writes_timeline else [])],
+            "target_action": outcome["next_step"],
+            "reason": outcome["reason"],
+            "target": _recurrence_work_item_target(item),
+        },
+        "read_model_route": {
+            "agent_id": "awooop_source_correlation_reviewer",
+            "tool_name": "channel_event_dossier.recurrence",
+            "required_scope": "record_history",
+            "is_shadow": False,
+            "flywheel_node": "source_correlation_review",
+        },
+        "next_step": outcome["next_step"] if allowed else "fix_preflight_checks",
+    }
+    if not allowed:
+        payload["executed"] = False
+    return payload
+
+
+async def _record_source_correlation_review_decision_history(
+    payload: dict[str, Any],
+) -> dict[str, Any]:
+    if not payload.get("allowed"):
+        return {"recorded": False, "reason": "preflight_failed"}
+
+    incident_id = str(payload.get("target_incident_id") or "") or None
+    history: dict[str, Any] = {
+        "recorded": False,
+        "alert_operation_id": None,
+        "timeline_event_id": None,
+    }
+    context = _source_correlation_review_context(payload)
+
+    try:
+        from src.repositories.alert_operation_log_repository import (
+            get_alert_operation_log_repository,
+        )
+
+        record = await get_alert_operation_log_repository().append(
+            "USER_ACTION",
+            incident_id=incident_id,
+            actor=_SOURCE_CORRELATION_REVIEW_ACTOR,
+            action_detail=f"source_correlation_review:{payload.get('decision')}"[:200],
+            success=True,
+            context=_json_safe(context),
+        )
+        if record is not None:
+            history["alert_operation_id"] = getattr(record, "id", None)
+    except Exception as exc:
+        logger.warning(
+            "awooop_source_correlation_review_alert_operation_failed",
+            work_item_id=payload.get("work_item_id"),
+            error=str(exc),
+        )
+
+    if incident_id and payload.get("decision") == "accepted":
+        try:
+            from src.services.approval_db import get_timeline_service
+
+            event = await get_timeline_service().add_event(
+                event_type="human",
+                status="success",
+                title="AwoooP source correlation review",
+                description=_source_correlation_review_description(context),
+                actor=_SOURCE_CORRELATION_REVIEW_ACTOR,
+                actor_role="source_correlation_review",
+                incident_id=incident_id,
+            )
+            if event:
+                history["timeline_event_id"] = event.get("id")
+        except Exception as exc:
+            logger.warning(
+                "awooop_source_correlation_review_timeline_failed",
+                incident_id=incident_id,
+                work_item_id=payload.get("work_item_id"),
+                error=str(exc),
+            )
+    else:
+        history["timeline_reason"] = "review_not_incident_timeline_scoped"
+
+    history["recorded"] = bool(
+        history.get("alert_operation_id") or history.get("timeline_event_id")
+    )
+    if not history["recorded"]:
+        history["reason"] = "history_sink_unavailable"
+    return history
+
+
 def build_dossier_coverage(
     rows: list[dict[str, Any]],
     *,
@@ -1293,6 +1639,48 @@ async def _fetch_auto_repair_summaries_by_incident(
     return summaries
 
 
+async def _fetch_source_review_decisions_by_work_item(
+    db: Any,
+    *,
+    project_id: str,
+    limit: int,
+) -> dict[str, dict[str, Any]]:
+    """Fetch latest source-correlation review decisions from event-sourced audit."""
+
+    result = await db.execute(
+        text("""
+            SELECT DISTINCT ON (context->>'work_item_id')
+                id,
+                incident_id,
+                actor,
+                action_detail,
+                success,
+                context,
+                created_at
+            FROM alert_operation_log
+            WHERE actor = :actor
+              AND context->>'schema_version' = :schema_version
+              AND COALESCE(context->>'project_id', :project_id) = :project_id
+            ORDER BY context->>'work_item_id', created_at DESC
+            LIMIT :limit
+        """),
+        {
+            "actor": _SOURCE_CORRELATION_REVIEW_ACTOR,
+            "schema_version": _SOURCE_CORRELATION_DECISION_SCHEMA_VERSION,
+            "project_id": project_id,
+            "limit": max(1, min(limit, _MAX_RECURRENCE_EVENTS)),
+        },
+    )
+
+    decisions: dict[str, dict[str, Any]] = {}
+    for row in result.mappings().all():
+        item = _normalize_source_review_decision(dict(row))
+        if not item:
+            continue
+        decisions[str(item["work_item_id"])] = item
+    return decisions
+
+
 async def fetch_channel_event_dossier(
     *,
     project_id: str | None,
@@ -1479,12 +1867,18 @@ async def fetch_channel_event_dossier_recurrence(
             db,
             _collect_incident_ids_from_rows(rows),
         )
+        source_review_decisions = await _fetch_source_review_decisions_by_work_item(
+            db,
+            project_id=effective_project_id,
+            limit=safe_limit,
+        )
 
     return build_dossier_recurrence(
         rows,
         project_id=effective_project_id,
         limit=safe_limit,
         repair_summaries_by_incident=repair_summaries,
+        source_review_decisions_by_work_item=source_review_decisions,
     )
 
 
@@ -1562,3 +1956,41 @@ async def fetch_recurrence_work_item_handoff(
     elif payload.get("allowed"):
         payload["handoff_status"] = "record_failed"
     return payload
+
+
+async def fetch_source_correlation_review_decision(
+    *,
+    project_id: str | None,
+    work_item_id: str,
+    decision: SourceCorrelationReviewDecision,
+    target_incident_id: str | None = None,
+    reviewer_id: str = "operator_console",
+    operator_note: str | None = None,
+    provider: str | None = None,
+    limit: int = _MAX_RECURRENCE_EVENTS,
+) -> dict[str, Any]:
+    """Fetch recurrence state and record an explicit source review decision."""
+
+    recurrence = await fetch_channel_event_dossier_recurrence(
+        project_id=project_id,
+        provider=provider,
+        limit=limit,
+    )
+    payload = build_source_correlation_review_decision(
+        recurrence,
+        work_item_id=work_item_id,
+        decision=decision,
+        target_incident_id=target_incident_id,
+        reviewer_id=reviewer_id,
+        operator_note=operator_note,
+    )
+    payload["history"] = await _record_source_correlation_review_decision_history(
+        payload
+    )
+    if payload["history"].get("recorded"):
+        payload["review_record_status"] = "recorded"
+    elif payload.get("allowed"):
+        payload["review_record_status"] = "record_failed"
+    else:
+        payload["review_record_status"] = "blocked"
+    return payload