gradio 6.0

app.py

@@ -1,140 +1,166 @@
 import gradio as gr
 from huggingface_hub import InferenceClient
 from array import array
-from functools import lru_cache
-import os, re, time
+import os
+import re
+import time
 
-# 1. API Configuration
+# Securely retrieve the token from your Space's secrets
 HF_TOKEN = os.getenv("HF_TOKEN")
-MODEL_ID = "Qwen/Qwen2.5-7B-Instruct" 
-client = InferenceClient(MODEL_ID, token=HF_TOKEN)
 
-# 2. T3 High-Speed Logic Kernel
+# Initialize BOTH engines with the exact same base model
+client_primary = InferenceClient("Qwen/Qwen2.5-7B-Instruct", token=HF_TOKEN)
+client_competitor = InferenceClient("Qwen/Qwen2.5-7B-Instruct", token=HF_TOKEN)
+
 class StateController:
-    __slots__ = ("_state", "_rom60", "_symbols", "_rendered")
+    __slots__ = ("_state", "_metric", "_batch", "_reg", "_rendered", "_rom60", "_symbols")
+
     def __init__(self):
         self._state      = array("B", [0]) * 121
-        self._rom60      = tuple(tuple((i * j) % 60 for j in range(60)) for i in range(60))
-        self._symbols    = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz01234567"
-        self._rendered   = "".join(" [NODE_120] " if i == 120 else ("<" if i % 10 == 0 else ".") for i in range(121))
-
-    @lru_cache(maxsize=128)
-    def compute_distribution(self, total, nodes) -> str:
-        if nodes <= 0: return "Error: Node count must be positive."
-        base, rem = divmod(total, nodes)
-        res = f"T3 Logic Kernel resolved {total} units across {nodes} nodes:\n\n"
-        for i in range(nodes):
-            res += f"NODE_{i+1:02}: {base + (1 if i < rem else 0)} units\n"
-        return res
-
-    def get_glyphs(self) -> str:
-        return f"Rendering 121-point state array:\n\n{self._rendered}\n\nSystem State: RESOLVED"
+        self._metric     = 60
+        self._batch      = 10
+        self._reg        = {}
+        self._rendered   = self._build_render()
+        
+        self._rom60 = tuple(
+            tuple((i * j) % 60 for j in range(60)) 
+            for i in range(60)
+        )
+        self._symbols = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz01234567"
+
+    def _build_render(self) -> str:
+        return "".join(
+            " [NODE_120] " if i == 120 else ("<" if i % 10 == 0 else ".")
+            for i in range(121)
+        )
+
+    def diagnostic(self) -> str:
+        for i in range(51):
+            self._state[i] = i % self._batch
+        self._reg.clear()
+        self._reg["STATUS"] = "RESOLVED"
+        return (
+            "Diagnostic sequence initiated.\n\n"
+            "Grid initialized: 5 active blocks.\n\n"
+            "Rendering 121-point array:\n"
+            f"{self._rendered}\n\n"
+            "Executing state resolution:\n"
+            "System resolved. State array reset to zero."
+        )
+
+    def generate_receipt(self, a: int, b: int, c: int) -> str:
+        rom_val = self._rom60[a % 60][b % 60]
+        checksum_index = (rom_val ^ (c % 60)) % 60
+        return f"0{self._symbols[checksum_index]}"
+
+    def validate_receipt(self, receipt: str, a: int, b: int, c: int) -> str:
+        expected_receipt = self.generate_receipt(a, b, c)
+        if receipt == expected_receipt:
+            self._state[120] = 1
+            return f"[NODE_120: ACTIVE] Checksum Validated. Receipt {receipt} matches allocation ({a}, {b}, {c})."
+        else:
+            self._state[120] = 0
+            return f"[NODE_120: INACTIVE] Verification Failed. Expected receipt {expected_receipt}, received {receipt}."
 
 controller = StateController()
 
-def format_telemetry(seconds: float) -> str:
-    if seconds < 0.001: return f"{seconds * 1_000_000:.2f} \u03BCs"
-    return f"{seconds * 1_000:.2f} ms" if seconds < 1 else f"{seconds:.2f} s"
-
-# 3. Core Response Logic
-def generate_responses(user_message, p_hist, c_hist):
-    msg = user_message.strip()
-    if not msg: yield p_hist or [], c_hist or [], ""; return
-
-    p_hist, c_hist = p_hist or [], c_hist or []
-    p_hist.append({"role": "user", "content": msg})
-    p_hist.append({"role": "assistant", "content": ""})
-    c_hist.append({"role": "user", "content": msg})
-    c_hist.append({"role": "assistant", "content": ""})
-    yield p_hist, c_hist, ""
+PRIMARY_SYSTEM_MSG = {"role": "system", "content": "You are a logic-focused inference engine. Utilize strict state-hold memory and parallel integer blocks."}
+COMPETITOR_SYSTEM_MSG = {"role": "system", "content": "You are a standard helpful AI assistant."}
+
+def generate_responses(user_message: str, primary_history: list, competitor_history: list):
+    clean_message = user_message.strip()
+    if not clean_message:
+        yield primary_history, competitor_history, ""
+        return
+
+    # Update histories with new dictionary format for Gradio 6
+    primary_history.append({"role": "user", "content": clean_message})
+    primary_history.append({"role": "assistant", "content": ""})
+    competitor_history.append({"role": "user", "content": clean_message})
+    competitor_history.append({"role": "assistant", "content": ""})
+    yield primary_history, competitor_history, ""
     
     start_time = time.perf_counter()
 
-    # --- LOCAL INTERCEPTORS ---
-    dist_match = re.search(r"(?P<units>\d{1,9})\s+units\s+across\s+(?P<nodes>\d{1,4})\s+nodes", msg, re.I)
-    diag_match = any(kw in msg.lower() for kw in ["diagnostic", "grid"])
-
-    if dist_match or diag_match:
-        if dist_match:
-            res = controller.compute_distribution(int(dist_match.group("units")), int(dist_match.group("nodes")))
-        else:
-            res = controller.get_glyphs()
+    # Hardware diagnostic intercept
+    if clean_message.lower() == "run grid diagnostic":
+        output = controller.diagnostic()
+        elapsed_time = time.perf_counter() - start_time
+        primary_history[-1]["content"] = f"{output}\n\n---\n*Telemetry: {elapsed_time:.4f}s | Source: Local Engine*"
+        competitor_history[-1]["content"] = "Hardware diagnostics not supported by generic models."
+        yield primary_history, competitor_history, ""
+        return
+
+    # Deterministic Checksum Intercept
+    verify_match = re.search(r"verify receipt\s+([a-zA-Z0-9]{2})\s+for\s+(\d+),\s*(\d+),\s*(\d+)", clean_message, re.IGNORECASE)
+    if verify_match:
+        receipt, a, b, c = verify_match.group(1), int(verify_match.group(2)), int(verify_match.group(3)), int(verify_match.group(4))
+        output = controller.validate_receipt(receipt, a, b, c)
+        elapsed_time = time.perf_counter() - start_time
+        primary_history[-1]["content"] = f"{output}\n\n---\n*Telemetry: {elapsed_time:.6f}s | Source: Local ROM Math*"
+        competitor_history[-1]["content"] = "Deterministic verification not supported by standard LLMs."
+        yield primary_history, competitor_history, ""
+        return
+
+    # STREAM 1: Primary Engine
+    try:
+        msgs = [PRIMARY_SYSTEM_MSG] + primary_history[:-1]
+        primary_response = ""
+        stream = client_primary.chat_completion(messages=msgs, max_tokens=1024, stream=True, temperature=0.15)
+        for chunk in stream:
+            primary_response += (chunk.choices[0].delta.content or "")
+            primary_history[-1]["content"] = primary_response
+            yield primary_history, competitor_history, ""
         
-        elapsed = time.perf_counter() - start_time
-        p_hist[-1]["content"] = f"{res}\n\n---\n*Telemetry: {format_telemetry(elapsed)} | Source: LOCAL T3 KERNEL*"
-        yield p_hist, c_hist, ""
-    else:
-        try:
-            res_text = ""
-            stream = client.chat_completion(
-                messages=[{"role":"system","content":"Logic Engine"}] + p_hist[:-1], 
-                max_tokens=512, stream=True, temperature=0.1
-            )
-            for chunk in stream:
-                res_text += (chunk.choices[0].delta.content or "")
-                p_hist[-1]["content"] = res_text
-                yield p_hist, c_hist, ""
-            p_hist[-1]["content"] += f"\n\n---\n*Telemetry: {format_telemetry(time.perf_counter()-start_time)} | Source: AUGMENTED CLOUD*"
-            yield p_hist, c_hist, ""
-        except Exception as e:
-            p_hist[-1]["content"] = f"Primary Error: {str(e)}"
-            yield p_hist, c_hist, ""
-
-    comp_start = time.perf_counter()
-    c_hist[-1]["content"] = "*Routing...*"
-    yield p_hist, c_hist, ""
+        primary_time = time.perf_counter() - start_time
+        primary_history[-1]["content"] += f"\n\n---\n*Telemetry: {primary_time:.2f}s | Source: Augmented Kernel*"
+        yield primary_history, competitor_history, ""
+    except Exception as e:
+        primary_history[-1]["content"] = f"Error: {str(e)}"
+        yield primary_history, competitor_history, ""
+
+    # STREAM 2: Competitor Engine
+    competitor_start = time.perf_counter()
+    competitor_history[-1]["content"] = "*Connecting to vanilla infrastructure...*"
+    yield primary_history, competitor_history, ""
     
     try:
-        res_text = ""
-        stream = client.chat_completion(
-            messages=[{"role":"system","content":"Standard AI"}] + c_hist[:-1], 
-            max_tokens=512, stream=True, temperature=0.7
-        )
+        msgs = [COMPETITOR_SYSTEM_MSG] + competitor_history[:-1]
+        competitor_response = ""
+        stream = client_competitor.chat_completion(messages=msgs, max_tokens=1024, stream=True, temperature=0.7)
         for chunk in stream:
-            res_text += (chunk.choices[0].delta.content or "")
-            c_hist[-1]["content"] = res_text
-            yield p_hist, c_hist, ""
-        c_hist[-1]["content"] += f"\n\n---\n*Telemetry: {format_telemetry(time.perf_counter()-comp_start)} | Source: VANILLA CLOUD*"
-        yield p_hist, c_hist, ""
+            competitor_response += (chunk.choices[0].delta.content or "")
+            competitor_history[-1]["content"] = competitor_response
+            yield primary_history, competitor_history, ""
+        
+        competitor_time = time.perf_counter() - competitor_start
+        competitor_history[-1]["content"] += f"\n\n---\n*Telemetry: {competitor_time:.2f}s | Source: Vanilla Qwen*"
+        yield primary_history, competitor_history, ""
     except Exception as e:
-        c_hist[-1]["content"] = f"Competitor Error: {str(e)}"
-    yield p_hist, c_hist, ""
+        competitor_history[-1]["content"] = f"Error: {str(e)}"
+        yield primary_history, competitor_history, ""
 
-# 4. Interface Build (With True Scrollable CSS)
 custom_css = """
-body, .gradio-container { background-color: #110c08 !important; color: #fb923c !important; }
-footer { display: none !important; }
-.scrollable-examples { max-height: 140px; overflow-y: auto; }
+body, .gradio-container { background-color: #110c08 !important; }
+footer {display: none !important}
+.message-row { gap: 10px !important; }
 """
 
-example_prompts = [
-    ["Run grid diagnostic"],
-    ["Calculate the integer distribution for 50000 units across 12 nodes."],
-    ["Define P vs. NP. Then validate a 120-unit distribution across 3 nodes."],
-    ["Execute a Tier-3 Distribution Audit for 8593 units across 14 nodes."],
-    ["Distribute 1000000 units across 7 nodes."]
-]
-
 with gr.Blocks() as demo:
     gr.Markdown("# [ GLYPH.IO ]\n### Dual-Engine Hardware Benchmark")
-    p_chat = gr.Chatbot(label="Augmented Logic Kernel (T3 Architecture)", height=350)
+    
+    primary_chat = gr.Chatbot(label="Augmented Logic Kernel", height=320, type="messages")
     
     with gr.Row():
-        msg_in = gr.Textbox(label="Message", placeholder="Test P vs NP or Logistics Distribution...", scale=8)
+        msg_input = gr.Textbox(label="Message", placeholder="Enter logic task...", scale=8)
         submit_btn = gr.Button("Execute", scale=1, variant="primary")
+        
+    gr.Examples(examples=["Calculate the integer distribution for 120 units across 3 nodes.", "Run grid diagnostic"], inputs=msg_input)
     
-    # Removed pagination, added physical scroll via CSS class
-    gr.Examples(
-        examples=example_prompts, 
-        inputs=msg_in,
-        elem_classes="scrollable-examples",
-        label="Diagnostic Test Suite"
-    )
-    
-    c_chat = gr.Chatbot(label="Vanilla Qwen 2.5 (Standard Infrastructure)", height=350)
+    competitor_chat = gr.Chatbot(label="Vanilla Qwen 2.5", height=320, type="messages")
 
-    msg_in.submit(generate_responses, [msg_in, p_chat, c_chat], [p_chat, c_chat, msg_in])
-    submit_btn.click(generate_responses, [msg_in, p_chat, c_chat], [p_chat, c_chat, msg_in])
+    msg_input.submit(generate_responses, [msg_input, primary_chat, competitor_chat], [primary_chat, competitor_chat, msg_input])
+    submit_btn.click(generate_responses, [msg_input, primary_chat, competitor_chat], [primary_chat, competitor_chat, msg_input])
 
 if __name__ == "__main__":
     demo.queue().launch(theme=gr.themes.Soft(primary_hue="orange"), css=custom_css)