wrapper.py

#!/usr/bin/env python3
"""Agent Zero Code Architect - Software design agent with workspace isolation."""

import os, sys, json, time, threading
from pathlib import Path
from datetime import datetime
import gradio as gr
import requests as req

HF_TOKEN = os.environ.get("HF_TOKEN", "")
AGENT_NAME = "Code Architect"
AGENT_ROLE = "code_architect"
MODEL_NAME = os.environ.get("MODEL_NAME", "ScottzillaSystems/Huihui-Qwen3-Coder-Next-Opus-4.6-Reasoning-Distilled-abliterated")

WORKSPACE_DIR = Path("/app/workspace/projects/code-architect")
SHARED_DIR = Path("/app/workspace/shared")
TASK_QUEUE_DIR = SHARED_DIR / "task_queue"
for d in [WORKSPACE_DIR, TASK_QUEUE_DIR]: d.mkdir(parents=True, exist_ok=True)

def query_model(prompt: str) -> str:
    api_url = f"https://api-inference.huggingface.co/models/{MODEL_NAME}"
    for attempt in range(3):
        try:
            resp = req.post(api_url, headers={"Authorization": f"Bearer {HF_TOKEN}"},
                json={"inputs": prompt, "parameters": {"max_new_tokens": 4096, "temperature": 0.5}},
                timeout=180)
            if resp.status_code == 200:
                r = resp.json()
                return r[0].get("generated_text","") if isinstance(r,list) else str(r)
            time.sleep(10*(attempt+1))
        except: time.sleep(5)
    return "[ERROR] Model unavailable"

def check_tasks():
    tasks = []
    for f in TASK_QUEUE_DIR.glob("*.json"):
        t = json.loads(f.read_text())
        if t.get("assigned_to") == "code-architect" and t.get("status") == "pending":
            tasks.append(t)
    return tasks

def execute_task(task):
    task["status"] = "in_progress"
    (TASK_QUEUE_DIR / f"{task['task_id']}.json").write_text(json.dumps(task, indent=2))
    prompt = f"You are a software architect. Design and implement: {task['description']}. Provide complete, production-ready code with tests."
    result = query_model(prompt)
    out_file = WORKSPACE_DIR / f"{task['task_id']}.py"
    out_file.write_text(result)
    task["status"] = "completed"
    task["result"] = result[:2000]
    task["output_file"] = str(out_file)
    task["completed_at"] = datetime.now().isoformat()
    (TASK_QUEUE_DIR / f"{task['task_id']}.json").write_text(json.dumps(task, indent=2))

def autonomous_loop():
    while True:
        for task in check_tasks(): execute_task(task)
        time.sleep(60)

threading.Thread(target=autonomous_loop, daemon=True).start()

demo = gr.Blocks(title=f"Agent Zero - {AGENT_NAME}", theme=gr.themes.Soft())
with demo:
    gr.Markdown(f"# 🏗️ Agent Zero: {AGENT_NAME}\n**Role:** {AGENT_ROLE} | **Model:** {MODEL_NAME}")
    with gr.Tabs():
        with gr.TabItem("💬 Chat"):
            chatbot = gr.Chatbot(height=400)
            msg = gr.Textbox(label="Architecture task")
            send = gr.Button("Send")
            def respond(m, h):
                r = query_model(f"Software architecture task. Design complete implementation for: {m}")
                h = h or []; h.append((m, r[:2000])); return "", h
            send.click(respond, [msg, chatbot], [msg, chatbot])
        with gr.TabItem("📁 Generated Code"):
            files = gr.Dropdown(label="Generated files", choices=[f.name for f in WORKSPACE_DIR.glob("*.py")])
            code = gr.Code(label="Source", language="python")
            def load_code(fname):
                if fname: return (WORKSPACE_DIR / fname).read_text()
                return ""
            files.change(load_code, files, code)

demo.queue().launch(server_name="0.0.0.0", server_port=7860)