{body}"
"").encode()
def e(s) -> str:
return html.escape(str(s))
SMILEYS = (("good", "🙂", "Went well"), ("neutral", "😐", "So-so"), ("poor", "🙁", "Poor"))
def score_line(c: dict) -> str:
return (f"tried {c['tried']} · succeeded {c['succeeded']} · failed {c['failed']} · "
f"abandoned {c['abandoned']} · stalled {c['stalled']}")
class App:
def __init__(self, roots: list[Path]):
self.roots = [Path(r) for r in roots]
self.token = secrets.token_urlsafe(16)
self.packs = [P.load(p) for r in roots for p in P.discover(r)]
def registry(self) -> Registry:
"""Fresh view of projects, attribute sets, templates and packs (files may have changed)."""
return Registry(self.roots)
def pack_index(self, pack_id: str) -> int | None:
return next((i for i, p in enumerate(self.packs) if p.id == pack_id), None)
def hidden(self) -> str:
return f""
def index(self) -> bytes:
rows = []
for i, p in enumerate(self.packs):
d = p.data
if not p.ok:
rows.append(f"
" if rows else ""
def capture_html(self, url: str, s: dict, st: dict, state: str, finished: bool) -> str:
saved = ""
for c in st.get("captures", []):
head = f"$ {c['command']}\n" if c.get("command") else ""
saved += f"
{e(head + c['output'][:4000])}
captured {e(c['at'][:19])} ({e(c['source'])})
"
if finished or state == "todo":
return f"Captured output{saved}" if saved else ""
paste = ""
if state == "running" and s["verify"].get("expect") == "output-contains":
paste = f""
return (f"Capture command and output (optional)"
f"{' - ' + str(len(st.get('captures', []))) + ' saved' if st.get('captures') else ''}{saved}"
f"")
def close_run_form(self, i: int, rid: str, r, p) -> str:
outcome = r.data.get("outcome")
if outcome:
note = f" - {e(r.data['outcome_note'])}" if r.data.get("outcome_note") else ""
box = f"
Run finished: {e(outcome)}{note}
"
if outcome == "succeeded":
if r.data.get("exercise_recorded"):
box += f"
Pack recorded as {e(r.data['exercise_recorded'])}. Commit the change in the owning repository.
"
else:
old, new = R.exercise_update(r)
line = new.split("exercise_status:", 1)[1].split("\n", 1)[0].strip()
box += (f"
Current pack status: {e(p.data['exercise_status'])}. "
f"Recording this run would set it to {e(line)} in {e(p.path)} "
"(a reviewable edit; commit it in the owning repository).
"
f"")
return f"
Close this run
{box}
"
states = [v["state"] for v in r.data["steps"].values()]
btns = []
if r.complete and all(x == "done" for x in states):
btns.append("")
if "failed" in states:
btns.append("")
btns.append("")
return (f"
Close this run
"
"
Finishing records the outcome in the pack's score. A finished run is read-only. "
"Unfinished runs count as stalled, and as abandoned after a week.
"
f"
")
def feedback_form(self, i: int, rid: str, r) -> str:
fb = R.latest_feedback(r)
btns = " ".join(f""
for k, face, t in SMILEYS)
form = (f"")
if fb: # collected: close the section, suggest the next step
face = next(f for k, f, _ in SMILEYS if k == fb["rating"])
return ("
Thank you. Recorded: "
f"{face} {e(fb['comment'] or '(no comment)')}
"
def run_page(self, i: int, rid: str, msg: str = "", query: dict | None = None) -> bytes:
p = self.packs[i]; r = R.load(p, rid); d = p.data
if p.spec == "runbook-pack/v0.2":
return UI2.run_page(self, i, rid, msg, query)
ctx = r.data.get("context") or {}
where = ""
if ctx.get("project"):
where = (f" · project {e(ctx['project'])}"
f"{' @' + e(ctx['target']) if ctx.get('target') else ''} · {e(ctx.get('environment', ''))}")
out = [f"
{e(d['title'])}
{e(d['exercise_status'])}
"
f"
run {e(rid)} · operator {e(r.data['operator'])}{where}
"]
if msg:
out.append(f"
{e(msg)}
")
blocked = False
waiting_for = ""
finished = bool(r.data.get("outcome"))
total = len(d["steps"])
done_n = sum(1 for x in d["steps"] if r.state(x["id"])["state"] in ("done", "skipped"))
out.append(f"
{done_n} of {total} steps done
"
f"")
current = next((x["id"] for x in d["steps"] if r.state(x["id"])["state"] == "running"), None)
if current is None:
current = next((x["id"] for x in d["steps"] if r.state(x["id"])["state"] in ("todo", "failed")), None)
for n, s in enumerate(d["steps"], 1):
st = r.state(s["id"]); state = st["state"]
act = ""
cmd = R.command_for(r, s["id"])
block = ""
if cmd and state == "running": # the command and Copy appear only once the step is started
block = f"
{e(cmd)}
"
elif cmd and state == "done":
block = f"
{e(cmd)}
" # kept as a record; no Copy once done
url = f"/run/{i}/{e(rid)}/{e(s['id'])}"
v = st.get("verification")
if state == "todo" and not blocked:
act = f""
elif state == "todo":
act = f"
Waiting for: {e(waiting_for)}
"
elif state == "running":
ver = s["verify"]
if ver.get("expect", "manual") == "manual":
act = (f"")
else:
act = (f"")
if not v:
act = "
1. Run the command, then check → 2. Mark done to lock it in
" + act
elif v["passed"]:
act += "
✔ Check passed. Press Mark done to lock it in; the next step opens after that.
"
else:
act += "
✘ Check not passed. Fix the cause, then check again, or mark the step failed.
"
ev = ""
if v and v["passed"]:
ev = "".join(f" " for x in s.get("evidence", []))
ev = f"
{ev}
" if ev else ""
done_btn = f" " if v and v["passed"] else ""
act += (f"")
elif state == "failed":
act = f"
Failed: {e(st.get('note') or '(no comment)')}
"
elif state == "done" and st.get("note"):
act = f"
Comment: {e(st['note'])}
"
if finished and state != "done":
act = "" # a finished run is read-only
if state not in ("done", "skipped") and not blocked:
blocked = True
waiting_for = f"step {n}, {s['title']}"
rb = f"
Rollback: {e(s['rollback'])}
" if s.get("rollback") and state in ("running", "failed") else ""
anchor = " id=current" if s["id"] == current else ""
fold = "" if state in ("done", "skipped") else " open" # finished steps start folded; the triangle toggles
out.append(f""
f"{n}. {e(s['title'])} "
f"[{e(s['owner'])}] {e(s.get('risk', 'read-only'))}"
f"{state}{block}"
f"{self.capture_html(url, s, st, state, finished)}"
f"
Done when: {e(s['verify']['done_when'])}
{rb}{act}")
if r.complete:
out.append("
All steps done. To record this run as exercising the pack, review and apply: "
f"python -m rtut run finish {e(p.path)} {e(rid)} --apply
")
return page(d["title"], "".join(out))
def make_handler(app: App, port_holder: list):
class H(BaseHTTPRequestHandler):
def log_message(self, *a): # quiet
pass
def _send(self, body: bytes, code=200):
self.send_response(code)
self.send_header("Content-Type", "text/html; charset=utf-8")
self.send_header("Content-Security-Policy", "default-src 'none'; style-src 'unsafe-inline'; script-src 'unsafe-inline'; form-action 'self'")
self.end_headers(); self.wfile.write(body)
def _host_ok(self) -> bool:
return (self.headers.get("Host", "").split(":")[0] in ("127.0.0.1", "localhost"))
def do_GET(self):
if not self._host_ok():
return self._send(b"forbidden", 403)
parts = [x for x in urlparse(self.path).path.split("/") if x]
try:
if not parts:
return self._send(app.index())
if parts[0] == "pack":
return self._send(app.pack_page(int(parts[1])))
if parts[0] == "run":
return self._send(app.run_page(int(parts[1]), parts[2], query=parse_qs(urlparse(self.path).query)))
if parts[0] == "project" and len(parts) == 2:
body = UP.project_page(app, parts[1])
if body:
return self._send(body)
if parts[0] == "template" and len(parts) == 2:
body = UP.template_page(app, parts[1])
if body:
return self._send(body)
except (IndexError, ValueError, R.RunError):
pass
self._send(page("Not found", "
Not found.
"), 404)
def do_POST(self):
if not self._host_ok():
return self._send(b"forbidden", 403)
n = int(self.headers.get("Content-Length", 0))
form = {k: v[0] for k, v in parse_qs(self.rfile.read(n).decode()).items()}
if form.get("_t") != app.token:
return self._send(b"bad token", 403)
parts = [x for x in urlparse(self.path).path.split("/") if x]
msg = ""
try:
if parts[0] == "run" and len(parts) >= 4 and parts[1].isdigit() and app.packs[int(parts[1])].spec == "runbook-pack/v0.2":
return self._redirect(UI2.handle_post(app, int(parts[1]), parts[2], parts[3:], form))
if len(parts) == 3 and parts[0] == "pack" and parts[2] == "review":
from . import maturity as M
i = int(parts[1])
try:
M.add_review(app.packs[i], form.get("by", ""), form.get("note", ""))
except M.ReviewError as ex:
raise ValueError(str(ex))
return self._redirect(f"/pack/{i}#maturity")
if len(parts) == 3 and parts[0] == "project" and parts[2] in ("attributes", "run"):
from . import project as J
reg = app.registry(); doc = reg.projects[parts[1]]
if parts[2] == "attributes":
J.set_attributes(reg, doc, {k[2:]: v for k, v in form.items() if k.startswith("a_")})
return self._redirect(f"/project/{parts[1]}")
r = J.start_run(reg, doc, form.get("pack", ""), form.get("operator", ""), form.get("target") or None, form.get("mode", "manual"))
i = app.pack_index(form.get("pack", ""))
return self._redirect(f"/run/{i}/{r.data['run_id']}#current")
if len(parts) == 3 and parts[0] == "template" and parts[2] == "new":
pid = UP.create_from_template(app, parts[1], form)
return self._redirect(f"/project/{pid}")
if parts[:2] == ["run", "new"]:
i = int(parts[2]); p = app.packs[i]
if p.spec == "runbook-pack/v0.2":
r = UI2.create_run(app, i, form)
return self._redirect(f"/run/{i}/{r.data['run_id']}#current")
params = {k[2:]: v for k, v in form.items() if k.startswith("p_")}
r = R.create(p, form.get("operator", ""), params)
return self._redirect(f"/run/{i}/{r.data['run_id']}#current")
if len(parts) == 4 and parts[0] == "run" and parts[3] == "feedback":
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"):
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", ""))
else:
R.apply_exercise(r)
app.packs[i] = P.load(app.packs[i].path) # pick up the new exercise status
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)
if action == "start": R.start(r, sid)
elif action == "verify":
R.verify(r, sid, confirm=bool(form.get("yes")), manual_ok=(form.get("ok") == "1") if not form.get("yes") else None)
elif action == "done":
R.complete(r, sid, {k[2:]: v for k, v in form.items() if k.startswith("e_")}, form.get("note", ""))
elif action == "fail": R.fail(r, sid, form.get("note", ""))
elif action == "capture": R.add_capture(r, sid, form.get("output", ""), form.get("command", ""))
elif action == "verify-pasted": R.verify_pasted(r, sid, form.get("output", ""), form.get("command", ""))
elif action == "retry": R.retry(r, sid)
return self._redirect(f"/run/{i}/{rid}#current")
except (R.RunError, KeyError, ValueError, IndexError) as ex:
msg = str(ex)
try:
if parts[0] == "project":
return self._send(UP.project_page(app, parts[1], msg) or page("Error", f"
{e(msg)}
"), 400)
if parts[0] == "template":
return self._send(UP.template_page(app, parts[1], msg) or page("Error", f"
{e(msg)}
"), 400)
if parts[0] == "pack":
return self._send(app.pack_page(int(parts[1])).replace(b"
", f"
{e(msg)}
".encode(), 1), 400)
if parts[:2] == ["run", "new"]:
return self._send(app.pack_page(int(parts[2])).replace(b"