fix(telegram): bind product alerts to canonical routes

apps/api/src/services/notification_matrix.py

@@ -11,6 +11,7 @@ from __future__ import annotations
 
 import json
 import re
+from collections.abc import Mapping
 from dataclasses import dataclass
 from enum import Enum
 from pathlib import Path
@@ -89,6 +90,23 @@ _SHARED_SRE_SIGNAL_FAMILIES = frozenset(
         "incident_lifecycle",
     }
 )
+_TRUSTED_ALERT_ROUTE_SOURCES = frozenset(
+    {
+        "hmac_alert_webhook",
+        "prometheus_alertmanager",
+        "sentry_webhook",
+    }
+)
+_ROUTE_SEVERITY_BY_SOURCE_VALUE = {
+    "critical": "P0",
+    "fatal": "P0",
+    "error": "P1",
+    "high": "P1",
+    "warning": "P2",
+    "medium": "P2",
+    "info": "P3",
+    "low": "P3",
+}
 _REQUIRED_ROUTE_FIELDS = frozenset(
     {
         "route_id",
@@ -230,6 +248,44 @@ def get_legacy_canonical_route_context(
     }
 
 
+def get_trusted_alert_canonical_route_context(
+    labels: Mapping[str, object] | None,
+    *,
+    source: str,
+    source_severity: str,
+) -> dict[str, str] | None:
+    """Extract an explicit product route from controlled, namespaced labels.
+
+    Product monitoring must never inherit the legacy TYPE-3 AWOOOI route.  A
+    recognized ingress may opt into canonical routing only with the committed
+    ``awoooi_*`` label triplet.  Missing or malformed labels return ``None`` so
+    the caller can emit a durable blocked/no-egress receipt.
+    """
+
+    if source not in _TRUSTED_ALERT_ROUTE_SOURCES or not isinstance(labels, Mapping):
+        return None
+
+    product_id = str(labels.get("awoooi_product_id") or "").strip()
+    signal_family = str(labels.get("awoooi_signal_family") or "").strip()
+    if not product_id or not signal_family:
+        return None
+
+    explicit_severity = str(labels.get("awoooi_route_severity") or "").strip().upper()
+    route_severity = explicit_severity or _ROUTE_SEVERITY_BY_SOURCE_VALUE.get(
+        str(source_severity or "").strip().lower(),
+        "",
+    )
+    if route_severity not in _ALLOWED_SEVERITIES:
+        return None
+
+    return {
+        "product_id": product_id,
+        "signal_family": signal_family,
+        "severity": route_severity,
+        "route_source": source,
+    }
+
+
 def resolve_chat_ids(
     notification_type: str,
     dm_chat_id: str,