runbook-tutorials/rtut/__main__.py

"""CLI: python -m rtut {list,validate,show,render} ..."""
from __future__ import annotations

import argparse
import sys
from pathlib import Path

from . import maturity as M
from . import pack as P


def _packs(root: str) -> list[P.Pack]:
    if Path(root).is_file():
        return [P.load(Path(root))]
    return [P.load(p) for p in P.discover(Path(root))]


def _run(pk, a) -> int:
    from . import run as R
    try:
        if a.action == "new":
            r = R.create(pk, a.operator, dict(kv.split("=", 1) for kv in a.param))
            print(r.data["run_id"])
            return 0
        if a.action == "list":
            print("\n".join(R.list_runs(pk)))
            return 0
        r = R.load(pk, a.run_id)
        if a.action == "status":
            for sid, st in r.data["steps"].items():
                v = st.get("verification", {})
                print(f"{st['state']:8} {sid}  {'verified' if v.get('passed') else ''}")
            return 0
        if a.action == "cmd":
            print(R.command_for(r, a.step)); return 0
        if a.action == "start":
            R.start(r, a.step); print(R.command_for(r, a.step) or ""); return 0
        if a.action == "verify":
            res = R.verify(r, a.step, confirm=a.yes, manual_ok=a.passed or None)
            print("passed" if res["passed"] else "FAILED"); return 0 if res["passed"] else 1
        if a.action == "done":
            R.complete(r, a.step, dict(kv.split("=", 1) for kv in a.evidence)); return 0
        if a.action == "fail":
            R.fail(r, a.step, a.note); return 0
        if a.action == "retry":
            R.retry(r, a.step); return 0
        if a.action == "skip":
            R.skip(r, a.step, a.note); return 0
        if a.action == "finalize":
            R.finalize(r, a.outcome or "", a.note)
            print(r.data["outcome"]); return 0
        if a.action == "capture":
            R.add_capture(r, a.step, sys.stdin.read(), a.command); print("captured"); return 0
        if a.action == "feedback":
            R.add_feedback(r, a.rating or "", a.note); print("recorded"); return 0
        raise ValueError(f"unknown action {a.action}")
    except (R.RunError, KeyError, ValueError) as e:
        print(f"error: {e}", file=sys.stderr)
        return 1


def _check(root: str) -> int:
    """One command for CI and for humans: everything declarative in a repository must validate."""
    from . import project as J
    from .attrs import Registry
    reg = Registry([Path(root)])
    bad = 0
    print("runbooks")
    for p in reg.packs.values():
        print(f"  {'ok  ' if p.ok else 'FAIL'} {p.id}")
        for err in p.errors:
            print(f"       - {err}")
        bad += not p.ok
    print("attribute sets")
    for sid, d in reg.sets.items():
        print(f"  {'ok  ' if d.ok else 'FAIL'} {sid}")
        for err in d.errors:
            print(f"       - {err}")
        bad += not d.ok
    print("projects")
    for pid, d in reg.projects.items():
        errs = J.validate(reg, d)
        print(f"  {'ok  ' if not errs else 'FAIL'} {pid}")
        for err in errs:
            print(f"       - {err}")
        bad += bool(errs)
    print("templates")
    for tid, d in reg.templates.items():
        print(f"  {'ok  ' if d.ok else 'FAIL'} {tid}")
        for err in d.errors:
            print(f"       - {err}")
        bad += not d.ok
    total = len(reg.packs) + len(reg.sets) + len(reg.projects) + len(reg.templates)
    if not total:
        print("nothing to check", file=sys.stderr)
        return 1
    print(f"{total - bad} of {total} ok")
    return 1 if bad else 0


def _project(a) -> int:
    from . import project as J
    from .attrs import Registry
    reg = Registry([Path(a.root)])
    try:
        if a.action == "list":
            for pid, d in reg.projects.items():
                st = J.state(reg, d)["state"] if d.ok else "invalid"
                print(f"{pid}\t{d.data.get('environment', '?')}\t{st}\t{d.data.get('title', '')}")
            for tid, d in reg.templates.items():
                print(f"{tid}\ttemplate\t-\t{d.data.get('title', '')}")
            return 0
        if a.action == "new":
            path = J.instantiate(reg, a.name, a.extra, a.environment, dict(kv.split("=", 1) for kv in a.set),
                                 title=a.title, description=a.description, goal=a.goal)
            print(path); return 0
        d = reg.projects.get(a.name)
        if d is None:
            print(f"error: no project {a.name}", file=sys.stderr); return 1
        errs = J.validate(reg, d)
        if a.action == "validate":
            print(("ok   " if not errs else "FAIL ") + d.id)
            for e in errs:
                print("     -", e)
            return 1 if errs else 0
        if errs:
            print("\n".join(errs), file=sys.stderr); return 1
        if a.action == "status":
            st = J.state(reg, d)
            print(f"{d.id}: {st['state']}  [{d.data['environment']}]  {d.data.get('goal', '')}")
            for m in st["members"]:
                print(f"  {m['status']:12} {m['pack']}" + (f" @{m['target']}" if m["target"] else ""))
            print("  next:", ", ".join(st["next"]) or "-")
            for n in st["needs"]:
                print(f"  needs: {n['attribute']} ({n['pack']})")
            return 0
        if a.action == "attrs":
            member = next(m for m in d.data["runbooks"] if m["pack"] == a.extra)
            res = J.member_attributes(reg, d.data, member)
            for r in res["rows"]:
                print(f"{r['state']:4} {r['requirement']:9} {r['id']} = {r['value']!r}  <- {r['source'] or 'runbook default' if 'default' in r else r['source'] or '-'}"
                      + (" (modified)" if r["modified"] else "") + (f"  ERROR: {r['error']}" if r.get("error") else ""))
            if res["unused"]:
                print("unused attributes:", ", ".join(res["unused"]))
            return 0
        if a.action == "run":
            r = J.start_run(reg, d, a.extra, a.operator, a.target, a.mode)
            print(r.data["run_id"]); return 0
    except Exception as e:                       # user-facing command: report the problem, do not print a trace
        print(f"error: {e}", file=sys.stderr); return 1
    return 0


def main(argv: list[str] | None = None) -> int:
    ap = argparse.ArgumentParser(prog="rtut")
    sub = ap.add_subparsers(dest="cmd", required=True)
    for name in ("list", "validate"):
        sp = sub.add_parser(name)
        sp.add_argument("root", nargs="?", default=".")
    sp = sub.add_parser("run", help="record a run of a native pack")
    sp.add_argument("action", choices=["new", "list", "status", "cmd", "start", "verify", "done", "fail", "retry", "skip", "feedback", "finalize", "capture"])
    sp.add_argument("manifest")
    sp.add_argument("run_id", nargs="?")
    sp.add_argument("step", nargs="?")
    sp.add_argument("--operator", "-o", default="")
    sp.add_argument("--param", "-p", action="append", default=[], help="name=value")
    sp.add_argument("--evidence", "-e", action="append", default=[], help="name=value (non-secret)")
    sp.add_argument("--note", default="")
    sp.add_argument("--outcome", choices=["succeeded", "failed", "abandoned"], help="finalize: how the run ended")
    sp.add_argument("--command", default="", help="capture: the command you actually ran; output is read from stdin")
    sp.add_argument("--rating", choices=["good", "neutral", "poor"], help="feedback: how the runbook felt")
    sp.add_argument("--yes", action="store_true", help="confirm running a verify command")
    sp.add_argument("--pass", dest="passed", action="store_true", help="manual check: operator confirms it passed")
    sp = sub.add_parser("ui", help="serve the local UI on 127.0.0.1")
    sp.add_argument("roots", nargs="*", default=["."])
    sp.add_argument("--port", type=int, default=8877)
    sp = sub.add_parser("seal", help="check the seal of an exported run or project-run record (JSON file)")
    sp.add_argument("record")
    sp = sub.add_parser("check", help="validate every pack, attribute set, project and template under a repository")
    sp.add_argument("root", nargs="?", default=".")
    sp = sub.add_parser("maturity", help="derived maturity of a pack at its current revision")
    sp.add_argument("manifest")
    sp = sub.add_parser("review", help="record that you reviewed the pack at its current revision")
    sp.add_argument("manifest"); sp.add_argument("--by", required=True); sp.add_argument("--note", default="")
    sp = sub.add_parser("project", help="projects: list, validate, status, attrs, new (from a template), run")
    sp.add_argument("action", choices=["list", "validate", "status", "attrs", "new", "run"])
    sp.add_argument("root")
    sp.add_argument("name", nargs="?", help="project id (new: template id)")
    sp.add_argument("extra", nargs="?", help="attrs/run: pack id; new: the new project id")
    sp.add_argument("--environment", "-e", default="")
    sp.add_argument("--set", "-s", action="append", default=[], help="new: attribute=value")
    sp.add_argument("--operator", "-o", default="")
    sp.add_argument("--title", default=None, help="new: the project's own name (default: its id)")
    sp.add_argument("--description", default="", help="new: what the project is")
    sp.add_argument("--goal", default="", help="new: what should be true when it is done")
    sp.add_argument("--target", default=None)
    sp.add_argument("--mode", default="manual", choices=["manual", "auto-confirm"])
    sp = sub.add_parser("score", help="how often each runbook was tried and how it went")
    sp.add_argument("root", nargs="?", default=".")
    sp = sub.add_parser("feedback", help="list the latest feedback of every run of a pack")
    sp.add_argument("manifest")
    sp = sub.add_parser("launch"); sp.add_argument("manifest")
    sp.add_argument("--timeout", type=float, default=30.0)
    sp = sub.add_parser("show"); sp.add_argument("manifest")
    sp = sub.add_parser("render"); sp.add_argument("manifest"); sp.add_argument("step")
    sp.add_argument("--param", "-p", action="append", default=[], help="name=value")
    a = ap.parse_args(argv)

    if a.cmd == "ui":
        from . import ui
        srv = ui.serve([Path(r) for r in a.roots], a.port)
        print(f"http://127.0.0.1:{a.port}  (Ctrl-C stops)")
        try:
            srv.serve_forever()
        except KeyboardInterrupt:
            pass
        return 0
    if a.cmd == "seal":
        import json
        from . import run as R
        try:
            state = R.record_seal_state(json.loads(Path(a.record).read_text()))
        except (OSError, ValueError) as e:
            print(f"error: {e}", file=sys.stderr); return 1
        print(state)
        return {"sealed": 0, "unsealed": 2, "open": 2, "broken": 1}[state]
    if a.cmd == "check":
        return _check(a.root)
    if a.cmd == "project":
        return _project(a)
    if a.cmd in ("maturity", "review"):
        pk = P.load(Path(a.manifest))
        if not pk.ok:
            print("\n".join(pk.errors), file=sys.stderr); return 1
        if a.cmd == "review":
            try:
                M.add_review(pk, a.by, a.note)
            except M.ReviewError as e:
                print(f"error: {e}", file=sys.stderr); return 1
            print("review recorded for revision", P.digest(pk)); return 0
        m = M.maturity(pk); ev = m["evidence"]
        print(f"{pk.id}: {m['level']}" + (f" (was {m['stale_from']})" if m["stale_from"] else "") + f"  revision {m['digest']}")
        print(f"  reviewed: {ev['reviewed']}  clean runs: {ev['clean_runs']} (rehearsal {ev['rehearsal_runs']})  "
              f"environments: {', '.join(ev['environments']) or '-'}  operators: {', '.join(ev['operators']) or '-'}")
        return 0
    if a.cmd == "score":
        from . import run as R
        for p in _packs(a.root):
            if p.ok and p.data.get("engine") == "native" and p.spec == "runbook-pack/v0.1":
                c = R.scorecard(p)
                print(f"{p.id}\ttried {c['tried']}\tsucceeded {c['succeeded']}\tfailed {c['failed']}\tabandoned {c['abandoned']}\tstalled {c['stalled']}\t[{M.maturity(p)['level']}]")
        return 0
    if a.cmd in ("list", "validate"):
        packs = _packs(a.root)
        if not packs:
            print("no packs found", file=sys.stderr)
            return 1
        bad = 0
        for p in packs:
            d = p.data
            if a.cmd == "list":
                level = M.maturity(p)["level"] if p.ok and d.get("engine") == "native" else "-"
                print(f"{p.id}\t{d.get('engine')}\t{level}\t{d.get('title')}")
            else:
                print(f"{'ok  ' if p.ok else 'FAIL'} {p.id}")
                for e in p.errors:
                    print(f"     - {e}")
            bad += not p.ok
        return 1 if bad and a.cmd == "validate" else 0
    pk = P.load(Path(a.manifest))
    if not pk.ok:
        print("\n".join(pk.errors), file=sys.stderr)
        return 1
    if a.cmd == "feedback":
        from . import run as R
        for f in R.all_feedback(pk):
            print(f"{f['at']}  {R.RATINGS[f['rating']]}  {f['operator']}  {f['run_id']}  {f['comment']}")
        return 0
    if a.cmd == "run":
        return _run(pk, a)
    if a.cmd == "launch":
        from . import legacy
        try:
            proc = legacy.launch(pk, a.timeout)
        except (ValueError, TimeoutError) as e:
            print(f"error: {e}", file=sys.stderr)
            return 1
        print(f"running: {pk.data['legacy'].get('url', '(no url)')}  (Ctrl-C stops it)")
        try:
            return proc.wait()
        except KeyboardInterrupt:
            proc.terminate()
            return 0
    if a.cmd == "show":
        d = pk.data
        print(f"{d['title']}  [{M.maturity(pk)['level']}]  engine={d['engine']}")
        for i, s in enumerate(d.get("steps", []), 1):
            print(f"{i}. [{s['owner']}] {s['title']}  ({s.get('risk', 'read-only')})")
            print(f"     done when: {s['verify']['done_when']}")
        return 0
    values = dict(kv.split("=", 1) for kv in a.param)
    step = next((s for s in pk.data.get("steps", []) if s["id"] == a.step), None)
    if not step or "command" not in step:
        print("no such step or step has no command", file=sys.stderr)
        return 1
    try:
        print(P.render(step["command"], values, pk))
    except (KeyError, ValueError) as e:
        print(f"error: {e}", file=sys.stderr)
        return 1
    return 0


def cli() -> None:
    """Console-script entry point (`rtut`)."""
    sys.exit(main())


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