"""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 set_datapoint(run: Run, dp_id: str, value: str, how: str = "manual") -> str: """Give a datapoint a value. Invalid input leaves it in state 'err' (never stores it).""" _writable(run) if dp_id not in flow(run).datapoints: raise RunError(f"no such datapoint: {dp_id}") clean, err = _coerce(run, dp_id, value) rec = run.data["datapoints"][dp_id] 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