Spaces:

Bc-AI
/

SACCP-head

Sleeping

App Files Files Community

Bc-AI commited on Apr 3

Commit

bca1298

verified ·

1 Parent(s): bd200f0

Update app.py

Browse files

Files changed (1) hide show

app.py +328 -391

app.py CHANGED Viewed

@@ -1,59 +1,53 @@
 """
-SharePUTER™ v2 Head Node — v2.0.3 FIXED
-Key fix: completion check happens IN submit_result, not background task.
 """
-from fastapi import FastAPI, HTTPException, BackgroundTasks, Request
 from fastapi.responses import HTMLResponse, JSONResponse
 from fastapi.middleware.cors import CORSMiddleware
 from pydantic import BaseModel
 from typing import Optional, Dict, Any
 import httpx, asyncio, uuid, os, json, time, ast, math
-from datetime import datetime, timedelta
-app = FastAPI(title="SharePUTER™ v2 Head Node")
 app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"])
-DB_URL = os.environ.get("SACCP_DB_URL", "https://saccpshareputer1.pythonanywhere.com/")
 DB_SECRET = os.environ.get("SACCP_DB_SECRET", "saccp-v2-master-key")
 HEAD_ID = os.environ.get("HEAD_NODE_ID", f"head-{uuid.uuid4().hex[:8]}")
 UNIT_PRICING = {
     "cpu": 0.5, "ram": 0.1,
     "gpu_t4": 4.0, "gpu_a10": 10.0, "gpu_a100": 25.0, "gpu_generic": 5.0,
 }
-NODE_PAY_RATES = {"RAM": 0.3, "CPU": 1.5, "GPU": 5.0}
 active_nodes: Dict[str, dict] = {}
-def db_headers():
-    return {"X-SACCP-Secret": DB_SECRET, "Content-Type": "application/json"}
-async def db_req(method: str, path: str, **kwargs):
     try:
-        async with httpx.AsyncClient(timeout=30.0) as c:
-            resp = await getattr(c, method)(f"{DB_URL}{path}", headers=db_headers(), **kwargs)
-            return resp
     except Exception as e:
-        print(f"[HEAD] DB ERROR {method} {path}: {e}")
         return None
-# ─── Models (Fixed Optional types for Pydantic v2) ─────────────────────────
 class UserCreate(BaseModel):
     username: str
     password: str
 class TaskSubmit(BaseModel):
     api_key: str
     code: str
     task_type: str = "general"
     config: dict = {}
     required_libs: list = []
-    ml_config: Optional[dict] = None
 class NodeRegister(BaseModel):
     node_type: str
@@ -61,97 +55,149 @@ class NodeRegister(BaseModel):
     owner: str = "anonymous"
     specs: dict = {}
     installed_libs: list = []
-    sandbox_enabled: bool = True
-# ─── Homepage ───────────────────────────────────────────────────────────────
 @app.get("/", response_class=HTMLResponse)
 async def homepage():
-    try:
-        sr = await db_req("get", "/stats")
-        s = sr.json() if sr and sr.status_code == 200 else {}
-    except:
-        s = {}
     return HTMLResponse(f"""<!DOCTYPE html>
-<html><head><title>SharePUTER™ v2</title>
-<style>body{{font-family:system-ui;background:#0a0a0f;color:#ddd;padding:40px;max-width:900px;margin:0 auto}}
-h1{{color:#00ff88}}.st{{display:inline-block;background:#111;padding:20px;margin:8px;border-radius:12px;
-min-width:120px;text-align:center}}.st .n{{font-size:2rem;color:#00ff88;font-weight:bold}}
-.st .l{{color:#888;font-size:.8rem}}</style></head><body>
-<h1>🍽️ SharePUTER™ v2.0.3</h1><p>Head Node — {HEAD_ID}</p>
-<div>
-<div class="st"><div class="n">{s.get('online_nodes',0)}</div><div class="l">Nodes</div></div>
-<div class="st"><div class="n">{s.get('total_users',0)}</div><div class="l">Users</div></div>
-<div class="st"><div class="n">{s.get('active_tasks',0)}</div><div class="l">Active</div></div>
-<div class="st"><div class="n">{s.get('completed_tasks',0)}</div><div class="l">Done</div></div>
-<div class="st"><div class="n">{s.get('pending_fragments',0)}</div><div class="l">Pending</div></div>
-<div class="st"><div class="n">{s.get('completed_fragments',0)}</div><div class="l">Frags Done</div></div>
-</div></body></html>""")
 # ─── Auth ───────────────────────────────────────────────────────────────────
 @app.post("/api/register")
-async def register(user: UserCreate):
-    r = await db_req("post", "/users", json={"username": user.username, "password": user.password})
-    if not r or r.status_code != 201:
-        raise HTTPException(r.status_code if r else 500, r.json().get("error") if r else "DB error")
     return r.json()
 @app.post("/api/login")
-async def login(user: UserCreate):
-    r = await db_req("post", "/users/auth", json={"username": user.username, "password": user.password})
-    if not r or r.status_code != 200:
-        raise HTTPException(401, "Invalid credentials")
     return r.json()
 @app.get("/api/balance")
 async def balance(api_key: str):
-    r = await db_req("post", "/users/by_api_key", json={"api_key": api_key})
-    if not r or r.status_code != 200:
-        raise HTTPException(401, "Invalid API key")
-    u = r.json()
-    return {"username": u["username"], "balance": u["balance"]}
 @app.get("/api/capabilities")
-async def capabilities():
-    r = await db_req("get", "/nodes/capabilities")
-    return r.json() if r and r.status_code == 200 else {}
 @app.get("/api/pricing")
-async def pricing():
-    return UNIT_PRICING
 @app.get("/api/available_libs")
-async def available_libs():
-    r = await db_req("get", "/libs/available")
-    return r.json() if r and r.status_code == 200 else {}
 # ─── Cost ───────────────────────────────────────────────────────────────────
-def calc_cost(config: dict) -> float:
-    cpus = config.get("cpus", 1)
-    ram = config.get("ram_gb", 4)
-    gpus = config.get("gpus", 0)
-    gt = config.get("gpu_type", "generic")
-    gk = f"gpu_{gt.lower()}" if f"gpu_{gt.lower()}" in UNIT_PRICING else "gpu_generic"
-    return round(cpus * UNIT_PRICING["cpu"] + ram * UNIT_PRICING["ram"] +
-                 (gpus * UNIT_PRICING.get(gk, 5) if gpus else 0), 4)
 # ─── Fragmentation ─────────────────────────────────────────────────────────
-def fragment_code(code: str, config: dict, task_id: str, required_libs: list) -> list:
     frags = []
-    cpus = max(1, config.get("cpus", 1))
-    needs_gpu = config.get("gpus", 0) > 0
-    ram = config.get("ram_gb", 4)
     target = max(2, min(cpus * 2, 8))
-    # Find splittable range loop
     loop = None
     try:
         tree = ast.parse(code)
@@ -161,412 +207,303 @@ def fragment_code(code: str, config: dict, task_id: str, required_libs: list) ->
                 if isinstance(fn, ast.Name) and fn.id == "range":
                     args = node.iter.args
                     try:
-                        if len(args) == 1:
-                            loop = {"var": node.target.id if isinstance(node.target, ast.Name) else "i",
-                                    "start": 0, "end": ast.literal_eval(args[0])}
-                        elif len(args) >= 2:
-                            loop = {"var": node.target.id if isinstance(node.target, ast.Name) else "i",
-                                    "start": ast.literal_eval(args[0]), "end": ast.literal_eval(args[1])}
                         if loop:
                             break
-                    except:
-                        loop = None
-    except:
-        pass
-    if loop and (loop["end"] - loop["start"]) >= 10:
-        total = loop["end"] - loop["start"]
         n = min(target, total)
         chunk = math.ceil(total / n)
-        # Build range patterns to search-replace
-        if loop["start"] == 0:
-            patterns = [f"range({loop['end']})", f"range(0, {loop['end']})",
-                        f"range(0,{loop['end']})"]
-        else:
-            patterns = [f"range({loop['start']}, {loop['end']})",
-                        f"range({loop['start']},{loop['end']})"]
         for idx in range(n):
-            cs = loop["start"] + idx * chunk
-            ce = min(loop["start"] + (idx + 1) * chunk, loop["end"])
-            if cs >= loop["end"]:
-                break
             mod = code
-            replacement = f"range({cs}, {ce})"
-            replaced = False
             for pat in patterns:
                 if pat in mod:
-                    mod = mod.replace(pat, replacement, 1)
-                    replaced = True
                     break
-            if not replaced:
-                mod = f"# SACCP chunk: {cs}-{ce}\n{code}"
             frags.append({
-                "fragment_id": f"{task_id}_frag_{idx}",
-                "task_id": task_id,
-                "fragment_index": idx,
-                "fragment_type": "compute",
-                "code": mod,
-                "input_data": json.dumps({"chunk": [cs, ce]}),
-                "required_libs": required_libs,
-                "required_gpu": needs_gpu,
-                "min_ram_gb": max(1, ram // n),
-                "timeout_seconds": 600,
             })
     if not frags:
         frags.append({
-            "fragment_id": f"{task_id}_frag_0",
-            "task_id": task_id,
-            "fragment_index": 0,
-            "fragment_type": "compute",
-            "code": code,
-            "input_data": json.dumps({"type": "full"}),
-            "required_libs": required_libs,
-            "required_gpu": needs_gpu,
-            "min_ram_gb": ram,
-            "timeout_seconds": 600,
         })
-    print(f"[HEAD] Fragmented task {task_id[:8]} into {len(frags)} pieces")
     return frags
-# ─── Task Submission ─────────────────────────────────────────────────────────
 @app.post("/api/submit_task")
-async def submit_task(task: TaskSubmit):
-    ur = await db_req("post", "/users/by_api_key", json={"api_key": task.api_key})
-    if not ur or ur.status_code != 200:
-        raise HTTPException(401, "Invalid API key")
     user = ur.json()
-    config = task.config
-    hourly = calc_cost(config)
-    if user["balance"] < hourly * 0.001:
-        raise HTTPException(402, "Insufficient balance")
     task_id = str(uuid.uuid4())
-    tr = await db_req("post", "/tasks", json={
         "task_id": task_id, "owner": user["username"], "code": task.code,
-        "task_type": task.task_type, "config": config, "required_libs": task.required_libs,
     })
-    if not tr or tr.status_code != 201:
-        raise HTTPException(500, "Failed to create task")
-    fragments = fragment_code(task.code, config, task_id, task.required_libs)
-    fr = await db_req("post", "/fragments/batch", json={"fragments": fragments})
-    if not fr or fr.status_code != 201:
-        raise HTTPException(500, "Failed to create fragments")
-    await db_req("patch", f"/tasks/{task_id}", json={
-        "status": "running", "total_fragments": len(fragments)
-    })
-    print(f"[HEAD] ✅ Task {task_id[:8]} submitted: {len(fragments)} fragments, {hourly}/hr")
-    return {
-        "task_id": task_id, "status": "running",
-        "total_fragments": len(fragments), "hourly_cost": hourly,
-    }
-# ─── THE KEY FIX: Assemble directly when results come in ────────────────────
-async def try_complete_task(task_id: str):
-    """Check if all fragments are done. If so, assemble and mark complete."""
-    fr = await db_req("get", f"/fragments/by_task/{task_id}")
-    if not fr or fr.status_code != 200:
-        return
     fragments = fr.json()
-    if not fragments:
-        return
-    total = len(fragments)
     completed = [f for f in fragments if f["status"] == "completed"]
     failed = [f for f in fragments if f["status"] == "failed"]
-    done_count = len(completed) + len(failed)
-    # Update progress
-    await db_req("patch", f"/tasks/{task_id}", json={
-        "completed_fragments": len(completed),
-        "failed_fragments": len(failed),
-    })
-    print(f"[HEAD] Task {task_id[:8]} progress: {done_count}/{total}")
-    # Not done yet
-    if done_count < total:
-        return
-    # ═══ ALL FRAGMENTS DONE — ASSEMBLE! ═══
-    print(f"[HEAD] Task {task_id[:8]} ALL DONE — assembling results!")
-    fragments.sort(key=lambda x: x["fragment_index"])
-    results = []
-    stdouts = []
-    errors = []
-    for f in fragments:
-        idx = f["fragment_index"]
-        if f["status"] == "completed":
-            if f.get("result") is not None:
-                results.append(f["result"])
-            if f.get("stdout"):
-                stdouts.append(f"[Fragment {idx}]\n{f['stdout']}")
-        elif f["status"] == "failed":
-            errors.append(f"[Fragment {idx}] {f.get('error', 'Unknown error')}")
     final = {
-        "fragments_total": total,
         "fragments_ok": len(completed),
         "fragments_failed": len(failed),
-        "results": results,
-        "stdout": "\n\n".join(stdouts),
         "errors": errors,
     }
-    status = "failed" if len(failed) == total else "completed"
-    # Calculate cost
-    task_resp = await db_req("get", f"/tasks/{task_id}")
-    elapsed_hours = 0.01
-    config = {}
-    owner = "unknown"
-    if task_resp and task_resp.status_code == 200:
-        task = task_resp.json()
-        owner = task.get("owner", "unknown")
-        config = task.get("config", {})
         try:
-            created_str = task["created_at"].split("+")[0].split("Z")[0]
-            created = datetime.fromisoformat(created_str)
-            elapsed_hours = max((datetime.utcnow() - created).total_seconds() / 3600, 0.001)
-        except:
-            elapsed_hours = 0.01
-    cost = round(calc_cost(config) * elapsed_hours, 6)
-    # Charge user
     if owner != "unknown":
-        await db_req("patch", f"/users/{owner}/balance", json={
-            "amount": -cost, "reason": f"task_{task_id}"
-        })
-    # Mark task complete
-    await db_req("patch", f"/tasks/{task_id}", json={
-        "status": status,
-        "result": json.dumps(final),
-        "cost": cost,
         "completed_at": datetime.utcnow().isoformat(),
         "error": "\n".join(errors) if errors else None,
     })
-    print(f"[HEAD] 🎉 Task {task_id[:8]} → {status} | Cost: {cost} SACCP | "
-          f"{len(results)} results | {len(errors)} errors")
-# ─── Submit Result (NO background task dependency!) ──────────────────────────
 @app.post("/api/submit_result")
 async def submit_result(request: Request):
-    """Worker submits fragment result. Uses raw request to avoid Pydantic issues."""
-    try:
-        data = await request.json()
-    except Exception as e:
-        print(f"[HEAD] submit_result: bad JSON: {e}")
-        raise HTTPException(400, "Invalid JSON")
-    fragment_id = data.get("fragment_id", "")
-    node_id = data.get("node_id", "")
-    status = data.get("status", "failed")
-    result_val = data.get("result")
-    error_val = data.get("error")
-    stdout_val = data.get("stdout", "")
-    resource_usage = data.get("resource_usage", {})
-    if not fragment_id:
-        raise HTTPException(400, "fragment_id required")
-    print(f"[HEAD] 📥 Result received: {fragment_id[:24]}... → {status}")
-    # Update fragment in database
-    update_resp = await db_req("patch", f"/fragments/{fragment_id}", json={
-        "status": status,
-        "result": result_val,
-        "error": error_val if error_val else None,
-        "stdout": stdout_val if stdout_val else "",
-        "completed_at": datetime.utcnow().isoformat(),
-        "resource_usage": resource_usage,
     })
-    if not update_resp or update_resp.status_code != 200:
-        print(f"[HEAD] ⚠️ Failed to update fragment {fragment_id[:16]} in DB!")
-        print(f"[HEAD] DB response: {update_resp.text if update_resp else 'None'}")
-    else:
-        print(f"[HEAD] ✅ Fragment {fragment_id[:16]} updated in DB")
-    # Pay node
-    if status == "completed" and node_id:
-        ninfo = active_nodes.get(node_id, {})
-        ntype = ninfo.get("node_type", "CPU")
-        pay = NODE_PAY_RATES.get(ntype, 0) / 360
-        if pay > 0:
-            await db_req("post", f"/nodes/{node_id}/pay", json={"amount": round(pay, 6)})
-    # ═══ KEY: Get task_id and check if task is complete ═══
-    frag_resp = await db_req("get", f"/fragments/{fragment_id}")
-    if frag_resp and frag_resp.status_code == 200:
-        frag_data = frag_resp.json()
-        task_id = frag_data.get("task_id", "")
-        if task_id:
-            print(f"[HEAD] Checking if task {task_id[:8]} is complete...")
-            await try_complete_task(task_id)
-    else:
-        print(f"[HEAD] ⚠️ Could not fetch fragment {fragment_id[:16]} to get task_id")
     return {"ok": True}
-# ─── Node Management ────────────────────────────────────────────────────────
 @app.post("/api/register_node")
-async def register_node(node: NodeRegister):
     nid = f"node-{uuid.uuid4().hex[:12]}"
-    await db_req("post", "/nodes", json={
         "node_id": nid, "node_type": node.node_type.upper(),
         "node_url": node.node_url, "owner": node.owner,
         "specs": node.specs, "installed_libs": node.installed_libs,
     })
     active_nodes[nid] = {"node_type": node.node_type.upper(), "last_hb": time.time()}
-    print(f"[HEAD] Node registered: {nid} ({node.node_type})")
     return {"node_id": nid, "pay_rate": f"{NODE_PAY_RATES.get(node.node_type.upper(), 0)} SACCP/hr"}
 @app.post("/api/node_heartbeat")
-async def heartbeat(data: dict):
     nid = data.get("node_id", "")
-    if nid in active_nodes:
-        active_nodes[nid]["last_hb"] = time.time()
-    await db_req("post", f"/nodes/{nid}/heartbeat", json=data)
     return {"ok": True}
 @app.get("/api/get_work")
-async def get_work(node_id: str, node_type: str = "CPU", has_gpu: bool = False,
-                   ram_gb: float = 4, libs: str = ""):
-    r = await db_req("get", "/fragments/pending?limit=10")
-    if not r or r.status_code != 200:
-        return {"work": None}
-    pending = r.json()
-    if not pending:
-        return {"work": None}
-    for frag in pending:
-        if frag.get("required_gpu") and not has_gpu:
-            continue
-        if frag.get("min_ram_gb", 0) > ram_gb > 0:
-            continue
-        frag_id = frag["fragment_id"]
-        await db_req("patch", f"/fragments/{frag_id}", json={
-            "status": "running",
-            "assigned_node": node_id,
-            "started_at": datetime.utcnow().isoformat()
-        })
-        print(f"[HEAD] 📤 Assigned {frag_id[:24]}... → {node_id}")
-        return {"work": frag}
     return {"work": None}
-# ─── Queries ────────────────────────────────────────────────────────────────
 @app.get("/api/task/{task_id}")
 async def get_task(task_id: str):
-    r = await db_req("get", f"/tasks/{task_id}")
-    if not r or r.status_code != 200:
-        raise HTTPException(404, "Not found")
     return r.json()
 @app.get("/api/task/{task_id}/fragments")
-async def task_fragments(task_id: str):
-    r = await db_req("get", f"/fragments/by_task/{task_id}")
     return r.json() if r and r.status_code == 200 else []
 @app.get("/api/my_tasks")
 async def my_tasks(api_key: str):
-    ur = await db_req("post", "/users/by_api_key", json={"api_key": api_key})
-    if not ur or ur.status_code != 200:
-        raise HTTPException(401, "Invalid API key")
-    r = await db_req("get", f"/tasks?owner={ur.json()['username']}")
     return r.json() if r and r.status_code == 200 else []
 @app.get("/api/stats")
 async def stats():
-    r = await db_req("get", "/stats")
-    return r.json() if r and r.status_code == 200 else {}
 @app.get("/health")
 async def health():
-    return {"status": "healthy", "head_id": HEAD_ID, "version": "2.0.3",
-            "active_nodes": len(active_nodes)}
-# ─── Safety Net: catch stuck tasks ──────────────────────────────────────────
 @app.on_event("startup")
 async def startup():
-    print(f"[HEAD] SharePUTER v2.0.3 starting — {HEAD_ID}")
-    asyncio.create_task(stuck_task_checker())
-    asyncio.create_task(stale_node_cleaner())
-async def stuck_task_checker():
-    """Every 15 seconds, check for tasks where all fragments done but task still running."""
     while True:
         await asyncio.sleep(15)
         try:
-            r = await db_req("get", "/tasks")
-            if not r or r.status_code != 200:
-                continue
-            for task in r.json():
-                if task["status"] not in ("running", "pending"):
-                    continue
-                tid = task["task_id"]
-                fr = await db_req("get", f"/fragments/by_task/{tid}")
-                if not fr or fr.status_code != 200:
-                    continue
-                frags = fr.json()
-                if not frags:
-                    continue
-                done = len([f for f in frags if f["status"] in ("completed", "failed")])
-                if done >= len(frags):
-                    print(f"[HEAD] 🔧 Stuck task found: {tid[:8]}, fixing...")
-                    await try_complete_task(tid)
-        except Exception as e:
-            print(f"[HEAD] Stuck checker error: {e}")
-async def stale_node_cleaner():
     while True:
         await asyncio.sleep(60)
-        cutoff = time.time() - 120
-        dead = [n for n, d in list(active_nodes.items()) if d.get("last_hb", 0) < cutoff]
         for n in dead:
             active_nodes.pop(n, None)
-            await db_req("post", f"/nodes/{n}/heartbeat", json={"status": "offline"})
 if __name__ == "__main__":
-    import uvicorn
-    uvicorn.run(app, host="0.0.0.0", port=7860)

 """
+SharePUTER™ v2.1 Head Node
+Result fusion, live streaming, IPU/TPU, fixed stats display
 """
+from fastapi import FastAPI, HTTPException, Request
 from fastapi.responses import HTMLResponse, JSONResponse
 from fastapi.middleware.cors import CORSMiddleware
 from pydantic import BaseModel
 from typing import Optional, Dict, Any
 import httpx, asyncio, uuid, os, json, time, ast, math
+from datetime import datetime
+app = FastAPI(title="SharePUTER™ v2.1")
 app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"])
+DB_URL = os.environ.get("SACCP_DB_URL", "http://localhost:5000")
 DB_SECRET = os.environ.get("SACCP_DB_SECRET", "saccp-v2-master-key")
 HEAD_ID = os.environ.get("HEAD_NODE_ID", f"head-{uuid.uuid4().hex[:8]}")
 UNIT_PRICING = {
     "cpu": 0.5, "ram": 0.1,
     "gpu_t4": 4.0, "gpu_a10": 10.0, "gpu_a100": 25.0, "gpu_generic": 5.0,
+    "tpu_v2": 15.0, "tpu_v4": 40.0,
+    "ipu_pod4": 20.0, "ipu_pod16": 60.0,
 }
+NODE_PAY_RATES = {"RAM": 0.3, "CPU": 1.5, "GPU": 5.0, "TPU": 8.0, "IPU": 7.0}
 active_nodes: Dict[str, dict] = {}
+def db_h(): return {"X-SACCP-Secret": DB_SECRET, "Content-Type": "application/json"}
+async def db(method, path, **kw):
     try:
+        async with httpx.AsyncClient(timeout=30) as c:
+            return await getattr(c, method)(f"{DB_URL}{path}", headers=db_h(), **kw)
     except Exception as e:
+        print(f"[HEAD] DB err {method} {path}: {e}")
         return None
 class UserCreate(BaseModel):
     username: str
     password: str
 class TaskSubmit(BaseModel):
     api_key: str
     code: str
     task_type: str = "general"
     config: dict = {}
     required_libs: list = []
 class NodeRegister(BaseModel):
     node_type: str
     owner: str = "anonymous"
     specs: dict = {}
     installed_libs: list = []
+# ─── Homepage (JS-loaded stats so page always renders) ─────────────────────
 @app.get("/", response_class=HTMLResponse)
 async def homepage():
     return HTMLResponse(f"""<!DOCTYPE html>
+<html><head><meta charset="UTF-8"><title>SharePUTER™ v2.1</title>
+<style>
+body{{font-family:system-ui;background:#08080d;color:#ddd;margin:0;padding:40px;max-width:1000px;margin:0 auto}}
+h1{{font-size:2.5rem;background:linear-gradient(135deg,#00ff88,#00aaff);-webkit-background-clip:text;-webkit-text-fill-color:transparent}}
+.g{{display:grid;grid-template-columns:repeat(auto-fill,minmax(140px,1fr));gap:12px;margin:24px 0}}
+.c{{background:#111;border:1px solid #222;border-radius:12px;padding:20px;text-align:center}}
+.c .n{{font-size:1.8rem;font-weight:700;color:#00ff88;font-family:monospace}}
+.c .l{{color:#888;font-size:.75rem;text-transform:uppercase;letter-spacing:1px;margin-top:4px}}
+.pulse{{display:inline-block;width:8px;height:8px;border-radius:50%;background:#00ff88;
+animation:p 2s infinite;margin-right:6px}}
+@keyframes p{{0%,100%{{box-shadow:0 0 0 0 rgba(0,255,136,.4)}}50%{{box-shadow:0 0 0 6px rgba(0,255,136,0)}}}}
+</style></head><body>
+<h1>SharePUTER™ v2.1</h1>
+<p><span class="pulse"></span>Head Node <code>{HEAD_ID}</code> — Online</p>
+<div class="g" id="stats"></div>
+<script>
+async function load(){{try{{const r=await fetch('/api/stats');const s=await r.json();
+const items=[
+['online_nodes','Nodes Online','#00ff88'],['cpu_nodes_online','CPU','#00aaff'],
+['gpu_nodes_online','GPU','#aa66ff'],['tpu_nodes_online','TPU','#ff8844'],
+['ipu_nodes_online','IPU','#00eeff'],['total_users','Users','#ffcc00'],
+['active_tasks','Active Tasks','#00aaff'],['completed_tasks','Completed','#00ff88'],
+['pending_fragments','Pending Frags','#ffcc00'],['completed_fragments','Done Frags','#00ff88'],
+];
+document.getElementById('stats').innerHTML=items.map(([k,l,c])=>
+`<div class="c"><div class="n" style="color:${{c}}">${{s[k]||0}}</div><div class="l">${{l}}</div></div>`).join('')
+}}catch(e){{document.getElementById('stats').innerHTML='<p style="color:#f44">Stats unavailable</p>'}}}}
+load();setInterval(load,5000);
+</script>
+</body></html>""")
 # ─── Auth ───────────────────────────────────────────────────────────────────
 @app.post("/api/register")
+async def register(u: UserCreate):
+    r = await db("post", "/users", json={"username": u.username, "password": u.password})
+    if not r or r.status_code != 201: raise HTTPException(r.status_code if r else 500, r.json().get("error") if r else "DB error")
     return r.json()
 @app.post("/api/login")
+async def login(u: UserCreate):
+    r = await db("post", "/users/auth", json={"username": u.username, "password": u.password})
+    if not r or r.status_code != 200: raise HTTPException(401, "Invalid credentials")
     return r.json()
 @app.get("/api/balance")
 async def balance(api_key: str):
+    r = await db("post", "/users/by_api_key", json={"api_key": api_key})
+    if not r or r.status_code != 200: raise HTTPException(401, "Invalid API key")
+    u = r.json(); return {"username": u["username"], "balance": u["balance"]}
 @app.get("/api/capabilities")
+async def caps():
+    r = await db("get", "/nodes/capabilities"); return r.json() if r and r.status_code == 200 else {}
 @app.get("/api/pricing")
+async def pricing(): return UNIT_PRICING
 @app.get("/api/available_libs")
+async def avlibs():
+    r = await db("get", "/libs/available"); return r.json() if r and r.status_code == 200 else {}
 # ─── Cost ───────────────────────────────────────────────────────────────────
+def calc_cost(cfg):
+    c = cfg.get("cpus", 1) * UNIT_PRICING["cpu"] + cfg.get("ram_gb", 4) * UNIT_PRICING["ram"]
+    gpus = cfg.get("gpus", 0)
+    gt = cfg.get("gpu_type", "generic")
+    if gpus: c += gpus * UNIT_PRICING.get(f"gpu_{gt}", 5)
+    tpus = cfg.get("tpus", 0)
+    tt = cfg.get("tpu_type", "v2")
+    if tpus: c += tpus * UNIT_PRICING.get(f"tpu_{tt}", 15)
+    ipus = cfg.get("ipus", 0)
+    it = cfg.get("ipu_type", "pod4")
+    if ipus: c += ipus * UNIT_PRICING.get(f"ipu_{it}", 20)
+    return round(c, 4)
+# ─── Result Fusion Engine ──────────────────────────────────────────────────
+def fuse_results(results):
+    """Merge fragment results into one unified result."""
+    if not results: return None
+    if len(results) == 1: return results[0]
+    # All dicts → merge keys (sum numbers, concat lists)
+    if all(isinstance(r, dict) for r in results):
+        merged = {}
+        all_keys = set()
+        for r in results: all_keys.update(r.keys())
+        for key in all_keys:
+            vals = [r[key] for r in results if key in r]
+            if not vals: continue
+            if all(isinstance(v, (int, float)) for v in vals):
+                merged[key] = sum(vals)
+            elif all(isinstance(v, list) for v in vals):
+                merged[key] = [item for v in vals for item in v]
+            elif all(isinstance(v, str) for v in vals):
+                merged[key] = "\n".join(vals)
+            else:
+                merged[key] = vals[-1]  # Take last value
+        return merged
+    # All lists → concatenate
+    if all(isinstance(r, list) for r in results):
+        return [item for r in results for item in r]
+    # All numbers → sum
+    if all(isinstance(r, (int, float)) for r in results):
+        return sum(results)
+    # All strings → merge (remove redundant lines)
+    if all(isinstance(r, str) for r in results):
+        return "\n".join(results)
+    return results
+def fuse_stdout(fragments):
+    """Merge fragment stdout into one continuous stream."""
+    merged = []
+    for f in sorted(fragments, key=lambda x: x["fragment_index"]):
+        stdout = f.get("stdout", "")
+        if stdout:
+            for line in stdout.split('\n'):
+                if line.strip():
+                    merged.append(line)
+    return "\n".join(merged)
 # ─── Fragmentation ─────────────────────────────────────────────────────────
+def fragment_code(code, cfg, task_id, libs):
     frags = []
+    cpus = max(1, cfg.get("cpus", 1))
+    needs_gpu = cfg.get("gpus", 0) > 0
+    ram = cfg.get("ram_gb", 4)
     target = max(2, min(cpus * 2, 8))
     loop = None
     try:
         tree = ast.parse(code)
                 if isinstance(fn, ast.Name) and fn.id == "range":
                     args = node.iter.args
                     try:
+                        if len(args) == 1: loop = {"s": 0, "e": ast.literal_eval(args[0])}
+                        elif len(args) >= 2: loop = {"s": ast.literal_eval(args[0]), "e": ast.literal_eval(args[1])}
                         if loop:
+                            loop["var"] = node.target.id if isinstance(node.target, ast.Name) else "i"
                             break
+                    except: loop = None
+    except: pass
+    if loop and (loop["e"] - loop["s"]) >= 20:
+        total = loop["e"] - loop["s"]
         n = min(target, total)
         chunk = math.ceil(total / n)
+        patterns = ([f"range({loop['e']})"] if loop["s"] == 0 else []) + [
+            f"range({loop['s']}, {loop['e']})", f"range({loop['s']},{loop['e']})"]
+        if loop["s"] == 0: patterns.append(f"range(0, {loop['e']})")
         for idx in range(n):
+            cs = loop["s"] + idx * chunk
+            ce = min(loop["s"] + (idx + 1) * chunk, loop["e"])
+            if cs >= loop["e"]: break
             mod = code
             for pat in patterns:
                 if pat in mod:
+                    mod = mod.replace(pat, f"range({cs}, {ce})", 1)
                     break
+            # Inject SACCP metadata
+            header = f"""# ═══ SACCP Fragment {idx}/{n} ═══
+__saccp_rank__ = {idx}
+__saccp_world_size__ = {n}
+__saccp_chunk__ = ({cs}, {ce})
+"""
             frags.append({
+                "fragment_id": f"{task_id}_frag_{idx}", "task_id": task_id,
+                "fragment_index": idx, "fragment_type": "compute",
+                "code": header + mod,
+                "input_data": json.dumps({"rank": idx, "world_size": n, "chunk": [cs, ce]}),
+                "required_libs": libs, "required_gpu": needs_gpu,
+                "min_ram_gb": max(1, ram // n), "timeout_seconds": 600,
             })
     if not frags:
+        header = """# ═══ SACCP Single Fragment ═══
+__saccp_rank__ = 0
+__saccp_world_size__ = 1
+"""
         frags.append({
+            "fragment_id": f"{task_id}_frag_0", "task_id": task_id,
+            "fragment_index": 0, "fragment_type": "compute",
+            "code": header + code,
+            "input_data": json.dumps({"rank": 0, "world_size": 1}),
+            "required_libs": libs, "required_gpu": needs_gpu,
+            "min_ram_gb": ram, "timeout_seconds": 600,
         })
+    print(f"[HEAD] Task {task_id[:8]} → {len(frags)} fragments")
     return frags
+# ─── Task Submit ─────────────────────────────────────────────────────────────
 @app.post("/api/submit_task")
+async def submit(task: TaskSubmit):
+    ur = await db("post", "/users/by_api_key", json={"api_key": task.api_key})
+    if not ur or ur.status_code != 200: raise HTTPException(401, "Invalid API key")
     user = ur.json()
+    cfg = task.config
+    hourly = calc_cost(cfg)
+    if user["balance"] < hourly * 0.001: raise HTTPException(402, "Insufficient balance")
     task_id = str(uuid.uuid4())
+    tr = await db("post", "/tasks", json={
         "task_id": task_id, "owner": user["username"], "code": task.code,
+        "task_type": task.task_type, "config": cfg, "required_libs": task.required_libs,
     })
+    if not tr or tr.status_code != 201: raise HTTPException(500, "Failed to create task")
+    fragments = fragment_code(task.code, cfg, task_id, task.required_libs)
+    fr = await db("post", "/fragments/batch", json={"fragments": fragments})
+    if not fr or fr.status_code != 201: raise HTTPException(500, "Failed to create fragments")
+    await db("patch", f"/tasks/{task_id}", json={"status": "running", "total_fragments": len(fragments)})
+    print(f"[HEAD] ✅ Task {task_id[:8]} submitted: {len(fragments)} frags, {hourly}/hr")
+    return {"task_id": task_id, "status": "running", "total_fragments": len(fragments), "hourly_cost": hourly}
+# ─── Completion Check (called from submit_result) ──────────────────────────
+async def try_complete(task_id):
+    fr = await db("get", f"/fragments/by_task/{task_id}")
+    if not fr or fr.status_code != 200: return
     fragments = fr.json()
+    if not fragments: return
     completed = [f for f in fragments if f["status"] == "completed"]
     failed = [f for f in fragments if f["status"] == "failed"]
+    await db("patch", f"/tasks/{task_id}", json={
+        "completed_fragments": len(completed), "failed_fragments": len(failed)})
+    if len(completed) + len(failed) < len(fragments): return
+    # ═══ ALL DONE — FUSE RESULTS ═══
+    print(f"[HEAD] Task {task_id[:8]} complete — fusing {len(fragments)} fragments")
+    # Collect raw results and parse JSON strings
+    raw_results = []
+    for f in sorted(completed, key=lambda x: x["fragment_index"]):
+        r = f.get("result")
+        if r is not None:
+            if isinstance(r, str):
+                try: r = json.loads(r)
+                except: pass
+            raw_results.append(r)
+    # Fuse results into one unified output
+    fused = fuse_results(raw_results)
+    merged_stdout = fuse_stdout(fragments)
+    errors = [f"[Fragment {f['fragment_index']}] {f.get('error', '?')}" for f in failed]
+    status = "failed" if len(failed) == len(fragments) else "completed"
     final = {
+        "fused_result": fused,
+        "stdout": merged_stdout,
+        "fragments_total": len(fragments),
         "fragments_ok": len(completed),
         "fragments_failed": len(failed),
         "errors": errors,
+        "per_fragment": raw_results,
     }
+    # Cost
+    tr = await db("get", f"/tasks/{task_id}")
+    owner, cfg = "unknown", {}
+    elapsed = 0.01
+    if tr and tr.status_code == 200:
+        t = tr.json(); owner = t.get("owner", "unknown"); cfg = t.get("config", {})
         try:
+            created = datetime.fromisoformat(t["created_at"].split("+")[0].split("Z")[0])
+            elapsed = max((datetime.utcnow() - created).total_seconds() / 3600, 0.001)
+        except: pass
+    cost = round(calc_cost(cfg) * elapsed, 6)
     if owner != "unknown":
+        await db("patch", f"/users/{owner}/balance", json={"amount": -cost, "reason": f"task_{task_id}"})
+    await db("patch", f"/tasks/{task_id}", json={
+        "status": status, "result": json.dumps(final), "cost": cost,
         "completed_at": datetime.utcnow().isoformat(),
         "error": "\n".join(errors) if errors else None,
     })
+    print(f"[HEAD] 🎉 Task {task_id[:8]} → {status} | Cost: {cost}")
+# ─── Submit Result ──────────────────────────────────────────────────────────
 @app.post("/api/submit_result")
 async def submit_result(request: Request):
+    data = await request.json()
+    fid = data.get("fragment_id", "")
+    if not fid: raise HTTPException(400, "fragment_id required")
+    await db("patch", f"/fragments/{fid}", json={
+        "status": data.get("status", "failed"),
+        "result": data.get("result"), "error": data.get("error") or None,
+        "stdout": data.get("stdout") or "", "completed_at": datetime.utcnow().isoformat(),
+        "resource_usage": data.get("resource_usage", {}),
     })
+    print(f"[HEAD] 📥 {fid[:20]}... → {data.get('status')}")
+    nid = data.get("node_id", "")
+    if data.get("status") == "completed" and nid:
+        nt = active_nodes.get(nid, {}).get("node_type", "CPU")
+        pay = NODE_PAY_RATES.get(nt, 0) / 360
+        if pay > 0: await db("post", f"/nodes/{nid}/pay", json={"amount": round(pay, 6)})
+    # Check completion
+    fg = await db("get", f"/fragments/{fid}")
+    if fg and fg.status_code == 200:
+        tid = fg.json().get("task_id")
+        if tid: await try_complete(tid)
     return {"ok": True}
+# ─── Live Task Stream ──────────────���───────────────────────────────────────
+@app.get("/api/task/{task_id}/live")
+async def live_task(task_id: str):
+    """Return current state with merged stdout for live streaming."""
+    tr = await db("get", f"/tasks/{task_id}")
+    if not tr or tr.status_code != 200: raise HTTPException(404, "Not found")
+    task = tr.json()
+    fr = await db("get", f"/fragments/by_task/{task_id}")
+    fragments = fr.json() if fr and fr.status_code == 200 else []
+    merged_stdout = fuse_stdout(fragments)
+    completed = len([f for f in fragments if f["status"] == "completed"])
+    running = len([f for f in fragments if f["status"] == "running"])
+    total = len(fragments)
+    return {
+        "task_id": task_id, "status": task["status"],
+        "completed": completed, "running": running, "total": total,
+        "stdout": merged_stdout, "result": task.get("result"),
+        "cost": task.get("cost", 0),
+    }
+# ─── Node + Task Endpoints ─────────────────────────────────────────────────
 @app.post("/api/register_node")
+async def reg_node(node: NodeRegister):
     nid = f"node-{uuid.uuid4().hex[:12]}"
+    await db("post", "/nodes", json={
         "node_id": nid, "node_type": node.node_type.upper(),
         "node_url": node.node_url, "owner": node.owner,
         "specs": node.specs, "installed_libs": node.installed_libs,
     })
     active_nodes[nid] = {"node_type": node.node_type.upper(), "last_hb": time.time()}
+    print(f"[HEAD] Node {nid} ({node.node_type})")
     return {"node_id": nid, "pay_rate": f"{NODE_PAY_RATES.get(node.node_type.upper(), 0)} SACCP/hr"}
 @app.post("/api/node_heartbeat")
+async def hb(data: dict):
     nid = data.get("node_id", "")
+    if nid in active_nodes: active_nodes[nid]["last_hb"] = time.time()
+    await db("post", f"/nodes/{nid}/heartbeat", json=data)
     return {"ok": True}
 @app.get("/api/get_work")
+async def get_work(node_id: str, node_type: str = "CPU", has_gpu: bool = False, ram_gb: float = 4, libs: str = ""):
+    r = await db("get", "/fragments/pending?limit=10")
+    if not r or r.status_code != 200: return {"work": None}
+    for f in r.json():
+        if f.get("required_gpu") and not has_gpu: continue
+        if f.get("min_ram_gb", 0) > ram_gb > 0: continue
+        fid = f["fragment_id"]
+        await db("patch", f"/fragments/{fid}", json={
+            "status": "running", "assigned_node": node_id, "started_at": datetime.utcnow().isoformat()})
+        print(f"[HEAD] 📤 {fid[:20]}... → {node_id}")
+        return {"work": f}
     return {"work": None}
 @app.get("/api/task/{task_id}")
 async def get_task(task_id: str):
+    r = await db("get", f"/tasks/{task_id}")
+    if not r or r.status_code != 200: raise HTTPException(404, "Not found")
     return r.json()
 @app.get("/api/task/{task_id}/fragments")
+async def get_frags(task_id: str):
+    r = await db("get", f"/fragments/by_task/{task_id}")
     return r.json() if r and r.status_code == 200 else []
 @app.get("/api/my_tasks")
 async def my_tasks(api_key: str):
+    ur = await db("post", "/users/by_api_key", json={"api_key": api_key})
+    if not ur or ur.status_code != 200: raise HTTPException(401, "Invalid API key")
+    r = await db("get", f"/tasks?owner={ur.json()['username']}")
     return r.json() if r and r.status_code == 200 else []
 @app.get("/api/stats")
 async def stats():
+    r = await db("get", "/stats"); return r.json() if r and r.status_code == 200 else {}
 @app.get("/health")
 async def health():
+    return {"status": "healthy", "head_id": HEAD_ID, "version": "2.1", "nodes": len(active_nodes)}
+# ─── Background ─────────────────────────────────────────────────────────────
 @app.on_event("startup")
 async def startup():
+    print(f"[HEAD] SharePUTER v2.1 — {HEAD_ID}")
+    asyncio.create_task(stuck_checker())
+    asyncio.create_task(node_cleaner())
+async def stuck_checker():
     while True:
         await asyncio.sleep(15)
         try:
+            r = await db("get", "/tasks")
+            if not r or r.status_code != 200: continue
+            for t in r.json():
+                if t["status"] not in ("running", "pending"): continue
+                fr = await db("get", f"/fragments/by_task/{t['task_id']}")
+                if not fr or fr.status_code != 200: continue
+                fs = fr.json()
+                if fs and len([f for f in fs if f["status"] in ("completed", "failed")]) >= len(fs):
+                    print(f"[HEAD] 🔧 Unsticking {t['task_id'][:8]}")
+                    await try_complete(t["task_id"])
+        except: pass
+async def node_cleaner():
     while True:
         await asyncio.sleep(60)
+        dead = [n for n, d in list(active_nodes.items()) if d.get("last_hb", 0) < time.time() - 120]
         for n in dead:
             active_nodes.pop(n, None)
+            await db("post", f"/nodes/{n}/heartbeat", json={"status": "offline"})
 if __name__ == "__main__":
+    import uvicorn; uvicorn.run(app, host="0.0.0.0", port=7860)