diff --git a/Makefile b/Makefile
index df24e7d..87f4ed4 100644
--- a/Makefile
+++ b/Makefile
@@ -210,7 +210,7 @@ bounded-execution-admission-check: ## Validate the example admissions (read-only
 	python3 tools/bounded-execution-admission/validate.py validate examples/bounded-execution-admission/*.yaml
 
 bounded-executor-test: ## Bounded executor RBAC and declaration tests; the pilot admission inside its declared grants (NK-WP-0056-T03)
-	python3 -m pytest sso-mfa/k8s/bounded-executor
+	python3 -m pytest sso-mfa/k8s/bounded-executor bounded-executor/tests
 	python3 tools/service-identity/validate.py tools/service-identity/declarations/bounded-executor.yaml \
 	  --admissions examples/bounded-execution-admission/bind-rotation-svc-authelia.yaml
 
diff --git a/WORK-RECORDS.md b/WORK-RECORDS.md
index fbaa2c2..904c467 100644
--- a/WORK-RECORDS.md
+++ b/WORK-RECORDS.md
@@ -353,7 +353,7 @@
 | task | NK-WP-0056-T08 | done | — | workplans/NK-WP-0056-bounded-executor-path.md |
 | task | NK-WP-0056-T09 | done | — | workplans/NK-WP-0056-bounded-executor-path.md |
 | task | NK-WP-0056-T10 | todo | — | workplans/NK-WP-0056-bounded-executor-path.md |
-| task | NK-WP-0056-T11 | todo | — | workplans/NK-WP-0056-bounded-executor-path.md |
+| task | NK-WP-0056-T11 | done | — | workplans/NK-WP-0056-bounded-executor-path.md |
 | task | NK-WP-0056-T12 | todo | — | workplans/NK-WP-0056-bounded-executor-path.md |
 | intake | NK-IN-0001 | closed | blue | docs/intakes/activity-core-ops-sso-operators.md |
 | intake | NK-IN-0002 | closed | blue | docs/intakes/activity-core-ops-sso-operators.md |
diff --git a/bounded-executor/README.md b/bounded-executor/README.md
new file mode 100644
index 0000000..572be73
--- /dev/null
+++ b/bounded-executor/README.md
@@ -0,0 +1,37 @@
+# Bounded executor (NK-WP-0056)
+
+Runs one founder-approved Bounded Execution Admission (`canon/standards/bounded-execution-admission_v0.1.md`)
+exactly once. Founder decision 6d1adce9: net-kingdom writes the executor; `rapp-bounded-executor` deploys it as one
+Job per admission in `sso` (T12).
+
+`executor.py` is the core runner (T11):
+
+- **preflight, no effect:**
+  - the admission is valid (BEA-1..14);
+  - the executed files match their pins;
+  - the trust states hold;
+  - every operation and check is known;
+  - the window is open and longer than the wall-time budget;
+  - the approval-engine approval binds this admission's digest, names this executor as actor and was approved
+    only by listed HUMAN approvers at aal2. The name-to-subject map is configured; no guessing.
+- **consume once:** `request_digest` = admission digest, `decision_id` = run id. An unclear outcome is resolved by
+  reading the approval back, never by a second consume.
+- **custody:** emit `execution.run.started` to audit-core. If the event can't be stored, the run stops.
+- **steps:** in plan order, one consumer at a time. Checks run after every mutating step, after their settle time.
+  The kill switch and the budget are checked between steps.
+- **failure:**
+  - A failed check, an exhausted budget or the kill switch reverts in reverse order.
+  - Each revert must pass its own checks.
+  - A clean revert may retry within `attempts`.
+  - An outcome that can't be determined stops without guessing.
+  - A failed revert stops and alerts.
+- **receipt:** one `execution.attribution.receipt` per run, with PDP `not-evaluated` in the pilot. It never holds a
+  value that an operation handled.
+
+Operations, checks, the approval carrier, the trust oracle, the audit sink, alerts and clocks are injected. T12 adds
+the pilot's real operations (OpenBao CAS write, LLDAP self-rotation, ExternalSecret sync, rollout, synthetic
+journey), the CLI, the Containerfile and the image workflow.
+
+```bash
+python3 -m pytest bounded-executor/tests     # or: make bounded-executor-test
+```
diff --git a/bounded-executor/executor.py b/bounded-executor/executor.py
new file mode 100644
index 0000000..9993a3e
--- /dev/null
+++ b/bounded-executor/executor.py
@@ -0,0 +1,348 @@
+#!/usr/bin/env python3
+"""Bounded executor: core runner (NK-WP-0056-T11).
+
+Runs one Bounded Execution Admission (canon/standards/bounded-execution-admission_v0.1.md) exactly once:
+
+  preflight (no effect)  the admission is valid (BEA-1..14); the executed source matches its pins; the
+                         trust states hold; every operation and check is known; the approval carried by
+                         approval-engine binds this admission's digest, names this executor as actor, was
+                         approved by listed HUMAN approvers at aal2, is unconsumed, and the window is open
+  consume                once, with request_digest = admission digest and decision_id = run id; an
+                         unclear outcome is resolved by reading the approval back, never by consuming again
+  started                `execution.run.started` to the audit sink; no custody, no run
+  steps                  in plan order, one consumer at a time; checks after each mutating step, after
+                         their settle time; the kill switch and the budget are checked between steps
+  failure                a failed check or an exhausted budget reverts the mutated steps in reverse
+                         order (or stops, as the admission says) and re-runs their checks; an error of
+                         unknown outcome stops without guessing; a failed revert stops and alerts
+  receipt                one `execution.attribution.receipt` per run (PDP `not-evaluated` in the pilot)
+
+Every effect goes through an injected interface: Carrier, operation and check callables, TrustOracle,
+AuditSink, Alert, clocks. The runner never prints, logs or stores a value an operation handles.
+"""
+
+from __future__ import annotations
+
+import dataclasses
+import datetime as dt
+import hashlib
+import importlib.util
+import json
+import pathlib
+import time
+import uuid
+from typing import Callable, Protocol
+
+HERE = pathlib.Path(__file__).resolve().parent
+REPO = HERE.parent
+_spec = importlib.util.spec_from_file_location("bea_validate", REPO / "tools/bounded-execution-admission/validate.py")
+bea = importlib.util.module_from_spec(_spec)
+_spec.loader.exec_module(bea)
+
+ACTION = "bounded-execution"
+STARTED = "execution.run.started"
+RECEIPT = "execution.attribution.receipt"
+
+
+class Refused(Exception):
+    """Refused before any effect."""
+
+
+class Unknown(Exception):
+    """An operation's outcome is unknown: stop, do not guess, do not revert blindly."""
+
+
+class Carrier(Protocol):
+    def get(self, approval_id: str) -> dict: ...
+    def consume(self, approval_id: str, request_digest: str, decision_id: str) -> dict: ...
+
+
+class AuditSink(Protocol):
+    def emit(self, event: dict) -> None: ...
+
+
+Operation = Callable[[dict, "Context"], None]   # (step, context) -> None, raises Unknown on unclear outcome
+Check = Callable[[dict, "Context"], bool]       # (check, context) -> passed
+
+
+@dataclasses.dataclass
+class Config:
+    executor_identity: str               # must equal spec.executor.identity and the approval's actor
+    runtime_principal: str               # e.g. system:serviceaccount:sso:nk-bind-executor
+    tenant: str
+    environment: str
+    approver_subjects: dict[str, str]    # admission approver name -> carrier subject_id (no guessing)
+    source_root: pathlib.Path = REPO
+    clock_source: str = "system-utc"
+
+
+@dataclasses.dataclass
+class Context:
+    """What operations and checks may see. Values they fetch stay inside them."""
+    run_id: str
+    admission_id: str
+    started_monotonic: float
+    data: dict = dataclasses.field(default_factory=dict)  # operation-private state (e.g. versions), never emitted
+
+
+@dataclasses.dataclass
+class Deps:
+    carrier: Carrier
+    operations: dict[str, Operation]
+    checks: dict[str, Check]
+    trust: Callable[[str], bool]
+    audit: AuditSink
+    alert: Callable[[str, dict], None]
+    should_stop: Callable[[], bool] = lambda: False
+    wall: Callable[[], dt.datetime] = lambda: dt.datetime.now(dt.UTC)
+    monotonic: Callable[[], float] = time.monotonic
+    sleep: Callable[[float], None] = time.sleep
+
+
+def iso(t: dt.datetime) -> str:
+    return t.astimezone(dt.UTC).isoformat(timespec="seconds").replace("+00:00", "Z")
+
+
+class Run:
+    def __init__(self, admission: dict, approval_id: str, config: Config, deps: Deps, schema: dict | None = None):
+        self.doc = admission
+        self.spec = admission.get("spec", {})
+        self.approval_id = approval_id
+        self.cfg = config
+        self.d = deps
+        self.schema = schema or json.loads(bea.SCHEMA.read_text())
+        self.run_id = str(uuid.uuid4())
+        self.steps: list[dict] = []
+        self.approver_subject: str | None = None
+        self.consumed_at: str | None = None
+        self.start_wall: dt.datetime | None = None
+        self.start_mono: float | None = None
+
+    # ------------------------------------------------------------------ preflight (no effect)
+    def preflight(self) -> None:
+        report = bea.evaluate(self.doc, self.schema)
+        if not report["valid"]:
+            raise Refused("admission_invalid:" + ",".join(sorted({f["rule"] for f in report["findings"]})))
+        spec = self.spec
+        if spec["executor"]["identity"] != self.cfg.executor_identity:
+            raise Refused("not_this_executor")
+        for pinned in spec["plan"]["source"]["files"]:
+            path = self.cfg.source_root / pinned["path"]
+            if not path.is_file() or hashlib.sha256(path.read_bytes()).hexdigest() != pinned["sha256"]:
+                raise Refused("source_drift")
+        for state in spec["trust_requires"]:
+            if not self.d.trust(state):
+                raise Refused("trust_not_satisfied:" + state)
+        for step in spec["plan"]["steps"] + spec["plan"].get("revert_steps", []):
+            if step["operation"] not in self.d.operations:
+                raise Refused("unknown_operation")
+        for check in spec["checks"]:
+            if check["id"] not in self.d.checks:
+                raise Refused("unknown_check")
+        self._check_window()
+        self._check_approval(self.d.carrier.get(self.approval_id), consumed_ok=False)
+
+    def _check_window(self) -> None:
+        now = self.d.wall()
+        start, end = (bea.parse_time(self.spec["window"][k]) for k in ("not_before", "not_after"))
+        if not start <= now < end:
+            raise Refused("outside_window")
+        if now + dt.timedelta(seconds=self.spec["budget"]["wall_time_seconds"]) > end:
+            raise Refused("window_shorter_than_budget")  # a run must not spill past its admission
+
+    def _check_approval(self, approval: dict, consumed_ok: bool) -> None:
+        spec = self.spec
+        binding = approval.get("binding") or {}
+        target = binding.get("target") or {}
+        if (binding.get("action") != ACTION or binding.get("actor") != self.cfg.executor_identity
+                or target.get("admission_id") != spec["id"]
+                or target.get("admission_digest") != bea.admission_digest(self.doc)
+                or binding.get("human_control") is not True):
+            raise Refused("approval_binding")
+        status = approval.get("status")
+        if status != "approved" and not (consumed_ok and status == "consumed"):
+            raise Refused("approval_status")
+        if not consumed_ok and approval.get("consumption"):
+            raise Refused("approval_already_consumed")
+        allowed = {self.cfg.approver_subjects.get(name) for name in spec["approval"]["approvers"]} - {None}
+        entries = approval.get("entries") or []
+        if not entries:
+            raise Refused("approval_no_entries")
+        for entry in entries:  # aal2 is recorded, not enforced, by approval-engine today: check every entry
+            if (entry.get("subject_id") not in allowed or str(entry.get("principal_type", "")).lower() != "human"
+                    or entry.get("assurance") != "aal2" or entry.get("subject_id") == self.cfg.executor_identity):
+                raise Refused("approval_entry")
+        self.approver_subject = entries[0]["subject_id"]
+        window_end = bea.parse_time(spec["window"]["not_after"])
+        expires = (approval.get("validity") or {}).get("expires_at")
+        if expires and bea.parse_time(expires) < window_end:
+            raise Refused("approval_expires_before_window_end")
+
+    # ------------------------------------------------------------------ consume (the one irreversible act)
+    def consume(self) -> None:
+        self._check_window()  # again, right before the one irreversible act
+        digest = bea.admission_digest(self.doc)
+        try:
+            response = self.d.carrier.consume(self.approval_id, digest, self.run_id)
+        except Exception:
+            response = None  # unclear outcome: read back, never consume twice
+        if response is not None:
+            if (response.get("status") != "consumed" or response.get("request_digest") != digest
+                    or response.get("decision_id") != self.run_id or response.get("idempotent") is not False):
+                raise Refused("consume_response")
+            self.consumed_at = response.get("consumed_at")
+        approval = self.d.carrier.get(self.approval_id)
+        self._check_approval(approval, consumed_ok=True)
+        consumption = approval.get("consumption") or {}
+        if approval.get("status") != "consumed" or consumption.get("request_digest") != digest \
+                or consumption.get("decision_id") != self.run_id:
+            raise Refused("consume_not_confirmed")
+        self.consumed_at = consumption.get("consumed_at", self.consumed_at)
+
+    # ------------------------------------------------------------------ execution
+    def _budget_left(self) -> float:
+        return self.spec["budget"]["wall_time_seconds"] - (self.d.monotonic() - self.start_mono)
+
+    def _record(self, step: dict, kind: str, outcome: str, checks: list[dict] | None = None) -> None:
+        self.steps.append({"id": step["id"], "kind": kind, "operation": step["operation"], "target": step["target"],
+                           "consumer": step["consumer"], "outcome": outcome, "checks": checks or []})
+
+    def _run_checks(self, ids: list[str], ctx: Context) -> tuple[bool, list[dict]]:
+        results = []
+        for cid in ids:
+            check = next(c for c in self.spec["checks"] if c["id"] == cid)
+            settle = check.get("settle_seconds", 0)
+            if settle:
+                if self._budget_left() < settle + check["timeout_seconds"]:
+                    results.append({"id": cid, "result": "budget"})
+                    return False, results
+                self.d.sleep(settle)
+            began = self.d.monotonic()
+            try:
+                passed = bool(self.d.checks[cid](check, ctx))
+            except Exception:
+                passed = False
+            if self.d.monotonic() - began > check["timeout_seconds"]:
+                passed = False
+            results.append({"id": cid, "result": "pass" if passed else "fail"})
+            if not passed:
+                return False, results
+        return True, results
+
+    def _revert(self, done: list[dict], ctx: Context) -> bool:
+        """Undo the mutated steps in reverse order; each revert step must pass its own checks."""
+        reverts = {s["id"]: s for s in self.spec["plan"].get("revert_steps", [])}
+        for step in reversed(done):
+            if not step["mutates"]:
+                continue
+            if step.get("irreversible"):
+                return False
+            revert = reverts[step["revert"]]
+            try:
+                self.d.operations[revert["operation"]](revert, ctx)
+            except Exception:
+                self._record(revert, "revert", "error")
+                return False
+            ok, results = self._run_checks(revert.get("checks_after", []), ctx)
+            self._record(revert, "revert", "done" if ok else "check_failed", results)
+            if not ok:
+                return False
+        return True
+
+    def _attempt(self, ctx: Context) -> str:
+        """One pass over the plan. Returns completed | reverted | revert_failed | stopped | unknown | budget."""
+        policy = self.spec["stop_and_revert"]
+        done: list[dict] = []
+        for step in self.spec["plan"]["steps"]:
+            if self.d.should_stop() or self._budget_left() <= 0:
+                reason = "stopped" if self.d.should_stop() else "budget"
+                action = policy["on_budget_exhausted"] if reason == "budget" else "revert"
+                if action == "revert" and done:
+                    return reason if self._revert(done, ctx) else "revert_failed"
+                return reason
+            try:
+                self.d.operations[step["operation"]](step, ctx)
+            except Unknown:
+                self._record(step, "step", "unknown")
+                return "unknown"
+            except Exception:
+                self._record(step, "step", "unknown")
+                return "unknown"
+            done.append(step)
+            ok, results = self._run_checks(step.get("checks_after", []), ctx) if step["mutates"] else (True, [])
+            self._record(step, "step", "done" if ok else "check_failed", results)
+            if not ok:
+                if policy["on_check_failure"] == "revert":
+                    return "reverted" if self._revert(done, ctx) else "revert_failed"
+                return "stopped"
+        return "completed"
+
+    # ------------------------------------------------------------------ evidence
+    def _event(self, kind: str, body: dict) -> dict:
+        return {"id": f"run:{self.run_id}:{'started' if kind == STARTED else 'receipt'}", "type": kind,
+                "correlation_id": self.run_id, "subject": f"admission:{self.spec['id']}",
+                "source": "net-kingdom.bounded-executor", "tenant": self.cfg.tenant, "data": body}
+
+    def _receipt(self, outcome: str, attempts: int) -> dict:
+        end = self.d.wall()
+        return self._event(RECEIPT, {
+            "run_id": self.run_id,
+            "initiator": {"sub": self.approver_subject, "role": "approver"},
+            "executing_workload": self.cfg.executor_identity,
+            "runtime_principal": self.cfg.runtime_principal,
+            "tenant": self.cfg.tenant, "environment": self.cfg.environment,
+            "source_commit": self.spec["plan"]["source"]["commit"],
+            "artifact_digest": self.spec["plan_digest"],
+            "target": f"admission:{self.spec['id']}", "action": ACTION,
+            "decision_refs": [
+                {"label": "admission", "ref": self.spec["id"]},
+                {"label": "admission_digest", "ref": bea.admission_digest(self.doc)},
+                {"label": "approval", "ref": self.approval_id},
+            ],
+            "pdp": "not-evaluated",
+            "time_interval": {"start": iso(self.start_wall), "end": iso(end), "clock_source": self.cfg.clock_source},
+            "outcome": outcome,
+            "reverted": any(s["kind"] == "revert" for s in self.steps),
+            "attempts": attempts,
+            "budget": {"limit": {"wall_time_seconds": self.spec["budget"]["wall_time_seconds"],
+                                 "attempts": self.spec["budget"]["attempts"]},
+                       "used": {"wall_time_seconds": round(self.d.monotonic() - self.start_mono, 3),
+                                "attempts": attempts}},
+            "steps": self.steps,
+        })
+
+    # ------------------------------------------------------------------ the run
+    def execute(self) -> dict:
+        try:
+            self.preflight()
+        except Refused as refusal:
+            return {"outcome": "refused", "reason": str(refusal), "run_id": None, "effects": False}
+        self.start_wall, self.start_mono = self.d.wall(), self.d.monotonic()
+        try:
+            self.consume()
+        except Refused as refusal:  # consumed or not, nothing else may happen now
+            self.d.alert("consume_unconfirmed", {"admission": self.spec["id"], "run_id": self.run_id})
+            return {"outcome": "refused", "reason": str(refusal), "run_id": self.run_id, "effects": "consume-only"}
+        try:
+            self.d.audit.emit(self._event(STARTED, {"run_id": self.run_id, "admission": self.spec["id"],
+                                                    "approval": self.approval_id, "consumed_at": self.consumed_at,
+                                                    "executing_workload": self.cfg.executor_identity}))
+        except Exception:
+            self.d.alert("started_not_in_custody", {"admission": self.spec["id"], "run_id": self.run_id})
+            return {"outcome": "stopped", "reason": "audit_unavailable", "run_id": self.run_id, "effects": "consume-only"}
+        ctx = Context(self.run_id, self.spec["id"], self.start_mono)
+        attempts, outcome = 0, "budget"
+        while attempts < self.spec["budget"]["attempts"]:
+            attempts += 1
+            outcome = self._attempt(ctx)
+            if outcome != "reverted" or self._budget_left() <= 0:
+                break  # only a clean revert may be retried, and only within budget
+        if outcome in ("unknown", "revert_failed"):
+            self.d.alert(outcome, {"admission": self.spec["id"], "run_id": self.run_id})
+        receipt = self._receipt(outcome, attempts)
+        try:
+            self.d.audit.emit(receipt)
+        except Exception:
+            self.d.alert("receipt_not_in_custody", {"admission": self.spec["id"], "run_id": self.run_id})
+            return {"outcome": outcome, "reason": "receipt_not_in_custody", "run_id": self.run_id, "receipt": receipt}
+        return {"outcome": outcome, "run_id": self.run_id, "receipt": receipt}
diff --git a/bounded-executor/tests/test_executor.py b/bounded-executor/tests/test_executor.py
new file mode 100644
index 0000000..570e52b
--- /dev/null
+++ b/bounded-executor/tests/test_executor.py
@@ -0,0 +1,372 @@
+"""Core runner tests (NK-WP-0056-T11). Fakes only: carrier, operations, checks, trust, audit, clocks."""
+
+from __future__ import annotations
+
+import copy
+import datetime as dt
+import hashlib
+import importlib.util
+import json
+import pathlib
+import sys
+import tempfile
+import unittest
+
+HERE = pathlib.Path(__file__).resolve().parent
+spec = importlib.util.spec_from_file_location("executor", HERE.parent / "executor.py")
+X = importlib.util.module_from_spec(spec)
+sys.modules["executor"] = X  # dataclasses resolve deferred annotations through sys.modules
+spec.loader.exec_module(X)
+bea = X.bea
+
+PILOT = bea.load(X.REPO / "examples/bounded-execution-admission/bind-rotation-svc-authelia.yaml")
+ADMIN_SUB = "kc-sub-0f3a"  # opaque KeyCape subject of the human approver `admin`
+NOW = dt.datetime(2026, 11, 3, 9, 30, tzinfo=dt.UTC)
+SECRET = "synthetic-bind-value-never-emitted"
+
+
+class Clock:
+    def __init__(self):
+        self.t = 0.0
+        self.sleeps = []
+
+    def monotonic(self):
+        return self.t
+
+    def sleep(self, s):
+        self.sleeps.append(s)
+        self.t += s
+
+
+class FakeCarrier:
+    """approval-engine's object and consume semantics (store.Approval.as_dict, Engine.consume)."""
+
+    def __init__(self, admission, **over):
+        self.approval = {
+            "id": "apr-1", "status": "approved",
+            "binding": {"action": "bounded-execution", "actor": admission["spec"]["executor"]["identity"],
+                        "principal": "admin", "purpose": admission["spec"]["purpose"],
+                        "target": {"admission_id": admission["spec"]["id"],
+                                   "admission_digest": bea.admission_digest(admission)},
+                        "human_control": True, "pdp_path": False},
+            "validity": {"not_before": admission["spec"]["window"]["not_before"],
+                         "expires_at": admission["spec"]["window"]["not_after"]},
+            "entries": [{"subject_id": ADMIN_SUB, "assurance": "aal2", "principal_type": "human",
+                         "approved_at": "2026-11-03T08:55:00Z"}],
+        }
+        for k, v in over.items():
+            self.approval[k] = v
+        self.consumes = 0
+        self.raise_on_consume = False
+        self.consume_lands = True
+
+    def get(self, approval_id):
+        return copy.deepcopy(self.approval)
+
+    def consume(self, approval_id, request_digest, decision_id):
+        self.consumes += 1
+        if self.consume_lands:
+            if self.approval["status"] == "consumed":
+                same = self.approval["consumption"]["request_digest"] == request_digest
+                if not same:
+                    raise RuntimeError("409")
+                return {"approval_id": approval_id, "status": "consumed", "request_digest": request_digest,
+                        "decision_id": self.approval["consumption"].get("decision_id"),
+                        "consumed_at": "x", "idempotent": True}
+            self.approval["status"] = "consumed"
+            self.approval["consumption"] = {"request_digest": request_digest, "decision_id": decision_id,
+                                            "consumed_at": "2026-11-03T09:30:00Z"}
+        if self.raise_on_consume:
+            raise TimeoutError("connection reset")
+        return {"approval_id": approval_id, "status": "consumed", "request_digest": request_digest,
+                "decision_id": decision_id, "consumed_at": "2026-11-03T09:30:00Z", "idempotent": False}
+
+
+class World:
+    def __init__(self, admission, clock):
+        self.calls = []
+        self.fail_check = {}      # check id -> number of times to fail
+        self.raise_op = set()     # operation names that raise
+        self.unknown_op = set()
+        self.slow_op = 0.0
+        self.events = []
+        self.alerts = []
+        self.audit_down = False
+        self.stop_after = None
+        self.clock = clock
+        self.carrier = FakeCarrier(admission)
+        self.trust_ok = True
+
+    def op(self, step, ctx):
+        if step["operation"] in self.unknown_op:
+            raise X.Unknown()
+        if step["id"] in self.raise_op:
+            raise RuntimeError("boom")
+        ctx.data["value"] = SECRET  # operations may hold values; the runner must never emit them
+        self.calls.append(step["id"])
+        self.clock.t += self.slow_op
+
+    def check(self, check, ctx):
+        self.calls.append("check:" + check["id"])
+        left = self.fail_check.get(check["id"], 0)
+        if left:
+            self.fail_check[check["id"]] = left - 1
+            return False
+        return True
+
+    def emit(self, event):
+        if self.audit_down:
+            raise ConnectionError()
+        self.events.append(event)
+
+    def should_stop(self):
+        return self.stop_after is not None and len([c for c in self.calls if not c.startswith("check:")]) >= self.stop_after
+
+    def deps(self, admission):
+        ops = {s["operation"]: self.op for s in admission["spec"]["plan"]["steps"] + admission["spec"]["plan"]["revert_steps"]}
+        checks = {c["id"]: self.check for c in admission["spec"]["checks"]}
+        audit = type("A", (), {"emit": lambda _s, e: self.emit(e)})()
+        return X.Deps(carrier=self.carrier, operations=ops, checks=checks, trust=lambda s: self.trust_ok,
+                      audit=audit, alert=lambda kind, info: self.alerts.append(kind),
+                      should_stop=self.should_stop, wall=lambda: NOW,
+                      monotonic=self.clock.monotonic, sleep=self.clock.sleep)
+
+
+def synthetic_admission(root: pathlib.Path, mutate=None) -> dict:
+    """The pilot example, with its pinned files replaced by temporary ones and the digest recomputed."""
+    doc = copy.deepcopy(PILOT)
+    files = []
+    for f in doc["spec"]["plan"]["source"]["files"]:
+        path = root / f["path"]
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text("synthetic " + f["path"])
+        files.append({"path": f["path"], "sha256": hashlib.sha256(path.read_bytes()).hexdigest()})
+    doc["spec"]["plan"]["source"]["files"] = files
+    by_path = {f["path"]: f["sha256"] for f in files}
+    for c in doc["spec"]["checks"]:
+        c["sha256"] = by_path[c["ref"]]
+    if mutate:
+        mutate(doc["spec"])
+    doc["spec"]["plan_digest"] = bea.plan_digest(doc)
+    return doc
+
+
+class Base(unittest.TestCase):
+    mutate = None
+
+    def setUp(self):
+        tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(tmp.cleanup)
+        self.root = pathlib.Path(tmp.name)
+        self.doc = synthetic_admission(self.root, self.mutate)
+        self.clock = Clock()
+        self.w = World(self.doc, self.clock)
+        self.cfg = X.Config(executor_identity="service:nk-bind-executor",
+                            runtime_principal="system:serviceaccount:sso:nk-bind-executor",
+                            tenant="tenant:platform", environment="PRODUCTION",
+                            approver_subjects={"admin": ADMIN_SUB}, source_root=self.root)
+
+    def run_it(self, doc=None):
+        doc = doc or self.doc
+        return X.Run(doc, "apr-1", self.cfg, self.w.deps(doc)).execute()
+
+    def ops(self):
+        return [c for c in self.w.calls if not c.startswith("check:")]
+
+    def assert_no_effect(self, result, reason):
+        self.assertEqual((result["outcome"], result["reason"]), ("refused", reason))
+        self.assertEqual(self.w.carrier.consumes, 0)
+        self.assertEqual(self.w.calls, [])
+        self.assertEqual(self.w.events, [])
+
+
+class HappyPathTests(Base):
+    def test_completes_with_started_and_receipt(self):
+        result = self.run_it()
+        self.assertEqual(result["outcome"], "completed", result)
+        self.assertEqual(self.w.carrier.consumes, 1)
+        self.assertEqual(self.ops(), ["generate", "set-directory", "deliver", "restart"])
+        self.assertEqual([e["type"] for e in self.w.events], [X.STARTED, X.RECEIPT])
+        started, receipt = self.w.events
+        self.assertEqual(started["id"], f"run:{result['run_id']}:started")
+        data = receipt["data"]
+        self.assertEqual(data["pdp"], "not-evaluated")
+        self.assertEqual({r["label"] for r in data["decision_refs"]}, {"admission", "admission_digest", "approval"})
+        self.assertEqual(data["initiator"]["sub"], ADMIN_SUB)
+        self.assertEqual(data["artifact_digest"], self.doc["spec"]["plan_digest"])
+        self.assertFalse(data["reverted"])
+        self.assertNotIn(SECRET, json.dumps(self.w.events))
+
+    def test_consumed_with_admission_digest_and_run_id(self):
+        result = self.run_it()
+        consumption = self.w.carrier.approval["consumption"]
+        self.assertEqual(consumption["request_digest"], bea.admission_digest(self.doc))
+        self.assertEqual(consumption["decision_id"], result["run_id"])
+
+
+class RefusalTests(Base):
+    def test_invalid_admission(self):
+        doc = copy.deepcopy(self.doc)
+        doc["spec"]["plan"]["steps"][0]["operation"] = "tampered"
+        self.w = World(doc, self.clock)
+        self.assertEqual(self.run_it(doc)["reason"], "admission_invalid:BEA-2")
+        self.assertEqual(self.w.carrier.consumes, 0)
+
+    def test_refusals_before_any_effect(self):
+        cases = {
+            "not_this_executor": lambda: setattr(self.cfg, "executor_identity", "service:other"),
+            "source_drift": lambda: (self.root / "tools/service-binds/bind-lane.py").write_text("changed"),
+            "trust_not_satisfied:runtime_secret_trust": lambda: setattr(self.w, "trust_ok", False),
+            "approval_binding": lambda: self.w.carrier.approval["binding"]["target"].update(admission_digest="sha256:" + "0" * 64),
+            "approval_status": lambda: self.w.carrier.approval.update(status="requested"),
+            "approval_entry": lambda: self.w.carrier.approval["entries"][0].update(assurance="aal1"),
+            "approval_expires_before_window_end": lambda: self.w.carrier.approval["validity"].update(expires_at="2026-11-03T10:00:00Z"),
+        }
+        for reason, mutate in cases.items():
+            with self.subTest(reason=reason):
+                self.setUp()
+                mutate()
+                self.assert_no_effect(self.run_it(), reason)
+
+    def test_approval_by_someone_not_listed_or_not_human(self):
+        for entry in ({"subject_id": "kc-sub-other", "assurance": "aal2", "principal_type": "human"},
+                      {"subject_id": ADMIN_SUB, "assurance": "aal2", "principal_type": "service"}):
+            with self.subTest(entry=entry):
+                self.setUp()
+                self.w.carrier.approval["entries"] = [entry]
+                self.assert_no_effect(self.run_it(), "approval_entry")
+
+    def test_actor_must_be_this_executor(self):
+        self.w.carrier.approval["binding"]["actor"] = "service:someone-else"
+        self.assert_no_effect(self.run_it(), "approval_binding")
+
+    def test_outside_window(self):
+        self.w.deps = (lambda orig: (lambda doc: (lambda d: (setattr(d, "wall", lambda: NOW + dt.timedelta(hours=3)), d)[1])(orig(doc))))(self.w.deps)
+        self.assert_no_effect(self.run_it(), "outside_window")
+
+    def test_window_too_short_for_the_budget(self):
+        late = dt.datetime(2026, 11, 3, 10, 50, tzinfo=dt.UTC)  # 10 min left, budget 20 min
+        orig = self.w.deps
+        self.w.deps = lambda doc: (lambda d: (setattr(d, "wall", lambda: late), d)[1])(orig(doc))
+        self.assert_no_effect(self.run_it(), "window_shorter_than_budget")
+
+    def test_already_consumed_is_never_reused(self):
+        self.w.carrier.approval["status"] = "consumed"
+        self.w.carrier.approval["consumption"] = {"request_digest": bea.admission_digest(self.doc), "decision_id": "earlier-run"}
+        self.assert_no_effect(self.run_it(), "approval_status")
+
+    def test_unknown_operation(self):
+        d = self.w.deps(self.doc)
+        d.operations.pop("eso.force-sync")
+        result = X.Run(self.doc, "apr-1", self.cfg, d).execute()
+        self.assert_no_effect(result, "unknown_operation")
+
+
+class ConsumeTests(Base):
+    def test_unclear_consume_resolved_by_reading_back(self):
+        self.w.carrier.raise_on_consume = True  # consume landed, response lost
+        result = self.run_it()
+        self.assertEqual(result["outcome"], "completed")
+        self.assertEqual(self.w.carrier.consumes, 1)  # never consumed twice
+
+    def test_unclear_consume_that_did_not_land_stops(self):
+        self.w.carrier.raise_on_consume = True
+        self.w.carrier.consume_lands = False
+        result = self.run_it()
+        self.assertEqual((result["outcome"], result["reason"]), ("refused", "consume_not_confirmed"))
+        self.assertEqual((self.w.calls, self.w.events), ([], []))
+        self.assertIn("consume_unconfirmed", self.w.alerts)
+
+    def test_idempotent_consume_is_not_a_fresh_run(self):
+        original = self.w.carrier.consume
+        self.w.carrier.consume = lambda a, d, r: dict(original(a, d, r), idempotent=True)
+        result = self.run_it()
+        self.assertEqual(result["reason"], "consume_response")
+        self.assertEqual(self.w.calls, [])
+
+    def test_no_custody_no_run(self):
+        self.w.audit_down = True
+        result = self.run_it()
+        self.assertEqual((result["outcome"], result["reason"]), ("stopped", "audit_unavailable"))
+        self.assertEqual(self.w.calls, [])
+        self.assertIn("started_not_in_custody", self.w.alerts)
+
+
+class FailureTests(Base):
+    def test_failed_check_reverts_in_reverse_and_verifies(self):
+        self.doc = synthetic_admission(self.root, lambda s: s["budget"].update(attempts=1))
+        self.w = World(self.doc, self.clock)
+        self.w.fail_check["eso-synced"] = 1
+        result = self.run_it()
+        self.assertEqual(result["outcome"], "reverted", result)
+        self.assertEqual(self.ops(), ["generate", "set-directory", "deliver", "redeliver", "restore-directory", "restore-kv"])
+        data = result["receipt"]["data"]
+        self.assertTrue(data["reverted"])
+        reverts = [x for x in data["steps"] if x["kind"] == "revert"]
+        self.assertEqual([x["checks"][0]["id"] for x in reverts], ["eso-synced", "bind-signs-in", "kv-advanced"])
+        self.assertTrue(all(x["outcome"] == "done" for x in reverts))
+        self.assertEqual(data["attempts"], 1)
+
+    def test_clean_revert_may_retry_within_attempts(self):
+        self.doc = synthetic_admission(self.root, lambda s: s["budget"].update(attempts=2))
+        self.w = World(self.doc, self.clock)
+        self.w.fail_check["synthetic-journey"] = 1
+        result = self.run_it()
+        self.assertEqual(result["outcome"], "completed")
+        self.assertEqual(result["receipt"]["data"]["attempts"], 2)
+        self.assertTrue(result["receipt"]["data"]["reverted"])
+
+    def test_revert_whose_check_fails_counts_as_failed(self):
+        self.doc = synthetic_admission(self.root, lambda s: s["budget"].update(attempts=1))
+        self.w = World(self.doc, self.clock)
+        self.w.fail_check["eso-synced"] = 2  # the forward check and the revert's own check
+        result = self.run_it()
+        self.assertEqual(result["outcome"], "revert_failed")
+        self.assertNotIn("restore-directory", self.ops())
+
+    def test_failed_revert_stops_and_alerts(self):
+        self.w.fail_check["eso-synced"] = 1
+        self.w.raise_op = {"restore-directory"}
+        result = self.run_it()
+        self.assertEqual(result["outcome"], "revert_failed")
+        self.assertIn("revert_failed", self.w.alerts)
+        self.assertNotIn("restore-kv", self.ops())  # stop at the failed revert, do not continue blindly
+
+    def test_unknown_outcome_stops_without_revert(self):
+        self.w.unknown_op = {"eso.force-sync"}
+        result = self.run_it()
+        self.assertEqual(result["outcome"], "unknown")
+        self.assertEqual(self.ops(), ["generate", "set-directory"])
+        self.assertFalse(result["receipt"]["data"]["reverted"])
+        self.assertIn("unknown", self.w.alerts)
+
+    def test_budget_exhausted_reverts(self):
+        self.w.slow_op = 700.0  # wall_time_seconds is 1200
+        result = self.run_it()
+        self.assertEqual(result["outcome"], "budget")
+        self.assertTrue(result["receipt"]["data"]["reverted"])
+
+    def test_kill_switch_reverts(self):
+        self.w.stop_after = 2
+        result = self.run_it()
+        self.assertEqual(result["outcome"], "stopped")
+        self.assertIn("restore-kv", self.ops())
+        self.assertNotIn("deliver", self.ops())
+
+    def test_settle_time_is_waited_and_counted(self):
+        self.doc = synthetic_admission(self.root, lambda s: s["checks"][3].update(settle_seconds=125))
+        self.w = World(self.doc, self.clock)
+        result = self.run_it()
+        self.assertEqual(result["outcome"], "completed")
+        self.assertIn(125, self.clock.sleeps)
+
+    def test_receipt_custody_failure_is_reported(self):
+        original = self.w.emit
+        self.w.emit = lambda e: (original(e) if e["type"] == X.STARTED else (_ for _ in ()).throw(ConnectionError()))
+        result = self.run_it()
+        self.assertEqual(result["reason"], "receipt_not_in_custody")
+        self.assertIn("receipt_not_in_custody", self.w.alerts)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/canon/standards/bounded-execution-admission_v0.1.md b/canon/standards/bounded-execution-admission_v0.1.md
index d0fc207..e35be2c 100644
--- a/canon/standards/bounded-execution-admission_v0.1.md
+++ b/canon/standards/bounded-execution-admission_v0.1.md
@@ -104,7 +104,7 @@ for up to 120 s, so a check run sooner proves nothing.
 | BEA-6 | Least privilege: every step and revert step is covered by a ceiling grant of the same mode, and every ceiling grant covers at least one step |
 | BEA-7 | No `human-intrinsic` step. A `root-class` step needs `high` risk |
 | BEA-8 | Every step's consumer is in the blast radius. `medium` and `high` risk change one consumer at a time |
-| BEA-9 | Every mutating step names checks. Every check is used and pinned: its `ref` and `sha256` appear in `plan.source.files` |
+| BEA-9 | Every mutating step names checks, revert steps included: a revert must prove the restored state with its own checks. Every check is used and pinned: its `ref` and `sha256` appear in `plan.source.files` |
 | BEA-10 | Every mutating step has a revert step, or is `irreversible` at `high` risk. Revert steps mutate and do not nest. Step ids are unique |
 | BEA-11 | A revertable plan reverts on check failure. A plan with nothing to revert stops |
 | BEA-12 | The wall time fits the window, the checks (timeouts plus settle times) fit the wall time, and attempts are at most 3 (1 at `high` risk) |
diff --git a/examples/bounded-execution-admission/bind-rotation-svc-authelia.yaml b/examples/bounded-execution-admission/bind-rotation-svc-authelia.yaml
index 85a90e2..5c162b3 100644
--- a/examples/bounded-execution-admission/bind-rotation-svc-authelia.yaml
+++ b/examples/bounded-execution-admission/bind-rotation-svc-authelia.yaml
@@ -49,7 +49,7 @@ spec:
   plan:
     source:
       repo: net-kingdom
-      commit: c9ab01e23e92
+      commit: 3b1abc405423
       files:
       - path: tools/service-binds/bind-lane.py
         sha256: 5b31b3567af07bba09530c6d3b0ed8b5c947df22bd13ac0442dc33c0d9b420c0
@@ -114,6 +114,8 @@ spec:
       requires:
         mode: OPERATE
         scope: openbao/platform/workloads/net-kingdom/directory-binds/svc-authelia
+      checks_after:
+      - kv-advanced
     - id: restore-directory
       operation: lldap.user.change-own-password-from-kv
       target: lldap:svc-authelia
@@ -123,6 +125,8 @@ spec:
       requires:
         mode: OPERATE
         scope: lldap/people/svc-authelia
+      checks_after:
+      - bind-signs-in
     - id: redeliver
       operation: eso.force-sync
       target: k8s:sso/externalsecret/authelia-secrets
@@ -132,6 +136,8 @@ spec:
       requires:
         mode: OPERATE
         scope: k8s/sso/externalsecret/authelia-secrets
+      checks_after:
+      - eso-synced
     - id: restart-again
       operation: k8s.rollout-restart
       target: k8s:sso/deployment/authelia
@@ -141,7 +147,9 @@ spec:
       requires:
         mode: OPERATE
         scope: k8s/sso/deployment/authelia
-  plan_digest: sha256:36328096076c71ade582173092511be6aee546281ef5edb3c2acaadaf7c9e1fb
+      checks_after:
+      - synthetic-journey
+  plan_digest: sha256:ce9253adf5084dc1b40c1004258d9f8480b236d8ed26fabbb2a983c853a4dc21
   checks:
   - id: kv-advanced
     description: KV svc-authelia has exactly one new version, CAS on the previous one
diff --git a/examples/bounded-execution-admission/bind-split-create-service-accounts.yaml b/examples/bounded-execution-admission/bind-split-create-service-accounts.yaml
index ecd7368..6f24c00 100644
--- a/examples/bounded-execution-admission/bind-split-create-service-accounts.yaml
+++ b/examples/bounded-execution-admission/bind-split-create-service-accounts.yaml
@@ -43,7 +43,7 @@ spec:
   plan:
     source:
       repo: net-kingdom
-      commit: c9ab01e23e92
+      commit: 3b1abc405423
       files:
       - path: sso-mfa/k8s/lldap/bind-split.py
         sha256: 4decc3bbe65120ff6c4cef5613b9a41a60168ae64ecd4c1380f61a2fcd37b6b3
@@ -84,6 +84,8 @@ spec:
       requires:
         mode: ADMINISTER
         scope: lldap
+      checks_after:
+      - accounts-read-only
     - id: destroy-kv
       operation: openbao.kv.destroy-created
       target: openbao:platform/workloads/net-kingdom/directory-binds
@@ -93,7 +95,9 @@ spec:
       requires:
         mode: OPERATE
         scope: openbao/platform/workloads/net-kingdom/directory-binds
-  plan_digest: sha256:139e7b874d8d8ba9419f141bf9e10341a9a0057bb75fa47eee32c2ee19b396ce
+      checks_after:
+      - accounts-sign-in
+  plan_digest: sha256:a4c60ad06ae93b7ebc4d12417678a7bf8e7507f6670d8861cb5b30c2cd517b68
   checks:
   - id: accounts-read-only
     description: each account exists in lldap_strict_readonly only and cannot write
diff --git a/examples/bounded-execution-admission/bind-split-read-only-verify.yaml b/examples/bounded-execution-admission/bind-split-read-only-verify.yaml
index b287e72..00cba4f 100644
--- a/examples/bounded-execution-admission/bind-split-read-only-verify.yaml
+++ b/examples/bounded-execution-admission/bind-split-read-only-verify.yaml
@@ -46,7 +46,7 @@ spec:
   plan:
     source:
       repo: net-kingdom
-      commit: c9ab01e23e92
+      commit: 3b1abc405423
       files:
       - path: sso-mfa/k8s/lldap/bind-split.py
         sha256: 4decc3bbe65120ff6c4cef5613b9a41a60168ae64ecd4c1380f61a2fcd37b6b3
@@ -73,7 +73,7 @@ spec:
         scope: k8s/sso
       checks_after:
       - split-shape
-  plan_digest: sha256:0459999b93bb8aee0dacb1b36ea95d114ac74d0ed775095f3ee5dc2ee66203cd
+  plan_digest: sha256:8c2195390ce140d2600667f2787c9fa2685a62ff918fb298d17e3b3aa3762bc5
   checks:
   - id: split-shape
     description: every consumer binds as its own svc- account; no service binds as admin
diff --git a/examples/bounded-execution-admission/bind-split-switch-super-admin.yaml b/examples/bounded-execution-admission/bind-split-switch-super-admin.yaml
index 4be7007..a7ab5ed 100644
--- a/examples/bounded-execution-admission/bind-split-switch-super-admin.yaml
+++ b/examples/bounded-execution-admission/bind-split-switch-super-admin.yaml
@@ -43,7 +43,7 @@ spec:
   plan:
     source:
       repo: net-kingdom
-      commit: c9ab01e23e92
+      commit: 3b1abc405423
       files:
       - path: sso-mfa/k8s/lldap/bind-split.py
         sha256: 4decc3bbe65120ff6c4cef5613b9a41a60168ae64ecd4c1380f61a2fcd37b6b3
@@ -84,6 +84,8 @@ spec:
       requires:
         mode: ADMINISTER
         scope: lldap
+      checks_after:
+      - ld-admin-administers
     - id: restart-lldap-again
       operation: k8s.rollout-restart
       target: k8s:sso/deployment/lldap
@@ -93,7 +95,9 @@ spec:
       requires:
         mode: OPERATE
         scope: k8s/sso/deployment/lldap
-  plan_digest: sha256:5f51601ce96a07da5fa56b5b00c2066c572e3b19a4677b1a16cee3178295cae0
+      checks_after:
+      - synthetic-journey
+  plan_digest: sha256:78e9b24a77955c0b7372d6733e8a6b4da17656eb5a9100a416c4d73eef981d6f
   checks:
   - id: ld-admin-administers
     description: ld-admin signs in and administers; admin keeps lldap_admin membership
diff --git a/tools/bounded-execution-admission/fixtures/invalid/human-intrinsic-step.yaml b/tools/bounded-execution-admission/fixtures/invalid/human-intrinsic-step.yaml
index 90efec5..0a48891 100644
--- a/tools/bounded-execution-admission/fixtures/invalid/human-intrinsic-step.yaml
+++ b/tools/bounded-execution-admission/fixtures/invalid/human-intrinsic-step.yaml
@@ -42,7 +42,7 @@ spec:
   plan:
     source:
       repo: net-kingdom
-      commit: c9ab01e23e92
+      commit: 3b1abc405423
       files:
       - path: sso-mfa/k8s/lldap/bind-split.py
         sha256: 4decc3bbe65120ff6c4cef5613b9a41a60168ae64ecd4c1380f61a2fcd37b6b3
@@ -95,6 +95,8 @@ spec:
       requires:
         mode: ADMINISTER
         scope: lldap
+      checks_after:
+      - ld-admin-administers
     - id: restart-lldap-again
       operation: k8s.rollout-restart
       target: k8s:sso/deployment/lldap
@@ -104,7 +106,9 @@ spec:
       requires:
         mode: OPERATE
         scope: k8s/sso/deployment/lldap
-  plan_digest: sha256:139f0ca7564184290c7cb1f6b20bdb9807db98e5609a79083ac6f7aea13bb50c
+      checks_after:
+      - synthetic-journey
+  plan_digest: sha256:80c14028288caf0994c1c5295e423527969c91f174f0ab2f7e08900f71283887
   checks:
   - id: ld-admin-administers
     description: ld-admin signs in and administers; admin keeps lldap_admin membership
diff --git a/tools/bounded-execution-admission/generate_examples.py b/tools/bounded-execution-admission/generate_examples.py
index a1b6f9b..6d973cb 100644
--- a/tools/bounded-execution-admission/generate_examples.py
+++ b/tools/bounded-execution-admission/generate_examples.py
@@ -61,10 +61,10 @@ rotation = doc(
      step("set-directory", "lldap.user.change-own-password-from-kv", "lldap:svc-authelia", "authelia", "delegable", True, "OPERATE", "lldap/people/svc-authelia", ["bind-signs-in"], "restore-directory"),
      step("deliver", "eso.force-sync", "k8s:sso/externalsecret/authelia-secrets", "authelia", "delegable", True, "OPERATE", "k8s/sso/externalsecret/authelia-secrets", ["eso-synced"], "redeliver"),
      step("restart", "k8s.rollout-restart", "k8s:sso/deployment/authelia", "authelia", "delegable", True, "OPERATE", "k8s/sso/deployment/authelia", ["synthetic-journey"], "restart-again")],
-    [step("restore-kv", "openbao.kv.write-previous-value-as-next-version", "openbao:" + KV[len("openbao/"):], "authelia", "delegable", True, "OPERATE", KV),
-     step("restore-directory", "lldap.user.change-own-password-from-kv", "lldap:svc-authelia", "authelia", "delegable", True, "OPERATE", "lldap/people/svc-authelia"),
-     step("redeliver", "eso.force-sync", "k8s:sso/externalsecret/authelia-secrets", "authelia", "delegable", True, "OPERATE", "k8s/sso/externalsecret/authelia-secrets"),
-     step("restart-again", "k8s.rollout-restart", "k8s:sso/deployment/authelia", "authelia", "delegable", True, "OPERATE", "k8s/sso/deployment/authelia")],
+    [step("restore-kv", "openbao.kv.write-previous-value-as-next-version", "openbao:" + KV[len("openbao/"):], "authelia", "delegable", True, "OPERATE", KV, ["kv-advanced"]),
+     step("restore-directory", "lldap.user.change-own-password-from-kv", "lldap:svc-authelia", "authelia", "delegable", True, "OPERATE", "lldap/people/svc-authelia", ["bind-signs-in"]),
+     step("redeliver", "eso.force-sync", "k8s:sso/externalsecret/authelia-secrets", "authelia", "delegable", True, "OPERATE", "k8s/sso/externalsecret/authelia-secrets", ["eso-synced"]),
+     step("restart-again", "k8s.rollout-restart", "k8s:sso/deployment/authelia", "authelia", "delegable", True, "OPERATE", "k8s/sso/deployment/authelia", ["synthetic-journey"])],
     [check("kv-advanced", "KV svc-authelia has exactly one new version, CAS on the previous one", LANE, 30),
      check("bind-signs-in", "the new KV value signs in to LLDAP as svc-authelia", LANE, 30),
      check("eso-synced", "the ExternalSecret is Ready with a refresh after the step started; the executor never reads the Secret", LANE, 120),
@@ -85,8 +85,8 @@ create = doc(
     [SPLIT, LANE],
     [step("create-accounts", "lldap.user.create-read-only-binds", "lldap:svc-authelia,svc-keycape,svc-privacyidea", "lldap", "root-class", True, "ADMINISTER", "lldap", ["accounts-read-only"], "delete-accounts"),
      step("hold-passwords", "openbao.kv.write-cas0", "openbao:platform/workloads/net-kingdom/directory-binds", "lldap", "root-class", True, "OPERATE", "openbao/platform/workloads/net-kingdom/directory-binds", ["accounts-sign-in"], "destroy-kv")],
-    [step("delete-accounts", "lldap.user.delete-created", "lldap:svc-authelia,svc-keycape,svc-privacyidea", "lldap", "root-class", True, "ADMINISTER", "lldap"),
-     step("destroy-kv", "openbao.kv.destroy-created", "openbao:platform/workloads/net-kingdom/directory-binds", "lldap", "root-class", True, "OPERATE", "openbao/platform/workloads/net-kingdom/directory-binds")],
+    [step("delete-accounts", "lldap.user.delete-created", "lldap:svc-authelia,svc-keycape,svc-privacyidea", "lldap", "root-class", True, "ADMINISTER", "lldap", ["accounts-read-only"]),
+     step("destroy-kv", "openbao.kv.destroy-created", "openbao:platform/workloads/net-kingdom/directory-binds", "lldap", "root-class", True, "OPERATE", "openbao/platform/workloads/net-kingdom/directory-binds", ["accounts-sign-in"])],
     [check("accounts-read-only", "each account exists in lldap_strict_readonly only and cannot write", SPLIT, 60),
      check("accounts-sign-in", "each KV value signs in to LLDAP as its account", LANE, 60)],
     {"on_check_failure": "revert", "on_budget_exhausted": "revert", "on_unknown": "stop",
@@ -104,8 +104,8 @@ superadmin = doc(
     [SPLIT, JOURNEY],
     [step("switch-user-dn", "lldap.config.set-ldap-user-dn", "k8s:sso/secret/lldap-secrets#LLDAP_LDAP_USER_DN", "lldap", "root-class", True, "ADMINISTER", "lldap", ["ld-admin-administers"], "switch-back"),
      step("restart-lldap", "k8s.rollout-restart", "k8s:sso/deployment/lldap", "lldap", "root-class", True, "OPERATE", "k8s/sso/deployment/lldap", ["synthetic-journey"], "restart-lldap-again")],
-    [step("switch-back", "lldap.config.set-ldap-user-dn", "k8s:sso/secret/lldap-secrets#LLDAP_LDAP_USER_DN", "lldap", "root-class", True, "ADMINISTER", "lldap"),
-     step("restart-lldap-again", "k8s.rollout-restart", "k8s:sso/deployment/lldap", "lldap", "root-class", True, "OPERATE", "k8s/sso/deployment/lldap")],
+    [step("switch-back", "lldap.config.set-ldap-user-dn", "k8s:sso/secret/lldap-secrets#LLDAP_LDAP_USER_DN", "lldap", "root-class", True, "ADMINISTER", "lldap", ["ld-admin-administers"]),
+     step("restart-lldap-again", "k8s.rollout-restart", "k8s:sso/deployment/lldap", "lldap", "root-class", True, "OPERATE", "k8s/sso/deployment/lldap", ["synthetic-journey"])],
     [check("ld-admin-administers", "ld-admin signs in and administers; admin keeps lldap_admin membership", SPLIT, 60),
      check("synthetic-journey", "all three binds still serve sign-in: syn-journey reaches aal2", JOURNEY, 90, settle=125)],
     {"on_check_failure": "revert", "on_budget_exhausted": "revert", "on_unknown": "stop",
diff --git a/tools/bounded-execution-admission/validate.py b/tools/bounded-execution-admission/validate.py
index b09008a..26f0de4 100644
--- a/tools/bounded-execution-admission/validate.py
+++ b/tools/bounded-execution-admission/validate.py
@@ -176,7 +176,7 @@ def evaluate(document: Any, schema: dict, as_of: dt.datetime | None = None) -> d
     if len(checks) != len(spec["checks"]):
         fail("BEA-9", "spec/checks:duplicate-id")
     used = set()
-    for step in steps:
+    for step in every_step:  # revert steps verify the restored state with their own checks
         if step["mutates"] and not step.get("checks_after"):
             fail("BEA-9", f"step/{step['id']}/checks_after:missing")
         for ref in step.get("checks_after", []):
diff --git a/workplans/NK-WP-0056-bounded-executor-path.md b/workplans/NK-WP-0056-bounded-executor-path.md
index c6fce0a..5a3dc3d 100644
--- a/workplans/NK-WP-0056-bounded-executor-path.md
+++ b/workplans/NK-WP-0056-bounded-executor-path.md
@@ -412,7 +412,7 @@ against a deliberately broken bind on a synthetic directory (T08).
 
 ```task
 id: NK-WP-0056-T11
-status: todo
+status: done
 priority: high
 assignee: claude-code
 depends_on: [NK-WP-0056-T01, NK-WP-0056-T03]
@@ -432,6 +432,40 @@ Tested against in-process fakes for the carrier, operations and audit sink.
 **Done when:** the runner passes its tests, including revert on a failed check, budget exhaustion, a stale or
 reused approval, and a step outside the ceiling. Every refusal must happen before any effect.
 
+**Done 2026-10-31.** `bounded-executor/executor.py` with 22 tests (`bounded-executor/tests/test_executor.py`; fakes
+for carrier, operations, checks, trust, audit, clocks), modelled on approval-engine's real object and consume
+response. Refusals before any effect cover:
+
+- an invalid admission;
+- another executor;
+- source drift;
+- unsatisfied trust;
+- an unknown operation;
+- outside the window, or a window shorter than the budget;
+- a binding digest or actor mismatch;
+- a non-human, aal1 or unlisted approver;
+- an already-consumed approval.
+
+Consume:
+
+- an unclear consume is read back and never repeated;
+- an idempotent response is refused;
+- with no audit custody, no run.
+
+Failure handling:
+
+- a failed check reverts in reverse order, each revert proving the restored state with its own checks;
+- a clean revert retries within `attempts`;
+- a failed revert, or a revert check that fails, stops and alerts;
+- an unknown outcome stops without reverting;
+- budget exhaustion and the kill switch revert;
+- settle times are waited and counted;
+- a receipt-custody failure is reported.
+
+Receipts carry no operation values.
+
+Standard and validator change: BEA-9 now applies to revert steps, which name their own checks; examples regenerated.
+
 ## Executor runtime: pilot operations and packaging
 
 ```task