rune/scripts/version_convergence.py

#!/usr/bin/env python3
"""Version convergence checker — the decision logic for the release safety net (#192).

Reads the repo's git log (via subprocess git) and reports whether master is in
a released state. Used by the push-triggered job in
``.gitea/workflows/auto-version.yml`` and by the maintenance drift
reconciliation job in ``.gitea/workflows/maintenance.yml``.

Decision (see ``assemble_release_decision``):

* ``warranted_bump`` — the highest bump kind warranted by commit subjects since
  the latest tag: ``feat!``/``BREAKING CHANGE``/``feat`` -> minor (0.x
  convention, matching the workflow's PR-title rule), ``fix``/``perf`` ->
  patch, anything else (docs/chore/test/refactor/...) -> null.
* ``converged`` — true when no release action is warranted: either master's
  ``__version__`` matches the latest tag and no release-worthy commit sits
  unreleased since it, or master's version is not ahead of the tag in any
  actionable way. ``converged: false`` means drift: an unreleased bump or a
  missing release.

Output: the plan's JSON contract on stdout (exit 0 in both converged and
drift states; exit 1 only on operational failure):

    {"converged": bool, "warranted_bump": "minor"|"patch"|null,
     "anomaly": "version_behind_tag"|null, "commits_since_tag": N,
     "sample": [...]}
"""

from __future__ import annotations

import argparse
import json
import re
import subprocess
import sys
from pathlib import Path
from typing import Any

SAMPLE_LIMIT = 10

_VERSION_RE = re.compile(r'__version__ = "([0-9]+\.[0-9]+\.[0-9]+)"')
_FEAT_RE = re.compile(r"^\s*feat(\([^)]*\))?:")
_WARRANTY_RE = re.compile(r"^\s*(fix|perf)(\([^)]*\))?:")
# Gitea merge-commit subject: the conventional title is quoted inside
# ("Merge pull request 'feat: x' (#N)"). Spec-gate finding 2 (#192): classify
# the inner title, not the "Merge ..." wrapper (which anchors as 'other').
_MERGE_COMMIT_RE = re.compile(r"^Merge pull request '(.+)' \(#\d+\)$")


def _git(args: list[str], repo: Path | None = None) -> str:
    proc = subprocess.run(
        ["git", *args],
        cwd=str(repo) if repo else None,
        check=True,
        capture_output=True,
        text=True,
    )
    return proc.stdout.strip()


def read_version(repo: Path | None = None) -> str | None:
    """Master's ``__version__`` from src/rune/__init__.py, or None if unreadable."""
    root = repo if repo else Path.cwd()
    init_py = root / "src" / "rune" / "__init__.py"
    try:
        match = _VERSION_RE.search(init_py.read_text(encoding="utf-8"))
    except OSError:
        return None
    return match.group(1) if match else None


def latest_tag(repo: Path | None = None) -> str | None:
    """Highest version tag (e.g. ``v0.54.0``), or None when the repo has no tags."""
    tags = _git(["tag", "--sort=-v:refname"], repo).splitlines()
    return tags[0] if tags else None


def subjects_since_tag(tag: str, repo: Path | None = None) -> list[str]:
    """Commit subjects (``%s``) between ``tag`` and HEAD."""
    out = _git(["log", f"{tag}..HEAD", "--pretty=%s"], repo)
    return out.splitlines() if out else []


def classify_subject(subject: str) -> str:
    """Classify one commit subject: ``breaking`` | ``feat`` | ``fix`` | ``other``.

    Mirrors the PR-title semver rule in auto-version.yml: ``feat!``/``BREAKING
    CHANGE`` are breaking; ``feat`` is a feature; ``fix``/``perf`` warrant a
    patch; everything else releases nothing.

    Merge commits (``Merge pull request '<title>' (#N)``) are classified by
    their inner PR title so a no-bump feat PR merged via merge-commit stays
    visible to the net and to maintenance (spec-gate finding 2, #192).
    """
    merge_match = _MERGE_COMMIT_RE.match(subject)
    if merge_match:
        subject = merge_match.group(1)
    if "BREAKING CHANGE" in subject or re.match(r"^\s*\w+(\([^)]*\))?!:", subject):
        return "breaking"
    if _FEAT_RE.match(subject):
        return "feat"
    if _WARRANTY_RE.match(subject):
        return "fix"
    return "other"


def highest_warranted_bump(subjects: list[str]) -> str | None:
    """Highest bump kind warranted by commit subjects: minor > patch > None."""
    kinds = {classify_subject(s) for s in subjects}
    if "breaking" in kinds or "feat" in kinds:
        return "minor"
    if "fix" in kinds:
        return "patch"
    return None


def next_version(version: str, kind: str) -> str:
    """Apply a bump kind to a version string (0.x convention: breaking -> minor)."""
    major, minor, patch = version.split(".")
    if kind == "major":
        return f"{major}.{int(minor) + 1}.0" if int(major) == 0 else f"{int(major) + 1}.0.0"
    if kind == "minor":
        return f"{major}.{int(minor) + 1}.0"
    if kind == "patch":
        return f"{major}.{minor}.{int(patch) + 1}"
    raise ValueError(f"unknown bump kind: {kind!r}")


def version_greater(a: str, b: str) -> bool:
    """Numeric semver comparison: is ``a`` strictly greater than ``b``?"""
    return tuple(int(part) for part in a.split(".")) > tuple(int(part) for part in b.split("."))


def assemble_release_decision(
    version: str | None,
    latest_tag: str | None,
    subjects: list[str],
    commit_count: int,
) -> dict[str, Any]:
    """Assemble the plan's JSON decision from raw inputs.

    ``converged`` is false (drift) when either:
      * master's ``__version__`` equals the latest tag but release-worthy
        commits sit unreleased since it (the bump never rode the merge), or
      * master's ``__version__`` is strictly ahead of the latest tag (a bump
        rode the merge but no release/tag was cut for it), or
      * master's ``__version__`` is strictly BEHIND the latest tag (spec-gate
        finding 3, #192: a bump merged without riding a release, or a tag
        moved backwards). That state is drift with ``anomaly:
        "version_behind_tag"`` and ``warranted_bump`` forced to None — the
        net must not act (bumping master cannot converge it with a tag that
        is already ahead); maintenance files/flags it instead.
    """
    warranted = highest_warranted_bump(subjects)
    tag_version = latest_tag[1:] if latest_tag and latest_tag.startswith("v") else latest_tag

    anomaly: str | None = None
    converged = True
    if version is None and tag_version is not None:
        converged = False  # master unreadable while a release history exists
    elif version is not None and tag_version is None:
        converged = True  # no tags: first-release territory, the net never acts
    elif version is not None and tag_version is not None:
        if version == tag_version:
            converged = warranted is None
        elif version_greater(version, tag_version):
            converged = False  # bump rode the merge but no release was cut
        else:
            converged = False  # version behind tag: drift, never silent
            warranted = None
            anomaly = "version_behind_tag"

    return {
        "converged": converged,
        "warranted_bump": warranted,
        "anomaly": anomaly,
        "commits_since_tag": commit_count,
        "sample": subjects[:SAMPLE_LIMIT],
    }


def decision_in_repo(repo: Path | None = None) -> dict[str, Any]:
    """Read the repo's git state and return the release decision."""
    version = read_version(repo)
    tag = latest_tag(repo)
    subjects = subjects_since_tag(tag, repo) if tag else []
    return assemble_release_decision(version, tag, subjects, len(subjects))


def main(argv: list[str] | None = None) -> dict[str, Any]:
    parser = argparse.ArgumentParser(description="Release convergence decision (JSON on stdout)")
    parser.add_argument("--repo", type=Path, default=None, help="repo root (default: cwd)")
    args = parser.parse_args(argv)
    decision = decision_in_repo(args.repo)
    print(json.dumps(decision))
    return decision


if __name__ == "__main__":
    try:
        main()
    except subprocess.CalledProcessError as exc:
        print(f"version_convergence: git failed ({exc.returncode}): {exc.stderr}", file=sys.stderr)
        sys.exit(1)