NK-WP-0056-T06: executor waits for every endpoint; errors named by phase, never mislabelled as no-effect

Two live runs of BEA-2026-0002 ended setup:URLError while four read-only diagnostic pods (same
image, labels, mounts and config) reached every endpoint. Likely cause: a fresh Job pod reaches
OpenBao before the sso/approval-engine ingress rules admit it (NetworkPolicy is programmed per
destination). Now:
- build waits until OpenBao, KeyCape, approval-engine, audit-core and LLDAP all answer HTTP
  (one shared 30 s budget); an unreachable one is named: setup:<name> not reachable.
- a preflight error is a refusal named preflight_error:<Type> (preflight has no effect).
- an error in the consume read-back is outcome unknown, consume_unconfirmed:<Type>, alerted;
  before, any exception after setup was reported as 'refused, no effect' even after consume.
- any other escaping error after setup is outcome unknown (exit 5), never 'no effect'.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 523928@bnt-lap001
Assistant-Session: 6c737afb-9c14-4152-9f02-477214b9c350

bounded-executor/cli.py

@@ -54,6 +54,7 @@ import pathlib
 import signal
 import sys
 import threading
+import urllib.parse
 
 HERE = pathlib.Path(__file__).resolve().parent
 
@@ -106,6 +107,19 @@ def distinct_roles(cfg: dict) -> None:
         raise ValueError("every lane must use its own OpenBao role")
 
 
+def run_endpoints(cfg: dict) -> dict:
+    """Every service the run talks to, with a path that answers HTTP without credentials (any status counts)."""
+    ends = {"openbao": (cfg["openbao"]["addr"], "/v1/sys/health"),
+            "lldap": (cfg["lldap"]["url"], "/"),
+            "audit-core": (cfg["audit_core"]["url"], "/healthz"),
+            "approval-engine": (cfg["approval_engine"]["url"], "/healthz")}
+    client = cfg["approval_engine"].get("client")
+    if client:  # the KeyCape token endpoint answers a GET with an HTTP error, which proves the path
+        token = urllib.parse.urlsplit(client["token_url"])
+        ends["keycape"] = (f"{token.scheme}://{token.netloc}", "/healthz")
+    return ends
+
+
 def build(cfg: dict, logins: list):
     """`logins` collects every OpenBao client as soon as it exists, so the caller revokes them all, even when a
     later step of setup fails."""
@@ -114,7 +128,7 @@ def build(cfg: dict, logins: list):
         raise ValueError("the journey read and the bind write must use separate roles")
     distinct_roles(cfg)
 
-    OPS.wait_reachable(o["addr"], ca_file=o.get("ca_file"))  # new pod: NetworkPolicy may not be applied yet
+    OPS.wait_all_reachable(run_endpoints(cfg))  # new pod: NetworkPolicy is programmed per destination
 
     def login(role: str):
         client = OPS.OpenBao(o["addr"], k8s_role=role, jwt_file=o["jwt_file"], ca_file=o.get("ca_file"))
@@ -336,9 +350,16 @@ def main(argv=None) -> int:
     logins: list = []
     try:
         config, deps = build(cfg, logins)
-        result = X.Run(admission, args.approval_id, config, deps).execute()
     except Exception as error:  # setup failed before preflight: nothing was consumed
-        result = {"outcome": "refused", "reason": "setup:" + type(error).__name__, "run_id": None, "effects": False}
+        where = str(error) if isinstance(error, OPS.Unreachable) else type(error).__name__
+        result = {"outcome": "refused", "reason": "setup:" + where, "run_id": None, "effects": False}
+    else:
+        run = X.Run(admission, args.approval_id, config, deps)
+        try:
+            result = run.execute()  # labels its own preflight and consume errors
+        except Exception as error:  # anything else after setup: the run's effect is not established
+            result = {"outcome": "unknown", "reason": "run:" + type(error).__name__, "run_id": run.run_id,
+                      "effects": "unknown"}
     finally:
         for client in logins:
             client.revoke_self()

bounded-executor/executor.py

@@ -354,6 +354,9 @@ class Run:
             self.preflight()
         except Refused as refusal:
             return {"outcome": "refused", "reason": str(refusal), "run_id": None, "effects": False}
+        except Exception as error:  # preflight has no effect: an error there is a refusal, named by its type
+            return {"outcome": "refused", "reason": "preflight_error:" + type(error).__name__, "run_id": None,
+                    "effects": False}
         if self.d.should_stop():  # the founder's stop before the one irreversible act: nothing is spent
             return {"outcome": "refused", "reason": "stopped_before_consume", "run_id": None, "effects": False}
         self.start_wall, self.start_mono = self.d.wall(), self.d.monotonic()
@@ -362,6 +365,10 @@ class Run:
         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"}
+        except Exception as error:  # the read-back failed: whether the approval is spent is not established
+            self.d.alert("consume_unconfirmed", {"admission": self.spec["id"], "run_id": self.run_id})
+            return {"outcome": "unknown", "reason": "consume_unconfirmed:" + type(error).__name__,
+                    "run_id": self.run_id, "effects": "consume-unknown"}
         try:
             self.d.audit.emit(self._event(STARTED, {
                 "receipt_version": RECEIPT_VERSION, "run_id": self.run_id, "admission": self.spec["id"],

bounded-executor/operations.py

@@ -62,7 +62,8 @@ def _opener(ca_file: str | None):
 
 
 def wait_reachable(addr: str, *, timeout: float = 30, interval: float = 1, ca_file: str | None = None,
-                   opener=None, clock=time.monotonic, sleep=time.sleep) -> float:
+                   opener=None, clock=time.monotonic, sleep=time.sleep, path: str = "/v1/sys/health",
+                   name: str = "openbao") -> float:
     """Wait until OpenBao answers on the network, before the first login. k3s applies NetworkPolicy to a new pod a
     moment after it starts, and connections in that window are refused (seen live 2026-10-31: refused at 0 s,
     open at 3 s). Polls the unauthenticated sys/health. Any HTTP answer, including 429/473/503, counts as reachable;
@@ -70,7 +71,7 @@ def wait_reachable(addr: str, *, timeout: float = 30, interval: float = 1, ca_fi
     open_ = opener or _opener(ca_file).open
     start = clock()
     while True:
-        request = urllib.request.Request(addr.rstrip("/") + "/v1/sys/health", method="GET")
+        request = urllib.request.Request(addr.rstrip("/") + path, method="GET")
         try:
             with open_(request, timeout=5):
                 return clock() - start
@@ -79,10 +80,23 @@ def wait_reachable(addr: str, *, timeout: float = 30, interval: float = 1, ca_fi
             return clock() - start
         except (urllib.error.URLError, OSError):
             if clock() - start >= timeout:
-                raise Unreachable("openbao not reachable") from None
+                raise Unreachable(name + " not reachable") from None
             sleep(interval)
 
 
+def wait_all_reachable(endpoints: dict, *, timeout: float = 30, **kwargs) -> dict:
+    """Wait until every named endpoint answers HTTP at all (any status counts), within one shared timeout. A new
+    pod's NetworkPolicy is programmed per destination: OpenBao can already be reachable while the sso and
+    approval-engine ingress rules do not yet admit the pod (NK-WP-0056-T06, two live runs refused setup:URLError).
+    `endpoints` maps name -> (base url, path). Returns seconds waited per endpoint; raises Unreachable(name)."""
+    clock = kwargs.get("clock", time.monotonic)
+    start, waited = clock(), {}
+    for name, (base, path) in endpoints.items():
+        left = max(timeout - (clock() - start), 0.5)
+        waited[name] = round(wait_reachable(base, timeout=left, path=path, name=name, **kwargs), 2)
+    return waited
+
+
 class Unreachable(Exception):
     pass
 

bounded-executor/tests/test_executor.py

@@ -402,6 +402,30 @@ class FailureTests(Base):
         self.assertIn("restore-kv", self.ops())
         self.assertNotIn("deliver", self.ops())
 
+    def test_preflight_error_is_a_named_refusal(self):
+        def broken(_id):
+            raise ConnectionRefusedError()
+        self.w.carrier.get = broken
+        result = self.run_it()
+        self.assertEqual((result["outcome"], result["reason"], result["effects"]),
+                         ("refused", "preflight_error:ConnectionRefusedError", False))
+        self.assertEqual(self.w.carrier.consumes, 0)
+
+    def test_consume_read_back_error_is_unknown_never_no_effect(self):
+        real_get, calls = self.w.carrier.get, []
+        def get(approval_id):
+            calls.append(approval_id)
+            if len(calls) > 1:  # preflight reads once; the consume read-back fails
+                raise ConnectionResetError()
+            return real_get(approval_id)
+        self.w.carrier.get = get
+        self.w.carrier.raise_on_consume = True
+        result = self.run_it()
+        self.assertEqual((result["outcome"], result["reason"], result["effects"]),
+                         ("unknown", "consume_unconfirmed:ConnectionResetError", "consume-unknown"))
+        self.assertIn("consume_unconfirmed", self.w.alerts)
+        self.assertEqual(self.w.events, [])
+
     def test_stop_before_consume_spends_nothing(self):
         self.w.stop_after = 0
         result = self.run_it()

bounded-executor/tests/test_openbao_lanes.py

@@ -92,14 +92,19 @@ class CliLanes(Base):
         return code, json.loads(out.getvalue())
 
     def cfg(self, journey_role="journey-read"):
-        # lldap is missing on purpose: setup fails after both logins
+        # every endpoint answers (the fake answers any GET); setup fails after both logins, at the Kubernetes client
+        # (no in-cluster service-account CA outside a pod)
         return {"openbao": {"addr": self.addr, "k8s_role": "bind-write", "jwt_file": str(self.root / "jwt")},
-                "journey": {"kv": "x", "k8s_role": journey_role, "config": {}}}
+                "journey": {"kv": "x", "k8s_role": journey_role, "config": {}},
+                "lldap": {"url": self.addr}, "audit_core": {"url": self.addr, "token_file": str(self.root / "jwt")},
+                "approval_engine": {"url": self.addr, "token_file": str(self.root / "jwt")}}
 
     def test_two_lanes_two_logins_both_revoked_when_setup_fails(self):
         code, result = self.run_cli(self.cfg())
-        self.assertEqual((code, result["reason"]), (4, "setup:KeyError"))
+        self.assertEqual((code, result["reason"]), (4, "setup:FileNotFoundError"))
         self.assertEqual([role for role, _ in self.bao.logins], ["bind-write", "journey-read"])
+        for path in ("/v1/sys/health", "/", "/healthz"):  # every endpoint was waited for before any login
+            self.assertIn(path, [p for p, _ in self.bao.reads])
         self.assertEqual(sorted(self.bao.revoked), ["tok-bind-write-1", "tok-journey-read-2"])
 
     def test_one_role_for_both_lanes_is_refused_before_any_login(self):
@@ -244,3 +249,39 @@ class JourneyUsesItsOwnLane(unittest.TestCase):
 
 if __name__ == "__main__":
     unittest.main()
+
+
+class EveryEndpointReachable(Base):
+    def test_unreachable_endpoint_is_named_and_nothing_logs_in(self):
+        closed = socket_port()
+        cfg = {"openbao": {"addr": self.addr, "k8s_role": "bind-write", "jwt_file": str(self.root / "jwt")},
+               "journey": {"kv": "x", "k8s_role": "journey-read", "config": {}},
+               "lldap": {"url": self.addr}, "audit_core": {"url": self.addr, "token_file": "x"},
+               "approval_engine": {"url": self.addr, "client": {"token_url": f"http://127.0.0.1:{closed}/token",
+                                                                "client_id": "c", "k8s_role": "a", "kv": "k"}}}
+        original = OPS.wait_reachable
+        self.addCleanup(setattr, OPS, "wait_reachable", original)
+        OPS.wait_reachable = lambda *a, **k: original(*a, **{**k, "timeout": 0.2, "interval": 0.05})
+        (self.root / "cfg.json").write_text(json.dumps(cfg))
+        admission = CLI.X.REPO / "examples/bounded-execution-admission/bind-rotation-svc-authelia.yaml"
+        with contextlib.redirect_stdout(io.StringIO()) as out:
+            code = CLI.main(["--admission", str(admission), "--approval-id", "apr-1", "--config", str(self.root / "cfg.json")])
+        result = json.loads(out.getvalue())
+        self.assertEqual((code, result["reason"], result["effects"]), (4, "setup:keycape not reachable", False))
+        self.assertEqual(self.bao.logins, [])
+
+    def test_endpoints_cover_every_service_the_run_uses(self):
+        cfg = {"openbao": {"addr": "http://bao:8200"}, "lldap": {"url": "http://lldap:17170"},
+               "audit_core": {"url": "http://audit:8080"},
+               "approval_engine": {"url": "http://ae:8080", "client": {"token_url": "http://kc:8080/token"}}}
+        self.assertEqual(CLI.run_endpoints(cfg), {
+            "openbao": ("http://bao:8200", "/v1/sys/health"), "lldap": ("http://lldap:17170", "/"),
+            "audit-core": ("http://audit:8080", "/healthz"), "approval-engine": ("http://ae:8080", "/healthz"),
+            "keycape": ("http://kc:8080", "/healthz")})
+
+
+def socket_port():
+    import socket
+    with socket.socket() as sock:
+        sock.bind(("127.0.0.1", 0))
+        return sock.getsockname()[1]