Ce Release Notes

#!/usr/bin/env python3
"""
list-plugin-releases.py — Fetch compound-engineering plugin releases from GitHub.

Output: a single JSON object on stdout. Always exits 0; failures are encoded
in the contract, never raised.

Usage:
  python3 list-plugin-releases.py [--limit N] [--api-base URL]

Environment:
  CE_RELEASE_NOTES_GH_BIN  Override the gh binary path (default: "gh"). Used
                            by the test harness; leave unset in production.

Contract:
  Success:
    {"ok": true, "source": "gh"|"anon", "fetched_at": "ISO8601",
     "releases": [{tag, version, name, published_at, url, body, linked_prs}]}
  Failure:
    {"ok": false, "error": {"code": "rate_limit"|"network_outage",
                            "message": "...", "user_hint": "..."}}
"""
import argparse
import json
import os
import re
import subprocess
import sys
import time
import urllib.error
import urllib.request
from datetime import datetime, timezone

OWNER = "EveryInc"
REPO = "compound-engineering-plugin"
TAG_PREFIX = "compound-engineering-v"
DEFAULT_API_BASE = "https://api.github.com"
GH_TIMEOUT_SECS = 10
ANON_TIMEOUT_SECS = 10
RELEASES_URL = "https://github.com/" + OWNER + "/" + REPO + "/releases"
PR_REGEX = re.compile(r"\[#(\d+)\]")


def _now_iso():
    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def _extract_linked_prs(body):
    if not body:
        return []
    seen = set()
    out = []
    for m in PR_REGEX.finditer(body):
        n = int(m.group(1))
        if n not in seen:
            seen.add(n)
            out.append(n)
    return out


def _version_from_tag(tag):
    if tag.startswith(TAG_PREFIX):
        return tag[len(TAG_PREFIX):]
    return tag


def _normalize_release(raw):
    """Coerce a raw release dict (gh shape OR API shape) into the contract shape."""
    tag = raw.get("tagName") or raw.get("tag_name") or ""
    if not tag:
        return None
    body = raw.get("body") or ""
    return {
        "tag": tag,
        "version": _version_from_tag(tag),
        "name": raw.get("name") or "",
        "published_at": raw.get("publishedAt") or raw.get("published_at") or "",
        "url": raw.get("html_url") or raw.get("url") or "",
        "body": body,
        "linked_prs": _extract_linked_prs(body),
    }


def _filter_and_sort(raw_list):
    out = []
    for raw in raw_list:
        if not isinstance(raw, dict):
            continue
        norm = _normalize_release(raw)
        if norm is None:
            continue
        if not norm["tag"].startswith(TAG_PREFIX):
            continue
        out.append(norm)
    out.sort(key=lambda r: r["published_at"], reverse=True)
    return out


def attempt_gh(limit):
    """
    Try to fetch via gh. Returns (success, releases).
      success=True  → caller emits the result with source="gh"
      success=False → caller falls back to attempt_anon
    Falls back when: gh missing, gh exits non-zero, gh times out, gh stdout is
    not parseable JSON, or gh returns zero plugin tags (covers the GitHub
    Enterprise silent-empty case).
    """
    gh_bin = os.environ.get("CE_RELEASE_NOTES_GH_BIN", "gh")
    # `gh release list --json` does NOT expose `body` or `url` (only metadata
    # fields). `gh api` returns the full GitHub Releases API response shape
    # (tag_name, html_url, body, published_at, ...) and uses gh's auth so
    # there is no rate limit. The normalizer already handles this shape.
    cmd = [
        gh_bin,
        "api",
        "/repos/" + OWNER + "/" + REPO + "/releases?per_page=" + str(limit),
    ]
    try:
        result = subprocess.run(
            cmd,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            timeout=GH_TIMEOUT_SECS,
            check=False,
        )
    except (FileNotFoundError, PermissionError, subprocess.TimeoutExpired):
        return False, None
    if result.returncode != 0:
        return False, None
    try:
        raw_list = json.loads(result.stdout)
    except json.JSONDecodeError: