231 lines
10 KiB
Python
231 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""merge_robot.py — deterministic merge-on-green for this repo (#203).
|
|
|
|
Extracted from the inline workflow script so its logic is TESTABLE
|
|
(bug-fix contract, docs/bugfix-contract.md: the silent no-op defect).
|
|
|
|
Behavior:
|
|
- GET /user first: verify the token's identity and log it. Scoped-token
|
|
failures are diagnosable from the job log instead of mysterious.
|
|
- Scan open PRs; for each: combined status must be success AND mergeable.
|
|
- Merge refusals (403/405/anything non-(200,405-already-merged)) FAIL THE JOB
|
|
with the full error body — never print-and-continue. A robot that cannot
|
|
merge must be visibly broken, not silently idle.
|
|
- 405 with "behind the base branch" is retried once after a self-update of
|
|
the head (informational), then fails loudly if still refused.
|
|
- Every scanned PR produces a decision line: merged / skipped(reason).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
from typing import Any
|
|
|
|
|
|
class MergeRefused(Exception):
|
|
"""A merge the robot expected to perform was refused by the forge."""
|
|
|
|
|
|
def _api(
|
|
base: str, token: str, path: str, method: str = "GET", data: Any = None, ok: tuple = (200,)
|
|
) -> tuple[int, Any]:
|
|
req = urllib.request.Request(
|
|
base + path,
|
|
data=json.dumps(data).encode() if data is not None else None,
|
|
headers={"Authorization": "token " + token, "Content-Type": "application/json"},
|
|
method=method,
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=30) as r:
|
|
body = r.read()
|
|
status = r.status
|
|
except urllib.error.HTTPError as e:
|
|
status = e.code
|
|
body = e.read()
|
|
except urllib.error.URLError as e:
|
|
raise MergeRefused(f"forge unreachable at {path}: {e.reason}") from e
|
|
try:
|
|
parsed: Any = json.loads(body) if body else {}
|
|
except json.JSONDecodeError:
|
|
if status not in ok:
|
|
raise MergeRefused(f"{method} {path} -> {status} (non-JSON body): {body[:200]}") from None
|
|
parsed = body # 2xx but not JSON — pass through raw
|
|
if method == "GET" and status not in ok:
|
|
raise MergeRefused(f"GET {path} -> {status}: {str(parsed)[:300]}")
|
|
return status, parsed
|
|
|
|
|
|
def verify_identity(base: str, token: str) -> str:
|
|
# Defense in depth: whatever scope the caller hands us, identity is an
|
|
# INSTANCE-scoped endpoint (#214 follow-up: repo-scoped base produced
|
|
# .../repos/o/r/user -> 404 -> unhandled JSONDecodeError).
|
|
stripped = base.rstrip("/")
|
|
instance = stripped.split("/repos/")[0] if "/repos/" in stripped else base
|
|
status, user = _api(instance, token, "/user")
|
|
login = user.get("login", "?")
|
|
print(f"merge-robot: authenticated as {login} (status {status})")
|
|
return login
|
|
|
|
|
|
def scan(base: str, token: str) -> list[dict]:
|
|
status, prs = _api(base, token, "/pulls?state=open&limit=50")
|
|
print(f"merge-robot: {len(prs)} open PR(s)")
|
|
return prs
|
|
|
|
|
|
def decide(base: str, token: str, pr: dict) -> str:
|
|
"""Return a decision line for one PR; raises MergeRefused on unexpected refusals."""
|
|
n = pr["number"]
|
|
sha = pr["head"]["sha"]
|
|
status, cs = _api(base, token, f"/commits/{sha}/status")
|
|
state = cs.get("state")
|
|
if state != "success":
|
|
return f"#{n}: skipped (status={state})"
|
|
status, detail = _api(base, token, f"/pulls/{n}")
|
|
if not detail.get("mergeable", False):
|
|
# CONFLICT-FIRST (Nick policy): resolve staleness immediately —
|
|
# attempt the API update NOW rather than waiting a sleep cycle.
|
|
# 409 (genuine conflict) still skips with the reason named.
|
|
ustatus, uresp = _api(base, token, f"/pulls/{n}/update", "POST", {"style": "merge"}, ok=(200, 409))
|
|
if ustatus == 200:
|
|
print(f"UPDATED {n} (conflict-first: mergeable=False resolved pre-merge)", flush=True)
|
|
# classify the update: docs-only → carried green stands; code-path
|
|
# touch → fresh CI required before merge (#208 slip prevention)
|
|
try:
|
|
_, fresh = _api(base, token, f"/pulls/{n}")
|
|
new_sha = fresh.get("head", {}).get("sha", sha)
|
|
if classify_update(base, token, pr, new_sha) == "RERUN_REQUIRED":
|
|
return f"#{n}: skipped (updated; CI re-running on new head)"
|
|
except MergeRefused:
|
|
return f"#{n}: skipped (updated; CI re-running on new head)"
|
|
status, detail = _api(base, token, f"/pulls/{n}")
|
|
if not detail.get("mergeable", False):
|
|
return f"#{n}: skipped (updated; CI re-running on new head)"
|
|
elif ustatus == 409:
|
|
cmsg = str(uresp.get("message", "")) if isinstance(uresp, dict) else str(uresp)
|
|
return f"#{n}: skipped (genuine conflict — update 409: {cmsg[:120]})"
|
|
else:
|
|
return f"#{n}: skipped (not mergeable; update returned {ustatus})"
|
|
mstatus, mresp = _api(base, token, f"/pulls/{n}/merge", "POST", {"Do": "merge"})
|
|
if mstatus == 200:
|
|
return f"#{n}: MERGED"
|
|
if mstatus == 405 and "try again later" in str(mresp.get("message", "")).lower():
|
|
# TRANSIENT: the forge is recomputing mergeability — back off and retry.
|
|
# (Fail-loud doctrine applies to refusals; this is "not ready".)
|
|
for attempt in range(3):
|
|
time.sleep(5 * (attempt + 1))
|
|
mstatus, mresp = _api(base, token, f"/pulls/{n}/merge", "POST", {"Do": "merge"})
|
|
if mstatus == 200:
|
|
return f"#{n}: MERGED (after transient retry)"
|
|
if mstatus != 405 or "try again later" not in str(mresp.get("message", "")).lower():
|
|
break
|
|
raise MergeRefused(f"#{n} merge still transient after 3 retries: {str(mresp)[:200]}")
|
|
if mstatus == 405:
|
|
msg = str(mresp.get("message", ""))
|
|
if "behind" in msg.lower():
|
|
# SELF-HEAL (inefficiency fix): the forge exposes a merge-style
|
|
# update; perform it and retry the merge once. Skipping here is
|
|
# what caused the nightly behind-base worktree dance.
|
|
ustatus, uresp = _api(base, token, f"/pulls/{n}/update", "POST", {"style": "merge"}, ok=(200, 409))
|
|
if ustatus == 409:
|
|
cmsg = str(uresp.get("message", "")) if isinstance(uresp, dict) else str(uresp)
|
|
return f"#{n}: skipped (genuine conflict — update 409: {cmsg[:120]})"
|
|
mstatus, mresp = _api(base, token, f"/pulls/{n}/merge", "POST", {"Do": "merge"})
|
|
if mstatus == 200:
|
|
return f"#{n}: MERGED (after behind-base self-update)"
|
|
if mstatus == 405 and "behind" in str(mresp.get("message", "")).lower():
|
|
# CARRY-FORWARD PROTOCOL: classify the tree diff between the
|
|
# tested head and the new head. Docs-only movement carries the
|
|
# prior green (no redundant re-run — Nick's policy); any
|
|
# code-path touch requires fresh CI before merge (#208 slip:
|
|
# required-checks satisfied by a stale green).
|
|
try:
|
|
_, fresh = _api(base, token, f"/pulls/{n}")
|
|
new_sha = fresh.get("head", {}).get("sha", sha)
|
|
verdict = classify_update(base, token, pr, new_sha)
|
|
except MergeRefused:
|
|
verdict = "RERUN_REQUIRED" # can't verify → safest default
|
|
if verdict == "RERUN_REQUIRED":
|
|
print(f"UPDATED {n} awaiting-ci (code-path touch)", flush=True)
|
|
return f"#{n}: skipped (updated; CI re-running on new head)"
|
|
print(f"UPDATED {n} carry-forward (docs-only diff)", flush=True)
|
|
mstatus, mresp = _api(base, token, f"/pulls/{n}/merge", "POST", {"Do": "merge"})
|
|
if mstatus == 200:
|
|
return f"#{n}: MERGED (carry-forward: docs-only update)"
|
|
return f"#{n}: skipped (carry-forward merge returned {mstatus})"
|
|
raise MergeRefused(f"#{n} merge refused {mstatus} after update: {str(mresp)[:200]}")
|
|
raise MergeRefused(f"#{n} merge refused 405: {msg}")
|
|
if mstatus == 403:
|
|
raise MergeRefused(f"#{n} merge FORBIDDEN 403 — token lacks pull-request write scope: {str(mresp)[:300]}")
|
|
raise MergeRefused(f"#{n} merge failed {mstatus}: {str(mresp)[:300]}")
|
|
|
|
|
|
CODE_PATHS = (
|
|
"src/",
|
|
"tests/",
|
|
"scripts/",
|
|
"shared/",
|
|
"config/",
|
|
".gitea/",
|
|
"henxels.yaml",
|
|
"henxels_checks.py",
|
|
"pyproject.toml",
|
|
"requirements-ci.txt",
|
|
"lint-baseline.json",
|
|
)
|
|
|
|
|
|
def classify_update(base: str, token: str, pr: dict, new_sha: str) -> str:
|
|
"""Classify a self-heal update: CARRY_FORWARD (docs-only tree diff — the
|
|
tested code is byte-identical, prior green stands) or RERUN_REQUIRED
|
|
(any code-path touch — semantic interactions are real, #208 slip)."""
|
|
old_sha = pr["head"]["sha"]
|
|
status, cmp_data = _api(base, token, f"/compare/{old_sha}...{new_sha}")
|
|
files = [f.get("filename", "") for f in cmp_data.get("files", []) if isinstance(f, dict)]
|
|
return "CARRY_FORWARD" if not touches_code_path(files) else "RERUN_REQUIRED"
|
|
|
|
|
|
def touches_code_path(changed_files: list[str], code_paths=CODE_PATHS) -> bool:
|
|
"""Carry-forward policy: True iff any changed file is a code path.
|
|
Docs-only base movement must NOT trigger a redundant full re-run."""
|
|
for f in changed_files:
|
|
for cp in code_paths:
|
|
if f == cp or f.startswith(cp):
|
|
return True
|
|
return False
|
|
|
|
|
|
def main_with(argv: list[str]) -> int:
|
|
token = argv[1] if len(argv) > 1 else ""
|
|
base = argv[2] if len(argv) > 2 else ""
|
|
if not token or not base:
|
|
print("usage: merge_robot.py <token> <api-base>")
|
|
return 2
|
|
try:
|
|
verify_identity(base, token)
|
|
prs = scan(base, token)
|
|
merged = 0
|
|
for pr in prs:
|
|
line = decide(base, token, pr)
|
|
print("merge-robot:", line)
|
|
if "MERGED" in line:
|
|
merged += 1
|
|
time.sleep(1)
|
|
print(f"merge-robot: done — {merged} merged of {len(prs)} scanned")
|
|
return 0
|
|
except MergeRefused as exc:
|
|
print(f"merge-robot: FAILING JOB — {exc}", file=sys.stderr)
|
|
return 1
|
|
|
|
|
|
def main() -> int:
|
|
return main_with(sys.argv)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|