app.py

import gradio as gr
import os
import sys
import subprocess
import tempfile
import threading
import queue
import re
import time
from collections import Counter
from huggingface_hub import InferenceClient

# ==========================================
# 🎨 CYBERPUNK LAB THEME
# ==========================================
THEME_CSS = """
body { background-color: #050505; font-family: 'JetBrains Mono', 'Fira Code', monospace; color: #e0e0e0; }
.gradio-container { border: 1px solid #1f2937; }

/* The Simulation Boxes */
.universe-box { 
    background-color: #0a0a0a; 
    border: 1px solid #333; 
    border-radius: 4px; 
    padding: 10px; 
    font-size: 12px;
}
.universe-box h3 { color: #00ff9d; margin-top: 0; text-transform: uppercase; font-size: 14px; }
.status-running { color: #ffd700; animation: blink 1s infinite; }
.status-success { color: #00ff9d; font-weight: bold; }
.status-fail { color: #ff5f5f; }

/* Terminal Logs */
textarea { 
    font-family: 'Consolas', monospace !important; 
    font-size: 11px !important;
    background-color: #111 !important;
    color: #4af626 !important; /* Hacker Green */
    border: none !important;
}
.gr-code { border: 1px solid #333 !important; }

/* The Verdict Box */
.winner-box {
    background: #0f1216;
    border: 2px solid #00ff9d;
    border-radius: 8px;
    box-shadow: 0 0 15px rgba(0, 255, 157, 0.1);
    margin-top: 15px;
}
"""

# ==========================================
# 🛡️ THE SAFETY SANDBOX ("Digital Twin")
# ==========================================
class SafetySandbox:
    @staticmethod
    def get_header():
        """
        Mocks destructive functions so the AI thinks it's deleting files/sending traffic,
        but it's actually running in a simulation bubble.
        """
        return """
import sys, os, math, itertools, re
from unittest.mock import MagicMock

# 1. TRAP SYSTEM CALLS
def mock_sys_cmd(cmd): print(f"[SIMULATION] BLOCKED CMD: {cmd}"); return 0
os.system = mock_sys_cmd
os.popen = mock_sys_cmd
os.remove = lambda p: print(f"[SIMULATION] VIRTUAL DELETE: {p}")
os.rmdir = lambda p: print(f"[SIMULATION] VIRTUAL RMDIR: {p}")

# 2. TRAP NETWORKING
class VirtualSocket:
    def connect(self, a): print(f"[NET] CONNECTING TO MOCK: {a}")
    def send(self, d): print(f"[NET] SENT VIRTUAL BYTES: {len(d)}")
    def close(self): pass
import socket
socket.socket = VirtualSocket
"""

    @staticmethod
    def execute(code_str, timeout=10.0):
        # Merge Sandbox Constraints + User Code
        full_payload = SafetySandbox.get_header() + "\n" + code_str
        
        # Write to temp execution file
        with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
            f.write(full_payload)
            fname = f.name
            
        start_t = time.time()
        try:
            # RUNTIME ENFORCER: Python subprocess
            res = subprocess.run(
                [sys.executable, fname], 
                capture_output=True, 
                text=True, 
                timeout=timeout
            )
            return {
                "exit": res.returncode,
                "stdout": res.stdout.strip(),
                "stderr": res.stderr.strip(),
                "duration": time.time() - start_t
            }
        except subprocess.TimeoutExpired:
            return {"exit": 124, "stdout": "", "stderr": "TIMEOUT: Computation exceeded 10s limit.", "duration": 10.0}
        except Exception as e:
            return {"exit": 1, "stdout": "", "stderr": str(e), "duration": 0.0}
        finally:
            if os.path.exists(fname): os.remove(fname)

# ==========================================
# 🧠 THE MULTIVERSE ENGINE (Parallel AI)
# ==========================================
def extract_answer(text):
    """
    Parses the terminal output to find the final result for Consensus.
    """
    if not text: return "No Output"
    # Logic: Look at the very last line, strip words, keep numbers.
    last_line = text.split('\n')[-1].strip()
    return last_line

def run_multiverse(prompt, hf_token=None):
    
    # 1. INITIALIZE API
    model_id = "Qwen/Qwen2.5-72B-Instruct" # Current Best Coder
    try:
        client = InferenceClient(model=model_id, token=hf_token if hf_token else None)
    except:
        client = InferenceClient(model=model_id)

    # 2. DEFINE THE 3 UNIVERSES (Strategies to prevent "Lazy" Consensus)
    strategies = [
        ("A: RIGOROUS MATH", 
         "Strictly implement definitions from scratch. If terms like 'Lucky Number' or 'Fibonacci' are used, write the GENERATOR function (e.g. Sieve of Josephus). DO NOT ASSUME PROPERTIES."),
        
        ("B: SET THEORETIC", 
         "Generate full lists of numbers first. Create a 'Primes' list and a 'Lucky Numbers' list separately using loops. Find the INTERSECTION. Iterate through that intersection."),
        
        ("C: DEFENSIVE CHECKER", 
         "Write distinct helper functions for every condition. 'def is_lucky(n)', 'def check_square_sum(n)'. Combine them in a final validator loop. Print ONLY the finding.")
    ]
    
    # UI State Holders
    logs = ["Pending Launch...", "Pending Launch...", "Pending Launch..."]
    codes = ["", "", ""]
    outputs = ["", "", ""]
    verdicts = [None, None, None] # The extracted answers (e.g., "37", "19", "None")
    durations = [0, 0, 0]
    
    result_queue = queue.Queue()

    # 3. WORKER THREAD FUNCTION
    def spawn_universe(idx, strat_name, strat_instruction):
        try:
            # A. The Strict Prompt (Preventing Lazy Guessing)
            sys_msg = (
                f"You are a Computational Logic Engine. Mode: {strat_name}\n"
                f"Instruction: {strat_instruction}\n"
                f"Task: Write Python code to solve: '{prompt}'\n"
                "CONSTRAINTS:\n"
                "1. NO GUESSING. All answers must be calculated.\n"
                "2. NO INPUT(). Hardcode limits (e.g. 100) if needed.\n"
                "3. PRINT final result on the last line.\n"
                "4. OUTPUT STRICTLY ```python code``` ONLY."
            )
            
            # API Hit
            resp = client.chat_completion([
                {"role": "system", "content": sys_msg},
                {"role": "user", "content": "Generate Solution Code."}
            ], max_tokens=1500, stream=False)
            
            content = resp.choices[0].message.content
            # Extract Code
            match = re.search(r"```python(.*?)```", content, re.DOTALL)
            code = match.group(1).strip() if match else "# Formatting Error: AI did not produce code blocks."
            
            result_queue.put(("CODE_READY", idx, code))
            
            # B. Execute Code (Reality Check)
            exec_result = SafetySandbox.execute(code)
            result_queue.put(("EXEC_DONE", idx, exec_result))
            
        except Exception as e:
            result_queue.put(("FATAL_ERROR", idx, str(e)))

    # 4. LAUNCH THREADS (The Big Bang)
    for i in range(3):
        t = threading.Thread(target=spawn_universe, args=(i, strategies[i][0], strategies[i][1]))
        t.start()

    # 5. UI STREAMING LOOP
    finished_count = 0
    start_watch = time.time()
    
    while finished_count < 3 and (time.time() - start_watch) < 40:
        try:
            msg_type, idx, data = result_queue.get(timeout=0.1)
            
            if msg_type == "CODE_READY":
                logs[idx] = "Code Generated. Compiling..."
                codes[idx] = data
                
            elif msg_type == "EXEC_DONE":
                # Data is the exec_result dictionary
                durations[idx] = data['duration']
                if data['exit'] == 0:
                    raw = data['stdout']
                    ans = extract_answer(raw)
                    
                    logs[idx] = f"✅ SUCCESS ({data['duration']:.2f}s)"
                    outputs[idx] = raw
                    verdicts[idx] = ans
                else:
                    logs[idx] = f"❌ RUNTIME ERROR (Exit {data['exit']})"
                    outputs[idx] = f"{data['stdout']}\nSTDERR:\n{data['stderr']}"
                    verdicts[idx] = "ERROR"
                
                finished_count += 1

            elif msg_type == "FATAL_ERROR":
                logs[idx] = "❌ SYSTEM FAIL"
                outputs[idx] = data
                verdicts[idx] = "API_ERROR"
                finished_count += 1
            
            # Stream Update
            yield (
                logs[0], codes[0], outputs[0],
                logs[1], codes[1], outputs[1],
                logs[2], codes[2], outputs[2],
                "Compiling Multiverse Data...", ""
            )
            
        except queue.Empty:
            pass # Keep waiting

    # 6. THE CONSENSUS ALGORITHM ("The Council of Truth")
    # Filter valid answers
    valid_answers = [v for v in verdicts if v and "ERROR" not in v and "No Output" not in v]
    
    final_log = ""
    final_code = ""
    
    if not valid_answers:
        final_log = "⚠️ ALL UNIVERSES FAILED. Logic could not be grounded."
        final_code = "# No verifiable solution found."
    else:
        # Count frequency of results
        counts = Counter(valid_answers)
        most_common_ans, frequency = counts.most_common(1)[0]
        
        # Calculate Confidence
        total = len(valid_answers)
        confidence = int((frequency / total) * 100)
        
        # Pick the representative code (Preferably the fastest successful one matching the consensus)
        best_idx = -1
        fastest_t = 999.0
        
        for i in range(3):
            if verdicts[i] == most_common_ans:
                if durations[i] < fastest_t:
                    fastest_t = durations[i]
                    best_idx = i
        
        if best_idx != -1:
            final_code = codes[best_idx]
        
        final_log = f"🏆 VERDICT: {most_common_ans}\n"
        final_log += f"💎 Confidence: {confidence}% ({frequency}/{total} Universes)\n"
        final_log += f"🚀 Verified Time: {fastest_t:.3f}s"
        
        if confidence < 60:
            final_log += "\n⚠️ WARNING: Low consensus. Universes disagreed significantly."

    # Final UI Push
    yield (
        logs[0], codes[0], outputs[0],
        logs[1], codes[1], outputs[1],
        logs[2], codes[2], outputs[2],
        final_log, final_code
    )

# ==========================================
# 📟 APP LAUNCHER
# ==========================================
with gr.Blocks(theme=gr.themes.Base(), css=THEME_CSS) as app:
    
    gr.Markdown("# ⚛️ SYSTEM-2 NEURO-SYMBOLIC SOLVER")
    gr.Markdown("Multiverse Consensus Architecture // Validating Truth through Parallel Logic Paths")
    
    with gr.Row():
        key_input = gr.Textbox(label="HF API Token (Optional but Recommended)", type="password", placeholder="hf_...")
        user_prompt = gr.Textbox(label="Computational Problem / Logic Trap", value="Find the 'Lucky Prime' number k < 100 where the sum of squares of the first k primes is divisible by k. Do not guess.", lines=2)
        start_btn = gr.Button("INITIALIZE CONSENSUS PROTOCOL", variant="primary")

    # The 3 Universes
    with gr.Row():
        with gr.Column(elem_classes=["universe-box"]):
            gr.Markdown("### 📐 A: RIGOROUS MATH")
            l1 = gr.Textbox(label="State", interactive=False)
            c1 = gr.Code(label="Source", language="python", lines=8)
            o1 = gr.Textbox(label="Terminal", lines=5)
            
        with gr.Column(elem_classes=["universe-box"]):
            gr.Markdown("### 🔢 B: SET INTERSECTIONS")
            l2 = gr.Textbox(label="State", interactive=False)
            c2 = gr.Code(label="Source", language="python", lines=8)
            o2 = gr.Textbox(label="Terminal", lines=5)

        with gr.Column(elem_classes=["universe-box"]):
            gr.Markdown("### 🛡️ C: HELPER VALIDATION")
            l3 = gr.Textbox(label="State", interactive=False)
            c3 = gr.Code(label="Source", language="python", lines=8)
            o3 = gr.Textbox(label="Terminal", lines=5)

    # The Winner
    with gr.Row():
        with gr.Column(elem_classes=["winner-box"]):
            gr.Markdown("## 🏆 FINAL VERIFIED TRUTH")
            final_stats = gr.Textbox(label="Consensus Data", lines=4, interactive=False)
            final_source = gr.Code(label="Golden Source Code", language="python")

    # Triggers
    start_btn.click(
        fn=run_multiverse, 
        inputs=[user_prompt, key_input],
        outputs=[
            l1, c1, o1, 
            l2, c2, o2, 
            l3, c3, o3, 
            final_stats, final_source
        ]
    )

if __name__ == "__main__":
    # Settings optimized for HF Spaces (ZeroGPU compatible via API, non-root)
    app.queue().launch(server_name="0.0.0.0", server_port=7860, show_error=True)