Update theme/app.py

theme/app.py

@@ -1,15 +1,17 @@
 import os
-import requests
+import uuid
+import json
+import time
 from pathlib import Path
 from fastapi import FastAPI, Request
 from fastapi.responses import JSONResponse, HTMLResponse
 from fastapi.templating import Jinja2Templates
 
 # =========================
-# APP INIT
+# CORE APP
 # =========================
 
-app = FastAPI(title="PyRunner Agent OS", version="3.2")
+app = FastAPI(title="PyRunner PAO v4 Distributed OS", version="4.0")
 
 BASE_DIR = Path(__file__).resolve().parent
 templates = Jinja2Templates(directory=str(BASE_DIR / "templates"))
@@ -21,199 +23,244 @@ templates = Jinja2Templates(directory=str(BASE_DIR / "templates"))
 OLLAMA_URL = os.getenv("OLLAMA_URL", "http://localhost:11434")
 
 MODELS = {
-    "orchestrator": os.getenv("OLLAMA_MODEL", "qwen2.5:1.5b"),
-    "coder": os.getenv("OLLAMA_CODE_MODEL", "qwen2.5-coder:1.5b"),
-    "fast": os.getenv("OLLAMA_FAST_MODEL", "deepseek-coder:1.3b"),
-    "fallback": os.getenv("OLLAMA_BACKUP_MODEL", "llama3.2:1b")
+    "planner": "qwen2.5:1.5b",
+    "worker": "qwen2.5-coder:1.5b",
+    "critic": "deepseek-coder:1.3b",
+    "synth": "llama3.2:1b"
 }
 
-TIMEOUT = 90
+# =========================
+# MEMORY GRAPH (NOW EVENT-SAFE)
+# =========================
+
+MEMORY_DB = {
+    "nodes": {},
+    "edges": []
+}
 
+def memory_store(task_id, data):
+    MEMORY_DB["nodes"][task_id] = data
+
+def memory_query(keyword):
+    results = []
+    for k, v in MEMORY_DB["nodes"].items():
+        if keyword.lower() in v.get("task", "").lower():
+            results.append(v)
+    return results
 
 # =========================
-# OLLAMA CORE CALL (SAFE)
+# EVENT BUS (SIMULATED QUEUE)
 # =========================
 
-def call_ollama(prompt: str, model: str):
-    try:
-        r = requests.post(
-            f"{OLLAMA_URL}/api/generate",
-            json={
-                "model": model,
-                "prompt": prompt,
-                "stream": False
-            },
-            timeout=TIMEOUT
-        )
+EVENT_QUEUE = []
 
-        if r.status_code != 200:
-            return {"error": f"Ollama HTTP {r.status_code}", "raw": r.text}
+def emit_event(event_type, payload):
+    EVENT_QUEUE.append({
+        "id": str(uuid.uuid4()),
+        "type": event_type,
+        "payload": payload,
+        "timestamp": time.time()
+    })
 
-        data = r.json()
+def consume_events():
+    events = EVENT_QUEUE[:]
+    EVENT_QUEUE.clear()
+    return events
 
-        return {
-            "response": data.get("response", "").strip(),
-            "raw": data
-        }
+# =========================
+# MODEL CALL
+# =========================
 
+def call_model(prompt, model):
+    import requests
+    try:
+        r = requests.post(
+            f"{OLLAMA_URL}/api/generate",
+            json={"model": model, "prompt": prompt, "stream": False},
+            timeout=120
+        )
+        return r.json().get("response", "")
     except Exception as e:
-        return {"error": str(e)}
-
+        return f"ERROR: {str(e)}"
 
 # =========================
-# CLEAN OUTPUT PARSER
+# ORCHESTRATOR (DAG BUILDER)
 # =========================
 
-def clean_category(text: str) -> str:
-    if not text:
-        return "general"
+def orchestrator(task):
 
-    text = text.lower().strip()
+    prompt = f"""
+Break into structured execution DAG:
 
-    # force valid categories only
-    valid = ["code", "vision", "fashion", "hairstyle", "general"]
+Task: {task}
 
-    for v in valid:
-        if v in text:
-            return v
+Return JSON array like:
+[
+  {{"step": "..." }},
+  {{"step": "..."}}
+]
+"""
 
-    return "general"
+    raw = call_model(prompt, MODELS["planner"])
 
+    try:
+        return json.loads(raw)
+    except:
+        # fallback
+        return [{"step": task}]
 
 # =========================
-# ORCHESTRATOR
+# WORKER (STATELESS EXECUTOR)
 # =========================
 
-def orchestrator(user_input: str):
+def worker(step, context=""):
 
     prompt = f"""
-Classify into ONE word only:
+Execute this step:
 
-code
-vision
-fashion
-hairstyle
-general
+{step}
 
-Input:
-{user_input}
+Context:
+{context}
 """
 
-    res = call_ollama(prompt, MODELS["orchestrator"])
+    return call_model(prompt, MODELS["worker"])
 
-    if "response" in res:
-        return clean_category(res["response"])
+# =========================
+# TOOL SYSTEM
+# =========================
 
-    return "general"
+def tool_calculator(expr):
+    try:
+        return eval(expr, {"__builtins__": {}})
+    except:
+        return "error"
 
+TOOLS = {
+    "calculator": tool_calculator
+}
+
+def tool_router(text):
+
+    if "calc:" in text:
+        expr = text.replace("calc:", "").strip()
+        return TOOLS["calculator"](expr)
+
+    return text
 
 # =========================
-# AGENTS
+# CRITIC (QUALITY GATE)
 # =========================
 
-def coder(prompt):
-    return call_ollama(prompt, MODELS["coder"])
+def critic(task, output):
+
+    prompt = f"""
+Score 0-10 and improve suggestion:
+
+Task: {task}
+Output: {output}
+"""
 
+    return call_model(prompt, MODELS["critic"])
 
-def fashion(prompt):
-    return call_ollama(f"Fashion analysis:\n{prompt}", MODELS["fast"])
+# =========================
+# SYNTHESIZER
+# =========================
 
+def synthesize(results):
 
-def hairstyle(prompt):
-    return call_ollama(f"Hairstyle analysis:\n{prompt}", MODELS["fast"])
+    combined = "\n".join(results)
 
+    prompt = f"""
+Create final clean response:
 
-def fallback(prompt):
-    return call_ollama(prompt, MODELS["fallback"])
+{combined}
+"""
 
+    return call_model(prompt, MODELS["synth"])
 
 # =========================
-# ROUTER
+# DISTRIBUTED EXECUTION ENGINE
 # =========================
 
-def route(category: str, prompt: str):
+def execute_dag(task, dag):
 
-    if category == "code":
-        return coder(prompt)
+    task_id = str(uuid.uuid4())
 
-    if category == "fashion":
-        return fashion(prompt)
+    emit_event("task_started", {"task_id": task_id, "task": task})
 
-    if category == "hairstyle":
-        return hairstyle(prompt)
+    results = []
 
-    return fallback(prompt)
+    for node in dag:
 
+        step = node.get("step")
 
-# =========================
-# UI
-# =========================
+        emit_event("step_started", {"step": step})
 
-@app.get("/", response_class=HTMLResponse)
-async def home(request: Request):
-    try:
-        return templates.TemplateResponse(
-            "index.html",
-            {"request": request}
-        )
-    except Exception as e:
-        return HTMLResponse(f"<h1>Template Error</h1><pre>{e}</pre>")
+        raw = worker(step)
 
+        processed = tool_router(raw)
 
-# =========================
-# HEALTH
-# =========================
+        crit = critic(task, processed)
+
+        results.append(processed)
+
+        emit_event("step_completed", {
+            "step": step,
+            "output": processed,
+            "critique": crit
+        })
+
+    final = synthesize(results)
+
+    memory_store(task_id, {
+        "task": task,
+        "result": final,
+        "steps": results
+    })
+
+    emit_event("task_completed", {"task_id": task_id})
 
-@app.get("/health")
-def health():
     return {
-        "status": "ok",
-        "models": MODELS
+        "task_id": task_id,
+        "dag": dag,
+        "events": consume_events(),
+        "final": final
     }
 
-
 # =========================
-# MAIN AGENT ENDPOINT
+# MAIN ORCHESTRATION PIPELINE
 # =========================
 
-@app.post("/api/agent")
-async def agent(request: Request):
+def run_pao(task):
 
-    try:
-        body = await request.json()
-        prompt = body.get("prompt", "")
+    dag = orchestrator(task)
 
-        if not prompt:
-            return JSONResponse(
-                {"error": "missing prompt"},
-                status_code=400
-            )
+    return execute_dag(task, dag)
 
-        category = orchestrator(prompt)
-        result = route(category, prompt)
+# =========================
+# API
+# =========================
 
-        if isinstance(result, dict) and "error" in result:
-            result = fallback(prompt)
+@app.get("/", response_class=HTMLResponse)
+async def home(request: Request):
+    return templates.TemplateResponse("index.html", {"request": request})
 
-        return {
-            "category": category,
-            "result": result.get("response", result)
-        }
+@app.post("/api/run")
+async def run(request: Request):
 
-    except Exception as e:
-        return JSONResponse(
-            {"error": str(e)},
-            status_code=500
-        )
+    body = await request.json()
+    task = body.get("task", "")
 
+    if not task:
+        return JSONResponse({"error": "missing task"}, status_code=400)
 
-# =========================
-# IMAGE PIPELINE (STABLE STUB)
-# =========================
+    return run_pao(task)
 
-@app.post("/api/image")
-async def image(request: Request):
-    return {
-        "status": "queued",
-        "pipeline": "vision_stub"
-    }
+@app.get("/api/memory/{task_id}")
+def get_memory(task_id: str):
+    return MEMORY_DB["nodes"].get(task_id, {})
+
+@app.get("/api/events")
+def get_events():
+    return {"events": EVENT_QUEUE}