v7 mech-6: production-evidence fail-closed live wiring seam (no real reader, no endpoint wiring yet)

- LiveEvidenceReader protocol: get_checks(repo, ref) /
  get_branch_protection(repo, branch), results carry status
  plus raw evidence; production code has no HTTP call, no
  token read, no real GitHub API call (AST-asserted).
- verify_live_evidence: no reader / no repo-ref / reader
  failure / mismatch all verified=False; comparison shared
  with verify_gate_evidence via _compare_claimed_vs_live.
- live_gate_decision: verified=False -> allowed=False,
  http_status=409, status=blocked; verified=True -> 200.
- Endpoints _merge_task/_release_task NOT wired yet (no
  production reader exists; wiring None would block all
  merges/releases); real live verification still pending
  B6 real App install and B1 user-side items.

(cherry picked from commit 7e8bfdee1f22c28329602314f3e166cde34eb6cf)

cloud/gate_verification.py

@@ -56,8 +56,95 @@ mismatch）当作阻断，核对器本身不用改。
 
 零第三方依赖，只用标准库。本模块不 import cloud
 其它模块（reader 由调用方注入），不发网络请求。
+
+v7 机制第 6 项（生产证据 fail-closed live 接线，本批新增，
+只动本模块，不动其它 7 项）：
+- 新增可注入 reader 接口 LiveEvidenceReader（协议）：
+  get_checks(repo, ref) 与 get_branch_protection(repo,
+  branch)，返回结构含状态与原始证据（LiveChecksResult /
+  LiveProtectionResult，或等价 dict）。本模块的生产代码
+  里绝不实现真实 HTTP 调用、绝不读真实 token、绝不调
+  用真实 GitHub API——接口是未来真实 reader 的插入缝，
+  不是真实 reader 本身。
+- 新增 verify_live_evidence(claimed, reader, repo, ref,
+  branch=None)：协议 reader 或旧式 callable reader 均可；
+  无 reader、无 repo/ref、reader 失败、reader 返回与声
+  称不一致时 verified 一律 False；声称 CI 全绿但实时
+  数据失败/缺失时记 mismatch。比对规则与
+  verify_gate_evidence 逐项一致（共用 _compare），不另
+  造口径；空声称不由本函数放行——缺失证据由硬门禁
+  （merge_gate_hard）阻断，本函数只核对“声称了的项”。
+- 新增 live_gate_decision：端点层判定函数，verified
+  为 False 时返回 allowed=False、http_status=409、
+  status="blocked"、blocked=True（mismatch 与未核对用
+  不同 code 区分：live_evidence_mismatch /
+  live_evidence_unverified）；verified 为 True 时返回
+  allowed=True、http_status=200、status="allowed"。
+- 诚实残留（不得冒充已接线）：HTTP 端点 _merge_task /
+  _release_task 本批未接 live_gate_decision——当前无
+  生产 reader 可注入：以 reader=None 接线会挡住全部合
+  并/发布（fail-closed 的本意不是停摆），以桩 reader
+  接线则是假求证。真实 live 验证仍待 B6 真实 App 安装
+  与 B1 等用户侧事项，未在本批完成。本批不改 #238
+  auto_apply（仍 False）、不改 #247“纠正为 pass 本身
+  不解封”、不改 v1.1 契约、不改任何阈值（阈值仍
+  test_default_unapproved）。
 """
 
+from dataclasses import dataclass
+
+try:
+    from typing import Protocol
+    from typing import runtime_checkable
+except ImportError:  # pragma: no cover - Python < 3.8
+    Protocol = object  # type: ignore
+
+    def runtime_checkable(cls):
+        return cls
+
+
+def _compare_claimed_vs_live(evidence, live):
+    """声称证据与实时数据的逐项比对（两核对函数共用，同一口径）。
+
+    只比对调用方声称的项；实时数据缺失某项时记 mismatch
+    （声称全绿却查无此项，不是“查不到所以通过”）。
+    返回 mismatches 列表（空即逐项全过）。
+    """
+    mismatches = []
+    live_checks = live.get("ci_checks") or {}
+    if not isinstance(live_checks, dict):
+        live_checks = {}
+    evidence = evidence or {}
+    if evidence.get("ci_status") == "success":
+        failed_live = sorted(
+            name for name, status in live_checks.items()
+            if status != "success")
+        if failed_live:
+            mismatches.append(
+                "ci_not_green_live: %s" % ", ".join(failed_live))
+    for check in (evidence.get("ci_checks") or []):
+        if not isinstance(check, dict):
+            continue
+        name = check.get("name")
+        if not name:
+            continue
+        if check.get("status") == "success" \
+                and live_checks.get(name) != "success":
+            mismatches.append(
+                "check_not_green_live: %s" % name)
+    for name in (evidence.get("required_checks") or []):
+        if live_checks.get(name) != "success":
+            mismatches.append(
+                "required_check_not_green_live: %s" % name)
+    claimed_protection = evidence.get("branch_protection")
+    live_protection = live.get("branch_protection")
+    if claimed_protection == "protected" \
+            and live_protection != "protected":
+        mismatches.append(
+            "branch_protection_not_protected_live: %s"
+            % live_protection)
+    return mismatches
+
 
 def verify_gate_evidence(evidence, reader=None, repo=None, ref=None):
     """只读核对调用方门禁证据与实时数据。
@@ -103,38 +190,246 @@ def verify_gate_evidence(evidence, reader=None, repo=None, ref=None):
         return {"checked": False, "verified": False,
                 "mismatches": [],
                 "reason": "reader_failed: non-dict result"}
-    mismatches = []
+    mismatches = _compare_claimed_vs_live(evidence, live)
-    live_checks = live.get("ci_checks") or {}
-    if not isinstance(live_checks, dict):
-        live_checks = {}
-    evidence = evidence or {}
-    if evidence.get("ci_status") == "success":
-        failed_live = sorted(
-            name for name, status in live_checks.items()
-            if status != "success")
-        if failed_live:
-            mismatches.append(
-                "ci_not_green_live: %s" % ", ".join(failed_live))
-    for check in (evidence.get("ci_checks") or []):
-        if not isinstance(check, dict):
-            continue
-        name = check.get("name")
-        if not name:
-            continue
-        if check.get("status") == "success" \
-                and live_checks.get(name) != "success":
-            mismatches.append(
-                "check_not_green_live: %s" % name)
-    for name in (evidence.get("required_checks") or []):
-        if live_checks.get(name) != "success":
-            mismatches.append(
-                "required_check_not_green_live: %s" % name)
-    claimed_protection = evidence.get("branch_protection")
-    live_protection = live.get("branch_protection")
-    if claimed_protection == "protected" \
-            and live_protection != "protected":
-        mismatches.append(
-            "branch_protection_not_protected_live: %s"
-            % live_protection)
     return {"checked": True, "verified": not mismatches,
             "mismatches": mismatches, "reason": None}
+
+
+@dataclass(frozen=True)
+class LiveChecksResult:
+    """get_checks 的返回结构：状态 + 原始证据。
+
+    checks：{检查名: 状态}，状态与声称口径一致
+    （"success" 为绿，其它一律非绿）。
+    raw：原始证据体（真实 reader 未来填 GitHub Checks
+    API 的原始 payload；假 reader 可填任意可追溯值，
+    也可为 None）。比对只用 checks，raw 只供审计追溯，
+    绝不参与“通过”判定。
+    """
+    checks: dict
+    raw: object = None
+
+
+@dataclass(frozen=True)
+class LiveProtectionResult:
+    """get_branch_protection 的返回结构：状态 + 原始证据。
+
+    status："protected" / "unprotected" / 其它；None 表
+    示实时数据无此项（声称 protected 时即 mismatch）。
+    raw：原始证据体（真实 reader 未来填分支保护 API
+    的原始 payload；假 reader 可填任意可追溯值，也可
+    为 None），同样只供追溯、不参与判定。
+    """
+    status: object = None
+    raw: object = None
+
+
+@runtime_checkable
+class LiveEvidenceReader(Protocol):
+    """可注入的实时证据 reader 接口（v7 机制第 6 项）。
+
+    生产代码只定义接口、不实现真实 reader：真实实现
+    （走 GitHub Checks / 分支保护只读接口）须在 B6 真
+    实 App 安装与 B1 等用户侧事项就绪后另行接入，本
+    模块绝不内置 HTTP 调用、token 读取或 GitHub API
+    调用。测试与验收一律用实现本协议的假 reader。
+    """
+
+    def get_checks(self, repo, ref):
+        """返回 LiveChecksResult（或等价 dict，见
+        _normalize_checks）：repo 在 ref 处的 CI 检查
+        实时状态与原始证据。"""
+        raise NotImplementedError
+
+    def get_branch_protection(self, repo, branch):
+        """返回 LiveProtectionResult（或等价 dict / 纯
+        状态字符串，见 _normalize_protection）：repo 的
+        branch 分支保护实时状态与原始证据。"""
+        raise NotImplementedError
+
+
+def _normalize_checks(result):
+    """归一化 get_checks 返回值 -> (checks dict, raw) 或 None。
+
+    接受 LiveChecksResult、含 "checks"/"ci_checks" dict
+    的 dict、以及本身就是 {名: 状态} 的 dict；其它形状
+    （非 dict checks、非字符串状态）返回 None，调用方
+    按 reader_failed 处理，不猜、不放行。
+    """
+    if isinstance(result, LiveChecksResult):
+        checks, raw = result.checks, result.raw
+    elif isinstance(result, dict):
+        raw = result.get("raw")
+        if isinstance(result.get("checks"), dict):
+            checks = result["checks"]
+        elif isinstance(result.get("ci_checks"), dict):
+            checks = result["ci_checks"]
+        else:
+            checks = result
+    else:
+        return None
+    if not isinstance(checks, dict):
+        return None
+    if any(not isinstance(status, str) for status in checks.values()):
+        return None
+    return checks, raw
+
+
+def _normalize_protection(result):
+    """归一化 get_branch_protection 返回值。
+
+    返回 (status, raw, malformed)：status 为字符串或
+    None（无此项）；malformed 为 True 时调用方按
+    reader_failed 处理（例如返回了整数等非法形状）。
+    纯字符串即状态本身；无 "status"/"branch_protection"
+    键的 dict 视为无此项（None），不是 malformed——
+    声称 protected 时它仍会记 mismatch。
+    """
+    if isinstance(result, LiveProtectionResult):
+        status, raw = result.status, result.raw
+    elif isinstance(result, str):
+        status, raw = result, None
+    elif result is None:
+        status, raw = None, None
+    elif isinstance(result, dict):
+        raw = result.get("raw")
+        if isinstance(result.get("status"), str):
+            status = result["status"]
+        elif isinstance(result.get("branch_protection"), str):
+            status = result["branch_protection"]
+        else:
+            status = None
+    else:
+        return None, None, True
+    if status is not None and not isinstance(status, str):
+        return None, None, True
+    return status, raw, False
+
+
+def _is_live_reader(reader):
+    """是否协议 reader：同时有 get_checks 与
+    get_branch_protection 两个方法（鸭子类型，不强制
+    isinstance，因为假 reader 未必显式继承协议）。"""
+    return callable(getattr(reader, "get_checks", None)) \
+        and callable(getattr(reader, "get_branch_protection", None))
+
+
+def verify_live_evidence(claimed, reader=None, repo=None, ref=None,
+                         branch=None):
+    """实时证据比对（v7 机制第 6 项，fail-closed）。
+
+    claimed：调用方声称的门禁证据 dict（与硬门禁同款
+    字段：ci_status、ci_checks[{name, status}]、
+    required_checks、branch_protection）。
+    reader：LiveEvidenceReader 协议实例，或旧式 callable
+    (repo, ref) -> {"ci_checks": {...}, "branch_protection":
+    str}（与 verify_gate_evidence 同款；兼容旧假 reader）。
+    repo/ref：求证目标；branch：分支保护的求证分支，
+    未给时用 ref 作分支名（调用方须保证 ref 在此语义
+    下就是分支；SHA 求证时请显式传 branch，绝不由本
+    函数猜分支与 SHA 的对应关系）。
+    返回 {"checked", "verified", "mismatches", "reason",
+    "live"}：live 为归一化后的实时数据（含原始证据
+    raw）或 None（未核对时）。verified 为 False 的情形：
+    无 reader、无 repo/ref、reader 抛异常、reader 返回
+    恶化形状、实时数据与声称不一致——一律不是“通过”。
+    比对规则与 verify_gate_evidence 共用 _compare，口
+    径一致；本函数不发网络请求、不读 token。
+    """
+    if reader is None:
+        return {"checked": False, "verified": False,
+                "mismatches": [],
+                "reason": "no_trusted_reader", "live": None}
+    if not repo or not ref:
+        return {"checked": False, "verified": False,
+                "mismatches": [],
+                "reason": "no_verification_target", "live": None}
+    if _is_live_reader(reader):
+        try:
+            checks_result = reader.get_checks(repo, ref)
+        except Exception as exc:
+            return {"checked": False, "verified": False,
+                    "mismatches": [],
+                    "reason": "reader_failed: %s" % exc,
+                    "live": None}
+        normalized = _normalize_checks(checks_result)
+        if normalized is None:
+            return {"checked": False, "verified": False,
+                    "mismatches": [],
+                    "reason": "reader_failed: malformed checks",
+                    "live": None}
+        checks, checks_raw = normalized
+        try:
+            protection_result = reader.get_branch_protection(
+                repo, branch if branch else ref)
+        except Exception as exc:
+            return {"checked": False, "verified": False,
+                    "mismatches": [],
+                    "reason": "reader_failed: %s" % exc,
+                    "live": None}
+        protection, protection_raw, malformed = \
+            _normalize_protection(protection_result)
+        if malformed:
+            return {"checked": False, "verified": False,
+                    "mismatches": [],
+                    "reason": "reader_failed: malformed protection",
+                    "live": None}
+        live = {"ci_checks": checks,
+                "branch_protection": protection,
+                "checks_raw": checks_raw,
+                "protection_raw": protection_raw}
+    elif callable(reader):
+        try:
+            live_raw = reader(repo, ref)
+        except Exception as exc:
+            return {"checked": False, "verified": False,
+                    "mismatches": [],
+                    "reason": "reader_failed: %s" % exc,
+                    "live": None}
+        if not isinstance(live_raw, dict):
+            return {"checked": False, "verified": False,
+                    "mismatches": [],
+                    "reason": "reader_failed: non-dict result",
+                    "live": None}
+        live = live_raw
+    else:
+        return {"checked": False, "verified": False,
+                "mismatches": [],
+                "reason": "reader_failed: unsupported reader type",
+                "live": None}
+    mismatches = _compare_claimed_vs_live(claimed, live)
+    return {"checked": True, "verified": not mismatches,
+            "mismatches": mismatches, "reason": None,
+            "live": live}
+
+
+def live_gate_decision(claimed, reader=None, repo=None, ref=None,
+                       branch=None):
+    """端点层 live 门禁判定（v7 机制第 6 项，fail-closed）。
+
+    verified=False 一律阻断，返回 allowed=False、
+    http_status=409、status="blocked"、blocked=True：
+    code 为 live_evidence_mismatch（已核对但不一致）
+    或 live_evidence_unverified（未核对：无 reader、
+    无目标、reader 失败）。verified=True 时返回
+    allowed=True、http_status=200、status="allowed"、
+    code="live_evidence_verified"。返回里附完整 live
+    比对结果（live 键），供端点写进 409 响应与审计。
+    注意：本函数是判定缝，不是端点接线——HTTP 端点
+    须在有生产 reader 可注入时才调用它；无 reader 时
+    调用它会阻断（这是 fail-closed 的正确语义），所
+    以在生产 reader 就绪前端点不得接线（见模块头残
+    留）。本函数本身是纯函数，不写审计、不改状态。
+    """
+    result = verify_live_evidence(claimed, reader, repo, ref,
+                                  branch=branch)
+    if result["verified"]:
+        return {"allowed": True, "blocked": False,
+                "status": "allowed", "http_status": 200,
+                "code": "live_evidence_verified",
+                "live": result}
+    code = "live_evidence_mismatch" if result["checked"] \
+        else "live_evidence_unverified"
+    return {"allowed": False, "blocked": True,
+            "status": "blocked", "http_status": 409,
+            "code": code, "live": result}

cloud/tests/test_scorecard_mech_live.py

@@ -0,0 +1,311 @@
+"""v7 机制第 6 项：生产证据 fail-closed live 接线定向测试。
+
+只测这一项，不碰其它 7 项，不改阈值（阈值仍
+test_default_unapproved，见 cloud/audit_scorecard.py，
+本文件不测阈值）、不改 #238 auto_apply、不改 #247
+“纠正为 pass 本身不解封”、不改 v1.1 契约。
+
+被测对象（cloud/gate_verification.py）：
+- LiveEvidenceReader 协议：get_checks(repo, ref) /
+  get_branch_protection(repo, branch)，返回结构含状态
+  与原始证据；生产代码里绝无真实 HTTP 调用、真实
+  token 读取、真实 GitHub API 调用（本文件以 AST 断
+  言钉住这一点）。
+- verify_live_evidence(claimed, reader, repo, ref)：无
+  reader、无 repo/ref、reader 失败、reader 返回与声称
+  不一致时 verified 一律 False；声称 CI 全绿但假 reader
+  返回失败/缺失时记 mismatch。
+- live_gate_decision：verified=False 时返回 409 语义
+  （allowed=False、http_status=409、status="blocked"、
+  blocked=True），verified=True 时返回 200 语义。
+
+诚实边界：本文件测的是判定函数，不冒充 HTTP 端点
+已接线——_merge_task / _release_task 本批未接
+live_gate_decision（无生产 reader 可注入；见
+gate_verification 模块头残留），真实 live 验证仍待
+B6 真实 App 安装与 B1 等用户侧事项。
+"""
+
+import ast
+import unittest
+from pathlib import Path
+
+try:
+    from cloud.gate_verification import LiveChecksResult
+    from cloud.gate_verification import LiveEvidenceReader
+    from cloud.gate_verification import LiveProtectionResult
+    from cloud.gate_verification import live_gate_decision
+    from cloud.gate_verification import verify_live_evidence
+except ImportError:
+    import sys
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.gate_verification import LiveChecksResult
+    from cloud.gate_verification import LiveEvidenceReader
+    from cloud.gate_verification import LiveProtectionResult
+    from cloud.gate_verification import live_gate_decision
+    from cloud.gate_verification import verify_live_evidence
+
+
+def _claimed_green():
+    return {
+        "ci_status": "success",
+        "ci_checks": [{"name": "ci", "status": "success"}],
+        "required_checks": ["ci"],
+        "branch_protection": "protected",
+    }
+
+
+class FakeReader(LiveEvidenceReader):
+    """实现协议的假 reader：不发网络请求，只返回预设值。
+
+    同时记录收到的 repo/ref/branch，断言接线把求证目标
+    原样传给 reader（不猜目标）。
+    """
+
+    def __init__(self, checks, protection, checks_raw=None,
+                 protection_raw=None):
+        self.checks = checks
+        self.protection = protection
+        self.checks_raw = checks_raw
+        self.protection_raw = protection_raw
+        self.seen = []
+
+    def get_checks(self, repo, ref):
+        self.seen.append(("checks", repo, ref))
+        return LiveChecksResult(checks=dict(self.checks),
+                                raw=self.checks_raw)
+
+    def get_branch_protection(self, repo, branch):
+        self.seen.append(("protection", repo, branch))
+        return LiveProtectionResult(status=self.protection,
+                                    raw=self.protection_raw)
+
+
+class FailingReader(LiveEvidenceReader):
+    def __init__(self, fail_on):
+        self.fail_on = fail_on
+
+    def get_checks(self, repo, ref):
+        if self.fail_on == "checks":
+            raise RuntimeError("github down")
+        return LiveChecksResult(checks={"ci": "success"})
+
+    def get_branch_protection(self, repo, branch):
+        if self.fail_on == "protection":
+            raise RuntimeError("github down")
+        return LiveProtectionResult(status="protected")
+
+
+class LiveEvidenceTestCase(unittest.TestCase):
+    def test_matching_fake_reader_verified(self):
+        reader = FakeReader({"ci": "success"}, "protected",
+                            checks_raw={"total_count": 1},
+                            protection_raw={"required": True})
+        result = verify_live_evidence(
+            _claimed_green(), reader, "o/r", "sha1",
+            branch="main")
+        self.assertTrue(result["checked"])
+        self.assertTrue(result["verified"])
+        self.assertEqual(result["mismatches"], [])
+        self.assertIsNone(result["reason"])
+        self.assertEqual(result["live"]["ci_checks"],
+                         {"ci": "success"})
+        self.assertEqual(result["live"]["branch_protection"],
+                         "protected")
+        self.assertEqual(result["live"]["checks_raw"],
+                         {"total_count": 1})
+        self.assertEqual(result["live"]["protection_raw"],
+                         {"required": True})
+        self.assertIn(("checks", "o/r", "sha1"), reader.seen)
+        self.assertIn(("protection", "o/r", "main"), reader.seen)
+
+    def test_lied_all_green_vs_failing_live_mismatch_to_409(self):
+        # 谎报全绿：声称 ci success + required ci，但实时数据
+        # 里 ci 是 failure、且 required 的 lint 查无此项。
+        claimed = _claimed_green()
+        claimed["required_checks"] = ["ci", "lint"]
+        reader = FakeReader({"ci": "failure"}, "protected")
+        result = verify_live_evidence(claimed, reader, "o/r", "sha1")
+        self.assertTrue(result["checked"])
+        self.assertFalse(result["verified"])
+        self.assertTrue(any("ci_not_green_live" in m
+                            for m in result["mismatches"]))
+        self.assertTrue(any("check_not_green_live: ci" in m
+                            for m in result["mismatches"]))
+        self.assertTrue(any("required_check_not_green_live: lint" in m
+                            for m in result["mismatches"]))
+        decision = live_gate_decision(claimed, reader, "o/r", "sha1")
+        self.assertFalse(decision["allowed"])
+        self.assertTrue(decision["blocked"])
+        self.assertEqual(decision["http_status"], 409)
+        self.assertEqual(decision["status"], "blocked")
+        self.assertEqual(decision["code"], "live_evidence_mismatch")
+        self.assertFalse(decision["live"]["verified"])
+
+    def test_protection_mismatch_blocks(self):
+        reader = FakeReader({"ci": "success"}, "unprotected")
+        result = verify_live_evidence(
+            _claimed_green(), reader, "o/r", "sha1")
+        self.assertTrue(result["checked"])
+        self.assertFalse(result["verified"])
+        self.assertTrue(any("branch_protection" in m
+                            for m in result["mismatches"]))
+        decision = live_gate_decision(
+            _claimed_green(), reader, "o/r", "sha1")
+        self.assertEqual(decision["http_status"], 409)
+        self.assertEqual(decision["status"], "blocked")
+
+    def test_no_reader_is_not_a_pass(self):
+        result = verify_live_evidence(
+            _claimed_green(), None, "o/r", "sha1")
+        self.assertFalse(result["checked"])
+        self.assertFalse(result["verified"])
+        self.assertEqual(result["reason"], "no_trusted_reader")
+        self.assertIsNone(result["live"])
+        decision = live_gate_decision(
+            _claimed_green(), None, "o/r", "sha1")
+        self.assertFalse(decision["allowed"])
+        self.assertEqual(decision["http_status"], 409)
+        self.assertEqual(decision["status"], "blocked")
+        self.assertEqual(decision["code"], "live_evidence_unverified")
+
+    def test_no_repo_or_ref_is_not_a_pass(self):
+        reader = FakeReader({"ci": "success"}, "protected")
+        for repo, ref in ((None, "sha1"), ("o/r", None),
+                          ("o/r", ""), ("", "sha1")):
+            result = verify_live_evidence(
+                _claimed_green(), reader, repo, ref)
+            self.assertFalse(result["checked"])
+            self.assertFalse(result["verified"])
+            self.assertEqual(result["reason"],
+                             "no_verification_target")
+            decision = live_gate_decision(
+                _claimed_green(), reader, repo, ref)
+            self.assertEqual(decision["http_status"], 409)
+
+    def test_reader_failure_is_not_a_pass(self):
+        for fail_on in ("checks", "protection"):
+            result = verify_live_evidence(
+                _claimed_green(), FailingReader(fail_on),
+                "o/r", "sha1")
+            self.assertFalse(result["checked"])
+            self.assertFalse(result["verified"])
+            self.assertTrue(
+                result["reason"].startswith("reader_failed"))
+            decision = live_gate_decision(
+                _claimed_green(), FailingReader(fail_on),
+                "o/r", "sha1")
+            self.assertEqual(decision["http_status"], 409)
+            self.assertEqual(decision["code"],
+                             "live_evidence_unverified")
+
+    def test_malformed_reader_result_is_not_a_pass(self):
+        class BadChecks(LiveEvidenceReader):
+            def get_checks(self, repo, ref):
+                return 42
+
+            def get_branch_protection(self, repo, branch):
+                return LiveProtectionResult(status="protected")
+
+        class BadStatus(LiveEvidenceReader):
+            def get_checks(self, repo, ref):
+                return LiveChecksResult(checks={"ci": 1})
+
+            def get_branch_protection(self, repo, branch):
+                return LiveProtectionResult(status="protected")
+
+        for reader in (BadChecks(), BadStatus()):
+            result = verify_live_evidence(
+                _claimed_green(), reader, "o/r", "sha1")
+            self.assertFalse(result["checked"])
+            self.assertFalse(result["verified"])
+            self.assertTrue(
+                result["reason"].startswith("reader_failed"))
+
+    def test_matching_fake_reader_decision_allows(self):
+        reader = FakeReader({"ci": "success"}, "protected")
+        decision = live_gate_decision(
+            _claimed_green(), reader, "o/r", "sha1",
+            branch="main")
+        self.assertTrue(decision["allowed"])
+        self.assertFalse(decision["blocked"])
+        self.assertEqual(decision["http_status"], 200)
+        self.assertEqual(decision["status"], "allowed")
+        self.assertEqual(decision["code"], "live_evidence_verified")
+
+    def test_branch_defaults_to_ref_and_explicit_branch_wins(self):
+        reader = FakeReader({"ci": "success"}, "protected")
+        verify_live_evidence(_claimed_green(), reader, "o/r", "sha1")
+        self.assertIn(("protection", "o/r", "sha1"), reader.seen)
+        reader2 = FakeReader({"ci": "success"}, "protected")
+        verify_live_evidence(_claimed_green(), reader2, "o/r", "sha1",
+                             branch="main")
+        self.assertIn(("protection", "o/r", "main"), reader2.seen)
+
+    def test_callable_legacy_reader_still_works(self):
+        def live_ok(repo, ref):
+            self.assertEqual((repo, ref), ("o/r", "sha1"))
+            return {"ci_checks": {"ci": "success"},
+                    "branch_protection": "protected"}
+
+        result = verify_live_evidence(
+            _claimed_green(), live_ok, "o/r", "sha1")
+        self.assertTrue(result["verified"])
+
+        def live_bad(repo, ref):
+            return {"ci_checks": {"ci": "failure"},
+                    "branch_protection": "protected"}
+
+        result = verify_live_evidence(
+            _claimed_green(), live_bad, "o/r", "sha1")
+        self.assertFalse(result["verified"])
+        self.assertTrue(any("ci_not_green_live" in m
+                            for m in result["mismatches"]))
+        decision = live_gate_decision(
+            _claimed_green(), live_bad, "o/r", "sha1")
+        self.assertEqual(decision["http_status"], 409)
+
+    def test_no_real_github_call_in_production_module(self):
+        # 注意：旧模块头 docstring 本来就以文字描述过“真实
+        # reader 用 github_app 的 _github_api_request”（未来
+        # 设想，非代码），所以断言须区分文字与代码：代码
+        # 位置（import / 调用 / 非 docstring 字符串）里不
+        # 得出现真实调用，docstring 文字不算。
+        path = Path(__file__).resolve().parents[1] / "gate_verification.py"
+        tree = ast.parse(path.read_text())
+        imported = set()
+        docstring_nodes = set()
+        for node in ast.walk(tree):
+            if isinstance(node, (ast.Module, ast.ClassDef,
+                                 ast.FunctionDef,
+                                 ast.AsyncFunctionDef)):
+                body = node.body
+                if body and isinstance(body[0], ast.Expr) \
+                        and isinstance(body[0].value, ast.Constant) \
+                        and isinstance(body[0].value.value, str):
+                    docstring_nodes.add(id(body[0].value))
+            if isinstance(node, ast.Import):
+                imported.update(a.name.split(".")[0]
+                                for a in node.names)
+            elif isinstance(node, ast.ImportFrom):
+                if node.module:
+                    imported.add(node.module.split(".")[0])
+        for banned in ("urllib", "requests", "http", "socket"):
+            self.assertNotIn(banned, imported,
+                             "production module must not import %s"
+                             % banned)
+        code_idents = set()
+        for node in ast.walk(tree):
+            if isinstance(node, ast.Attribute):
+                code_idents.add(node.attr)
+            elif isinstance(node, ast.Name):
+                code_idents.add(node.id)
+            elif isinstance(node, ast.Constant) \
+                    and isinstance(node.value, str) \
+                    and id(node) not in docstring_nodes:
+                self.assertNotIn("api.github.com", node.value)
+        self.assertNotIn("_github_api_request", code_idents)
+
+
+if __name__ == "__main__":
+    unittest.main()