feat(telegram): expose AI automation matrix

apps/api/src/services/awoooi_priority_work_order_readback.py

@@ -320,9 +320,12 @@ _COMMANDER_INSERTED_REQUIREMENT_WORK_ITEMS: list[dict[str, Any]] = [
             "每個 Telegram 告警都要映射到 DB/log receipt、AI route、controlled queue、work item、"
             "post verifier 與 learning writeback；direct send 或 manual default 都列缺口。"
         ),
-        "current_state": "本輪已把 workflow direct send 收斂到 0 並暴露 API direct gap；仍需全面 alert matrix。",
+        "current_state": (
+            "Telegram alert AI automation matrix endpoint 已實作；目前明確讀回 workflow direct send 0、"
+            "ops script direct gap 4、API direct gap 1，下一步收斂 direct Bot API 旁路。"
+        ),
         "acceptance": "Telegram alert matrix 顯示 total、DB receipt、AI route、controlled queue、manual/default gap 與 direct send gap。",
-        "next_action": "擴充 Telegram alert inventory/readback，讓人工預設只允許 critical/break-glass 或歷史證據。",
+        "next_action": "先把 apps/api/src/services/channel_hub.py direct sender 改走 TelegramGateway / AwoooP outbound receipt path。",
         "mapped_workplan_id": "P0-006",
     },
     {

apps/api/src/services/telegram_alert_ai_automation_matrix.py

@@ -0,0 +1,414 @@
+"""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"
+
+
+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)
+
+    direct_gap_count = int(egress_summary["direct_bot_api_call_count"])
+    direct_gap_file_count = int(egress_summary["direct_bot_api_file_count"])
+    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_delivery_receipt_readback_required",
+            "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_open_requires_controlled_migration",
+            "known_item_count": direct_gap_count,
+            "db_or_log_receipt": "missing_or_not_proven_for_direct_path",
+            "ai_route": "missing_or_not_proven_for_direct_path",
+            "controlled_queue": "missing_or_not_proven_for_direct_path",
+            "post_verifier": "metadata_inventory_ready",
+            "learning_writeback": "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": "draft_only_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_apply_candidates_ready",
+            "known_item_count": int(migration_summary["migration_candidate_count"]),
+            "db_or_log_receipt": "required_before_candidate_close",
+            "ai_route": "migration_plan_ready",
+            "controlled_queue": "ready_for_controlled_apply_review",
+            "post_verifier": "owner_acceptance_ledger_ready",
+            "learning_writeback": "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": "read_only_query_no_send",
+            "post_verifier": "ready_source_contract_present",
+            "learning_writeback": "ready_delivery_receipt_summary",
+            "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",
+            ],
+        },
+    ]
+
+    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,
+    )
+
+    return {
+        "schema_version": _SCHEMA_VERSION,
+        "status": (
+            "telegram_alert_ai_automation_matrix_ready_with_direct_send_gaps"
+            if direct_gap_count
+            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": [
+            {
+                "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.",
+            },
+            {
+                "action_id": "bind_ai_alert_cards_to_learning_writeback",
+                "scope": "ai_automation_alert_card_v1 outbound mirror metadata",
+                "priority": "P0",
+                "why": "Telegram must become an AI automation learning input, not a manual terminal state.",
+                "requires_secret": False,
+                "requires_runtime_send": False,
+                "post_verifier": "/api/v1/platform/runs/ai-alert-cards readback shows delivery_receipt_readback_required.",
+            },
+        ],
+        "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 remain the concrete AI automation gap; they must not be treated as completed delivery receipts.",
+            "Manual/default semantics are only acceptable for critical/break-glass or historical evidence; direct send gaps remain controlled migration candidates.",
+        ],
+    }
+
+
+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, int | bool]:
+    source_checks = {
+        "telegram_gateway_mirror_present": (
+            repo_root / "apps/api/src/services/telegram_gateway.py",
+            "_mirror_outbound_message",
+        ),
+        "telegram_gateway_record_outbound_present": (
+            repo_root / "apps/api/src/services/telegram_gateway.py",
+            "record_outbound_message",
+        ),
+        "ai_alert_card_metadata_present": (
+            repo_root / "apps/api/src/services/telegram_gateway.py",
+            "ai_automation_alert_card_mirror_v1",
+        ),
+        "outbound_message_model_present": (
+            repo_root / "apps/api/src/db/awooop_models.py",
+            "__tablename__ = \"awooop_outbound_message\"",
+        ),
+        "ai_alert_card_delivery_readback_endpoint_present": (
+            repo_root / "apps/api/src/api/v1/platform/operator_runs.py",
+            "/runs/ai-alert-cards",
+        ),
+    }
+    result: dict[str, int | bool] = {}
+    for key, (path, marker) in source_checks.items():
+        text = path.read_text(encoding="utf-8", errors="replace") if path.exists() else ""
+        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
+
+    required = [key for key, value in result.items() if key.endswith("_present") and value is not True]
+    if required:
+        raise ValueError(f"Telegram matrix source contract missing markers: {required}")
+    return result
+
+
+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, int | bool],
+) -> dict[str, Any]:
+    ready = lambda key, prefix: 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"
+        ],
+        "next_priority_action_id": "migrate_api_direct_sender_to_gateway",
+        "next_priority_work_item_id": "CIR-P0-TG-001",
+    }