agent-orchestrator/tangled_pr_merge.py

#!/usr/bin/env python3
"""Merge or close a Tangled pull as the bot, with a mergeability pre-check.

WHY: `main` advances only by operator-approved merges of the stacked pulls, and the
stack must be merged strictly bottom-up. The appview exposes both actions as bare
htmx POSTs with no body:
    POST /{owner}/{repo}/pulls/{n}/merge
    POST /{owner}/{repo}/pulls/{n}/close
Neither tells you up front whether the pull is mergeable — that lives in a separate
fragment, GET /{owner}/{repo}/pulls/{n}/round/{r}/actions, which reports "No conflicts,
ready to merge" / "patch is empty" / a conflict. This checks that fragment first and
refuses to merge anything it does not understand, so a bad rung stops the frontier
instead of landing broken.

Note: a pull whose fragment says "patch is empty" (a branch identical to its parent —
e.g. an index slot whose content moved out of the repo) CANNOT be merged; close it.

USAGE:
  tangled_pr_merge.py --owner X --repo Y --pull 76 --check
  tangled_pr_merge.py --owner X --repo Y --pull 76 --merge
  tangled_pr_merge.py --owner X --repo Y --pull 75 --close --reason "empty after restack"
"""
import argparse, os, re, sys, urllib.error, urllib.request

BASE = "https://tangled.org"


def load_cookie():
    sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
    import secrets as _store
    c = _store.get("tangled.cookie")
    if not c:
        sys.exit("no tangled.cookie in the secret store")
    return c


def get(url, cookie):
    req = urllib.request.Request(url, headers={"Cookie": cookie, "User-Agent": "tangled-pr-bot"})
    return urllib.request.urlopen(req, timeout=60).read().decode()


def post(url, cookie, referer):
    req = urllib.request.Request(
        url,
        data=b"",
        headers={
            "Cookie": cookie,
            "User-Agent": "tangled-pr-bot",
            "HX-Request": "true",
            "Content-Type": "application/x-www-form-urlencoded",
            "Referer": referer,
        },
    )
    return urllib.request.urlopen(req, timeout=120)


def latest_round(doc):
    idxs = re.findall(r'name="pull-round-idx"[^>]*value="(\d+)"', doc)
    return max((int(i) for i in idxs), default=0)


def state(owner, repo, pull, cookie):
    """(verdict, round) — verdict is the appview's own mergeability text."""
    page = get(f"{BASE}/{owner}/{repo}/pulls/{pull}", cookie)
    rnd = latest_round(page)
    frag = get(f"{BASE}/{owner}/{repo}/pulls/{pull}/round/{rnd}/actions", cookie)
    txt = re.sub(r"\s+", " ", re.sub(r"<[^>]+>", " ", frag)).strip()
    for phrase in ("No conflicts, ready to merge", "patch is empty", "merged", "conflict"):
        if phrase.lower() in txt.lower():
            return phrase, rnd, txt
    return "UNKNOWN", rnd, txt


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--owner", required=True)
    ap.add_argument("--repo", required=True)
    ap.add_argument("--pull", required=True, type=int)
    g = ap.add_mutually_exclusive_group(required=True)
    g.add_argument("--check", action="store_true")
    g.add_argument("--merge", action="store_true")
    g.add_argument("--close", action="store_true")
    ap.add_argument("--reason", default=None, help="close only: logged, not posted")
    a = ap.parse_args()

    cookie = load_cookie()
    verdict, rnd, txt = state(a.owner, a.repo, a.pull, cookie)
    print(f"#{a.pull} round {rnd}: {verdict}")

    if a.check:
        print(f"  raw: {txt[:200]}")
        return

    referer = f"{BASE}/{a.owner}/{a.repo}/pulls/{a.pull}"
    if a.merge:
        if verdict != "No conflicts, ready to merge":
            sys.exit(f"refusing to merge #{a.pull}: appview says {verdict!r}")
        r = post(f"{referer}/merge", cookie, referer)
        print(f"  merged #{a.pull} — HTTP {r.status}")
    else:
        r = post(f"{referer}/close", cookie, referer)
        print(f"  closed #{a.pull} — HTTP {r.status}" + (f" ({a.reason})" if a.reason else ""))


if __name__ == "__main__":
    main()