troupe-standup/wizard-work/null/check.py

#!/usr/bin/env python3
"""null verification for the Wizard Troupe standup.

Exit code is the verdict: 0 = everything holds, 1 = something broke.

Run:  python3 /root/troupe-standup/wizard-work/null/check.py

Self-run guard: the recursive probe runs this same file with
NULL_CHECK_SELF=1, which makes the probe inert (report success without
re-executing).  Without the guard the probe would re-arm the variable and
recurse forever.
"""
import os
import re
import subprocess
import sys

MYDIR = os.path.dirname(os.path.abspath(__file__))
REPO = os.path.dirname(os.path.dirname(MYDIR))
PASSED, FAILED = [], []


def check(name, ok, detail=""):
    (PASSED if ok else FAILED).append(name)
    print(("PASS  " if ok else "FAIL  ") + name + ("  -> " + detail if detail else ""))


def main():
    # --- 1. artifact presence (all three files) ---
    for fn in ("wizard.log.md", "wizard.py", "check.py"):
        check("artifact present: " + fn, os.path.isfile(os.path.join(MYDIR, fn)))

    # --- 2. wizard.py compiles, exposes the expected API ---
    src = open(os.path.join(MYDIR, "wizard.py")).read()
    check("wizard.py compiles", _compiles(src))
    ns = {"__name__": "wizard_module", "__file__": os.path.join(MYDIR, "wizard.py")}
    if _compiles(src):
        exec(compile(src, "wizard.py", "exec"), ns)
    check("API: craft exists", callable(ns.get("craft")))
    check("API: VERDICT/CHECK markers exist", "VERDICT:" in src and "CHECK:" in src)

    # --- 3. craft() unit test: deterministic, non-empty, format correct ---
    try:
        verdict, chk, summary = ns["craft"](
            "fix the build", "staging", "Codeforge", "d7a1f", "alex"
        )
        ok = bool(verdict) and chk.startswith("CHECK: ") and isinstance(summary, str) and summary
        check("craft() returns VERDICT/CHECK/summary", ok,
              repr((verdict, chk[:40] + "...", summary[:24])))
    except Exception as e:  # noqa: BLE001
        check("craft() returns VERDICT/CHECK/summary", False, "%s: %s" % (type(e).__name__, e))

    # --- 4. report parses: exactly one VERDICT: and one CHECK: ---
    log = open(os.path.join(MYDIR, "wizard.log.md")).read()
    nv = len(re.findall(r"(?m)^VERDICT:", log))
    nc = len(re.findall(r"(?m)^CHECK:", log))
    check("wizard.log.md has exactly one VERDICT:", nv == 1, "found %d" % nv)
    check("wizard.log.md has exactly one CHECK:", nc == 1, "found %d" % nc)

    # --- 5. self-verification: run check.py and require exit 0 ---
    check("check.py self-run exits 0", _self_run(), "")

    # --- 6. git hygiene: commit exists and pushed HEAD is a descendant of it ---
    ok, detail = _git_ok()
    check("commit pushed to origin/main", ok, detail)

    print("\n%d passed, %d failed" % (len(PASSED), len(FAILED)))
    if FAILED:
        print("FAILED: " + ", ".join(FAILED))
        return 1
    print("ALL CHECKS GREEN")
    return 0


def _self_run():
    """Re-execute check.py under an inert marker env var; True iff it exits 0.

    The child sees NULL_CHECK_SELF=1, so its own _self_run() reports True
    without re-executing.  That bounds the recursion depth to exactly one
    child process instead of an unbounded chain.
    """
    if os.environ.get("NULL_CHECK_SELF") == "1":
        return True
    env = dict(os.environ, NULL_CHECK_SELF="1")
    try:
        r = subprocess.run([sys.executable, os.path.join(MYDIR, "check.py")],
                           capture_output=True, text=True, env=env, timeout=120)
        return r.returncode == 0
    except Exception:  # noqa: BLE001
        return False


def _compiles(src):
    try:
        compile(src, "wizard.py", "exec")
        return True
    except SyntaxError:
        return False


def _git_ok():
    """True iff this dir's artifact is committed by null and pushed to origin/main.

    Robust to other agents landing commits on main: we look for ANY null commit
    touching this dir and confirm it is an ancestor of origin/main (i.e. pushed),
    rather than assuming it is HEAD.
    """
    try:
        commits = subprocess.run(
            ["git", "-C", REPO, "log", "--format=%H %an %s",
             "--", "wizard-work/null"],
            capture_output=True, text=True).stdout.splitlines()
        mine = [c for c in commits if c.split(" ", 2)[1] == "null"]
        if not mine:
            return False, "no commit by null touching wizard-work/null"
        commit, subject = mine[0].split(" ", 1)[0], mine[0].split(" ", 2)[2]
        if not subject.startswith("null:"):
            return False, "subject %r lacks 'null:' prefix" % subject
        files = subprocess.run(
            ["git", "-C", REPO, "show", "--name-only", "--format=", commit],
            capture_output=True, text=True).stdout.split()
        need = {"wizard-work/null/wizard.log.md", "wizard-work/null/wizard.py",
                "wizard-work/null/check.py"}
        if not need.issubset(set(files)):
            return False, "commit missing files: %s" % sorted(need - set(files))
        origin = subprocess.run(["git", "-C", REPO, "rev-parse", "origin/main"],
                                capture_output=True, text=True).stdout.strip()
        anc = subprocess.run(
            ["git", "-C", REPO, "merge-base", "--is-ancestor", commit, origin])
        if anc.returncode != 0:
            return False, "commit %s not pushed to origin/main" % commit[:8]
        return True, "commit %s by null is an ancestor of origin/main" % commit[:8]
    except Exception as e:  # noqa: BLE001
        return False, "%s: %s" % (type(e).__name__, e)


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