File size: 6,636 Bytes
5f65366 4e1e99e 97c22de 7a7f0f6 5f65366 9242d90 5f65366 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 | #!/usr/bin/env python3
"""Emit JSON/Markdown alignment badge for community nodes (Phase 2)."""
from __future__ import annotations
import argparse
import json
import subprocess
import sys
from datetime import datetime, timezone
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
SIGNATURE = "Δ9Φ963-PHASE2-DEPLOYMENT"
def _run(cmd: list[str], cwd: Path) -> tuple[int, str]:
cp = subprocess.run(cmd, cwd=cwd, capture_output=True, text=True, timeout=300)
out = (cp.stdout or "") + (cp.stderr or "")
return cp.returncode, out.strip()
def collect_badge(*, quick: bool = False) -> dict:
badge: dict = {
"signature": SIGNATURE,
"timestamp": datetime.now(timezone.utc).isoformat(),
"repo": str(ROOT),
"checks": {},
"status": "UNKNOWN",
}
golden = ROOT / "protocol0_nano_kernel" / "fixtures" / "p0_canonical.sha256"
badge["checks"]["p0_golden_sha"] = golden.is_file()
stack = None
sys.path.insert(0, str(ROOT / "stack"))
for sub in (
"protocol0_nano_kernel/src/python",
"protocol1_memory_mycelium/src/python",
"protocol2_cognitive_bridge/src/python",
"protocol3_vortex_consensus/src/python",
"protocol4_ascension_engine/src/python",
"protocol5_harmony_node/src/python",
):
p = ROOT / sub
if str(p) not in sys.path:
sys.path.insert(0, str(p))
try:
from lygo_stack import deploy_stack # noqa: E402
stack = deploy_stack("BADGE_PROBE")
demo = stack.demo_cycle()
badge["checks"]["stack_demo"] = demo.get("p0", {}).get("verdict") == "AMPLIFY"
badge["stack_version"] = stack.version
except Exception as exc:
badge["checks"]["stack_demo"] = False
badge["stack_error"] = str(exc)
reg_path = ROOT / "data" / "kernel_eggs" / "registry.json"
if reg_path.is_file():
try:
kr = json.loads(reg_path.read_text(encoding="utf-8"))
badge["kernel_egg_registry_merkle_root"] = kr.get("registry_merkle_root")
ke = ROOT / "tests" / "kernel_eggs_last_run.json"
if ke.is_file():
kv = json.loads(ke.read_text(encoding="utf-8"))
badge["checks"]["kernel_eggs"] = kv.get("all_pass", False)
else:
badge["checks"]["kernel_eggs"] = None
except Exception:
badge["checks"]["kernel_eggs"] = False
else:
badge["checks"]["kernel_eggs"] = None
sr_path = ROOT / "data" / "scalable_registry" / "registry.json"
if sr_path.is_file():
try:
sr = json.loads(sr_path.read_text(encoding="utf-8"))
badge["scalable_registry_merkle_root"] = sr.get("global_merkle_root")
srun = ROOT / "tests" / "scalable_registry_last_run.json"
if srun.is_file():
sv = json.loads(srun.read_text(encoding="utf-8"))
badge["checks"]["scalable_registry"] = sv.get("all_pass", False)
else:
badge["checks"]["scalable_registry"] = None
except Exception:
badge["checks"]["scalable_registry"] = False
else:
badge["checks"]["scalable_registry"] = None
mesh_art = ROOT / "tests" / "mesh_scale_last_run.json"
if mesh_art.is_file():
try:
mr = json.loads(mesh_art.read_text(encoding="utf-8"))
badge["checks"]["phase5_mesh_scale"] = bool(mr.get("under_10_rounds"))
badge["mesh_convergence_rounds"] = mr.get("convergence_rounds")
except Exception:
badge["checks"]["phase5_mesh_scale"] = False
else:
badge["checks"]["phase5_mesh_scale"] = False
try:
if stack is not None:
coord = stack.elasticity
fed = stack.federation
coord.scatter_prioritized({"probe": True}, "BADGE_PROBE", verdict_hint="AMPLIFY")
fed.announce_alignment({"status": "PROBING", "signature": SIGNATURE})
badge["checks"]["phase1_elasticity"] = coord.status()["version"].startswith("Δ9")
badge["checks"]["phase3_4_federation"] = fed.version.startswith("Δ9")
else:
badge["checks"]["phase1_elasticity"] = False
badge["checks"]["phase3_4_federation"] = False
except Exception as exc:
badge["checks"]["phase1_elasticity"] = False
badge["checks"]["phase3_4_federation"] = False
badge["elasticity_error"] = str(exc)
if not quick:
rc, _ = _run([sys.executable, "tools/run_grok_audit_demo.py", "--no-report"], ROOT)
badge["checks"]["grok_audit_cli"] = rc == 0
rc2, out2 = _run([sys.executable, "tools/verify_lattice_alignment.py"], ROOT)
badge["checks"]["lattice"] = rc2 == 0 and "ALIGNED" in out2
else:
badge["checks"]["grok_audit_cli"] = None
badge["checks"]["lattice"] = None
required = [k for k, v in badge["checks"].items() if v is not None]
badge["status"] = "ALIGNED" if all(badge["checks"][k] for k in required) else "NEEDS_FIX"
return badge
def to_markdown(badge: dict) -> str:
status = badge.get("status", "UNKNOWN")
icon = "✅" if status == "ALIGNED" else "⚠️"
lines = [
f"### {icon} LYGO Alignment Badge",
"",
f"**Status:** `{status}` · **Signature:** `{badge.get('signature')}`",
f"**Stack:** `{badge.get('stack_version', 'n/a')}` · **UTC:** `{badge.get('timestamp')}`",
"",
"| Check | OK |",
"|-------|-----|",
]
for k, v in sorted((badge.get("checks") or {}).items()):
if v is None:
lines.append(f"| {k} | skipped |")
else:
lines.append(f"| {k} | {'✅' if v else '❌'} |")
return "\n".join(lines)
def main() -> int:
ap = argparse.ArgumentParser(description="LYGO node alignment badge")
ap.add_argument("--quick", action="store_true", help="Skip full audit + lattice (faster Docker health)")
ap.add_argument("--format", choices=("json", "md", "both"), default="both")
ap.add_argument("--out", type=Path, default=ROOT / "tests" / "alignment_badge.json")
args = ap.parse_args()
badge = collect_badge(quick=args.quick)
if args.format in ("json", "both"):
args.out.write_text(json.dumps(badge, indent=2), encoding="utf-8")
print(json.dumps(badge, indent=2))
if args.format in ("md", "both"):
print("\n" + to_markdown(badge))
if badge["status"] == "ALIGNED":
print("BADGE VALID")
return 0 if badge["status"] == "ALIGNED" else 1
if __name__ == "__main__":
raise SystemExit(main()) |