coulomb/runbook-tutorials
5
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Retire exercise_status: maturity is the only notion; add CLI tests (RBT-WP-0009-T01)

Browse source 
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
This commit is contained in:
tegwick 2026-09-29 21:28:56 +02:00
parent b89e8eaa74
commit 6d3414d8b7
29 changed files with 444 additions and 218 deletions
Download patch file Download diff file Expand all files Collapse all files

21
rtut/__main__.py
View file

@ -5,6 +5,7 @@ import argparse
import sys
from pathlib import Path
from . import maturity as M
from . import pack as P
@ -47,19 +48,12 @@ def _run(pk, a) -> int:
R.skip(r, a.step, a.note); return 0
if a.action == "finalize":
R.finalize(r, a.outcome or "", a.note)
if r.data["outcome"] == "succeeded" and a.apply:
print("recorded:", R.apply_exercise(r))
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
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
raise ValueError(f"unknown action {a.action}")
except (R.RunError, KeyError, ValueError) as e:
print(f"error: {e}", file=sys.stderr)
return 1
@ -125,7 +119,7 @@ def main(argv: list[str] | None = None) -> int:
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", "feedback", "finalize", "capture"])
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="?")
@ -138,7 +132,6 @@ def main(argv: list[str] | None = None) -> int:
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.add_argument("--apply", action="store_true", help="finish: write the exercise-status change")
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)
@ -182,7 +175,6 @@ def main(argv: list[str] | None = None) -> int:
if a.cmd == "project":
return _project(a)
if a.cmd in ("maturity", "review"):
from . import maturity as M
pk = P.load(Path(a.manifest))
if not pk.ok:
print("\n".join(pk.errors), file=sys.stderr); return 1
@ -202,7 +194,7 @@ def main(argv: list[str] | None = None) -> int:
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[{p.data['exercise_status']}]")
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)
@ -213,7 +205,8 @@ def main(argv: list[str] | None = None) -> int:
for p in packs:
d = p.data
if a.cmd == "list":
print(f"{p.id}\t{d.get('engine')}\t{d.get('exercise_status')}\t{d.get('title')}")
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:
@ -246,7 +239,7 @@ def main(argv: list[str] | None = None) -> int:
return 0
if a.cmd == "show":
d = pk.data
print(f"{d['title']} [{d['exercise_status']}] engine={d['engine']}")
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']}")

2
rtut/pack.py
View file

@ -143,7 +143,7 @@ REVIEW_FILE = "REVIEWS.yaml"
def digest(pack: "Pack") -> str:
"""Revision of a pack: hash of every file in its directory, except review records and local state.
`exercise_status:` lines are ignored, so recording an exercise never changes the revision.
`exercise_status:` lines (a deprecated field) are ignored, so dropping it from a pack is not a new revision.
"""
import hashlib
h = hashlib.sha256()

43
rtut/run.py
View file

@ -44,19 +44,19 @@ class Run:
def path(self) -> Path:
return runs_dir(self.pack) / f"{self.data['run_id']}.json"
def save(self, allow: tuple = ()) -> None:
"""Write the record. A closed run is locked: nothing may change except the keys named in `allow`."""
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"} - set(allow)
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; the allowed update reopens it briefly
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)
@ -270,25 +270,6 @@ def skip(run: Run, step_id: str, reason: str) -> None:
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")
if run.data.get("deviations"):
raise RunError("a run with deviations (overrides, waivers, push-through) cannot 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
RATINGS = {"good": ":)", "neutral": ":|", "poor": ":("}
@ -379,7 +360,8 @@ STALE_DAYS = 7
def _seal_of(data: dict) -> str:
"""Fingerprint of a closed record. Timestamps of allowed later updates and the exercise marker are not part of it."""
"""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()
@ -503,16 +485,3 @@ def scorecard(pack: Pack, now: datetime | None = None, stale_days: int = STALE_D
card["fastest_seconds"] = durations[0] if durations else None
card["timed_runs"] = len(durations)
return card
def apply_exercise(run: Run, today: date | None = None) -> str:
"""Write 'exercised <date> by <operator>' into pack.yaml for a succeeded run. Returns the new status."""
if run.data.get("outcome") != "succeeded":
raise RunError("only a run finished as succeeded can exercise the pack")
_, new = exercise_update(run, today)
run.pack.path.write_text(new)
status = STATUS_LINE.search(new).group(0).split(": ", 1)[1]
run.pack.data["exercise_status"] = status
run.data["exercise_recorded"] = status
run.save(allow=("exercise_recorded",)) # the one thing a closed run may still receive
return status

26
rtut/ui.py
View file

@ -260,7 +260,7 @@ class App:
if x.get("values") else f"<input name='p_{x['id']}' value='{e(x.get('default', ''))}'>")
+ f"</label><br><span class=note>{e(x.get('help', ''))} Never enter secrets.</span></p>"
for x in d.get("parameters", []))
return page(d["title"], f"<h1>{e(d['title'])}</h1><p class=pill>{e(d['exercise_status'])}</p>"
return page(d["title"], f"<h1>{e(d['title'])}</h1><p>{UP.maturity_pill(p)}</p>"
f"<p>{e(d.get('outcome', ''))}</p><p class=note>{e(score_line(R.scorecard(p)))}</p>"
f"{UP.maturity_block(self, i, p)}{self.runs_list(i, p)}<h2>Prerequisites</h2><ul>{prereq}</ul>"
f"<form method=post action='/run/new/{i}'>{self.hidden()}{fields}"
@ -302,17 +302,6 @@ class App:
if outcome:
note = f" - {e(r.data['outcome_note'])}" if r.data.get("outcome_note") else ""
box = f"<p><b>Run finished: {e(outcome)}</b>{note}</p>{self.locked_note(r)}"
if outcome == "succeeded":
if r.data.get("exercise_recorded"):
box += f"<p>Pack recorded as <b>{e(r.data['exercise_recorded'])}</b>. Commit the change in the owning repository.</p>"
else:
old, new = R.exercise_update(r)
line = new.split("exercise_status:", 1)[1].split("\n", 1)[0].strip()
box += (f"<p>Current pack status: <span class=pill>{e(p.data['exercise_status'])}</span>. "
f"Recording this run would set it to <b>{e(line)}</b> in <code>{e(p.path)}</code> "
"(a reviewable edit; commit it in the owning repository).</p>"
f"<form method=post action='/run/{i}/{e(rid)}/exercise'>{self.hidden()}"
"<button>Record pack as exercised</button></form>")
return f"<div id=close class=step><h2>Close this run</h2>{box}</div>"
states = [v["state"] for v in r.data["steps"].values()]
btns = []
@ -383,7 +372,7 @@ class App:
where = (f" · project <a href='/project/{e(ctx['project'])}'>{e(ctx['project'])}</a>"
f"{' @' + e(ctx['target']) if ctx.get('target') else ''} · {e(ctx.get('environment', ''))}"
f"{' · part of <a href=' + chr(39) + '/project-run/' + e(ctx['project_run']) + chr(39) + '>project run ' + e(ctx['project_run']) + '</a>' if ctx.get('project_run') else ''}")
out = [f"<h1>{e(d['title'])}</h1><p class=pill>{e(d['exercise_status'])}</p>"
out = [f"<h1>{e(d['title'])}</h1><p>{UP.maturity_pill(p)}</p>"
f"<p class=note>run {e(rid)} · operator {e(r.data['operator'])}{where}</p>"
f"<p class=note>{UI2.times_line(r)}</p>"]
step_secs = R.run_times(r)["step_seconds"]
@ -607,15 +596,10 @@ def make_handler(app: App, port_holder: list):
i, rid = int(parts[1]), parts[2]
R.add_feedback(R.load(app.packs[i], rid), form.get("rating", ""), form.get("comment", ""))
return self._redirect(f"/run/{i}/{rid}#feedback")
if len(parts) == 4 and parts[0] == "run" and parts[3] in ("finalize", "exercise"):
if len(parts) == 4 and parts[0] == "run" and parts[3] == "finalize":
i, rid = int(parts[1]), parts[2]
r = R.load(app.packs[i], rid)
if parts[3] == "finalize":
R.finalize(r, form.get("outcome", ""), form.get("note", ""), rating=form.get("rating") or None,
comment=form.get("fb_comment", ""))
else:
R.apply_exercise(r)
app.packs[i] = P.load(app.packs[i].path) # pick up the new exercise status
R.finalize(R.load(app.packs[i], rid), form.get("outcome", ""), form.get("note", ""),
rating=form.get("rating") or None, comment=form.get("fb_comment", ""))
return self._redirect(f"/run/{i}/{rid}#close")
i, rid, sid, action = int(parts[1]), parts[2], parts[3], parts[4]
p = app.packs[i]; r = R.load(p, rid)