tegwick
79996ab249
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
232 lines
8 KiB
Python
232 lines
8 KiB
Python
"""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")
|
|
if pack.spec != "runbook-pack/v0.1":
|
|
raise RunError(f"{pack.spec} packs cannot be run yet (RBT-WP-0007)")
|
|
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
|