szl-command-lab / server.py
betterwithage's picture
deploy(hf): sync szl-holdings/szl-command-lab@ad181ebfbee8706fd1f714da9db862f4dab6cd7c derived COPY set
7c8a5fa verified
Raw
History Blame Contribute Delete
21.6 kB
#!/usr/bin/env python3
# SPDX-License-Identifier: Apache-2.0
# Copyright 2026 SZL Holdings
# Signed-off-by: Lutar, Stephen P. <stephenlutar2@gmail.com>
"""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 "<html" in t.lower() or "<!doctype" in t.lower():
honesty = "REACHABLE"
detail = "html 200 · not a kernel healthz"
else:
honesty = "REACHABLE"
detail = "http 200"
return {"id": ident, "role": role, "href": href, "honesty": honesty, "detail": detail, "http": http}
with ThreadPoolExecutor(max_workers=8) as pool:
futs = [pool.submit(one, row) for row in SURFACES]
surfaces = [f.result() for f in futs]
live_n = sum(1 for s in surfaces if s["honesty"] == "LIVE")
reach_n = sum(1 for s in surfaces if s["honesty"] == "REACHABLE")
payload = {
"captured_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
"source": "SZLHOLDINGS/szl-command-lab",
"kernel": {
"ok": body["live_count"] == 5 and not body["blocked"],
"live_count": body["live_count"],
"blocked": body["blocked"],
"verdict": body["verdict"],
"conjecture_1": "OPEN",
"proven_trust": False,
"reason": body["reason"],
"organs": body["organs"],
"energy": energy,
},
"surfaces": surfaces,
"live_surfaces": live_n,
"reachable_surfaces": reach_n,
}
_estate_cache, _estate_at = payload, now
return payload
HTML = r"""<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8"/>
<meta name="viewport" content="width=device-width,initial-scale=1"/>
<title>SZL Command lab</title>
<style>
:root{--bg:#05070d;--void:#080c14;--panel:#0b121b;--line:#1b2734;--proof:#3af4c8;--gold:#e8c074;--cream:#eef3f6;--para:#9fb1bf;--dim:#6c7d8c;--false:#d76a6a}
*{box-sizing:border-box}
html,body{margin:0;background:var(--bg);color:var(--cream);font:15px/1.5 ui-sans-serif,system-ui,sans-serif;min-height:100%}
body{background:radial-gradient(120% 90% at 50% -10%,rgba(232,192,116,.09),transparent 60%),var(--bg)}
main{max-width:960px;margin:0 auto;padding:28px 18px 64px}
.eyebrow{font:10.5px ui-monospace,monospace;letter-spacing:1.4px;text-transform:uppercase;color:var(--gold)}
h1{font-size:clamp(22px,4vw,32px);margin:.2em 0 .4em}
.lede{color:var(--para);max-width:70ch}
.badges{display:flex;flex-wrap:wrap;gap:8px;margin:14px 0 22px}
.badge{font:10.5px ui-monospace,monospace;padding:3px 9px;border:1px solid var(--line);border-radius:6px;color:var(--para);background:#10171f}
.badge b{color:var(--gold)}
.grid{display:grid;grid-template-columns:repeat(auto-fit,minmax(140px,1fr));gap:10px;margin:18px 0}
.card{border:1px solid var(--line);border-radius:11px;background:var(--panel);padding:14px}
.card .k{font:10px ui-monospace,monospace;letter-spacing:.8px;text-transform:uppercase;color:var(--dim)}
.card .v{font:22px/1.2 ui-monospace,monospace;margin-top:4px}
.ok{color:var(--proof)} .hold{color:var(--gold)} .down{color:var(--false)}
button{min-height:44px;padding:0 16px;border-radius:8px;border:1px solid var(--proof);background:var(--proof);color:var(--bg);font:600 12px ui-monospace,monospace;letter-spacing:.8px;text-transform:uppercase;cursor:pointer}
.row{display:flex;flex-wrap:wrap;gap:8px;margin:12px 0}
label{font:12px ui-monospace,monospace;color:var(--para);display:flex;align-items:center;gap:6px}
pre{overflow:auto;background:var(--void);border:1px solid var(--line);border-radius:11px;padding:12px;font:11px/1.45 ui-monospace,monospace;color:var(--cream)}
a{color:var(--proof)}
.stage{height:280px;border:1px solid var(--line);border-radius:14px;position:relative;background:linear-gradient(180deg,rgba(58,244,200,.05),var(--void));margin:16px 0;overflow:hidden}
.node{position:absolute;transform:translate(-50%,-50%);border:1px solid var(--proof);border-radius:10px;padding:8px 10px;font:11px ui-monospace,monospace;background:rgba(11,18,27,.9)}
.node.down{border-color:var(--false);color:var(--false)}
.foot{margin-top:28px;border-top:1px solid var(--line);padding-top:14px;color:var(--dim);font:11px ui-monospace,monospace}
.surfaces{display:grid;grid-template-columns:repeat(auto-fit,minmax(160px,1fr));gap:8px;margin-top:12px}
</style>
</head>
<body>
<main>
<div class="eyebrow">SZL Holdings · Command lab · GitHub canonical · Hub operational</div>
<h1>Holographic command body. Fail closed.</h1>
<p class="lede">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.</p>
<div class="badges" id="badges"></div>
<div class="stage" id="stage"></div>
<div class="grid" id="metrics"></div>
<div class="row">
<label><input type="checkbox" id="zero_heart"> Zero Yuyay</label>
<label><input type="checkbox" id="tamper_chain"> Tamper YAWAR</label>
<label><input type="checkbox" id="fabricate_joule"> Fabricate joule</label>
<button id="run">Run organ cycle</button>
</div>
<p class="eyebrow" style="margin-top:28px">Estate recapture</p>
<div class="surfaces" id="surfaces"></div>
<pre id="out">loading…</pre>
<p class="foot">Source <a href="https://github.com/szl-holdings/szl-command-lab">szl-holdings/szl-command-lab</a> · product <a href="https://a-11-oy.com">a-11-oy.com</a> · proof <a href="https://a11oy.net">a11oy.net</a></p>
</main>
<script>
const slots={BRAIN:['18%','50%'],HEART:['38%','50%'],CIRCULATORY:['58%','50%'],NERVOUS:['58%','22%'],SKELETON:['82%','50%']};
async function cycle(){
const q=new URLSearchParams({
zero_heart: document.getElementById('zero_heart').checked? '1':'0',
tamper_chain: document.getElementById('tamper_chain').checked? '1':'0',
fabricate_joule: document.getElementById('fabricate_joule').checked? '1':'0',
});
const r=await fetch('/api/organs/integrity?'+q);
const j=await r.json();
const body=j.body||j;
const energy=j.energy||{};
document.getElementById('out').textContent=JSON.stringify({verdict:body.verdict, live:body.live_count, energy, reason:body.reason, head:body.chain_head},null,2);
const eh=energy.honesty||'UNAVAILABLE';
const ch=energy.channel||'LIVE';
document.getElementById('badges').innerHTML=[
['organs', (body.live_count||0)+'/5'],
['energy channel', ch],
['joule', eh],
['Λ', body.conjecture_1||'OPEN'],
['proven_trust', String(body.proven_trust===true)],
].map(([k,v])=>`<span class="badge">${k} <b>${v}</b></span>`).join('');
document.getElementById('metrics').innerHTML=(body.organs||[]).map(o=>`<div class="card"><div class="k">${o.name}</div><div class="v ${o.status==='DOWN'?'down':o.honesty==='UNAVAILABLE'?'hold':'ok'}">${o.status}</div></div>`).join('');
document.getElementById('stage').innerHTML=(body.organs||[]).map(o=>{
const [t,l]=slots[o.name]||['50%','50%'];
return `<div class="node ${o.status==='DOWN'?'down':''}" style="top:${t};left:${l}">${o.name}<br>${o.status}</div>`;
}).join('');
}
async function estate(){
const r=await fetch('/api/estate');
const j=await r.json();
document.getElementById('surfaces').innerHTML=(j.surfaces||[]).map(s=>{
const cls=s.honesty==='LIVE'?'ok':s.honesty==='REACHABLE'?'hold':'down';
return `<div class="card"><div class="k">${s.role}</div><div class="v ${cls}">${s.honesty}</div><div class="k">${s.id}</div></div>`;
}).join('');
}
document.getElementById('run').onclick=cycle;
cycle();
estate();
</script>
</body>
</html>
"""
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())