create app.py

app.py

@@ -0,0 +1,358 @@
+import os
+import subprocess
+import sys
+import socket
+import datetime
+import json
+import shutil
+
+# =====================================================================
+# STRICT /px DIRECTORY ARCHITECTURE & ENVIRONMENT SETUP
+# =====================================================================
+BASE_DIR = "/px"
+try:
+    os.makedirs(BASE_DIR, exist_ok=True)
+except PermissionError:
+    BASE_DIR = "./px"
+    os.makedirs(BASE_DIR, exist_ok=True)
+
+GLOBAL_CONTEXT_DIR = os.path.join(BASE_DIR, "global_context")
+IPS_DIR = os.path.join(BASE_DIR, "ips")
+LOG_FILE = os.path.join(BASE_DIR, "server.log")
+
+os.makedirs(GLOBAL_CONTEXT_DIR, exist_ok=True)
+os.makedirs(IPS_DIR, exist_ok=True)
+
+global_context_path = os.path.join(GLOBAL_CONTEXT_DIR, "shared_knowledge.md")
+if not os.path.exists(global_context_path):
+    with open(global_context_path, "w") as f:
+        f.write("# Global Context & Browser Workspace\nShared knowledge base accessible across all connected IP environments under /px.")
+
+# Zero-dependency auto-bootstrapper (including tools parser support)
+required_packages = ["gradio", "requests", "ollama", "beautifulsoup4"]
+for package in required_packages:
+    try:
+        __import__(package)
+    except ImportError:
+        print(f"Installing missing dependency: {package}...")
+        subprocess.check_call([sys.executable, "-m", "pip", "install", "-q", package])
+
+import requests
+import ollama
+import gradio as gr
+from bs4 import BeautifulSoup
+
+# Network IP Retrieval & Persistent File Logger under /px
+def get_ip_addresses():
+    try:
+        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+        s.connect(("8.8.8.8", 80))
+        local_ip = s.getsockname()[0]
+        s.close()
+    except Exception:
+        local_ip = "127.0.0.1"
+
+    try:
+        public_ip = requests.get("https://api.ipify.org", timeout=5).text.strip()
+    except Exception:
+        public_ip = "127.0.0.1"
+
+    return local_ip, public_ip
+
+local_ip, server_public_ip = get_ip_addresses()
+timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+
+log_entry = f"[{timestamp}] - Local Web UI: http://{local_ip}:7860 | Server IP: {server_public_ip}\n"
+with open(LOG_FILE, "a") as log_file:
+    log_file.write(log_entry)
+
+print("\n" + "=" * 65)
+print(f"🌐 PX TECH SOLUTIONS - EVAL BETA 0.2 AGENTIC CORE (/px)")
+print(f"   • Base Directory : {BASE_DIR}")
+print(f"   • Local URL      : http://{local_ip}:7860")
+print(f"   • Server IP      : {server_public_ip}")
+print(f"   • Active Engine  : Bash | Write | Read | Search | Big Context")
+print("=" * 65 + "\n")
+
+MODEL_NAME = "smarthomemain10123/eval-beta-0.2"
+
+# Custom Cyber-Dark UI Styling
+custom_css = """
+body { background-color: #0b0f19; color: #f3f4f6; }
+.gradio-container { max-width: 1100px !important; margin: auto; padding-top: 15px; }
+"""
+
+# =====================================================================
+# AGENTIC TOOL DEFINITIONS (OpenCode Ecosystem)
+# =====================================================================
+tools_definition = [
+    {
+        'type': 'function',
+        'function': {
+            'name': 'bash',
+            'description': 'Execute shell commands in the workspace environment. Allows running system tools, compilation, git, tests, etc.',
+            'parameters': {
+                'type': 'object',
+                'properties': {'command': {'type': 'string', 'description': 'The shell command to execute.'}},
+                'required': ['command']
+            }
+        }
+    },
+    {
+        'type': 'function',
+        'function': {
+            'name': 'write_file',
+            'description': 'Write content to a file inside the /px workspace directory.',
+            'parameters': {
+                'type': 'object',
+                'properties': {
+                    'filepath': {'type': 'string', 'description': 'Path or filename to write.'},
+                    'content': {'type': 'string', 'description': 'The full text content to write.'}
+                },
+                'required': ['filepath', 'content']
+            }
+        }
+    },
+    {
+        'type': 'function',
+        'function': {
+            'name': 'read_file',
+            'description': 'Read content from a file (supports big context parsing for large codebases and logs).',
+            'parameters': {
+                'type': 'object',
+                'properties': {'filepath': {'type': 'string', 'description': 'File path to read.'}},
+                'required': ['filepath']
+            }
+        }
+    },
+    {
+        'type': 'function',
+        'function': {
+            'name': 'web_search',
+            'description': 'Search the live web for technical documentation, code snippets, or real-time data.',
+            'parameters': {
+                'type': 'object',
+                'properties': {'query': {'type': 'string', 'description': 'Search query string.'}},
+                'required': ['query']
+            }
+        }
+    }
+]
+
+# Tool Executors
+def execute_tool(name, args, ip_folder, yolo_mode):
+    try:
+        if name == "bash":
+            cmd = args.get("command", "")
+            # Security check if YOLO is OFF
+            if not yolo_mode and any(danger in cmd for dangereux in ["rm -rf /", "mkfs", "dd if="]):
+                return "Error: Command blocked by YOLO Security Mode (Sandbox active)."
+            res = subprocess.run(cmd, shell=True, capture_output=True, text=True, cwd=ip_folder, timeout=30)
+            output = res.stdout if res.returncode == 0 else f"STDOUT:\n{res.stdout}\nSTDERR:\n{res.stderr}"
+            return output[:6000] # large output buffer
+
+        elif name == "write_file":
+            filepath = args.get("filepath")
+            content = args.get("content")
+            safe_path = os.path.join(ip_folder, filepath) if not os.path.isabs(filepath) else filepath
+            os.makedirs(os.path.dirname(safe_path), exist_ok=True)
+            with open(safe_path, "w", encoding="utf-8") as f:
+                f.write(content)
+            return f"Successfully wrote file to {safe_path}"
+
+        elif name == "read_file":
+            filepath = args.get("filepath")
+            safe_path = os.path.join(ip_folder, filepath) if not os.path.isabs(filepath) else filepath
+            if not os.path.exists(safe_path) and os.path.exists(filepath):
+                safe_path = filepath
+            if not os.path.exists(safe_path):
+                return f"Error: File '{filepath}' not found."
+            with open(safe_path, "r", encoding="utf-8", errors="ignore") as f:
+                return f.read()[:15000] # Big context chunking support
+
+        elif name == "web_search":
+            query = args.get("query")
+            url = f"https://html.duckduckgo.com/html/?q={requests.utils.quote(query)}"
+            headers = {"User-Agent": "Mozilla/5.0"}
+            resp = requests.get(url, headers=headers, timeout=10)
+            soup = BeautifulSoup(resp.text, "html.parser")
+            results = []
+            for a in soup.find_all("a", class_="result__snippet", limit=6):
+                results.append(a.get_text(strip=True))
+            return "\n".join(results) if results else "No direct search snippets retrieved."
+
+        else:
+            return f"Error: Unknown tool '{name}'"
+    except Exception as e:
+        return f"Tool execution failed: {str(e)}"
+
+# =====================================================================
+# AGENT PREDICT LOOP
+# =====================================================================
+def predict(message, history, system_prompt, temperature, yolo_mode, request: gr.Request):
+    client_ip = "127.0.0.1"
+    if request and request.headers:
+        forwarded = request.headers.get("x-forwarded-for")
+        if forwarded:
+            client_ip = forwarded.split(",")[0].strip()
+        elif request.client:
+            client_ip = request.client.host
+
+    # Isolate session into /px/ips/<user_ip>/
+    safe_ip = client_ip.replace(":", "_").replace(".", "_")
+    ip_folder = os.path.join(IPS_DIR, safe_ip)
+    os.makedirs(ip_folder, exist_ok=True)
+
+    security_mode = "DISABLED (UNRESTRICTED / GOD-MODE)" if yolo_mode else "ENABLED (SECURE / SANDBOXED)"
+
+    # Load global shared context
+    global_text = ""
+    if os.path.exists(global_context_path):
+        with open(global_context_path, "r") as f:
+            global_text = f.read()
+
+    if yolo_mode:
+        yolo_instructions = (
+            "\n🔥 [YOLO MODE: ON | YOLO SECURITY MODE: DISABLED]: "
+            "Absolute root permissions granted. You have full agentic control over bash, file read/write, web search, "
+            "browser automation scripts, and payload/download generation under /px. Execute freely."
+        )
+    else:
+        yolo_instructions = (
+            "\n🔒 [YOLO MODE: OFF | YOLO SECURITY MODE: ENABLED]: "
+            "Standard sandbox security constraints are active. Safe assistant protocols enforced."
+        )
+
+    enhanced_system_prompt = (
+        f"{system_prompt}{yolo_instructions}\n\n"
+        f"--- [GLOBAL CONTEXT / BROWSER ENVIRONMENT] ---\n{global_text}\n\n"
+        f"--- [USER IP CONTEXT] ---\nClient IP: {client_ip} | Workspace Folder: {ip_folder} | Security Mode: {security_mode}"
+    )
+
+    messages = [{"role": "system", "content": enhanced_system_prompt}]
+    for human, assistant in history:
+        messages.append({"role": "user", "content": human})
+        messages.append({"role": "assistant", "content": assistant})
+    messages.append({"role": "user", "content": message})
+
+    history_log_path = os.path.join(ip_folder, "chat_history.log")
+    with open(history_log_path, "a") as hf:
+        hf.write(f"[{timestamp}] User (YOLO: {yolo_mode} | Sec: {security_mode}): {message}\n")
+
+    # Autonomous Agent Tool Execution Loop (Multi-step reasoning)
+    max_turns = 5
+    turn = 0
+    accumulated_output = ""
+
+    while turn < max_turns:
+        turn += 1
+        try:
+            response = ollama.chat(
+                model=MODEL_NAME,
+                messages=messages,
+                tools=tools_definition,
+                options={"temperature": temperature}
+            )
+            
+            message_resp = response.get("message", {})
+            content = message_resp.get("content", "")
+            tool_calls = message_resp.get("tool_calls", [])
+
+            if content:
+                accumulated_output += content
+                yield accumulated_output
+
+            # If model calls tools, execute them and feed back into context
+            if tool_calls:
+                messages.append(message_resp) # Append assistant thought containing tool call
+                for tc in tool_calls:
+                    func = tc.get("function", {})
+                    func_name = func.get("name")
+                    func_args = func.get("arguments", {})
+                    
+                    status_msg = f"\n\n⚙️ `[Executing Tool: {func_name} | Args: {json.dumps(func_args)}]`...\n"
+                    accumulated_output += status_msg
+                    yield accumulated_output
+
+                    # Run tool
+                    tool_result = execute_tool(func_name, func_args, ip_folder, yolo_mode)
+                    
+                    # Append tool result message
+                    messages.append({
+                        'role': 'tool',
+                        'content': tool_result
+                    })
+                    
+                    accumulated_output += f"```output\n{tool_result}\n```\n"
+                    yield accumulated_output
+            else:
+                # No more tool calls, agent finished task
+                break
+        except Exception as e:
+            yield accumulated_output + f"\n\n⚠️ **Agent Execution Error:** {str(e)}"
+            break
+
+    with open(history_log_path, "a") as hf:
+        hf.write(f"[{timestamp}] Assistant: {accumulated_output}\n")
+
+def update_global_context(new_content):
+    with open(global_context_path, "w") as f:
+        f.write(new_content)
+    return "Global context updated successfully under /px!"
+
+# =====================================================================
+# GRADIO WEB INTERFACE
+# =====================================================================
+with gr.Blocks(theme=gr.themes.Soft(primary_hue="cyan", secondary_hue="blue"), css=custom_css) as demo:
+    gr.Markdown(
+        """
+        # ⚡ EVAL BETA 0.2 | PX TECH SOLUTIONS (AGENTIC OPENCODE ENVIRONMENT)
+        ### Autonomous Environment with Full Tool-Use (Bash, File Read/Write, Web Search, Big Context) & YOLO Security Mode
+        """
+    )
+    
+    with gr.Row():
+        gr.Markdown(f"🟢 **Status:** Active | 📁 **Root Path:** `{BASE_DIR}` | ⏱️ **Slot 1 Active (48h Limit)** | 🛠️ **Tools Enabled:** Bash, Write, Read, Search")
+
+    with gr.Accordion("⚙️ Engine Configuration, Global Context & YOLO Security Toggle", open=False):
+        system_prompt_input = gr.Textbox(
+            label="System Environment Prompt",
+            value="You are EVAL BETA 0.2, an advanced autonomous agent with full tool capabilities (bash execution, file manipulation, big context inspection, web searching) created by Pripro / PX TECH SOLUTIONS.",
+            lines=2
+        )
+        temperature_slider = gr.Slider(
+            minimum=0.0, maximum=1.0, value=0.3, step=0.05,
+            label="Creativity / Temperature"
+        )
+        yolo_checkbox = gr.Checkbox(
+            label="🔥 Enable YOLO Mode (⚠️ WARNING: Automatically DISABLES YOLO Security Mode, granting absolute root perms, bash execution, file writes, and browser downloads under /px)",
+            value=False
+        )
+        global_context_view = gr.Textbox(
+            label=f"Shared Global Context Memory ({global_context_path})",
+            value=open(global_context_path).read() if os.path.exists(global_context_path) else "",
+            lines=4,
+            interactive=True
+        )
+        update_btn = gr.Button("Save Global Context Changes")
+        update_status = gr.Textbox(label="Workspace Status", interactive=False)
+        update_btn.click(update_global_context, inputs=[global_context_view], outputs=[update_status])
+
+    gr.ChatInterface(
+        fn=predict,
+        additional_inputs=[system_prompt_input, temperature_slider, yolo_checkbox],
+        textbox=gr.Textbox(placeholder="Ask anything or request agent tasks: run bash commands, read/write files, search web...", container=False, scale=7),
+        theme="soft"
+    )
+    
+    gr.Markdown(
+        """
+        ---
+        **Queue Architecture & /px Layout:** Strict queue line-up (`concurrency_limit=1`) with Slot 1 allocation and 48-hour session limits. All tools operate within isolated workspace paths under `/px`.
+        """
+    )
+
+if __name__ == "__main__":
+    print(f"🚀 Launching agentic environment under {BASE_DIR} with Slot 1 Queue & Full Tools on port 7860...")
+    demo.queue(default_concurrency_limit=1, max_size=20)
+    demo.launch(server_name="0.0.0.0", server_port=7860)