feat(telegram): audit monitoring alert coverage

apps/api/tests/test_telegram_alert_monitoring_coverage_readback_api.py

@@ -0,0 +1,218 @@
+from __future__ import annotations
+
+import json
+
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from src.api.v1 import agents
+from src.api.v1.agents import router
+from src.services.telegram_alert_monitoring_coverage_readback import (
+    _inspect_source_contract,
+    build_telegram_alert_monitoring_coverage_readback,
+)
+
+
+def _source_contract() -> dict[str, bool]:
+    return {
+        "alertmanager_alert_log_append_present": True,
+        "alert_operation_log_telegram_sent_present": True,
+        "alert_operation_log_repository_present": True,
+        "telegram_gateway_outbound_mirror_present": True,
+        "telegram_gateway_outbound_record_present": True,
+        "ai_alert_card_delivery_readback_present": True,
+        "ai_alert_card_learning_refs_present": True,
+        "post_apply_verifier_present": True,
+        "awooop_coverage_surface_present": True,
+    }
+
+
+def _payload() -> dict:
+    return build_telegram_alert_monitoring_coverage_readback(
+        project_id="awoooi",
+        monitoring_inventory={
+            "schema_version": "monitoring_alerting_observability_inventory_v1",
+            "summary": {
+                "surface_count": 60,
+                "source_exists_count": 60,
+                "live_evidence_received_count": 0,
+                "alert_rule_surface_count": 13,
+                "telegram_surface_count": 3,
+            },
+        },
+        monitoring_readback_plan={
+            "schema_version": "monitoring_post_incident_readback_plan_v1",
+            "summary": {
+                "readback_candidate_count": 60,
+            },
+        },
+        telegram_matrix={
+            "schema_version": "telegram_alert_ai_automation_matrix_v1",
+            "status": "telegram_alert_ai_automation_matrix_ready",
+            "summary": {
+                "telegram_alert_surface_count": 9,
+                "known_direct_send_gap_count": 0,
+                "gateway_normalized_callsite_count": 57,
+                "db_or_log_receipt_ready_surface_count": 9,
+                "ai_route_ready_surface_count": 9,
+                "controlled_queue_ready_surface_count": 9,
+                "post_verifier_ready_surface_count": 9,
+                "learning_writeback_ready_surface_count": 9,
+            },
+        },
+        post_apply_verifier={
+            "schema_version": "telegram_alert_learning_context_post_apply_verifier_v1",
+            "status": "telegram_alert_learning_context_post_apply_verified",
+            "rollups": {
+                "telegram_alert_learning_context_post_apply_verifier_ready": True,
+                "verified_context_receipt_count": 36,
+                "verified_target_count": 6,
+                "verified_ai_agent_context_receipt_count": 6,
+            },
+        },
+        ai_alert_card_delivery_readback={
+            "status": "ok",
+            "summary": {
+                "total": 36,
+                "learning_writeback_ready_total": 36,
+            },
+        },
+        runtime_log_readback={
+            "status": "ok",
+            "summary": {
+                "alert_operation_log_total_7d": 41,
+                "telegram_sent_event_count_7d": 8,
+                "km_converted_event_count_7d": 4,
+                "playbook_draft_event_count_7d": 3,
+            },
+        },
+        source_contract=_source_contract(),
+    )
+
+
+def test_telegram_alert_monitoring_coverage_readback_surfaces_live_gaps():
+    payload = _payload()
+
+    assert payload["schema_version"] == (
+        "telegram_alert_monitoring_coverage_readback_v1"
+    )
+    assert payload["priority_work_item_id"] == "CIR-P0-TG-001"
+    assert payload["status"] == (
+        "blocked_telegram_alert_monitoring_coverage_gaps_present"
+    )
+    assert payload["operator_answer"]["all_known_telegram_egress_routes_controlled"] is True
+    assert (
+        payload["operator_answer"][
+            "all_known_telegram_egress_routes_have_db_or_log_receipt"
+        ]
+        is True
+    )
+    assert (
+        payload["operator_answer"][
+            "all_verified_ai_alert_context_receipts_reusable_by_ai_agent"
+        ]
+        is True
+    )
+    assert (
+        payload["operator_answer"][
+            "all_monitoring_inventory_surfaces_have_accepted_live_receipt"
+        ]
+        is False
+    )
+    assert payload["operator_answer"]["manual_default_terminal_state_allowed"] is False
+
+    summary = payload["summary"]
+    assert summary["monitoring_surface_count"] == 60
+    assert summary["monitoring_live_receipt_gap_count"] == 60
+    assert summary["known_direct_send_gap_count"] == 0
+    assert summary["ai_alert_card_delivery_total"] == 36
+    assert summary["verified_context_receipt_count"] == 36
+    assert summary["verified_target_count"] == 6
+    assert summary["runtime_telegram_sent_event_count_7d"] == 8
+    assert summary["source_contract_ready"] is True
+    assert "monitoring_live_receipt_gap:60" in payload["active_blockers"]
+
+    matrix = {item["surface_id"]: item for item in payload["coverage_matrix"]}
+    assert matrix["telegram_gateway_outbound_mirror"]["gap_count"] == 0
+    assert matrix["telegram_alert_ai_loop_context_verifier"]["status"] == (
+        "verified_context_ready"
+    )
+    assert matrix["monitoring_inventory_static_scope"]["gap_count"] == 60
+    assert payload["operation_boundaries"]["telegram_send_performed"] is False
+    assert payload["operation_boundaries"]["raw_alert_payload_stored"] is False
+    assert payload["operation_boundaries"]["secret_value_read"] is False
+
+    serialized = json.dumps(payload, ensure_ascii=False)
+    for marker in [
+        "TELEGRAM_BOT_TOKEN",
+        "authorization_header",
+        "raw Telegram payload",
+        "批准！繼續",
+        "My request for Codex",
+    ]:
+        assert marker not in serialized
+
+
+def test_telegram_alert_monitoring_coverage_endpoint(monkeypatch):
+    async def fake_loader():
+        return _payload()
+
+    monkeypatch.setattr(
+        agents,
+        "load_latest_telegram_alert_monitoring_coverage_readback",
+        fake_loader,
+    )
+    app = FastAPI()
+    app.include_router(router, prefix="/api/v1")
+    client = TestClient(app)
+
+    response = client.get(
+        "/api/v1/agents/telegram-alert-monitoring-coverage-readback"
+    )
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["schema_version"] == (
+        "telegram_alert_monitoring_coverage_readback_v1"
+    )
+    assert data["summary"]["monitoring_live_receipt_gap_count"] == 60
+    assert data["summary"]["known_direct_send_gap_count"] == 0
+    assert data["operation_boundaries"]["bot_api_call_performed"] is False
+
+
+def test_telegram_alert_monitoring_coverage_source_contract(tmp_path):
+    (tmp_path / "apps/api/src/api/v1").mkdir(parents=True)
+    (tmp_path / "apps/api/src/repositories").mkdir(parents=True)
+    (tmp_path / "apps/api/src/services").mkdir(parents=True)
+    (tmp_path / "apps/web/src/components/awooop").mkdir(parents=True)
+    (tmp_path / "apps/api/src/api/v1/webhooks.py").write_text(
+        "get_alert_operation_log_repository\n"
+        '"TELEGRAM_SENT"\n',
+        encoding="utf-8",
+    )
+    (tmp_path / "apps/api/src/repositories/alert_operation_log_repository.py").write_text(
+        "ALERT_EVENT_TYPES\n",
+        encoding="utf-8",
+    )
+    (tmp_path / "apps/api/src/services/telegram_gateway.py").write_text(
+        "_mirror_outbound_message\n"
+        "record_outbound_message\n",
+        encoding="utf-8",
+    )
+    (tmp_path / "apps/api/src/services/platform_operator_service.py").write_text(
+        "list_ai_alert_card_delivery_readback\n"
+        "ai_alert_card_learning_writeback_refs_v1\n",
+        encoding="utf-8",
+    )
+    (tmp_path / "apps/api/src/services/telegram_alert_learning_context_post_apply_verifier.py").write_text(
+        "telegram_alert_learning_context_post_apply_verifier_v1\n",
+        encoding="utf-8",
+    )
+    (tmp_path / "apps/web/src/components/awooop/autonomous-runtime-receipt-panel.tsx").write_text(
+        "telegram-alert-monitoring-coverage-readback\n",
+        encoding="utf-8",
+    )
+
+    proof = _inspect_source_contract(tmp_path)
+
+    assert proof == _source_contract()