diff --git a/docs/pack-spec-v0.2.md b/docs/pack-spec-v0.2.md
index ed8dc40..d4f1081 100644
--- a/docs/pack-spec-v0.2.md
+++ b/docs/pack-spec-v0.2.md
@@ -34,6 +34,10 @@ pack
 | **Step** | `todo`, `running`, `done`, `failed` (`skipped` for read-only) | Has it been applied? |
 | **Gate** | `open`, `passed`, `failed`, `waived` | Is the quality condition met? |
 
+A datapoint becomes `ok` when the step or gate named in its `confirmed_by` passes. A gate must not
+require a datapoint (`datapoints_ok`) that it is itself `confirmed_by`: that is circular and the validator
+rejects it.
+
 A v0.1 `parameter` is a datapoint that has a value type and is asked as a
 question. Steps `uses:` datapoints; steps may `produces:` datapoints
 (for example an artefact location).
diff --git a/docs/run-spec.md b/docs/run-spec.md
index 6eb1f66..d8bffab 100644
--- a/docs/run-spec.md
+++ b/docs/run-spec.md
@@ -60,3 +60,41 @@ back, as documentation or for inspection when the engine cannot reach the system
 detect every secret, so the operator stays responsible. For `output-contains` steps the output can
 be **pasted for verification** (`source: pasted-for-verification`), which checks the `contains` text
 without running anything. Exit-code checks cannot be verified from pasted output.
+
+## Run record `runbook-run/v0.2` (packs of `runbook-pack/v0.2`)
+
+Implemented in `rtut/run2.py` (graph in `rtut/flow.py`). It adds to the v0.1 record:
+
+| Field | Meaning |
+| --- | --- |
+| `mode` | `manual` or `autopilot` |
+| `datapoints.<id>` | `state` nil, set, err or ok; `value`, `how` (initial, manual, produced, check, manual confirmation), `error` |
+| `steps.<id>` | as v0.1, plus `alternative` (chosen `one_of` branch), `authorized_by`, `pushed_through` |
+| `gates.<id>` | `state` open, passed, failed or waived; per-check `results`; `waiver` |
+| `deviations[]` | every override, waiver and push-through (see `docs/pack-spec-v0.2.md`) |
+| `autopilot` | `preauthorized` irreversible steps, `authorized_by`, and `stopped` if a timeout stopped the run |
+
+Rules the engine enforces:
+
+- **Preconditions** of an item are its `requires` (and those of enclosing stages), the `set`/`ok`
+  state of the datapoints it `uses`, and, for a gate, completion of the steps of its stage.
+- **Recommended next action**: the running item first, else the first ready item in the pack's canonical
+  order. `next_actions(run)` returns the recommended item, all ready items (parallel work) and the blocked
+  ones with what they wait for.
+- **Alternatives**: a `one_of` step is started with an alternative (default: the first whose
+  `recommended_when` matches the datapoints).
+- **Datapoints**: invalid input leaves the datapoint `err` and is never stored; a step's `produces` become `set`
+  when it completes; `confirmed_by` promotes them to `ok` when that step or gate passes.
+- **Override** (manual mode only): starting an item with unmet preconditions needs a reason and an acknowledged
+  downstream risk; it is recorded as a deviation.
+- **Waiver**: only for gates that declare `waivable`, with a reason; recorded as a deviation.
+- **Confirmation**: irreversible steps need the exact confirmation phrase at start. In autopilot they run only
+  if the run pre-authorized them, and never if the step is `autopilot: manual-only`.
+- **Autopilot** (`autopilot_tick`): the engine applies confirmation when preconditions are met: it starts the
+  recommended item, runs its checks (manual checks are auto-confirmed and marked so) and completes it. It
+  never runs a step's own command; the work is done by the operator or, later, by opt-in execution. A step
+  with `evidence` waits for the operator. If checks keep failing until the item's `timeout` (or the pack
+  default), the run stops, or with `on_timeout: push-through` the item is marked done with
+  `pushed_through` and a deviation is recorded.
+- **Deviations and exercise**: a run with any deviation or pushed-through step can be finalized but never exercises a pack.
+  `finalize(succeeded)` also needs every gate passed or waived.
diff --git a/rtut/flow.py b/rtut/flow.py
new file mode 100644
index 0000000..4f1003c
--- /dev/null
+++ b/rtut/flow.py
@@ -0,0 +1,69 @@
+"""The flow graph of a runbook-pack/v0.2: stages, steps, alternatives, gates, canonical order."""
+from __future__ import annotations
+
+
+class Flow:
+    def __init__(self, data: dict):
+        self.data = data
+        self.stages: dict[str, dict] = {}
+        self.steps: dict[str, dict] = {}
+        self.gates: dict[str, dict] = {}
+        self.alts: dict[str, str] = {}            # alternative id -> parent step id
+        self.parent: dict[str, str | None] = {}   # flow id -> enclosing stage id
+        self.order: list[str] = []                # canonical order of steps and gates
+        self.datapoints = {d["id"]: d for d in data.get("datapoints", [])}
+        self.actors = {a["id"]: a for a in data.get("actors", [])}
+        self._walk(data["stages"], None)
+
+    def _walk(self, stages: list[dict], parent: str | None) -> None:
+        for st in stages:
+            self.stages[st["id"]] = st
+            self.parent[st["id"]] = parent
+            for s in st.get("steps", []):
+                self.steps[s["id"]] = s
+                self.parent[s["id"]] = st["id"]
+                self.order.append(s["id"])
+                for a in s.get("one_of", []):
+                    self.alts[a["id"]] = s["id"]
+            self._walk(st.get("stages", []), st["id"])
+            for g in st.get("gates", []):
+                self.gates[g["id"]] = g
+                self.parent[g["id"]] = st["id"]
+                self.order.append(g["id"])
+
+    # -- structure ------------------------------------------------------------------------
+    def kind(self, id_: str) -> str:
+        return "stage" if id_ in self.stages else "step" if id_ in self.steps else "gate" if id_ in self.gates else "?"
+
+    def node(self, id_: str) -> dict:
+        return self.stages.get(id_) or self.steps.get(id_) or self.gates[id_]
+
+    def resolve(self, ref: str) -> str | None:
+        if ref.startswith("gate:"):
+            return ref[5:] if ref[5:] in self.gates else None
+        return ref if ref in self.stages or ref in self.steps else None
+
+    def ancestors(self, id_: str) -> list[str]:
+        out, p = [], self.parent.get(id_)
+        while p:
+            out.append(p)
+            p = self.parent.get(p)
+        return out
+
+    def members(self, stage_id: str, include_gates: bool = True) -> list[str]:
+        """Steps (and gates) inside a stage, recursively."""
+        out = []
+        st = self.stages[stage_id]
+        out += [s["id"] for s in st.get("steps", [])]
+        for child in st.get("stages", []):
+            out += self.members(child["id"], include_gates)
+        if include_gates:
+            out += [g["id"] for g in st.get("gates", [])]
+        return out
+
+    def requires(self, id_: str) -> list[str]:
+        """Own requires plus those of every enclosing stage, as resolved flow ids."""
+        refs = list(self.node(id_).get("requires", []))
+        for a in self.ancestors(id_):
+            refs += self.stages[a].get("requires", [])
+        return [r for r in (self.resolve(x) for x in refs) if r]
diff --git a/rtut/pack.py b/rtut/pack.py
index d703df0..a8afad1 100644
--- a/rtut/pack.py
+++ b/rtut/pack.py
@@ -112,7 +112,7 @@ def load(path: Path) -> Pack:
 
 def render(command: str, values: dict[str, str], pack: Pack) -> str:
     """Substitute parameters into a command template. Unknown or unset -> error."""
-    params = {p["id"]: p for p in pack.data.get("parameters", [])}
+    params = {p["id"]: p for p in (pack.data.get("parameters") or pack.data.get("datapoints") or [])}
 
     def sub(m: re.Match) -> str:
         name = m.group(1)
diff --git a/rtut/run.py b/rtut/run.py
index 78321d6..353447d 100644
--- a/rtut/run.py
+++ b/rtut/run.py
@@ -72,8 +72,11 @@ def _writable(run: Run) -> None:
 def create(pack: Pack, operator: str, params: dict[str, str] | None = None) -> Run:
     if not pack.ok or pack.data.get("engine") != "native":
         raise RunError("only valid native packs can be run")
+    if pack.spec == "runbook-pack/v0.2":
+        from . import run2
+        return run2.create(pack, operator, params)
     if pack.spec != "runbook-pack/v0.1":
-        raise RunError(f"{pack.spec} packs cannot be run yet (RBT-WP-0007)")
+        raise RunError(f"{pack.spec} packs cannot be run")
     operator = _clean(operator.strip(), "operator")
     if not re.fullmatch(r"\S+", operator):
         raise RunError("operator must be one word (name or handle)")
@@ -239,6 +242,10 @@ def exercise_update(run: Run, today: date | None = None) -> tuple[str, str]:
     """
     if any(v["state"] != "done" for v in run.data["steps"].values()):
         raise RunError("every step must be done (not skipped) to exercise the pack")
+    if run.data.get("deviations"):
+        raise RunError("a run with deviations (overrides, waivers, push-through) cannot exercise the pack")
+    if any(v.get("pushed_through") for v in run.data["steps"].values()):
+        raise RunError("a run with pushed-through steps cannot exercise the pack")
     today = today or date.today()
     old = run.pack.path.read_text()
     new = STATUS_LINE.sub(f"exercise_status: exercised {today.isoformat()} by {run.data['operator']}",
@@ -337,6 +344,8 @@ def finalize(run: Run, outcome: str, note: str = "") -> None:
     states = [v["state"] for v in run.data["steps"].values()]
     if outcome == "succeeded" and any(x != "done" for x in states):
         raise RunError("a run succeeds only when every step is done (skipped steps do not count)")
+    if outcome == "succeeded" and any(g["state"] not in ("passed", "waived") for g in run.data.get("gates", {}).values()):
+        raise RunError("a run succeeds only when every gate has passed or been waived")
     if outcome == "failed" and "failed" not in states:
         raise RunError("a run can be finished as failed only when a step has failed; otherwise abandon it")
     run.data["outcome"] = outcome
diff --git a/rtut/run2.py b/rtut/run2.py
new file mode 100644
index 0000000..c218443
--- /dev/null
+++ b/rtut/run2.py
@@ -0,0 +1,539 @@
+"""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 autopilot 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, render
+from .run import Run, RunError, _clean, _now, runs_dir  # noqa: F401  (re-exported for callers)
+
+DEFAULT_PHRASE = "I understand this cannot be undone"
+VERIFY_TIMEOUT = 60
+DONE_STEP = ("done", "skipped")
+DONE_GATE = ("passed", "waived")
+
+
+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) -> 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", "autopilot"):
+        raise RunError("mode must be manual or autopilot")
+    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": [],
+        "autopilot": {"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"):
+                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["autopilot"]["preauthorized"]
+        if step_id in pre and run.data["mode"] == "autopilot":
+            st["authorized_by"] = run.data["autopilot"]["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")
+        cmd = render(ck["command"], run.data["params"], run.pack)
+        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) -> None:
+    _writable(run)
+    st = run.data["steps"][step_id]
+    if st["state"] != "failed":
+        raise RunError(f"step {step_id} has not failed")
+    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()
+
+
+# ---- autopilot ----------------------------------------------------------------------------------------------------------
+
+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["autopilot"]["stopped"] = {"item": item, "reason": why, "at": _now()}
+    run.save()
+    return {"action": "stopped", "item": item, "detail": why}
+
+
+def autopilot_tick(run: Run, now: datetime | None = None) -> dict:
+    """One autopilot 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"] != "autopilot":
+        raise RunError("autopilot_tick needs an autopilot run")
+    ap = run.data["autopilot"]
+    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 _autopilot_step(run, item, now)
+    return _autopilot_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 _autopilot_step(run: Run, sid: str, now: datetime) -> dict:
+    fl = flow(run)
+    s = fl.steps[sid]
+    st = run.data["steps"][sid]
+    if s.get("autopilot") == "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["autopilot"]["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="done", finished=_now(), pushed_through=True)
+            _deviate(run, "push-through", sid, ["verification did not pass"],
+                     f"autopilot 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"] == "nil":
+                    rec.update(state="err", error="pushed through without verification", at=_now())
+            run.save()
+        return out
+    return {"action": "waiting", "item": sid, "detail": "verification did not pass yet"}
+
+
+def _autopilot_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="waived", waiver={"by_actor": "autopilot", "who": run.data["operator"],
+                                              "reason": f"push-through after {cfg['seconds']}s", "at": _now()})
+            _deviate(run, "push-through", gid, failed, f"autopilot 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"}
diff --git a/rtut/v2.py b/rtut/v2.py
index 0828f8b..a36e8d5 100644
--- a/rtut/v2.py
+++ b/rtut/v2.py
@@ -151,6 +151,11 @@ def _check_refs(m: _Model, data: dict) -> None:
             m.errs.append(f"datapoint {d['id']}: produced_by names unknown step {d['produced_by']}")
         if d.get("confirmed_by") and d["confirmed_by"] not in steps | gates:
             m.errs.append(f"datapoint {d['id']}: confirmed_by names unknown step or gate")
+    for g_id in gates:                     # a gate that requires a datapoint to be ok cannot be what confirms it
+        for ck in m.flow[g_id]["checks"]:
+            for d in ck.get("datapoints_ok", []):
+                if dps.get(d, {}).get("confirmed_by") == g_id:
+                    m.errs.append(f"gate {g_id}: checks datapoint {d} that it is itself confirmed_by (circular)")
     for a in data.get("artefacts", []):
         if a["location"]["datapoint"] not in dps:
             m.errs.append(f"artefact {a['id']}: location names unknown datapoint")
diff --git a/runbooks/showcase-lighthouse/pack.v0.2.yaml b/runbooks/showcase-lighthouse/pack.v0.2.yaml
index dd4290c..1e11ae3 100644
--- a/runbooks/showcase-lighthouse/pack.v0.2.yaml
+++ b/runbooks/showcase-lighthouse/pack.v0.2.yaml
@@ -33,7 +33,7 @@ datapoints:
   - {id: weather, label: Tonight's weather, type: enum, values: [clear, fog, storm], default: fog, dimensions: {kind: condition}, ask_in: dusk}
   - {id: motto, label: Motto, type: text, default: Steady light for those at sea, dimensions: {actor: keeper, kind: identity}, ask_in: dusk}
   - {id: logdir, label: Scratch folder, type: path, pattern: "/tmp/rtut-[a-z0-9-]+", default: /tmp/rtut-lighthouse, dimensions: {kind: location, subsystem: logbook}, ask_in: dusk}
-  - {id: lamp_lit, label: Lamp is lit, type: bool, dimensions: {actor: keeper, kind: state}, produced_by: light-lamp, confirmed_by: lamp-check}
+  - {id: lamp_lit, label: Lamp is lit, type: bool, dimensions: {actor: keeper, kind: state}, produced_by: light-lamp, confirmed_by: light-lamp}
 
 stages:
   - id: dusk
diff --git a/tests/test_run2.py b/tests/test_run2.py
new file mode 100644
index 0000000..87fe20c
--- /dev/null
+++ b/tests/test_run2.py
@@ -0,0 +1,316 @@
+import copy
+import subprocess
+import uuid
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+import pytest
+import yaml
+
+from rtut import pack as P, run as R1, run2 as R
+
+ROOT = Path(__file__).resolve().parent.parent
+SRC = yaml.safe_load((ROOT / "runbooks/showcase-lighthouse/pack.v0.2.yaml").read_text())
+PHRASE = "I understand this cannot be undone"
+
+
+def mkpack(tmp_path, mutate=None):
+    d = copy.deepcopy(SRC)
+    if mutate:
+        mutate(d)
+    f = tmp_path / "runbooks" / "lh" / "pack.yaml"
+    f.parent.mkdir(parents=True, exist_ok=True)
+    f.write_text(yaml.safe_dump(d, sort_keys=False))
+    pk = P.load(f)
+    assert pk.ok, pk.errors
+    return pk
+
+
+_SCRATCH: list[str] = []
+
+
+@pytest.fixture(autouse=True)
+def _cleanup_scratch():
+    yield
+    import shutil
+    while _SCRATCH:
+        shutil.rmtree(_SCRATCH.pop(), ignore_errors=True)
+
+
+def new(tmp_path, mutate=None, **kw):
+    pk = mkpack(tmp_path, mutate)
+    logdir = f"/tmp/rtut-{uuid.uuid4().hex[:10]}"
+    _SCRATCH.append(logdir)
+    vals = {"logdir": logdir, **kw.pop("values", {})}
+    return R.create(pk, "tester", vals, **kw)
+
+
+def do_step(r, sid, alt=None, phrase=None, manual=True, run_cmd=True, evidence=None):
+    """What an operator does: start, run the command, check, mark done."""
+    R.start(r, sid, alt=alt, phrase=phrase)
+    cmd = R.command_for(r, sid)
+    if cmd and run_cmd:
+        subprocess.run(cmd, shell=True, check=True, capture_output=True)
+    R.verify(r, sid, confirm=True, manual_ok=manual)
+    R.complete(r, sid, evidence)
+
+
+def pass_gate(r, gid):
+    for i, ck in enumerate(R.flow(r).gates[gid]["checks"]):
+        R.gate_check(r, gid, i, confirm=True, manual_ok=True)
+
+
+# ---- datapoints -----------------------------------------------------------------------------------
+
+def test_create_sets_datapoint_states(tmp_path):
+    r = new(tmp_path)
+    dp = r.data["datapoints"]
+    assert dp["keeper_name"]["state"] == "set" and dp["keeper_name"]["value"] == "Ada"
+    assert dp["lamp_lit"]["state"] == "nil"
+    assert r.data["params"]["weather"] == "fog"
+
+
+def test_invalid_datapoint_is_err_and_not_stored(tmp_path):
+    r = new(tmp_path)
+    assert R.set_datapoint(r, "weather", "hail") == "err"
+    assert "value" not in r.data["datapoints"]["weather"] and "weather" not in r.data["params"]
+    assert R.set_datapoint(r, "weather", "storm") == "set"
+    assert R.set_datapoint(r, "logdir", "/etc") == "err"
+    assert R.set_datapoint(r, "motto", 'x"; rm -rf /') == "err"
+    with pytest.raises(R.RunError):
+        R.create(mkpack(tmp_path), "t", {"weather": "hail"})
+
+
+# ---- readiness, recommendation, parallelism, alternatives ---------------------------------------------
+
+def test_recommended_next_is_first_ready_in_canonical_order_and_others_are_parallel(tmp_path):
+    r = new(tmp_path)
+    na = R.next_actions(r)
+    assert na["recommended"] == "read-the-sign"
+    assert na["ready"] == ["read-the-sign", "light-lamp", "read-the-sky"]           # parallel, canonical order
+    blocked = {b["id"]: b["unmet"] for b in na["blocked"]}
+    assert "lamp-check" in blocked and "warn" in blocked and "close-log" in blocked
+    R.start(r, "light-lamp")                                                        # any ready item may be taken
+    assert R.next_actions(r)["recommended"] == "light-lamp"                         # a running item comes first
+
+
+def test_alternatives_recommended_by_datapoint_and_explicit_choice(tmp_path):
+    for weather, alt in (("fog", "sound-horn"), ("clear", "wave-flags"), ("storm", "shelter")):
+        r = new(tmp_path, values={"weather": weather})
+        assert R.recommended_alt(r, "warn") == alt
+    r = new(tmp_path)
+    for s in ("read-the-sign", "light-lamp", "read-the-sky"):
+        do_step(r, s)
+    pass_gate(r, "lamp-check")
+    with pytest.raises(R.RunError, match="choose one of"):
+        R.start(r, "warn", alt="dance")
+    R.start(r, "warn", alt="wave-flags")                                             # operator may override the recommendation
+    assert R.command_for(r, "warn").startswith("echo") and "signals with flags" in R.command_for(r, "warn")
+    assert r.data["steps"]["warn"]["alternative"] == "wave-flags"
+
+
+def test_datapoint_ok_via_confirmed_by_and_gate_flow(tmp_path):
+    r = new(tmp_path)
+    assert R.unmet(r, "lamp-check")                                                  # stage steps not done yet
+    do_step(r, "light-lamp")
+    dp = r.data["datapoints"]["lamp_lit"]
+    assert dp["state"] == "ok" and dp["how"]                                          # produced, then confirmed by the step's own check
+    for s in ("read-the-sign", "read-the-sky"):
+        do_step(r, s)
+    assert R.unmet(r, "lamp-check") == []
+    R.gate_check(r, "lamp-check", 0, manual_ok=True)
+    assert r.data["gates"]["lamp-check"]["state"] == "open"                          # not all checks evaluated yet
+    R.gate_check(r, "lamp-check", 1)                                                  # datapoints_ok is automatic
+    assert r.data["gates"]["lamp-check"]["state"] == "passed"
+    assert "warn" in R.next_actions(r)["ready"]
+
+
+def test_failed_step_marks_produced_datapoint_err_and_retry_clears_it(tmp_path):
+    r = new(tmp_path)
+    R.start(r, "light-lamp"); R.fail(r, "light-lamp", "no oil")
+    assert r.data["datapoints"]["lamp_lit"]["state"] == "err"
+    R.retry(r, "light-lamp")
+    assert r.data["datapoints"]["lamp_lit"]["state"] == "nil"
+
+
+# ---- overrides, waivers, deviations ------------------------------------------------------------------
+
+def test_override_needs_reason_and_acknowledged_risk_and_is_recorded(tmp_path):
+    r = new(tmp_path)
+    with pytest.raises(R.RunError, match="Risk:"):
+        R.start(r, "warn", alt="sound-horn")                                          # gate not passed: names unmet + risk
+    with pytest.raises(R.RunError, match="acknowledge"):
+        R.start(r, "warn", alt="sound-horn", override_reason="need to see the horn text")
+    R.start(r, "warn", alt="sound-horn", override_reason="need to see the horn text", acknowledge_risk=True)
+    dev = r.data["deviations"][0]
+    assert dev["kind"] == "override" and dev["item"] == "warn" and dev["mode"] == "manual"
+    assert any("lamp-check" in u for u in dev["unmet"]) and dev["risk_acknowledged"] and dev["reason"]
+    assert "not complete" in dev["unmet"][0]
+
+
+def test_waiver_only_where_declared_with_reason_and_recorded(tmp_path):
+    r = new(tmp_path)
+    for s in ("read-the-sign", "light-lamp", "read-the-sky"):
+        do_step(r, s)
+    R.gate_check(r, "lamp-check", 0, manual_ok=False)
+    R.gate_check(r, "lamp-check", 1)
+    assert r.data["gates"]["lamp-check"]["state"] == "failed"
+    with pytest.raises(R.RunError, match="reason"):
+        R.waive(r, "lamp-check", "  ")
+    R.waive(r, "lamp-check", "harbour master accepts the risk")
+    assert r.data["gates"]["lamp-check"]["state"] == "waived"
+    assert r.data["deviations"][-1]["kind"] == "waiver" and "check 0 not passed" in r.data["deviations"][-1]["unmet"]
+    assert "warn" in R.next_actions(r)["ready"]                                       # a waived gate unblocks dependents
+    with pytest.raises(R.RunError, match="not waivable"):
+        R.waive(r, "night-on-record", "try")
+
+
+def test_deviations_block_exercise_but_not_finalize(tmp_path):
+    r = new(tmp_path)
+    R.start(r, "warn", alt="sound-horn", override_reason="probe", acknowledge_risk=True)
+    r.data["steps"]["warn"]["state"] = "failed"
+    assert r.data["deviations"]
+    with pytest.raises(R1.RunError, match="deviations"):
+        for k in r.data["steps"]:
+            r.data["steps"][k]["state"] = "done"
+        R1.exercise_update(r)
+
+
+# ---- confirmation styles ------------------------------------------------------------------------------
+
+def test_irreversible_step_needs_the_typed_phrase(tmp_path):
+    r = new(tmp_path)
+    assert R.confirmation_style(r, "read-the-sign") == "confirm" and R.confirmation_style(r, "close-log") == "phrase"
+    for s in ("read-the-sign", "light-lamp", "read-the-sky"):
+        do_step(r, s)
+    pass_gate(r, "lamp-check")
+    do_step(r, "warn")
+    do_step(r, "write-logbook")
+    with pytest.raises(R.RunError, match="confirmation phrase"):
+        R.start(r, "close-log")
+    with pytest.raises(R.RunError, match="confirmation phrase"):
+        R.start(r, "close-log", phrase="yes")
+    R.start(r, "close-log", phrase=PHRASE)
+    assert r.data["steps"]["close-log"]["authorized_by"] == "tester"
+
+
+def test_preauthorization_only_for_irreversible_steps(tmp_path):
+    with pytest.raises(R.RunError, match="only irreversible"):
+        new(tmp_path, mode="autopilot", preauthorize=["light-lamp"])
+
+
+# ---- autopilot ------------------------------------------------------------------------------------------------
+
+def drive(r, ticks=60, now=None, operator=None):
+    log = []
+    for _ in range(ticks):
+        out = R.autopilot_tick(r, now)
+        log.append(out)
+        if out["action"] in ("finished", "stopped", "awaiting-operator", "waiting"):
+            break
+        if out["action"] == "started" and operator:
+            operator(out["item"])                   # the operator runs the step's command while autopilot confirms
+    return log
+
+
+def test_autopilot_confirms_when_preconditions_are_met_and_waits_for_the_operator_on_manual_only(tmp_path):
+    r = new(tmp_path, mode="autopilot")
+
+    def operator(sid):
+        cmd = R.command_for(r, sid)
+        if cmd:
+            subprocess.run(cmd, shell=True, check=True, capture_output=True)
+
+    log = drive(r, operator=operator)
+    assert log[-1] == {"action": "awaiting-operator", "item": "close-log", "detail": "step is manual-only"}
+    done = [x["item"] for x in log if x["action"] == "completed"]
+    assert done[:3] == ["read-the-sign", "light-lamp", "read-the-sky"] and "lamp-check" in done and "write-logbook" in done
+    assert r.data["steps"]["warn"]["alternative"] == "sound-horn"                    # recommended by the fog datapoint
+    assert r.data["gates"]["lamp-check"]["state"] == "passed"
+    assert r.data["steps"]["read-the-sign"]["verification"].get("auto_confirmed")
+    assert r.data["deviations"] == []
+    R.start(r, "close-log", phrase=PHRASE)                                            # the operator takes over the last step
+    subprocess.run(R.command_for(r, "close-log"), shell=True, check=True, capture_output=True)
+    R.verify(r, "close-log", confirm=True); R.complete(r, "close-log")
+    assert R.autopilot_tick(r)["action"] == "completed"                               # gate night-on-record
+    assert R.autopilot_tick(r)["action"] == "finished"
+    R1.finalize(r, "succeeded")
+    assert r.data["outcome"] == "succeeded" and not r.data["deviations"]
+
+
+def test_autopilot_timeout_stop(tmp_path):
+    def m(d):
+        d["timeout"] = {"seconds": 30, "on_timeout": "stop"}
+    r = new(tmp_path, m, mode="autopilot")
+    log = drive(r)                                     # write-logbook's file check cannot pass: nobody ran its command
+    assert log[-1]["action"] == "waiting" and log[-1]["item"] == "write-logbook"
+    late = datetime.now(timezone.utc) + timedelta(seconds=31)
+    out = R.autopilot_tick(r, late)
+    assert out["action"] == "stopped" and "timed out after 30s" in out["detail"]
+    assert R.autopilot_tick(r)["action"] == "stopped"                                # stays stopped
+    assert r.data["deviations"] == []                                                # stopping is not a deviation
+
+
+def test_autopilot_push_through_records_deviation_and_moves_on(tmp_path):
+    def m(d):
+        d["timeout"] = {"seconds": 30, "on_timeout": "push-through"}
+    r = new(tmp_path, m, mode="autopilot")
+    drive(r)
+    late = datetime.now(timezone.utc) + timedelta(seconds=31)
+    out = R.autopilot_tick(r, late)
+    assert out["action"] == "push-through" and out["item"] == "write-logbook"
+    assert r.data["steps"]["write-logbook"]["state"] == "done" and r.data["steps"]["write-logbook"]["pushed_through"]
+    dev = r.data["deviations"][0]
+    assert dev["kind"] == "push-through" and dev["mode"] == "autopilot" and dev["risk_acknowledged"] is False
+    nxt = R.autopilot_tick(r, late)                                                    # continues to the manual-only step
+    assert nxt["item"] == "close-log" and nxt["action"] == "awaiting-operator"
+    with pytest.raises(R1.RunError, match="deviations"):
+        for k in r.data["steps"]:
+            r.data["steps"][k]["state"] = "done"
+        R1.exercise_update(r)
+
+
+def test_autopilot_gate_push_through_and_stop(tmp_path):
+    def m(d):
+        d["stages"][0]["gates"][0]["checks"].append({"done_when": "always fails", "command": "false", "expect": "exit-0"})
+        d["stages"][0]["gates"][0]["timeout"] = {"seconds": 10, "on_timeout": "push-through"}
+    r = new(tmp_path, m, mode="autopilot")
+    drive(r)
+    assert r.data["gates"]["lamp-check"]["state"] == "failed"        # all checks evaluated, one failing
+    out = R.autopilot_tick(r, datetime.now(timezone.utc) + timedelta(seconds=11))
+    assert out["action"] == "push-through" and r.data["gates"]["lamp-check"]["state"] == "waived"
+    assert r.data["deviations"][0]["item"] == "lamp-check" and r.data["deviations"][0]["kind"] == "push-through"
+
+
+def test_autopilot_never_confirms_irreversible_steps_unless_preauthorized(tmp_path):
+    def m(d):
+        for st in d["stages"]:
+            for s in st.get("steps", []):
+                s.pop("autopilot", None)                                             # allow automation of close-log
+    r = new(tmp_path, m, mode="autopilot")
+    drive(r, operator=lambda sid: (subprocess.run(R.command_for(r, sid), shell=True, check=True, capture_output=True) if R.command_for(r, sid) else None))
+    last = R.autopilot_tick(r)
+    assert last["action"] == "awaiting-operator" and "not pre-authorized" in last["detail"]
+    r2 = new(tmp_path, m, mode="autopilot", preauthorize=["close-log"])
+    drive(r2, operator=lambda sid: (subprocess.run(R.command_for(r2, sid), shell=True, check=True, capture_output=True) if R.command_for(r2, sid) else None))
+    assert r2.data["steps"]["close-log"]["state"] in ("running", "done")
+    assert r2.data["steps"]["close-log"]["authorized_by"] == "tester"
+
+
+def test_manual_mode_rejects_autopilot_and_autopilot_rejects_overrides(tmp_path):
+    with pytest.raises(R.RunError, match="autopilot run"):
+        R.autopilot_tick(new(tmp_path))
+    r = new(tmp_path, mode="autopilot")
+    with pytest.raises(R.RunError, match="manual mode"):
+        R.start(r, "warn", alt="sound-horn", override_reason="x", acknowledge_risk=True)
+
+
+def test_finished_v02_run_is_read_only_and_scores_count_it(tmp_path):
+    r = new(tmp_path)
+    R1.finalize(r, "abandoned")
+    with pytest.raises(R.RunError, match="finished"):
+        R.start(r, "read-the-sign")
+    with pytest.raises(R.RunError, match="finished"):
+        R.set_datapoint(r, "weather", "clear")
+    assert R1.scorecard(r.pack)["abandoned"] == 1
+    assert R1.load(r.pack, r.data["run_id"]).data["spec"] == "runbook-run/v0.2"      # reloads through the shared loader
diff --git a/tests/test_v2.py b/tests/test_v2.py
index dc7c13e..f1b8f00 100644
--- a/tests/test_v2.py
+++ b/tests/test_v2.py
@@ -35,9 +35,10 @@ def test_showcase_v02_is_valid_and_loads_through_pack_load():
     assert pk.spec == "runbook-pack/v0.2"
 
 
-def test_v02_packs_cannot_be_run_yet():
-    with pytest.raises(R.RunError, match="cannot be run yet"):
-        R.create(P.load(FILE), "tester")
+def test_v02_packs_run_through_run_create():
+    import uuid
+    r = R.create(P.load(FILE), "tester", {"logdir": f"/tmp/rtut-{uuid.uuid4().hex[:8]}"})
+    assert r.data["spec"] == "runbook-run/v0.2" and r.data["mode"] == "manual"
 
 
 CASES = [
@@ -104,3 +105,8 @@ def test_legacy_v02_needs_only_schema():
     d = {"spec": "runbook-pack/v0.2", "id": "x.y", "title": "Legacy x", "owner": "o",
          "exercise_status": "unexercised", "engine": "legacy-command", "legacy": {"command": ["true"]}}
     assert v2.validate(d) == []
+
+
+def test_gate_cannot_confirm_a_datapoint_it_requires():
+    out = errs(lambda d: next(x for x in d["datapoints"] if x["id"] == "lamp_lit").update(confirmed_by="lamp-check"))
+    assert any("circular" in e for e in out), out
diff --git a/workplans/RBT-WP-0002-runbook-pack-spec-and-validator.md b/workplans/RBT-WP-0002-runbook-pack-spec-and-validator.md
index 4fc7e70..657983c 100644
--- a/workplans/RBT-WP-0002-runbook-pack-spec-and-validator.md
+++ b/workplans/RBT-WP-0002-runbook-pack-spec-and-validator.md
@@ -105,7 +105,7 @@ state_hub_task_id: "6740457f-81a7-5761-a768-b9b0faee7552"
 scratch logbook under /tmp/rtut-*). `pack.yaml` (v0.1) runs today and is the test bed for
 UI refinement. `pack.v0.2.yaml` expresses the same story in the proposed v0.2 model
 (stages, parallel steps, alternatives, gates, actors, datapoints, artefacts, irreversible
-confirmation). It now validates against the v0.2 schema and semantic checks. It cannot be run until RBT-WP-0007 lands.
+confirmation). It validates against the v0.2 schema and semantic checks and can be run by the v0.2 engine (`rtut/run2.py`); a UI for it is RBT-WP-0004.
 
 ## Acceptance Criteria
 
diff --git a/workplans/RBT-WP-0004-local-runbook-ui.md b/workplans/RBT-WP-0004-local-runbook-ui.md
index 72e2e08..a2e2c44 100644
--- a/workplans/RBT-WP-0004-local-runbook-ui.md
+++ b/workplans/RBT-WP-0004-local-runbook-ui.md
@@ -91,12 +91,12 @@ step instead of the top of the page.
 
 ```task
 id: RBT-WP-0004-T06
-status: wait
+status: todo
 priority: high
 state_hub_task_id: "261bf74f-f5cd-55bb-9a96-12767418c980"
 ```
 
-Depends on RBT-WP-0007. The serial guided flow is the main interface: stages with
+Unblocked: RBT-WP-0007 (engine) is finished. The serial guided flow is the main interface: stages with
 progress, the recommended next action, other ready actions, framing shown at the
 start of each stage, and gates shown at stage exit.
 
@@ -104,12 +104,12 @@ start of each stage, and gates shown at stage exit.
 
 ```task
 id: RBT-WP-0004-T07
-status: wait
+status: todo
 priority: high
 state_hub_task_id: "16db09c7-7707-5d17-aab3-5d8eab2c1fce"
 ```
 
-Depends on RBT-WP-0007. A random-access table of datapoints (with `nil/set/err/ok`
+Unblocked: RBT-WP-0007 (engine) is finished. A random-access table of datapoints (with `nil/set/err/ok`
 state) grouped or filtered by dimension (actor, role, privilege, subsystem, kind),
 editable independently of the flow, plus the artefacts view.
 
@@ -117,12 +117,12 @@ editable independently of the flow, plus the artefacts view.
 
 ```task
 id: RBT-WP-0004-T08
-status: wait
+status: todo
 priority: high
 state_hub_task_id: "23d3f5cc-e844-5077-9a4a-0435c189d98b"
 ```
 
-Depends on RBT-WP-0007. State transparency at a glance: stage progress, gate states
+Unblocked: RBT-WP-0007 (engine) is finished. State transparency at a glance: stage progress, gate states
 and waivers, datapoint completeness by dimension.
 
 ## Visual design
@@ -154,12 +154,12 @@ Confirm button for reversible steps, typed phrase for irreversible ones, per the
 
 ```task
 id: RBT-WP-0004-T11
-status: wait
+status: todo
 priority: high
 state_hub_task_id: "99320417-8261-556b-8932-4947cfd8ce11"
 ```
 
-Depends on RBT-WP-0007 T06-T07. Choose manual or autopilot per run (with timeout and
+Unblocked: RBT-WP-0007 T06-T07 are finished. Choose manual or autopilot per run (with timeout and
 push-through controls). In manual mode, an override dialog lists unmet preconditions and
 the downstream risk before continuing; deviations show in the overview.
 
diff --git a/workplans/RBT-WP-0007-engine-v0-2-graph-datapoints-and-gates.md b/workplans/RBT-WP-0007-engine-v0-2-graph-datapoints-and-gates.md
index 3733469..f53ab4b 100644
--- a/workplans/RBT-WP-0007-engine-v0-2-graph-datapoints-and-gates.md
+++ b/workplans/RBT-WP-0007-engine-v0-2-graph-datapoints-and-gates.md
@@ -23,7 +23,7 @@ a dimension view and an overview from the same run data.
 
 ```task
 id: RBT-WP-0007-T01
-status: todo
+status: done
 priority: high
 state_hub_task_id: "55dd86f9-5f89-59a2-8a0a-85fbedd95483"
 ```
@@ -35,7 +35,7 @@ canonical order; explain what a blocked item waits for.
 
 ```task
 id: RBT-WP-0007-T02
-status: todo
+status: done
 priority: high
 state_hub_task_id: "571e8685-6eeb-54ce-bb1a-4616ea951f81"
 ```
@@ -47,7 +47,7 @@ how `ok` is reached (check or recorded manual confirmation).
 
 ```task
 id: RBT-WP-0007-T03
-status: todo
+status: done
 priority: high
 state_hub_task_id: "707cca55-24a1-5355-8722-10740c5a2862"
 ```
@@ -59,7 +59,7 @@ exit, record waivers with who, when and why.
 
 ```task
 id: RBT-WP-0007-T04
-status: todo
+status: done
 priority: medium
 state_hub_task_id: "73307bb9-2739-563b-92c7-a034a360b444"
 ```
@@ -70,7 +70,7 @@ state_hub_task_id: "73307bb9-2739-563b-92c7-a034a360b444"
 
 ```task
 id: RBT-WP-0007-T05
-status: todo
+status: done
 priority: medium
 state_hub_task_id: "0933ff4a-8478-50d9-aa99-991948bc2379"
 ```
@@ -82,7 +82,7 @@ engine so no renderer can skip it.
 
 ```task
 id: RBT-WP-0007-T06
-status: todo
+status: done
 priority: high
 state_hub_task_id: "1656597c-3748-5480-b136-e2872436bab4"
 ```
@@ -96,7 +96,7 @@ are never auto-confirmed without a recorded pre-authorization.
 
 ```task
 id: RBT-WP-0007-T07
-status: todo
+status: done
 priority: high
 state_hub_task_id: "0ef958ef-88b6-565b-8797-df730cf11ad3"
 ```
@@ -107,6 +107,15 @@ deviation. Waivers only where a gate declares `waivable`. Autopilot timeouts
 (`stop` or `push-through`) record a deviation and, for push-through, move on. A run with
 any deviation is flagged and can never exercise a pack.
 
+## Delivered 2026-09-29
+
+`rtut/flow.py` and `rtut/run2.py`, documented in `docs/run-spec.md` (run record v0.2) and covered by
+`tests/test_run2.py` (25 tests: readiness and recommendation, parallel items, alternatives, datapoint
+states, gates, overrides, waivers, confirmation phrase, autopilot confirm / stop / push-through,
+manual-only and pre-authorization, read-only finished runs). Two spec corrections came out of it:
+a gate cannot require a datapoint it is itself `confirmed_by` (validator rule), and `run.finalize`
+also requires gates passed or waived. Not done here: any UI or CLI for v0.2 runs (RBT-WP-0004).
+
 ## Acceptance Criteria
 
 - The old console's stages, gates and datapoints can be expressed and computed