codex-app-server/cloud/merge_gate_hard.py

"""服务端合并硬门禁：五项全查、缺一 409、fail-closed（纯函数）。

本模块堆叠在 #209（cloud/merge_gate.py）之上，不改 #209 纯函数
本身的语义，但服务端不再经由它决定模式：
- #209 的 ``evaluate_merge_gate`` / ``gate_from_request`` 只是
  纯函数（record 默认、无证据不阻断、unknown 只记风险不阻断），
  自 A1 起不再接线到任何服务端端点；它们的 record/enforce/off
  语义是纯函数的历史语义，不是服务端行为，不得据此推断服务端
  会放行（见 cloud/merge_gate.py 头部的 A1 声明）。
- 服务端唯一门禁就是本模块的 hard：POST /tasks/{id}/merge
  一律按 hard 求值；环境变量 CLOUD_MERGE_GATE_MODE 与请求体
  ``merge_gate_mode`` 在服务端一律被忽略——旧值 record/enforce/off
  不再是可选项，设了也等同 hard，不存在“显式降级”的服务端路径。
  （此前 hard_mode_from_env 曾把显式 record/enforce/off 原样沿用，
  A1 已删除该解析与 HARD_MODES：删模式，不是改默认值。）

模式（服务端唯一：hard）：

硬门禁五项（全部是调用方在 /merge 请求体里提交的证据；
本服务没有可信渠道直连 GitHub 核实，所以门禁强制的是“证据齐全
且全绿”，不是“服务端已向 GitHub 求证”——这条诚实边界必须写清，
调用方谎报证据不在本模块的验证范围内，由审计（第 3 轨）与
PR/Checks 的只读核对去覆盖）：
  ① CI 全绿：``ci_status`` 必须为 ``"success"``，且 ``ci_checks``
     非空、每项 status 均为 ``"success"``，且 ``required_checks``
     非空、每项都在 ``ci_checks`` 里且为 success。空
     ``required_checks``（``[]``）= 未配置，按分支保护 payload
     的同一口径处理（见 docs/merge-gate-branch-protection.md），
     不是“没有必须项所以通过”。
  ② 目录隔离：复用 #209 的 ``check_directory_isolation``（v0.3 口径：
     merge-base 范围 ``git diff --name-only``，拥有目录前缀检查，
     含 ``apps/web2`` 非子目录边界与 ``../``、绝对路径逃逸）。
     ``files_changed`` 缺失 = 无证据，直接阻断（#209 里是跳过，
     硬模式里是阻断，这是本轨改动的核心之一）；越目录只能靠
     ``approved_exceptions``（协调员批准的契约变更文件）放行，
     显式确认绝不覆盖目录违规。
  ③ 合同版本：``contract_version`` 非空，且 ``contract_frozen``
     为 true；若同时给了 ``contract_expected_version``（冻结版），
     两者必须相等，否则阻断。服务端没有独立的合同版本存储，
     冻结版由调用方（协调员侧）声明， mismatch 时阻断。
  ④ 协调员批准：``coordinator_approved`` 必须为 true；缺失与
     显式 false 用不同的 block 码区分（缺证据 vs 已否决）。
  ⑤ 无未解决评审意见：``unresolved_review_comments`` 必须为 0
     （整数）或空列表；缺失 = 无证据，阻断；大于 0 / 非空 = 阻断。

分支保护（硬模式下的第六条隐含项，来自 #209 的保护定义）：
  ``branch_protection`` 必须为 ``"protected"``。``"unprotected"``
  即使带 ``merge_gate_confirm`` 也阻断——硬模式没有“确认即放行”
  的逃生口（旧 record 模式下确认可覆盖保护风险的语义已随 A1
  删模式一并失效，服务端不再有确认放行的路径）。``"unknown"``
  （未声明或非法值）同样阻断：unknown 不等于 protected，这是
  #209 里“unknown 只记风险”的过渡语义在硬模式下的收紧。

与第 3 轨（独立审计 + 自动推回）的接口约定（本模块不造推回状态机）：
  - 阻塞依据 = 409 响应里的 ``merge_gate`` 对象：``mode``、
    ``blocks``（稳定的机器可读码，下面的 _BLOCK_DESCRIPTIONS 有
    逐条中文说明）、以及 ``ci`` / ``directory`` / ``contract`` /
    ``coordinator`` / ``reviews`` 五个分项。推回包逐条取 blocks，
    不要解析审计 detail 文本。
  - 审计依据 = 每次门禁阻断写一条 ``merge_gate_blocked`` 审计
    （action 名固定，detail 为 ``"<task_id> mode=<mode>
    blocks=<逗号分隔的 blocks>"``），任务状态不变（UPDATE 不执行）。
  - 门禁结论是单次调用的判定，不持久化为任务状态：一次 200 不
    代表之后另一次合并自动通过；修复后必须重新提交完整证据、
    重新走全部五项（完整重审，非只审改动点）。
  - 本模块不建推回任务、不发通知、不改任务状态机：推回执行、
    轮次上限、升级路径全部归第 3 轨。
"""

from cloud.merge_gate import (
    PROTECTION_STATES,
    RISK_UNKNOWN,
    RISK_UNPROTECTED,
    check_directory_isolation,
)

_BLOCK_DESCRIPTIONS = {
    "ci_evidence_missing": "CI 证据缺失（ci_status / ci_checks / required_checks 有一项未给）",
    "ci_not_green": "CI 状态不是 success",
    "ci_check_failed": "CI 检查项中有非 success 项",
    "ci_required_checks_unconfigured": "required_checks 为空 = 未配置",
    "ci_required_check_missing": "required_checks 中有项不在 CI 检查里或不是 success",
    "directory_evidence_missing": "目录证据缺失（files_changed 未给）",
    "directory_owned_dir_unknown": "拥有目录未知，无法核对目录隔离",
    "directory_isolation_violation": "目录隔离违规（越拥有目录）",
    "contract_version_missing": "合同版本缺失",
    "contract_not_frozen": "合同未冻结（contract_frozen 不是 true）",
    "contract_version_mismatch": "合同版本与冻结版不一致",
    "coordinator_approval_missing": "协调员批准证据缺失",
    "coordinator_not_approved": "协调员未批准",
    "review_evidence_missing": "评审意见证据缺失",
    "unresolved_review_comments": "存在未解决评审意见",
    "branch_protection_unknown": "分支保护状态未知（unknown 不等于 protected）",
    "branch_protection_unprotected": "分支未保护（硬模式下显式确认也不放行）",
}


def _ci_checks_index(ci_checks):
    """ci_checks 列表 -> {name: status}；条目不合规时返回 None。"""
    index = {}
    for entry in ci_checks:
        if not isinstance(entry, dict):
            return None
        name = entry.get("name")
        status = entry.get("status")
        if not isinstance(name, str) or not name \
                or not isinstance(status, str):
            return None
        index[name] = status
    return index


def evaluate_hard_merge_gate(files_changed=None, owned_dir=None,
                             approved_exceptions=(), protection="unknown",
                             ci_status=None, ci_checks=None,
                             required_checks=None, contract_version=None,
                             contract_frozen=False,
                             contract_expected_version=None,
                             coordinator_approved=None,
                             unresolved_review_comments=None):
    """硬门禁纯函数：五项 + 保护，缺一即 allowed=False。

    files_changed=None、ci_checks=None、required_checks=None、
    unresolved_review_comments=None 全部是“无证据”，不是“跳过”。
    protection 非法值归一化为 unknown，随后按 unknown 阻断。
    返回 dict：allowed / mode / protection / directory / ci /
    contract / coordinator / reviews / risks / blocks。
    """
    protection = protection if protection in PROTECTION_STATES \
        else "unknown"
    risks = []
    blocks = []

    # ① CI 全绿。
    if ci_status is None or ci_status == "" \
            or ci_checks is None or required_checks is None:
        blocks.append("ci_evidence_missing")
    else:
        if ci_status != "success":
            blocks.append("ci_not_green")
        if not isinstance(ci_checks, list) or not ci_checks:
            blocks.append("ci_evidence_missing")
            index = None
        else:
            index = _ci_checks_index(ci_checks)
            if index is None:
                blocks.append("ci_check_failed")
            elif any(status != "success" for status in index.values()):
                blocks.append("ci_check_failed")
        if isinstance(required_checks, list) and not required_checks:
            blocks.append("ci_required_checks_unconfigured")
        elif isinstance(required_checks, list) and index is not None:
            missing = [name for name in required_checks
                       if index.get(name) != "success"]
            if missing:
                blocks.append("ci_required_check_missing")
        elif not isinstance(required_checks, list):
            blocks.append("ci_evidence_missing")
    ci = {"status": ci_status, "checks": ci_checks,
          "required_checks": required_checks,
          "all_green": False}  # 末尾按 CI 分项 blocks 重算，见下。

    # ② 目录隔离（复用 #209，不另造口径）。
    directory = None
    if files_changed is None:
        blocks.append("directory_evidence_missing")
    else:
        directory = check_directory_isolation(
            files_changed, owned_dir, approved_exceptions)
        if directory["inconclusive"]:
            blocks.append("directory_owned_dir_unknown")
        elif not directory["passed"]:
            blocks.append("directory_isolation_violation")

    # ③ 合同版本。
    if contract_version is None or (
            isinstance(contract_version, str)
            and not contract_version.strip()):
        blocks.append("contract_version_missing")
    if contract_frozen is not True:
        blocks.append("contract_not_frozen")
    if contract_expected_version is not None \
            and contract_version is not None \
            and contract_version != contract_expected_version:
        blocks.append("contract_version_mismatch")
    contract = {"version": contract_version, "frozen": contract_frozen,
                "expected": contract_expected_version}

    # ④ 协调员批准。
    if coordinator_approved is None:
        blocks.append("coordinator_approval_missing")
    elif coordinator_approved is not True:
        blocks.append("coordinator_not_approved")
    coordinator = {"approved": coordinator_approved}

    # ⑤ 无未解决评审意见。
    if unresolved_review_comments is None:
        blocks.append("review_evidence_missing")
        reviews = {"unresolved": None}
    elif isinstance(unresolved_review_comments, bool):
        blocks.append("review_evidence_missing")
        reviews = {"unresolved": None}
    elif isinstance(unresolved_review_comments, list):
        reviews = {"unresolved": unresolved_review_comments}
        if unresolved_review_comments:
            blocks.append("unresolved_review_comments")
    elif isinstance(unresolved_review_comments, int):
        reviews = {"unresolved": unresolved_review_comments}
        if unresolved_review_comments > 0:
            blocks.append("unresolved_review_comments")
        elif unresolved_review_comments < 0:
            blocks.append("review_evidence_missing")
    else:
        blocks.append("review_evidence_missing")
        reviews = {"unresolved": None}

    # 分支保护：硬模式下 unknown 与 unprotected 一律阻断，确认无效。
    if protection == "unprotected":
        risks.append(RISK_UNPROTECTED)
        blocks.append("branch_protection_unprotected")
    elif protection == "unknown":
        risks.append(RISK_UNKNOWN)
        blocks.append("branch_protection_unknown")

    # ci.all_green 只反映 CI 分项本身，不被其它项的 blocks 污染。
    ci_block_codes = ("ci_evidence_missing", "ci_not_green",
                      "ci_check_failed", "ci_required_checks_unconfigured",
                      "ci_required_check_missing")
    ci["all_green"] = not any(code in blocks for code in ci_block_codes)

    return {"allowed": not blocks, "mode": "hard",
            "protection": protection, "directory": directory,
            "ci": ci, "contract": contract,
            "coordinator": coordinator, "reviews": reviews,
            "risks": risks, "blocks": blocks}


def block_descriptions(blocks):
    """blocks 码 -> 中文说明（供第 3 轨推回包逐条引用）。"""
    return [_BLOCK_DESCRIPTIONS.get(code, code) for code in blocks or []]


def blocked_audit_detail(task_id, gate):
    """merge_gate_blocked 审计的 detail：只含任务 id、模式与 blocks 码。

    推回包不要解析这段文本，取 409 响应的 merge_gate 对象为准。
    """
    return "%s mode=%s blocks=%s" % (
        task_id, gate.get("mode", "hard"),
        ",".join(gate.get("blocks") or []))