rune/scripts/auto_bump_pr.py

#!/usr/bin/env python3
"""Protected-branch-safe bump-PR machinery for the Rune release pipeline (#192).

Extracted verbatim from the 'Auto-bump' step heredoc in
``.gitea/workflows/auto-version.yml`` so both the pull_request:closed path and
the push-triggered safety net drive the same code: branch create -> contents-API
edits of the five version surfaces -> PR -> wait for the protected check ->
API merge -> confirm.

Env-driven: GITEA_TOKEN, RELEASE_SERVER_URL, RELEASE_REPO, RELEASE_VERSION,
NEW_VERSION, PR_TITLE, PR_BODY. When running inside Actions it writes
RELEASE_VERSION/AUTO_BUMPED back to GITHUB_ENV for the release step.
"""

from __future__ import annotations

import argparse
import base64
import json
import os
import sys
import time
import urllib.error
import urllib.request
from typing import Any
import sys as _sys
_SCRIPT_DIR = str(__import__('pathlib').Path(__file__).resolve().parent)
if _SCRIPT_DIR not in _sys.path:
    _sys.path.insert(0, _SCRIPT_DIR)
import gitea_auth

BUMP_BRANCH_PREFIX = "chore/auto-bump-"

# The version surface is NOT optional (spec-gate finding 4, #192): a GET
# failure here would produce a bump PR that silently bumps nothing. A GET
# failure on any other surface stays optional (pattern absent — skip).
VERSION_SURFACE = "src/rune/__init__.py"


def is_surface_fatal(path: str) -> bool:
    return path == VERSION_SURFACE


def apply_surface_edits(
    branch: str,
    version: str,
    new_version: str,
    api: Any,
    on_commit: Any = None,
) -> None:
    """Commit the version-surface edits onto ``branch`` via the contents API.

    ``api`` is the request callable; ``on_commit`` (optional) is called with
    each committed path — test seam only.
    """
    surfaces = {
        "src/rune/__init__.py": [
            (f'__version__ = "{version}"', f'__version__ = "{new_version}"'),
        ],
        "src/rune/rune_tui.py": [
            (f"Version: v{version}", f"Version: v{new_version}"),
            (f"Rune v{version} |", f"Rune v{new_version} |"),
        ],
        "src/rune/canary_gitea_automation.py": [
            (f'"{version}"', f'"{new_version}"'),
        ],
        "README.md": [
            (f"v{version}", f"v{new_version}"),
        ],
    }
    for path, pairs in surfaces.items():
        status, cur = api(f"/contents/{path}?ref={branch}")
        if status != 200:
            if is_surface_fatal(path):
                print(f"  FATAL: version surface {path} unreadable ({status}) — refusing to file a no-op bump PR")
                sys.exit(1)
            # genuinely optional surface (e.g. pattern absent) — skip gracefully
            print(f"  skip {path} ({status})")
            continue
        text = base64.b64decode(cur["content"]).decode("utf-8")
        changed = text
        for old, newv in pairs:
            changed = changed.replace(old, newv)
        if changed == text:
            print(f"  no-op {path}")
            continue
        status, r = api(
            f"/contents/{path}",
            "PUT",
            {
                "message": f"chore(release): auto-bump version to {new_version} (post-merge automation)",
                "content": base64.b64encode(changed.encode()).decode(),
                "branch": branch,
                "sha": cur["sha"],
            },
        )
        if status not in (200, 201):
            print(f"  commit {path} failed ({status})")
            sys.exit(1)
        if on_commit is not None:
            on_commit(path)


def run_bump() -> None:
    TOKEN = gitea_auth.get_token()
    BASE = os.environ["RELEASE_SERVER_URL"] + "/api/v1/repos/" + os.environ["RELEASE_REPO"]
    VERSION = os.environ["RELEASE_VERSION"]
    NEW = os.environ["NEW_VERSION"]
    PR_TITLE = os.environ.get("PR_TITLE", "")
    PR_BODY = os.environ.get("PR_BODY", "")
    BRANCH = f"chore/auto-bump-{NEW}"

    def api(
        path: str,
        method: str = "GET",
        data: dict[str, Any] | None = None,
        ok: tuple[int, ...] = (200,),
    ) -> tuple[int, dict[str, 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()
                return r.status, (json.loads(body) if body else {})
        except urllib.error.HTTPError as e:
            body = e.read().decode("utf-8", "replace")
            print(f"  API {method} {path} -> {e.code}: {body[:200]}")
            return e.code, {"_body": body}

    # 1. bump branch from master
    status, _ = api("/branches", "POST", {"new_branch_name": BRANCH, "old_branch_name": "master"})
    if status not in (201, 409):
        print(f"branch create failed ({status})")
        sys.exit(1)

    # 2. commit the version-surface edits via the contents API
    apply_surface_edits(BRANCH, VERSION, NEW, api)

    # 3. bump PR
    status, pr = api(
        "/pulls",
        "POST",
        {
            "title": f"chore(release): auto-bump version to {NEW} (post-merge automation)",
            "body": f"Automatic version bump after merge of: {PR_TITLE}\n\n{PR_BODY[:1500]}",
            "head": BRANCH,
            "base": "master",
        },
    )
    if status not in (200, 201):
        print(f"PR create failed ({status})")
        sys.exit(1)
    pr_number = pr["number"]
    head_sha = pr["head"]["sha"]
    print(f"  bump PR #{pr_number} head {head_sha[:12]}")

    # 4. wait for the protected check on the bump PR (same gate as everything else)
    deadline = time.time() + 900  # 15 min ceiling
    state = None
    while time.time() < deadline:
        st, cs = api(f"/commits/{head_sha}/status")
        state = cs.get("state")
        if state in ("success", "failure"):
            break
        time.sleep(15)
    if state != "success":
        print(f"bump PR CI did not go green (state={state})")
        sys.exit(1)

    # 5. merge via API (respects branch protection; user_can_merge)
    status, _ = api(f"/pulls/{pr_number}/merge", "POST", {"Do": "merge"})
    if status not in (200, 405):
        print(f"merge failed ({status})")
        sys.exit(1)

    # 6. confirm merged, then export the version for the release step
    st, pr2 = api(f"/pulls/{pr_number}")
    if not pr2.get("merged"):
        print("merge did not take")
        sys.exit(1)
    print("AUTO_BUMPED=true")
    gh_env = os.environ.get("GITHUB_ENV")
    if gh_env:
        with open(gh_env, "a") as env:
            env.write(f"RELEASE_VERSION={NEW}\n")
            env.write("AUTO_BUMPED=true\n")


def list_branch_names(base: str, token: str) -> list[str]:
    req = urllib.request.Request(
        base + "/branches?limit=50",
        headers={"Authorization": "token " + token, "Content-Type": "application/json"},
    )
    with urllib.request.urlopen(req, timeout=30) as r:
        return [b.get("name", "") for b in json.loads(r.read())]


def bump_branch_in_flight(names: list[str], new_version: str | None = None) -> bool:
    """True when the bump branch for THIS merge's in-flight version exists —
    evidence the PR-closed path is mid-bump.

    Spec-gate finding 7a (#192): the guard used to match ANY chore/auto-bump-*
    branch, so one stale historical branch (they are never cleaned up) stood
    the net down permanently. It now matches exactly chore/auto-bump-{NEW};
    pass new_version=None to restore the legacy prefix-wide behavior.
    """
    if new_version is None:
        return any(name.startswith(BUMP_BRANCH_PREFIX) for name in names)
    return f"{BUMP_BRANCH_PREFIX}{new_version}" in names


def wait_out_closed_path(
    base: str, token: str, grace_seconds: int, poll_seconds: int = 15, new_version: str | None = None
) -> bool:
    """Wait for evidence that the PR-closed path is handling this merge.

    Both events fire from the same merge, so a living closed path creates its
    chore/auto-bump-{new_version} branch within seconds of job start. Returns
    True (safety net stands down) as soon as it appears, or False (closed
    event genuinely absent — the #190/#192 defect) once the grace window
    elapses. Best-effort attribution only (see docs/release-flow.md): a stale
    branch of the same version, or a >grace delayed closed path, can still
    misattribute — double-bump risk is bounded by release tag idempotency.
    """
    deadline = time.time() + grace_seconds
    while True:
        if bump_branch_in_flight(list_branch_names(base, token), new_version):
            return True
        if time.time() >= deadline:
            return False
        time.sleep(poll_seconds)


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description="Protected-branch-safe bump-PR machinery (#192)")
    parser.add_argument("--dry-run", action="store_true", help="print the decision without executing any API call")
    parser.add_argument(
        "--safety-net",
        action="store_true",
        help="push path: first wait out a possibly-alive PR-closed path before acting",
    )
    parser.add_argument(
        "--attribution-grace-seconds",
        type=int,
        default=90,
        help="how long --safety-net waits for the PR-closed path's bump branch to appear",
    )
    args = parser.parse_args(argv)

    if args.dry_run:
        version = os.environ.get("RELEASE_VERSION", "")
        new = os.environ.get("NEW_VERSION", "")
        print(
            json.dumps(
                {
                    "dry_run": True,
                    "release_version": version,
                    "new_version": new,
                    "branch": f"{BUMP_BRANCH_PREFIX}{new}",
                    "pr_title": f"chore(release): auto-bump version to {new} (post-merge automation)",
                }
            )
        )
        return 0

    if args.safety_net:
        base = os.environ["RELEASE_SERVER_URL"] + "/api/v1/repos/" + os.environ["RELEASE_REPO"]
        token = gitea_auth.get_token()
        # Attribute against exactly this merge's would-be bump branch: a stale
        # chore/auto-bump-* branch from an older version must not stand the
        # net down (spec-gate finding 7a, #192).
        new_version = os.environ.get("NEW_VERSION") or None
        try:
            closed_path_alive = wait_out_closed_path(
                base, token, args.attribution_grace_seconds, new_version=new_version
            )
        except (urllib.error.HTTPError, urllib.error.URLError, OSError) as exc:
            print(f"attribution check failed ({exc}) — refusing to act (conservative)")
            return 1
        if closed_path_alive:
            print("PR-closed path is mid-bump (bump branch for this version exists) — safety net stands down")
            return 0
        print(
            f"no bump branch after {args.attribution_grace_seconds}s — closed event genuinely absent; safety net acts"
        )

    run_bump()
    return 0


if __name__ == "__main__":
    sys.exit(main())