awoooi/apps/api/src/services/notification_matrix.py

"""Fail-closed Telegram notification routing policy.

The legacy ``TYPE-*`` matrix remains for existing callers, but an unknown type
no longer falls through to the shared SRE room.  Product monitoring routes are
loaded from a committed, machine-readable registry.  This module only returns
policy decisions: it never reads Telegram credentials or chat IDs, sends a
message, writes a receipt, or changes a runtime route.
"""

from __future__ import annotations

import json
import re
from collections.abc import Mapping
from dataclasses import dataclass
from datetime import UTC, datetime
from enum import Enum
from pathlib import Path
from typing import Any

from src.services.snapshot_paths import default_security_dir


class Destination(str, Enum):
    DM = "dm"  # legacy alias; formal alerts do not fall back to DM
    GROUP = "group"  # symbolic shared group selected by the caller
    BOTH = "both"  # legacy alias; retained for import compatibility
    BLOCKED = "blocked"


@dataclass(frozen=True)
class RoutingRule:
    destination: Destination
    strip_buttons_for_group: bool = False


# ADR-093 legacy notification types.  The global product routing registry below
# supersedes its unsafe "unknown -> shared group" default.
NOTIFICATION_ROUTING: dict[str, RoutingRule] = {
    "TYPE-1": RoutingRule(Destination.BLOCKED),
    "TYPE-2": RoutingRule(Destination.GROUP),
    "TYPE-3": RoutingRule(Destination.GROUP),
    "TYPE-4": RoutingRule(Destination.GROUP),
    "TYPE-4D": RoutingRule(Destination.GROUP),
    "TYPE-5S": RoutingRule(Destination.GROUP),
    "TYPE-6B": RoutingRule(Destination.BLOCKED),
    "TYPE-7E": RoutingRule(Destination.GROUP),
    "TYPE-8M": RoutingRule(Destination.GROUP),
}

_DEFAULT_RULE = RoutingRule(Destination.BLOCKED)
_LEGACY_CANONICAL_ROUTES: dict[str, tuple[str, str, str]] = {
    "TYPE-2": ("awoooi", "incident_lifecycle", "P1"),
    "TYPE-3": ("awoooi", "incident_lifecycle", "P1"),
    "TYPE-4": ("awoooi", "incident_lifecycle", "P1"),
    "TYPE-4D": ("awoooi", "security_incident", "P1"),
    "TYPE-5S": ("awoooi", "security_incident", "P1"),
    "TYPE-7E": ("awoooi", "incident_lifecycle", "P0"),
    "TYPE-8M": ("awoooi", "shared_infrastructure", "P1"),
}
_SCHEMA_VERSION = "telegram_canonical_routing_registry_v1"
_REGISTRY_FILENAME = "telegram-canonical-routing-registry.snapshot.json"
_DEFAULT_SECURITY_DIR = default_security_dir(Path(__file__))
_EXPECTED_PRODUCT_IDS = frozenset(
    {
        "2026fifa",
        "agent-bounty-protocol",
        "awooogo",
        "awoooi",
        "bitan-pharmacy",
        "clawbot-openclaw",
        "momo-pro",
        "stockplatform-v2",
        "tsenyang-website",
        "vibework",
        "vtuber",
    }
)
_ALLOWED_SEVERITIES = frozenset({"P0", "P1", "P2", "P3"})
_ALLOWED_EGRESS_STATUSES = frozenset(
    {"allowed", "blocked", "disabled", "not_implemented"}
)
_SHARED_SRE_CHAT_ALIAS = "awoooi_sre_war_room"
_SHARED_MONITORING_BOT_ALIAS = "tsenyang_bot"
_SHARED_MONITORING_BOT_OWNER = "sre_shared_runtime"
_SHARED_SRE_SIGNAL_FAMILIES = frozenset(
    {
        "shared_infrastructure",
        "security_incident",
        "disaster_recovery",
        "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",
        "product_id",
        "signal_family",
        "severity",
        "scope",
        "bot_alias",
        "chat_alias",
        "allowed_destination",
        "receipt_backend",
        "owner",
        "egress_status",
    }
)
_TOP_LEVEL_FIELDS = frozenset(
    {
        "schema_version",
        "generated_at",
        "status",
        "mode",
        "supersedes",
        "policy",
        "destination_roles",
        "products",
        "routes",
        "rollups",
        "execution_boundaries",
    }
)
_SUPERSEDES_FIELDS = frozenset(
    {
        "policy_ref",
        "superseded_behavior",
        "superseded_by",
        "owner",
        "expiry",
        "replacement",
        "verifier",
    }
)
_POLICY_FIELDS = frozenset(
    {
        "default_decision",
        "unknown_product_decision",
        "unknown_signal_family_decision",
        "unknown_severity_decision",
        "unknown_destination_decision",
        "legacy_runtime_resolver",
        "legacy_shared_sre_allowed_types",
        "legacy_shared_sre_blocked_types",
        "raw_chat_id_allowed",
        "shared_sre_rule",
        "shared_sre_forbidden",
    }
)
_DESTINATION_ROLE_FIELDS = frozenset(
    {
        "visible_label",
        "proven_runtime_alias",
        "runtime_alias_evidence",
        "chat_alias",
        "bot_alias",
        "bot_owner",
        "monitoring_owner",
        "role",
        "monitoring_egress_policy",
        "runtime_identifier_in_registry",
    }
)
_PRODUCT_FIELDS = frozenset(
    {
        "product_id",
        "display_name",
        "owner",
        "source_audit_status",
        "egress_status",
    }
)
_ROLLUP_FIELDS = frozenset(
    {
        "product_count",
        "route_count",
        "allowed_route_count",
        "blocked_route_count",
        "disabled_route_count",
        "not_implemented_route_count",
        "shared_sre_allowed_route_count",
        "raw_numeric_destination_count",
        "runtime_route_change_count",
        "telegram_send_count",
    }
)
_EXECUTION_BOUNDARY_FIELDS = frozenset(
    {
        "telegram_send_allowed",
        "bot_api_call_allowed",
        "secret_value_read_allowed",
        "raw_chat_id_storage_allowed",
        "runtime_route_change_allowed",
        "production_deploy_allowed",
    }
)
_SAFE_SENSITIVE_CONTROL_KEYS = frozenset(
    {
        "raw_chat_id_allowed",
        "raw_chat_id_storage_allowed",
        "secret_value_read_allowed",
    }
)
_SENSITIVE_KEY_PATTERN = re.compile(
    r"(?:^|_)(?:chat_id|bot_token|token|secret|secret_value|authorization(?:_header)?|"
    r"webhook_url|api_key|password|credential|private_key|cookie|session)(?:$|_)",
    re.IGNORECASE,
)
_RAW_NUMERIC_IDENTIFIER_PATTERN = re.compile(r"^-?\d{8,}$")
_BOT_TOKEN_VALUE_PATTERN = re.compile(r"\d{6,}:[A-Za-z0-9_-]{20,}")
_BOT_API_TOKEN_URL_PATTERN = re.compile(r"api\.telegram\.org/bot[^/\s]+", re.IGNORECASE)


def get_routing_rule(notification_type: str) -> RoutingRule:
    """Return a legacy route; unknown types are explicitly blocked."""
    return NOTIFICATION_ROUTING.get(notification_type, _DEFAULT_RULE)


def get_legacy_canonical_route_context(
    notification_type: str,
) -> dict[str, str] | None:
    """Map an allowlisted legacy type to explicit canonical route dimensions."""
    canonical_route = _LEGACY_CANONICAL_ROUTES.get(notification_type)
    if canonical_route is None:
        return None
    product_id, signal_family, severity = canonical_route
    return {
        "product_id": product_id,
        "signal_family": signal_family,
        "severity": severity,
        "legacy_notification_type": notification_type,
    }


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,
    group_chat_id: str,
    *,
    tg_group_cutover: bool = False,
    registry: dict[str, Any] | None = None,
) -> list[str]:
    """Resolve a legacy type through the canonical product policy."""
    _ = (dm_chat_id, tg_group_cutover)
    rule = get_routing_rule(notification_type)
    if rule.destination is not Destination.GROUP:
        return []

    canonical_route = get_legacy_canonical_route_context(notification_type)
    if canonical_route is None:
        return []
    try:
        decision = resolve_canonical_telegram_route(
            canonical_route["product_id"],
            canonical_route["signal_family"],
            canonical_route["severity"],
            requested_destination=("telegram_chat_alias:awoooi_sre_war_room"),
            registry=registry,
        )
    except (FileNotFoundError, TypeError, ValueError, json.JSONDecodeError):
        return []
    if decision["decision"] != "allow":
        return []
    return [group_chat_id] if group_chat_id else []


def load_canonical_telegram_routing_registry(
    security_dir: Path | None = None,
) -> dict[str, Any]:
    """Load and validate the committed 11-product routing registry."""
    registry_path = (security_dir or _DEFAULT_SECURITY_DIR) / _REGISTRY_FILENAME
    if not registry_path.is_file():
        raise FileNotFoundError(
            f"canonical Telegram routing registry missing: {registry_path}"
        )

    with registry_path.open(encoding="utf-8") as handle:
        payload = json.load(handle)

    if not isinstance(payload, dict):
        raise ValueError(f"{registry_path}: expected JSON object")
    _validate_canonical_registry(payload, str(registry_path))
    return payload


def build_canonical_telegram_routing_runtime_readback(
    security_dir: Path | None = None,
) -> dict[str, Any]:
    """Project the committed registry through the live fail-closed resolver.

    The source snapshot deliberately records the no-send boundaries of the run
    that created it.  Once the gateway guard is deployed, that historical
    ``source_status`` must not be presented as the current runtime state.  This
    readback resolves every route with the same code path used before Telegram
    provider calls and returns symbolic aliases only.
    """

    registry = load_canonical_telegram_routing_registry(security_dir)
    resolved_routes: list[dict[str, Any]] = []
    for route in registry["routes"]:
        decision = resolve_canonical_telegram_route(
            route["product_id"],
            route["signal_family"],
            route["severity"][0],
            requested_destination=route["allowed_destination"],
            registry=registry,
        )
        resolved_routes.append(
            {
                **route,
                "decision": decision["decision"],
                "decision_reason": decision["reason"],
            }
        )

    allowed_route_count = sum(
        route["decision"] == "allow" for route in resolved_routes
    )
    blocked_route_count = len(resolved_routes) - allowed_route_count
    unknown_probe = resolve_canonical_telegram_route(
        "unknown-product",
        "unknown-signal",
        "P1",
        registry=registry,
    )

    return {
        "schema_version": "telegram_canonical_routing_runtime_readback_v1",
        "status": "runtime_policy_enforced_product_delivery_receipts_partial",
        "source_status": registry["status"],
        "source_mode": registry["mode"],
        "source_generated_at": registry["generated_at"],
        "runtime_generated_at": datetime.now(UTC).isoformat(),
        "runtime_enforcement": {
            "canonical_resolver_active": True,
            "gateway_pre_send_gate_active": True,
            "trusted_ingress_context_required": True,
            "unknown_route_decision": unknown_probe["decision"],
            "unknown_route_reason": unknown_probe["reason"],
            "raw_numeric_destination_count": registry["rollups"][
                "raw_numeric_destination_count"
            ],
            "destination_identity_exposed": False,
            "all_product_delivery_receipts_ready": blocked_route_count == 0,
        },
        "rollups": {
            **registry["rollups"],
            "runtime_allowed_route_count": allowed_route_count,
            "runtime_blocked_route_count": blocked_route_count,
        },
        "policy": registry["policy"],
        "destination_roles": registry["destination_roles"],
        "products": registry["products"],
        "routes": resolved_routes,
        "operation_boundaries": {
            "read_only": True,
            "telegram_send_performed": False,
            "bot_api_call_performed": False,
            "runtime_route_changed": False,
            "secret_value_read": False,
            "raw_identifier_included": False,
        },
        "source_refs": {
            "registry": "docs/security/telegram-canonical-routing-registry.snapshot.json",
            "resolver": "apps/api/src/services/notification_matrix.py",
            "gateway_gate": "apps/api/src/services/telegram_gateway.py",
            "verifier": "apps/api/tests/test_telegram_canonical_routing_registry.py",
        },
    }


def resolve_canonical_telegram_route(
    product_id: str,
    signal_family: str,
    severity: str,
    *,
    requested_destination: str | None = None,
    registry: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Return one symbolic route decision, defaulting to a no-egress terminal."""
    payload = (
        load_canonical_telegram_routing_registry() if registry is None else registry
    )
    _validate_canonical_registry(payload, "in-memory canonical Telegram registry")

    product = next(
        (
            product
            for product in payload["products"]
            if product["product_id"] == product_id
        ),
        None,
    )

    normalized_severity = severity.strip().upper()
    candidates = [
        route
        for route in payload["routes"]
        if route["product_id"] == product_id
        and route["signal_family"] in {signal_family, "*"}
        and normalized_severity in route["severity"]
    ]
    candidates.sort(key=lambda route: route["signal_family"] == "*", reverse=False)

    if not candidates:
        return _blocked_decision(
            product_id=product_id,
            signal_family=signal_family,
            severity=normalized_severity,
            reason="unknown_product_signal_or_severity",
        )

    route = dict(candidates[0])
    route["decision"] = (
        "allow"
        if product is not None
        and product["egress_status"] == "allowed"
        and route["egress_status"] == "allowed"
        else "block"
    )
    route["reason"] = route.get("blocked_reason") or "canonical_route_allowed"

    if (
        requested_destination is not None
        and requested_destination != route["allowed_destination"]
    ):
        route.update(
            {
                "decision": "block",
                "egress_status": "blocked",
                "reason": "requested_destination_not_allowed",
                "requested_destination": requested_destination,
            }
        )
    return route


def _blocked_decision(
    *, product_id: str, signal_family: str, severity: str, reason: str
) -> dict[str, Any]:
    return {
        "route_id": None,
        "product_id": product_id,
        "signal_family": signal_family,
        "severity": [severity],
        "scope": "unknown",
        "bot_alias": "none",
        "chat_alias": "none",
        "allowed_destination": "blocked",
        "receipt_backend": "none",
        "owner": "unassigned",
        "egress_status": "blocked",
        "decision": "block",
        "reason": reason,
    }


def _validate_canonical_registry(payload: dict[str, Any], label: str) -> None:
    _require_exact_object_fields(payload, _TOP_LEVEL_FIELDS, label)
    if payload.get("schema_version") != _SCHEMA_VERSION:
        raise ValueError(f"{label}: expected schema_version={_SCHEMA_VERSION}")
    if payload.get("status") != "source_policy_ready_runtime_not_applied":
        raise ValueError(f"{label}: unsupported status")
    if payload.get("mode") != "source_only_no_secret_no_telegram_send_no_route_change":
        raise ValueError(f"{label}: unsupported mode")
    if not isinstance(payload.get("generated_at"), str) or not payload["generated_at"]:
        raise ValueError(f"{label}: generated_at must be a non-empty string")

    _validate_no_sensitive_runtime_material(payload, label)

    supersedes = payload["supersedes"]
    _require_exact_object_fields(
        supersedes,
        _SUPERSEDES_FIELDS,
        f"{label}.supersedes",
    )
    for field in _SUPERSEDES_FIELDS:
        if not isinstance(supersedes[field], str) or not supersedes[field].strip():
            raise ValueError(f"{label}.supersedes.{field} must be non-empty")
    if supersedes["superseded_by"] != "global_product_governance_v2":
        raise ValueError(f"{label}: superseded_by must be global governance v2")

    policy = payload["policy"]
    _require_exact_object_fields(policy, _POLICY_FIELDS, f"{label}.policy")
    for field in (
        "default_decision",
        "unknown_product_decision",
        "unknown_signal_family_decision",
        "unknown_severity_decision",
        "unknown_destination_decision",
    ):
        if policy[field] != "block":
            raise ValueError(f"{label}.policy.{field} must be block")
    if policy["raw_chat_id_allowed"] is not False:
        raise ValueError(f"{label}.policy.raw_chat_id_allowed must be false")
    expected_legacy_allowed = set(_LEGACY_CANONICAL_ROUTES)
    legacy_allowed = policy["legacy_shared_sre_allowed_types"]
    legacy_blocked = policy["legacy_shared_sre_blocked_types"]
    shared_forbidden = policy["shared_sre_forbidden"]
    if (
        not isinstance(legacy_allowed, list)
        or len(legacy_allowed) != len(set(legacy_allowed))
        or set(legacy_allowed) != expected_legacy_allowed
    ):
        raise ValueError(f"{label}: legacy allowed types drifted")
    if (
        not isinstance(legacy_blocked, list)
        or len(legacy_blocked) != len(set(legacy_blocked))
        or set(legacy_blocked) != {"TYPE-1", "TYPE-6B", "UNKNOWN"}
    ):
        raise ValueError(f"{label}: legacy blocked types drifted")
    if not isinstance(shared_forbidden, list) or not shared_forbidden or any(
        not isinstance(item, str) or not item.strip() for item in shared_forbidden
    ):
        raise ValueError(f"{label}: shared_sre_forbidden must be non-empty strings")
    for field in ("legacy_runtime_resolver", "shared_sre_rule"):
        if not isinstance(policy[field], str) or not policy[field].strip():
            raise ValueError(f"{label}.policy.{field} must be non-empty")

    boundaries = payload["execution_boundaries"]
    _require_exact_object_fields(
        boundaries,
        _EXECUTION_BOUNDARY_FIELDS,
        f"{label}.execution_boundaries",
    )
    unsafe = sorted(
        key for key in _EXECUTION_BOUNDARY_FIELDS if boundaries[key] is not False
    )
    if unsafe:
        raise ValueError(f"{label}: execution boundaries must remain false: {unsafe}")

    destination_roles = payload["destination_roles"]
    if not isinstance(destination_roles, list) or not destination_roles:
        raise ValueError(f"{label}: destination_roles must be a non-empty array")
    observed_role_bindings: set[tuple[str, str]] = set()
    for index, role in enumerate(destination_roles):
        role_label = f"{label}.destination_roles[{index}]"
        _require_exact_object_fields(role, _DESTINATION_ROLE_FIELDS, role_label)
        for field in (
            "visible_label",
            "proven_runtime_alias",
            "runtime_alias_evidence",
            "chat_alias",
            "bot_alias",
            "bot_owner",
            "role",
            "monitoring_egress_policy",
        ):
            if not isinstance(role[field], str) or not role[field].strip():
                raise ValueError(f"{role_label}.{field} must be non-empty")
        if role["runtime_identifier_in_registry"] is not False:
            raise ValueError(
                f"{role_label}.runtime_identifier_in_registry must be false"
            )
        if not isinstance(role["monitoring_owner"], bool):
            raise ValueError(f"{role_label}.monitoring_owner must be boolean")
        if role["runtime_alias_evidence"] == "visible_label_only":
            if role["proven_runtime_alias"] != "none":
                raise ValueError(
                    f"{role_label}: visible label must not claim a proven runtime alias"
                )
        elif role["runtime_alias_evidence"] == "proven_runtime_receipt":
            if role["proven_runtime_alias"] == "none":
                raise ValueError(
                    f"{role_label}: proven runtime receipt requires an alias"
                )
        elif role["runtime_alias_evidence"] not in {
            "source_config_only",
            "configured_unreachable",
        }:
            raise ValueError(f"{role_label}: unsupported runtime_alias_evidence")
        for field in ("chat_alias", "bot_alias"):
            if _RAW_NUMERIC_IDENTIFIER_PATTERN.fullmatch(role[field]):
                raise ValueError(f"{role_label}: raw numeric runtime identifier")
        binding = (role["chat_alias"], role["bot_alias"])
        if binding in observed_role_bindings:
            raise ValueError(f"{role_label}: duplicate symbolic destination binding")
        observed_role_bindings.add(binding)

    products = payload["products"]
    routes = payload["routes"]
    if not isinstance(products, list) or not isinstance(routes, list):
        raise ValueError(f"{label}: products and routes must be arrays")

    for index, product in enumerate(products):
        product_label = f"{label}.products[{index}]"
        _require_exact_object_fields(product, _PRODUCT_FIELDS, product_label)
        for field in ("product_id", "display_name", "owner", "source_audit_status"):
            if not isinstance(product[field], str) or not product[field].strip():
                raise ValueError(f"{product_label}.{field} must be non-empty")
        if product["egress_status"] not in _ALLOWED_EGRESS_STATUSES:
            raise ValueError(f"{product_label}: unsupported product egress_status")
    product_ids = [product["product_id"] for product in products]
    if len(product_ids) != len(set(product_ids)):
        raise ValueError(f"{label}: duplicate product_id")
    if set(product_ids) != _EXPECTED_PRODUCT_IDS:
        raise ValueError(
            f"{label}: product registry must cover the canonical 11 products"
        )

    route_ids: set[str] = set()
    covered_products: set[str] = set()
    for index, route in enumerate(routes):
        route_label = f"{label}.routes[{index}]"
        _require_exact_object_fields(
            route,
            _REQUIRED_ROUTE_FIELDS,
            route_label,
            optional_fields=frozenset({"blocked_reason"}),
        )

        route_id = str(route["route_id"])
        if not route_id or route_id in route_ids:
            raise ValueError(
                f"{label}: route_id must be non-empty and unique: {route_id!r}"
            )
        route_ids.add(route_id)

        product_id = str(route["product_id"])
        if product_id not in _EXPECTED_PRODUCT_IDS:
            raise ValueError(f"{label}: unknown product_id in route: {product_id}")
        covered_products.add(product_id)

        severity = route["severity"]
        if (
            not isinstance(severity, list)
            or not severity
            or len(severity) != len(set(severity))
        ):
            raise ValueError(f"{label}: {route_id} severity must be a non-empty array")
        if not set(severity).issubset(_ALLOWED_SEVERITIES):
            raise ValueError(f"{label}: {route_id} has unsupported severity")

        egress_status = str(route["egress_status"])
        if egress_status not in _ALLOWED_EGRESS_STATUSES:
            raise ValueError(f"{label}: {route_id} has unsupported egress_status")
        if route["scope"] not in {"shared", "product"}:
            raise ValueError(f"{label}: {route_id} has unsupported scope")
        for field in (
            "route_id",
            "product_id",
            "signal_family",
            "scope",
            "bot_alias",
            "chat_alias",
            "allowed_destination",
            "receipt_backend",
            "owner",
            "egress_status",
        ):
            if not isinstance(route[field], str) or not route[field].strip():
                raise ValueError(f"{route_label}.{field} must be non-empty")

        _validate_no_runtime_identifiers(route, label, route_id)
        _validate_route_destination(route, label, route_id)
        _validate_shared_sre_boundary(route, label, route_id)

    if covered_products != _EXPECTED_PRODUCT_IDS:
        raise ValueError(
            f"{label}: every canonical product requires at least one route"
        )

    by_product = {product["product_id"]: product for product in products}
    for product_id, product in by_product.items():
        expected_status = product.get("egress_status")
        if expected_status not in _ALLOWED_EGRESS_STATUSES:
            raise ValueError(
                f"{label}: {product_id} has unsupported product egress_status"
            )
        product_routes = [
            route for route in routes if route["product_id"] == product_id
        ]
        allowed_route_count = sum(
            route["egress_status"] == "allowed" for route in product_routes
        )
        if expected_status != "allowed" and allowed_route_count:
            raise ValueError(f"{label}: {product_id} must not have an allowed route")
        if expected_status == "allowed" and allowed_route_count == 0:
            raise ValueError(
                f"{label}: {product_id} requires at least one allowed route"
            )
        if expected_status in {"disabled", "not_implemented"} and any(
            route["egress_status"] != expected_status for route in product_routes
        ):
            raise ValueError(
                f"{label}: {product_id} route status must match product disposition"
            )

    # Ownership is a cross-object invariant. Validate it after product-level
    # disposition so a disabled product cannot be made allowed merely by also
    # forging a destination-role binding.
    for route in routes:
        _validate_route_role_ownership(
            route,
            destination_roles,
            label,
            str(route["route_id"]),
        )

    _validate_registry_rollups(payload, label)


def _require_exact_object_fields(
    value: object,
    required_fields: frozenset[str],
    label: str,
    *,
    optional_fields: frozenset[str] = frozenset(),
) -> None:
    if not isinstance(value, dict):
        raise ValueError(f"{label}: expected JSON object")
    missing = sorted(required_fields - value.keys())
    unexpected = sorted(value.keys() - required_fields - optional_fields)
    if missing:
        raise ValueError(f"{label}: missing required fields: {missing}")
    if unexpected:
        raise ValueError(f"{label}: unexpected fields: {unexpected}")


def _validate_no_runtime_identifiers(
    route: dict[str, Any], label: str, route_id: str
) -> None:
    forbidden_fields = {"chat_id", "bot_token", "token", "secret", "webhook_url"}
    present = sorted(forbidden_fields & route.keys())
    if present:
        raise ValueError(
            f"{label}: {route_id} contains forbidden runtime fields: {present}"
        )

    for field in ("bot_alias", "chat_alias", "allowed_destination"):
        value = str(route[field])
        if value.lstrip("-").isdigit():
            raise ValueError(
                f"{label}: {route_id} must not contain a raw Telegram identifier"
            )


def _validate_no_sensitive_runtime_material(payload: Any, label: str) -> None:
    """Reject secret-shaped keys/values and raw numeric identifiers recursively."""

    def walk(value: Any, path: str) -> None:
        if isinstance(value, dict):
            for raw_key, nested in value.items():
                key = str(raw_key)
                child_path = f"{path}.{key}"
                if _SENSITIVE_KEY_PATTERN.search(key):
                    safe_control = (
                        key in _SAFE_SENSITIVE_CONTROL_KEYS and nested is False
                    )
                    if not safe_control:
                        raise ValueError(
                            f"{label}: forbidden sensitive runtime key at {child_path}"
                        )
                walk(nested, child_path)
            return
        if isinstance(value, list):
            for index, nested in enumerate(value):
                walk(nested, f"{path}[{index}]")
            return
        if isinstance(value, bool) or value is None:
            return
        if isinstance(value, int):
            if abs(value) >= 1_000_000_000:
                raise ValueError(f"{label}: raw numeric runtime identifier at {path}")
            return
        if not isinstance(value, str):
            return
        if _RAW_NUMERIC_IDENTIFIER_PATTERN.fullmatch(value):
            raise ValueError(f"{label}: raw numeric runtime identifier at {path}")
        if _BOT_TOKEN_VALUE_PATTERN.search(value) or _BOT_API_TOKEN_URL_PATTERN.search(
            value
        ):
            raise ValueError(f"{label}: secret-shaped runtime value at {path}")

    walk(payload, "registry")


def _validate_route_destination(
    route: dict[str, Any], label: str, route_id: str
) -> None:
    status = route["egress_status"]
    destination = route["allowed_destination"]
    if status == "allowed":
        expected = f"telegram_chat_alias:{route['chat_alias']}"
        if destination != expected:
            raise ValueError(
                f"{label}: {route_id} allowed destination must match chat_alias"
            )
        if route["bot_alias"] == "none" or route["chat_alias"] == "none":
            raise ValueError(
                f"{label}: {route_id} allowed route requires symbolic aliases"
            )
        if route["receipt_backend"] == "none":
            raise ValueError(
                f"{label}: {route_id} allowed route requires a receipt backend"
            )
    else:
        if destination != "blocked":
            raise ValueError(f"{label}: {route_id} non-allowed route must be blocked")
        if not route.get("blocked_reason"):
            raise ValueError(
                f"{label}: {route_id} non-allowed route requires blocked_reason"
            )


def _validate_shared_sre_boundary(
    route: dict[str, Any], label: str, route_id: str
) -> None:
    if route["chat_alias"] != _SHARED_SRE_CHAT_ALIAS:
        return
    invalid = (
        route["product_id"] != "awoooi"
        or route["scope"] != "shared"
        or route["signal_family"] not in _SHARED_SRE_SIGNAL_FAMILIES
        or not set(route["severity"]).issubset({"P0", "P1"})
        or route["bot_alias"] != _SHARED_MONITORING_BOT_ALIAS
    )
    if invalid:
        raise ValueError(
            f"{label}: {route_id} violates shared SRE P0/P1 infra/security/DR/lifecycle boundary"
        )


def _validate_route_role_ownership(
    route: dict[str, Any],
    destination_roles: list[dict[str, Any]],
    label: str,
    route_id: str,
) -> None:
    """Cross-check every allowed route against its declared destination owner."""
    if route["egress_status"] != "allowed":
        return
    role = next(
        (
            item
            for item in destination_roles
            if item["chat_alias"] == route["chat_alias"]
            and item["bot_alias"] == route["bot_alias"]
        ),
        None,
    )
    if role is None:
        raise ValueError(
            f"{label}: {route_id} has no matching destination_roles ownership binding"
        )
    if role["monitoring_egress_policy"] != "allowlist_only":
        raise ValueError(
            f"{label}: {route_id} destination role is not monitoring allowlisted"
        )
    if role["monitoring_owner"] is not True:
        raise ValueError(
            f"{label}: {route_id} destination role is not the monitoring owner"
        )
    if role["bot_owner"] != _SHARED_MONITORING_BOT_OWNER:
        raise ValueError(
            f"{label}: {route_id} bot ownership does not match shared monitoring policy"
        )


def _validate_registry_rollups(payload: dict[str, Any], label: str) -> None:
    rollups = payload["rollups"]
    _require_exact_object_fields(
        rollups,
        _ROLLUP_FIELDS,
        f"{label}.rollups",
    )
    routes = payload["routes"]
    expected = {
        "product_count": len(payload["products"]),
        "route_count": len(routes),
        "allowed_route_count": sum(
            route["egress_status"] == "allowed" for route in routes
        ),
        "blocked_route_count": sum(
            route["egress_status"] == "blocked" for route in routes
        ),
        "disabled_route_count": sum(
            route["egress_status"] == "disabled" for route in routes
        ),
        "not_implemented_route_count": sum(
            route["egress_status"] == "not_implemented" for route in routes
        ),
        "shared_sre_allowed_route_count": sum(
            route["egress_status"] == "allowed"
            and route["chat_alias"] == _SHARED_SRE_CHAT_ALIAS
            for route in routes
        ),
        "raw_numeric_destination_count": 0,
        "runtime_route_change_count": 0,
        "telegram_send_count": 0,
    }
    drifted = sorted(
        key for key, value in expected.items() if rollups.get(key) != value
    )
    if drifted:
        raise ValueError(f"{label}: registry rollup mismatch: {drifted}")