feat(alerting): add Telegram callback ingress sensor

apps/api/tests/test_telegram_callback_monitoring_contract.py

@@ -0,0 +1,164 @@
+from __future__ import annotations
+
+import re
+from pathlib import Path
+
+import pytest
+import yaml
+
+from src.core.metrics import (
+    TELEGRAM_CALLBACK_INGRESS_LAST_RECEIPT_TIMESTAMP,
+)
+from src.services import telegram_gateway as telegram_gateway_module
+from src.services.controlled_alert_target_router import resolve_typed_alert_target
+from src.services.telegram_gateway import TelegramGateway
+
+ROOT = Path(__file__).resolve().parents[3]
+CATALOG = ROOT / "apps" / "api" / "src" / "services" / "awooop_ansible_audit_service.py"
+POST_VERIFIER = (
+    ROOT / "apps" / "api" / "src" / "services" / "awooop_ansible_post_verifier.py"
+)
+PLAYBOOK = (
+    ROOT
+    / "infra"
+    / "ansible"
+    / "playbooks"
+    / "110-telegram-callback-ingress-sensor.yml"
+)
+RULE_SOURCES = (
+    ROOT / "ops" / "monitoring" / "alerts-unified.yml",
+    ROOT / "ops" / "monitoring" / "alerts.yml",
+    ROOT / "k8s" / "monitoring" / "alert-chain-monitor.yaml",
+)
+
+
+class _ConfiguredGateway(TelegramGateway):
+    @property
+    def bot_token(self) -> str:
+        return "configured"
+
+    @property
+    def chat_id(self) -> str:
+        return "configured"
+
+
+class _Redis:
+    def __init__(self) -> None:
+        self.values: dict[str, str] = {}
+
+    async def setex(self, key: str, _ttl: int, value: str) -> None:
+        self.values[key] = value
+
+    async def get(self, key: str) -> str | None:
+        return self.values.get(key)
+
+
+def _rule(path: Path) -> dict:
+    payload = yaml.safe_load(path.read_text(encoding="utf-8"))
+    groups = payload["spec"]["groups"] if "spec" in payload else payload["groups"]
+    return next(
+        rule
+        for group in groups
+        for rule in group.get("rules", [])
+        if rule.get("alert") == "TelegramCallbackIngressUnverified"
+    )
+
+
+def test_callback_ingress_sensor_is_identical_and_exact_typed_in_all_sources() -> None:
+    rules = [_rule(path) for path in RULE_SOURCES]
+    expressions = {re.sub(r"\s+", "", str(rule["expr"])) for rule in rules}
+
+    assert len(expressions) == 1
+    expression = expressions.pop()
+    assert "awoooi_telegram_callback_ingress_last_receipt_timestamp_seconds" in expression
+    assert ">300" in expression
+    assert "absent(" in expression
+
+    for rule in rules:
+        assert rule["for"] == "2m"
+        assert rule["labels"] == {
+            "severity": "warning",
+            "layer": "docker-188",
+            "component": "openclaw",
+            "host": "188",
+            "team": "platform",
+            "auto_repair": "true",
+            "alert_category": "ai_agent",
+            "canonical_asset_id": "service:openclaw:host188",
+            "domain_router": "docker_container",
+        }
+
+        route = resolve_typed_alert_target(
+            alertname="TelegramCallbackIngressUnverified",
+            target_resource=rule["labels"]["component"],
+            namespace="default",
+            labels=rule["labels"],
+            alert_category=rule["labels"]["alert_category"],
+        )
+        assert route["canonical_asset_id"] == "service:openclaw:host188"
+        assert route["allowed_catalog_ids"] == [
+            "ansible:188-openclaw-callback-forwarder"
+        ]
+        assert route["cross_domain_fallback_allowed"] is False
+
+
+def test_sensor_has_exact_host110_catalog_rollback_and_independent_verifier() -> None:
+    catalog = CATALOG.read_text(encoding="utf-8")
+    verifier = POST_VERIFIER.read_text(encoding="utf-8")
+    playbook = PLAYBOOK.read_text(encoding="utf-8")
+
+    assert '"catalog_id": "ansible:110-telegram-callback-ingress-sensor"' in catalog
+    assert '"inventory_hosts": ["host_110"]' in catalog
+    assert '"auto_apply_enabled": False' in catalog
+    assert '"canonical_asset_id": "service:prometheus"' in catalog
+    assert '"ansible:110-telegram-callback-ingress-sensor": (' in verifier
+    assert "host_110_telegram_callback_ingress_rule_healthy" in verifier
+    assert "host_110_telegram_callback_ingress_metric_visible" in verifier
+    assert "host_110_telegram_callback_ingress_rule_hash_converged" in verifier
+    assert "promtool" in playbook
+    assert "callback-sensor.bak" in playbook
+    assert "rescue:" in playbook
+    assert "Require exact rollback proof after any started apply" in playbook
+    assert "TelegramCallbackIngressUnverified" in playbook
+
+    play = yaml.safe_load(playbook)[0]
+    assert play["hosts"] == "host_110"
+    assert "--force-recreate" in playbook
+    assert "prometheus_runtime_rules" in playbook
+    assert "sensor_rollback_hashes" in playbook
+
+
+@pytest.mark.asyncio
+async def test_durable_callback_receipt_updates_prometheus_timestamp(
+    monkeypatch,
+) -> None:
+    redis = _Redis()
+    monkeypatch.setattr(telegram_gateway_module, "get_redis", lambda: redis)
+    TELEGRAM_CALLBACK_INGRESS_LAST_RECEIPT_TIMESTAMP.set(0)
+
+    gateway = _ConfiguredGateway()
+    await gateway.record_callback_ingress_receipt(
+        source="openclaw_188_hmac_forwarder",
+        update_id=99,
+    )
+
+    assert TELEGRAM_CALLBACK_INGRESS_LAST_RECEIPT_TIMESTAMP._value.get() > 0
+
+
+@pytest.mark.asyncio
+async def test_separate_scraped_replica_refreshes_shared_receipt_metric(
+    monkeypatch,
+) -> None:
+    redis = _Redis()
+    monkeypatch.setattr(telegram_gateway_module, "get_redis", lambda: redis)
+    writer = _ConfiguredGateway()
+    scraped_replica = _ConfiguredGateway()
+
+    await writer.record_callback_ingress_receipt(
+        source="openclaw_188_hmac_forwarder",
+        update_id=100,
+    )
+    TELEGRAM_CALLBACK_INGRESS_LAST_RECEIPT_TIMESTAMP.set(0)
+
+    assert await scraped_replica.refresh_callback_ingress_receipt() is True
+    assert TELEGRAM_CALLBACK_INGRESS_LAST_RECEIPT_TIMESTAMP._value.get() > 0