codex-app-server/tools/chv_merge_precheck.py

#!/usr/bin/env python3
"""CHV 合并前检查（只读，绝不合并、绝不改 CI、绝不写库）。

用途：合并 §13.1/§14 链上 PR 之前，逐个核对 base/head、
mergeable、CI 状态、与 main 头的关系与堆叠顺序，输出
诚实结论。CI 仍是 UNSTABLE 就如实报 NOT-READY，不把
账单模式的秒级失败读成代码失败，也不把 UNSTABLE 读成绿。

用法：
    python3 tools/chv_merge_precheck.py [PR号 ...]
缺省检查整条 CHV 链（不含本轨新 PR；新 PR 建好后把
它的号追加到参数末尾再跑一次）：
    python3 tools/chv_merge_precheck.py 216 221 230 232 234 \\
        239 246 252 254 257 264 <本轨新 PR 号>

依赖：gh（已登录）与 git（仅读本地 remote，不 fetch、
不 checkout、不 merge）。gh 不可用或 API 失败时退出码 2。
退出码：0 = 全部就绪（可按顺序合并）；1 = 未就绪（见原因）；
2 = 工具/查询失败（结论不可信，不要据此合并）。
"""

import json
import subprocess
import sys

DEFAULT_STACK = [216, 221, 230, 232, 234, 239, 246, 252, 254, 257, 264]


def _run(cmd):
    proc = subprocess.run(cmd, capture_output=True, text=True)
    if proc.returncode != 0:
        raise RuntimeError("%s 失败：%s" % (
            " ".join(cmd), (proc.stderr or proc.stdout).strip()[:500]))
    return proc.stdout


def _gh_json(args):
    return json.loads(_run(["gh"] + args))


def _check_summary(rollup):
    """把 statusCheckRollup 归纳成一句话，不猜失败原因。"""
    if not rollup:
        return "NO-CHECKS"
    states = []
    for item in rollup:
        name = item.get("name") or item.get("workflowName") or "?"
        conclusion = item.get("conclusion")
        status = item.get("status")
        states.append("%s:%s" % (name, conclusion or status))
    conclusions = {item.get("conclusion") for item in rollup}
    if "FAILURE" in conclusions or "CANCELLED" in conclusions:
        verdict = "FAIL"
    elif conclusions <= {"SUCCESS", "SKIPPED", "NEUTRAL", None} \
            and all(item.get("status") == "COMPLETED" for item in rollup):
        verdict = "GREEN"
    else:
        verdict = "PENDING"
    return "%s [%s]" % (verdict, "; ".join(states))


def main(argv):
    try:
        prs = [int(a) for a in argv] or list(DEFAULT_STACK)
    except ValueError:
        print("参数须为 PR 号（整数）", file=sys.stderr)
        return 2
    try:
        repo = _gh_json(["repo", "view", "--json", "nameWithOwner"])[
            "nameWithOwner"]
        main_head = _run(
            ["gh", "api", "repos/%s/commits/main" % repo, "--jq",
             ".sha"]).strip()
    except Exception as exc:
        print("查询仓库/main 头失败：%s" % exc, file=sys.stderr)
        return 2
    print("仓库：%s" % repo)
    print("当前 main 头：%s" % main_head)
    print("链（按合并顺序）：%s" % " → ".join(str(p) for p in prs))
    print("")
    problems = []
    tool_failed = False
    prev_head_branch = "main"
    for number in prs:
        try:
            pr = _gh_json(["pr", "view", str(number), "--json",
                           "number,title,state,baseRefName,headRefName,"
                           "headRefOid,mergeable,mergeStateStatus,"
                           "statusCheckRollup,additions,deletions,"
                           "changedFiles"])
        except Exception as exc:
            print("#%d：查询失败：%s" % (number, exc))
            tool_failed = True
            continue
        chain_ok = pr["baseRefName"] == prev_head_branch
        ci = _check_summary(pr.get("statusCheckRollup") or [])
        ready = (pr["state"] == "OPEN" and pr["mergeable"] == "MERGEABLE"
                 and pr["mergeStateStatus"] == "CLEAN" and chain_ok
                 and ci.startswith("GREEN"))
        print("#%(number)d %(title)s" % pr)
        print("  base=%(baseRefName)s head=%(headRefName)s@%(headRefOid)s"
              % pr)
        print("  state=%(state)s mergeable=%(mergeable)s "
              "mergeState=%(mergeStateStatus)s" % pr)
        print("  diff: +%s/-%s, %s files" % (
            pr.get("additions"), pr.get("deletions"),
            pr.get("changedFiles")))
        print("  CI: %s" % ci)
        print("  链衔接（base 应为上一 PR 的 head 分支 %s）：%s" % (
            prev_head_branch, "OK" if chain_ok else "MISMATCH"))
        print("  本项：%s" % ("READY" if ready else "NOT-READY"))
        if not ready:
            reasons = []
            if pr["state"] != "OPEN":
                reasons.append("state=%s" % pr["state"])
            if pr["mergeable"] != "MERGEABLE":
                reasons.append("mergeable=%s" % pr["mergeable"])
            if pr["mergeStateStatus"] != "CLEAN":
                reasons.append("mergeState=%s"
                               % pr["mergeStateStatus"])
            if not chain_ok:
                reasons.append("base 衔接断裂")
            if not ci.startswith("GREEN"):
                reasons.append("CI 未绿（%s）" % ci)
            problems.append("#%d：%s" % (number, "; ".join(reasons)))
        prev_head_branch = pr["headRefName"]
    print("")
    if tool_failed:
        print("结论：查询失败，结论不可信，不要据此合并。")
        return 2
    if problems:
        print("结论：NOT-READY，禁止合并。原因：")
        for problem in problems:
            print("  - %s" % problem)
        print("注意：CI 的秒级 verify 失败此前被定性为账单模式，"
              "脚本不据此改判——账单恢复变绿前，结论保持 NOT-READY。")
        return 1
    print("结论：READY，可按上面顺序逐个合并（合并后仍默认关闭，"
          "再跑 §14 灰度清单）。")
    return 0


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))