runbook-tutorials/rtut/run.py

"""Run records (runbook-run/v0.1): non-secret progress and evidence for one pack run."""
from __future__ import annotations

import json
import re
import subprocess
import time
import uuid
from dataclasses import dataclass
from datetime import date, datetime, timezone
from pathlib import Path

from .legacy import repo_root
from .pack import PLACEHOLDER, SECRET_RE, Pack, render

STATES = ("todo", "running", "done", "failed", "skipped")
VERIFY_TIMEOUT = 60


class RunError(Exception):
    pass


def _now() -> str:
    return datetime.now(timezone.utc).isoformat(timespec="seconds")


def _clean(value: str, what: str) -> str:
    if SECRET_RE.search(value):
        raise RunError(f"{what} contains a secret-looking marker; not recorded")
    return value


def runs_dir(pack: Pack) -> Path:
    return repo_root(pack) / ".local" / "runbook-runs"


@dataclass
class Run:
    pack: Pack
    data: dict

    @property
    def path(self) -> Path:
        return runs_dir(self.pack) / f"{self.data['run_id']}.json"

    def save(self) -> None:
        self.data["updated"] = _now()
        self.path.parent.mkdir(parents=True, exist_ok=True)
        self.path.write_text(json.dumps(self.data, indent=2, sort_keys=True) + "\n")

    def step_def(self, step_id: str) -> dict:
        for s in self.pack.data.get("steps", []):
            if s["id"] == step_id:
                return s
        raise RunError(f"no such step: {step_id}")

    def state(self, step_id: str) -> dict:
        return self.data["steps"][step_id]

    @property
    def complete(self) -> bool:
        return all(v["state"] in ("done", "skipped") for v in self.data["steps"].values())


def create(pack: Pack, operator: str, params: dict[str, str] | None = None) -> Run:
    if not pack.ok or pack.data.get("engine") != "native":
        raise RunError("only valid native packs can be run")
    operator = _clean(operator.strip(), "operator")
    if not re.fullmatch(r"\S+", operator):
        raise RunError("operator must be one word (name or handle)")
    values = dict(params or {})
    known = {p["id"] for p in pack.data.get("parameters", [])}
    for k in values:
        if k not in known:
            raise RunError(f"unknown parameter {k}")
    resolved: dict[str, str] = {}
    for p in pack.data.get("parameters", []):
        # render() validates values/pattern/unsafe characters and applies defaults
        try:
            resolved[p["id"]] = render("{{%s}}" % p["id"], values, pack)
        except (KeyError, ValueError) as e:
            raise RunError(str(e)) from e
        _clean(resolved[p["id"]], f"parameter {p['id']}")
    data = {
        "spec": "runbook-run/v0.1",
        "run_id": time.strftime("%Y%m%d-%H%M%S-") + uuid.uuid4().hex[:6],
        "pack_id": pack.id,
        "pack_path": str(pack.path),
        "operator": operator,
        "created": _now(),
        "params": resolved,
        "steps": {s["id"]: {"state": "todo"} for s in pack.data["steps"]},
    }
    run = Run(pack, data)
    run.save()
    return run


def load(pack: Pack, run_id: str) -> Run:
    f = runs_dir(pack) / f"{run_id}.json"
    if not f.exists():
        raise RunError(f"no such run: {run_id}")
    return Run(pack, json.loads(f.read_text()))


def list_runs(pack: Pack) -> list[str]:
    d = runs_dir(pack)
    return sorted(p.stem for p in d.glob("*.json")) if d.exists() else []


def _order(run: Run, step_id: str) -> None:
    for s in run.pack.data["steps"]:
        if s["id"] == step_id:
            return
        if run.state(s["id"])["state"] not in ("done", "skipped"):
            raise RunError(f"step {s['id']} must be done or skipped before {step_id}")


def command_for(run: Run, step_id: str) -> str | None:
    cmd = run.step_def(step_id).get("command")
    return render(cmd, run.data["params"], run.pack) if cmd else None


def start(run: Run, step_id: str) -> None:
    run.step_def(step_id)
    _order(run, step_id)
    st = run.state(step_id)
    if st["state"] in ("done", "skipped"):
        raise RunError(f"step {step_id} already {st['state']}")
    st.update(state="running", started=_now())
    run.save()


def verify(run: Run, step_id: str, confirm: bool = False, manual_ok: bool | None = None) -> dict:
    """Run the step's check. Commands run only when the operator confirms.

    Stored: exit code, duration, output size, and whether 'contains' matched.
    Output text is never stored.
    """
    v = run.step_def(step_id)["verify"]
    st = run.state(step_id)
    if st["state"] != "running":
        raise RunError(f"step {step_id} is not running")
    expect = v.get("expect", "manual")
    if expect == "manual":
        if manual_ok is None:
            raise RunError("manual verification needs the operator's yes/no")
        result = {"kind": "manual", "passed": bool(manual_ok)}
    else:
        if not confirm:
            raise RunError("running a verify command needs operator confirmation")
        cmd = render(v["command"], run.data["params"], run.pack)
        t0 = time.monotonic()
        try:
            cp = subprocess.run(cmd, shell=True, capture_output=True, text=True,
                                timeout=VERIFY_TIMEOUT, cwd=repo_root(run.pack))
            code, out = cp.returncode, cp.stdout + cp.stderr
        except subprocess.TimeoutExpired:
            code, out = -1, ""
        result = {"kind": expect, "exit_code": code, "output_bytes": len(out.encode()),
                  "seconds": round(time.monotonic() - t0, 2)}
        if expect == "exit-0":
            result["passed"] = code == 0
        else:
            needle = render(v["contains"], run.data["params"], run.pack)
            result["passed"] = code == 0 and needle in out
    result["at"] = _now()
    st["verification"] = result
    run.save()
    return result


def complete(run: Run, step_id: str, evidence: dict[str, str] | None = None) -> None:
    sd = run.step_def(step_id)
    st = run.state(step_id)
    if st["state"] != "running":
        raise RunError(f"step {step_id} is not running")
    if not st.get("verification", {}).get("passed"):
        raise RunError(f"step {step_id} has no passing verification")
    evidence = evidence or {}
    missing = [n for n in sd.get("evidence", []) if n not in evidence]
    if missing:
        raise RunError(f"evidence required: {', '.join(missing)}")
    st["evidence"] = {k: _clean(str(v), f"evidence {k}") for k, v in evidence.items()}
    st.update(state="done", finished=_now())
    run.save()


def fail(run: Run, step_id: str, note: str) -> None:
    st = run.state(step_id)
    if st["state"] != "running":
        raise RunError(f"step {step_id} is not running")
    st.update(state="failed", note=_clean(note, "note"), finished=_now())
    run.save()


def retry(run: Run, step_id: str) -> None:
    st = run.state(step_id)
    if st["state"] != "failed":
        raise RunError(f"step {step_id} has not failed")
    st.pop("verification", None)
    st.update(state="running", started=_now())
    run.save()


def skip(run: Run, step_id: str, reason: str) -> None:
    sd = run.step_def(step_id)
    if sd.get("risk", "read-only") != "read-only":
        raise RunError("only read-only steps may be skipped")
    _order(run, step_id)
    run.state(step_id).update(state="skipped", note=_clean(reason, "reason"), finished=_now())
    run.save()


STATUS_LINE = re.compile(r"^exercise_status:.*$", re.M)


def exercise_update(run: Run, today: date | None = None) -> tuple[str, str]:
    """Return (old, new) pack.yaml text with exercised status. Does not write.

    A run counts only if every step is done (skips do not exercise the pack).
    """
    if any(v["state"] != "done" for v in run.data["steps"].values()):
        raise RunError("every step must be done (not skipped) to exercise the pack")
    today = today or date.today()
    old = run.pack.path.read_text()
    new = STATUS_LINE.sub(f"exercise_status: exercised {today.isoformat()} by {run.data['operator']}",
                          old, count=1)
    return old, new