claude-plugins/.github/scripts/discover_bumps.py

#!/usr/bin/env python3
"""Discover plugins in marketplace.json whose upstream repo has moved past
their pinned SHA, update the file in place, and emit a summary.

Adapted from claude-plugins-community-internal's discover_bumps.py for the
single-file marketplace.json format used by claude-plugins-official.

Usage: discover_bumps.py [--plugin NAME] [--max N] [--dry-run]
"""

import argparse
import json
import os
import re
import subprocess
import sys
from datetime import datetime, timezone
from typing import Any


MARKETPLACE_PATH = ".claude-plugin/marketplace.json"


def gh_api(path: str) -> Any:
    """GET from the GitHub API. None on not-found; raises on other errors.

    "Not found" covers both 404 (resource gone) and 422 "No commit found
    for SHA" (force-pushed away). Both mean the thing we asked for isn't
    there — treating them the same lets callers handle dead refs uniformly.
    """
    r = subprocess.run(
        ["gh", "api", path], capture_output=True, text=True
    )
    if r.returncode != 0:
        combined = r.stdout + r.stderr
        if any(s in combined for s in ("404", "Not Found", "No commit found")):
            return None
        raise RuntimeError(f"gh api {path}: {r.stderr.strip() or r.stdout.strip()}")
    return json.loads(r.stdout)


def parse_github_repo(url: str) -> tuple[str, str] | None:
    """Extract (owner, repo) from a URL or owner/repo shorthand."""
    # Full URL: https://github.com/owner/repo(.git)(/...)
    m = re.match(r"https?://github\.com/([^/]+)/([^/]+?)(?:\.git)?(?:/|$)", url)
    if m:
        return m.group(1), m.group(2)
    # Shorthand: owner/repo
    m = re.match(r"^([\w.-]+)/([\w.-]+)$", url)
    if m:
        return m.group(1), m.group(2)
    return None


def latest_sha(owner: str, repo: str, *, ref: str | None, path: str | None) -> str | None:
    """Latest commit SHA for the repo, optionally scoped to a ref and/or path."""
    if path:
        # Scoped to a subdirectory — use the commits list endpoint with path filter.
        q = f"repos/{owner}/{repo}/commits?per_page=1&path={path}"
        if ref:
            q += f"&sha={ref}"
        commits = gh_api(q)
        if not commits:
            return None
        return commits[0]["sha"]
    # Whole repo — the single-ref endpoint is cheaper.
    if not ref:
        meta = gh_api(f"repos/{owner}/{repo}")
        if not meta:
            return None
        ref = meta["default_branch"]
    c = gh_api(f"repos/{owner}/{repo}/commits/{ref}")
    return c["sha"] if c else None


def pinned_age_days(owner: str, repo: str, sha: str) -> int | None:
    """Days since the pinned commit was authored. Used for oldest-first rotation."""
    c = gh_api(f"repos/{owner}/{repo}/commits/{sha}")
    if not c:
        return None
    dt = datetime.fromisoformat(
        c["commit"]["committer"]["date"].replace("Z", "+00:00")
    )
    return (datetime.now(timezone.utc) - dt).days


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--plugin", help="only check this plugin")
    ap.add_argument("--max", type=int, default=20, help="cap bumps emitted")
    ap.add_argument("--dry-run", action="store_true", help="don't write marketplace.json")
    args = ap.parse_args()

    with open(MARKETPLACE_PATH) as f:
        marketplace = json.load(f)

    plugins = marketplace.get("plugins", [])
    bumps: list[dict] = []
    dead: list[str] = []
    skipped_non_github = 0
    checked = 0

    for plugin in plugins:
        name = plugin.get("name", "?")
        src = plugin.get("source")

        # Only process object sources with a sha field
        if not isinstance(src, dict) or "sha" not in src:
            continue

        # Filter to specific plugin if requested
        if args.plugin and name != args.plugin:
            continue

        checked += 1
        kind = src.get("source")
        url = src.get("url", "")
        path = src.get("path")
        ref = src.get("ref")
        pinned = src.get("sha")

        slug = parse_github_repo(url)
        if not slug:
            skipped_non_github += 1
            continue
        owner, repo = slug

        try:
            latest = latest_sha(owner, repo, ref=ref, path=path)
        except RuntimeError as e:
            print(f"::warning::{name}: {e}", file=sys.stderr)
            continue

        if latest is None:
            dead.append(f"{name} ({owner}/{repo})")
            continue

        if latest == pinned:
            continue  # up to date

        # Age lookup for rotation — oldest-pinned first prevents starvation.
        try:
            age = pinned_age_days(owner, repo, pinned) if pinned else None
        except RuntimeError as e:
            print(f"::warning::{name}: age lookup failed: {e}", file=sys.stderr)
            age = None

        bumps.append({
            "name": name,
            "kind": kind,
            "url": url,
            "path": path or "",
            "ref": ref or "",
            "old_sha": pinned or "",
            "new_sha": latest,
            "age_days": age if age is not None else 10**6,
        })

    # Oldest-pinned first so nothing starves under the cap.
    bumps.sort(key=lambda b: -b["age_days"])
    emitted = bumps[: args.max]

    # Apply bumps to marketplace data
    if emitted and not args.dry_run:
        bump_map = {b["name"]: b["new_sha"] for b in emitted}
        for plugin in plugins:
            name = plugin.get("name")
            src = plugin.get("source")
            if isinstance(src, dict) and name in bump_map:
                src["sha"] = bump_map[name]

        with open(MARKETPLACE_PATH, "w") as f:
            json.dump(marketplace, f, indent=2, ensure_ascii=False)
            f.write("\n")

    # Write GitHub outputs
    out = os.environ.get("GITHUB_OUTPUT")
    if out:
        bumped_names = ",".join(b["name"] for b in emitted)
        with open(out, "a") as fh:
            fh.write(f"count={len(emitted)}\n")
            fh.write(f"bumped_names={bumped_names}\n")

    # Write GitHub step summary
    summary = os.environ.get("GITHUB_STEP_SUMMARY")
    if summary:
        with open(summary, "a") as fh:
            fh.write("## SHA Bump Discovery\n\n")
            fh.write(f"- Checked: {checked} SHA-pinned entries\n")
            fh.write(f"- Stale: {len(bumps)} (applying {len(emitted)}, cap {args.max})\n")
            if skipped_non_github:
                fh.write(f"- Skipped non-GitHub: {skipped_non_github}\n")
            if dead:
                fh.write(f"- **Dead upstream** ({len(dead)}): {', '.join(dead)}\n")
            if emitted:
                fh.write("\n| Plugin | Old | New | Age |\n|---|---|---|---|\n")
                for b in emitted:
                    old = b["old_sha"][:8] if b["old_sha"] else "(unpinned)"
                    fh.write(f"| {b['name']} | `{old}` | `{b['new_sha'][:8]}` | {b['age_days']}d |\n")

    # Write PR body for the workflow to use
    pr_body_path = os.environ.get("PR_BODY_PATH", "/tmp/bump-pr-body.md")
    if emitted:
        with open(pr_body_path, "w") as fh:
            fh.write("Upstream repos moved. Bumping pinned SHAs so plugins track latest.\n\n")
            fh.write("| Plugin | Old | New | Upstream |\n")
            fh.write("|--------|-----|-----|----------|\n")
            for b in emitted:
                old = b["old_sha"][:8] if b["old_sha"] else "(unpinned)"
                slug_str = re.sub(r"https?://github\.com/", "", b["url"])
                slug_str = re.sub(r"\.git$", "", slug_str)
                compare = f"https://github.com/{slug_str}/compare/{b['old_sha'][:12]}...{b['new_sha'][:12]}"
                fh.write(f"| `{b['name']}` | `{old}` | `{b['new_sha'][:8]}` | [diff]({compare}) |\n")
            fh.write(f"\n---\n_Auto-generated by `bump-plugin-shas.yml` on {datetime.now(timezone.utc).strftime('%Y-%m-%d')}_\n")

    # Console summary
    print(f"Checked {checked} SHA-pinned plugins", file=sys.stderr)
    print(f"Stale: {len(bumps)}, applying: {len(emitted)}", file=sys.stderr)
    if dead:
        print(f"Dead upstream: {', '.join(dead)}", file=sys.stderr)
    for b in emitted:
        old = b["old_sha"][:8] if b["old_sha"] else "unpinned"
        print(f"  {b['name']}: {old} -> {b['new_sha'][:8]} ({b['age_days']}d)", file=sys.stderr)

    return 0


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