7e58623cb3
實查(2026-08-27):inkstone org 有 7 個 open PR,最久的三個躺了兩星期 (Arcrun#104 15 天、Arcrun#116 14 天、arcrun-rag#91 14 天), 而 v0.9.0 的 46 支閘一支都沒有在管 PR。票看起來「已交付」, 東西卻沒進 main ⇒ 沒進版本 ⇒ leo 手上永遠不會出現它。 新增 hooks/pr-verdict-guard.sh(Stop,F 組) open、沒有人被指派、也沒有 REQUEST_CHANGES review 的 PR ⇒ 擋一次並點名 (含開了幾天與網址);merge 了但 head 分支還留著的一起點名 ——票上寫死「merge → 併完當場刪掉那條 branch」,分支還在=結論只給了一半。 🔴 票上點名要避開的錯路(ISEP v0.6.0 divergence §B4): 「佇列非空即 block」會讓總管永遠停不下來。所以擋的是遺忘,不是等待—— ① 有人被指派/掛 Human ⇒ 不點名(棒子在他手上,是 Gitea 原生欄位不是措辭) ② 有 REQUEST_CHANGES review ⇒ 結論給過了,球在對方腳下 ③ 這一回合碰過它 ⇒ 不點名 ④ 響過就退讓:每個 PR 各自 1→4→8→16 「碰過」不讀任何一句話:① updated_at 跟上次收工比變了(外部系統的事實) ② 上次收工時還不存在(剛開出來,不在誕生的回合開罵) ③ 這回合的 tool call 出現它的識別碼(接住「結論寫在票上」這種合法結論)。 ③ 不是關鍵字黑名單——黑名單拿措辭來「擋」,這裡拿識別碼來「放行」, 比對錯了只會少擋一次。leo 2026-08-17 證偽的是前者。 刻意只掛 Stop 不掛 SubagentStop:給 PR 結論是總管的權,subagent 連 main 都不准推。 新增 scripts/pr-verdict —— merge/reject/changes 各是一個動作 merge 併 + 刪分支 + **當場複驗分支真的不見了**(下了刪除指令 ≠ 它被刪掉) reject 理由寫進票 + 關 PR + 刪分支 changes 修改要求寫進票 + 留 REQUEST_CHANGES(讓「給過結論」變成閘查得到的事實) list / --dry-run(按下去之前先看會打哪幾通 API,一通都不送出) 測試 hooks/tests/pr-verdict-guard.test.sh:52 條全過,離線走 fixture, 不打 Gitea、不開測試 PR、不留測試票。A 群 8 條全在證「不該擋」。 另用真實 Gitea 唯讀重演過票上那三個真跡(一次點名 7 個)。 順手:plugin.json 的「27 支腳本」是過期數字,實數 34 (ls -p scripts | grep -v / | wc -l)——同一個病,只是換一欄。 版本 0.9.0 → 0.10.0。tag 待併入 main 後由總管打(release-tag-guard 管那道)。 Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016ZBu4Sa1cGntKFRBYNZ6xs
258 lines
11 KiB
Python
Executable File
258 lines
11 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""pr-verdict — 給 PR 一個結論,一個動作把該做的全做完(inkstone/ISEP#81)。
|
||
|
||
leo 的問題(票上原文):「我交出去的 PR 沒有人去看它⋯⋯票看起來是『已交付』,
|
||
但東西沒有進 main,也就沒有進版本,**leo 手上永遠不會出現它**。」
|
||
|
||
「有結論」只有三種,缺一不可:
|
||
|
||
pr-verdict merge <owner/repo#N> 併 + **當場刪掉那條 branch**
|
||
pr-verdict reject <owner/repo#N> --ticket <票> -F <檔> 退回:理由寫**票上** + 關 PR
|
||
pr-verdict changes <owner/repo#N> --ticket <票> -F <檔> 要求修改:寫**票上** + REQUEST_CHANGES
|
||
pr-verdict list 撈一次:誰還沒有結論
|
||
|
||
🔴 為什麼要有這支、而不是叫人記得多做一步:
|
||
「merge 完要刪 branch」「退回理由要寫在票上」都是**第二個動作**,
|
||
而所有掉棒的形狀都長一樣——第一個動作做了,第二個沒有。
|
||
把兩件事綁成一個指令,就沒有「忘了做第二件」這個狀態可以存在。
|
||
|
||
🔴 `--dry-run` 會把要打的每一通 API 印出來但不送出。
|
||
加在任何子指令上都有效;**收工前想確認自己按下去會發生什麼,用它。**
|
||
"""
|
||
import json
|
||
import os
|
||
import re
|
||
import subprocess
|
||
import sys
|
||
import urllib.error
|
||
import urllib.parse
|
||
import urllib.request
|
||
|
||
HOST = os.environ.get("PR_VERDICT_HOST", "https://git.uncle6.me")
|
||
DRY = False
|
||
|
||
|
||
def die(msg, code=2):
|
||
print(msg, file=sys.stderr)
|
||
sys.exit(code)
|
||
|
||
|
||
def token():
|
||
root = os.environ.get("CLAUDE_PROJECT_DIR") or os.getcwd()
|
||
host = HOST.split("//")[-1].rstrip("/")
|
||
try:
|
||
out = subprocess.run(["git", "-C", root, "remote", "-v"],
|
||
capture_output=True, text=True, timeout=20).stdout
|
||
except Exception:
|
||
out = ""
|
||
for line in out.splitlines():
|
||
if host in line:
|
||
m = re.search(r"//[^:/]+:([^@]+)@", line)
|
||
if m:
|
||
return m.group(1)
|
||
for env in ("GITEA_TOKEN_CLAUDE_CODE", "GITEA_TOKEN"):
|
||
if os.environ.get(env):
|
||
return os.environ[env]
|
||
die(f"🔴 拿不到 {host} 的 token(remote 沒帶憑證,也沒有 GITEA_TOKEN_CLAUDE_CODE)")
|
||
|
||
|
||
def api(path, payload=None, method=None, soft=False):
|
||
verb = method or ("POST" if payload is not None else "GET")
|
||
if DRY and verb != "GET":
|
||
print(f" [dry-run] {verb} {path}"
|
||
+ (f"\n {json.dumps(payload, ensure_ascii=False)}" if payload else ""))
|
||
return {}
|
||
url = path if path.startswith("http") else f"{HOST}/api/v1{path}"
|
||
data = json.dumps(payload).encode() if payload is not None else None
|
||
req = urllib.request.Request(
|
||
url, data=data, method=verb,
|
||
headers={"Authorization": f"token {token()}", "Content-Type": "application/json"})
|
||
try:
|
||
raw = urllib.request.urlopen(req, timeout=40).read()
|
||
return json.loads(raw) if raw.strip() else {}
|
||
except urllib.error.HTTPError as e:
|
||
if soft:
|
||
return {"__status__": e.code}
|
||
die(f"🔴 Gitea {e.code} on {verb} {path}:{e.read().decode()[:300]}")
|
||
|
||
|
||
def parse_ref(s):
|
||
m = re.match(r"^([\w.-]+)/([\w.-]+)#(\d+)$", s.strip())
|
||
if not m:
|
||
die(f"🔴 寫法是 owner/repo#N,你給的是:{s}")
|
||
return m.group(1), m.group(2), int(m.group(3))
|
||
|
||
|
||
def read_body(argv, what):
|
||
if "-F" not in argv:
|
||
die(f"🚫 缺 -F <檔>:{what}要**寫在票上**,不是寫在訊息裡。\n"
|
||
f" (inkstone/ISEP#81:「退回理由寫在票上(寫票,不是寫訊息)」)")
|
||
p = argv[argv.index("-F") + 1]
|
||
try:
|
||
with open(p) as f:
|
||
return f.read().strip()
|
||
except Exception as e:
|
||
die(f"🔴 讀不到 {p}:{e}")
|
||
|
||
|
||
def opt(argv, name, default=None):
|
||
return argv[argv.index(name) + 1] if name in argv else default
|
||
|
||
|
||
def say_on_ticket(ref, body):
|
||
"""把結論貼進**票**(不是貼進 PR 的討論串——PR 會隨自己沉掉,票才是那條線的家)。
|
||
|
||
內文第一行帶【總管】身份欄,對齊 reply-identity-guard 的規約
|
||
(leo 2026-08-27:「subagent 回覆時要表明身份」)。
|
||
"""
|
||
owner, repo, num = parse_ref(ref)
|
||
if DRY:
|
||
print(f" [dry-run] POST /repos/{owner}/{repo}/issues/{num}/comments")
|
||
print(f" {body[:120]}…")
|
||
return
|
||
api(f"/repos/{owner}/{repo}/issues/{num}/comments", {"body": body})
|
||
|
||
|
||
# ── merge ────────────────────────────────────────────────────────────────
|
||
def cmd_merge(argv):
|
||
if not argv:
|
||
die("用法:pr-verdict merge <owner/repo#N> [--style merge|squash|rebase] [--dry-run]")
|
||
owner, repo, num = parse_ref(argv[0])
|
||
style = opt(argv, "--style", "merge")
|
||
|
||
pr = api(f"/repos/{owner}/{repo}/pulls/{num}")
|
||
branch = ((pr.get("head") or {}).get("ref")) or ""
|
||
same_repo = ((pr.get("head") or {}).get("repo") or {}).get("full_name") == f"{owner}/{repo}"
|
||
|
||
if not pr.get("merged"):
|
||
if pr.get("state") != "open":
|
||
die(f"🔴 {owner}/{repo}#{num} 不是 open(state={pr.get('state')})——沒得 merge。")
|
||
api(f"/repos/{owner}/{repo}/pulls/{num}/merge",
|
||
{"Do": style, "delete_branch_after_merge": True})
|
||
if not DRY:
|
||
print(f"✅ {owner}/{repo}#{num} 已 merge({style})")
|
||
else:
|
||
print(f"ℹ️ {owner}/{repo}#{num} 本來就是 merged,只補刪分支")
|
||
|
||
# 🔴 第二件事:**當場複驗分支真的不見了**。
|
||
# Gitea 的 delete_branch_after_merge 在某些設定下不會生效(保護分支、
|
||
# 跨 fork),而「我下了刪除的指令」不等於「它被刪掉了」——
|
||
# 本 repo 心法:交出去之前,我知不知道它能不能用。
|
||
if not branch:
|
||
print("⚠️ 拿不到 head 分支名,無法複驗——請自己確認分支已刪。")
|
||
return
|
||
if not same_repo:
|
||
print(f"ℹ️ head 在別的 repo(fork),本指令不跨 repo 刪分支:{branch}")
|
||
return
|
||
if DRY:
|
||
print(f" [dry-run] DELETE /repos/{owner}/{repo}/branches/{branch}(若還在)")
|
||
return
|
||
if _branch_alive(owner, repo, branch):
|
||
api(f"/repos/{owner}/{repo}/branches/{urllib.parse.quote(branch, safe='')}",
|
||
method="DELETE", soft=True)
|
||
if _branch_alive(owner, repo, branch):
|
||
die(f"🔴 分支 {branch} 刪不掉(保護分支?權限?)——**結論只給了一半**,"
|
||
f"pr-verdict-guard 會繼續點名它。")
|
||
print(f"✅ 分支 {branch} 已刪除(複驗過:查不到了)")
|
||
|
||
|
||
def _branch_alive(owner, repo, br):
|
||
r = api(f"/repos/{owner}/{repo}/branches/{urllib.parse.quote(br, safe='')}", soft=True)
|
||
return isinstance(r, dict) and "name" in r
|
||
|
||
|
||
# ── reject / changes ─────────────────────────────────────────────────────
|
||
def _verdict_with_reason(argv, kind):
|
||
if not argv:
|
||
die(f"用法:pr-verdict {kind} <owner/repo#N> --ticket <owner/repo#N> -F <檔> [--dry-run]")
|
||
owner, repo, num = parse_ref(argv[0])
|
||
ticket = opt(argv, "--ticket")
|
||
if not ticket:
|
||
die("🚫 缺 --ticket <owner/repo#N>:結論要寫在**票**上。\n"
|
||
" PR 的討論串會隨 PR 一起沉掉;票才是那條線的家。")
|
||
body = read_body(argv, "退回理由" if kind == "reject" else "修改要求")
|
||
|
||
pr = api(f"/repos/{owner}/{repo}/pulls/{num}")
|
||
head = f"{owner}/{repo}#{num}"
|
||
url = (pr.get("html_url") or "")
|
||
title = "退回" if kind == "reject" else "要求修改"
|
||
say_on_ticket(ticket, f"【總管】**PR {title}**:{head} {url}\n\n{body}")
|
||
if not DRY:
|
||
print(f"✅ {title}理由已寫進票 {ticket}")
|
||
|
||
if kind == "reject":
|
||
api(f"/repos/{owner}/{repo}/issues/{num}", {"state": "closed"}, method="PATCH")
|
||
if not DRY:
|
||
print(f"✅ {head} 已關閉")
|
||
br = ((pr.get("head") or {}).get("ref")) or ""
|
||
same = ((pr.get("head") or {}).get("repo") or {}).get("full_name") == f"{owner}/{repo}"
|
||
if br and same and not DRY and _branch_alive(owner, repo, br):
|
||
api(f"/repos/{owner}/{repo}/branches/{urllib.parse.quote(br, safe='')}",
|
||
method="DELETE", soft=True)
|
||
print(f"✅ 分支 {br} 已刪除")
|
||
else:
|
||
# REQUEST_CHANGES:**讓「已經給過結論」變成 Gitea 查得到的事實**,
|
||
# 這樣 pr-verdict-guard 就不會再把它當成被遺忘的 PR(=擋等待)。
|
||
api(f"/repos/{owner}/{repo}/pulls/{num}/reviews",
|
||
{"event": "REQUEST_CHANGES", "body": f"修改要求寫在 {ticket}"})
|
||
if not DRY:
|
||
print(f"✅ {head} 已留下 REQUEST_CHANGES review(閘看得到,不會再點名它)")
|
||
if DRY:
|
||
print("\n[dry-run] 以上一通都沒有送出。")
|
||
return
|
||
print("\n📌 兩件事是同一個動作——不會只寫了理由而 PR 還開著"
|
||
"(reject),也不會留了 review 而票上查不到理由(changes)。")
|
||
|
||
|
||
# ── list ─────────────────────────────────────────────────────────────────
|
||
def cmd_list(argv):
|
||
orgs = [o.strip().lower() for o in
|
||
os.environ.get("PR_VERDICT_ORGS", "inkstone").split(",") if o.strip()]
|
||
rows = api("/repos/issues/search?type=pulls&state=open&limit=50")
|
||
import time
|
||
now = time.time()
|
||
n = 0
|
||
for pr in rows:
|
||
full = (pr.get("repository") or {}).get("full_name") or ""
|
||
if "/" not in full or (orgs and full.split("/", 1)[0].lower() not in orgs):
|
||
continue
|
||
try:
|
||
t = time.mktime(time.strptime(pr["created_at"][:19], "%Y-%m-%dT%H:%M:%S"))
|
||
age = int((now - t) / 86400)
|
||
except Exception:
|
||
age = -1
|
||
who = [a["login"] for a in (pr.get("assignees") or [])]
|
||
mark = "🕒" if who else "❌"
|
||
print(f"{mark} {full}#{pr['number']:<5} 開了 {age:>3} 天 "
|
||
f"{'指派=' + ','.join(who) if who else '沒有人被指派'} {pr['title'][:34]}")
|
||
n += 1
|
||
print(f"\n共 {n} 個 open PR(org:{'/'.join(orgs) or '全部'})。"
|
||
f" ❌=沒有人在管,pr-verdict-guard 會點名它")
|
||
|
||
|
||
def main():
|
||
global DRY
|
||
argv = sys.argv[1:]
|
||
if "--dry-run" in argv:
|
||
DRY = True
|
||
argv.remove("--dry-run")
|
||
if not argv:
|
||
print(__doc__)
|
||
sys.exit(0)
|
||
cmd, rest = argv[0], argv[1:]
|
||
if cmd == "merge":
|
||
cmd_merge(rest)
|
||
elif cmd == "reject":
|
||
_verdict_with_reason(rest, "reject")
|
||
elif cmd == "changes":
|
||
_verdict_with_reason(rest, "changes")
|
||
elif cmd == "list":
|
||
cmd_list(rest)
|
||
else:
|
||
print(__doc__)
|
||
die(f"🔴 沒有這個子指令:{cmd}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|