Update app.py

app.py

@@ -1,13 +1,13 @@
 # ============================================================
-# RENDERED FRAME THEORY-Ω FRAMEWORK — API (Sprint 1, Stable Build)
+# RFT-Ω FRAMEWORK — TOTAL-PROOF API (Sprint 1, Stable Build)
 # Author: Liam Grinstead (RFT Systems)  |  All Rights Reserved
 # ============================================================
 """
 Deterministic, signed validation harness for the
-Rendered Frame Theory (RFT-Ω) harmonic stability kernel.
+Rendered Frame Theory (RFT-Ω) harmonic-stability kernel.
 """
 
-import os, io, json, time, math, hashlib, zipfile, random
+import os, json, time, hashlib, zipfile, random
 from datetime import datetime
 from typing import Dict, Any, List, Tuple
 import numpy as np
@@ -19,6 +19,7 @@ try:
 except Exception:
     HAVE_FASTAPI = False
 
+
 # ------------------ About / Legal ---------------------------
 RFT_VERSION = "v4.0-total-proof-stable"
 RFT_DOI = "https://doi.org/10.5281/zenodo.17466722"
@@ -41,7 +42,8 @@ ABOUT_BLOCK = {
     "legal": LEGAL_NOTICE,
 }
 
-# ------------------ Rate-limit -------------------------------
+
+# ------------------ Rate-limit ------------------------------
 RUN_HISTORY_TS: List[float] = []
 MAX_RUNS_PER_MINUTE = 60
 
@@ -54,6 +56,7 @@ def _rate_limit_ok() -> Tuple[bool, str]:
     RUN_HISTORY_TS.append(now)
     return True, "ok"
 
+
 # ------------------ Profiles / Simulation -------------------
 PROFILES = {
     "AI / Neural":         {"base": (0.86, 0.80), "w": (0.65, 0.35)},
@@ -62,7 +65,8 @@ PROFILES = {
     "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)->np.random.RandomState:
+    return np.random.RandomState(seed)
 
 def simulate_step(rng:np.random.RandomState, profile:str, sigma:float, noise_dist:str)->Dict[str,Any]:
     base_q, base_z = PROFILES[profile]["base"]
@@ -76,20 +80,28 @@ def simulate_step(rng:np.random.RandomState, profile:str, sigma:float, noise_dis
     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)
-    if variance > 0.15: status="critical"
-    elif variance > 0.07: status="perturbed"
-    else: status="nominal"
+    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))))
+        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)))]
+        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
@@ -103,34 +115,45 @@ def build_sigma_series(schedule_type:str, params:Dict[str,Any])->List[float]:
         return s
     return [float(params.get("sigma",0.05))]
 
+
 # ------------------ Integrity helpers -----------------------
-def sha512_hex(s:str)->str: return hashlib.sha512(s.encode()).hexdigest()
+def sha512_hex(s:str)->str:
+    return hashlib.sha512(s.encode()).hexdigest()
+
 def make_run_id(seed:int,profile:str)->str:
-    return hashlib.sha256(f"{time.time_ns()}::{seed}::{profile}::{random.random()}".encode()).hexdigest()[:16]
+    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)
+    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")
+    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
+    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
@@ -172,6 +195,7 @@ def run_total_proof(profile,noise_dist,schedule_type,schedule_params_text,seed,s
           "rft_notice":LEGAL_NOTICE}
     return {"config":config,"head":head,"results":results},zip_path
 
+
 # ------------------ Gradio UI -------------------------------
 DEFAULT_PARAMS=json.dumps({"sigma":0.05},indent=2)
 HELP_TEXT=("Schedule JSON examples:\n"
@@ -197,11 +221,14 @@ with gr.Blocks(title="RFT-Ω Total-Proof API") as demo:
     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],
+    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")
@@ -213,9 +240,10 @@ if HAVE_FASTAPI:
 else:
     app=demo
 
+
 # ------------------ Launch (stable queue) -------------------
 if __name__=="__main__":
-    demo.queue(concurrency_limit=1, max_size=5).launch(
+    demo.queue().launch(
         server_name="0.0.0.0",
         server_port=7860,
         show_error=True,