agent-exec transport db893a08b5594e8086689933bcd883a5/3bc5f8142666

.gitea/workflows/ci.yaml

@@ -1,7 +1,6 @@
 name: ci
 on:
   pull_request:
-    branches: [main]
   push:
     branches: [main]
 
@@ -156,6 +155,8 @@ jobs:
           print("Checkout complete; credentials were not persisted in Git configuration")
           PY_CHECKOUT
           unset TOKEN SUBMODULE_TOKEN
+      - name: PR base gate
+        run: python3 scripts/task_pr_check.py ci
       - name: Install Go 1.27.1
         run: |
           set -euo pipefail

Makefile

@@ -13,7 +13,7 @@ PLUGINS    := plugins
 COMBO      := core=$(shell git -C $(CORE) rev-parse --short HEAD 2>/dev/null)+backend=$(shell git -C $(BACKEND) rev-parse --short HEAD 2>/dev/null)+frontend=$(shell git -C $(FRONTEND) rev-parse --short HEAD 2>/dev/null)+ui=$(shell git -C $(UI) rev-parse --short HEAD 2>/dev/null)+agentcmd=$(shell git -C deps/agent-command rev-parse --short HEAD 2>/dev/null)+documents=$(shell git -C modules/documents rev-parse --short HEAD 2>/dev/null)+native=$(shell git -C deps/agent-native rev-parse --short HEAD 2>/dev/null)+squad=$(shell git -C deps/agent-squad rev-parse --short HEAD 2>/dev/null)+intent=$(shell git -C deps/agent-intent rev-parse --short HEAD 2>/dev/null)
 VERSION    := $(shell git describe --tags --always --dirty 2>/dev/null)+$(COMBO)
 
-.PHONY: help sync status build frontend svc svc-list svc-info svc-commits svc-history svc-update svc-recover svc-watchdog test-backend test-backend-pg test-multi-instance test-lifecycle test-frontend test-ui test-core test-agentcmd test-pins test-libs dev pin clean setup precheck task-triage task-triage-decide task-start task-resume task-sync task-close task-finish facts-verify evidence version agent-exec-validate agent-exec-apply agent-exec-admit
+.PHONY: help sync status build frontend svc svc-list svc-info svc-commits svc-history svc-update svc-recover svc-watchdog test-backend test-backend-pg test-multi-instance test-lifecycle test-frontend test-ui test-core test-agentcmd test-pins test-libs dev pin clean setup precheck task-triage task-triage-decide task-start task-resume task-sync task-close task-finish task-pr-check facts-verify evidence version agent-exec-validate agent-exec-apply agent-exec-admit
 
 version:
 	@echo "agent-system $(VERSION)"
@@ -60,6 +60,10 @@ task-close:
 task-finish:
 	@bash $(TASK) finish $(ABANDON)
 
+# PR base/head 门禁（两级分支）：本地分支卡片期望目标 vs Gitea PR 与 origin；FINAL=1 用于 initiative 收口。
+task-pr-check:
+	python3 scripts/task_pr_check.py local $(if $(PR),--pr $(PR)) $(if $(FINAL),--final)
+
 # 证据写入：执行命令并把结果行追加进当前任务卡（AGENTS.md「证据与验证门禁」）。
 # 本地执行只记 SELF；RECEIPT 捕获启动收据，IMPORT_RECEIPT 导入原始收据而不重跑。
 # 用法：CMD="<单行命令>" make evidence [TAIL=15]；退出码随被执行命令。
@@ -93,6 +97,7 @@ help:
 	@echo "make task-triage REQ='<需求>' TRIAGE_RECEIPT=<path>  以远端为准发现接纳中的 initiative 并写分流收据（候选存在时 exit 5，待决策）"
 	@echo "make task-triage-decide TRIAGE_RECEIPT=<path> VERDICT=JOIN|NEW [INITIATIVE=<branch>] REASON='<理由>'  记录显式分流决策（不按数量自动 JOIN）"
 	@echo "make task-start NAME=feature/<slug> TRIAGE_RECEIPT=<path> [KIND=...] [INITIATIVE=...]  建任务前核验分流收据（失败不建分支）"
+	@echo "make task-pr-check [PR=N] [FINAL=1]  核验当前任务分支 PR 的 base/head 门禁（远端为准；FINAL=1 用于 initiative 收口）"
 
 agent-exec-validate:
 	PYTHONDONTWRITEBYTECODE=1 python3 scripts/agent_exec.py validate --repo .

scripts/task_pr_check.py

@@ -0,0 +1,291 @@
+#!/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/ 顶层取卡片，核对 GITHUB_BASE_REF。
+      不联网、不需要 token；非 pull_request 事件跳过（退出 0）。
+
+期望目标：child → immediate_target；其余（initiative / standalone / 无卡）→ root_target（无卡即 main）。
+
+退出码：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, http_get_json,
+                         open_pulls, read_token, repo_slug, target_branch)
+
+EXIT_FAILED = 1
+EXIT_USAGE = 2
+CHILD_COLUMNS = ("子任务分支", "必需", "状态", "PR")
+DONE_STATES = ("已合入", "已延期")
+
+
+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 find_card(branch):
+    """只看检出树的 .task/ 顶层（archive 不参与）；返回 (路径, 文本)，无卡返回 (None, None)。"""
+    found = []
+    for path in sorted(Path(".task").glob("*.md")):
+        text = path.read_text(encoding="utf-8", errors="replace")
+        if card_branch(text) == branch:
+            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, 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 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))
+    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([], 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())

tests/test_task_pr_check.py

@@ -0,0 +1,401 @@
+"""PR base/head 门禁（scripts/task_pr_check.py）：临时真实 Git 仓库（bare origin + clone）+ 进程内 Gitea 桩。
+
+全部离线运行：不联网、不读取真实 ~/.secrets（HOME 指向临时目录），token 为合成值。
+"""
+import http.server
+import json
+import os
+import socket
+import subprocess
+import sys
+import tempfile
+import threading
+import unittest
+import urllib.parse
+from pathlib import Path
+
+SOURCE = Path(__file__).resolve().parents[1]
+SCRIPT = SOURCE / "scripts" / "task_pr_check.py"
+
+TOKEN = "synthetic-gitea-token-not-real"
+
+
+def card_text(branch, kind, root="main", immediate="main", accepting="false", registry=None):
+    slug = branch.replace("/", "-")
+    lines = [
+        f"# {branch} 任务卡",
+        "任务主记录：fixture；状态：进行中；owner：PR Check Test",
+        "- 权威仓：fixture；目标分支：main；R0（initial_base_sha）= 0",
+        f"- 工作区：fixture（worktree）；分支：{branch}",
+        f"- 分支类型：{kind}；initiative_id：{slug}；root_target：{root}；"
+        f"immediate_target：{immediate}；accepting_requirements：{accepting}",
+        "",
+        "## 目标与非目标",
+        "",
+        "- 交付 PR 门禁",
+        "",
+    ]
+    if registry is not None:
+        lines += ["## 子任务登记", "", "| 子任务分支 | 必需 | 状态 | PR |", "| --- | --- | --- | --- |"]
+        lines += [f"| {branch_name} | {required} | {status} | {pr} |"
+                  for branch_name, required, status, pr in registry]
+        lines.append("")
+    return "\n".join(lines) + "\n"
+
+
+def pr(number, branch, base, sha, state="open", merged=False):
+    return {"number": number, "state": state, "merged": merged,
+            "base": {"ref": base}, "head": {"ref": branch, "sha": sha}}
+
+
+class FakeGitea:
+    """Gitea pulls API 桩（127.0.0.1）：校验 token，列表（分页+state）与单 PR 查询；从不联网。"""
+
+    PAGE = 50
+
+    def __init__(self, token=TOKEN):
+        self.token = token
+        self.prs = []
+        stub = self
+
+        class Handler(http.server.BaseHTTPRequestHandler):
+            def log_message(self, *args):
+                pass
+
+            def _send(self, status, obj):
+                body = json.dumps(obj).encode()
+                self.send_response(status)
+                self.send_header("Content-Type", "application/json")
+                self.send_header("Content-Length", str(len(body)))
+                self.end_headers()
+                self.wfile.write(body)
+
+            def do_GET(self):
+                if self.headers.get("Authorization") != f"token {stub.token}":
+                    return self._send(401, {"message": "unauthorized"})
+                parsed = urllib.parse.urlsplit(self.path)
+                query = urllib.parse.parse_qs(parsed.query)
+                parts = parsed.path.split("/")
+                if parts[1:4] != ["api", "v1", "repos"] or len(parts) not in (7, 8) or parts[6] != "pulls":
+                    return self._send(404, {"message": "not found"})
+                if len(parts) == 8:
+                    found = [item for item in stub.prs if str(item["number"]) == parts[7]]
+                    return self._send(200, found[0]) if found else self._send(404, {"message": "not found"})
+                state = query.get("state", [""])[0]
+                items = [item for item in stub.prs if not state or item["state"] == state]
+                page = int(query.get("page", ["1"])[0])
+                start = (page - 1) * stub.PAGE
+                return self._send(200, items[start:start + stub.PAGE])
+
+        self.server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+        self.url = f"http://127.0.0.1:{self.server.server_address[1]}/api/v1"
+        threading.Thread(target=self.server.serve_forever, daemon=True).start()
+
+    def close(self):
+        self.server.shutdown()
+        self.server.server_close()
+
+
+def closed_port_url():
+    with socket.socket() as sock:
+        sock.bind(("127.0.0.1", 0))
+        port = sock.getsockname()[1]
+    return f"http://127.0.0.1:{port}/api/v1"
+
+
+class PrCheckTests(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory(prefix="pr-check-")
+        self.addCleanup(self.tmp.cleanup)
+        root = Path(self.tmp.name)
+        self.remote = root / "remote.git"
+        self.repo = root / "repo"
+        self.env = {k: v for k, v in os.environ.items()
+                    if not k.startswith(("GITHUB_", "GIT_")) and k not in {
+                        "GITEA_API_URL", "GITEA_REPO", "GITEA_TOKEN", "TASK_TARGET_BRANCH"}}
+        self.env.update({"GIT_CONFIG_NOSYSTEM": "1", "GIT_CONFIG_GLOBAL": "/dev/null",
+                         "GIT_ALLOW_PROTOCOL": "file", "GIT_TERMINAL_PROMPT": "0",
+                         "GIT_AUTHOR_NAME": "PR Check Test", "GIT_AUTHOR_EMAIL": "pr-check@example.invalid",
+                         "GIT_COMMITTER_NAME": "PR Check Test", "GIT_COMMITTER_EMAIL": "pr-check@example.invalid",
+                         "GIT_CONFIG_COUNT": "2", "GIT_CONFIG_KEY_0": "gc.auto",
+                         "GIT_CONFIG_VALUE_0": "0", "GIT_CONFIG_KEY_1": "maintenance.auto",
+                         "GIT_CONFIG_VALUE_1": "false",
+                         "HOME": str(root / "no-home"),
+                         "NO_PROXY": "127.0.0.1,localhost", "no_proxy": "127.0.0.1,localhost",
+                         "GITEA_REPO": "fixture/repo", "GITEA_TOKEN": TOKEN})
+        self.git(root, "init", "--bare", "--initial-branch=main", str(self.remote))
+        self.git(root, "init", "--initial-branch=main", str(self.repo))
+        (self.repo / "README").write_text("fixture\n")
+        self.git(self.repo, "add", "README")
+        self.git(self.repo, "commit", "-m", "Create fixture")
+        self.git(self.repo, "remote", "add", "origin", str(self.remote))
+        self.git(self.repo, "push", "-u", "origin", "main")
+        self.gitea = FakeGitea()
+        self.addCleanup(self.gitea.close)
+        self.env["GITEA_API_URL"] = self.gitea.url
+
+    # ---- 夹具 ----
+
+    def git(self, cwd, *args):
+        return subprocess.run(["git", *args], cwd=cwd, env=self.env, capture_output=True,
+                              text=True, check=True).stdout.strip()
+
+    def head(self, ref="HEAD"):
+        return self.git(self.repo, "rev-parse", ref)
+
+    def make_branch(self, branch, base="main"):
+        self.git(self.repo, "switch", "-c", branch, base)
+        self.git(self.repo, "push", "-u", "origin", branch)
+        return self.head()
+
+    def commit_card(self, branch, text, base="main"):
+        self.make_branch(branch, base)
+        path = self.repo / ".task" / (branch.replace("/", "-") + ".md")
+        path.parent.mkdir(exist_ok=True)
+        path.write_text(text, encoding="utf-8")
+        self.git(self.repo, "add", ".task")
+        self.git(self.repo, "commit", "-m", f"Record {branch}")
+        self.git(self.repo, "push", "origin", branch)
+        return self.head()
+
+    def set_card(self, branch, text):
+        self.git(self.repo, "switch", branch)
+        path = self.repo / ".task" / (branch.replace("/", "-") + ".md")
+        path.write_text(text, encoding="utf-8")
+        self.git(self.repo, "add", ".task")
+        self.git(self.repo, "commit", "-m", f"Update {branch}")
+        self.git(self.repo, "push", "origin", branch)
+        return self.head()
+
+    def commit_work(self, branch, push=True):
+        self.git(self.repo, "switch", branch)
+        count = len(list(self.repo.glob("work-*.txt")))
+        (self.repo / f"work-{count}.txt").write_text("more work\n")
+        self.git(self.repo, "add", ".")
+        self.git(self.repo, "commit", "-m", "More work")
+        if push:
+            self.git(self.repo, "push", "origin", "HEAD")
+        return self.head()
+
+    def build_child(self):
+        self.commit_card("feature/init", card_text("feature/init", "initiative", accepting="true"))
+        self.commit_card("feature/init-child",
+                         card_text("feature/init-child", "child", root="main", immediate="feature/init"),
+                         base="feature/init")
+        return self.head()
+
+    def invoke(self, *args, env=None, drop=()):
+        merged = {k: v for k, v in (self.env | (env or {})).items() if k not in drop}
+        result = subprocess.run([sys.executable, str(SCRIPT), *args], cwd=self.repo, env=merged,
+                                capture_output=True, text=True, timeout=60)
+        result.out = result.stdout + result.stderr
+        self.assertNotIn(TOKEN, result.out)
+        return result
+
+    def local(self, *args, env=None, drop=()):
+        return self.invoke("local", *args, env=env, drop=drop)
+
+    @staticmethod
+    def fails(result):
+        return [line for line in result.stdout.splitlines() if line.startswith("[pr-check] FAIL")]
+
+    # ---- local：基址与分支类型 ----
+
+    def test_child_pr_into_main_is_rejected(self):
+        self.build_child()
+        self.gitea.prs = [pr(1, "feature/init", "main", self.head("feature/init")),
+                          pr(2, "feature/init-child", "main", self.head())]
+        result = self.local()
+        self.assertEqual(result.returncode, 1, result.out)
+        self.assertEqual(len(self.fails(result)), 1, result.out)
+        self.assertIn("expected 'feature/init' for child branch feature/init-child", result.out)
+        self.assertIn("[verdict] PR_CHECK_FAILED", result.out)
+
+    def test_child_pr_into_initiative_passes(self):
+        self.build_child()
+        head = self.head()
+        self.gitea.prs = [pr(1, "feature/init", "main", self.head("feature/init")),
+                          pr(2, "feature/init-child", "feature/init", head)]
+        result = self.local()
+        self.assertEqual(result.returncode, 0, result.out)
+        self.assertIn(f"[verdict] PR_CHECK_OK base=feature/init head={head} kind=child", result.out)
+
+    def test_child_is_rejected_when_parent_initiative_pr_is_closed(self):
+        self.build_child()
+        self.gitea.prs = [pr(1, "feature/init", "main", self.head("feature/init"), state="closed"),
+                          pr(2, "feature/init-child", "feature/init", self.head())]
+        result = self.local()
+        self.assertEqual(result.returncode, 1, result.out)
+        self.assertEqual(len(self.fails(result)), 1, result.out)
+        self.assertIn("parent initiative feature/init has no open PR into main", result.out)
+
+    def test_standalone_and_initiative_base_gate(self):
+        self.make_branch("feature/other")
+        self.commit_card("feature/solo", card_text("feature/solo", "standalone"))
+        self.commit_card("feature/init-x", card_text("feature/init-x", "initiative", accepting="true"))
+        for branch, kind in (("feature/solo", "standalone"), ("feature/init-x", "initiative")):
+            self.git(self.repo, "switch", branch)
+            head = self.head()
+            for base, expected_ok in (("main", True), ("feature/other", False)):
+                with self.subTest(kind=kind, base=base):
+                    self.gitea.prs = [pr(1, branch, base, head)]
+                    result = self.local()
+                    if expected_ok:
+                        self.assertEqual(result.returncode, 0, result.out)
+                        self.assertIn(f"[verdict] PR_CHECK_OK base=main head={head} kind={kind}", result.out)
+                    else:
+                        self.assertEqual(result.returncode, 1, result.out)
+                        self.assertIn(f"expected 'main' for {kind} branch {branch}", result.out)
+
+    # ---- local：PR 状态与 head ----
+
+    def test_stale_pr_head_is_rejected(self):
+        self.commit_card("feature/solo", card_text("feature/solo", "standalone"))
+        stale = self.head()
+        self.gitea.prs = [pr(1, "feature/solo", "main", stale)]
+        fresh = self.commit_work("feature/solo")
+        result = self.local()
+        self.assertEqual(result.returncode, 1, result.out)
+        self.assertEqual(len(self.fails(result)), 1, result.out)
+        self.assertIn(f"PR #1 head {stale} != local HEAD {fresh}", result.out)
+
+    def test_remote_branch_head_differing_from_local_is_rejected(self):
+        self.commit_card("feature/solo", card_text("feature/solo", "standalone"))
+        remote = self.head()
+        local = self.commit_work("feature/solo", push=False)
+        self.gitea.prs = [pr(1, "feature/solo", "main", local)]
+        result = self.local()
+        self.assertEqual(result.returncode, 1, result.out)
+        self.assertEqual(len(self.fails(result)), 1, result.out)
+        self.assertIn(f"origin/feature/solo is {remote}, local HEAD is {local}", result.out)
+
+    def test_closed_and_merged_prs_are_rejected(self):
+        self.commit_card("feature/solo", card_text("feature/solo", "standalone"))
+        head = self.head()
+        for kwargs, reason in (({"state": "closed"}, "PR #1 is not open (state=closed)"),
+                               ({"state": "closed", "merged": True}, "PR #1 is already merged")):
+            with self.subTest(reason=reason):
+                self.gitea.prs = [pr(1, "feature/solo", "main", head, **kwargs)]
+                result = self.local("--pr", "1")
+                self.assertEqual(result.returncode, 1, result.out)
+                self.assertIn(reason, result.out)
+
+    def test_pr_lookup_requires_one_open_pr_or_explicit_number(self):
+        self.commit_card("feature/solo", card_text("feature/solo", "standalone"))
+        head = self.head()
+        self.make_branch("feature/other")
+        self.git(self.repo, "switch", "feature/solo")
+        self.gitea.prs = [pr(1, "feature/solo", "feature/other", head), pr(2, "feature/solo", "main", head)]
+        ambiguous = self.local()
+        self.assertEqual(ambiguous.returncode, 1, ambiguous.out)
+        self.assertIn("2 open PRs have head feature/solo; pass --pr N", ambiguous.out)
+        explicit = self.local("--pr", "2")
+        self.assertEqual(explicit.returncode, 0, explicit.out)
+        self.gitea.prs = []
+        missing = self.local()
+        self.assertEqual(missing.returncode, 1, missing.out)
+        self.assertIn("no open PR has head feature/solo", missing.out)
+
+    # ---- 远端不可核验：fail closed ----
+
+    def test_unreachable_or_unauthorized_api_fails_closed_with_exit_6(self):
+        self.build_child()
+        self.gitea.prs = [pr(1, "feature/init", "main", self.head("feature/init")),
+                          pr(2, "feature/init-child", "feature/init", self.head())]
+        for label, env in (("unreachable", {"GITEA_API_URL": closed_port_url()}),
+                           ("unauthorized", {"GITEA_TOKEN": "wrong-token"})):
+            with self.subTest(label):
+                result = self.local(env=env)
+                self.assertEqual(result.returncode, 6, result.out)
+                self.assertIn("[verdict] REMOTE_UNVERIFIED", result.out)
+
+    def test_token_is_first_access_token_line_of_secrets_file(self):
+        self.commit_card("feature/solo", card_text("feature/solo", "standalone"))
+        home = Path(self.tmp.name) / "secrets-home"
+        (home / ".secrets").mkdir(parents=True)
+        (home / ".secrets" / "gitea").write_text(
+            f"URL_PREFIX=https://example.invalid\n- ACCESS_TOKEN={TOKEN}\nACCESS_TOKEN=other-token\n")
+        self.gitea.prs = [pr(1, "feature/solo", "main", self.head())]
+        result = self.local(env={"HOME": str(home)}, drop=("GITEA_TOKEN",))
+        self.assertEqual(result.returncode, 0, result.out)
+
+    # ---- local --final ----
+
+    def test_final_rejects_initiative_still_accepting_requirements(self):
+        self.commit_card("feature/init", card_text("feature/init", "initiative", accepting="true", registry=[]))
+        self.gitea.prs = [pr(1, "feature/init", "main", self.head())]
+        result = self.local("--final")
+        self.assertEqual(result.returncode, 1, result.out)
+        self.assertIn("still accepts requirements", result.out)
+
+    def test_final_requires_required_children_merged_or_deferred(self):
+        self.commit_card("feature/init", card_text("feature/init", "initiative", accepting="true"))
+        scenarios = [
+            ("merged and deferred pass",
+             [("feature/init-a", "是", "已合入", "#2"), ("feature/init-b", "否", "进行中", "-"),
+              ("feature/init-c", "是", "已延期", "-")],
+             [pr(2, "feature/init-a", "feature/init", "a" * 40, state="closed", merged=True)],
+             None),
+            ("required child still in progress",
+             [("feature/init-d", "是", "进行中", "-")], [], "required child feature/init-d is '进行中'"),
+            ("open PR is not merged",
+             [("feature/init-e", "是", "已合入", "#3")],
+             [pr(3, "feature/init-e", "feature/init", "b" * 40, state="open")],
+             "child feature/init-e PR #3 is not merged into feature/init"),
+            ("merged into main instead",
+             [("feature/init-f", "是", "已合入", "#4")],
+             [pr(4, "feature/init-f", "main", "c" * 40, state="closed", merged=True)],
+             "child feature/init-f PR #4 is not merged into feature/init"),
+        ]
+        for label, registry, children, reason in scenarios:
+            with self.subTest(label):
+                head = self.set_card("feature/init", card_text("feature/init", "initiative",
+                                                               accepting="false", registry=registry))
+                self.gitea.prs = [pr(1, "feature/init", "main", head)] + children
+                result = self.local("--final")
+                if reason is None:
+                    self.assertEqual(result.returncode, 0, result.out)
+                    self.assertIn("[verdict] PR_CHECK_OK base=main", result.out)
+                else:
+                    self.assertEqual(result.returncode, 1, result.out)
+                    self.assertIn(reason, result.out)
+
+    # ---- ci 模式 ----
+
+    def ci(self, base=None, head=None, event="pull_request"):
+        env = {"GITHUB_EVENT_NAME": event}
+        if base is not None:
+            env["GITHUB_BASE_REF"] = base
+        if head is not None:
+            env["GITHUB_HEAD_REF"] = head
+        # ci 模式不得需要网络或 token：指向关闭端口并移除 token，证明零依赖。
+        return self.invoke("ci", env=env | {"GITEA_API_URL": closed_port_url()}, drop=("GITEA_TOKEN",))
+
+    def test_ci_mode_gates_pull_request_base_against_head_card(self):
+        self.build_child()
+        result = self.ci(base="feature/init", head="feature/no-card")
+        self.assertEqual(result.returncode, 1, result.out)  # 无卡 → standalone → 期望 main
+        self.assertIn("expected 'main' for standalone branch feature/no-card", result.out)
+        cases = [
+            ("skips non pull_request events", dict(event="push", base="main", head="feature/init-child"), 0,
+             "[pr-check] skip: event push is not pull_request"),
+            ("child into main is rejected", dict(base="main", head="feature/init-child"), 1,
+             "expected 'feature/init' for child branch feature/init-child"),
+            ("child into initiative passes", dict(base="feature/init", head="feature/init-child"), 0,
+             "[verdict] PR_CHECK_OK base=feature/init head=feature/init-child kind=child"),
+            ("card-less head defaults to main", dict(base="main", head="feature/no-card"), 0,
+             "[verdict] PR_CHECK_OK base=main head=feature/no-card kind=standalone"),
+            ("card-less head into other base is rejected", dict(base="feature/init", head="feature/no-card"), 1,
+             "expected 'main' for standalone branch feature/no-card"),
+            ("missing base ref is rejected", dict(base=None, head="feature/init-child"), 1,
+             "lacks GITHUB_BASE_REF or GITHUB_HEAD_REF"),
+        ]
+        for label, kwargs, code, text in cases:
+            with self.subTest(label):
+                result = self.ci(**kwargs)
+                self.assertEqual(result.returncode, code, result.out)
+                self.assertIn(text, result.out)
+
+
+if __name__ == "__main__":
+    unittest.main()