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

"""Telegram alert AI automation matrix readback.

Builds a repo-committed, metadata-only matrix that shows whether Telegram alert
surfaces are wired into AI automation receipts instead of ending as manual
messages. This module never sends Telegram, calls Bot API, reads secrets, or
writes runtime queues.
"""

from __future__ import annotations

import json
from pathlib import Path
from typing import Any

from src.services.snapshot_paths import (
    default_evaluations_dir,
    default_security_dir,
    resolve_repo_root,
)

_SCHEMA_VERSION = "telegram_alert_ai_automation_matrix_v1"
_EGRESS_INVENTORY = "telegram-notification-egress-inventory.snapshot.json"
_OWNER_ACCEPTANCE = "telegram-notification-egress-owner-response-acceptance.snapshot.json"
_MIGRATION_PLAN = "telegram-notification-egress-migration-plan-draft.snapshot.json"
_READABILITY_GUARD = "telegram-alert-readability-guard.snapshot.json"
_DIGEST_POLICY_PATTERN = "ai_agent_telegram_action_required_digest_policy_*.json"
_AWOOOP_VERIFIED_CONTEXT_SURFACE_SOURCE = (
    "apps/web/src/components/awooop/autonomous-runtime-receipt-panel.tsx"
)
_AWOOOP_VERIFIED_CONTEXT_SURFACE_MARKER = (
    "telegram-alert-post-apply-verifier-readback"
)


def load_latest_telegram_alert_ai_automation_matrix(
    security_dir: Path | None = None,
    evaluations_dir: Path | None = None,
) -> dict[str, Any]:
    """Return the latest Telegram alert AI automation matrix."""
    anchor = Path(__file__)
    security = security_dir or default_security_dir(anchor)
    evaluations = evaluations_dir or default_evaluations_dir(anchor)
    repo_root = resolve_repo_root(anchor)

    egress_inventory = _load_required_json(security / _EGRESS_INVENTORY)
    owner_acceptance = _load_required_json(security / _OWNER_ACCEPTANCE)
    migration_plan = _load_required_json(security / _MIGRATION_PLAN)
    readability_guard = _load_required_json(security / _READABILITY_GUARD)
    digest_policy = _load_latest_json(evaluations, _DIGEST_POLICY_PATTERN)

    _require_schema(
        egress_inventory,
        "telegram_notification_egress_inventory_v1",
        _EGRESS_INVENTORY,
    )
    _require_schema(
        owner_acceptance,
        "telegram_notification_egress_owner_response_acceptance_v1",
        _OWNER_ACCEPTANCE,
    )
    _require_schema(
        migration_plan,
        "telegram_notification_egress_migration_plan_draft_v1",
        _MIGRATION_PLAN,
    )
    _require_schema(
        readability_guard,
        "telegram_alert_readability_guard_v1",
        _READABILITY_GUARD,
    )
    _require_schema(
        digest_policy,
        "ai_agent_telegram_action_required_digest_policy_v1",
        _DIGEST_POLICY_PATTERN,
    )
    _require_safe_boundaries(
        egress_inventory,
        owner_acceptance,
        migration_plan,
        readability_guard,
        digest_policy,
    )

    egress_summary = egress_inventory["summary"]
    owner_summary = owner_acceptance["summary"]
    migration_summary = migration_plan["summary"]
    readability_summary = readability_guard["summary"]
    digest_rollups = digest_policy["rollups"]
    source_proof = _inspect_source_contract(repo_root)
    source_contract_missing_markers = list(
        source_proof.get("source_contract_missing_markers") or []
    )
    learning_writeback_refs_present = bool(
        source_proof["ai_alert_card_learning_writeback_refs_present"]
    )
    learning_registry_present = bool(
        source_proof["ai_alert_card_learning_registry_readback_present"]
    )
    ai_loop_context_present = bool(
        source_proof["telegram_alert_ai_loop_consumer_context_readback_present"]
    )
    post_apply_verifier_present = bool(
        source_proof["telegram_alert_post_apply_verifier_readback_present"]
    )
    post_apply_verifier_surface_present = bool(
        source_proof["telegram_alert_post_apply_verifier_awooop_surface_present"]
    )

    direct_gap_count = int(egress_summary["direct_bot_api_call_count"])
    direct_gap_file_count = int(egress_summary["direct_bot_api_file_count"])
    api_direct_gap_count = int(egress_summary["api_direct_bot_api_call_count"])
    ops_script_gap_count = int(egress_summary["ops_script_direct_bot_api_call_count"])
    migration_candidate_count = int(migration_summary["migration_candidate_count"])
    direct_gap_closed = direct_gap_count == 0
    direct_gap_refs = [
        {
            "surface_id": item["egress_surface_id"],
            "surface_kind": item["surface_kind"],
            "path": item["path"],
            "line": item["line"],
            "line_hash": item["line_hash"],
        }
        for item in egress_inventory.get("direct_bot_api_calls", [])
    ]

    matrix = [
        {
            "surface_id": "telegram_gateway_ai_alert_cards",
            "priority_work_item_id": "CIR-P0-TG-001",
            "surface_kind": "gateway_final_exit",
            "status": "ai_controlled_receipt_path_ready",
            "known_item_count": int(readability_summary["final_exit_contract_count"]),
            "db_or_log_receipt": "ready_via_awooop_outbound_message_mirror",
            "ai_route": "ready_ai_automation_alert_card_v1",
            "controlled_queue": "ready_controlled_playbook_queue",
            "post_verifier": "ready_telegram_alert_readability_guard",
            "learning_writeback": "ready_learning_writeback_refs",
            "manual_default_gap_count": 0,
            "direct_send_gap_count": 0,
            "evidence_refs": [
                "docs/security/telegram-alert-readability-guard.snapshot.json",
                "apps/api/src/services/telegram_gateway.py",
                "/api/v1/platform/runs/ai-alert-cards",
            ],
        },
        {
            "surface_id": "telegram_direct_bot_api_egress_gap",
            "priority_work_item_id": "CIR-P0-TG-001",
            "surface_kind": "direct_bot_api_gap",
            "status": (
                "gap_closed_no_direct_bot_api_routes_detected"
                if direct_gap_closed
                else "gap_open_requires_controlled_migration"
            ),
            "known_item_count": direct_gap_count,
            "db_or_log_receipt": (
                "ready_no_direct_routes_remaining"
                if direct_gap_closed
                else "missing_or_not_proven_for_direct_path"
            ),
            "ai_route": (
                "ready_no_direct_routes_remaining"
                if direct_gap_closed
                else "missing_or_not_proven_for_direct_path"
            ),
            "controlled_queue": (
                "ready_not_applicable_no_direct_migration_queue_needed"
                if direct_gap_closed
                else "missing_or_not_proven_for_direct_path"
            ),
            "post_verifier": (
                "ready_metadata_inventory_clear"
                if direct_gap_closed
                else "metadata_inventory_ready"
            ),
            "learning_writeback": (
                "ready_bind_gateway_receipts_to_learning_writeback"
                if direct_gap_closed
                else "migration_candidate_ready_not_applied"
            ),
            "manual_default_gap_count": direct_gap_count,
            "direct_send_gap_count": direct_gap_count,
            "evidence_refs": [
                "docs/security/telegram-notification-egress-inventory.snapshot.json",
                "docs/security/telegram-notification-egress-migration-plan-draft.snapshot.json",
            ],
        },
        {
            "surface_id": "telegram_action_required_digest_policy",
            "priority_work_item_id": "CIR-P0-TG-001",
            "surface_kind": "ai_digest_policy",
            "status": "no_send_policy_ready",
            "known_item_count": int(digest_rollups["rule_count"]),
            "db_or_log_receipt": "ready_source_policy_receipt_no_runtime_write",
            "ai_route": "ready_action_required_and_failure_only_rules",
            "controlled_queue": "ready_policy_gate_no_direct_send",
            "post_verifier": "ready_policy_loader_validation",
            "learning_writeback": "ready_next_task_ids",
            "manual_default_gap_count": 0,
            "direct_send_gap_count": 0,
            "evidence_refs": [
                "docs/evaluations/ai_agent_telegram_action_required_digest_policy_2026-06-11.json",
                "/api/v1/agents/agent-telegram-action-required-digest-policy",
            ],
        },
        {
            "surface_id": "telegram_egress_controlled_migration_candidates",
            "priority_work_item_id": "CIR-P0-TG-001",
            "surface_kind": "migration_candidate",
            "status": (
                "controlled_migration_complete_no_candidates"
                if migration_candidate_count == 0
                else "controlled_apply_candidates_ready"
            ),
            "known_item_count": migration_candidate_count,
            "db_or_log_receipt": (
                "ready_no_direct_migration_candidates"
                if migration_candidate_count == 0
                else "required_before_candidate_close"
            ),
            "ai_route": (
                "ready_no_direct_migration_candidates"
                if migration_candidate_count == 0
                else "migration_plan_ready"
            ),
            "controlled_queue": (
                "ready_not_applicable_no_direct_migration_queue_needed"
                if migration_candidate_count == 0
                else "ready_for_controlled_apply_review"
            ),
            "post_verifier": (
                "ready_owner_acceptance_ledger_clear"
                if migration_candidate_count == 0
                else "owner_acceptance_ledger_ready"
            ),
            "learning_writeback": (
                "ready_bind_gateway_receipts_to_learning_writeback"
                if migration_candidate_count == 0
                else "candidate_acceptance_metadata_ready"
            ),
            "manual_default_gap_count": 0,
            "direct_send_gap_count": direct_gap_count,
            "evidence_refs": [
                "docs/security/telegram-notification-egress-owner-response-acceptance.snapshot.json",
                "docs/security/telegram-notification-egress-migration-plan-draft.snapshot.json",
            ],
        },
        {
            "surface_id": "telegram_ai_alert_card_delivery_readback_endpoint",
            "priority_work_item_id": "CIR-P0-TG-001",
            "surface_kind": "production_readback_endpoint",
            "status": "readback_endpoint_present",
            "known_item_count": source_proof["ai_alert_card_delivery_readback_endpoint_count"],
            "db_or_log_receipt": "ready_awooop_outbound_message_query",
            "ai_route": "ready_event_type_lane_target_filters",
            "controlled_queue": "ready_not_applicable_read_only_query_no_send",
            "post_verifier": "ready_source_contract_present",
            "learning_writeback": "ready_delivery_receipt_summary_and_learning_writeback_refs",
            "manual_default_gap_count": 0,
            "direct_send_gap_count": 0,
            "evidence_refs": [
                "apps/api/src/api/v1/platform/operator_runs.py",
                "apps/api/src/db/awooop_models.py",
            ],
        },
        {
            "surface_id": "telegram_alert_learning_registry_readback",
            "priority_work_item_id": "CIR-P0-TG-001",
            "surface_kind": "learning_registry_readback",
            "status": (
                "registry_readback_present"
                if learning_registry_present
                else "registry_readback_missing"
            ),
            "known_item_count": source_proof[
                "ai_alert_card_learning_registry_readback_count"
            ],
            "db_or_log_receipt": "ready_awooop_outbound_message_learning_refs",
            "ai_route": "ready_ai_loop_agent_consumer_context",
            "controlled_queue": "ready_metadata_only_consumer_binding",
            "post_verifier": "ready_registry_source_contract_present",
            "learning_writeback": (
                "ready_km_rag_playbook_mcp_verifier_ai_agent_registry"
                if learning_registry_present
                else "missing_registry_readback"
            ),
            "manual_default_gap_count": 0,
            "direct_send_gap_count": 0,
            "evidence_refs": [
                "apps/api/src/services/platform_operator_service.py",
                "/api/v1/platform/runs/ai-alert-cards",
            ],
        },
        {
            "surface_id": "telegram_ai_loop_agent_consumer_context",
            "priority_work_item_id": "CIR-P0-TG-001",
            "surface_kind": "ai_loop_agent_consumer_context",
            "status": (
                "ai_loop_context_readback_present"
                if ai_loop_context_present
                else "ai_loop_context_readback_missing"
            ),
            "known_item_count": source_proof[
                "telegram_alert_ai_loop_consumer_context_readback_count"
            ],
            "db_or_log_receipt": "ready_ai_alert_card_registry_receipts",
            "ai_route": (
                "ready_ai_loop_agent_context_reuse"
                if ai_loop_context_present
                else "missing_ai_loop_agent_context_reuse"
            ),
            "controlled_queue": (
                "ready_metadata_only_context_receipt"
                if ai_loop_context_present
                else "missing_context_receipt"
            ),
            "post_verifier": (
                "ready_ai_loop_context_source_contract_present"
                if ai_loop_context_present
                else "missing_ai_loop_context_source_contract"
            ),
            "learning_writeback": (
                "ready_agent_context_receipt_readback"
                if ai_loop_context_present
                else "missing_agent_context_receipt_readback"
            ),
            "manual_default_gap_count": 0,
            "direct_send_gap_count": 0,
            "evidence_refs": [
                "apps/api/src/services/ai_agent_log_controlled_writeback_consumer_readback.py",
                "/api/v1/agents/agent-log-controlled-writeback-consumer-readback",
            ],
        },
        {
            "surface_id": "telegram_ai_loop_post_apply_verifier",
            "priority_work_item_id": "CIR-P0-TG-001",
            "surface_kind": "ai_loop_post_apply_verifier",
            "status": (
                "post_apply_verifier_readback_present"
                if post_apply_verifier_present
                else "post_apply_verifier_readback_missing"
            ),
            "known_item_count": source_proof[
                "telegram_alert_post_apply_verifier_readback_count"
            ],
            "db_or_log_receipt": "ready_ai_loop_context_receipt_source",
            "ai_route": (
                "ready_verified_context_reuse"
                if post_apply_verifier_present
                else "missing_verified_context_reuse"
            ),
            "controlled_queue": (
                "ready_metadata_only_verifier"
                if post_apply_verifier_present
                else "missing_metadata_only_verifier"
            ),
            "post_verifier": (
                "ready_context_receipt_post_apply_verifier"
                if post_apply_verifier_present
                else "missing_context_receipt_post_apply_verifier"
            ),
            "learning_writeback": (
                "ready_verified_context_receipt_for_km_playbook_trust"
                if post_apply_verifier_present
                else "missing_verified_context_receipt"
            ),
            "manual_default_gap_count": 0,
            "direct_send_gap_count": 0,
            "evidence_refs": [
                "apps/api/src/services/telegram_alert_learning_context_post_apply_verifier.py",
                "/api/v1/agents/telegram-alert-learning-context-post-apply-verifier",
            ],
        },
        {
            "surface_id": "telegram_verified_context_awooop_operator_surface",
            "priority_work_item_id": "CIR-P0-TG-001",
            "surface_kind": "awooop_operator_surface",
            "status": (
                "verified_context_receipt_surface_present"
                if post_apply_verifier_surface_present
                else "verified_context_receipt_surface_missing"
            ),
            "known_item_count": source_proof[
                "telegram_alert_post_apply_verifier_awooop_surface_count"
            ],
            "db_or_log_receipt": "ready_consumes_post_apply_verifier_readback",
            "ai_route": (
                "ready_operator_visible_ai_loop_receipts"
                if post_apply_verifier_surface_present
                else "missing_operator_visible_ai_loop_receipts"
            ),
            "controlled_queue": "ready_read_only_ui_surface_no_send_no_apply",
            "post_verifier": (
                "ready_awooop_surface_source_contract_present"
                if post_apply_verifier_surface_present
                else "missing_awooop_surface_source_contract"
            ),
            "learning_writeback": (
                "ready_human_observable_ai_agent_learning_receipts"
                if post_apply_verifier_surface_present
                else "missing_human_observable_ai_agent_learning_receipts"
            ),
            "manual_default_gap_count": 0,
            "direct_send_gap_count": 0,
            "evidence_refs": [
                "apps/web/src/components/awooop/autonomous-runtime-receipt-panel.tsx",
                "/zh-TW/awooop/runs",
                "/zh-TW/awooop/work-items",
                "/zh-TW/awooop/alerts",
            ],
        },
    ]

    next_priority_action = _next_priority_action(
        api_direct_gap_count=api_direct_gap_count,
        ops_script_gap_count=ops_script_gap_count,
        direct_gap_count=direct_gap_count,
        learning_writeback_refs_present=learning_writeback_refs_present,
        learning_registry_present=learning_registry_present,
        ai_loop_context_present=ai_loop_context_present,
        post_apply_verifier_present=post_apply_verifier_present,
        post_apply_verifier_surface_present=post_apply_verifier_surface_present,
    )
    next_controlled_actions = _ordered_controlled_actions(next_priority_action)
    summary = _build_summary(
        matrix=matrix,
        direct_gap_count=direct_gap_count,
        direct_gap_file_count=direct_gap_file_count,
        egress_summary=egress_summary,
        owner_summary=owner_summary,
        migration_summary=migration_summary,
        readability_summary=readability_summary,
        digest_rollups=digest_rollups,
        source_proof=source_proof,
        next_priority_action=next_priority_action,
    )

    return {
        "schema_version": _SCHEMA_VERSION,
        "status": (
            "telegram_alert_ai_automation_matrix_ready_with_direct_send_gaps"
            if direct_gap_count
            else "blocked_telegram_alert_ai_automation_matrix_source_contract_gaps"
            if source_contract_missing_markers
            else "telegram_alert_ai_automation_matrix_ready"
        ),
        "mode": "metadata_only_no_secret_no_telegram_send_no_runtime_write",
        "priority_work_item_id": "CIR-P0-TG-001",
        "source_refs": {
            "egress_inventory": f"docs/security/{_EGRESS_INVENTORY}",
            "owner_acceptance": f"docs/security/{_OWNER_ACCEPTANCE}",
            "migration_plan": f"docs/security/{_MIGRATION_PLAN}",
            "readability_guard": f"docs/security/{_READABILITY_GUARD}",
            "digest_policy": "docs/evaluations/ai_agent_telegram_action_required_digest_policy_2026-06-11.json",
        },
        "operation_boundaries": {
            "telegram_send_performed": False,
            "bot_api_call_performed": False,
            "gateway_queue_write_performed": False,
            "runtime_write_performed": False,
            "secret_value_read": False,
            "raw_payload_stored": False,
            "receiver_route_changed": False,
            "workflow_dispatch_performed": False,
            "production_write_performed": False,
            "metadata_only": True,
        },
        "summary": summary,
        "matrix": matrix,
        "direct_send_gap_refs": direct_gap_refs,
        "next_controlled_actions": next_controlled_actions,
        "operator_interpretation": [
            "TelegramGateway AI alert cards already expose AI route, controlled queue, final-exit formatter, outbound mirror metadata, and readback endpoint support.",
            (
                "Known direct Bot API surfaces are clear; keep future Telegram sends on the gateway receipt path."
                if direct_gap_closed
                else "Known direct Bot API surfaces remain the concrete AI automation gap; API direct senders are clear when api_direct_send_gap_count is 0."
            ),
            (
                (
                    (
                        (
                            "Manual/default semantics are only acceptable for critical/break-glass or historical evidence; "
                            + (
                                (
                                    "Telegram alert AI Loop context receipts are visible in AwoooP operator surfaces; "
                                    "the next P0 step is full monitoring alert DB/log coverage audit."
                                )
                                if post_apply_verifier_surface_present
                                else (
                                    "Telegram alert AI Loop context has a post-apply verifier readback, "
                                    "so the next P0 step is surfacing verified receipts in Runs / Work Items / Alerts."
                                )
                            )
                        )
                        if post_apply_verifier_present
                        else (
                            "Manual/default semantics are only acceptable for critical/break-glass or historical evidence; "
                            "Telegram alert registry is now consumable by AI Loop Agent context, "
                            "so the next P0 step is post-apply verifier receipt."
                        )
                    )
                    if ai_loop_context_present
                    else (
                        "Manual/default semantics are only acceptable for critical/break-glass or historical evidence; "
                        "KM / PlayBook / RAG / MCP / verifier / AI Agent registry readback is present, "
                        "so the next P0 step is AI Loop Agent consumer context."
                    )
                    if learning_registry_present
                    else (
                        "Manual/default semantics are only acceptable for critical/break-glass or historical evidence; "
                        "learning refs are present, so the next P0 step is KM / PlayBook / RAG / MCP registry readback."
                    )
                    if learning_writeback_refs_present
                    else "Manual/default semantics are only acceptable for critical/break-glass or historical evidence; the next P0 step is learning/writeback readback."
                )
                if direct_gap_closed
                else "Manual/default semantics are only acceptable for critical/break-glass or historical evidence; direct send gaps remain controlled migration candidates."
            ),
        ],
    }


def _ordered_controlled_actions(next_priority_action: dict[str, Any]) -> list[dict[str, Any]]:
    actions = [
        next_priority_action,
        {
            "action_id": "migrate_api_direct_sender_to_gateway",
            "scope": "apps/api/src/services/channel_hub.py",
            "priority": "P0",
            "why": "API direct Bot API path is the remaining high-blast-radius bypass.",
            "requires_secret": False,
            "requires_runtime_send": False,
            "post_verifier": "telegram-notification-egress-inventory direct gap count decreases and TelegramGateway mirror tests pass.",
        },
        {
            "action_id": "migrate_ops_script_direct_senders_to_receipt_path",
            "scope": "scripts/ops direct Bot API sendMessage surfaces",
            "priority": "P0",
            "why": "Ops script direct sends can bypass DB/log receipt and AI automation card metadata.",
            "requires_secret": False,
            "requires_runtime_send": False,
            "post_verifier": "egress inventory and owner acceptance matrix show no unreviewed direct sends.",
        },
        _legacy_bind_ai_alert_cards_action(),
    ]
    ordered: list[dict[str, Any]] = []
    seen: set[str] = set()
    for action in actions:
        action_id = str(action["action_id"])
        if action_id in seen:
            continue
        seen.add(action_id)
        ordered.append(action)
    return ordered


def _load_required_json(path: Path) -> dict[str, Any]:
    if not path.exists():
        raise FileNotFoundError(f"required Telegram matrix source missing: {path}")
    with path.open(encoding="utf-8") as handle:
        payload = json.load(handle)
    if not isinstance(payload, dict):
        raise ValueError(f"{path}: expected JSON object")
    return payload


def _load_latest_json(directory: Path, pattern: str) -> dict[str, Any]:
    candidates = sorted(directory.glob(pattern))
    if not candidates:
        raise FileNotFoundError(f"no matching snapshots found in {directory}: {pattern}")
    return _load_required_json(candidates[-1])


def _require_schema(payload: dict[str, Any], expected: str, label: str) -> None:
    actual = payload.get("schema_version")
    if actual != expected:
        raise ValueError(f"{label}: expected schema_version={expected}, got {actual!r}")


def _require_safe_boundaries(*payloads: dict[str, Any]) -> None:
    forbidden_true_keys = {
        "telegram_send_authorized",
        "bot_api_call_authorized",
        "telegram_direct_send_allowed",
        "telegram_gateway_queue_write_allowed",
        "runtime_execution_authorized",
        "runtime_execution_allowed",
        "runtime_gate_open",
        "workflow_dispatch_authorized",
        "production_deploy_authorized",
        "production_write_authorized",
        "secret_value_collection_allowed",
        "secret_plaintext_allowed",
        "raw_payload_storage_allowed",
        "conversation_transcript_allowed",
    }
    for payload in payloads:
        for section_key in (
            "operation_boundaries",
            "execution_boundaries",
            "approval_boundaries",
        ):
            section = payload.get(section_key) or {}
            if not isinstance(section, dict):
                continue
            unsafe = sorted(
                key
                for key in forbidden_true_keys
                if section.get(key) not in (None, False)
            )
            if unsafe:
                raise ValueError(
                    f"{payload.get('schema_version')}.{section_key}: unsafe true flags {unsafe}"
                )


def _inspect_source_contract(repo_root: Path) -> dict[str, Any]:
    source_checks = {
        "telegram_gateway_mirror_present": (
            "apps/api/src/services/telegram_gateway.py",
            "_mirror_outbound_message",
        ),
        "telegram_gateway_record_outbound_present": (
            "apps/api/src/services/telegram_gateway.py",
            "record_outbound_message",
        ),
        "ai_alert_card_metadata_present": (
            "apps/api/src/services/telegram_gateway.py",
            "ai_automation_alert_card_mirror_v1",
        ),
        "outbound_message_model_present": (
            "apps/api/src/db/awooop_models.py",
            "__tablename__ = \"awooop_outbound_message\"",
        ),
        "ai_alert_card_delivery_readback_endpoint_present": (
            "apps/api/src/api/v1/platform/operator_runs.py",
            "/runs/ai-alert-cards",
        ),
        "ai_alert_card_learning_writeback_refs_present": (
            "apps/api/src/api/v1/platform/operator_runs.py",
            "learning_writeback_refs",
        ),
        "ai_alert_card_learning_writeback_summary_present": (
            "apps/api/src/api/v1/platform/operator_runs.py",
            "learning_writeback_ready_total",
        ),
        "ai_alert_card_learning_registry_readback_present": (
            "apps/api/src/services/platform_operator_service.py",
            "ai_alert_card_learning_registry_readback_v1",
        ),
        "ai_alert_card_ai_agent_context_ref_present": (
            "apps/api/src/services/platform_operator_service.py",
            "ai_agent_context_ref",
        ),
        "telegram_alert_ai_loop_consumer_context_readback_present": (
            "apps/api/src/services/ai_agent_log_controlled_writeback_consumer_readback.py",
            "telegram_alert_ai_loop_consumer_context_readback_v1",
        ),
        "telegram_alert_post_apply_verifier_readback_present": (
            "apps/api/src/services/telegram_alert_learning_context_post_apply_verifier.py",
            "telegram_alert_learning_context_post_apply_verifier_v1",
        ),
        "telegram_alert_post_apply_verifier_awooop_surface_present": (
            _AWOOOP_VERIFIED_CONTEXT_SURFACE_SOURCE,
            _AWOOOP_VERIFIED_CONTEXT_SURFACE_MARKER,
        ),
    }
    result: dict[str, int | bool] = {}
    for key, (relative_path, marker) in source_checks.items():
        text = _read_source_contract_text(repo_root, relative_path)
        present = marker in text
        result[key] = present
        if key == "ai_alert_card_delivery_readback_endpoint_present":
            result["ai_alert_card_delivery_readback_endpoint_count"] = 1 if present else 0
        if key == "ai_alert_card_learning_writeback_refs_present":
            result["ai_alert_card_learning_writeback_refs_count"] = 1 if present else 0
        if key == "ai_alert_card_learning_registry_readback_present":
            result["ai_alert_card_learning_registry_readback_count"] = (
                1 if present else 0
            )
        if key == "telegram_alert_ai_loop_consumer_context_readback_present":
            result["telegram_alert_ai_loop_consumer_context_readback_count"] = (
                1 if present else 0
            )
        if key == "telegram_alert_post_apply_verifier_readback_present":
            result["telegram_alert_post_apply_verifier_readback_count"] = (
                1 if present else 0
            )
        if key == "telegram_alert_post_apply_verifier_awooop_surface_present":
            result["telegram_alert_post_apply_verifier_awooop_surface_count"] = (
                1 if present else 0
            )

    required = [
        key
        for key, value in result.items()
        if key.endswith("_present") and value is not True
    ]
    result["source_contract_ready"] = not required
    result["source_contract_missing_markers"] = required
    return result


def _read_source_contract_text(repo_root: Path, relative_path: str) -> str:
    candidates = [repo_root / relative_path]
    api_prefix = "apps/api/"
    if relative_path.startswith(api_prefix):
        candidates.append(repo_root / relative_path.removeprefix(api_prefix))
    if (
        relative_path == _AWOOOP_VERIFIED_CONTEXT_SURFACE_SOURCE
        and not (repo_root / "apps").exists()
    ):
        candidates.append(repo_root / "src/services/telegram_alert_ai_automation_matrix.py")

    for path in candidates:
        if path.exists():
            return path.read_text(encoding="utf-8", errors="replace")
    return ""


def _next_priority_action(
    *,
    api_direct_gap_count: int,
    ops_script_gap_count: int,
    direct_gap_count: int,
    learning_writeback_refs_present: bool,
    learning_registry_present: bool,
    ai_loop_context_present: bool,
    post_apply_verifier_present: bool,
    post_apply_verifier_surface_present: bool,
) -> dict[str, Any]:
    if api_direct_gap_count > 0:
        return {
            "action_id": "migrate_api_direct_sender_to_gateway",
            "scope": "apps/api/src/services/channel_hub.py",
            "priority": "P0",
            "why": "API direct Bot API path is the remaining high-blast-radius bypass.",
            "requires_secret": False,
            "requires_runtime_send": False,
            "post_verifier": "telegram-notification-egress-inventory direct gap count decreases and TelegramGateway mirror tests pass.",
        }
    if ops_script_gap_count > 0:
        return {
            "action_id": "migrate_ops_script_direct_senders_to_receipt_path",
            "scope": "scripts/ops direct Bot API sendMessage surfaces",
            "priority": "P0",
            "why": "Only ops script direct Bot API paths remain after API sender convergence.",
            "requires_secret": False,
            "requires_runtime_send": False,
            "post_verifier": "telegram-notification-egress-inventory reports ops=0 and owner acceptance candidates=0 for direct Bot API surfaces.",
        }
    if direct_gap_count > 0:
        return {
            "action_id": "migrate_remaining_direct_senders_to_receipt_path",
            "scope": "remaining direct Bot API sendMessage surfaces",
            "priority": "P0",
            "why": "Direct Bot API paths still bypass the controlled receipt path.",
            "requires_secret": False,
            "requires_runtime_send": False,
            "post_verifier": "telegram-notification-egress-inventory direct gap count reaches 0.",
        }
    if not learning_writeback_refs_present:
        return {
            "action_id": "bind_ai_alert_cards_to_learning_writeback",
            "scope": "ai_automation_alert_card_v1 outbound mirror metadata",
            "priority": "P0",
            "why": "All direct Bot API gaps are clear; close the learning/writeback loop.",
            "requires_secret": False,
            "requires_runtime_send": False,
            "post_verifier": "/api/v1/platform/runs/ai-alert-cards readback shows delivery_receipt_readback_required and learning writeback refs.",
        }
    if not learning_registry_present:
        return {
            "action_id": "promote_telegram_alert_learning_refs_to_km_rag_mcp_registry",
            "scope": "AI alert card receipt refs -> KM / PlayBook / RAG / MCP registry readback",
            "priority": "P0",
            "why": "Alert cards now expose learning refs; the next closure is registry/readback integration for AI Loop Agent reuse.",
            "requires_secret": False,
            "requires_runtime_send": False,
            "post_verifier": "KM / PlayBook / RAG / MCP registry readback includes CIR-P0-TG-001 alert-card receipt refs.",
        }
    if not ai_loop_context_present:
        return {
            "action_id": "consume_telegram_alert_learning_registry_in_ai_loop_agent_context",
            "scope": "KM / PlayBook / RAG / MCP / verifier registry -> AI Loop Agent consumer context",
            "priority": "P0",
            "why": "Telegram alert learning registry is readable; the next closure is AI Loop Agent consumer context reuse and apply receipt.",
            "requires_secret": False,
            "requires_runtime_send": False,
            "post_verifier": "AI Loop Agent context readback references the CIR-P0-TG-001 Telegram alert learning registry.",
        }
    if not post_apply_verifier_present:
        return {
            "action_id": "verify_telegram_alert_learning_context_receipts_with_ai_loop_post_apply_verifier",
            "scope": "AI Loop Agent Telegram alert consumer context -> post-apply verifier receipt",
            "priority": "P0",
            "why": "Telegram alert learning registry is readable by AI Loop Agent context; the next closure is verifier receipt and controlled apply reuse.",
            "requires_secret": False,
            "requires_runtime_send": False,
            "post_verifier": "AI Loop post-apply verifier reads Telegram alert context receipts and reports reusable decision context.",
        }
    if not post_apply_verifier_surface_present:
        return {
            "action_id": "surface_verified_telegram_alert_context_receipts_in_awooop_runs_work_items_alerts",
            "scope": "Verified Telegram alert context receipts -> AwoooP Runs / Work Items / Alerts",
            "priority": "P0",
            "why": "Post-apply verifier is source-controlled; the next closure is visible AI automation proof across operator surfaces.",
            "requires_secret": False,
            "requires_runtime_send": False,
            "post_verifier": "Runs, Work Items, and Alerts surfaces show verified Telegram alert AI Loop receipts without manual-default terminal states.",
        }
    return {
        "action_id": "audit_all_telegram_monitoring_alerts_db_log_ai_automation_coverage",
        "scope": "Telegram monitoring alerts -> DB/log ledger, KM/RAG/MCP/PlayBook, and AI Agent context coverage",
        "priority": "P0",
        "why": "Verified Telegram alert AI Loop receipts are now visible in AwoooP; the next closure is proving every monitoring alert path is persisted and reusable by AI Agent automation.",
        "requires_secret": False,
        "requires_runtime_send": False,
        "post_verifier": "Telegram alert matrix shows every monitoring alert surface has DB/log receipt, AI route, controlled queue, and no manual-default terminal state.",
    }


def _legacy_bind_ai_alert_cards_action() -> dict[str, Any]:
    return {
        "action_id": "bind_ai_alert_cards_to_learning_writeback",
        "scope": "ai_automation_alert_card_v1 outbound mirror metadata",
        "priority": "P0",
        "why": "All direct Bot API gaps are clear; close the learning/writeback loop.",
        "requires_secret": False,
        "requires_runtime_send": False,
        "post_verifier": "/api/v1/platform/runs/ai-alert-cards readback shows delivery_receipt_readback_required and learning writeback refs.",
    }


def _build_summary(
    *,
    matrix: list[dict[str, Any]],
    direct_gap_count: int,
    direct_gap_file_count: int,
    egress_summary: dict[str, Any],
    owner_summary: dict[str, Any],
    migration_summary: dict[str, Any],
    readability_summary: dict[str, Any],
    digest_rollups: dict[str, Any],
    source_proof: dict[str, Any],
    next_priority_action: dict[str, Any],
) -> dict[str, Any]:
    def ready(key: str, prefix: str) -> int:
        return sum(
            1 for item in matrix if str(item[key]).startswith(prefix)
        )

    return {
        "telegram_alert_surface_count": len(matrix),
        "known_direct_send_gap_count": direct_gap_count,
        "known_direct_send_gap_file_count": direct_gap_file_count,
        "workflow_direct_send_gap_count": int(egress_summary["workflow_direct_bot_api_call_count"]),
        "ops_script_direct_send_gap_count": int(egress_summary["ops_script_direct_bot_api_call_count"]),
        "ci_script_direct_send_gap_count": int(egress_summary["ci_script_direct_bot_api_call_count"]),
        "api_direct_send_gap_count": int(egress_summary["api_direct_bot_api_call_count"]),
        "gateway_normalized_callsite_count": int(egress_summary["gateway_normalized_callsite_count"]),
        "db_or_log_receipt_ready_surface_count": ready("db_or_log_receipt", "ready"),
        "ai_route_ready_surface_count": ready("ai_route", "ready"),
        "controlled_queue_ready_surface_count": ready("controlled_queue", "ready"),
        "post_verifier_ready_surface_count": ready("post_verifier", "ready"),
        "learning_writeback_ready_surface_count": ready("learning_writeback", "ready"),
        "manual_default_gap_count": sum(int(item["manual_default_gap_count"]) for item in matrix),
        "direct_gap_migration_candidate_count": int(migration_summary["migration_candidate_count"]),
        "owner_acceptance_candidate_count": int(owner_summary["acceptance_candidate_count"]),
        "owner_response_required_as_evidence_count": int(migration_summary["owner_response_required_count"]),
        "readability_final_exit_contract_count": int(readability_summary["final_exit_contract_count"]),
        "digest_rule_count": int(digest_rollups["rule_count"]),
        "digest_action_required_rule_count": len(digest_rollups["action_required_rule_ids"]),
        "digest_failure_escalation_rule_count": len(digest_rollups["failure_escalation_rule_ids"]),
        "digest_suppressed_success_rule_count": len(digest_rollups["suppressed_success_rule_ids"]),
        "telegram_send_performed_count": 0,
        "bot_api_call_performed_count": 0,
        "secret_value_read_count": 0,
        "runtime_write_performed_count": 0,
        "ai_alert_card_delivery_readback_endpoint_count": source_proof[
            "ai_alert_card_delivery_readback_endpoint_count"
        ],
        "ai_alert_card_learning_writeback_refs_count": source_proof[
            "ai_alert_card_learning_writeback_refs_count"
        ],
        "ai_alert_card_learning_registry_readback_count": source_proof[
            "ai_alert_card_learning_registry_readback_count"
        ],
        "telegram_alert_ai_loop_consumer_context_readback_count": source_proof[
            "telegram_alert_ai_loop_consumer_context_readback_count"
        ],
        "telegram_alert_post_apply_verifier_readback_count": source_proof[
            "telegram_alert_post_apply_verifier_readback_count"
        ],
        "telegram_alert_post_apply_verifier_awooop_surface_count": source_proof[
            "telegram_alert_post_apply_verifier_awooop_surface_count"
        ],
        "source_contract_ready": bool(source_proof.get("source_contract_ready")),
        "source_contract_missing_marker_count": len(
            source_proof.get("source_contract_missing_markers") or []
        ),
        "source_contract_missing_markers": list(
            source_proof.get("source_contract_missing_markers") or []
        ),
        "next_priority_action_id": next_priority_action["action_id"],
        "next_priority_work_item_id": "CIR-P0-TG-001",
    }