#!/usr/bin/env python3 # SPDX-License-Identifier: Apache-2.0 # Copyright 2026 SZL Holdings # Signed-off-by: Lutar, Stephen P. """Flatten payload for Hub. Kernel + energy + estate recapture. Stdlib HTTP 7860.""" from __future__ import annotations import json import os import sys import time from concurrent.futures import ThreadPoolExecutor from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from urllib.parse import parse_qs, urlparse from urllib.request import Request, urlopen HERE = Path(__file__).resolve().parent sys.path.insert(0, str(HERE)) sys.path.insert(0, str(HERE / "python")) try: from energy import measure_run, probe from kernel import evaluate_anatomy, selftest except ImportError: # Immune flatten historically copied only server.py. Keep a local fallback. import hashlib import math from datetime import datetime, timezone LOCKED_EIGHT = ("F1", "F4", "F7", "F11", "F12", "F18", "F19", "F22") YUYAY_FLOORS = (0.95, 0.95) + (0.90,) * 11 ZERO = "0" * 64 CHAIN_OPS = ("anatomy.brain", "anatomy.heart", "anatomy.skeleton") POWERCAP = Path("/sys/class/powercap") RAPL = Path("/sys/class/powercap/intel-rapl:0/energy_uj") def _rapl_uj(): candidates = [RAPL] try: if POWERCAP.is_dir(): candidates.extend(sorted(POWERCAP.glob("intel-rapl:*/energy_uj"))) except OSError: pass seen = set() for path in candidates: if path in seen: continue seen.add(path) try: if path.is_file(): return int(path.read_text().strip()) except (OSError, ValueError): continue return None def _nvml_mj(): try: import pynvml # type: ignore except ImportError: return None try: pynvml.nvmlInit() handle = pynvml.nvmlDeviceGetHandleByIndex(0) mj = float(pynvml.nvmlDeviceGetTotalEnergyConsumption(handle)) pynvml.nvmlShutdown() return mj except Exception: try: pynvml.nvmlShutdown() except Exception: pass return None def probe(*, sample_s: float = 0.05): a = _rapl_uj() if a is not None: time.sleep(max(0.0, sample_s)) b = _rapl_uj() or a return { "channel": "LIVE", "honesty": "MEASURED", "source": "intel-rapl", "package_energy_j": b / 1_000_000.0, "sample_delta_j": max(0.0, (b - a) / 1_000_000.0), "inference_energy_j": None, "energy_j": None, "note": "RAPL package counter MEASURED.", } mj = _nvml_mj() if mj is not None: return { "channel": "LIVE", "honesty": "MEASURED", "source": "nvml", "package_energy_j": mj / 1000.0, "sample_delta_j": None, "inference_energy_j": None, "energy_j": None, "note": "NVML total energy MEASURED.", } return { "channel": "LIVE", "honesty": "UNAVAILABLE", "source": None, "package_energy_j": None, "sample_delta_j": None, "inference_energy_j": None, "energy_j": None, "note": "No RAPL, no NVML. Channel is live. Never a fabricated joule.", } def measure_run(fn): a = _rapl_uj() t0 = time.perf_counter() result = fn() dt = time.perf_counter() - t0 b = _rapl_uj() energy = probe(sample_s=0.0) energy["duration_s"] = dt if a is not None and b is not None: energy["honesty"] = "MEASURED" energy["inference_energy_j"] = max(0.0, (b - a) / 1_000_000.0) energy["energy_j"] = energy["inference_energy_j"] energy["note"] = f"RAPL delta around kernel · {dt:.4f}s" return result, energy def _sha256_hex(text: str) -> str: return hashlib.sha256(text.encode("utf-8")).hexdigest() def wgm(xs, ws): if len(xs) != len(ws) or not xs: return 0.0 if any((not math.isfinite(x)) or x <= 0.0 for x in xs): return 0.0 if abs(sum(ws) - 1.0) >= 1e-9: return 0.0 value = math.exp(sum(w * math.log(x) for x, w in zip(xs, ws))) return value if math.isfinite(value) else 0.0 def evaluate_lambda(axes): n = len(axes) weights = tuple(1.0 / n for _ in range(n)) if n else () value = wgm(axes, weights) return {"value": float(value), "blocked": value == 0.0} def yawar_chain(seed: int, tamper: bool): hops = [] prev = ZERO for seq, op in enumerate(CHAIN_OPS): digest = _sha256_hex(f"{seq}|{op}|{prev}|{int(seed)}") hops.append({"seq": seq, "op": op, "prev": prev, "digest": digest}) prev = digest if tamper and len(hops) > 1: hops[1] = dict(hops[1]) hops[1]["prev"] = "deadbeef" + hops[1]["prev"][8:] walk = ZERO ok = True for hop in hops: expect = _sha256_hex(f"{hop['seq']}|{hop['op']}|{hop['prev']}|{int(seed)}") if hop["prev"] != walk or expect != hop["digest"]: ok = False break walk = hop["digest"] return {"ok": ok, "head": hops[-1]["digest"] if hops else ZERO} def evaluate_anatomy(*, zero_heart=False, tamper_chain=False, fabricate_joule=False, seed=11): axes = list(YUYAY_FLOORS) if zero_heart: axes[0] = 0.0 heart = evaluate_lambda(axes) chain = yawar_chain(int(seed), bool(tamper_chain)) organs = [ {"name": "BRAIN", "status": "LIVE", "honesty": "LIVE"}, {"name": "HEART", "status": "DOWN" if heart["blocked"] else "LIVE", "honesty": "ADVISORY"}, {"name": "CIRCULATORY", "status": "DOWN" if not chain["ok"] else "LIVE", "honesty": "LIVE"}, {"name": "NERVOUS", "status": "DOWN" if fabricate_joule else "LIVE", "honesty": "UNAVAILABLE"}, {"name": "SKELETON", "status": "LIVE", "honesty": "ADVISORY"}, ] live = sum(1 for o in organs if o["status"] == "LIVE") blocked = any(o["status"] == "DOWN" for o in organs) return { "organs": organs, "live_count": live, "blocked": blocked, "verdict": "BLOCKED" if blocked else "ADVISORY_BODY", "energy": "UNAVAILABLE", "energy_j": None, "conjecture_1": "OPEN", "locked_proven": 8, "locked_ids": list(LOCKED_EIGHT), "proven_trust": False, "chain_head": chain["head"], "reason": ( f"organ integrity {live}/5 LIVE · Λ advisory · energy UNAVAILABLE · Conjecture 1 OPEN" if not blocked else "organ integrity FAIL · fail closed" ), "checked_at": datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z"), } def selftest(): healthy = evaluate_anatomy(seed=11) assert healthy["live_count"] == 5 and healthy["blocked"] is False assert evaluate_anatomy(zero_heart=True)["blocked"] is True assert evaluate_anatomy(tamper_chain=True)["blocked"] is True assert evaluate_anatomy(fabricate_joule=True)["blocked"] is True return {"ok": True, "cases": 4, "healthy_head": healthy["chain_head"]} SURFACES = [ ("command-lab", "Operator kernel", "https://huggingface.co/spaces/SZLHOLDINGS/szl-command-lab", None), ("a11oy", "Product command", "https://huggingface.co/spaces/SZLHOLDINGS/a11oy", "https://szlholdings-a11oy.hf.space/healthz"), ("killinchu", "Bounded vertical", "https://huggingface.co/spaces/SZLHOLDINGS/killinchu", "https://szlholdings-killinchu.hf.space/healthz"), ("khipu", "Python kernels", "https://huggingface.co/spaces/SZLHOLDINGS/szl-khipu", "https://szlholdings-szl-khipu.hf.space/"), ("anatomy", "Living body", "https://huggingface.co/spaces/SZLHOLDINGS/anatomy", "https://szlholdings-anatomy.hf.space/healthz"), ("immune", "Defense matrix", "https://huggingface.co/spaces/SZLHOLDINGS/immune", "https://szlholdings-immune.hf.space/healthz"), ("sovereign-os", "Operator OS", "https://huggingface.co/spaces/SZLHOLDINGS/szl-sovereign-os", "https://szlholdings-szl-sovereign-os.hf.space/healthz"), ("real-estate", "Public records", "https://huggingface.co/spaces/SZLHOLDINGS/szl-real-estate", "https://szlholdings-szl-real-estate.hf.space/healthz"), ("cosmos", "Estate map", "https://huggingface.co/spaces/SZLHOLDINGS/cosmos", "https://szlholdings-cosmos.hf.space/"), ("counsel", "Counsel hologram", "https://huggingface.co/spaces/SZLHOLDINGS/counsel", "https://szlholdings-counsel.hf.space/"), ] _estate_cache: dict | None = None _estate_at = 0.0 def _hit(url: str, timeout: float = 4.0) -> tuple[int | None, str]: req = Request(url, headers={"User-Agent": "szl-command-lab-operator", "Accept": "application/json, text/html;q=0.8"}) try: with urlopen(req, timeout=timeout) as res: return res.status, res.read(240).decode("utf-8", "replace") except Exception as exc: code = getattr(exc, "code", None) return code, "" def recapture_estate() -> dict: global _estate_cache, _estate_at now = time.time() if _estate_cache and now - _estate_at < 20: return _estate_cache energy = probe() body = evaluate_anatomy(seed=11) surfaces = [] def one(row): ident, role, href, url = row if url is None: return { "id": ident, "role": role, "href": href, "honesty": "LIVE" if body["live_count"] == 5 else "UNAVAILABLE", "detail": f"local kernel {body['live_count']}/5", "http": 200, } http, text = _hit(url) honesty = "UNAVAILABLE" detail = "no answer" if http is None else f"HTTP {http}" if http == 200: t = text.strip() if t.startswith("{") or t.startswith("["): honesty = "LIVE" detail = "json 200" elif " SZL Command lab
SZL Holdings · Command lab · GitHub canonical · Hub operational

Holographic command body. Fail closed.

One kernel. Ten estate surfaces. Energy channel LIVE. Joules MEASURED only from RAPL or NVML. Never a fabricated joule. Λ uniqueness is Conjecture 1 OPEN. Not a-11-oy.com. Not an ATO. Not an elevation.

Estate recapture

loading…

Source szl-holdings/szl-command-lab · product a-11-oy.com · proof a11oy.net

""" JSON_PATHS = {"/healthz", "/readyz", "/api/energy", "/api/organs/integrity", "/v1/organs/integrity", "/api/estate"} HTML_PATHS = {"/", "/index.html"} def _flag(qs, name: str) -> bool: v = (qs.get(name) or ["0"])[0] return v in {"1", "true", "on", "yes"} class Handler(BaseHTTPRequestHandler): def log_message(self, fmt: str, *args) -> None: # noqa: A003 sys.stderr.write("%s - %s\n" % (self.address_string(), fmt % args)) def do_HEAD(self) -> None: # noqa: N802 path = urlparse(self.path).path.rstrip("/") or "/" self.send_response(200 if path in HTML_PATHS or path in JSON_PATHS else 404) self.send_header("Content-Type", "text/html; charset=utf-8" if path in HTML_PATHS else "application/json; charset=utf-8") self.send_header("Content-Length", "0") self.send_header("Cache-Control", "no-store") self.end_headers() def do_GET(self) -> None: # noqa: N802 parsed = urlparse(self.path) path = parsed.path.rstrip("/") or "/" qs = parse_qs(parsed.query) if path in {"/healthz", "/readyz"}: self._send(200, {"ok": True, "energy": probe(), "proven_trust": False, "channel": "LIVE", "space": "szl-command-lab"}) return if path == "/api/energy": self._send(200, probe()) return if path == "/api/estate": self._send(200, recapture_estate()) return if path in {"/api/organs/integrity", "/v1/organs/integrity"}: def run(): return evaluate_anatomy( zero_heart=_flag(qs, "zero_heart"), tamper_chain=_flag(qs, "tamper_chain"), fabricate_joule=_flag(qs, "fabricate_joule"), seed=11, ) body, energy = measure_run(run) if _flag(qs, "fabricate_joule"): energy = {"channel": "LIVE", "honesty": "UNAVAILABLE", "energy_j": None, "inference_energy_j": None, "note": "fabricated joule refused"} body["blocked"] = True body["verdict"] = "BLOCKED" body["energy"] = energy.get("honesty") body["energy_j"] = energy.get("energy_j") self._send(200, {"ok": True, "body": body, "energy": energy}) return if path in HTML_PATHS: raw = HTML.encode("utf-8") self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Content-Length", str(len(raw))) self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(raw) return self._send(404, {"ok": False, "error": "not found"}) def _send(self, status: int, obj) -> None: raw = json.dumps(obj, indent=2, default=str).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(raw))) self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(raw) def main() -> int: selftest() host, port = "0.0.0.0", int(os.environ.get("PORT", "7860")) httpd = ThreadingHTTPServer((host, port), Handler) print(f"[szl-command-lab] {host}:{port} · kernel + estate recapture · never a fabricated joule", file=sys.stderr) httpd.serve_forever() return 0 if __name__ == "__main__": raise SystemExit(main())