fix(telegram): fail closed on unverified delivery

scripts/security/telegram-alert-readability-guard.py

@@ -8,6 +8,7 @@ Bot API、不讀 secret、不連線主機，也不啟動任何 runtime gate。
 from __future__ import annotations
 
 import argparse
+import ast
 import json
 import subprocess
 from datetime import datetime, timedelta, timezone
@@ -164,11 +165,44 @@ def require_contains(label: str, text: str, marker: str) -> None:
         raise SystemExit(f"BLOCKED {label}: missing {marker!r}")
 
 
-def function_segment(text: str, marker: str, *, limit: int = 2200) -> str:
-    start = text.find(marker)
-    if start == -1:
-        raise SystemExit(f"BLOCKED source function marker missing: {marker!r}")
-    return text[start : start + limit]
+def function_segment(text: str, marker: str) -> str:
+    """Return the complete named function using Python AST source boundaries."""
+    marker_prefix, separator, function_name = marker.partition("def ")
+    function_name = function_name.strip()
+    if not separator or not function_name.isidentifier():
+        raise SystemExit(f"BLOCKED invalid source function marker: {marker!r}")
+
+    expected_type: type[ast.FunctionDef] | type[ast.AsyncFunctionDef]
+    expected_type = (
+        ast.AsyncFunctionDef
+        if marker_prefix.strip() == "async"
+        else ast.FunctionDef
+    )
+    try:
+        tree = ast.parse(text)
+    except SyntaxError as exc:
+        raise SystemExit(
+            f"BLOCKED telegram gateway source is not valid Python: {exc.msg}"
+        ) from exc
+
+    matches = [
+        node
+        for node in ast.walk(tree)
+        if isinstance(node, expected_type) and node.name == function_name
+    ]
+    if len(matches) != 1:
+        raise SystemExit(
+            "BLOCKED source function marker must resolve exactly once: "
+            f"{marker!r}, matches={len(matches)}"
+        )
+
+    node = matches[0]
+    if node.end_lineno is None:
+        raise SystemExit(
+            f"BLOCKED source function boundary unavailable: {marker!r}"
+        )
+    lines = text.splitlines(keepends=True)
+    return "".join(lines[node.lineno - 1 : node.end_lineno])
 
 
 def git_commit(root: Path) -> str:

scripts/security/telegram-notification-egress-inventory.py

@@ -1,14 +1,16 @@
 #!/usr/bin/env python3
 """Build a repo-only Telegram notification egress inventory.
 
-This scanner identifies Telegram Bot API sendMessage paths that can bypass
-TelegramGateway's final-exit formatter. It does not read secrets, call
-Telegram, modify workflows, or send notifications.
+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
@@ -24,6 +26,7 @@ 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"),
@@ -42,17 +45,81 @@ GUARDED_BOT_METHODS = (
     "sendAudio",
     "sendVoice",
 )
-DIRECT_BOT_API_RE = re.compile(
+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_URL_RE = DIRECT_BOT_API_RE
+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",
@@ -167,10 +234,429 @@ def sanitize_excerpt(line: str) -> str:
         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/"):
@@ -186,6 +672,12 @@ def surface_kind(relative_path: str) -> str:
     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]
@@ -200,42 +692,50 @@ def build_report(root: Path, generated_at: str | None = None) -> dict[str, Any]:
     for path in files:
         relative_path = path.relative_to(root).as_posix()
         text = path.read_text(encoding="utf-8", errors="replace")
-        for line_number, line in enumerate(text.splitlines(), start=1):
-            direct_match = DIRECT_BOT_API_RE.search(line)
-            if direct_match:
-                kind = surface_kind(relative_path)
-                direct_calls.append(
-                    {
-                        "egress_surface_id": f"telegram_egress:{kind}:{relative_path}:{line_number}",
-                        "surface_kind": kind,
-                        "path": relative_path,
-                        "line": line_number,
-                        "line_hash": line_hash(relative_path, line_number, line),
-                        "method": direct_match.group("method"),
-                        "sanitized_excerpt": sanitize_excerpt(line),
-                        "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_gate": False,
-                        "action_buttons_allowed": False,
-                        "not_authorization": True,
-                    }
-                )
+        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(
                     {
@@ -245,14 +745,56 @@ def build_report(root: Path, generated_at: str | None = None) -> dict[str, Any]:
                     }
                 )
 
+    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})
-    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"]
+    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
+    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",
@@ -265,12 +807,24 @@ def build_report(root: Path, generated_at: str | None = None) -> dict[str, Any]:
             "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,
+            "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),
@@ -284,6 +838,7 @@ def build_report(root: Path, generated_at: str | None = None) -> dict[str, Any]:
             "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,
@@ -296,6 +851,8 @@ def build_report(root: Path, generated_at: str | None = None) -> dict[str, Any]:
             "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,
@@ -311,7 +868,8 @@ def build_report(root: Path, generated_at: str | None = None) -> dict[str, Any]:
         "direct_bot_api_calls": direct_calls,
         "gateway_normalized_callsite_refs": gateway_calls,
         "operator_interpretation": [
-            "direct_bot_api_call_count 大於 0 代表仍有 workflow / ops / API 旁路可能繞過 TelegramGateway formatter。",
+            "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 與維護窗口。",
         ],
@@ -321,11 +879,15 @@ def build_report(root: Path, generated_at: str | None = None) -> dict[str, Any]:
 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")
+        raise SystemExit(
+            "BLOCKED telegram egress inventory: gateway formatter not found"
+        )
 
 
 def main() -> None:
-    parser = argparse.ArgumentParser(description="Build Telegram notification egress inventory")
+    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")
@@ -342,6 +904,9 @@ def main() -> None:
     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']} "

scripts/security/telegram-notification-egress-no-new-bypass-guard.py

@@ -2,14 +2,16 @@
 """檢查 Telegram 通知出口不可新增未登記 direct Bot API 旁路。
 
 本 guard 只掃描 repo 原始碼與 committed snapshot，不讀 secret、不呼叫
-Telegram、不修改 workflow / script / API sender。既有 direct send 仍是待
-owner response 的基線；任何新增或變形的 direct Bot API endpoint 都必須先
-進 inventory / owner request / migration plan，而不是直接合併。
+Telegram、不修改或執行 workflow / script / API sender。Committed baseline
+僅供 drift 比對；目前 active source 的 direct endpoint 或 custom sender 必須
+真正為零。Local ``.github/workflows`` 只以 frozen legacy source truth 稽核，
+不得以 regex 漏掃、歷史基線或 frozen source 產生 false-green。
 """
 
 from __future__ import annotations
 
 import argparse
+import ast
 import json
 import re
 import subprocess
@@ -22,10 +24,13 @@ from typing import Any
 
 TAIPEI = timezone(timedelta(hours=8))
 
-SOURCE_SNAPSHOT = Path("docs/security/telegram-notification-egress-inventory.snapshot.json")
+SOURCE_SNAPSHOT = Path(
+    "docs/security/telegram-notification-egress-inventory.snapshot.json"
+)
 SCAN_ROOTS = (
     Path("agent99-control-plane.ps1"),
     Path(".gitea/workflows"),
+    Path(".github/workflows"),
     Path("scripts/ops"),
     Path("scripts/ci"),
     Path("scripts/reboot-recovery"),
@@ -50,6 +55,65 @@ BOT_ENDPOINT_RE = re.compile(
     + 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,
+)
 SECRET_INTERPOLATION_RE = re.compile(r"\$\{\{\s*secrets\.[^}]+\}\}")
 BOT_TOKEN_URL_RE = re.compile(
     r"api\.telegram\.org/bot.*?/(?P<method>"
@@ -57,6 +121,12 @@ BOT_TOKEN_URL_RE = re.compile(
     + r")\b",
     re.IGNORECASE,
 )
+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,
+)
 
 
 def git_short_sha(root: Path) -> str:
@@ -96,13 +166,436 @@ def sanitize_excerpt(line: str) -> str:
         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 signature(path: str, method: str, sanitized_excerpt: str) -> str:
     return f"{path}::{method.lower()}::{sanitized_excerpt}"
 
 
+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 load_source_snapshot(root: Path) -> dict[str, Any]:
     snapshot_path = root / SOURCE_SNAPSHOT
     return json.loads(snapshot_path.read_text(encoding="utf-8"))
@@ -111,9 +604,16 @@ def load_source_snapshot(root: Path) -> dict[str, Any]:
 def build_baseline(source_snapshot: dict[str, Any]) -> Counter[str]:
     baseline: Counter[str] = Counter()
     for item in source_snapshot.get("direct_bot_api_calls", []):
+        if (
+            source_truth_classification(str(item["path"]))
+            == "frozen_legacy_source_truth"
+        ):
+            continue
         excerpt = item.get("sanitized_excerpt", "")
         match = BOT_ENDPOINT_RE.search(excerpt)
-        method = match.group("method") if match else "sendMessage"
+        method = str(
+            item.get("method") or (match.group("method") if match else "dynamic")
+        )
         baseline[signature(item["path"], method, excerpt)] += 1
     return baseline
 
@@ -123,19 +623,25 @@ def scan_current_direct_endpoints(root: Path) -> list[dict[str, Any]]:
     for path in iter_scannable_files(root):
         relative_path = path.relative_to(root).as_posix()
         text = path.read_text(encoding="utf-8", errors="replace")
-        for line_number, line in enumerate(text.splitlines(), start=1):
-            for match in BOT_ENDPOINT_RE.finditer(line):
-                method = match.group("method")
-                sanitized = sanitize_excerpt(line)
-                findings.append(
-                    {
-                        "path": relative_path,
-                        "line": line_number,
-                        "method": method,
-                        "sanitized_excerpt": sanitized,
-                        "signature": signature(relative_path, method, sanitized),
-                    }
-                )
+        for finding in scan_direct_bot_api_surfaces(relative_path, text):
+            method = finding["method"]
+            sanitized = finding["sanitized_excerpt"]
+            findings.append(
+                {
+                    "path": relative_path,
+                    "line": finding["line"],
+                    "method": method,
+                    "detection_kind": finding["detection_kind"],
+                    "function": finding["function"],
+                    "sanitized_excerpt": sanitized,
+                    "signature": signature(relative_path, method, sanitized),
+                    "source_truth_classification": source_truth_classification(
+                        relative_path
+                    ),
+                    "runtime_execution_authorized": False,
+                    "workflow_execution_authorized": False,
+                }
+            )
     return findings
 
 
@@ -153,7 +659,17 @@ def build_report(root: Path, generated_at: str | None = None) -> dict[str, Any]:
     generated = generated_at or datetime.now(TAIPEI).isoformat(timespec="seconds")
     source_snapshot = load_source_snapshot(root)
     baseline = build_baseline(source_snapshot)
-    current_findings = scan_current_direct_endpoints(root)
+    detected_findings = scan_current_direct_endpoints(root)
+    current_findings = [
+        item
+        for item in detected_findings
+        if item["source_truth_classification"] == "active_repo_source_truth"
+    ]
+    frozen_legacy_findings = [
+        item
+        for item in detected_findings
+        if item["source_truth_classification"] == "frozen_legacy_source_truth"
+    ]
     remaining_baseline = baseline.copy()
     new_bypass_findings: list[dict[str, Any]] = []
 
@@ -172,23 +688,48 @@ def build_report(root: Path, generated_at: str | None = None) -> dict[str, Any]:
     current_files = sorted({item["path"] for item in current_findings})
     new_bypass_files = sorted({item["path"] for item in new_bypass_findings})
     counts_by_method = method_counts(current_findings)
+    custom_direct_senders = [
+        item
+        for item in current_findings
+        if item["detection_kind"] == "custom_direct_sender"
+    ]
+    direct_endpoints = [
+        item
+        for item in current_findings
+        if item["detection_kind"] == "direct_bot_api_endpoint"
+    ]
     source_summary = source_snapshot["summary"]
 
     return {
         "schema_version": "telegram_notification_egress_no_new_bypass_guard_v1",
         "generated_at": generated,
         "git_commit": git_short_sha(root),
-        "status": "pass_no_new_bypass" if not new_bypass_findings else "blocked_new_bypass_detected",
+        "status": "pass_no_direct_or_custom_bypass"
+        if not current_findings
+        else "blocked_direct_or_custom_bypass_detected",
         "mode": "repo_source_scan_no_secret_value_no_telegram_send",
         "source_snapshot": SOURCE_SNAPSHOT.as_posix(),
         "guarded_roots": [path.as_posix() for path in SCAN_ROOTS],
         "guarded_bot_methods": list(GUARDED_BOT_METHODS),
         "summary": {
-            "source_direct_bot_api_call_count": source_summary["direct_bot_api_call_count"],
-            "source_direct_bot_api_file_count": source_summary["direct_bot_api_file_count"],
+            "source_direct_bot_api_call_count": source_summary[
+                "direct_bot_api_call_count"
+            ],
+            "source_direct_bot_api_file_count": source_summary[
+                "direct_bot_api_file_count"
+            ],
             "baseline_signature_count": sum(baseline.values()),
+            "detected_direct_bot_api_call_count": len(detected_findings),
             "current_direct_bot_api_call_count": len(current_findings),
             "current_direct_bot_api_file_count": len(current_files),
+            "current_direct_bot_api_endpoint_count": len(direct_endpoints),
+            "current_custom_direct_sender_count": len(custom_direct_senders),
+            "github_frozen_legacy_direct_bot_api_call_count": len(
+                frozen_legacy_findings
+            ),
+            "github_frozen_legacy_direct_bot_api_file_count": len(
+                {item["path"] for item in frozen_legacy_findings}
+            ),
             "guarded_method_count": len(GUARDED_BOT_METHODS),
             "sendMessage_call_count": counts_by_method["sendMessage"],
             "sendDocument_call_count": counts_by_method["sendDocument"],
@@ -209,7 +750,9 @@ def build_report(root: Path, generated_at: str | None = None) -> dict[str, Any]:
             ),
             "new_bypass_count": len(new_bypass_findings),
             "new_bypass_file_count": len(new_bypass_files),
-            "removed_baseline_call_count": sum(item["removed_count"] for item in removed_baseline_signatures),
+            "removed_baseline_call_count": sum(
+                item["removed_count"] for item in removed_baseline_signatures
+            ),
             "runtime_gate_count": 0,
             "action_button_count": 0,
         },
@@ -218,6 +761,8 @@ def build_report(root: Path, generated_at: str | None = None) -> dict[str, Any]:
             "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,
             "api_sender_refactor_authorized": False,
             "secret_value_collection_allowed": False,
@@ -231,15 +776,18 @@ def build_report(root: Path, generated_at: str | None = None) -> dict[str, Any]:
             "not_authorization": True,
         },
         "current_direct_bot_api_calls": current_findings,
+        "detected_direct_bot_api_calls": detected_findings,
+        "frozen_legacy_direct_bot_api_calls": frozen_legacy_findings,
         "new_bypass_findings": new_bypass_findings,
         "removed_baseline_signatures": removed_baseline_signatures,
         "operator_interpretation": [
-            "new_bypass_count 維持 0 才代表沒有新增未登記 Telegram Bot API 直送旁路。",
+            "current_direct_bot_api_call_count 必須為 0，才代表沒有 Telegram direct/custom bypass。",
+            ".github/workflows 僅以 frozen_legacy_source_truth 盤點，不計入 active/new failure；runtime 與 workflow execution 均未授權且禁止執行。",
             (
                 f"committed inventory 目前有 {source_summary['direct_bot_api_call_count']} 個 direct Bot API call site；"
-                "若數值大於 0 仍是待 controlled migration 的基線，不代表已批准保留。"
+                "baseline 只做 drift 比對，不能批准或隱藏目前 source 旁路。"
             ),
-            "sendDocument / sendPhoto / sendMediaGroup 等附件型出口若出現在 repo source，會被視為新增旁路並阻擋。",
+            "分段 URL、requests/httpx/urllib、Invoke-WebRequest/Invoke-RestMethod 與自訂 sender 都會被掃描。",
             "本 guard 只讀 repo source 與 committed snapshot，不送 Telegram、不讀 Bot token、不修改 workflow / script / API sender。",
         ],
     }
@@ -248,10 +796,11 @@ def build_report(root: Path, generated_at: str | None = None) -> dict[str, Any]:
 def validate(root: Path) -> None:
     report = build_report(root)
     errors: list[str] = []
-    if report["summary"]["new_bypass_count"]:
-        for item in report["new_bypass_findings"]:
+    if report["summary"]["current_direct_bot_api_call_count"]:
+        for item in report["current_direct_bot_api_calls"]:
             errors.append(
-                f"{item['path']}:{item['line']}: 新增未登記 Telegram Bot API 旁路 {item['method']}"
+                f"{item['path']}:{item['line']}: Telegram direct/custom bypass "
+                f"{item['detection_kind']} {item['method']} ({item['function']})"
             )
 
     if errors:
@@ -262,10 +811,14 @@ def validate(root: Path) -> None:
 
 
 def main() -> None:
-    parser = argparse.ArgumentParser(description="檢查 Telegram 通知出口不可新增未登記 direct Bot API 旁路")
+    parser = argparse.ArgumentParser(
+        description="檢查 Telegram 通知出口不可新增未登記 direct Bot API 旁路"
+    )
     parser.add_argument("--root", default=".", help="repository root")
     parser.add_argument("--output", help="寫出 JSON 報告")
-    parser.add_argument("--generated-at", help="固定 generated_at 時間，供 committed snapshot 使用")
+    parser.add_argument(
+        "--generated-at", help="固定 generated_at 時間，供 committed snapshot 使用"
+    )
     args = parser.parse_args()
 
     root = Path(args.root).resolve()
@@ -273,13 +826,18 @@ def main() -> None:
     if args.output:
         output = Path(args.output)
         output.parent.mkdir(parents=True, exist_ok=True)
-        output.write_text(json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+        output.write_text(
+            json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True) + "\n",
+            encoding="utf-8",
+        )
 
     validate(root)
     summary = report["summary"]
     print(
         "TELEGRAM_NOTIFICATION_EGRESS_NO_NEW_BYPASS_GUARD_OK "
         f"current={summary['current_direct_bot_api_call_count']} "
+        f"custom={summary['current_custom_direct_sender_count']} "
+        f"frozen_legacy={summary['github_frozen_legacy_direct_bot_api_call_count']} "
         f"baseline={summary['baseline_signature_count']} "
         f"new={summary['new_bypass_count']} "
         f"sendDocument={summary['sendDocument_call_count']} "