Update app.py

app.py

@@ -1,251 +1,91 @@
 # ============================================================
-# RFT-Ω FRAMEWORK — TOTAL-PROOF API (Sprint 1, Stable Build)
-# Author: Liam Grinstead (RFT Systems)  |  All Rights Reserved
+# RFT-Ω FRAMEWORK — TOTAL-PROOF API (Gradio Stand-Alone Build)
+# Author: Liam Grinstead | RFT Systems | All Rights Reserved
 # ============================================================
-"""
-Deterministic, signed validation harness for the
-Rendered Frame Theory (RFT-Ω) harmonic-stability kernel.
-"""
 
 import os, json, time, hashlib, zipfile, random
 from datetime import datetime
-from typing import Dict, Any, List, Tuple
 import numpy as np
 import gradio as gr
 
-try:
-    from fastapi import FastAPI
-    HAVE_FASTAPI = True
-except Exception:
-    HAVE_FASTAPI = False
-
-
 # ------------------ About / Legal ---------------------------
 RFT_VERSION = "v4.0-total-proof-stable"
 RFT_DOI = "https://doi.org/10.5281/zenodo.17466722"
-HF_URL = "https://rftsystems-rft-omega-api.hf.space"
 LEGAL_NOTICE = (
     "All Rights Reserved — RFT-IPURL v1.0 (UK / Berne). "
-    "Research validation use only. No reverse-engineering or derivative kernels without written consent."
+    "Research validation use only. No reverse-engineering without written consent."
 )
 
-ABOUT_BLOCK = {
-    "name": "RFT-Ω Framework — Total-Proof API",
-    "version": RFT_VERSION,
-    "doi": RFT_DOI,
-    "space": HF_URL,
-    "profiles": ["AI / Neural", "SpaceX / Aerospace", "Energy / RHES", "Extreme Perturbation"],
-    "noise_distributions": ["gauss", "uniform"],
-    "schedules": ["single", "ramp", "random", "impulse", "step"],
-    "metrics": ["QΩ", "ζ_sync", "status"],
-    "integrity": ["run_id", "sha512(config+results)", "bundle(zip)"],
-    "legal": LEGAL_NOTICE,
-}
-
-
-# ------------------ Rate-limit ------------------------------
-RUN_HISTORY_TS: List[float] = []
-MAX_RUNS_PER_MINUTE = 60
-
-def _rate_limit_ok() -> Tuple[bool, str]:
-    now = time.time()
-    while RUN_HISTORY_TS and now - RUN_HISTORY_TS[0] > 60:
-        RUN_HISTORY_TS.pop(0)
-    if len(RUN_HISTORY_TS) >= MAX_RUNS_PER_MINUTE:
-        return False, "Rate limit exceeded (demo fairness). Please retry shortly."
-    RUN_HISTORY_TS.append(now)
-    return True, "ok"
-
-
-# ------------------ Profiles / Simulation -------------------
 PROFILES = {
-    "AI / Neural":         {"base": (0.86, 0.80), "w": (0.65, 0.35)},
-    "SpaceX / Aerospace":  {"base": (0.84, 0.79), "w": (0.60, 0.40)},
-    "Energy / RHES":       {"base": (0.83, 0.78), "w": (0.55, 0.45)},
-    "Extreme Perturbation":{"base": (0.82, 0.77), "w": (0.50, 0.50)},
+    "AI / Neural": {"base": (0.86, 0.80), "w": (0.65, 0.35)},
+    "SpaceX / Aerospace": {"base": (0.84, 0.79), "w": (0.60, 0.40)},
+    "Energy / RHES": {"base": (0.83, 0.78), "w": (0.55, 0.45)},
+    "Extreme Perturbation": {"base": (0.82, 0.77), "w": (0.50, 0.50)},
 }
 
-def _rng(seed:int)->np.random.RandomState:
-    return np.random.RandomState(seed)
+def _rng(seed:int): return np.random.RandomState(seed)
 
-def simulate_step(rng:np.random.RandomState, profile:str, sigma:float, noise_dist:str)->Dict[str,Any]:
+def simulate_step(rng, profile, sigma, dist):
     base_q, base_z = PROFILES[profile]["base"]
     wq, wz = PROFILES[profile]["w"]
-    if noise_dist == "uniform":
-        q_noise = rng.uniform(-sigma, sigma)
-        z_noise = rng.uniform(-sigma*0.8, sigma*0.8)
+    if dist == "uniform":
+        qn = rng.uniform(-sigma, sigma)
+        zn = rng.uniform(-sigma*0.8, sigma*0.8)
     else:
-        q_noise = rng.normal(0, sigma)
-        z_noise = rng.normal(0, sigma*0.8)
-    q = float(np.clip(base_q + wq*q_noise, 0.0, 0.99))
-    z = float(np.clip(base_z + wz*z_noise, 0.0, 0.99))
-    variance = abs(q_noise)+abs(z_noise)
+        qn = rng.normal(0, sigma)
+        zn = rng.normal(0, sigma*0.8)
+    q = float(np.clip(base_q + wq*qn, 0.0, 0.99))
+    z = float(np.clip(base_z + wz*zn, 0.0, 0.99))
+    variance = abs(qn)+abs(zn)
     if variance > 0.15:
         status="critical"
     elif variance > 0.07:
         status="perturbed"
     else:
         status="nominal"
-    return {"sigma": round(sigma,6),"QΩ":q,"ζ_sync":z,"status":status}
-
-
-# ------------------ Schedules -------------------------------
-def build_sigma_series(schedule_type:str, params:Dict[str,Any])->List[float]:
-    if schedule_type=="single":
-        return [float(params.get("sigma",0.05))]
-    elif schedule_type=="ramp":
-        return list(np.linspace(float(params.get("start",0.0)),
-                                float(params.get("stop",0.3)),
-                                int(params.get("steps",10))))
-    elif schedule_type=="random":
-        r = random.Random(int(params.get("seed",0)))
-        return [r.uniform(float(params.get("min",0.0)),
-                          float(params.get("max",0.3)))
-                for _ in range(int(params.get("steps",10)))]
-    elif schedule_type=="impulse":
-        base,spike,at,steps = float(params.get("base",0.05)), float(params.get("spike",0.25)), int(params.get("at",5)), int(params.get("steps",10))
-        s=[base]*steps
-        if 0<=at<len(s): s[at]=spike
-        return s
-    elif schedule_type=="step":
-        before,after,at,steps = float(params.get("before",0.05)), float(params.get("after",0.2)), int(params.get("at",5)), int(params.get("steps",10))
-        s=[before]*steps
-        for i in range(steps):
-            if i>=at: s[i]=after
-        return s
-    return [float(params.get("sigma",0.05))]
-
-
-# ------------------ Integrity helpers -----------------------
-def sha512_hex(s:str)->str:
-    return hashlib.sha512(s.encode()).hexdigest()
-
-def make_run_id(seed:int,profile:str)->str:
-    raw=f"{time.time_ns()}::{seed}::{profile}::{random.random()}"
-    return hashlib.sha256(raw.encode()).hexdigest()[:16]
-
-def write_bundle(run_dir:str,config:Dict[str,Any],results:Dict[str,Any])->Tuple[str,str]:
-    os.makedirs(run_dir,exist_ok=True)
-    with open(os.path.join(run_dir,"run.json"),"w") as f:
-        json.dump({"config":config,"results":results},f,indent=2)
-    canonical=json.dumps({"config":config,"results":results},sort_keys=True)
-    digest=sha512_hex(canonical)
-    with open(os.path.join(run_dir,"run.sha512"),"w") as f:
-        f.write(digest+"\n")
-    with open(os.path.join(run_dir,"ABOUT.json"),"w") as f:
-        json.dump(ABOUT_BLOCK,f,indent=2)
-    with open(os.path.join(run_dir,"NOTICE.txt"),"w") as f:
-        f.write(LEGAL_NOTICE+"\n")
-    zip_path=os.path.join(run_dir,"rft_run_bundle.zip")
-    with zipfile.ZipFile(zip_path,"w",zipfile.ZIP_DEFLATED) as z:
-        for fn in ["run.json","run.sha512","ABOUT.json","NOTICE.txt"]:
-            z.write(os.path.join(run_dir,fn),arcname=fn)
-    return zip_path,digest
-
-
-# ------------------ Core runner -----------------------------
-def run_total_proof(profile,noise_dist,schedule_type,schedule_params_text,seed,samples):
-    ok,msg=_rate_limit_ok()
-    if not ok:
-        return {"error":msg,"rft_notice":LEGAL_NOTICE},None
-    try:
-        params=json.loads(schedule_params_text) if schedule_params_text.strip() else {}
-        if not isinstance(params,dict): raise ValueError
-    except Exception as e:
-        return {"error":f"Invalid schedule_params JSON: {e}","rft_notice":LEGAL_NOTICE},None
-
-    sigma_series=build_sigma_series(schedule_type,params)
-    rng_master=_rng(seed)
-    results_steps=[]; trigger_count=0
-    for i,sigma in enumerate(sigma_series):
-        qs, zs, statuses=[],[],[]
-        sub_seed=int(rng_master.randint(0,2**31-1))
-        step_rng=_rng(sub_seed)
-        for _ in range(max(1,samples)):
-            o=simulate_step(step_rng,profile,float(sigma),noise_dist)
-            qs.append(o["QΩ"]); zs.append(o["ζ_sync"]); statuses.append(o["status"])
-        q_mean=float(np.mean(qs)); z_mean=float(np.mean(zs))
-        counts={"nominal":0,"perturbed":0,"critical":0}
-        for s in statuses: counts[s]+=1
-        majority=max(counts.items(),key=lambda kv:(kv[1],["nominal","perturbed","critical"].index(kv[0])))[0]
-        if majority!="nominal": trigger_count+=1
-        results_steps.append({"index":i,"sigma":round(float(sigma),6),
-                              "QΩ_mean":round(q_mean,6),"ζ_sync_mean":round(z_mean,6),
-                              "status_majority":majority,"samples":int(samples)})
-    statuses=[r["status_majority"] for r in results_steps]
-    summary={"steps":len(results_steps),
-             "nominal_count":statuses.count("nominal"),
-             "perturbed_count":statuses.count("perturbed"),
-             "critical_count":statuses.count("critical"),
-             "triggers_non_nominal":trigger_count}
-    run_id=make_run_id(seed,profile)
-    config={"run_id":run_id,"timestamp_utc":datetime.utcnow().isoformat()+"Z",
-            "profile":profile,"noise_distribution":noise_dist,
-            "schedule_type":schedule_type,"schedule_params":params,
-            "seed":int(seed),"samples_per_step":int(samples),
-            "about_version":RFT_VERSION,"about_doi":RFT_DOI}
-    results={"series":results_steps,"summary":summary,"rft_notice":LEGAL_NOTICE}
-    run_dir=f"/tmp/rft_run_{run_id}"
-    zip_path,digest=write_bundle(run_dir,config,results)
-    head={"run_id":run_id,"sha512":digest,"sha512_short":digest[:16]+"…",
-          "steps":summary["steps"],
-          "counts":{"nominal":summary["nominal_count"],
-                    "perturbed":summary["perturbed_count"],
-                    "critical":summary["critical_count"]},
-          "rft_notice":LEGAL_NOTICE}
-    return {"config":config,"head":head,"results":results},zip_path
-
+    return {"σ": round(sigma,6),"QΩ":q,"ζ_sync":z,"status":status}
+
+# ------------------ Main Runner -----------------------------
+def run(profile, dist, sigma, seed, samples):
+    rng = _rng(int(seed))
+    results = []
+    for _ in range(samples):
+        results.append(simulate_step(rng, profile, sigma, dist))
+    q_mean = np.mean([r["QΩ"] for r in results])
+    z_mean = np.mean([r["ζ_sync"] for r in results])
+    summary = {
+        "QΩ_mean": round(float(q_mean),6),
+        "ζ_sync_mean": round(float(z_mean),6),
+        "samples": samples,
+        "profile": profile,
+        "noise": sigma,
+        "dist": dist,
+        "status_majority": max(
+            ["nominal","perturbed","critical"],
+            key=lambda s: sum(1 for r in results if r["status"]==s)
+        ),
+        "rft_notice": LEGAL_NOTICE
+    }
+    return summary
 
 # ------------------ Gradio UI -------------------------------
-DEFAULT_PARAMS=json.dumps({"sigma":0.05},indent=2)
-HELP_TEXT=("Schedule JSON examples:\n"
-"- single: {\"sigma\": 0.05}\n"
-"- ramp: {\"start\": 0.0, \"stop\": 0.3, \"steps\": 12}\n"
-"- random: {\"min\": 0.01, \"max\": 0.3, \"steps\": 10, \"seed\": 0}\n"
-"- impulse: {\"base\":0.05,\"spike\":0.25,\"at\":5,\"steps\":12}\n"
-"- step: {\"before\":0.05,\"after\":0.2,\"at\":6,\"steps\":12}\n")
-
-with gr.Blocks(title="RFT-Ω Total-Proof API") as demo:
-    gr.Markdown(f"### RFT-Ω Total-Proof API ({RFT_VERSION})  \n"
-                f"Stable display mode — deterministic, signed validation harness.  \n"
-                f"**DOI:** {RFT_DOI}  \n**Legal:** {LEGAL_NOTICE}")
+with gr.Blocks(title="RFT-Ω Total-Proof Kernel") as demo:
+    gr.Markdown(f"### RFT-Ω Total-Proof Kernel ({RFT_VERSION})  \n"
+                f"DOI: [{RFT_DOI}]({RFT_DOI})  \n{LEGAL_NOTICE}")
+    
     with gr.Row():
-        profile=gr.Radio(list(PROFILES.keys()),label="System Profile",value="AI / Neural")
-        noise_dist=gr.Radio(["gauss","uniform"],label="Noise Distribution",value="gauss")
+        profile = gr.Dropdown(list(PROFILES.keys()), label="System Profile", value="AI / Neural")
+        dist = gr.Radio(["gauss","uniform"], label="Noise Distribution", value="gauss")
     with gr.Row():
-        schedule_type=gr.Radio(["single","ramp","random","impulse","step"],label="Schedule Type",value="single")
-        seed_in=gr.Number(value=123,precision=0,label="Seed (int)")
-        samples_in=gr.Slider(1,20,value=5,step=1,label="Samples per Step")
-    schedule_params=gr.Code(value=DEFAULT_PARAMS,language="json",label="Schedule Parameters (JSON)")
-    gr.Markdown(HELP_TEXT)
-    run_btn=gr.Button("Run Deterministic Simulation & Sign Results")
-    out_json=gr.JSON(label="Signed Run Summary")
-    out_bundle=gr.File(label="Download Signed Bundle (zip)")
-    def _on_run(p,nd,st,sp_json,seed,samples):
-        return run_total_proof(p,nd,st,sp_json,int(seed),int(samples))
-    run_btn.click(_on_run,
-                  inputs=[profile,noise_dist,schedule_type,schedule_params,seed_in,samples_in],
-                  outputs=[out_json,out_bundle])
-    gr.Markdown("Ops endpoints: `/healthz`, `/about` (FastAPI mounted if available).")
-
-
-# ------------------ FastAPI Mount ---------------------------
-if HAVE_FASTAPI:
-    api=FastAPI(title="RFT-Ω Total-Proof Ops")
-    @api.get("/healthz")
-    def healthz(): return {"ok":True,"service":"RFT-Ω Total-Proof","version":RFT_VERSION}
-    @api.get("/about")
-    def about(): return ABOUT_BLOCK
-    app=gr.mount_gradio_app(api,demo,path="/")
-else:
-    app=demo
-
-
-# ------------------ Launch (stable queue) -------------------
-if __name__=="__main__":
-    demo.queue().launch(
-        server_name="0.0.0.0",
-        server_port=7860,
-        show_error=True,
-        debug=False
-    )
+        sigma = gr.Slider(0.0, 0.3, value=0.05, step=0.01, label="Noise Scale (σ)")
+        seed = gr.Number(value=123, precision=0, label="Seed")
+        samples = gr.Slider(1, 20, value=5, step=1, label="Samples")
+    
+    run_btn = gr.Button("Run Simulation")
+    output = gr.JSON(label="Run Summary")
+
+    run_btn.click(run, inputs=[profile, dist, sigma, seed, samples], outputs=[output])
+
+# ------------------ Launch -------------------------------
+if __name__ == "__main__":
+    demo.launch(server_name="0.0.0.0", server_port=7860, show_error=True, debug=False)