rune/scripts/release_history_cli.py

#!/usr/bin/env python3
"""release_history_cli.py — thin workflow entry points for release history.

SCAFFOLD-TRANCHE-009B §4: the workflows invoke THIS tested module; no
note-building logic lives in YAML. Two moments:

    init   RELEASE INITIALIZATION — auto-version creates the evidence-
           derived Gitea release when the version/tag appears.
    refresh  RELEASE CONVERGENCE REFRESH — deploy promotes a later SHA of
           the same version; the SAME release is updated with the final
           deployed identity + newly-governed facts. Idempotent.

Fail-closed boundary (§5/§16): `init` exits non-zero on publication
failure — Rune cannot silently create another empty historical release.
`refresh` ALSO exits non-zero (visible failure + retry converges), but the
workflow marks it non-blocking for deployment validity: valid software is
never rolled back over documentation. Failure state is visible either way.
"""
from __future__ import annotations

import argparse
import json
import os
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "src"))
from rune import release_notes as rn  # noqa: E402


def _token() -> str:
    tok = os.environ.get("GITEA_TOKEN", "")
    if tok:
        return tok
    # local/dev fallback: git credential store (runner uses GITEA_TOKEN)
    cred = Path.home() / ".git-credentials"
    if cred.exists():
        import re
        m = re.search(r"https://[^:]+:([^@]+)@", cred.read_text())
        if m:
            return m.group(1)
    return ""


def main(argv=None) -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    sub = ap.add_subparsers(dest="cmd", required=True)

    p_init = sub.add_parser("init", help="release initialization")
    p_init.add_argument("--version", required=True)
    p_init.add_argument("--server", required=True)
    p_init.add_argument("--repo", required=True)
    p_init.add_argument("--store", default="store/scaffold/current.json")
    p_init.add_argument("--release-sha", default="")
    p_init.add_argument("--deployed-sha", default="")
    p_init.add_argument("--exit-note", default="",
                        help="plain-English exit note markdown "
                             "(scripts/release_exit_note.py output)")

    p_ref = sub.add_parser("refresh", help="convergence refresh")
    p_ref.add_argument("--version", required=True)
    p_ref.add_argument("--server", required=True)
    p_ref.add_argument("--repo", required=True)
    p_ref.add_argument("--store", default="store/scaffold/current.json")
    p_ref.add_argument("--release-sha", default="")
    p_ref.add_argument("--deployed-sha", required=True,
                       help="converged deployed commit (from the deployment "
                            "authority — required, never assumed)")

    args = ap.parse_args(argv)
    token = _token()
    if not token:
        print(json.dumps({"failed": True,
                          "error": "no Gitea token available"}))
        return 1

    if args.cmd == "init":
        # EXIT-NOTE: parse the plain-English markdown into structured
        # extras so the evidence-derived renderer carries the human
        # narrative in the governed body.
        extra = None
        if args.exit_note:
            items = [l[2:].strip() for l in args.exit_note.splitlines()
                     if l.startswith("- ")]
            headline = ""
            for l in args.exit_note.splitlines():
                if l and not l.startswith(("#", "-", " ")) and "What changed" not in l:
                    headline = l.strip()
                    break
            extra = {"improved": ([headline] if headline else []) + items}
        note = rn.auto_initial_note(
            version=args.version,
            summary=f"Rune v{args.version} — evidence-derived release "
                    f"history (automatic initialization)",
            scaffold_store=args.store,
            release_sha=args.release_sha,
            deployed_sha=args.deployed_sha or args.release_sha,
            extra=extra,
        )
        result = rn.publish_release(args.server, args.repo, token, note)
        print(json.dumps(result))
        # fail-closed: initialization failure must be visible so Rune cannot
        # silently create another empty historical release (§5/§16)
        return 0 if not result.get("failed") else 1

    if args.cmd == "refresh":
        # §12C RACE-SAFE REVALIDATION: between promotion and refresh, a
        # NEWER deploy may have superseded this candidate. The deployment
        # authority is re-read NOW: if current's build identity differs
        # from the candidate, this job is stale → refuse (preserve T11
        # invariant: stale jobs never overwrite release history). When
        # current AGREES, the candidate IS the converged truth.
        current_link = os.environ.get(
            "RUNE_DEPLOY_CURRENT", "/mnt/workdrive/rune-code/current")
        try:
            deployed_now = rn.converged_deployed_sha(
                str(Path(current_link).parent))
        except (OSError, ValueError):
            deployed_now = ""
        if args.deployed_sha and deployed_now and \
                not deployed_now.startswith(args.deployed_sha[:12]):
            print(json.dumps({
                "failed": True, "action": "refresh-superseded",
                "error": f"deployment authority shows {deployed_now[:12]}, "
                         f"candidate {args.deployed_sha[:12]} was superseded "
                         f"— stale refresh refused (history preserved)",
                "authority": deployed_now[:12],
            }))
            return 1  # visible; retry by the WINNING deploy converges

        existing = rn.fetch_release(args.server, args.repo, token,
                                    f"v{args.version}")
        if existing is None:
            # no release yet for this version — init instead (convergence
            # may legitimately arrive before/without an init if the
            # auto-version run failed); idempotent recovery path
            note = rn.build_from_deployment(
                version=args.version, deployed_sha=args.deployed_sha,
                scaffold_store=args.store, release_sha=args.release_sha)
        else:
            note = rn.convergence_refresh(
                rn.ReleaseNote.from_dict({
                    "version": args.version,
                    "summary": _summary_from_body(existing.get("body", "")),
                    "provenance": [],
                }),
                deployed_sha=args.deployed_sha, scaffold_store=args.store,
                extra_provenance="automatic convergence refresh (deploy "
                                 "workflow)")
            # preserve structural fields the stored body carries
            note.release_sha = _sha_from_body(existing.get("body", ""),
                                              "tag") or args.release_sha
        result = rn.publish_release(args.server, args.repo, token, note)
        print(json.dumps(result))
        # visible failure; retry converges (§10). Deployment validity is
        # NOT rolled back over documentation — the workflow marks this
        # step appropriately.
        return 0 if not result.get("failed") else 1
    return 2


def _summary_from_body(body: str) -> str:
    """Delegate to the single authority in rune.release_notes (§14B)."""
    return rn.summary_from_body(body)


def _sha_from_body(body: str, which: str) -> str:
    return rn.sha_from_body(body, which)


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