"""UI for runbook-pack/v0.2 runs.
Three views of one run:
flow the guided questionnaire: stages, the recommended next action, steps, gates, questions (main)
data every datapoint, grouped or filtered by dimension, editable independently of the flow
overview progress, gate states, datapoint states, deviations, mode history
Server-rendered forms; the only script is the small one in ui.page().
"""
from __future__ import annotations
import html
from urllib.parse import quote
from . import artefacts as A, run as R1, run2 as R
from .flow import Flow
DONE = ("done", "skipped", "pushed")
def e(s) -> str:
return html.escape(str(s))
def sp(state: str) -> str:
return f"{e(state)}"
# ---- starting a run -------------------------------------------------------------------------------------------
def start_form(app, i: int, pack) -> str:
fl = Flow(pack.data)
fields = ""
for d in fl.datapoints.values():
if d.get("produced_by"):
continue
name, default = f"p_{d['id']}", d.get("default", "")
if d.get("values"):
ctl = f""
elif d["type"] == "bool":
ctl = f""
else:
ctl = f""
dims = " ".join(f"{e(k)}: {e(v)}" for k, v in (d.get("dimensions") or {}).items())
fields += f"
{dims} {e(d.get('help', ''))} Never enter secrets.
"
irrev = [s for s in fl.steps.values() if not s.get("reversible", True)]
pre = ""
if irrev:
pre = ("")
return (f"")
def create_run(app, i: int, form: dict):
pack = app.packs[i]
params = {k[2:]: v for k, v in form.items() if k.startswith("p_") and v != ""}
pre = [k[4:] for k, v in form.items() if k.startswith("pre_")]
return R.create(pack, form.get("operator", ""), params, mode=form.get("mode", "manual"), preauthorize=pre)
# ---- context -------------------------------------------------------------------------------------------------------
class Ctx:
def __init__(self, app, i: int, rid: str, run, view: str = "flow", query: dict | None = None):
self.app, self.i, self.rid, self.run, self.view = app, i, rid, run, view
self.q = query or {}
self.fl = R.flow(run)
self.base = f"/run/{i}/{rid}"
self.na = R.next_actions(run)
self.num = {sid: k + 1 for k, sid in enumerate(x for x in self.fl.order if self.fl.kind(x) == "step")}
self.manual = run.data["mode"] == "manual"
self.finished = bool(run.data.get("outcome"))
def h(self) -> str:
return self.app.hidden()
# ---- page ---------------------------------------------------------------------------------------------------------------
def run_page(app, i: int, rid: str, msg: str = "", query: dict | None = None) -> bytes:
from .ui import page
pack = app.packs[i]
run = R1.load(pack, rid)
q = {k: (v[0] if isinstance(v, list) else v) for k, v in (query or {}).items()}
view = q.get("view", "flow") if q.get("view") in ("flow", "data", "overview") else "flow"
c = Ctx(app, i, rid, run, view, q)
d = run.data
ctx = d.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', ''))}"
f"{' · part of project run ' + e(ctx['project_run']) + '' if ctx.get('project_run') else ''}")
ov = R.overview(run)
prog = ov["progress"]
out = [f"
{prog['done']} of {prog['total']} done"
f"{' · ' + str(prog['pushed']) + ' pushed through' if prog['pushed'] else ''}
"
f""]
if d["deviations"]:
out.append(f"
⚑ {len(d['deviations'])} deviation(s) recorded: this run can be finished but will never count towards "
"maturity. See them.
")
if msg:
out.append(f"
{e(msg)}
")
tabs = " ".join(f"{label}"
for v, label in (("flow", "Guided flow"), ("data", "Data"), ("overview", "Overview")))
out.append(f"")
if view == "flow":
out.append(_flow_view(c, ov))
elif view == "data":
out.append(_data_view(c))
else:
out.append(_overview_view(c, ov))
return page(pack.data["title"], "".join(out))
def stamp(iso: str | None) -> str:
return iso[:16].replace("T", " ") + " UTC" if iso else "-"
def times_line(run) -> str:
"""Started, finished and how long: the run's timestamps at a glance."""
t = R1.run_times(run)
if t["finished_run"]:
return (f"Started {e(stamp(t['started']))} · finished {e(stamp(t['finished']))} · took {e(R1.fmt_duration(t['elapsed']))}"
f"{' · working ' + e(R1.fmt_duration(t['working'])) if t['working'] else ''}")
return (f"Started {e(stamp(t['started']))} · running for {e(R1.fmt_duration(t['elapsed']))}"
f"{' · working ' + e(R1.fmt_duration(t['working'])) if t['working'] else ''}")
# ---- flow view ---------------------------------------------------------------------------------------------------------
def _flow_view(c: Ctx, ov: dict) -> str:
run, fl = c.run, c.fl
out = []
framing = list(run.pack.data.get("framing", []))
if framing:
out.append("Keep in mind
" + "".join(f"
{e(x)}
" for x in framing) + "
")
out.append(_mode_panel(c, ov))
out.append(_next_banner(c))
for st in fl.data["stages"]:
out.append(_stage(c, st))
if c.finished or _all_complete(run) or R1.latest_feedback(run):
out.append(c.app.feedback_form(c.i, c.rid, run))
out.append(_close(c))
return "".join(out)
def _all_complete(run) -> bool:
fl = R.flow(run)
return all(R.is_complete(run, x) for x in fl.order)
def _mode_panel(c: Ctx, ov: dict) -> str:
run = c.run
if c.finished:
return ""
if c.manual:
return (f"
Manual mode.You confirm every check and gate. Unmet preconditions can be "
f"overridden with a reason; each override is recorded."
f"
")
ap = run.data["auto_confirm"]
last = ap.get("last")
status = ""
if ap.get("stopped"):
status = f"
Stopped at {e(ap['stopped']['item'])}: {e(ap['stopped']['reason'])}
Auto-confirm mode.The engine confirms when preconditions hold. It never "
"runs commands: do the work yourself, then check again. Irreversible steps wait for you unless pre-authorized"
f"{' (' + e(', '.join(pre)) + ')' if pre else ''}.{status}"
f""
f"
")
def _title_of(c: Ctx, id_: str) -> str:
fl = c.fl
node = fl.node(id_) if fl.kind(id_) != "?" else {}
return node.get("title", id_)
def _next_banner(c: Ctx) -> str:
na = c.na
if c.finished:
return ""
rec = na["recommended"]
if rec is None:
if _all_complete(c.run):
return "
Everything is done. Close the run below.
"
b = na["blocked"][0] if na["blocked"] else None
return f"
Nothing can start. {e(_title_of(c, b['id'])) if b else ''} waits: {e('; '.join(b['unmet'])) if b else ''}
"
kind = c.fl.kind(rec)
others = [x for x in na["ready"] + na["running"] if x != rec]
links = " ".join(f"{e(_title_of(c, x))}" for x in others)
verb = "Continue" if rec in na["running"] else ("Evaluate gate" if kind == "gate" else "Next")
return (f"
{verb}:{e(_title_of(c, rec))}(recommended)"
f"{' Also possible in parallel: ' + links if links else ''}
")
def _stage(c: Ctx, st: dict) -> str:
run, fl = c.run, c.fl
done = R.is_complete(run, st["id"])
members = fl.members(st["id"], include_gates=False)
n_done = sum(run.data["steps"][m]["state"] in DONE for m in members)
body = [f"
Goal: {e(st['goal'])}
"]
if st.get("framing"):
body.append("
" + "".join(f"
{e(x)}
" for x in st["framing"]) + "
")
q = _questions(c, st)
if q:
body.append(q)
for s in st.get("steps", []):
body.append(_step(c, s))
for child in st.get("stages", []):
body.append(_stage(c, child))
for g in st.get("gates", []):
body.append(_gate(c, g))
return (f""
f"{e(st['title'])}{n_done}/{len(members)} steps"
f"{sp('complete' if done else 'in progress')}{''.join(body)}")
def _questions(c: Ctx, st: dict) -> str:
dps = [d for d in c.fl.datapoints.values() if d.get("ask_in") == st["id"] and not d.get("produced_by")]
if not dps or c.finished:
return ""
rows = []
locked = False
for d in dps:
rec = c.run.data["datapoints"][d["id"]]
users = R.datapoint_users(c.run, d["id"])
locked = locked or bool(users and rec.get("value") is not None)
rows.append(f"
Questions for this stage(values used by its steps; never enter secrets)"
f"
")
def _override_fields(c: Ctx, locked: bool) -> str:
"""Only shown when some value is already in use by started steps: changing it is a recorded deviation."""
if not locked:
return ""
if not c.manual:
return "
Values already in use by steps can only be changed in manual mode: take over first.
"
return ("Changing a value that steps already used (override)"
"
Risk: steps that already ran used the previous value, so their results may no longer match the new one. "
"The change is recorded as a deviation.
"
""
"")
def _input(d: dict, rec: dict, editable: bool = True, users: list[str] | None = None) -> str:
name = f"d_{d['id']}"
val = rec.get("value", "")
dis = "" if editable else " disabled"
if d.get("values"):
ctl = f""
elif d["type"] == "bool":
ctl = (f"")
else:
ctl = f""
err = f" {e(rec.get('error', ''))}" if rec["state"] == "err" else ""
use = (f" in use by {e(', '.join(users))}"
if users and rec.get("value") is not None else "")
return f"{err}"
# ---- steps ------------------------------------------------------------------------------------------------------------------
def _step(c: Ctx, s: dict) -> str:
run, fl = c.run, c.fl
sid = s["id"]
st = run.data["steps"][sid]
state = st["state"]
eff = R.step_def(run, sid)
n = c.num[sid]
url = f"{c.base}/{e(sid)}"
marks = []
if s.get("idempotent") or eff.get("idempotent"):
marks.append(f"{e(eff.get('idempotent') or s.get('idempotent'))}")
if not eff.get("reversible", True) or any(not a.get("reversible", s.get("reversible", True)) for a in s.get("one_of", [])):
marks.append("irreversible")
if s.get("auto_confirm") == "manual-only":
marks.append("manual-only")
took = R1.run_times(run)["step_seconds"].get(sid)
if took is not None and state in DONE:
marks.append(f"took {e(R1.fmt_duration(took))}")
title = eff.get("title", s["title"]) if st.get("alternative") else s["title"]
actor = fl.actors.get(s["actor"], {}).get("role", s["actor"])
body = []
if s.get("description"):
body.append(f"
{e(s['description'])}
")
unmet = R.unmet(run, sid) if state == "todo" else []
if state == "todo":
body.append(_start_control(c, s, sid, url, unmet))
elif state == "running":
body.append(_running(c, s, eff, st, sid, url))
elif state == "failed":
idem = eff.get("idempotent") or s.get("idempotent")
ack = ("" if idem == "idempotent" else
" ")
body.append(f"
Failed: {e(st.get('note') or '(no comment)')}
"
f"")
elif state == "pushed":
body.append("
Passed over by push-through without verification: not done, and recorded as a deviation.
")
elif state == "done" and st.get("note"):
body.append(f"
Comment: {e(st['note'])}
")
if state == "done" or state == "pushed":
cmd = _command(run, sid)
if cmd:
body.insert(0, f"
{e(cmd)}
")
body.append(c.app.capture_html(url, eff, st, state, c.finished))
current = " id=current" if sid == c.na["recommended"] else ""
fold = "" if state in DONE else " open"
return (f""
f"{n}. {e(title)}[{e(actor)} · {e(s['owner'])}] {' '.join(marks)}"
f"{sp(state)}{''.join(body)}")
def _command(run, sid: str) -> str | None:
try:
return R.command_for(run, sid)
except (KeyError, ValueError) as ex:
return f"(needs a value that is not set: {ex})"
def _start_control(c: Ctx, s: dict, sid: str, url: str, unmet: list[str]) -> str:
run = c.run
if c.finished:
return ""
alts = s.get("one_of", [])
rec = R.recommended_alt(run, sid) if alts else None
alt_html = ""
if alts:
alt_html = "
Choose one: " + "".join(
f" " for a in alts) + "
"
irrev = (not s.get("reversible", True)) or any(not a.get("reversible", s.get("reversible", True)) for a in alts)
phrase = ""
if irrev:
phrase = (f"
This step is irreversible. Type the confirmation phrase to start it: "
f"{e(R.phrase_for(run, sid))}
")
if not unmet:
if s.get("auto_confirm") == "manual-only" and not c.manual:
return "
Manual-only: auto-confirm will not start this step. Take over to do it yourself.
")
ver = eff["verify"]
exp = ver.get("expect") or (None if ver.get("datapoints_ok") else "manual")
said = ver.get("done_when") or ("datapoints ok: " + ", ".join(ver.get("datapoints_ok", [])))
v = st.get("verification")
if exp == "manual":
out.append(f"")
else:
what = (f"{e(_render_check(run, ver))}" if ver.get("command") else "the datapoint states")
out.append(f"")
if not v:
out.insert(0, "
1. Do the work, then check → 2. Mark done to lock it in
")
elif v["passed"]:
out.append("
✔ Check passed. Press Mark done to lock it in.
")
else:
out.append("
✘ Check not passed. Fix the cause and 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 ""
out.append(f"")
if eff.get("rollback"):
out.append(f"
Rollback: {e(_render_text(run, eff['rollback']))}
")
return "".join(out)
def _render_text(run, text: str) -> str:
try:
return R.render(text, run.data["params"], run.pack)
except (KeyError, ValueError):
return text
def _render_check(run, ver: dict) -> str:
return _render_text(run, ver.get("command", ""))
# ---- gates ---------------------------------------------------------------------------------------------------------------------
def _gate(c: Ctx, g: dict) -> str:
run, fl = c.run, c.fl
gid = g["id"]
gs = run.data["gates"][gid]
state = gs["state"]
unmet = R.unmet(run, gid) if state not in R.DONE_GATE else []
rows = []
for idx, ck in enumerate(g["checks"]):
res = gs["results"].get(str(idx))
exp = ck.get("expect") or (None if ck.get("datapoints_ok") else "manual")
text = ck.get("done_when") or "datapoints ok: " + ", ".join(ck.get("datapoints_ok", []))
pill = sp("passed" if res and res["passed"] else "failed") if res else sp("not checked")
ctl = ""
if state not in R.DONE_GATE and not c.finished:
ctl = _gate_check_form(c, gid, idx, ck, exp, unmet)
rows.append(f"
{pill} {e(text)}{' ' + ctl if ctl else ''}
")
waive = ""
w = g.get("waivable")
if w and state in ("open", "failed") and not c.finished:
role = fl.actors.get(w["by"], {}).get("role", w["by"])
waive = (f"Waive this gate (declared waivable by {e(role)})"
f"")
extra = ""
if state == "pushed":
extra = "
Passed over by push-through without passing: recorded as a deviation.
"
if state == "waived":
extra = f"
Waived: {e(gs['waiver']['reason'])}
"
current = " open" if state not in R.DONE_GATE else ""
marker = " id=current" if gid == c.na["recommended"] else ""
return (f""
f"Gate: {e(g['title'])}quality goal: {e(g['goal'])}"
f"{' · blocking' if g.get('blocking', True) else ''}{sp(state)}"
f"
{''.join(rows)}
{extra}{waive}")
def _gate_check_form(c: Ctx, gid: str, idx: int, ck: dict, exp: str | None, unmet: list[str]) -> str:
run = c.run
action = f"{c.base}/gate/{e(gid)}/check"
hidden = f"{c.h()}"
if exp == "manual":
ctl = " "
else:
cmd = f" {e(_render_check(run, ck))}" if ck.get("command") else ""
ctl = f"{cmd}"
if not unmet:
return f""
if not c.manual:
return f"Waiting for: {e('; '.join(unmet))}"
return (f"Waiting for: {e('; '.join(unmet))}"
f"Check anyway (override)"
f"")
# ---- data view -----------------------------------------------------------------------------------------------------------------
def _data_view(c: Ctx) -> str:
run, fl = c.run, c.fl
group = c.q.get("group") or None
flt = c.q.get("filter") or None
keys = R.dimension_keys(run)
base = "?view=data"
def link(g=None, f=None, label="", cls=""):
qs = base + (f"&group={quote(g)}" if g else "") + (f"&filter={quote(f)}" if f else "")
return f"{e(label)}"
ctl = ("
Group by " + link(None, flt, "none", " on" if not group else "") + " "
+ " ".join(link(k, flt, k, " on" if group == k else "") for k in keys) + "
"
groups = R.datapoint_rows(run, group=group, flt=flt)
out = [f"
Every datapoint of this run, independent of the flow. Tags (dimensions) come from the pack; click one to filter.
{ctl}"]
editable = not c.finished
any_locked = any(R.datapoint_users(run, x["id"]) and x["value"] is not None and x["editable"]
for rows_ in groups.values() for x in rows_)
out.append(f"")
arts = R.artefact_rows(run)
if arts:
out.append("
Artefacts
Artefact
Where it lives
Open
Produced by
Until
" + "".join(
f"
{e(a['title'])}
{e(a['location'] or '(not known yet)')}
"
f"
{_artefact_links(c, a)}
"
f"
{e(a['produced_by'])} {sp(a['produced_state'])}
{e(a['lives_until'] or '-')}
" for a in arts) + "
")
if fl.actors:
counts: dict[str, int] = {}
for s in fl.steps.values():
counts[s["actor"]] = counts.get(s["actor"], 0) + 1
out.append("
Actors
Role
Description
Privileges
Steps
" + "".join(
f"
{e(a['role'])} ({e(a['id'])})
{e(a['description'])}
"
f"
{e(', '.join(a.get('privileges', [])))}
{counts.get(a['id'], 0)}
" for a in fl.actors.values()) + "
")
return "".join(out)
def _artefact_links(c: Ctx, a: dict) -> str:
loc = a["location"]
if loc and A.is_url(loc):
return f"Open link"
path, why = A.check(loc)
if path is None:
return f"{e(why)}"
url = f"{c.base}/artefact/{e(a['id'])}"
return f"Open · Download"
def serve_artefact(app, i: int, rid: str, aid: str) -> tuple[int, bytes, str, str | None]:
"""(status, body, content type, download filename or None). Refusals come back as short plain text."""
run = R1.load(app.packs[i], rid)
if run.pack.spec != "runbook-pack/v0.2":
return 404, b"no artefacts", "text/plain; charset=utf-8", None
row = next((a for a in R.artefact_rows(run) if a["id"] == aid), None)
if row is None:
return 404, b"no such artefact", "text/plain; charset=utf-8", None
path, why = A.check(row["location"])
if path is None:
return 403, f"not served: {why}".encode(), "text/plain; charset=utf-8", None
body, ctype, is_text = A.content(path)
return 200, body, ctype, (None if is_text else path.name)
# ---- overview ------------------------------------------------------------------------------------------------------------------
def _overview_view(c: Ctx, ov: dict) -> str:
rows = ""
for s in ov["stages"]:
gates = " ".join(f"{e(g['title'])} {sp(g['state'])}" for g in s["gates"])
prob = ""
if s["steps_failed"]:
prob += f" {s['steps_failed']} failed"
if s["steps_pushed"]:
prob += f" {s['steps_pushed']} pushed"
if s["steps_running"]:
prob += f" {s['steps_running']} running"
rows += (f"
None. A run without deviations can count towards maturity.
")
if ov["mode_history"]:
out.append("
Mode changes
" + "".join(f"
{e(m['from'])} → {e(m['to'])} · {e(m['at'][:19])}
" for m in ov["mode_history"]) + "
")
if ov["stopped"]:
out.append(f"
Auto-confirm stopped at {e(ov['stopped']['item'])}: {e(ov['stopped']['reason'])}
")
return "".join(out)
# ---- closing the run -----------------------------------------------------------------------------------------------------
def _close(c: Ctx) -> str:
run = c.run
if c.finished:
note = f"
Comment: {e(run.data['outcome_note'])}
" if run.data.get("outcome_note") else ""
return (f"
Run finished: {e(run.data['outcome'])}
{note}{c.app.locked_note(run)}"
"
The outcome is part of this runbook's score and maturity evidence at this revision.
")
fl = c.fl
steps = [run.data["steps"][s]["state"] for s in fl.steps]
gates = [run.data["gates"][g]["state"] for g in fl.gates]
btns = []
if all(x == "done" for x in steps) and all(x in ("passed", "waived") for x in gates):
btns.append("")
if {"failed", "pushed"} & set(steps) or {"failed", "pushed"} & set(gates):
btns.append("")
btns.append("")
return ("
Close this run
"
"
Finishing records the outcome. A run with deviations can be finished but does not count towards maturity. "
"A finished run is read-only.
"
f"
")
# ---- actions -------------------------------------------------------------------------------------------------------------------
def _safe_next(app, form: dict, default: str) -> str:
n = form.get("next", "")
return n if n.startswith("/run/") and "//" not in n and "\n" not in n and "\r" not in n else default
def handle_post(app, i: int, rid: str, parts: list[str], form: dict) -> str:
"""Perform a v0.2 run action; return the URL to redirect to. Raises RunError/ValueError/KeyError on a refused action."""
pack = app.packs[i]
run = R1.load(pack, rid)
base = f"/run/{i}/{rid}"
if parts == ["feedback"]:
R1.add_feedback(run, form.get("rating", ""), form.get("comment", ""))
return f"{base}#feedback"
if parts == ["finalize"]:
R1.finalize(run, form.get("outcome", ""), form.get("note", ""), rating=form.get("rating") or None, comment=form.get("fb_comment", ""))
return f"{base}#close"
if parts == ["mode"]:
R.switch_mode(run, form.get("to", ""))
return base
if parts == ["tick"]:
n = max(1, min(int(form.get("n", "1") or 1), 50))
last = None
for _ in range(n):
last = R.auto_confirm_tick(run)
if last["action"] in ("waiting", "awaiting-operator", "stopped", "finished", "push-through"):
break
run.data["auto_confirm"]["last"] = last
run.save()
return f"{base}#current"
if parts == ["data"]:
bad = []
changes = {}
for k, v in form.items():
if not k.startswith("d_"):
continue
dp = k[2:]
cur = run.data["datapoints"].get(dp)
if cur is None or v == "" or (cur["state"] in ("set", "ok") and cur.get("value") == v):
continue # unchanged or left empty
changes[dp] = v
need = R.changes_needing_override(run, changes) # all or nothing: check before applying any
reason, ack = form.get("override_reason", ""), bool(form.get("acknowledge_risk"))
if need and not (reason.strip() and ack):
raise R1.RunError("already in use by started steps: " + "; ".join(f"{d} ({', '.join(u)})" for d, u in need.items())
+ ". To change them give a reason and acknowledge the risk; each change is recorded as a deviation.")
for dp, v in changes.items():
if R.set_datapoint(run, dp, v, override_reason=reason, acknowledge_risk=ack) == "err":
bad.append(f"{dp} ({run.data['datapoints'][dp]['error']})")
if bad:
raise ValueError("invalid value for: " + ", ".join(bad))
return _safe_next(app, form, base)
if len(parts) == 3 and parts[0] == "gate" and parts[2] in ("check", "waive"):
gid = parts[1]
if parts[2] == "waive":
R.waive(run, gid, form.get("reason", ""))
else:
R.gate_check(run, gid, int(form.get("idx", "-1")), confirm=bool(form.get("yes")),
manual_ok=(form.get("ok") == "1") if not form.get("yes") else None,
override_reason=form.get("override_reason", ""), acknowledge_risk=bool(form.get("acknowledge_risk")))
return f"{base}#item-{gid}"
if len(parts) == 2:
sid, action = parts
if action == "start":
R.start(run, sid, alt=form.get("alt") or None, phrase=form.get("phrase") or None,
override_reason=form.get("override_reason", ""), acknowledge_risk=bool(form.get("acknowledge_risk")))
elif action == "verify":
R.verify(run, sid, confirm=bool(form.get("yes")), manual_ok=(form.get("ok") == "1") if not form.get("yes") else None)
elif action == "done":
R.complete(run, sid, {k[2:]: v for k, v in form.items() if k.startswith("e_")}, form.get("note", ""))
elif action == "fail":
R.fail(run, sid, form.get("note", ""))
elif action == "retry":
R.retry(run, sid, acknowledge_partial=bool(form.get("acknowledge_partial")))
elif action == "capture":
R1.add_capture(run, sid, form.get("output", ""), form.get("command", ""))
elif action == "verify-pasted":
R1.verify_pasted(run, sid, form.get("output", ""), form.get("command", ""))
else:
raise ValueError(f"unknown action {action}")
return f"{base}#current"
raise ValueError("unknown action")