server.py

from __future__ import annotations
import math, time, random, threading
from fastapi import FastAPI
from prometheus_client import Gauge, CollectorRegistry, generate_latest, CONTENT_TYPE_LATEST
from fastapi.responses import Response, JSONResponse

app = FastAPI(title="HAL Demo Metrics Server")
registry = CollectorRegistry()
g_sigma = Gauge("hal_sigma","coherence", registry=registry)
g_entropy = Gauge("hal_entropy","entropy", registry=registry)
g_queue = Gauge("hal_queue","queue length", registry=registry)
g_phi = Gauge("hal_phi","informational potential", registry=registry)
g_p = Gauge("hal_p","policy knob p", registry=registry)

state = {"running": False}

def loop():
    t=0; p=0.6
    while state["running"]:
        # toy signals just for a visible dashboard line; not tied to any secret math
        p = max(0.0, min(1.0, p + random.uniform(-0.02,0.02)))
        sigma = 0.6 + 0.3*math.sin(t/50.0)
        H = 1.0 + 0.2*math.cos(t/60.0)
        q = max(0, int(50 + 20*math.sin(t/45.0) + random.randint(-5,5)))
        Phi = (1.0 - sigma)**2 + 0.4*((q-50)/50.0)**2 + 0.2*(1.2 - H)**2
        g_sigma.set(sigma); g_entropy.set(H); g_queue.set(q); g_phi.set(Phi); g_p.set(p)
        t += 1; time.sleep(0.1)

@app.on_event("startup")
def start():
    state["running"]=True; threading.Thread(target=loop, daemon=True).start()

@app.on_event("shutdown")
def stop():
    state["running"]=False

@app.get("/healthz")
def healthz(): return {"ok": True}

@app.get("/metrics")
def metrics():
    return Response(generate_latest(registry), media_type=CONTENT_TYPE_LATEST)

@app.get("/live")
def live():
    return JSONResponse({"sigma": g_sigma._value.get(), "H": g_entropy._value.get(),
                         "queue": g_queue._value.get(), "p": g_p._value.get(), "Phi": g_phi._value.get()})