check-desktop-beta-freshness.py

#!/usr/bin/env python3
"""Hourly alarm: desktop beta is stale versus the candidate train.

Uses GH_TOKEN (default repo token) and unauthenticated HTTPS. Diagnoses from
one command's output. Idempotent: healthy runs print recovered and exit 0.
"""

from __future__ import annotations

import argparse
import importlib.util
import json
import os
import re
import subprocess
import sys
import urllib.error
import urllib.request
from pathlib import Path

PLANNER_PATH = Path(__file__).with_name("plan-desktop-release.py")
SPEC = importlib.util.spec_from_file_location("plan_desktop_release", PLANNER_PATH)
assert SPEC and SPEC.loader
planner = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(planner)

BETA_APPCAST_URL = "https://api.omi.me/v2/desktop/appcast.xml?identity=beta"
CODEMAGIC_CHECK_NAME = planner.CODEMAGIC_CHECK_NAME
PROMOTION_LAG_SECONDS = 3 * 60 * 60
WEDGED_TRAIN_SECONDS = 6 * 60 * 60
AUTO_RELEASE_WORKFLOW = "desktop_auto_release.yml"
TAG_BUILD_RE = re.compile(r"^v[0-9]+\.[0-9]+\.[0-9]+\+(\d+)-macos$")
SPARKLE_VERSION_RE = re.compile(r"<sparkle:version>(\d+)</sparkle:version>|sparkle:version=\"(\d+)\"")


class FreshnessError(RuntimeError):
    pass


def parse_tag_build(tag: str) -> int | None:
    match = TAG_BUILD_RE.fullmatch(tag)
    if match is None:
        return None
    return int(match.group(1))


def live_beta_build(appcast: str) -> int | None:
    builds = [int(a or b) for a, b in SPARKLE_VERSION_RE.findall(appcast)]
    return max(builds) if builds else None


def fetch_beta_appcast() -> str:
    request = urllib.request.Request(BETA_APPCAST_URL, headers={"Accept": "application/xml, text/xml, */*"})
    try:
        with urllib.request.urlopen(request, timeout=30) as response:
            return response.read().decode("utf-8", errors="replace")
    except (urllib.error.URLError, TimeoutError, OSError) as error:
        raise FreshnessError(f"could not fetch beta appcast {BETA_APPCAST_URL}: {error}") from error


def is_releasable_path(path: str) -> bool:
    if path == "desktop/macos/CHANGELOG.json":
        return False
    if path == "desktop/macos/AGENTS.md":
        return False
    if path.startswith("desktop/macos/changelog/"):
        return False
    return True


def oldest_unreleased_releasable_main_commit(latest_tag: str) -> tuple[str, int] | None:
    """Return the oldest unshipped macOS update so merge churn cannot reset age."""
    try:
        output = planner.git(
            [
                "log",
                "--first-parent",
                "--reverse",
                "--format=%H",
                f"{latest_tag}..HEAD",
                "--",
                *planner.DESKTOP_RELEASE_PATHS,
            ]
        )
    except subprocess.CalledProcessError:
        return None
    for sha in output.splitlines():
        if not sha:
            continue
        try:
            files = planner.git(
                [
                    "diff-tree",
                    "--no-commit-id",
                    "--name-only",
                    "--diff-filter=ACDMR",
                    "-r",
                    f"{sha}^1",
                    sha,
                    "--",
                    *planner.DESKTOP_RELEASE_PATHS,
                ]
            )
        except subprocess.CalledProcessError:
            continue
        if any(is_releasable_path(path) for path in files.splitlines() if path):
            age = planner.commit_age_seconds(sha)
            if age is None:
                return None
            return sha, age
    return None


def is_ancestor(commit: str, ref: str) -> bool:
    result = subprocess.run(
        ["git", "merge-base", "--is-ancestor", commit, ref],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
    )
    return result.returncode == 0


def latest_auto_release_run_url(repository: str) -> str:
    result = subprocess.run(
        [
            "gh",
            "api",
            f"repos/{repository}/actions/workflows/{AUTO_RELEASE_WORKFLOW}/runs?per_page=1",
        ],
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )
    if result.returncode != 0:
        return f"(could not read latest {AUTO_RELEASE_WORKFLOW} run: {result.stderr.strip() or result.stdout.strip()})"
    try:
        payload = json.loads(result.stdout)
        runs = payload.get("workflow_runs") if isinstance(payload, dict) else None
        if isinstance(runs, list) and runs and isinstance(runs[0], dict):
            url = runs[0].get("html_url")
            if isinstance(url, str) and url:
                return url
    except json.JSONDecodeError:
        pass
    return f"(no {AUTO_RELEASE_WORKFLOW} runs found)"


def diagnose_promotion_lag(repository: str, tag: str, tag_sha: str) -> str:
    status, conclusion, html_url, error = planner.github_check_status(repository, tag_sha, CODEMAGIC_CHECK_NAME)
    check_url = html_url or f"(no {CODEMAGIC_CHECK_NAME} URL)"
    if error:
        return f"could not read {CODEMAGIC_CHECK_NAME} on {tag} ({tag_sha}): {error}. " f"Inspect {check_url}."
    if status is None:
        return (
            f"{CODEMAGIC_CHECK_NAME} is absent on {tag} ({tag_sha}): no Codemagic intake. "
            f"Dispatch workflow \"Build Desktop Release Candidate\" "
            f"(.github/workflows/{AUTO_RELEASE_WORKFLOW})."
        )
    if status == "completed" and conclusion == "failure":
        return (
            f"{CODEMAGIC_CHECK_NAME} failed on {tag}: {check_url}. "
            f"Re-run that Codemagic build or dispatch \"Desktop Release Recovery Required\"."
        )
    if status == "completed" and conclusion == "success":
        return (
            f"{CODEMAGIC_CHECK_NAME} succeeded on {tag}: {check_url}. "
            f"The build published but promotion did not land. "
            f"Dispatch \"Retry Desktop Beta Promotion\" "
            f"(.github/workflows/desktop_recover_beta.yml) with release_tag={tag}."
        )
    return (
        f"{CODEMAGIC_CHECK_NAME} on {tag} is {status}"
        + (f"/{conclusion}" if conclusion else "")
        + f": {check_url}. Wait for Codemagic, or dispatch \"Build Desktop Release Candidate\"."
    )


def evaluate(*, repository: str) -> tuple[int, list[str]]:
    lines: list[str] = ["Desktop beta freshness"]
    alarms: list[str] = []

    tag = planner.latest_desktop_tag()
    if tag is None:
        lines.append("No v*-macos candidate tag exists yet.")
        return 0, lines

    candidate_build = parse_tag_build(tag)
    if candidate_build is None:
        raise FreshnessError(f"newest desktop tag is not a canonical v*-macos tag: {tag}")

    appcast = fetch_beta_appcast()
    live_build = live_beta_build(appcast)
    tag_sha = planner.tag_sha(tag) or ""
    tag_age = planner.tag_age_seconds(tag)
    lines.append(f"candidate_tag={tag}")
    lines.append(f"candidate_build={candidate_build}")
    lines.append(f"live_beta_build={live_build if live_build is not None else 'missing'}")
    lines.append(f"candidate_tag_age_seconds={tag_age if tag_age is not None else 'unknown'}")

    if live_build is None:
        alarms.append(f"beta appcast {BETA_APPCAST_URL} has no sparkle:version")
    elif candidate_build > live_build and tag_age is not None and tag_age > PROMOTION_LAG_SECONDS:
        diagnosis = diagnose_promotion_lag(repository, tag, tag_sha)
        alarms.append(
            f"Promotion lag: candidate_build {candidate_build} > live_beta_build {live_build} "
            f"and {tag} is {tag_age}s old (>{PROMOTION_LAG_SECONDS}s). {diagnosis}"
        )

    oldest_unreleased = oldest_unreleased_releasable_main_commit(tag)
    if oldest_unreleased is not None:
        sha, age = oldest_unreleased
        lines.append(f"oldest_unreleased_main_sha={sha}")
        lines.append(f"oldest_unreleased_main_age_seconds={age}")
        if age > WEDGED_TRAIN_SECONDS and not is_ancestor(sha, tag):
            run_url = latest_auto_release_run_url(repository)
            alarms.append(
                f"Wedged train: oldest unshipped releasable first-parent main commit {sha} is {age}s old "
                f"(>{WEDGED_TRAIN_SECONDS}s) and is not an ancestor of {tag}. "
                f"Inspect the latest \"Build Desktop Release Candidate\" run: {run_url}. "
                f"Dispatch .github/workflows/{AUTO_RELEASE_WORKFLOW} to retry tagging."
            )
    else:
        lines.append("oldest_unreleased_main_sha=none")

    if alarms:
        lines.append("status=unhealthy")
        lines.extend(alarms)
        return 1, lines

    lines.append("status=healthy")
    return 0, lines


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--repository", default=os.environ.get("GITHUB_REPOSITORY", "BasedHardware/omi"))
    parser.add_argument("--summary", type=Path)
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    try:
        exit_code, lines = evaluate(repository=args.repository)
    except FreshnessError as error:
        lines = [f"Desktop beta freshness failed: {error}"]
        exit_code = 1
    text = "\n".join(lines) + "\n"
    sys.stdout.write(text)
    if args.summary is not None:
        args.summary.parent.mkdir(parents=True, exist_ok=True)
        args.summary.write_text(text, encoding="utf-8")
    return exit_code


if __name__ == "__main__":
    raise SystemExit(main())