weichen
ba1d3b2b81
Why: PR #682 merged a stale branch that reset five component pins to pre-batch values, silently dropping the batch-1 composition on main. Evidence: materialized byte-exact from the reviewed transport (gatemat-run.json); regression/forward/same/skip/missing-object cases (verify-pingate.json); make test-governance and test-pins re-run below. Impact: new check script wired into check-pins and task-pr-check; PIN_REGRESSION_SKIP=1 documents deliberate downgrades. Refs: feature/audit-followups-pin-gate-m5
320 lines
13 KiB
Python
320 lines
13 KiB
Python
#!/usr/bin/env python3
|
||
"""PR base/head 门禁（两级分支模型：child → initiative → main）。
|
||
|
||
子命令：
|
||
local [--pr N] [--final]
|
||
核验当前检出的任务分支及其 PR（事实来源是 Gitea 与 origin）：PR 处于开放状态、base 为卡片期望目标、
|
||
PR head 与本地 HEAD 一致、origin 分支头与本地 HEAD 一致、期望目标分支存在；child 的父 initiative
|
||
须仍有开放 PR 指向其 root_target。--final（仅 initiative）：accepting_requirements=false，且
|
||
`## 子任务登记` 中「必需=是」的子任务为「已合入」（PR 已合并入本 initiative）或「已延期」。
|
||
ci
|
||
仅 pull_request 事件：按 GITHUB_HEAD_REF 在检出的 .task/ 顶层取卡片（顶层无卡时取 .task/archive/
|
||
中状态为「已完成（合入前闭合）」的卡），核对 GITHUB_BASE_REF。
|
||
不联网、不需要 token；非 pull_request 事件跳过（退出 0）。
|
||
|
||
期望目标：child → immediate_target；其余（initiative / standalone / 无卡）→ root_target（无卡即 main）。
|
||
local 与 ci 均运行 check_pin_regression.sh（gitlink 相对 origin/main 不得倒退）；非零即拒绝，
|
||
PIN_REGRESSION_SKIP=1 由该脚本降级为告警（有意降级须 owner Review）。
|
||
|
||
退出码：0 通过；1 门禁拒绝（PR_CHECK_FAILED）；2 用法；6 远端无法核验（REMOTE_UNVERIFIED，fail closed）。
|
||
仅使用标准库，复用 task_triage 的 Gitea 客户端与 token 读取。token 不打印、不写入任何产物。
|
||
"""
|
||
import argparse
|
||
import os
|
||
import re
|
||
import subprocess
|
||
import sys
|
||
import urllib.parse
|
||
from pathlib import Path
|
||
|
||
from task_triage import (EXIT_REMOTE, Stop, api_base, as_dict, card_branch, card_status, http_get_json,
|
||
open_pulls, read_token, repo_slug, target_branch)
|
||
|
||
EXIT_FAILED = 1
|
||
EXIT_USAGE = 2
|
||
PRE_MERGE_CLOSED = "已完成（合入前闭合）"
|
||
CHILD_COLUMNS = ("子任务分支", "必需", "状态", "PR")
|
||
DONE_STATES = ("已合入", "已延期")
|
||
PIN_REGRESSION = Path(__file__).resolve().parent / "check_pin_regression.sh"
|
||
|
||
|
||
def sh(*args):
|
||
return subprocess.run(["git", *args], capture_output=True, text=True,
|
||
encoding="utf-8", errors="replace")
|
||
|
||
|
||
def standalone(target):
|
||
return {"kind": "standalone", "initiative_id": "", "root_target": target,
|
||
"immediate_target": target, "accepting": "false"}
|
||
|
||
|
||
def card_meta(text, target):
|
||
"""两级分支元数据行；无该行即旧卡，按 standalone 处理（与 task.sh 的 card_branch_kind 一致）。"""
|
||
match = re.search(r"^- 分支类型：([^\n]*)$", text, re.M)
|
||
if not match:
|
||
return standalone(target)
|
||
line = match.group(1)
|
||
|
||
def field(name, default):
|
||
found = re.search("；" + name + r"：([^；]*)", line)
|
||
return found.group(1).strip() if found else default
|
||
|
||
return {"kind": line.split("；", 1)[0].strip(), "initiative_id": field("initiative_id", ""),
|
||
"root_target": field("root_target", target),
|
||
"immediate_target": field("immediate_target", target),
|
||
"accepting": field("accepting_requirements", "false")}
|
||
|
||
|
||
def expected_base(meta):
|
||
return meta["immediate_target"] if meta["kind"] == "child" else meta["root_target"]
|
||
|
||
|
||
def matching_card(directory, branch, archived):
|
||
"""directory 下声明该分支的唯一卡片；archived=True 时只认状态以「已完成（合入前闭合）」开头的卡。"""
|
||
found = []
|
||
for path in sorted(Path(directory).glob("*.md")):
|
||
text = path.read_text(encoding="utf-8", errors="replace")
|
||
if card_branch(text) != branch:
|
||
continue
|
||
if archived and not card_status(text).startswith(PRE_MERGE_CLOSED):
|
||
continue
|
||
found.append((path, text))
|
||
if len(found) > 1:
|
||
raise Stop(EXIT_FAILED, f"ambiguous task cards for {branch}: {', '.join(str(p) for p, _ in found)}")
|
||
return found[0] if found else None
|
||
|
||
|
||
def find_card(branch):
|
||
"""先看检出树的 .task/ 顶层；顶层无卡时再看 .task/archive/ 中合入前闭合（close --self）的卡，
|
||
其余归档卡不参与。返回 (路径, 文本)，无卡返回 (None, None)。"""
|
||
return (matching_card(".task", branch, False)
|
||
or matching_card(".task/archive", branch, True)
|
||
or (None, None))
|
||
|
||
|
||
def branch_meta(branch, target):
|
||
_, text = find_card(branch)
|
||
return (standalone(target) if text is None else card_meta(text, target)), text
|
||
|
||
|
||
def current_branch():
|
||
proc = sh("symbolic-ref", "--quiet", "--short", "HEAD")
|
||
if proc.returncode != 0 or not proc.stdout.strip():
|
||
raise Stop(EXIT_USAGE, "detached HEAD: check out the task branch first")
|
||
return proc.stdout.strip()
|
||
|
||
|
||
def local_head():
|
||
proc = sh("rev-parse", "--verify", "-q", "HEAD^{commit}")
|
||
if proc.returncode != 0:
|
||
raise Stop(EXIT_USAGE, "cannot resolve local HEAD")
|
||
return proc.stdout.strip()
|
||
|
||
|
||
def remote_head(branch):
|
||
proc = sh("ls-remote", "origin", f"refs/heads/{branch}")
|
||
if proc.returncode != 0:
|
||
raise Stop(EXIT_REMOTE, "git ls-remote origin failed")
|
||
for line in proc.stdout.splitlines():
|
||
sha, _, ref = line.partition("\t")
|
||
if ref == f"refs/heads/{branch}":
|
||
return sha
|
||
return None
|
||
|
||
|
||
def remote_context():
|
||
origin = sh("remote", "get-url", "origin")
|
||
if origin.returncode != 0:
|
||
raise Stop(EXIT_REMOTE, "origin remote is not configured")
|
||
slug = repo_slug(origin.stdout.strip())
|
||
token = read_token()
|
||
if not token:
|
||
raise Stop(EXIT_REMOTE, "no Gitea token: set GITEA_TOKEN or ~/.secrets/gitea ACCESS_TOKEN")
|
||
return api_base(), slug, token
|
||
|
||
|
||
def pull_request(api, slug, token, number):
|
||
owner, name = (urllib.parse.quote(part, safe="") for part in slug.split("/"))
|
||
pr = http_get_json(f"{api}/repos/{owner}/{name}/pulls/{number}", token)
|
||
if not isinstance(pr, dict):
|
||
raise Stop(EXIT_REMOTE, f"Gitea PR #{number} response is malformed")
|
||
return pr
|
||
|
||
|
||
def base_ref(pr):
|
||
return as_dict(pr.get("base")).get("ref")
|
||
|
||
|
||
def head_ref(pr):
|
||
return as_dict(pr.get("head")).get("ref")
|
||
|
||
|
||
def head_sha(pr):
|
||
return as_dict(pr.get("head")).get("sha")
|
||
|
||
|
||
def parent_pr_open(api, slug, token, parent, root):
|
||
return any(isinstance(pr, dict) and pr.get("state") == "open"
|
||
and head_ref(pr) == parent and base_ref(pr) == root
|
||
for pr in open_pulls(api, slug, token))
|
||
|
||
|
||
def pr_number(cell):
|
||
for pattern in (r"pulls/(\d+)", r"#(\d+)", r"^(\d+)$"):
|
||
found = re.search(pattern, cell.strip())
|
||
if found:
|
||
return int(found.group(1))
|
||
return None
|
||
|
||
|
||
def child_registry(text):
|
||
"""解析 `## 子任务登记` 表（列：子任务分支 | 必需 | 状态 | PR）；缺失即拒绝（fail closed）。"""
|
||
lines = text.splitlines()
|
||
starts = [i for i, line in enumerate(lines) if line.strip() == "## 子任务登记"]
|
||
if not starts:
|
||
raise Stop(EXIT_FAILED, "initiative card has no ## 子任务登记 section")
|
||
rows, columns = [], None
|
||
for line in lines[starts[0] + 1:]:
|
||
if line.startswith("## "):
|
||
break
|
||
if not line.strip().startswith("|"):
|
||
continue
|
||
cells = [cell.strip() for cell in line.strip().strip("|").split("|")]
|
||
if columns is None:
|
||
if not all(name in cells for name in CHILD_COLUMNS):
|
||
raise Stop(EXIT_FAILED, "子任务登记 table must have columns 子任务分支 | 必需 | 状态 | PR")
|
||
columns = {name: cells.index(name) for name in CHILD_COLUMNS}
|
||
continue
|
||
if all(re.fullmatch(r":?-{3,}:?", cell) for cell in cells):
|
||
continue
|
||
if len(cells) <= max(columns.values()):
|
||
raise Stop(EXIT_FAILED, f"malformed 子任务登记 row: {line.strip()}")
|
||
rows.append({name: cells[index] for name, index in columns.items()})
|
||
if columns is None:
|
||
raise Stop(EXIT_FAILED, "子任务登记 section has no table")
|
||
return rows
|
||
|
||
|
||
def final_problems(api, slug, token, branch, meta, text):
|
||
problems = []
|
||
if meta["accepting"] != "false":
|
||
problems.append(f"initiative {branch} still accepts requirements "
|
||
f"(accepting_requirements={meta['accepting']}); set false before --final")
|
||
for row in child_registry(text):
|
||
name, required, status, pr_cell = row["子任务分支"], row["必需"], row["状态"], row["PR"]
|
||
if required == "否":
|
||
continue
|
||
if required != "是":
|
||
problems.append(f"child {name} has invalid 必需 value {required!r}")
|
||
elif status == "已合入":
|
||
number = pr_number(pr_cell)
|
||
if number is None:
|
||
problems.append(f"child {name} is 已合入 but PR cell {pr_cell!r} names no PR")
|
||
continue
|
||
pr = pull_request(api, slug, token, number)
|
||
if pr.get("merged") is not True or base_ref(pr) != branch:
|
||
problems.append(f"child {name} PR #{number} is not merged into {branch}")
|
||
elif status not in DONE_STATES:
|
||
problems.append(f"required child {name} is {status!r}; must be 已合入 or 已延期")
|
||
return problems
|
||
|
||
|
||
def pin_regression_problems():
|
||
proc = subprocess.run(["bash", str(PIN_REGRESSION)], capture_output=True, text=True,
|
||
encoding="utf-8", errors="replace")
|
||
for line in (proc.stdout + proc.stderr).splitlines():
|
||
print(f"[pin-regression] {line}")
|
||
if proc.returncode != 0:
|
||
return [f"pin regression gate failed (exit {proc.returncode}): a gitlink moves backward against "
|
||
"origin/main or cannot be verified; PIN_REGRESSION_SKIP=1 only for owner-reviewed downgrades"]
|
||
return []
|
||
|
||
|
||
def verdict(problems, kind, base, head):
|
||
if problems:
|
||
for problem in problems:
|
||
print(f"[pr-check] FAIL {problem}")
|
||
print("[verdict] PR_CHECK_FAILED")
|
||
return EXIT_FAILED
|
||
print(f"[verdict] PR_CHECK_OK base={base} head={head} kind={kind}")
|
||
return 0
|
||
|
||
|
||
def cmd_local(args):
|
||
if args.pr is not None and args.pr < 1:
|
||
raise Stop(EXIT_USAGE, "--pr must be a positive PR number")
|
||
branch = current_branch()
|
||
head = local_head()
|
||
meta, text = branch_meta(branch, target_branch())
|
||
kind, expected = meta["kind"], expected_base(meta)
|
||
api, slug, token = remote_context()
|
||
if args.pr is not None:
|
||
pr = pull_request(api, slug, token, args.pr)
|
||
else:
|
||
pulls = [pr for pr in open_pulls(api, slug, token) if isinstance(pr, dict) and head_ref(pr) == branch]
|
||
if not pulls:
|
||
raise Stop(EXIT_FAILED, f"no open PR has head {branch}")
|
||
if len(pulls) > 1:
|
||
raise Stop(EXIT_FAILED, f"{len(pulls)} open PRs have head {branch}; pass --pr N")
|
||
pr = pulls[0]
|
||
number = pr.get("number")
|
||
problems = []
|
||
if pr.get("merged") is True:
|
||
problems.append(f"PR #{number} is already merged")
|
||
elif pr.get("state") != "open":
|
||
problems.append(f"PR #{number} is not open (state={pr.get('state')})")
|
||
if base_ref(pr) != expected:
|
||
problems.append(f"PR #{number} base is {base_ref(pr)!r}; expected {expected!r} for {kind} branch {branch}")
|
||
if head_sha(pr) != head:
|
||
problems.append(f"PR #{number} head {head_sha(pr) or '(none)'} != local HEAD {head}")
|
||
remote = remote_head(branch)
|
||
if remote is None:
|
||
problems.append(f"origin has no branch {branch}")
|
||
elif remote != head:
|
||
problems.append(f"origin/{branch} is {remote}, local HEAD is {head}")
|
||
if remote_head(expected) is None:
|
||
problems.append(f"base branch {expected} is absent on origin")
|
||
if kind == "child" and not parent_pr_open(api, slug, token, expected, meta["root_target"]):
|
||
problems.append(f"parent initiative {expected} has no open PR into {meta['root_target']}")
|
||
if args.final and kind == "initiative":
|
||
problems.extend(final_problems(api, slug, token, branch, meta, text))
|
||
problems.extend(pin_regression_problems())
|
||
return verdict(problems, kind, expected, head)
|
||
|
||
|
||
def cmd_ci(_args):
|
||
event = os.environ.get("GITHUB_EVENT_NAME", "")
|
||
if event != "pull_request":
|
||
print(f"[pr-check] skip: event {event or '(none)'} is not pull_request")
|
||
return 0
|
||
base = os.environ.get("GITHUB_BASE_REF", "").strip()
|
||
head = os.environ.get("GITHUB_HEAD_REF", "").strip()
|
||
if not base or not head:
|
||
raise Stop(EXIT_FAILED, "pull_request event lacks GITHUB_BASE_REF or GITHUB_HEAD_REF")
|
||
meta, _ = branch_meta(head, target_branch())
|
||
expected = expected_base(meta)
|
||
if base != expected:
|
||
raise Stop(EXIT_FAILED, f"PR base {base!r} differs from expected {expected!r} "
|
||
f"for {meta['kind']} branch {head}")
|
||
return verdict(pin_regression_problems(), meta["kind"], base, head)
|
||
|
||
|
||
def main(argv=None):
|
||
parser = argparse.ArgumentParser(prog="task_pr_check.py", description="PR base/head 门禁（两级分支模型）")
|
||
sub = parser.add_subparsers(dest="cmd", required=True)
|
||
local = sub.add_parser("local")
|
||
local.add_argument("--pr", type=int)
|
||
local.add_argument("--final", action="store_true")
|
||
sub.add_parser("ci")
|
||
args = parser.parse_args(argv)
|
||
try:
|
||
return cmd_local(args) if args.cmd == "local" else cmd_ci(args)
|
||
except Stop as stop:
|
||
print(f"[pr-check] FAIL {stop.message}")
|
||
print("[verdict] REMOTE_UNVERIFIED" if stop.code == EXIT_REMOTE else "[verdict] PR_CHECK_FAILED")
|
||
return stop.code
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|