tegwick
8d62c1fbe8
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
683 lines
32 KiB
Python
683 lines
32 KiB
Python
"""Runs of runbook-pack/v0.2 packs (record runbook-run/v0.2).
|
|
|
|
Adds to the v0.1 run: datapoint states (nil/set/err/ok), a dependency graph with a recommended
|
|
next action, alternatives, QA gates with waivers, manual and auto-confirm modes with timeouts, and
|
|
a deviation record. Shared helpers (feedback, finalize, captures, score) live in run.py.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
import subprocess
|
|
import time
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
|
|
from .flow import Flow
|
|
from .pack import Pack, digest, render
|
|
from .run import Run, RunError, _clean, _context, _now, runs_dir # noqa: F401 (re-exported for callers)
|
|
|
|
DEFAULT_PHRASE = "I understand this cannot be undone"
|
|
VERIFY_TIMEOUT = 60
|
|
# "pushed" is its own state: the item was passed over by auto-confirm push-through without being verified.
|
|
# It lets dependents proceed but is neither done nor an error, and it is tracked separately.
|
|
DONE_STEP = ("done", "skipped", "pushed")
|
|
DONE_GATE = ("passed", "waived", "pushed")
|
|
|
|
|
|
def flow(run: Run) -> Flow:
|
|
if not hasattr(run, "_flow"):
|
|
run._flow = Flow(run.pack.data) # type: ignore[attr-defined]
|
|
return run._flow # type: ignore[attr-defined]
|
|
|
|
|
|
def _values(run: Run) -> dict[str, str]:
|
|
return {k: v["value"] for k, v in run.data["datapoints"].items() if v["state"] in ("set", "ok")}
|
|
|
|
|
|
def _sync(run: Run) -> None:
|
|
run.data["params"] = _values(run)
|
|
|
|
|
|
# ---- datapoints ---------------------------------------------------------------------------------
|
|
|
|
def _coerce(run: Run, dp_id: str, value: str) -> tuple[str | None, str | None]:
|
|
"""(clean value, None) or (None, error)."""
|
|
dp = flow(run).datapoints[dp_id]
|
|
value = str(value)
|
|
if dp["type"] == "int" and not re.fullmatch(r"-?\d+", value):
|
|
return None, "not an integer"
|
|
if dp["type"] == "bool" and value not in ("true", "false"):
|
|
return None, "must be true or false"
|
|
try:
|
|
render("{{%s}}" % dp_id, {dp_id: value}, run.pack) # values, pattern, unsafe characters
|
|
_clean(value, f"datapoint {dp_id}")
|
|
except (ValueError, KeyError, RunError) as e:
|
|
return None, str(e)
|
|
return value, None
|
|
|
|
|
|
def datapoint_users(run: Run, dp_id: str) -> list[str]:
|
|
"""Steps that have already started and use this datapoint: they worked with its current value."""
|
|
return [sid for sid, s in flow(run).steps.items()
|
|
if dp_id in s.get("uses", []) and run.data["steps"][sid]["state"] in ("running", "done", "failed", "pushed")]
|
|
|
|
|
|
def changes_needing_override(run: Run, changes: dict[str, str]) -> dict[str, list[str]]:
|
|
"""Of proposed {datapoint: new value}, those that steps already used and that really change the value."""
|
|
out = {}
|
|
for dp, v in changes.items():
|
|
rec = run.data["datapoints"].get(dp)
|
|
if rec is None or rec.get("value") is None or str(v) == str(rec["value"]):
|
|
continue
|
|
users = datapoint_users(run, dp)
|
|
if users:
|
|
out[dp] = users
|
|
return out
|
|
|
|
|
|
def set_datapoint(run: Run, dp_id: str, value: str, how: str = "manual",
|
|
override_reason: str = "", acknowledge_risk: bool = False) -> str:
|
|
"""Give a datapoint a value. Invalid input leaves it in state 'err' (never stores it).
|
|
|
|
Changing a value that steps have already used is a deviation: it needs the operator's override
|
|
(reason and acknowledged risk), only in manual mode, and is recorded as 'changed-after-use'.
|
|
"""
|
|
_writable(run)
|
|
if dp_id not in flow(run).datapoints:
|
|
raise RunError(f"no such datapoint: {dp_id}")
|
|
rec = run.data["datapoints"][dp_id]
|
|
users = datapoint_users(run, dp_id) if how != "initial" else []
|
|
if users and rec.get("value") is not None and str(value) != str(rec["value"]):
|
|
risk = "Steps that already ran used the previous value, so their results may no longer match this one."
|
|
if run.data["mode"] != "manual":
|
|
raise RunError(f"{dp_id} was already used by {', '.join(users)}; take over (manual mode) to change it")
|
|
if not override_reason.strip():
|
|
raise RunError(f"{dp_id} was already used by {', '.join(users)}. Risk: {risk} "
|
|
"To change it give a reason and acknowledge the risk; this is recorded as a deviation.")
|
|
if not acknowledge_risk:
|
|
raise RunError("changing a value that steps already used needs the operator to acknowledge the risk")
|
|
_deviate(run, "changed-after-use", dp_id, [f"{s} already used {dp_id} = {rec['value']!r}" for s in users],
|
|
override_reason, risk, True)
|
|
clean, err = _coerce(run, dp_id, value)
|
|
if err:
|
|
rec.update(state="err", error=err, at=_now())
|
|
rec.pop("value", None)
|
|
else:
|
|
rec.update(state="set", value=clean, how=how, at=_now())
|
|
rec.pop("error", None)
|
|
_sync(run)
|
|
run.save()
|
|
return rec["state"]
|
|
|
|
|
|
def _confirm_datapoints(run: Run, by: str, how: str) -> None:
|
|
for d in flow(run).datapoints.values():
|
|
rec = run.data["datapoints"][d["id"]]
|
|
if d.get("confirmed_by") == by and rec["state"] == "set":
|
|
rec.update(state="ok", how=how, at=_now())
|
|
|
|
|
|
# ---- creation -------------------------------------------------------------------------------------
|
|
|
|
def create(pack: Pack, operator: str, values: dict[str, str] | None = None, mode: str = "manual",
|
|
preauthorize: list[str] | None = None, context: dict | None = None) -> Run:
|
|
if not pack.ok or pack.data.get("engine") != "native" or pack.spec != "runbook-pack/v0.2":
|
|
raise RunError("only valid native runbook-pack/v0.2 packs can be run here")
|
|
if mode not in ("manual", "auto-confirm"):
|
|
raise RunError("mode must be manual or auto-confirm")
|
|
operator = _clean(operator.strip(), "operator")
|
|
if not re.fullmatch(r"\S+", operator):
|
|
raise RunError("operator must be one word (name or handle)")
|
|
fl = Flow(pack.data)
|
|
for u in values or {}:
|
|
if u not in fl.datapoints:
|
|
raise RunError(f"unknown datapoint {u}")
|
|
pre = list(preauthorize or [])
|
|
for sid in pre:
|
|
if sid not in fl.steps or effective_step(pack, fl, sid, None).get("reversible", True):
|
|
raise RunError(f"{sid}: only irreversible steps can be pre-authorized")
|
|
data = {
|
|
"spec": "runbook-run/v0.2", "run_id": time.strftime("%Y%m%d-%H%M%S-") + uuid.uuid4().hex[:6],
|
|
"pack_id": pack.id, "pack_path": str(pack.path), "operator": operator, "created": _now(),
|
|
"mode": mode, "params": {}, "deviations": [],
|
|
"pack_digest": digest(pack), "context": _context(context),
|
|
"auto_confirm": {"preauthorized": pre, "authorized_by": operator if pre else None},
|
|
"datapoints": {i: {"state": "nil"} for i in fl.datapoints},
|
|
"steps": {i: {"state": "todo"} for i in fl.steps},
|
|
"gates": {i: {"state": "open", "results": {}} for i in fl.gates},
|
|
}
|
|
run = Run(pack, data)
|
|
run._flow = fl # type: ignore[attr-defined]
|
|
for i, d in fl.datapoints.items():
|
|
v = (values or {}).get(i, d.get("default"))
|
|
if v is not None:
|
|
set_datapoint(run, i, str(v).lower() if isinstance(v, bool) else str(v), how="initial")
|
|
if any(x["state"] == "err" for x in data["datapoints"].values()):
|
|
bad = {k: x["error"] for k, x in data["datapoints"].items() if x["state"] == "err"}
|
|
raise RunError(f"invalid datapoint values: {bad}")
|
|
run.save()
|
|
return run
|
|
|
|
|
|
def _writable(run: Run) -> None:
|
|
if run.data.get("outcome"):
|
|
raise RunError(f"run is finished (outcome: {run.data['outcome']}); start a new run")
|
|
|
|
|
|
# ---- step definitions and alternatives ------------------------------------------------------------------
|
|
|
|
def effective_step(pack: Pack, fl: Flow, step_id: str, alt: str | None) -> dict:
|
|
"""The step with the chosen alternative's command, verify, reversibility and rollback applied."""
|
|
s = dict(fl.steps[step_id])
|
|
if s.get("one_of") and alt:
|
|
a = next(x for x in s["one_of"] if x["id"] == alt)
|
|
for k in ("command", "verify", "reversible", "rollback", "title"):
|
|
if k in a:
|
|
s[k] = a[k]
|
|
return s
|
|
|
|
|
|
def recommended_alt(run: Run, step_id: str) -> str | None:
|
|
vals = _values(run)
|
|
for a in flow(run).steps[step_id].get("one_of", []):
|
|
rw = a.get("recommended_when") or {}
|
|
if rw and all(vals.get(k) in v for k, v in rw.items()):
|
|
return a["id"]
|
|
return None
|
|
|
|
|
|
def step_def(run: Run, step_id: str) -> dict:
|
|
if step_id not in flow(run).steps:
|
|
raise RunError(f"no such step: {step_id}")
|
|
return effective_step(run.pack, flow(run), step_id, run.data["steps"][step_id].get("alternative"))
|
|
|
|
|
|
def command_for(run: Run, step_id: str) -> str | None:
|
|
cmd = step_def(run, step_id).get("command")
|
|
return render(cmd, run.data["params"], run.pack) if cmd else None
|
|
|
|
|
|
def phrase_for(run: Run, step_id: str) -> str:
|
|
return flow(run).steps[step_id].get("confirm_phrase") or run.pack.data.get("confirm_phrase") or DEFAULT_PHRASE
|
|
|
|
|
|
def confirmation_style(run: Run, step_id: str) -> str:
|
|
"""'confirm' (button) for reversible steps, 'phrase' (typed) for irreversible ones."""
|
|
return "confirm" if step_def(run, step_id).get("reversible", True) else "phrase"
|
|
|
|
|
|
# ---- readiness, unmet preconditions, the recommended next action -----------------------------------------------
|
|
|
|
def is_complete(run: Run, id_: str) -> bool:
|
|
fl = flow(run)
|
|
k = fl.kind(id_)
|
|
if k == "step":
|
|
return run.data["steps"][id_]["state"] in DONE_STEP
|
|
if k == "gate":
|
|
return run.data["gates"][id_]["state"] in DONE_GATE
|
|
return all(is_complete(run, m) for m in fl.members(id_))
|
|
|
|
|
|
def unmet(run: Run, id_: str) -> list[str]:
|
|
fl = flow(run)
|
|
out = []
|
|
for r in fl.requires(id_):
|
|
if not is_complete(run, r):
|
|
out.append(f"{fl.kind(r)} {r} is not complete")
|
|
if fl.kind(id_) == "step":
|
|
for dp in fl.steps[id_].get("uses", []):
|
|
st = run.data["datapoints"][dp]["state"]
|
|
if st not in ("set", "ok", "pushed"):
|
|
out.append(f"datapoint {dp} is {st}")
|
|
elif fl.kind(id_) == "gate":
|
|
stage = fl.parent[id_]
|
|
for m in fl.members(stage, include_gates=False):
|
|
if not is_complete(run, m):
|
|
out.append(f"step {m} of the stage is not done")
|
|
return out
|
|
|
|
|
|
def _item_open(run: Run, id_: str) -> bool:
|
|
fl = flow(run)
|
|
return not is_complete(run, id_) if fl.kind(id_) in ("step", "gate") else False
|
|
|
|
|
|
def next_actions(run: Run) -> dict:
|
|
"""{'recommended': id|None, 'ready': [...], 'blocked': [{'id', 'unmet'}...]} in canonical order."""
|
|
fl = flow(run)
|
|
running = [i for i in fl.order if fl.kind(i) == "step" and run.data["steps"][i]["state"] == "running"]
|
|
ready, blocked = [], []
|
|
for i in fl.order:
|
|
if not _item_open(run, i) or i in running:
|
|
continue
|
|
u = unmet(run, i)
|
|
(blocked if u else ready).append({"id": i, "unmet": u} if u else i)
|
|
rec = (running + ready + [None])[0]
|
|
return {"recommended": rec, "running": running, "ready": ready, "blocked": blocked}
|
|
|
|
|
|
# ---- deviations --------------------------------------------------------------------------------------------------
|
|
|
|
def _deviate(run: Run, kind: str, item: str, unmet_list: list[str], reason: str, risk: str, ack: bool) -> None:
|
|
run.data["deviations"].append({
|
|
"kind": kind, "item": item, "mode": run.data["mode"], "unmet": unmet_list,
|
|
"downstream_risk": risk, "risk_acknowledged": ack, "reason": _clean(reason.strip(), "reason"),
|
|
"who": run.data["operator"], "at": _now()})
|
|
|
|
|
|
def _risk_text(run: Run, id_: str) -> str:
|
|
return flow(run).node(id_).get("downstream_risk") or "Steps that rely on this may break or give wrong results."
|
|
|
|
|
|
def _gate_precondition(run: Run, id_: str, override_reason: str, acknowledge_risk: bool) -> None:
|
|
u = unmet(run, id_)
|
|
if not u:
|
|
return
|
|
if not override_reason.strip():
|
|
raise RunError(f"{id_} has unmet preconditions: {'; '.join(u)}. Risk: {_risk_text(run, id_)} "
|
|
"To continue anyway give an override reason and acknowledge the risk.")
|
|
if not acknowledge_risk:
|
|
raise RunError("an override needs the operator to acknowledge the downstream risk")
|
|
if run.data["mode"] != "manual":
|
|
raise RunError("overrides are made by the operator in manual mode")
|
|
_deviate(run, "override", id_, u, override_reason, _risk_text(run, id_), True)
|
|
|
|
|
|
# ---- steps -----------------------------------------------------------------------------------------------------------
|
|
|
|
def start(run: Run, step_id: str, alt: str | None = None, phrase: str | None = None,
|
|
override_reason: str = "", acknowledge_risk: bool = False) -> None:
|
|
_writable(run)
|
|
fl = flow(run)
|
|
if step_id not in fl.steps:
|
|
raise RunError(f"no such step: {step_id}")
|
|
st = run.data["steps"][step_id]
|
|
if st["state"] in DONE_STEP or st["state"] == "running":
|
|
raise RunError(f"step {step_id} already {st['state']}")
|
|
_gate_precondition(run, step_id, override_reason, acknowledge_risk)
|
|
s = fl.steps[step_id]
|
|
if s.get("one_of"):
|
|
alt = alt or recommended_alt(run, step_id)
|
|
if alt not in {a["id"] for a in s["one_of"]}:
|
|
raise RunError(f"step {step_id}: choose one of {', '.join(a['id'] for a in s['one_of'])}")
|
|
st["alternative"] = alt
|
|
elif alt:
|
|
raise RunError(f"step {step_id} has no alternatives")
|
|
eff = effective_step(run.pack, fl, step_id, st.get("alternative"))
|
|
if not eff.get("reversible", True): # irreversible: the typed phrase is enforced here
|
|
pre = run.data["auto_confirm"]["preauthorized"]
|
|
if step_id in pre and run.data["mode"] == "auto-confirm":
|
|
st["authorized_by"] = run.data["auto_confirm"]["authorized_by"]
|
|
elif (phrase or "") != phrase_for(run, step_id):
|
|
raise RunError(f"step {step_id} is irreversible: type the confirmation phrase exactly: "
|
|
f"{phrase_for(run, step_id)!r}")
|
|
else:
|
|
st["authorized_by"] = run.data["operator"]
|
|
st.pop("verification", None)
|
|
st.update(state="running", started=_now())
|
|
run.save()
|
|
|
|
|
|
def _check_result(run: Run, ck: dict, confirm: bool, manual_ok: bool | None, auto_manual: bool = False) -> dict:
|
|
parts, passed = {}, True
|
|
dps = ck.get("datapoints_ok", [])
|
|
if dps:
|
|
states = {d: run.data["datapoints"][d]["state"] for d in dps}
|
|
parts["datapoints_ok"] = all(v == "ok" for v in states.values())
|
|
passed &= parts["datapoints_ok"]
|
|
exp = ck.get("expect") or (None if dps else "manual")
|
|
if exp == "manual":
|
|
if auto_manual:
|
|
parts["manual"] = True
|
|
elif manual_ok is None:
|
|
raise RunError("a manual check needs the operator's yes/no")
|
|
else:
|
|
parts["manual"] = bool(manual_ok)
|
|
passed &= parts["manual"]
|
|
elif exp in ("exit-0", "output-contains"):
|
|
if not confirm:
|
|
raise RunError("running a check command needs operator confirmation")
|
|
try:
|
|
cmd = render(ck["command"], run.data["params"], run.pack)
|
|
except (KeyError, ValueError) as e: # e.g. a datapoint that was pushed through has no value
|
|
return {"passed": False, "kind": exp, "parts": {"command": False, "error": str(e)}, "at": _now()}
|
|
t0 = time.monotonic()
|
|
try:
|
|
cp = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=VERIFY_TIMEOUT,
|
|
cwd=run.pack.path.resolve().parents[2])
|
|
code, out = cp.returncode, cp.stdout + cp.stderr
|
|
except subprocess.TimeoutExpired:
|
|
code, out = -1, ""
|
|
parts.update(exit_code=code, output_bytes=len(out.encode()), seconds=round(time.monotonic() - t0, 2))
|
|
ok = code == 0
|
|
if exp == "output-contains":
|
|
ok = ok and render(ck["contains"], run.data["params"], run.pack) in out
|
|
parts["command"] = ok
|
|
passed &= ok
|
|
return {"passed": bool(passed), "kind": exp or "datapoints", "parts": parts, "at": _now(),
|
|
**({"auto_confirmed": True} if auto_manual and exp == "manual" else {})}
|
|
|
|
|
|
def verify(run: Run, step_id: str, confirm: bool = False, manual_ok: bool | None = None) -> dict:
|
|
_writable(run)
|
|
st = run.data["steps"][step_id]
|
|
if st["state"] != "running":
|
|
raise RunError(f"step {step_id} is not running")
|
|
res = _check_result(run, step_def(run, step_id)["verify"], confirm, manual_ok)
|
|
st["verification"] = res
|
|
run.save()
|
|
return res
|
|
|
|
|
|
def complete(run: Run, step_id: str, evidence: dict[str, str] | None = None, note: str = "") -> None:
|
|
_writable(run)
|
|
st = run.data["steps"][step_id]
|
|
if st["state"] != "running":
|
|
raise RunError(f"step {step_id} is not running")
|
|
if not st.get("verification", {}).get("passed"):
|
|
raise RunError(f"step {step_id} has no passing verification")
|
|
s = flow(run).steps[step_id]
|
|
evidence = evidence or {}
|
|
missing = [n for n in s.get("evidence", []) if n not in evidence]
|
|
if missing:
|
|
raise RunError(f"evidence required: {', '.join(missing)}")
|
|
st["evidence"] = {k: _clean(str(v), f"evidence {k}") for k, v in evidence.items()}
|
|
if note.strip():
|
|
st["note"] = _clean(note.strip(), "note")
|
|
st.update(state="done", finished=_now())
|
|
for dp in s.get("produces", []):
|
|
if dp in run.data["datapoints"] and run.data["datapoints"][dp]["state"] not in ("set", "ok"):
|
|
val = st["evidence"].get(dp) or ("true" if flow(run).datapoints[dp]["type"] == "bool" else "(produced)")
|
|
run.data["datapoints"][dp].update(state="set", value=val, how="produced", at=_now())
|
|
_confirm_datapoints(run, step_id, st["verification"]["kind"] if st["verification"]["kind"] != "datapoints" else "check")
|
|
_sync(run)
|
|
run.save()
|
|
|
|
|
|
def fail(run: Run, step_id: str, note: str = "") -> None:
|
|
_writable(run)
|
|
st = run.data["steps"][step_id]
|
|
if st["state"] != "running":
|
|
raise RunError(f"step {step_id} is not running")
|
|
st.update(state="failed", note=_clean(note.strip(), "note"), finished=_now())
|
|
for dp in flow(run).steps[step_id].get("produces", []):
|
|
if dp in run.data["datapoints"] and run.data["datapoints"][dp]["state"] == "nil":
|
|
run.data["datapoints"][dp].update(state="err", error=f"producing step {step_id} failed", at=_now())
|
|
run.save()
|
|
|
|
|
|
def retry(run: Run, step_id: str, acknowledge_partial: bool = False) -> None:
|
|
"""Retry a failed step. Unless it is declared idempotent, the operator must confirm that the earlier
|
|
attempt did not partly apply (a repeated non-idempotent step can double its effect)."""
|
|
_writable(run)
|
|
st = run.data["steps"][step_id]
|
|
if st["state"] != "failed":
|
|
raise RunError(f"step {step_id} has not failed")
|
|
idem = step_def(run, step_id).get("idempotent") or flow(run).steps[step_id].get("idempotent")
|
|
if idem != "idempotent":
|
|
if not acknowledge_partial:
|
|
raise RunError(f"step {step_id} is {idem or 'not declared idempotent'}: confirm that the earlier attempt did not "
|
|
"partly apply before retrying it")
|
|
st.setdefault("retries", []).append({"at": _now(), "acknowledged_partial": True})
|
|
for dp in flow(run).steps[step_id].get("produces", []):
|
|
rec = run.data["datapoints"].get(dp)
|
|
if rec and rec["state"] == "err" and rec.get("error", "").startswith("producing step"):
|
|
run.data["datapoints"][dp] = {"state": "nil"}
|
|
st.pop("verification", None)
|
|
st.update(state="running", started=_now())
|
|
run.save()
|
|
|
|
|
|
# ---- gates ------------------------------------------------------------------------------------------------------------
|
|
|
|
def gate_check(run: Run, gate_id: str, index: int, confirm: bool = False, manual_ok: bool | None = None,
|
|
override_reason: str = "", acknowledge_risk: bool = False, _auto: bool = False) -> dict:
|
|
"""Run one check of a gate and update the gate state (passed when all checks pass)."""
|
|
_writable(run)
|
|
g = flow(run).gates.get(gate_id)
|
|
if g is None:
|
|
raise RunError(f"no such gate: {gate_id}")
|
|
gs = run.data["gates"][gate_id]
|
|
if gs["state"] in DONE_GATE:
|
|
raise RunError(f"gate {gate_id} is already {gs['state']}")
|
|
if index < 0 or index >= len(g["checks"]):
|
|
raise RunError("no such check")
|
|
if not gs.get("overridden") and unmet(run, gate_id):
|
|
_gate_precondition(run, gate_id, override_reason, acknowledge_risk)
|
|
gs["overridden"] = True
|
|
gs.setdefault("started", _now())
|
|
res = _check_result(run, g["checks"][index], confirm, manual_ok, auto_manual=_auto)
|
|
gs["results"][str(index)] = res
|
|
done = all(str(i) in gs["results"] for i in range(len(g["checks"])))
|
|
if done:
|
|
gs["state"] = "passed" if all(r["passed"] for r in gs["results"].values()) else "failed"
|
|
if gs["state"] == "passed":
|
|
_confirm_datapoints(run, gate_id, "check")
|
|
_sync(run)
|
|
run.save()
|
|
return res
|
|
|
|
|
|
def waive(run: Run, gate_id: str, reason: str) -> None:
|
|
"""Waive a gate that declares 'waivable'. The waiver is a recorded deviation."""
|
|
_writable(run)
|
|
g = flow(run).gates.get(gate_id)
|
|
if g is None:
|
|
raise RunError(f"no such gate: {gate_id}")
|
|
w = g.get("waivable")
|
|
if not w:
|
|
raise RunError(f"gate {gate_id} is not waivable; use an override at the dependent step instead")
|
|
if not reason.strip():
|
|
raise RunError("a waiver needs a reason")
|
|
gs = run.data["gates"][gate_id]
|
|
if gs["state"] == "passed":
|
|
raise RunError("the gate has already passed")
|
|
gs.update(state="waived", waiver={"by_actor": w["by"], "who": run.data["operator"], "reason": _clean(reason.strip(), "reason"), "at": _now()})
|
|
_deviate(run, "waiver", gate_id, [f"check {i} not passed" for i, r in gs["results"].items() if not r["passed"]] or ["not evaluated"],
|
|
reason, _risk_text(run, gate_id), True)
|
|
run.save()
|
|
|
|
|
|
# ---- auto-confirm ----------------------------------------------------------------------------------------------------------
|
|
|
|
def _timeout_for(run: Run, id_: str) -> dict | None:
|
|
return flow(run).node(id_).get("timeout") or run.pack.data.get("timeout")
|
|
|
|
|
|
def _elapsed(started: str, now: datetime) -> float:
|
|
return (now - datetime.fromisoformat(started)).total_seconds()
|
|
|
|
|
|
def _stop(run: Run, item: str, why: str) -> dict:
|
|
run.data["auto_confirm"]["stopped"] = {"item": item, "reason": why, "at": _now()}
|
|
run.save()
|
|
return {"action": "stopped", "item": item, "detail": why}
|
|
|
|
|
|
def auto_confirm_tick(run: Run, now: datetime | None = None) -> dict:
|
|
"""One auto-confirm decision. Confirmation is applied when preconditions are met; a timeout stops
|
|
the run or pushes through, recording a deviation. The engine never runs a step's own command."""
|
|
_writable(run)
|
|
if run.data["mode"] != "auto-confirm":
|
|
raise RunError("auto_confirm_tick needs an auto-confirm run")
|
|
ap = run.data["auto_confirm"]
|
|
if ap.get("stopped"):
|
|
return {"action": "stopped", "item": ap["stopped"]["item"], "detail": ap["stopped"]["reason"]}
|
|
now = now or datetime.now(timezone.utc)
|
|
fl = flow(run)
|
|
na = next_actions(run)
|
|
item = na["recommended"]
|
|
if item is None:
|
|
if all(is_complete(run, i) for i in fl.order):
|
|
return {"action": "finished", "item": None, "detail": "every step is done and every gate passed or waived"}
|
|
first = na["blocked"][0]
|
|
return _stop(run, first["id"], "nothing can advance: " + "; ".join(first["unmet"]))
|
|
if fl.kind(item) == "step":
|
|
return _auto_confirm_step(run, item, now)
|
|
return _auto_confirm_gate(run, item, now)
|
|
|
|
|
|
def _on_timeout(run: Run, item: str, cfg: dict, detail: str) -> dict:
|
|
if cfg.get("on_timeout", "stop") == "stop":
|
|
return _stop(run, item, f"timed out after {cfg['seconds']}s: {detail}")
|
|
return {"action": "push-through", "item": item, "detail": detail}
|
|
|
|
|
|
def _auto_confirm_step(run: Run, sid: str, now: datetime) -> dict:
|
|
fl = flow(run)
|
|
s = fl.steps[sid]
|
|
st = run.data["steps"][sid]
|
|
if s.get("auto_confirm") == "manual-only":
|
|
return {"action": "awaiting-operator", "item": sid, "detail": "step is manual-only"}
|
|
if st["state"] == "failed":
|
|
return {"action": "awaiting-operator", "item": sid, "detail": "step failed; retry or stop"}
|
|
alt = st.get("alternative") or recommended_alt(run, sid)
|
|
eff = effective_step(run.pack, fl, sid, alt)
|
|
if not eff.get("reversible", True) and sid not in run.data["auto_confirm"]["preauthorized"]:
|
|
return {"action": "awaiting-operator", "item": sid, "detail": "irreversible and not pre-authorized"}
|
|
if s.get("one_of") and not alt:
|
|
return {"action": "awaiting-operator", "item": sid, "detail": "no alternative is recommended; the operator must choose"}
|
|
if st["state"] == "todo":
|
|
start(run, sid, alt=alt)
|
|
return {"action": "started", "item": sid, "detail": alt or ""}
|
|
res = _check_result(run, eff["verify"], True, None, auto_manual=True)
|
|
st["verification"] = res
|
|
if res["passed"]:
|
|
if s.get("evidence"):
|
|
run.save()
|
|
return {"action": "awaiting-operator", "item": sid, "detail": "evidence must be provided by the operator: " + ", ".join(s["evidence"])}
|
|
complete(run, sid)
|
|
return {"action": "completed", "item": sid, "detail": ""}
|
|
cfg = _timeout_for(run, sid)
|
|
run.save()
|
|
if cfg and _elapsed(st["started"], now) >= cfg["seconds"]:
|
|
out = _on_timeout(run, sid, cfg, "verification did not pass")
|
|
if out["action"] == "push-through":
|
|
st.update(state="pushed", finished=_now())
|
|
_deviate(run, "push-through", sid, ["verification did not pass"],
|
|
f"auto-confirm timeout after {cfg['seconds']}s", _risk_text(run, sid), False)
|
|
for dp in s.get("produces", []):
|
|
rec = run.data["datapoints"].get(dp)
|
|
if rec and rec["state"] in ("nil", "err"):
|
|
rec.pop("error", None)
|
|
rec.update(state="pushed", how="push-through", at=_now())
|
|
run.save()
|
|
return out
|
|
return {"action": "waiting", "item": sid, "detail": "verification did not pass yet"}
|
|
|
|
|
|
def _auto_confirm_gate(run: Run, gid: str, now: datetime) -> dict:
|
|
g = flow(run).gates[gid]
|
|
gs = run.data["gates"][gid]
|
|
gs.setdefault("started", _now())
|
|
gs["results"] = {}
|
|
gs["state"] = "open"
|
|
for i in range(len(g["checks"])):
|
|
gate_check(run, gid, i, confirm=True, _auto=True)
|
|
if gs["state"] == "passed":
|
|
return {"action": "completed", "item": gid, "detail": "gate passed"}
|
|
cfg = _timeout_for(run, gid)
|
|
run.save()
|
|
if cfg and _elapsed(gs["started"], now) >= cfg["seconds"]:
|
|
out = _on_timeout(run, gid, cfg, "gate did not pass")
|
|
if out["action"] == "push-through":
|
|
failed = [f"check {i} not passed" for i, r in gs["results"].items() if not r["passed"]]
|
|
gs.update(state="pushed", pushed={"who": run.data["operator"],
|
|
"reason": f"push-through after {cfg['seconds']}s", "at": _now()})
|
|
_deviate(run, "push-through", gid, failed, f"auto-confirm timeout after {cfg['seconds']}s", _risk_text(run, gid), False)
|
|
run.save()
|
|
return out
|
|
return {"action": "waiting", "item": gid, "detail": "gate did not pass yet"}
|
|
|
|
|
|
# ---- views for renderers: overview, datapoint table with dimensions, artefacts, mode switch -----------------------------
|
|
|
|
def switch_mode(run: Run, mode: str) -> None:
|
|
"""Take over (auto-confirm to manual) or hand over (manual to auto-confirm). The change is recorded."""
|
|
_writable(run)
|
|
if mode not in ("manual", "auto-confirm"):
|
|
raise RunError("mode must be manual or auto-confirm")
|
|
if mode == run.data["mode"]:
|
|
raise RunError(f"the run is already in {mode} mode")
|
|
run.data.setdefault("mode_history", []).append({"from": run.data["mode"], "to": mode, "at": _now(), "who": run.data["operator"]})
|
|
run.data["mode"] = mode
|
|
if mode == "auto-confirm":
|
|
run.data["auto_confirm"].pop("stopped", None)
|
|
run.save()
|
|
|
|
|
|
def overview(run: Run) -> dict:
|
|
"""Progress per stage, datapoint states, deviations: the state of the run at a glance."""
|
|
fl = flow(run)
|
|
stages = []
|
|
|
|
def walk(items: list[dict], depth: int) -> None:
|
|
for st in items:
|
|
steps = fl.members(st["id"], include_gates=False)
|
|
gates = [g for g in fl.members(st["id"]) if fl.kind(g) == "gate"]
|
|
sstates = [run.data["steps"][s]["state"] for s in steps]
|
|
stages.append({
|
|
"id": st["id"], "title": st["title"], "goal": st["goal"], "depth": depth,
|
|
"steps_total": len(steps), "steps_done": sum(x in ("done", "skipped") for x in sstates),
|
|
"steps_pushed": sstates.count("pushed"), "steps_failed": sstates.count("failed"),
|
|
"steps_running": sstates.count("running"),
|
|
"gates": [{"id": g, "title": fl.gates[g]["title"], "state": run.data["gates"][g]["state"]} for g in gates],
|
|
"complete": is_complete(run, st["id"]),
|
|
"unmet": [] if is_complete(run, st["id"]) else unmet(run, st["id"]),
|
|
})
|
|
walk(st.get("stages", []), depth + 1)
|
|
|
|
walk(fl.data["stages"], 0)
|
|
dps: dict[str, int] = {}
|
|
for v in run.data["datapoints"].values():
|
|
dps[v["state"]] = dps.get(v["state"], 0) + 1
|
|
items = [i for i in fl.order]
|
|
done = sum(1 for i in items if is_complete(run, i) and not (fl.kind(i) == "step" and run.data["steps"][i]["state"] == "pushed")
|
|
and not (fl.kind(i) == "gate" and run.data["gates"][i]["state"] == "pushed"))
|
|
pushed = sum(1 for i in items if (fl.kind(i) == "step" and run.data["steps"][i]["state"] == "pushed")
|
|
or (fl.kind(i) == "gate" and run.data["gates"][i]["state"] == "pushed"))
|
|
return {"stages": stages, "datapoints": dps, "deviations": list(run.data["deviations"]),
|
|
"progress": {"done": done, "pushed": pushed, "total": len(items)}, "mode": run.data["mode"],
|
|
"mode_history": run.data.get("mode_history", []), "stopped": run.data["auto_confirm"].get("stopped")}
|
|
|
|
|
|
def dimension_keys(run: Run) -> list[str]:
|
|
keys: set[str] = set()
|
|
for d in flow(run).datapoints.values():
|
|
keys |= set((d.get("dimensions") or {}).keys())
|
|
return sorted(keys)
|
|
|
|
|
|
def datapoint_rows(run: Run, group: str | None = None, flt: str | None = None) -> dict[str, list[dict]]:
|
|
"""Datapoints with their state, optionally filtered by 'dimension:value' and grouped by a dimension.
|
|
|
|
Returns {group label: [rows]}; without grouping there is one group, ''.
|
|
"""
|
|
fl = flow(run)
|
|
fkey, _, fval = (flt or "").partition(":")
|
|
groups: dict[str, list[dict]] = {}
|
|
for d in fl.datapoints.values():
|
|
dims = d.get("dimensions") or {}
|
|
if flt and dims.get(fkey) != fval:
|
|
continue
|
|
rec = run.data["datapoints"][d["id"]]
|
|
row = {"id": d["id"], "label": d["label"], "type": d["type"], "values": d.get("values"), "state": rec["state"],
|
|
"value": rec.get("value"), "how": rec.get("how"), "error": rec.get("error"), "dimensions": dims,
|
|
"ask_in": d.get("ask_in"), "produced_by": d.get("produced_by"), "confirmed_by": d.get("confirmed_by"),
|
|
"editable": not d.get("produced_by")}
|
|
groups.setdefault(dims.get(group, "(none)") if group else "", []).append(row)
|
|
return dict(sorted(groups.items()))
|
|
|
|
|
|
def artefact_rows(run: Run) -> list[dict]:
|
|
"""What the run produced or will produce, and where it lives (a reference, never a secret)."""
|
|
fl = flow(run)
|
|
out = []
|
|
for a in run.pack.data.get("artefacts", []):
|
|
dp = run.data["datapoints"].get(a["location"]["datapoint"], {})
|
|
base = dp.get("value") if dp.get("state") in ("set", "ok") else None
|
|
loc = (base.rstrip("/") + "/" + a["location"]["path"]) if base and a["location"].get("path") else base
|
|
out.append({"id": a["id"], "title": a["title"], "location": loc, "produced_by": a["produced_by"],
|
|
"produced_state": run.data["steps"][a["produced_by"]]["state"] if a["produced_by"] in run.data["steps"] else "?",
|
|
"lives_until": a.get("lives_until")})
|
|
return out
|