#!/usr/bin/env python3 """Compile owner-reviewed cadence bindings from the ecosystem coordination frame.""" from __future__ import annotations import argparse import hashlib import importlib.util import json import re import sys from pathlib import Path import yaml ROOT = Path(__file__).resolve().parents[2] FRAME_REF = "coordination-engine/config/coordination-frame.json" def owner_frame(workspace: Path): source = workspace / "coordination-engine/src/coordination_engine/timing.py" spec = importlib.util.spec_from_file_location("coordination_frame_owner", source) if spec is None or spec.loader is None: raise ValueError("coordination-engine frame API is required; no fallback") module = importlib.util.module_from_spec(spec) sys.modules[spec.name] = module spec.loader.exec_module(module) return module.load_frame(workspace / FRAME_REF) def generated_module(frame, digest): return ( '"""Generated cadence timings; regenerate with NetKingdom coordination-frame tool."""\n' f"# Source: {FRAME_REF}\n" f'FRAME_CONFIG_SHA256 = "{digest}"\n' f"FRAME_SECONDS = {frame.frame_seconds}\n" f"RECEIVER_WAIT_SECONDS = {frame.seconds(2)}\n" f"EFFECTIVE_DEADLINE_SECONDS = {frame.seconds(3)}\n" ) def bind_frame_annotation(text, digest, metadata_prefix): key = "coordination.coulomb.social/frame-sha256" pattern = r"(?m)^(\s*" + re.escape(key) + r": )[^\n]+" if re.search(pattern, text): updated, count = re.subn(pattern, lambda match: match[1] + json.dumps(digest), text) else: replacement = ( metadata_prefix + "annotations:\n" + " " * (len(metadata_prefix.splitlines()[-1]) + 2) + key + ": " + json.dumps(digest) + "\n" ) # metadata_prefix ends with the indentation for annotations. Existing # annotation maps retain their contents; otherwise insert a new map. if metadata_prefix + "annotations:\n" in text: updated = text.replace(metadata_prefix + "annotations:\n", replacement, 1) count = 1 else: updated = text.replace( metadata_prefix, replacement + metadata_prefix.rsplit("\n", 1)[-1], 1 ) count = int(updated != text) if count != 1: raise ValueError("expected one runtime frame provenance binding") return updated def compile_bindings(workspace: Path): frame = owner_frame(workspace) digest = hashlib.sha256((workspace / FRAME_REF).read_bytes()).hexdigest() provenance = { "ref": FRAME_REF, "sha256": digest, "frame_seconds": frame.frame_seconds, "heartbeat_frames": 1, "registered_wait_frames": 2, "effective_deadline_frames": 3, } outputs = {} for relative in ( "approval-engine/approval_engine/coordination.py", "qonto-assistant/src/qonto_assistant/coordination.py", ): outputs[workspace / relative] = generated_module(frame, digest) approval_path = workspace / "approval-engine/cadence.yaml" approval = json.loads(approval_path.read_text()) for entry in approval["sources"]: if "heartbeat" in entry: beat = entry["heartbeat"] beat["interval_seconds"] = frame.frame_seconds ext = beat["extensions"]["approval-engine"] ext["requested_observer_max_gap_seconds"] = frame.seconds(2) ext["effective_deadline_seconds"] = frame.seconds(3) approval["extensions"].setdefault("netkingdom", {})["coordination_frame"] = provenance outputs[approval_path] = json.dumps(approval, indent=2) + "\n" qonto_path = workspace / "qonto-assistant/specs/audit-emission-cadence.yaml" qonto = yaml.safe_load(qonto_path.read_text()) for entry in qonto["sources"]: if "heartbeat" in entry: entry["heartbeat"]["interval"] = f"PT{frame.frame_seconds}S" heartbeat = qonto["extensions"]["qonto-assistant"]["heartbeat"] heartbeat["interval"] = f"PT{frame.frame_seconds}S" heartbeat["default_interval_seconds"] = frame.frame_seconds qonto["extensions"]["netkingdom"]["coordination_frame"] = provenance outputs[qonto_path] = ( "# Source-owned cadence; coordination timings generated from " + FRAME_REF + "\n" + yaml.safe_dump(qonto, sort_keys=False) ) deploy_path = workspace / "approval-engine/deploy/approval-engine.yaml" deploy = deploy_path.read_text() if re.search(r"^\s+- --heartbeat-seconds\s*$", deploy, re.M): deploy, count = re.subn( r"(?m)^(\s+- --heartbeat-seconds\n\s+- )[^\n]+", lambda m: m[1] + json.dumps(str(frame.frame_seconds)), deploy, ) else: # Explicit runtime arg also works with the already pinned image; no image # pin, storage migration, authority or unrelated deployment change here. deploy, count = re.subn( r"(?m)^(\s+)- --audit-url$", lambda m: ( f'{m[1]}- --heartbeat-seconds\n{m[1]}- "{frame.frame_seconds}"\n{m[1]}- --audit-url' ), deploy, ) if count != 1: raise ValueError("expected one Approval Engine heartbeat argument binding") outputs[deploy_path] = bind_frame_annotation( deploy, digest, " template:\n metadata:\n " ) deploy_path = workspace / "rapp-qonto/runtime/knative/service.yaml" deploy, count = re.subn( r"(?m)(name: QONTO_AUDIT_HEARTBEAT_INTERVAL_SECONDS\n\s+value: )[^\n]+", lambda m: m[1] + json.dumps(str(frame.frame_seconds)), deploy_path.read_text(), ) if count != 1: raise ValueError("expected one Qonto heartbeat environment binding") outputs[deploy_path] = bind_frame_annotation( deploy, digest, " template:\n metadata:\n " ) fixture_path = ( workspace / "kings-guard/src/kings_guard/fixtures/qonto_assistant_source_cadence.json" ) fixture = json.loads(fixture_path.read_text()) for entry in fixture["sources"]: if "heartbeat" in entry: entry["heartbeat"]["interval_seconds"] = frame.frame_seconds fixture["extensions"].setdefault("netkingdom", {})["coordination_frame"] = provenance outputs[fixture_path] = json.dumps(fixture, indent=2) + "\n" receiver = { "status": "proposed-not-applied", "coordination_frame": provenance, "sender": "approval-engine", "source": "approval-engine", "tenant": "tenant:platform", "may_read": False, "emission_cadence_ref": "approval-engine/cadence.yaml", "emission_cadence_sha256": hashlib.sha256(outputs[approval_path].encode()).hexdigest(), "heartbeat_classes": { name: frame.seconds(2) for name in ("issuance", "use", "supersession", "revocation") }, "receiver_grace_factor": 1.5, "effective_deadline_seconds": frame.seconds(3), "preserve_existing_credentials_and_other_registry_fields": True, } outputs[ workspace / "net-kingdom/generated/coordination-frame/approval-engine-receiver.json" ] = json.dumps(receiver, indent=2) + "\n" return frame, outputs def main(argv=None): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--workspace", type=Path, default=ROOT.parent) mode = parser.add_mutually_exclusive_group() mode.add_argument("--check", action="store_true") mode.add_argument("--write", action="store_true") args = parser.parse_args(argv) try: frame, outputs = compile_bindings(args.workspace) changed = { p: text for p, text in outputs.items() if not p.exists() or p.read_text() != text } if args.write: for path, text in changed.items(): path.parent.mkdir(parents=True, exist_ok=True) path.write_text(text) report = { "frame": frame.as_dict(), "mode": "write" if args.write else "check" if args.check else "plan", "changed": [str(p.relative_to(args.workspace)) for p in changed], "live_operations": False, } print(json.dumps(report, indent=2, sort_keys=True)) return 1 if args.check and changed else 0 except (OSError, ValueError, KeyError, TypeError) as exc: parser.error(str(exc)) if __name__ == "__main__": raise SystemExit(main())