Prepare guarded cadence registry updates and Qonto receiver admission

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a13eab-1e6a-7b33-9c37-372e90ada3a2

scripts/register-approval-engine-cadence.py

@@ -0,0 +1,78 @@
+#!/usr/bin/env python3
+"""Silent attended registry CAS; no rollout, no credential mint, no local secrets.
+
+Run through the existing warden attended --exec envelope. Default is preflight
+only. --apply authorizes exactly the proposal's two metadata fields. Receipt
+contains metadata only. Receiver ESO synchronization/restart is a separate gate.
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import os
+from pathlib import Path
+import subprocess
+import sys
+from datetime import datetime, timezone
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+from audit_core.cadence_registration import prepare, strict_json
+
+REGISTRY = "platform/data/workloads/audit-core/senders"
+
+
+def bao(args, payload=None):
+    proc = subprocess.run(["bao", *args], input=None if payload is None else json.dumps(payload).encode(),
+                          stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False)
+    if proc.returncode:
+        raise RuntimeError("bao_operation_failed")
+    return strict_json(proc.stdout)
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--proposal", required=True, type=Path)
+    parser.add_argument("--declaration", required=True, type=Path)
+    parser.add_argument("--receipt", required=True, type=Path)
+    parser.add_argument("--apply", action="store_true")
+    args = parser.parse_args()
+    receipt = {"observed_at": datetime.now(timezone.utc).isoformat(),
+               "registry": "platform/workloads/audit-core/senders",
+               "credential_values_emitted": False, "applied": False}
+    try:
+        if Path.home().parent.name != ".warden-attended-login":
+            raise ValueError("attended_warden_envelope_required")
+        if any(os.environ.get(k) for k in ("BAO_TOKEN", "VAULT_TOKEN", "OPENBAO_TOKEN")):
+            raise ValueError("token_environment_forbidden")
+        identity = bao(["token", "lookup", "-format=json"])["data"]
+        policies = identity.get("policies", [])
+        if "platform-admin" not in policies or "root" in policies:
+            raise ValueError("attended_platform_admin_required")
+        raw = bao(["read", "-format=json", REGISTRY])["data"]
+        version = raw["metadata"]["version"]
+        body = raw["data"]
+        updated, detail = prepare(body, args.declaration.read_bytes(), strict_json(args.proposal.read_bytes()))
+        receipt.update(detail, registry_version_before=version)
+        if args.apply and detail["changed"]:
+            result = bao(["write", "-format=json", REGISTRY, "-"],
+                         {"options": {"cas": version}, "data": updated})
+            receipt["applied"] = True
+            receipt["request_id"] = result.get("request_id")
+            after = bao(["read", "-format=json", REGISTRY])["data"]
+            if after["metadata"]["version"] != version + 1 or after["data"] != updated:
+                raise RuntimeError("readback_mismatch")
+            receipt["registry_version_after"] = after["metadata"]["version"]
+        receipt["status"] = "registered" if args.apply else "preflight_only"
+        code = 0
+    except Exception:
+        # Never propagate exceptions containing parser input or subprocess
+        # output: either may contain an incumbent registry credential.
+        receipt["status"] = "failed_closed"
+        code = 1
+    args.receipt.parent.mkdir(parents=True, exist_ok=True)
+    args.receipt.write_text(json.dumps(receipt, indent=2) + "\n")
+    return code
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())