NK-WP-0058-T03 step 2: executor --preflight-only
All checks were successful
Build and Publish bounded-executor / build-and-push (push) Successful in 14s
CI Smoke / host-smoke (push) Successful in 0s
CI Smoke / container-smoke (push) Successful in 1s
Identity provider journey acceptance / provider (push) Successful in 52s

Reachability per endpoint, setup without the window-scoped write role (no Kubernetes or directory
client), then every preflight check (admission, executor, source, trust, registry, window,
approval) with its phase; no consume, emit or write. Exit 0 ready, 4 not. preflight() is now the
same named checks, stopping at the first. Setup reasons keep the clients' fixed texts.

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

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 943911@bnt-lap001
Assistant-Session: fc322940-1c8e-4bed-ab39-86cec99f23a0
This commit is contained in:
tegwick 2026-11-05 23:27:13 +01:00
parent 7350ba8003
commit 64cec27c28
3 changed files with 179 additions and 32 deletions

View file

@ -155,28 +155,61 @@ class Run:
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"]})))
def _preflight_steps(self) -> list[tuple[str, Callable[[], None]]]:
"""The preflight as named checks, in order. preflight() stops at the first refusal; preflight_report()
runs them all (NK-WP-0058-T03 --preflight-only). None of them has an effect."""
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 admission():
report = bea.evaluate(self.doc, self.schema)
if not report["valid"]:
raise Refused("admission_invalid:" + ",".join(sorted({f["rule"] for f in report["findings"]})))
def executor():
if spec["executor"]["identity"] != self.cfg.executor_identity:
raise Refused("not_this_executor")
def source():
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")
def trust():
for state in spec["trust_requires"]:
if not self.d.trust(state):
raise Refused("trust_not_satisfied:" + state)
def registry():
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")
def approval():
self._check_approval(self.d.carrier.get(self.approval_id), consumed_ok=False)
return [("admission", admission), ("executor", executor), ("source", source), ("trust", trust),
("registry", registry), ("window", self._check_window), ("approval", approval)]
def preflight(self) -> None:
for _, check in self._preflight_steps():
check()
def preflight_report(self) -> list[dict]:
"""Every preflight check with its result; value-free. An error that is not a refusal is named by type."""
results = []
for name, check in self._preflight_steps():
try:
check()
results.append({"check": name, "result": "pass"})
except Refused as refusal:
results.append({"check": name, "result": "fail", "reason": str(refusal)})
except Exception as error:
results.append({"check": name, "result": "fail", "reason": "error:" + type(error).__name__})
return results
def _check_window(self) -> None:
now = self.d.wall()