tegwick
f070999aa2
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
634 lines
34 KiB
Python
634 lines
34 KiB
Python
"""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 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"<span class=pill data-s='{e(state)}'>{e(state)}</span>"
|
|
|
|
|
|
# ---- 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"<select name='{name}'>" + "".join(f"<option{' selected' if o == default else ''}>{e(o)}</option>" for o in d["values"]) + "</select>"
|
|
elif d["type"] == "bool":
|
|
ctl = f"<select name='{name}'><option value=''>(not set)</option><option>true</option><option>false</option></select>"
|
|
else:
|
|
ctl = f"<input name='{name}' value='{e(default)}'>"
|
|
dims = " ".join(f"<span class=chip>{e(k)}: {e(v)}</span>" for k, v in (d.get("dimensions") or {}).items())
|
|
fields += f"<p><label>{e(d['label'])}<br>{ctl}</label> {dims}<br><span class=note>{e(d.get('help', ''))} Never enter secrets.</span></p>"
|
|
irrev = [s for s in fl.steps.values() if not s.get("reversible", True)]
|
|
pre = ""
|
|
if irrev:
|
|
pre = ("<fieldset><legend>Pre-authorize irreversible steps (auto-confirm only)</legend>"
|
|
"<p class=note>An auto-confirm run never confirms an irreversible step by itself unless you name it here. Steps marked "
|
|
"manual-only are never automated.</p>"
|
|
+ "".join(f"<label><input type=checkbox name='pre_{e(s['id'])}'> {e(s['title'])}"
|
|
f"{' (manual-only)' if s.get('auto_confirm') == 'manual-only' else ''}</label><br>" for s in irrev) + "</fieldset>")
|
|
return (f"<form method=post action='/run/new/{i}'>{app.hidden()}{fields}"
|
|
"<fieldset><legend>Mode</legend>"
|
|
"<label><input type=radio name=mode value=manual checked> <b>Manual</b>: you confirm every check and gate.</label><br>"
|
|
"<label><input type=radio name=mode value=auto-confirm> <b>Auto-confirm</b>: the engine confirms when preconditions hold. "
|
|
"It does not run commands; you (or another tool) do the work.</label></fieldset>"
|
|
f"{pre}<p><label>Your name (one word)<br><input name=operator required></label></p><button>Start run</button></form>")
|
|
|
|
|
|
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 <a href='/project/{e(ctx['project'])}'>{e(ctx['project'])}</a>"
|
|
f"{' @' + e(ctx['target']) if ctx.get('target') else ''} · {e(ctx.get('environment', ''))}")
|
|
ov = R.overview(run)
|
|
prog = ov["progress"]
|
|
out = [f"<h1>{e(pack.data['title'])}</h1>"
|
|
f"<p>{sp(d['mode'])} {sp(d['outcome']) if d.get('outcome') else ''} "
|
|
f"<span class=note>run {e(rid)} · operator {e(d['operator'])}{where}</span></p>"
|
|
f"<p class=eyebrow>{prog['done']} of {prog['total']} done"
|
|
f"{' · ' + str(prog['pushed']) + ' pushed through' if prog['pushed'] else ''}</p>"
|
|
f"<progress max={prog['total']} value={prog['done'] + prog['pushed']} style='width:100%'></progress>"]
|
|
if d["deviations"]:
|
|
out.append(f"<p class=note>⚑ {len(d['deviations'])} deviation(s) recorded: this run can be finished but will never count towards "
|
|
"maturity. <a href='?view=overview'>See them</a>.</p>")
|
|
if msg:
|
|
out.append(f"<p class=bad>{e(msg)}</p>")
|
|
tabs = " ".join(f"<a class='tab{' on' if view == v else ''}' href='?view={v}'>{label}</a>"
|
|
for v, label in (("flow", "Guided flow"), ("data", "Data"), ("overview", "Overview")))
|
|
out.append(f"<nav class=tabs>{tabs}</nav>")
|
|
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))
|
|
|
|
|
|
# ---- 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("<details class=framing open><summary><b>Keep in mind</b></summary><ul>" + "".join(f"<li>{e(x)}</li>" for x in framing) + "</ul></details>")
|
|
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):
|
|
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"<div class=panel><b>Manual mode.</b> <span class=note>You confirm every check and gate. Unmet preconditions can be "
|
|
f"overridden with a reason; each override is recorded.</span>"
|
|
f"<form method=post action='{c.base}/mode' class=row>{c.h()}<input type=hidden name=to value=auto-confirm>"
|
|
"<button>Hand over to auto-confirm</button></form></div>")
|
|
ap = run.data["auto_confirm"]
|
|
last = ap.get("last")
|
|
status = ""
|
|
if ap.get("stopped"):
|
|
status = f"<p class=bad>Stopped at {e(ap['stopped']['item'])}: {e(ap['stopped']['reason'])}</p>"
|
|
elif last:
|
|
status = f"<p class=note>Last: {e(last['action'])}{' ' + e(last['item']) if last.get('item') else ''} {e(last.get('detail', ''))}</p>"
|
|
pre = ap.get("preauthorized") or []
|
|
return (f"<div class=panel data-s='auto'><b>Auto-confirm mode.</b> <span class=note>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 ''}.</span>{status}"
|
|
f"<form method=post action='{c.base}/tick' class=row>{c.h()}"
|
|
"<button name=n value=1>Check once</button> <button name=n value=50>Run until it waits</button></form>"
|
|
f"<form method=post action='{c.base}/mode' class=row>{c.h()}<input type=hidden name=to value=manual>"
|
|
"<button>Take over (manual)</button></form></div>")
|
|
|
|
|
|
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 "<div class=banner><b>Everything is done.</b> Close the run below.</div>"
|
|
b = na["blocked"][0] if na["blocked"] else None
|
|
return f"<div class=banner><b>Nothing can start.</b> {e(_title_of(c, b['id'])) if b else ''} waits: {e('; '.join(b['unmet'])) if b else ''}</div>"
|
|
kind = c.fl.kind(rec)
|
|
others = [x for x in na["ready"] + na["running"] if x != rec]
|
|
links = " ".join(f"<a class=chip href='#item-{e(x)}'>{e(_title_of(c, x))}</a>" for x in others)
|
|
verb = "Continue" if rec in na["running"] else ("Evaluate gate" if kind == "gate" else "Next")
|
|
return (f"<div class=banner><b>{verb}:</b> <a href='#item-{e(rec)}'>{e(_title_of(c, rec))}</a> <span class=note>(recommended)</span>"
|
|
f"{'<br><span class=note>Also possible in parallel:</span> ' + links if links else ''}</div>")
|
|
|
|
|
|
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"<p class=note><b>Goal:</b> {e(st['goal'])}</p>"]
|
|
if st.get("framing"):
|
|
body.append("<ul class=note>" + "".join(f"<li>{e(x)}</li>" for x in st["framing"]) + "</ul>")
|
|
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"<details class='stage' id='stage-{e(st['id'])}'{'' if done else ' open'}><summary><span class=head>"
|
|
f"<span><b>{e(st['title'])}</b> <span class=note>{n_done}/{len(members)} steps</span></span>"
|
|
f"{sp('complete' if done else 'in progress')}</span></summary>{''.join(body)}</details>")
|
|
|
|
|
|
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 = []
|
|
for d in dps:
|
|
rec = c.run.data["datapoints"][d["id"]]
|
|
rows.append(f"<p>{_input(d, rec)}<br><span class=note>{e(d.get('help', ''))}</span></p>")
|
|
return (f"<div class=questions><b>Questions for this stage</b> <span class=note>(values used by its steps; never enter secrets)</span>"
|
|
f"<form method=post action='{c.base}/data'>{c.h()}<input type=hidden name=next value='{c.base}#stage-{e(st['id'])}'>"
|
|
f"{''.join(rows)}<button>Save answers</button></form></div>")
|
|
|
|
|
|
def _input(d: dict, rec: dict, editable: bool = True) -> str:
|
|
name = f"d_{d['id']}"
|
|
val = rec.get("value", "")
|
|
dis = "" if editable else " disabled"
|
|
if d.get("values"):
|
|
ctl = f"<select name='{name}'{dis}>" + "".join(f"<option{' selected' if o == val else ''}>{e(o)}</option>" for o in d["values"]) + "</select>"
|
|
elif d["type"] == "bool":
|
|
ctl = (f"<select name='{name}'{dis}><option value=''>(not set)</option>"
|
|
+ "".join(f"<option{' selected' if o == val else ''}>{o}</option>" for o in ("true", "false")) + "</select>")
|
|
else:
|
|
ctl = f"<input name='{name}' value='{e(val)}'{dis}>"
|
|
err = f" <span class=bad>{e(rec.get('error', ''))}</span>" if rec["state"] == "err" else ""
|
|
return f"<label>{e(d['label'])} {sp(rec['state'])}<br>{ctl}</label>{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"<span class=chip>{e(eff.get('idempotent') or s.get('idempotent'))}</span>")
|
|
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("<span class=chip data-s='irreversible'>irreversible</span>")
|
|
if s.get("auto_confirm") == "manual-only":
|
|
marks.append("<span class=chip>manual-only</span>")
|
|
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"<p class=note>{e(s['description'])}</p>")
|
|
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
|
|
"<label><input type=checkbox name=acknowledge_partial value=1> I have checked that the earlier attempt did not partly apply</label> ")
|
|
body.append(f"<p class=bad>Failed: {e(st.get('note') or '(no comment)')}</p>"
|
|
f"<form method=post action='{url}/retry'>{c.h()}<p class=row>{ack}<button>Retry</button></p></form>")
|
|
elif state == "pushed":
|
|
body.append("<p class=note>Passed over by push-through without verification: not done, and recorded as a deviation.</p>")
|
|
elif state == "done" and st.get("note"):
|
|
body.append(f"<p class=note>Comment: {e(st['note'])}</p>")
|
|
if state == "done" or state == "pushed":
|
|
cmd = _command(run, sid)
|
|
if cmd:
|
|
body.insert(0, f"<pre>{e(cmd)}</pre>")
|
|
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"<details class='step {e(state)}' id='item-{e(sid)}'{fold}><summary{current}><span class=head>"
|
|
f"<span><b>{n}. {e(title)}</b> <span class=note>[{e(actor)} · {e(s['owner'])}]</span> {' '.join(marks)}</span>"
|
|
f"{sp(state)}</span></summary>{''.join(body)}</details>")
|
|
|
|
|
|
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 = "<p><b>Choose one:</b><br>" + "".join(
|
|
f"<label><input type=radio name=alt value='{e(a['id'])}'{' checked' if (a['id'] == rec or (rec is None and a is alts[0])) else ''}> "
|
|
f"{e(a['title'])}{' <span class=note>(recommended)</span>' if a['id'] == rec else ''}</label><br>" for a in alts) + "</p>"
|
|
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"<p>This step is <b>irreversible</b>. Type the confirmation phrase to start it:<br>"
|
|
f"<code>{e(R.phrase_for(run, sid))}</code><br><input name=phrase autocomplete=off class=grow style='width:100%'></p>")
|
|
if not unmet:
|
|
if s.get("auto_confirm") == "manual-only" and not c.manual:
|
|
return "<p class=note>Manual-only: auto-confirm will not start this step. Take over to do it yourself.</p>"
|
|
return f"<form method=post action='{url}/start'>{c.h()}{alt_html}{phrase}<button>Start this step</button></form>"
|
|
risk = R._risk_text(run, sid)
|
|
wait = f"<p class=note>Waiting for: {e('; '.join(unmet))}</p>"
|
|
if not c.manual:
|
|
return wait + "<p class=note>Overrides are made in manual mode.</p>"
|
|
return (wait + f"<details class=override><summary>Start anyway (override)</summary>"
|
|
f"<form method=post action='{url}/start'>{c.h()}<p class=bad>Downstream risk: {e(risk)}</p>{alt_html}{phrase}"
|
|
"<p><input name=override_reason class=grow style='width:100%' placeholder='why you continue anyway (required)' required></p>"
|
|
"<p><label><input type=checkbox name=acknowledge_risk value=1 required> I acknowledge the risk. Record this as a deviation.</label></p>"
|
|
"<button>Override and start</button></form></details>")
|
|
|
|
|
|
def _running(c: Ctx, s: dict, eff: dict, st: dict, sid: str, url: str) -> str:
|
|
run = c.run
|
|
out = []
|
|
cmd = _command(run, sid)
|
|
if cmd:
|
|
out.append(f"<pre>{e(cmd)}</pre><button type=button data-copy>Copy</button>")
|
|
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"<form method=post action='{url}/verify'>{c.h()}<p>Is this true? <b>{e(said)}</b></p>"
|
|
"<button name=ok value=1>Yes, it is</button> <button name=ok value=0>Not yet</button></form>")
|
|
else:
|
|
what = (f"<code>{e(_render_check(run, ver))}</code>" if ver.get("command") else "the datapoint states")
|
|
out.append(f"<form method=post action='{url}/verify'>{c.h()}<input type=hidden name=yes value=1>"
|
|
f"<p class=note>Checks: {what}</p><button>Run check</button></form>")
|
|
if not v:
|
|
out.insert(0, "<p class=note>1. Do the work, then check → 2. Mark done to lock it in</p>")
|
|
elif v["passed"]:
|
|
out.append("<p><b>✔ Check passed.</b> Press <b>Mark done</b> to lock it in.</p>")
|
|
else:
|
|
out.append("<p class=bad>✘ Check not passed. Fix the cause and check again, or mark the step failed.</p>")
|
|
ev = ""
|
|
if v and v["passed"]:
|
|
ev = "".join(f"<label>{e(x)} <input name='e_{e(x)}' required></label> " for x in s.get("evidence", []))
|
|
ev = f"<p>{ev}</p>" if ev else ""
|
|
done_btn = f"<button formaction='{url}/done'>Mark done</button> " if v and v["passed"] else ""
|
|
out.append(f"<form method=post action='{url}/fail'>{c.h()}{ev}<p class=row>{done_btn}"
|
|
f"<button formaction='{url}/fail' formnovalidate data-fail>Mark failed</button> "
|
|
"<input name=note class=grow placeholder='comment (what happened; no secrets)'></p>"
|
|
"<p class='note bad' data-warn hidden>Add a comment about what went wrong, or click <b>Mark failed</b> again to fail without one.</p></form>")
|
|
if eff.get("rollback"):
|
|
out.append(f"<p class=note>Rollback: {e(_render_text(run, eff['rollback']))}</p>")
|
|
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"<li>{pill} {e(text)}{'<br>' + ctl if ctl else ''}</li>")
|
|
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"<details class=override><summary>Waive this gate (declared waivable by {e(role)})</summary>"
|
|
f"<form method=post action='{c.base}/gate/{e(gid)}/waive'>{c.h()}"
|
|
f"<p class=bad>Downstream risk: {e(R._risk_text(run, gid))}</p>"
|
|
"<p><input name=reason class=grow style='width:100%' placeholder='reason (required)' required></p>"
|
|
"<button>Waive and record as a deviation</button></form></details>")
|
|
extra = ""
|
|
if state == "pushed":
|
|
extra = "<p class=note>Passed over by push-through without passing: recorded as a deviation.</p>"
|
|
if state == "waived":
|
|
extra = f"<p class=note>Waived: {e(gs['waiver']['reason'])}</p>"
|
|
current = " open" if state not in R.DONE_GATE else ""
|
|
marker = " id=current" if gid == c.na["recommended"] else ""
|
|
return (f"<details class='step gate {e(state)}' id='item-{e(gid)}'{current}><summary{marker}><span class=head>"
|
|
f"<span><b>Gate: {e(g['title'])}</b> <span class=note>quality goal: {e(g['goal'])}"
|
|
f"{' · blocking' if g.get('blocking', True) else ''}</span></span>{sp(state)}</span></summary>"
|
|
f"<ul class=checks>{''.join(rows)}</ul>{extra}{waive}</details>")
|
|
|
|
|
|
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()}<input type=hidden name=idx value={idx}>"
|
|
if exp == "manual":
|
|
ctl = "<button name=ok value=1>Yes, it is</button> <button name=ok value=0>Not yet</button>"
|
|
else:
|
|
cmd = f" <code>{e(_render_check(run, ck))}</code>" if ck.get("command") else ""
|
|
ctl = f"<input type=hidden name=yes value=1><button>Run check</button>{cmd}"
|
|
if not unmet:
|
|
return f"<form method=post action='{action}'>{hidden}<p class=row>{ctl}</p></form>"
|
|
if not c.manual:
|
|
return f"<span class=note>Waiting for: {e('; '.join(unmet))}</span>"
|
|
return (f"<span class=note>Waiting for: {e('; '.join(unmet))}</span>"
|
|
f"<details class=override><summary>Check anyway (override)</summary>"
|
|
f"<form method=post action='{action}'>{hidden}<p class=bad>Downstream risk: {e(R._risk_text(run, gid))}</p>"
|
|
"<p><input name=override_reason class=grow style='width:100%' placeholder='why you continue anyway (required)' required></p>"
|
|
"<p><label><input type=checkbox name=acknowledge_risk value=1 required> I acknowledge the risk. Record this as a deviation.</label></p>"
|
|
f"<p class=row>{ctl}</p></form></details>")
|
|
|
|
|
|
# ---- 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"<a class='chip{cls}' href='{qs.replace('&', '&')}'>{e(label)}</a>"
|
|
|
|
ctl = ("<p><b>Group by</b> " + link(None, flt, "none", " on" if not group else "") + " "
|
|
+ " ".join(link(k, flt, k, " on" if group == k else "") for k in keys) + "</p>")
|
|
if flt:
|
|
ctl += f"<p><b>Filter</b> <span class=chip on>{e(flt)}</span> " + link(group, None, "clear filter") + "</p>"
|
|
groups = R.datapoint_rows(run, group=group, flt=flt)
|
|
out = [f"<p class=note>Every datapoint of this run, independent of the flow. Tags (dimensions) come from the pack; click one to filter.</p>{ctl}"]
|
|
editable = not c.finished
|
|
out.append(f"<form method=post action='{c.base}/data'>{c.h()}<input type=hidden name=next value='{c.base}?view=data"
|
|
f"{'&group=' + quote(group) if group else ''}{'&filter=' + quote(flt) if flt else ''}'>")
|
|
for gname, rows in groups.items():
|
|
if gname:
|
|
out.append(f"<h3>{e(group)}: {e(gname)}</h3>")
|
|
trs = ""
|
|
for r in rows:
|
|
rec = run.data["datapoints"][r["id"]]
|
|
d = fl.datapoints[r["id"]]
|
|
dims = " ".join(link(group, f"{k}:{v}", f"{k}: {v}") for k, v in r["dimensions"].items())
|
|
src = " · ".join(x for x in (f"asked in {e(r['ask_in'])}" if r["ask_in"] else "", f"produced by {e(r['produced_by'])}" if r["produced_by"] else "",
|
|
f"confirmed by {e(r['confirmed_by'])}" if r["confirmed_by"] else "", f"how: {e(r['how'])}" if r["how"] else "") if x)
|
|
ctl = _input(d, rec, editable and r["editable"]) if (editable and r["editable"]) else (
|
|
f"{e(r['label'])} {sp(r['state'])}<br><code>{e(r['value'] if r['value'] is not None else '')}</code>")
|
|
trs += f"<tr><td>{ctl}</td><td>{dims}</td><td class=note>{src}</td></tr>"
|
|
out.append(f"<table class=data><tr><th>Datapoint</th><th>Dimensions</th><th>Provenance</th></tr>{trs}</table>")
|
|
if not groups:
|
|
out.append("<p class=note>No datapoints match.</p>")
|
|
if editable:
|
|
out.append("<button>Save changed values</button>")
|
|
out.append("</form>")
|
|
arts = R.artefact_rows(run)
|
|
if arts:
|
|
out.append("<h2>Artefacts</h2><table><tr><th>Artefact</th><th>Where it lives</th><th>Produced by</th><th>Until</th></tr>" + "".join(
|
|
f"<tr><td>{e(a['title'])}</td><td><code>{e(a['location'] or '(not known yet)')}</code></td>"
|
|
f"<td>{e(a['produced_by'])} {sp(a['produced_state'])}</td><td>{e(a['lives_until'] or '-')}</td></tr>" for a in arts) + "</table>")
|
|
if fl.actors:
|
|
counts: dict[str, int] = {}
|
|
for s in fl.steps.values():
|
|
counts[s["actor"]] = counts.get(s["actor"], 0) + 1
|
|
out.append("<h2>Actors</h2><table><tr><th>Role</th><th>Description</th><th>Privileges</th><th>Steps</th></tr>" + "".join(
|
|
f"<tr><td>{e(a['role'])} <span class=note>({e(a['id'])})</span></td><td>{e(a['description'])}</td>"
|
|
f"<td>{e(', '.join(a.get('privileges', [])))}</td><td>{counts.get(a['id'], 0)}</td></tr>" for a in fl.actors.values()) + "</table>")
|
|
return "".join(out)
|
|
|
|
|
|
# ---- 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" <span class=bad>{s['steps_failed']} failed</span>"
|
|
if s["steps_pushed"]:
|
|
prob += f" {s['steps_pushed']} pushed"
|
|
if s["steps_running"]:
|
|
prob += f" <span class=note>{s['steps_running']} running</span>"
|
|
rows += (f"<tr><td style='padding-left:{0.5 + s['depth'] * 1.2}rem'><a href='?view=flow#stage-{e(s['id'])}'>{e(s['title'])}</a>"
|
|
f"<br><span class=note>{e(s['goal'])}</span></td>"
|
|
f"<td>{s['steps_done']}/{s['steps_total']}{prob}</td><td>{gates or '-'}</td>"
|
|
f"<td>{sp('complete' if s['complete'] else 'in progress')}"
|
|
f"{'<br><span class=note>' + e('; '.join(s['unmet'])) + '</span>' if s['unmet'] else ''}</td></tr>")
|
|
dps = " ".join(f"{sp(k)} {v}" for k, v in sorted(ov["datapoints"].items()))
|
|
out = ["<h2>Stages</h2><table><tr><th>Stage</th><th>Steps</th><th>Gates</th><th>State</th></tr>" + rows + "</table>",
|
|
f"<h2>Datapoints</h2><p>{dps}</p><p class=note><a href='?view=data'>Open the data view</a></p>"]
|
|
if ov["deviations"]:
|
|
out.append("<h2>Deviations</h2><table><tr><th>Kind</th><th>Item</th><th>Mode</th><th>Unmet</th><th>Reason</th><th>When</th></tr>" + "".join(
|
|
f"<tr><td>{sp(d['kind'])}</td><td>{e(d['item'])}</td><td>{e(d['mode'])}</td><td>{e('; '.join(d['unmet']))}</td>"
|
|
f"<td>{e(d['reason'])}{'<br><span class=note>risk: ' + e(d['downstream_risk']) + '</span>' if d.get('downstream_risk') else ''}</td>"
|
|
f"<td class=note>{e(d['at'][:19])}</td></tr>" for d in ov["deviations"]) + "</table>")
|
|
else:
|
|
out.append("<h2>Deviations</h2><p class=note>None. A run without deviations can count towards maturity.</p>")
|
|
if ov["mode_history"]:
|
|
out.append("<h2>Mode changes</h2><ul>" + "".join(f"<li>{e(m['from'])} → {e(m['to'])} · {e(m['at'][:19])}</li>" for m in ov["mode_history"]) + "</ul>")
|
|
if ov["stopped"]:
|
|
out.append(f"<p class=bad>Auto-confirm stopped at {e(ov['stopped']['item'])}: {e(ov['stopped']['reason'])}</p>")
|
|
return "".join(out)
|
|
|
|
|
|
# ---- closing the run -----------------------------------------------------------------------------------------------------
|
|
|
|
def _close(c: Ctx) -> str:
|
|
run = c.run
|
|
if c.finished:
|
|
note = f" - {e(run.data['outcome_note'])}" if run.data.get("outcome_note") else ""
|
|
return (f"<div id=close class=step><h2>Run finished: {e(run.data['outcome'])}</h2><p>{note}</p>"
|
|
"<p class=note>The outcome is part of this runbook's score and maturity evidence at this revision. "
|
|
"<a href='/'>← Back to the runbook list</a></p></div>")
|
|
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("<button name=outcome value=succeeded>Finish: succeeded</button>")
|
|
if {"failed", "pushed"} & set(steps) or {"failed", "pushed"} & set(gates):
|
|
btns.append("<button name=outcome value=failed onclick=\"return confirm('Finish this run as failed?')\">Finish: failed</button>")
|
|
btns.append("<button name=outcome value=abandoned onclick=\"return confirm('Abandon this run?')\">Abandon run</button>")
|
|
return ("<div id=close class=step><h2>Close this run</h2>"
|
|
"<p class=note>Finishing records the outcome. A run with deviations can be finished but does not count towards maturity. "
|
|
"A finished run is read-only.</p>"
|
|
f"<form method=post action='{c.base}/finalize'>{c.h()}<p><input name=note class=grow style='width:100%' placeholder='comment (optional)'></p>"
|
|
f"<p class=row>{' '.join(btns)}</p></form></div>")
|
|
|
|
|
|
# ---- 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", ""))
|
|
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 = []
|
|
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
|
|
if R.set_datapoint(run, dp, v) == "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")
|