"""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 = ("
Pre-authorize irreversible steps (auto-confirm only)" "

An auto-confirm run never confirms an irreversible step by itself unless you name it here. Steps marked " "manual-only are never automated.

" + "".join(f"
" for s in irrev) + "
") return (f"
{app.hidden()}{fields}" "
Mode" "
" "
" f"{pre}{app.who()}
") 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', ''))}") ov = R.overview(run) prog = ov["progress"] out = [f"

{e(pack.data['title'])}

" f"

{sp(d['mode'])} {sp(d['outcome']) if d.get('outcome') else ''} " f"run {e(rid)} · operator {e(d['operator'])}{where}

" f"

{times_line(run)}

" 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
") 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"
{c.h()}" "
") 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'])}

" elif last: status = f"

Last: {e(last['action'])}{' ' + e(last['item']) if last.get('item') else ''} {e(last.get('detail', ''))}

" pre = ap.get("preauthorized") or [] return (f"
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"
{c.h()}" "
" f"
{c.h()}" "
") 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 "" b = na["blocked"][0] if na["blocked"] else None return f"" 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"") 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("") 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"

{_input(d, rec, users=users)}
{e(d.get('help', ''))}

") return (f"
Questions for this stage (values used by its steps; never enter secrets)" f"
{c.h()}" f"{''.join(rows)}{_override_fields(c, locked)}
") 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"
{c.h()}

{ack}

") 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.

" return f"
{c.h()}{alt_html}{phrase}
" risk = R._risk_text(run, sid) wait = f"

Waiting for: {e('; '.join(unmet))}

" if not c.manual: return wait + "

Overrides are made in manual mode.

" return (wait + f"
Start anyway (override)" f"
{c.h()}

Downstream risk: {e(risk)}

{alt_html}{phrase}" "

" "

" "
") 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"
{e(cmd)}
") 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"
{c.h()}

Is this true? {e(said)}

" "
") else: what = (f"{e(_render_check(run, ver))}" if ver.get("command") else "the datapoint states") out.append(f"
{c.h()}" f"

Checks: {what}

") 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"
{c.h()}{ev}

{done_btn}" 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"
    {c.h()}" f"

    Downstream risk: {e(R._risk_text(run, gid))}

    " "

    " "
    ") 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"{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"
    {hidden}

    {ctl}

    " if not c.manual: return f"Waiting for: {e('; '.join(unmet))}" return (f"Waiting for: {e('; '.join(unmet))}" f"
    Check anyway (override)" f"
    {hidden}

    Downstream risk: {e(R._risk_text(run, gid))}

    " "

    " "

    " f"

    {ctl}

    ") # ---- 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) + "

    ") if flt: ctl += f"

    Filter {e(flt)} " + link(group, None, "clear filter") + "

    " 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"
    {c.h()}") for gname, rows in groups.items(): if gname: out.append(f"

    {e(group)}: {e(gname)}

    ") 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"], R.datapoint_users(run, r["id"])) if (editable and r["editable"]) else ( f"{e(r['label'])} {sp(r['state'])}
    {e(r['value'] if r['value'] is not None else '')}") trs += f"{ctl}{dims}{src}" out.append(f"{trs}
    DatapointDimensionsProvenance
    ") if not groups: out.append("

    No datapoints match.

    ") if editable: out.append(_override_fields(c, any_locked) + "") out.append("
    ") arts = R.artefact_rows(run) if arts: out.append("

    Artefacts

    " + "".join( f"" f"" f"" for a in arts) + "
    ArtefactWhere it livesOpenProduced byUntil
    {e(a['title'])}{e(a['location'] or '(not known yet)')}{_artefact_links(c, a)}{e(a['produced_by'])} {sp(a['produced_state'])}{e(a['lives_until'] or '-')}
    ") 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

    " + "".join( f"" f"" for a in fl.actors.values()) + "
    RoleDescriptionPrivilegesSteps
    {e(a['role'])} ({e(a['id'])}){e(a['description'])}{e(', '.join(a.get('privileges', [])))}{counts.get(a['id'], 0)}
    ") 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"{e(s['title'])}" f"
    {e(s['goal'])}" f"{s['steps_done']}/{s['steps_total']}{prob}{gates or '-'}" f"{sp('complete' if s['complete'] else 'in progress')}" f"{'
    ' + e('; '.join(s['unmet'])) + '' if s['unmet'] else ''}") dps = " ".join(f"{sp(k)} {v}" for k, v in sorted(ov["datapoints"].items())) out = ["

    Stages

    " + rows + "
    StageStepsGatesState
    ", f"

    Datapoints

    {dps}

    Open the data view

    "] if ov["deviations"]: out.append("

    Deviations

    " + "".join( f"" f"" f"" for d in ov["deviations"]) + "
    KindItemModeUnmetReasonWhen
    {sp(d['kind'])}{e(d['item'])}{e(d['mode'])}{e('; '.join(d['unmet']))}{e(d['reason'])}{'
    risk: ' + e(d['downstream_risk']) + '' if d.get('downstream_risk') else ''}
    {e(d['at'][:19])}
    ") else: out.append("

    Deviations

    None. A run without deviations can count towards maturity.

    ") if ov["mode_history"]: out.append("

    Mode changes

    ") 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" - {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"
    {c.h()}

    " f"{c.app.close_feedback_fields(run)}

    {' '.join(btns)}

    ") # ---- 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")