#!/usr/bin/env python3
"""Build a repo-only Telegram notification egress inventory.

This scanner identifies guarded Telegram Bot API paths that can bypass
TelegramGateway's final-exit formatter. Local ``.github/workflows`` files are
audited only as frozen legacy source truth. It does not read secrets, call
Telegram, modify or execute workflows, or send notifications.
"""

from __future__ import annotations

import argparse
import ast
import hashlib
import json
import re
import subprocess
import sys
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any


TAIPEI = timezone(timedelta(hours=8))

SCAN_ROOTS = (
    Path("agent99-control-plane.ps1"),
    Path(".gitea/workflows"),
    Path(".github/workflows"),
    Path("scripts/ops"),
    Path("scripts/ci"),
    Path("scripts/reboot-recovery"),
    Path("apps/api/src"),
)
SCAN_SUFFIXES = {".py", ".ps1", ".sh", ".js", ".yml", ".yaml"}

GUARDED_BOT_METHODS = (
    "sendMessage",
    "sendDocument",
    "sendPhoto",
    "sendMediaGroup",
    "editMessageText",
    "sendAnimation",
    "sendVideo",
    "sendAudio",
    "sendVoice",
)
BOT_TOKEN_URL_RE = re.compile(
    r"api\.telegram\.org/bot.*?/(?P<method>"
    + "|".join(re.escape(method) for method in GUARDED_BOT_METHODS)
    + r")\b",
    re.IGNORECASE,
)
GUARDED_BOT_METHOD_RE = re.compile(
    r"\b(?P<method>"
    + "|".join(re.escape(method) for method in GUARDED_BOT_METHODS)
    + r")\b",
    re.IGNORECASE,
)
COMPACT_BOT_ENDPOINT_RE = re.compile(
    r"api\.telegram\.org.{0,1024}?bot.{0,512}?/(?P<method>"
    + "|".join(re.escape(method) for method in GUARDED_BOT_METHODS)
    + r")",
    re.IGNORECASE,
)
DIRECT_HTTP_TRANSPORT_RE = re.compile(
    r"(?:"
    r"requests\s*\.\s*Session\s*\([^)]*\)\s*\.\s*(?:post|request|send)\s*\(|"
    r"requests\s*\.\s*(?:post|request)\s*\(|"
    r"httpx\s*\.\s*(?:AsyncClient|Client)\s*\([^)]*\)\s*\.\s*(?:post|request|send)\s*\(|"
    r"httpx\s*\.\s*(?:post|request)\s*\(|"
    r"(?:urllib\s*\.\s*request\s*\.\s*)?build_opener\s*\([^)]*\)\s*\.\s*open\s*\(|"
    r"urllib\s*\.\s*request\s*\.\s*(?:urlopen|Request)\s*\(|"
    r"\burlopen\s*\(|"
    r"\b(?:_?opener)\s*\.\s*open\s*\(|"
    r"\bInvoke-(?:WebRequest|RestMethod)\b|"
    r"\.\s*PostAsync\s*\(|"
    r"\.\s*Upload(?:String|Data|File)\s*\(|"
    r"\b(?:fetch|curl|wget)\b|"
    r"\b(?:_http_client|client|session)\s*\.\s*(?:post|request|send)\s*\("
    r")",
    re.IGNORECASE,
)
TELEGRAM_CONTEXT_RE = re.compile(
    r"(?:telegram|bot[_-]?token|chat[_-]?id)", re.IGNORECASE
)
BOT_TOKEN_CONTEXT_RE = re.compile(
    r"(?:\bbot[_-]?token\b|\btelegram[_-]?(?:bot[_-]?)?token\b|\$Token\b)",
    re.IGNORECASE,
)
CHAT_CONTEXT_RE = re.compile(
    r"(?:\bchat[_-]?id\b|\$ChatId\b)",
    re.IGNORECASE,
)
POWERSHELL_FUNCTION_RE = re.compile(
    r"^function\s+(?P<name>[A-Za-z0-9_-]+)\s*\{", re.IGNORECASE | re.MULTILINE
)
CANONICAL_FINAL_EXITS = {
    (
        "apps/api/src/services/telegram_gateway.py",
        "TelegramGateway._send_request",
    ),
}
_AST_IMPORT_NODE_TYPES = (ast.Import, ast.ImportFrom)
_AST_FUNCTION_NODE_TYPES = (ast.FunctionDef, ast.AsyncFunctionDef)
_AST_SEQUENCE_TARGET_NODE_TYPES = (ast.Tuple, ast.List)
_AST_NON_MODULE_SCOPE_NODE_TYPES = (
    ast.ClassDef,
    ast.FunctionDef,
    ast.AsyncFunctionDef,
    ast.Lambda,
)
GATEWAY_CALLSITE_RE = re.compile(
    r"(?:send_alert_notification\(|\b(?:tg|gw|gateway|telegram)\.send_text\(|_send_request\(\s*[\"']sendMessage[\"'])"
)
SECRET_INTERPOLATION_RE = re.compile(r"\$\{\{\s*secrets\.[^}]+\}\}")
BOT_TOKEN_FRAGMENT_RE = re.compile(
    r"bot[^/\s\"']+/(?P<method>"
    + "|".join(re.escape(method) for method in GUARDED_BOT_METHODS)
    + r")\b",
    re.IGNORECASE,
)

REQUIRED_OWNER_FIELDS = [
    "egress_surface_id",
    "owner_role_or_team",
    "routing_purpose",
    "current_sender",
    "target_chat_route",
    "message_shape_contract",
    "redaction_contract",
    "formatter_convergence_plan",
    "delivery_receipt_ref",
    "dedup_or_fingerprint_plan",
    "fallback_or_degraded_mode",
    "migration_or_exception_reason",
    "maintenance_window",
    "rollback_owner",
    "postcheck_evidence_ref",
    "no_secret_value_attestation",
    "no_raw_payload_attestation",
    "no_false_green_attestation",
]

REVIEWER_CHECKS = [
    "direct_bot_api_surface_identified",
    "owner_role_present",
    "target_route_is_sre_owned",
    "message_shape_is_ai_automation_card_or_documented_exception",
    "redaction_contract_present",
    "formatter_convergence_path_present",
    "delivery_receipt_metadata_only",
    "dedup_or_fingerprint_present",
    "fallback_mode_does_not_leak_raw_payload",
    "secret_name_only_no_value",
    "workflow_or_script_change_requires_separate_approval",
    "telegram_send_not_executed_by_inventory",
    "no_false_green_claim",
    "runtime_gate_stays_zero",
]

OUTCOME_LANES = [
    "waiting_owner_response",
    "request_owner_route_supplement",
    "request_formatter_convergence_plan",
    "request_redaction_contract",
    "request_delivery_receipt_metadata",
    "quarantine_secret_or_raw_payload",
    "reject_false_green_claim",
    "ready_for_notification_egress_review",
    "waiting_runtime_gate",
]

BLOCKED_ACTIONS = [
    "telegram_send",
    "bot_api_call",
    "workflow_modification",
    "script_modification_without_owner",
    "secret_value_collection",
    "secret_hash_collection",
    "partial_token_collection",
    "chat_id_collection_without_owner",
    "store_raw_message_payload",
    "store_unredacted_workflow_log",
    "change_chat_route",
    "change_bot_token",
    "rotate_secret",
    "workflow_dispatch",
    "production_deploy",
    "accept_route_200_as_delivery_receipt",
    "accept_cd_success_as_notification_acceptance",
    "accept_ui_visible_as_notification_acceptance",
    "skip_formatter_convergence",
    "skip_redaction_review",
    "open_runtime_gate",
    "add_action_button",
]


def git_short_sha(root: Path) -> str:
    try:
        result = subprocess.run(
            ["git", "rev-parse", "--short", "HEAD"],
            cwd=root,
            check=True,
            capture_output=True,
            text=True,
        )
        return result.stdout.strip()
    except Exception:
        return "unknown"


def iter_scannable_files(root: Path) -> list[Path]:
    files: list[Path] = []
    for scan_root in SCAN_ROOTS:
        absolute_root = root / scan_root
        if not absolute_root.exists():
            continue
        if absolute_root.is_file():
            if absolute_root.suffix in SCAN_SUFFIXES:
                files.append(absolute_root)
            continue
        for path in absolute_root.rglob("*"):
            if path.is_file() and path.suffix in SCAN_SUFFIXES:
                files.append(path)
    return sorted(files)


def sanitize_excerpt(line: str) -> str:
    excerpt = line.strip()
    excerpt = SECRET_INTERPOLATION_RE.sub("${{ secrets.<redacted> }}", excerpt)
    excerpt = BOT_TOKEN_URL_RE.sub(
        lambda match: f"api.telegram.org/bot<redacted>/{match.group('method')}",
        excerpt,
    )
    excerpt = BOT_TOKEN_FRAGMENT_RE.sub(
        lambda match: f"bot<redacted>/{match.group('method')}",
        excerpt,
    )
    return excerpt[:180]


def _compact_source(source: str) -> str:
    without_string_prefixes = re.sub(
        r"(?i)\b(?:fr|rf|f|r|u|b)(?=[\"'])",
        "",
        source,
    )
    return re.sub(r"[\s\"'`+()]+", "", without_string_prefixes)


_PYTHON_HTTP_TRANSPORT_CALL_RE = re.compile(
    r"^(?:"
    r"requests\.(?:post|request)|"
    r"requests\.Session\(\)\.(?:post|request|send)|"
    r"httpx\.(?:post|request)|"
    r"httpx\.(?:AsyncClient|Client)\(\)\.(?:post|request|send)|"
    r"urllib\.request\.(?:urlopen|Request)|"
    r"urllib\.request\.build_opener\(\)\.open"
    r")$"
)


def _record_python_import_alias(
    node: ast.AST,
    aliases: dict[str, str],
) -> None:
    if isinstance(node, ast.Import):
        for imported in node.names:
            bound_name = imported.asname or imported.name.split(".", 1)[0]
            aliases[bound_name] = (
                imported.name if imported.asname else bound_name
            )
        return

    module = str(node.module or "").strip(".")
    if not module:
        return
    for imported in node.names:
        if imported.name == "*":
            continue
        aliases[imported.asname or imported.name] = (
            f"{module}.{imported.name}"
        )


def _resolve_python_expression(
    node: ast.AST,
    aliases: dict[str, str],
) -> str | None:
    if isinstance(node, ast.Name):
        return aliases.get(node.id, node.id)
    if isinstance(node, ast.Attribute):
        owner = _resolve_python_expression(node.value, aliases)
        return f"{owner}.{node.attr}" if owner else None
    if isinstance(node, ast.Call):
        callable_name = _resolve_python_expression(node.func, aliases)
        return f"{callable_name}()" if callable_name else None
    return None


def _assignment_target_names(node: ast.AST) -> list[str]:
    if isinstance(node, ast.Name):
        return [node.id]
    if isinstance(node, _AST_SEQUENCE_TARGET_NODE_TYPES):
        return [
            name
            for child in node.elts
            for name in _assignment_target_names(child)
        ]
    return []


def _python_parent_map(tree: ast.AST) -> dict[ast.AST, ast.AST]:
    return {
        child: parent
        for parent in ast.walk(tree)
        for child in ast.iter_child_nodes(parent)
    }


def _python_function_qualified_name(
    function: ast.FunctionDef | ast.AsyncFunctionDef,
    parents: dict[ast.AST, ast.AST],
) -> str:
    parts = [function.name]
    parent = parents.get(function)
    while parent is not None:
        if isinstance(parent, (ast.ClassDef, *_AST_FUNCTION_NODE_TYPES)):
            parts.append(parent.name)
        parent = parents.get(parent)
    return ".".join(reversed(parts))


def _nearest_python_function(
    node: ast.AST,
    parents: dict[ast.AST, ast.AST],
) -> ast.FunctionDef | ast.AsyncFunctionDef | None:
    parent = parents.get(node)
    while parent is not None:
        if isinstance(parent, _AST_FUNCTION_NODE_TYPES):
            return parent
        parent = parents.get(parent)
    return None


def _is_python_module_scope(
    node: ast.AST,
    parents: dict[ast.AST, ast.AST],
) -> bool:
    parent = parents.get(node)
    while parent is not None:
        if isinstance(parent, _AST_NON_MODULE_SCOPE_NODE_TYPES):
            return False
        parent = parents.get(parent)
    return True


def _resolve_python_transport_calls(
    nodes: list[ast.AST],
    aliases: dict[str, str],
) -> list[tuple[int, str]]:
    resolved_calls: list[tuple[int, str]] = []
    for node in sorted(
        nodes,
        key=lambda item: (
            int(getattr(item, "lineno", 0)),
            int(getattr(item, "col_offset", 0)),
            0 if isinstance(item, _AST_IMPORT_NODE_TYPES) else 1,
        ),
    ):
        if isinstance(node, _AST_IMPORT_NODE_TYPES):
            _record_python_import_alias(node, aliases)
            continue
        if isinstance(node, ast.Assign):
            resolved_value = _resolve_python_expression(node.value, aliases)
            for target in node.targets:
                for name in _assignment_target_names(target):
                    if resolved_value:
                        aliases[name] = resolved_value
                    else:
                        aliases.pop(name, None)
            continue
        if isinstance(node, ast.AnnAssign):
            resolved_value = (
                _resolve_python_expression(node.value, aliases)
                if node.value is not None
                else None
            )
            for name in _assignment_target_names(node.target):
                if resolved_value:
                    aliases[name] = resolved_value
                else:
                    aliases.pop(name, None)
            continue
        if not isinstance(node, ast.Call):
            continue
        qualified_call = _resolve_python_expression(node.func, aliases)
        if qualified_call and _PYTHON_HTTP_TRANSPORT_CALL_RE.fullmatch(
            qualified_call
        ):
            resolved_calls.append((node.lineno, qualified_call))
    return sorted(set(resolved_calls))


def _python_transport_calls_by_scope(
    text: str,
) -> tuple[
    dict[tuple[str, int, int], list[tuple[int, str]]],
    list[tuple[int, str]],
]:
    """Resolve HTTP aliases in function and true module-level AST scopes."""
    try:
        tree = ast.parse(text)
    except SyntaxError:
        return {}, []

    parents = _python_parent_map(tree)
    module_aliases: dict[str, str] = {}
    module_nodes = [
        node
        for node in ast.walk(tree)
        if _is_python_module_scope(node, parents)
    ]
    module_calls = _resolve_python_transport_calls(
        module_nodes,
        module_aliases,
    )

    transports: dict[tuple[str, int, int], list[tuple[int, str]]] = {}
    function_nodes = [
        node
        for node in ast.walk(tree)
        if isinstance(node, _AST_FUNCTION_NODE_TYPES)
        and node.end_lineno is not None
    ]
    for function in function_nodes:
        aliases = dict(module_aliases)
        arguments = [
            *function.args.posonlyargs,
            *function.args.args,
            *function.args.kwonlyargs,
        ]
        if function.args.vararg is not None:
            arguments.append(function.args.vararg)
        if function.args.kwarg is not None:
            arguments.append(function.args.kwarg)
        for argument in arguments:
            aliases.pop(argument.arg, None)

        scoped_nodes = [
            node
            for node in ast.walk(function)
            if node is function
            or _nearest_python_function(node, parents) is function
        ]
        resolved_calls = _resolve_python_transport_calls(scoped_nodes, aliases)

        if resolved_calls:
            qualified_name = _python_function_qualified_name(function, parents)
            transports[
                (
                    qualified_name,
                    function.lineno,
                    int(function.end_lineno),
                )
            ] = resolved_calls
    return transports, module_calls


def _transport_window_units(
    text: str,
    *,
    excluded_line_ranges: list[tuple[int, int]],
) -> list[tuple[str, str, int, int, str]]:
    lines = text.splitlines()
    units: list[tuple[str, str, int, int, str]] = []
    for match in DIRECT_HTTP_TRANSPORT_RE.finditer(text):
        line_number = text.count("\n", 0, match.start()) + 1
        if any(start <= line_number <= end for start, end in excluded_line_ranges):
            continue
        start_line = max(1, line_number - 20)
        end_line = min(len(lines), line_number + 20)
        units.append(
            (
                "<module>",
                "<module>",
                start_line,
                end_line,
                "\n".join(lines[start_line - 1 : end_line]),
            )
        )
    return units


def _source_units(
    relative_path: str,
    text: str,
) -> list[tuple[str, str, int, int, str]]:
    lines = text.splitlines()
    units: list[tuple[str, str, int, int, str]] = []
    ranges: list[tuple[int, int]] = []
    if relative_path.endswith(".py"):
        try:
            tree = ast.parse(text)
        except SyntaxError:
            tree = None
        if tree is not None:
            parents = _python_parent_map(tree)
            nodes = [
                node
                for node in ast.walk(tree)
                if isinstance(node, _AST_FUNCTION_NODE_TYPES)
                and getattr(node, "end_lineno", None)
            ]
            for node in sorted(
                nodes, key=lambda item: (item.lineno, item.end_lineno or item.lineno)
            ):
                end_line = int(node.end_lineno or node.lineno)
                ranges.append((node.lineno, end_line))
                units.append(
                    (
                        node.name,
                        _python_function_qualified_name(node, parents),
                        node.lineno,
                        end_line,
                        "\n".join(lines[node.lineno - 1 : end_line]),
                    )
                )
    elif relative_path.endswith(".ps1"):
        matches = list(POWERSHELL_FUNCTION_RE.finditer(text))
        for index, match in enumerate(matches):
            start_line = text.count("\n", 0, match.start()) + 1
            end_offset = (
                matches[index + 1].start() if index + 1 < len(matches) else len(text)
            )
            end_line = text.count("\n", 0, end_offset) + 1
            ranges.append((start_line, end_line))
            units.append(
                (
                    match.group("name"),
                    match.group("name"),
                    start_line,
                    end_line,
                    text[match.start() : end_offset],
                )
            )
    return units + _transport_window_units(text, excluded_line_ranges=ranges)


def scan_direct_bot_api_surfaces(
    relative_path: str,
    text: str,
) -> list[dict[str, Any]]:
    """Find direct Telegram transports, including split URLs and custom senders."""

    source_lines = text.splitlines()
    findings: list[dict[str, Any]] = []
    seen: set[tuple[int, str, str]] = set()
    python_transports, module_transports = (
        _python_transport_calls_by_scope(text)
        if relative_path.endswith(".py")
        else ({}, [])
    )
    source_units = _source_units(relative_path, text)
    for line_number, _qualified_call in module_transports:
        start_line = max(1, line_number - 20)
        end_line = min(len(source_lines), line_number + 20)
        source_units.append(
            (
                "<module>",
                "<module>",
                start_line,
                end_line,
                "\n".join(source_lines[start_line - 1 : end_line]),
            )
        )

    for (
        function_name,
        function_qualified,
        start_line,
        _end_line,
        source,
    ) in source_units:
        if (relative_path, function_qualified) in CANONICAL_FINAL_EXITS:
            continue
        resolved_python_transports = (
            [
                item
                for item in module_transports
                if start_line <= item[0] <= _end_line
            ]
            if function_qualified == "<module>"
            else python_transports.get(
                (function_qualified, start_line, _end_line),
                [],
            )
        )
        transport_match = DIRECT_HTTP_TRANSPORT_RE.search(source)
        if transport_match is None and not resolved_python_transports:
            continue
        method_match = GUARDED_BOT_METHOD_RE.search(source)
        endpoint_match = COMPACT_BOT_ENDPOINT_RE.search(_compact_source(source))
        telegram_named = bool(
            re.search(r"(?:telegram|bot)", function_name, re.IGNORECASE)
        )
        telegram_context = bool(TELEGRAM_CONTEXT_RE.search(source))
        token_and_chat_context = bool(
            BOT_TOKEN_CONTEXT_RE.search(source) and CHAT_CONTEXT_RE.search(source)
        )
        if (
            endpoint_match is None
            and method_match is None
            and not ((telegram_named and telegram_context) or token_and_chat_context)
        ):
            continue

        method = (
            endpoint_match.group("method")
            if endpoint_match is not None
            else method_match.group("method")
            if method_match is not None
            else "dynamic"
        )
        line_number = (
            resolved_python_transports[0][0]
            if resolved_python_transports
            else start_line + source[: transport_match.start()].count("\n")
        )
        detection_kind = (
            "direct_bot_api_endpoint"
            if endpoint_match is not None
            else "custom_direct_sender"
        )
        key = (line_number, method.lower(), detection_kind)
        if key in seen:
            continue
        seen.add(key)
        source_line = (
            source_lines[line_number - 1] if line_number <= len(source_lines) else ""
        )
        findings.append(
            {
                "line": line_number,
                "method": method,
                "detection_kind": detection_kind,
                "function": function_name,
                "function_qualified": function_qualified,
                "sanitized_excerpt": sanitize_excerpt(source_line),
            }
        )
    return findings


def surface_kind(relative_path: str) -> str:
    if relative_path.startswith(".github/workflows/"):
        return "github_frozen_legacy_workflow_direct_bot_api"
    if relative_path.startswith(".gitea/workflows/"):
        return "gitea_workflow_direct_bot_api"
    if relative_path.startswith("scripts/ops/"):
        return "ops_script_direct_bot_api"
    if relative_path.startswith("scripts/ci/"):
        return "ci_script_direct_bot_api"
    if relative_path == "agent99-control-plane.ps1":
        return "agent99_control_plane_direct_bot_api"
    if relative_path.startswith("scripts/reboot-recovery/"):
        return "reboot_recovery_direct_bot_api"
    if relative_path.startswith("apps/api/src/"):
        return "api_direct_bot_api"
    return "other_direct_bot_api"


def source_truth_classification(relative_path: str) -> str:
    if relative_path.startswith(".github/workflows/"):
        return "frozen_legacy_source_truth"
    return "active_repo_source_truth"


def line_hash(relative_path: str, line_number: int, line: str) -> str:
    payload = f"{relative_path}:{line_number}:{line.strip()}".encode("utf-8")
    return hashlib.sha256(payload).hexdigest()[:16]


def build_report(root: Path, generated_at: str | None = None) -> dict[str, Any]:
    generated = generated_at or datetime.now(TAIPEI).isoformat(timespec="seconds")
    files = iter_scannable_files(root)
    direct_calls: list[dict[str, Any]] = []
    gateway_calls: list[dict[str, Any]] = []

    for path in files:
        relative_path = path.relative_to(root).as_posix()
        text = path.read_text(encoding="utf-8", errors="replace")
        text_lines = text.splitlines()
        for finding in scan_direct_bot_api_surfaces(relative_path, text):
            line_number = int(finding["line"])
            line = text_lines[line_number - 1] if line_number <= len(text_lines) else ""
            kind = surface_kind(relative_path)
            truth_classification = source_truth_classification(relative_path)
            direct_calls.append(
                {
                    "egress_surface_id": f"telegram_egress:{kind}:{relative_path}:{line_number}",
                    "surface_kind": kind,
                    "source_truth_classification": truth_classification,
                    "path": relative_path,
                    "line": line_number,
                    "line_hash": line_hash(relative_path, line_number, line),
                    "method": finding["method"],
                    "detection_kind": finding["detection_kind"],
                    "function": finding["function"],
                    "sanitized_excerpt": finding["sanitized_excerpt"],
                    "required_owner_fields": REQUIRED_OWNER_FIELDS,
                    "reviewer_checks": REVIEWER_CHECKS,
                    "outcome_lanes": OUTCOME_LANES,
                    "blocked_actions": BLOCKED_ACTIONS,
                    "owner_response_received": False,
                    "owner_response_accepted": False,
                    "formatter_convergence_accepted": False,
                    "redaction_contract_accepted": False,
                    "delivery_receipt_accepted": False,
                    "direct_bot_api_migration_authorized": False,
                    "telegram_send_authorized": False,
                    "bot_api_call_authorized": False,
                    "workflow_modification_authorized": False,
                    "script_modification_authorized": False,
                    "secret_value_collection_allowed": False,
                    "raw_payload_storage_allowed": False,
                    "production_write_authorized": False,
                    "runtime_execution_authorized": False,
                    "workflow_execution_authorized": False,
                    "runtime_gate": False,
                    "action_buttons_allowed": False,
                    "not_authorization": True,
                }
            )

        for line_number, line in enumerate(text_lines, start=1):
            if GATEWAY_CALLSITE_RE.search(line):
                gateway_calls.append(
                    {
                        "path": relative_path,
                        "line": line_number,
                        "line_hash": line_hash(relative_path, line_number, line),
                    }
                )

    active_direct_calls = [
        item
        for item in direct_calls
        if item["source_truth_classification"] == "active_repo_source_truth"
    ]
    frozen_legacy_direct_calls = [
        item
        for item in direct_calls
        if item["source_truth_classification"] == "frozen_legacy_source_truth"
    ]
    direct_files = sorted({item["path"] for item in direct_calls})
    active_direct_files = sorted({item["path"] for item in active_direct_calls})
    frozen_legacy_direct_files = sorted(
        {item["path"] for item in frozen_legacy_direct_calls}
    )
    workflow_direct_calls = [
        item
        for item in direct_calls
        if item["surface_kind"] == "gitea_workflow_direct_bot_api"
    ]
    ops_direct_calls = [
        item
        for item in direct_calls
        if item["surface_kind"] == "ops_script_direct_bot_api"
    ]
    ci_direct_calls = [
        item
        for item in direct_calls
        if item["surface_kind"] == "ci_script_direct_bot_api"
    ]
    api_direct_calls = [
        item for item in direct_calls if item["surface_kind"] == "api_direct_bot_api"
    ]
    custom_direct_senders = [
        item
        for item in direct_calls
        if item["detection_kind"] == "custom_direct_sender"
    ]
    direct_endpoints = [
        item
        for item in direct_calls
        if item["detection_kind"] == "direct_bot_api_endpoint"
    ]
    telegram_gateway_path = root / "apps/api/src/services/telegram_gateway.py"
    telegram_gateway_text = telegram_gateway_path.read_text(
        encoding="utf-8", errors="replace"
    )
    gateway_formatter_present = (
        "normalize_telegram_send_message_payload" in telegram_gateway_text
    )

    return {
        "schema_version": "telegram_notification_egress_inventory_v1",
        "generated_at": generated,
        "git_commit": git_short_sha(root),
        "status": "inventory_ready_no_runtime_action",
        "mode": "repo_only_scan_no_secret_value_no_telegram_send",
        "scan_roots": [path.as_posix() for path in SCAN_ROOTS],
        "summary": {
            "scanned_file_count": len(files),
            "direct_bot_api_file_count": len(direct_files),
            "direct_bot_api_call_count": len(direct_calls),
            "active_direct_bot_api_file_count": len(active_direct_files),
            "active_direct_bot_api_call_count": len(active_direct_calls),
            "github_frozen_legacy_direct_bot_api_file_count": len(
                frozen_legacy_direct_files
            ),
            "github_frozen_legacy_direct_bot_api_call_count": len(
                frozen_legacy_direct_calls
            ),
            "direct_bot_api_endpoint_count": len(direct_endpoints),
            "custom_direct_sender_count": len(custom_direct_senders),
            "workflow_direct_bot_api_call_count": len(workflow_direct_calls),
            "ops_script_direct_bot_api_call_count": len(ops_direct_calls),
            "ci_script_direct_bot_api_call_count": len(ci_direct_calls),
            "api_direct_bot_api_call_count": len(api_direct_calls),
            "gateway_normalized_callsite_count": len(gateway_calls),
            "gateway_final_exit_formatter_present_count": 1
            if gateway_formatter_present
            else 0,
            "required_owner_field_count": len(REQUIRED_OWNER_FIELDS),
            "reviewer_check_count": len(REVIEWER_CHECKS),
            "outcome_lane_count": len(OUTCOME_LANES),
            "blocked_action_count": len(BLOCKED_ACTIONS),
            "owner_response_received_count": 0,
            "owner_response_accepted_count": 0,
            "formatter_convergence_accepted_count": 0,
            "redaction_contract_accepted_count": 0,
            "delivery_receipt_accepted_count": 0,
            "direct_bot_api_migration_authorized_count": 0,
            "telegram_send_authorized_count": 0,
            "bot_api_call_authorized_count": 0,
            "workflow_modification_authorized_count": 0,
            "workflow_execution_authorized_count": 0,
            "script_modification_authorized_count": 0,
            "secret_value_collection_allowed_count": 0,
            "raw_payload_storage_allowed_count": 0,
            "production_write_authorized_count": 0,
            "runtime_gate_count": 0,
            "action_button_count": 0,
        },
        "execution_boundaries": {
            "runtime_execution_authorized": False,
            "telegram_send_authorized": False,
            "bot_api_call_authorized": False,
            "workflow_modification_authorized": False,
            "workflow_execution_authorized": False,
            "github_workflow_execution_authorized": False,
            "script_modification_authorized": False,
            "secret_value_collection_allowed": False,
            "secret_hash_collection_allowed": False,
            "partial_token_collection_allowed": False,
            "raw_payload_storage_allowed": False,
            "chat_route_change_authorized": False,
            "bot_token_change_authorized": False,
            "workflow_dispatch_authorized": False,
            "production_deploy_authorized": False,
            "action_buttons_allowed": False,
            "not_authorization": True,
        },
        "direct_bot_api_calls": direct_calls,
        "gateway_normalized_callsite_refs": gateway_calls,
        "operator_interpretation": [
            "active_direct_bot_api_call_count 大於 0 代表仍有 active workflow / ops / API 旁路可能繞過 TelegramGateway formatter。",
            ".github/workflows 僅列為 frozen_legacy_source_truth 供盤點，runtime_execution_authorized=false、workflow_execution_authorized=false，禁止執行。",
            "本清冊只建立 metadata-only egress surface，不送 Telegram、不修改 workflow / script、不讀 secret value。",
            "後續要收斂 direct Bot API 必須另走 owner response、formatter convergence、redaction contract、delivery receipt 與維護窗口。",
        ],
    }


def validate(root: Path) -> None:
    report = build_report(root)
    if report["summary"]["gateway_final_exit_formatter_present_count"] != 1:
        raise SystemExit(
            "BLOCKED telegram egress inventory: gateway formatter not found"
        )


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Build Telegram notification egress inventory"
    )
    parser.add_argument("--root", default=".", help="repository root")
    parser.add_argument("--output", help="write JSON snapshot")
    parser.add_argument("--generated-at", help="fixed generated_at timestamp")
    args = parser.parse_args()

    root = Path(args.root).resolve()
    report = build_report(root, args.generated_at)
    payload = json.dumps(report, ensure_ascii=False, indent=2) + "\n"
    if args.output:
        Path(args.output).write_text(payload, encoding="utf-8")
    else:
        sys.stdout.write(payload)

    print(
        "TELEGRAM_NOTIFICATION_EGRESS_INVENTORY_OK "
        f"direct_calls={report['summary']['direct_bot_api_call_count']} "
        f"active={report['summary']['active_direct_bot_api_call_count']} "
        f"frozen_legacy={report['summary']['github_frozen_legacy_direct_bot_api_call_count']} "
        f"custom={report['summary']['custom_direct_sender_count']} "
        f"files={report['summary']['direct_bot_api_file_count']} "
        f"workflow={report['summary']['workflow_direct_bot_api_call_count']} "
        f"ops={report['summary']['ops_script_direct_bot_api_call_count']} "
        f"api={report['summary']['api_direct_bot_api_call_count']} "
        f"runtime_gate={report['summary']['runtime_gate_count']}",
        file=sys.stderr,
    )


if __name__ == "__main__":
    main()