tegwick
6d70f4d97d
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
492 lines
19 KiB
Python
492 lines
19 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, digest, render
|
|
|
|
STATES = ("todo", "running", "done", "failed", "skipped")
|
|
VERIFY_TIMEOUT = 60
|
|
|
|
|
|
class RunError(Exception):
|
|
pass
|
|
|
|
|
|
def _now() -> str:
|
|
return datetime.now(timezone.utc).isoformat(timespec="microseconds") # sortable even within one second
|
|
|
|
|
|
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:
|
|
"""Write the record. A closed run is locked: nothing may change."""
|
|
disk = _read_json(self.path)
|
|
if disk and disk.get("outcome"):
|
|
keys = set(disk) | set(self.data)
|
|
changed = {k for k in keys if disk.get(k) != self.data.get(k)} - {"updated"}
|
|
if changed:
|
|
raise RunError(f"this run is closed and locked ({disk['outcome']}); it cannot be changed "
|
|
f"({', '.join(sorted(changed))})")
|
|
self.data["updated"] = _now()
|
|
self.path.parent.mkdir(parents=True, exist_ok=True)
|
|
if self.path.exists():
|
|
self.path.chmod(0o644) # a sealed file is read-only; only the closing write reaches here
|
|
self.path.write_text(json.dumps(self.data, indent=2, sort_keys=True) + "\n")
|
|
if self.data.get("outcome"):
|
|
self.path.chmod(0o444)
|
|
|
|
def step_def(self, step_id: str) -> dict:
|
|
if self.pack.spec == "runbook-pack/v0.2": # steps live in stages; alternatives are applied
|
|
from . import run2
|
|
return run2.step_def(self, step_id)
|
|
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 _writable(run: Run) -> None:
|
|
if run.data.get("outcome"):
|
|
raise RunError(f"run is finished (outcome: {run.data['outcome']}); start a new run")
|
|
|
|
|
|
ENVIRONMENTS = ("rehearsal", "staging", "production")
|
|
|
|
|
|
def _context(ctx: dict | None) -> dict:
|
|
"""Where a run applies: project id, target and environment class. Non-secret; all optional."""
|
|
ctx = dict(ctx or {})
|
|
if ctx.get("environment") and ctx["environment"] not in ENVIRONMENTS:
|
|
raise RunError(f"environment must be one of {', '.join(ENVIRONMENTS)}")
|
|
return {k: _clean(str(v), f"context {k}") for k, v in ctx.items() if v and k in ("project", "target", "environment", "project_run")}
|
|
|
|
|
|
def create(pack: Pack, operator: str, params: dict[str, str] | None = None, context: dict | 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.2":
|
|
from . import run2
|
|
return run2.create(pack, operator, params, context=context)
|
|
if pack.spec != "runbook-pack/v0.1":
|
|
raise RunError(f"{pack.spec} packs cannot 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"]},
|
|
"pack_digest": digest(pack),
|
|
"context": _context(context),
|
|
}
|
|
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]:
|
|
"""Run ids of this pack. All packs of a repository share one run folder, so filter by pack id."""
|
|
d = runs_dir(pack)
|
|
if not d.exists():
|
|
return []
|
|
out = []
|
|
for f in sorted(d.glob("*.json")):
|
|
try:
|
|
if json.loads(f.read_text()).get("pack_id") == pack.id:
|
|
out.append(f.stem)
|
|
except (OSError, ValueError):
|
|
continue
|
|
return out
|
|
|
|
|
|
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:
|
|
_writable(run)
|
|
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.
|
|
"""
|
|
_writable(run)
|
|
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, note: str = "") -> None:
|
|
_writable(run)
|
|
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()}
|
|
if note.strip():
|
|
st["note"] = _clean(note.strip(), "note")
|
|
st.update(state="done", finished=_now())
|
|
run.save()
|
|
|
|
|
|
def fail(run: Run, step_id: str, note: str) -> None:
|
|
_writable(run)
|
|
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:
|
|
_writable(run)
|
|
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:
|
|
_writable(run)
|
|
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()
|
|
|
|
|
|
RATINGS = {"good": ":)", "neutral": ":|", "poor": ":("}
|
|
|
|
|
|
def _feedback_entry(rating: str, comment: str) -> dict:
|
|
if rating not in RATINGS:
|
|
raise RunError(f"rating must be one of {', '.join(RATINGS)}")
|
|
return {"rating": rating, "comment": _clean(comment.strip(), "comment"),
|
|
"at": datetime.now(timezone.utc).isoformat(timespec="microseconds")} # sortable even within one second
|
|
|
|
|
|
def add_feedback(run: Run, rating: str, comment: str = "") -> dict:
|
|
"""Record how the runbook felt. Possible while the run is open (the latest entry counts); a closed run is locked,
|
|
so give the feedback before closing, or with the closing form."""
|
|
if run.data.get("outcome"):
|
|
raise RunError("this run is closed and locked: its feedback can no longer be changed")
|
|
entry = _feedback_entry(rating, comment)
|
|
run.data.setdefault("feedback", []).append(entry)
|
|
run.save()
|
|
return entry
|
|
|
|
|
|
def latest_feedback(run: Run) -> dict | None:
|
|
fb = run.data.get("feedback") or []
|
|
return fb[-1] if fb else None
|
|
|
|
|
|
def all_feedback(pack: Pack) -> list[dict]:
|
|
"""Latest feedback of every run of this pack, newest first."""
|
|
out = []
|
|
for rid in list_runs(pack):
|
|
r = load(pack, rid)
|
|
fb = latest_feedback(r)
|
|
if fb:
|
|
out.append({"run_id": rid, "operator": r.data["operator"], **fb})
|
|
return sorted(out, key=lambda x: x["at"], reverse=True)
|
|
|
|
|
|
# ---- captures: optional documentation of what was run and what came back --------------------
|
|
|
|
CAPTURE_MAX_OUTPUT = 20000
|
|
CAPTURE_MAX_COMMAND = 2000
|
|
|
|
|
|
def add_capture(run: Run, step_id: str, output: str, command: str = "", source: str = "operator") -> dict:
|
|
"""Keep the command actually run and its output as documentation or for inspection.
|
|
|
|
The engine cannot detect every secret. It refuses obvious markers and oversized text;
|
|
the operator remains responsible for what is pasted.
|
|
"""
|
|
_writable(run)
|
|
run.step_def(step_id)
|
|
st = run.state(step_id)
|
|
if st["state"] == "todo":
|
|
raise RunError("start the step before capturing its output")
|
|
if len(output) > CAPTURE_MAX_OUTPUT or len(command) > CAPTURE_MAX_COMMAND:
|
|
raise RunError(f"capture too large (output max {CAPTURE_MAX_OUTPUT}, command max {CAPTURE_MAX_COMMAND} characters)")
|
|
if not output.strip() and not command.strip():
|
|
raise RunError("nothing to capture")
|
|
entry = {"command": _clean(command.strip(), "captured command"), "output": _clean(output.rstrip(), "captured output"),
|
|
"source": source, "at": _now()}
|
|
st.setdefault("captures", []).append(entry)
|
|
run.save()
|
|
return entry
|
|
|
|
|
|
def verify_pasted(run: Run, step_id: str, output: str, command: str = "") -> dict:
|
|
"""Check an output-contains step against pasted output, for systems the engine cannot reach."""
|
|
_writable(run)
|
|
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")
|
|
if v.get("expect") != "output-contains":
|
|
raise RunError("only output-contains checks can be verified from pasted output")
|
|
add_capture(run, step_id, output, command, source="pasted-for-verification")
|
|
needle = render(v["contains"], run.data["params"], run.pack)
|
|
result = {"kind": "output-contains", "source": "pasted", "passed": needle in output,
|
|
"output_bytes": len(output.encode()), "at": _now()}
|
|
st["verification"] = result
|
|
run.save()
|
|
return result
|
|
|
|
|
|
# ---- outcome and score ------------------------------------------------------------------------
|
|
|
|
OUTCOMES = ("succeeded", "failed", "abandoned")
|
|
STALE_DAYS = 7
|
|
|
|
|
|
def _seal_of(data: dict) -> str:
|
|
"""Fingerprint of a closed record. `updated` is not part of it, nor is `exercise_recorded`, which records closed
|
|
before exercise recording was retired may still carry."""
|
|
import hashlib
|
|
body = {k: v for k, v in data.items() if k not in ("seal", "updated", "exercise_recorded")}
|
|
return hashlib.sha256(json.dumps(body, sort_keys=True).encode()).hexdigest()
|
|
|
|
|
|
def _read_json(path) -> dict | None:
|
|
try:
|
|
return json.loads(Path(path).read_text())
|
|
except (OSError, ValueError):
|
|
return None
|
|
|
|
|
|
def record_seal_state(data: dict) -> str:
|
|
"""'open' | 'sealed' | 'broken' | 'unsealed' for a run or project-run record, judged from its content alone,
|
|
so an exported record can be checked by anyone (`rtut seal <file>`)."""
|
|
if not data.get("outcome"):
|
|
return "open"
|
|
if not data.get("seal"):
|
|
return "unsealed"
|
|
return "sealed" if data["seal"] == _seal_of(data) else "broken"
|
|
|
|
|
|
def seal_state(run: Run) -> str:
|
|
"""'open' | 'sealed' | 'broken' (changed after closing, outside the app) | 'unsealed' (closed before sealing existed).
|
|
|
|
Judged on what is on disk, not on the in-memory copy."""
|
|
return record_seal_state(_read_json(run.path) or run.data)
|
|
|
|
|
|
def finalize(run: Run, outcome: str, note: str = "", rating: str | None = None, comment: str = "") -> None:
|
|
"""Close a run: its outcome is recorded, its record sealed and locked. An optional rating and comment are
|
|
recorded with it, because a closed run can no longer take feedback.
|
|
|
|
'succeeded' needs every step done; 'failed' needs a failed or pushed-through step."""
|
|
if run.data.get("outcome"):
|
|
raise RunError(f"run already finished ({run.data['outcome']})")
|
|
if outcome not in OUTCOMES:
|
|
raise RunError(f"outcome must be one of {', '.join(OUTCOMES)}")
|
|
states = [v["state"] for v in run.data["steps"].values()]
|
|
if outcome == "succeeded" and any(x != "done" for x in states):
|
|
raise RunError("a run succeeds only when every step is done (skipped steps do not count)")
|
|
if outcome == "succeeded" and any(g["state"] not in ("passed", "waived") for g in run.data.get("gates", {}).values()):
|
|
raise RunError("a run succeeds only when every gate has passed or been waived")
|
|
gstates = [g["state"] for g in run.data.get("gates", {}).values()]
|
|
if outcome == "failed" and not ({"failed", "pushed"} & set(states) or {"failed", "pushed"} & set(gstates)):
|
|
raise RunError("a run can be finished as failed only when a step or gate has failed or was pushed through; otherwise abandon it")
|
|
entry = _feedback_entry(rating, comment) if rating else None # validated before anything changes
|
|
run.data["outcome"] = outcome
|
|
run.data["finished"] = _now()
|
|
if note.strip():
|
|
run.data["outcome_note"] = _clean(note.strip(), "note")
|
|
if entry:
|
|
run.data.setdefault("feedback", []).append(entry)
|
|
run.data["seal"] = _seal_of(run.data)
|
|
run.save()
|
|
|
|
|
|
def effective_outcome(run: Run, now: datetime | None = None, stale_days: int = STALE_DAYS) -> str:
|
|
"""finished outcome, or 'stalled' (unfinished), or 'abandoned' for an unfinished run untouched for stale_days."""
|
|
if run.data.get("outcome"):
|
|
return run.data["outcome"]
|
|
now = now or datetime.now(timezone.utc)
|
|
last = datetime.fromisoformat(run.data.get("updated") or run.data["created"])
|
|
return "abandoned" if (now - last).days >= stale_days else "stalled"
|
|
|
|
|
|
def _ts(s: str) -> datetime:
|
|
return datetime.fromisoformat(s)
|
|
|
|
|
|
def fmt_duration(seconds: float | None) -> str:
|
|
"""Human duration: '45 s', '3 min 20 s', '1 h 5 min', '2 d 3 h'."""
|
|
if seconds is None:
|
|
return "-"
|
|
s = int(round(seconds))
|
|
if s < 60:
|
|
return f"{s} s"
|
|
if s < 3600:
|
|
return f"{s // 60} min {s % 60} s" if s % 60 else f"{s // 60} min"
|
|
if s < 86400:
|
|
return f"{s // 3600} h {(s % 3600) // 60} min" if (s % 3600) // 60 else f"{s // 3600} h"
|
|
return f"{s // 86400} d {(s % 86400) // 3600} h" if (s % 86400) // 3600 else f"{s // 86400} d"
|
|
|
|
|
|
def run_times(run: Run, now: datetime | None = None) -> dict:
|
|
"""Start, finish and durations of one run.
|
|
|
|
elapsed: start to finish (to now while unfinished); working: the time steps were actually in progress;
|
|
step_seconds: per step, from start to finish.
|
|
"""
|
|
now = now or datetime.now(timezone.utc)
|
|
start = _ts(run.data["created"])
|
|
fin = run.data.get("finished")
|
|
end = _ts(fin) if fin else now
|
|
steps = {}
|
|
for sid, st in run.data["steps"].items():
|
|
if st.get("started") and st.get("finished"):
|
|
steps[sid] = max(0.0, (_ts(st["finished"]) - _ts(st["started"])).total_seconds())
|
|
return {"started": run.data["created"], "finished": fin, "elapsed": max(0.0, (end - start).total_seconds()),
|
|
"finished_run": bool(fin), "working": sum(steps.values()), "step_seconds": steps}
|
|
|
|
|
|
def scorecard(pack: Pack, now: datetime | None = None, stale_days: int = STALE_DAYS) -> dict:
|
|
"""How often a runbook was tried and how it went, derived from its run records.
|
|
|
|
avg_seconds / median_seconds / fastest_seconds are the elapsed times of *succeeded* runs only:
|
|
unfinished and failed runs say nothing about how long a working run takes.
|
|
"""
|
|
card = {"tried": 0, "succeeded": 0, "failed": 0, "abandoned": 0, "stalled": 0,
|
|
"feedback": {"good": 0, "neutral": 0, "poor": 0}}
|
|
durations: list[float] = []
|
|
for rid in list_runs(pack):
|
|
r = load(pack, rid)
|
|
card["tried"] += 1
|
|
outcome = effective_outcome(r, now, stale_days)
|
|
card[outcome] += 1
|
|
if outcome == "succeeded" and r.data.get("finished"):
|
|
durations.append(run_times(r)["elapsed"])
|
|
fb = latest_feedback(r)
|
|
if fb:
|
|
card["feedback"][fb["rating"]] += 1
|
|
durations.sort()
|
|
card["avg_seconds"] = sum(durations) / len(durations) if durations else None
|
|
card["median_seconds"] = (durations[len(durations) // 2] if len(durations) % 2 else
|
|
(durations[len(durations) // 2 - 1] + durations[len(durations) // 2]) / 2) if durations else None
|
|
card["fastest_seconds"] = durations[0] if durations else None
|
|
card["timed_runs"] = len(durations)
|
|
return card
|