Add run records, step state machine, verification and exercise-status diff (RBT-WP-0003)

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: sonnet
Assistant-Process: 295952@bnt-lap001
Assistant-Session: e93f64ad-516c-46eb-9666-aad8d300c477

rtut/__main__.py

@@ -12,12 +12,66 @@ def _packs(root: str) -> list[P.Pack]:
     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
+        old, new = R.exercise_update(r)
+        import difflib
+        print("".join(difflib.unified_diff(old.splitlines(True), new.splitlines(True), "pack.yaml", "pack.yaml")))
+        if a.apply:
+            pk.path.write_text(new); print("written")
+        return 0
+    except (R.RunError, KeyError, ValueError) as e:
+        print(f"error: {e}", file=sys.stderr)
+        return 1
+
+
 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", "finish"])
+    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("--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.add_argument("--apply", action="store_true", help="finish: write the exercise-status change")
     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")
@@ -45,6 +99,8 @@ def main(argv: list[str] | None = None) -> int:
     if not pk.ok:
         print("\n".join(pk.errors), file=sys.stderr)
         return 1
+    if a.cmd == "run":
+        return _run(pk, a)
     if a.cmd == "launch":
         from . import legacy
         try:

rtut/run.py

@@ -0,0 +1,230 @@
+"""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)
+            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