feat(awooop): verify source correlation links in status chain

apps/api/src/services/platform_operator_service.py

@@ -1332,18 +1332,24 @@ def _source_correlation_empty(
         "incident_ids": incident_ids,
         "direct_ref_total": 0,
         "candidate_total": 0,
+        "applied_link_total": 0,
         "provider_event_total": 0,
+        "latest_applied_link_at": None,
+        "verification_status": status_value,
         "providers": {
             provider: {
                 "direct_ref_total": 0,
                 "candidate_total": 0,
+                "applied_link_total": 0,
                 "latest_event_at": None,
                 "latest_heartbeat_at": None,
+                "latest_applied_link_at": None,
             }
             for provider in _SOURCE_CORRELATION_PROVIDERS
         },
         "top_candidates": [],
         "matching_criteria": [
+            "source_correlation_linked_stage",
             "direct_source_ref",
             "fingerprint_overlap",
             "alertname_overlap",
@@ -1557,6 +1563,17 @@ def _score_source_correlation_event(
     }
 
 
+def _is_source_correlation_applied_link(
+    event_context: dict[str, Any],
+    scored: dict[str, Any],
+) -> bool:
+    """Applied source links must be append-only events that still match directly."""
+    return (
+        str(event_context.get("stage") or "").lower() == "source_correlation_linked"
+        and bool(scored.get("is_direct"))
+    )
+
+
 async def _fetch_source_correlation_summary(
     *,
     project_id: str,
@@ -1689,6 +1706,19 @@ async def _fetch_source_correlation_summary(
             summary["candidate_total"] += 1
             provider_item["candidate_total"] += 1
 
+        is_applied_link = _is_source_correlation_applied_link(
+            event_context,
+            best_match,
+        )
+        if is_applied_link:
+            applied_at = _iso_or_none(row.get("received_at"))
+            summary["applied_link_total"] += 1
+            provider_item["applied_link_total"] += 1
+            if summary.get("latest_applied_link_at") is None:
+                summary["latest_applied_link_at"] = applied_at
+            if provider_item.get("latest_applied_link_at") is None:
+                provider_item["latest_applied_link_at"] = applied_at
+
         top_candidates.append(
             {
                 "provider": provider,
@@ -1696,19 +1726,40 @@ async def _fetch_source_correlation_summary(
                 "stage": str(event_context.get("stage") or ""),
                 "score": best_match["score"],
                 "match_type": "direct" if best_match["is_direct"] else "candidate",
+                "link_state": (
+                    "applied"
+                    if is_applied_link
+                    else "direct_ref"
+                    if best_match["is_direct"]
+                    else "candidate"
+                ),
+                "verification_status": (
+                    "applied_link_verified"
+                    if is_applied_link
+                    else "direct_ref_verified"
+                    if best_match["is_direct"]
+                    else "candidate_only"
+                ),
                 "reasons": best_match["reasons"],
                 "received_at": _iso_or_none(row.get("received_at")),
             }
         )
 
-    if summary["direct_ref_total"] > 0:
+    if summary["applied_link_total"] > 0:
         summary["status"] = "linked"
+        summary["verification_status"] = "applied_link_verified"
+        summary["missing_reason"] = None
+    elif summary["direct_ref_total"] > 0:
+        summary["status"] = "linked"
+        summary["verification_status"] = "direct_ref_verified"
         summary["missing_reason"] = None
     elif summary["candidate_total"] > 0:
         summary["status"] = "candidate_found"
+        summary["verification_status"] = "candidate_only"
         summary["missing_reason"] = None
     elif any(item.get("latest_heartbeat_at") for item in providers.values()):
         summary["status"] = "provider_fresh_no_match"
+        summary["verification_status"] = "provider_fresh_no_match"
         summary["missing_reason"] = "provider_heartbeat_present_but_no_incident_match"
 
     summary["top_candidates"] = sorted(

apps/api/tests/test_awooop_operator_timeline_labels.py

@@ -21,6 +21,7 @@ from src.services.platform_operator_service import (
     _callback_reply_event_item,
     _callback_reply_summary_matches_status,
     _collect_run_incident_ids,
+    _is_source_correlation_applied_link,
     _legacy_mcp_timeline_status,
     _legacy_mcp_timeline_summary,
     _list_filter_context_limit,
@@ -33,6 +34,7 @@ from src.services.platform_operator_service import (
     _run_callback_reply_summary,
     _run_remediation_list_summary,
     _score_source_correlation_event,
+    _source_event_correlation_context,
     _timeline_sort_key,
     _validate_ai_route_workload,
     _validate_callback_reply_action_filter,
@@ -612,12 +614,22 @@ def test_awooop_status_chain_includes_source_provider_correlation() -> None:
         source_correlation={
             "schema_version": "source_provider_correlation_v1",
             "status": "candidate_found",
+            "verification_status": "candidate_only",
             "direct_ref_total": 0,
             "candidate_total": 2,
+            "applied_link_total": 0,
             "provider_event_total": 2,
             "providers": {
-                "sentry": {"direct_ref_total": 0, "candidate_total": 1},
-                "signoz": {"direct_ref_total": 0, "candidate_total": 1},
+                "sentry": {
+                    "direct_ref_total": 0,
+                    "candidate_total": 1,
+                    "applied_link_total": 0,
+                },
+                "signoz": {
+                    "direct_ref_total": 0,
+                    "candidate_total": 1,
+                    "applied_link_total": 0,
+                },
             },
             "top_candidates": [
                 {
@@ -634,10 +646,56 @@ def test_awooop_status_chain_includes_source_provider_correlation() -> None:
     correlation = chain["source_refs"]["correlation"]
     assert correlation["status"] == "candidate_found"
     assert correlation["candidate_total"] == 2
+    assert correlation["verification_status"] == "candidate_only"
     assert correlation["providers"]["sentry"]["candidate_total"] == 1
     assert correlation["top_candidates"][0]["provider_event_id"] == "sentry:issue:1"
 
 
+def test_awooop_status_chain_preserves_applied_source_link_verification() -> None:
+    chain = _build_awooop_status_chain(
+        incident_ids=["INC-20260520-4D1124"],
+        source_id="INC-20260520-4D1124",
+        source_correlation={
+            "schema_version": "source_provider_correlation_v1",
+            "status": "linked",
+            "verification_status": "applied_link_verified",
+            "direct_ref_total": 1,
+            "candidate_total": 0,
+            "applied_link_total": 1,
+            "latest_applied_link_at": "2026-05-21T02:03:04",
+            "provider_event_total": 1,
+            "providers": {
+                "sentry": {
+                    "direct_ref_total": 1,
+                    "candidate_total": 0,
+                    "applied_link_total": 1,
+                    "latest_applied_link_at": "2026-05-21T02:03:04",
+                },
+            },
+            "top_candidates": [
+                {
+                    "provider": "sentry",
+                    "provider_event_id": "sentry:source_correlation_linked:issue-1",
+                    "stage": "source_correlation_linked",
+                    "score": 100,
+                    "match_type": "direct",
+                    "link_state": "applied",
+                    "verification_status": "applied_link_verified",
+                    "reasons": ["direct_incident_ref"],
+                }
+            ],
+        },
+    )
+
+    correlation = chain["source_refs"]["correlation"]
+    assert correlation["status"] == "linked"
+    assert correlation["verification_status"] == "applied_link_verified"
+    assert correlation["applied_link_total"] == 1
+    assert correlation["latest_applied_link_at"] == "2026-05-21T02:03:04"
+    assert correlation["providers"]["sentry"]["applied_link_total"] == 1
+    assert correlation["top_candidates"][0]["link_state"] == "applied"
+
+
 def test_source_correlation_scoring_distinguishes_direct_and_candidate() -> None:
     incident_context = {
         "incident_ids": ["INC-20260520-4D1124"],
@@ -684,6 +742,62 @@ def test_source_correlation_scoring_distinguishes_direct_and_candidate() -> None
     assert unrelated["is_candidate"] is False
 
 
+def test_source_correlation_applied_link_requires_stage_and_direct_match() -> None:
+    incident_context = {
+        "incident_ids": ["INC-20260520-4D1124"],
+        "alertnames": ["sentry issue"],
+        "severities": ["error"],
+        "fingerprints": ["fp-sentry-1"],
+        "namespaces": ["awoooi-prod"],
+        "targets": ["web"],
+    }
+    event_context = _source_event_correlation_context({
+        "provider": "sentry",
+        "provider_event_id": "sentry:source_correlation_linked:issue-1",
+        "received_at": datetime(2026, 5, 21, 2, 3, 4),
+        "source_envelope": {
+            "provider": "sentry",
+            "stage": "source_correlation_linked",
+            "source_refs": {
+                "incident_ids": ["INC-20260520-4D1124"],
+                "fingerprints": ["fp-sentry-1"],
+            },
+            "log_correlation": {
+                "alertname": "Sentry Issue",
+                "severity": "error",
+                "namespace": "awoooi-prod",
+                "target_resource": "web",
+            },
+        },
+    })
+    scored = _score_source_correlation_event(incident_context, event_context)
+
+    assert scored["is_direct"] is True
+    assert _is_source_correlation_applied_link(event_context, scored) is True
+
+    candidate_only_context = {
+        **event_context,
+        "incident_ids": [],
+        "fingerprints": [],
+    }
+    candidate_only = _score_source_correlation_event(
+        incident_context,
+        candidate_only_context,
+    )
+    non_link_stage = {
+        **event_context,
+        "stage": "upstream_canary",
+    }
+
+    assert candidate_only["is_direct"] is False
+    assert candidate_only["is_candidate"] is True
+    assert _is_source_correlation_applied_link(
+        candidate_only_context,
+        candidate_only,
+    ) is False
+    assert _is_source_correlation_applied_link(non_link_stage, scored) is False
+
+
 def test_awooop_status_chain_marks_read_only_manual_gate() -> None:
     chain = _build_awooop_status_chain(
         incident_ids=["INC-20260513-79ED5E"],