Update app.py

app.py

@@ -1,12 +1,13 @@
 """
-SharePUTER™ v2 Worker Node — FIXED
+SharePUTER™ v2 Worker Node — v2.0.3 FIXED
+Key fix: separate httpx client for get_work vs submit_result
 """
 
 from fastapi import FastAPI
-from fastapi.responses import HTMLResponse, JSONResponse
+from fastapi.responses import HTMLResponse
 from fastapi.middleware.cors import CORSMiddleware
 import httpx, asyncio, uuid, os, sys, io, json, time, traceback
-import subprocess, multiprocessing
+import multiprocessing
 from datetime import datetime
 from contextlib import redirect_stdout, redirect_stderr
 from typing import Optional
@@ -21,18 +22,16 @@ AUTO_REG = os.environ.get("SACCP_AUTO_REGISTER", "true").lower() == "true"
 NODE_ID: Optional[str] = None
 REGISTERED = False
 
+import psutil
 
 def detect_specs():
-    import psutil
     s = {
         "cpu_count": multiprocessing.cpu_count(),
         "ram_total_gb": round(psutil.virtual_memory().total / (1024**3), 2),
         "ram_available_gb": round(psutil.virtual_memory().available / (1024**3), 2),
         "platform": sys.platform,
         "python": sys.version.split()[0],
-        "gpu_available": False,
-        "gpu_name": "",
-        "gpu_memory_gb": 0,
+        "gpu_available": False, "gpu_name": "", "gpu_memory_gb": 0,
     }
     try:
         import torch
@@ -40,111 +39,88 @@ def detect_specs():
             s["gpu_available"] = True
             s["gpu_name"] = torch.cuda.get_device_name(0)
             s["gpu_memory_gb"] = round(torch.cuda.get_device_properties(0).total_mem / (1024**3), 2)
-    except ImportError:
+    except:
         pass
     return s
 
-
 def detect_libs():
-    common = ["numpy", "pandas", "scipy", "scikit-learn", "torch", "transformers",
-              "tensorflow", "keras", "pillow", "requests", "httpx", "fastapi", "pydantic"]
-    installed = []
+    common = ["numpy","pandas","scipy","scikit-learn","torch","transformers",
+              "tensorflow","keras","pillow","requests","httpx","fastapi","pydantic"]
+    out = []
     for lib in common:
         try:
-            __import__(lib.replace("-", "_"))
-            installed.append(lib)
-        except ImportError:
+            __import__(lib.replace("-","_"))
+            out.append(lib)
+        except:
             pass
-    return installed
-
+    return out
 
-import psutil
 SPECS = detect_specs()
 INSTALLED_LIBS = detect_libs()
 
 
 def safe_execute(code: str, timeout: int = 600) -> dict:
-    """Execute Python code safely."""
-    result = {
-        "status": "completed",
-        "result": None,
-        "stdout": "",
-        "error": None,
-        "execution_time": 0,
-        "resource_usage": {}
-    }
-
+    result = {"status": "completed", "result": None, "stdout": "", "error": None,
+              "execution_time": 0, "resource_usage": {}}
     stdout_cap = io.StringIO()
     stderr_cap = io.StringIO()
     namespace = {}
-
     start = time.time()
     start_mem = psutil.Process().memory_info().rss
 
     try:
         with redirect_stdout(stdout_cap), redirect_stderr(stderr_cap):
             exec(compile(code, "<fragment>", "exec"), namespace)
-
         result["execution_time"] = round(time.time() - start, 3)
         result["stdout"] = stdout_cap.getvalue()
-
         end_mem = psutil.Process().memory_info().rss
         result["resource_usage"] = {
             "time_seconds": result["execution_time"],
             "memory_delta_mb": round((end_mem - start_mem) / (1024**2), 2),
         }
-
-        # Capture result variable
         for var in ["result", "__saccp_result__", "__saccp_results__", "output"]:
             if var in namespace and namespace[var] is not None:
                 val = namespace[var]
                 try:
                     json.dumps(val)
                     result["result"] = val
-                except (TypeError, ValueError):
+                except:
                     result["result"] = str(val)[:5000]
                 break
-
         if result["result"] is None and result["stdout"]:
             result["result"] = result["stdout"].strip()[-5000:]
-
     except Exception as e:
         result["status"] = "failed"
         result["error"] = f"{type(e).__name__}: {str(e)}\n{traceback.format_exc()[-1500:]}"
         result["execution_time"] = round(time.time() - start, 3)
         result["stdout"] = stdout_cap.getvalue()
-
     return result
 
 
 @app.get("/", response_class=HTMLResponse)
 async def homepage():
-    import psutil
-    ram_avail = round(psutil.virtual_memory().available / (1024**3), 2)
-    return HTMLResponse(f"""<!DOCTYPE html>
-<html><head><title>SharePUTER Worker</title>
+    ram = round(psutil.virtual_memory().available / (1024**3), 2)
+    return HTMLResponse(f"""<!DOCTYPE html><html><head><title>Worker</title>
 <style>body{{font-family:system-ui;background:#0a0a0f;color:#ddd;padding:40px;max-width:600px;margin:0 auto}}
-h1{{color:#00ff88}}.stat{{background:#111;padding:12px;margin:8px 0;border-radius:8px}}</style></head>
-<body><h1>🤖 SharePUTER Worker</h1>
-<p>Node ID: <strong>{NODE_ID or 'Not registered'}</strong></p>
-<p>Type: <strong>{NODE_TYPE}</strong></p>
-<p>Status: <strong style="color:#00ff88">{'ACTIVE' if REGISTERED else 'WAITING'}</strong></p>
-<div class="stat">CPUs: {SPECS['cpu_count']}</div>
-<div class="stat">RAM: {ram_avail} / {SPECS['ram_total_gb']} GB</div>
-<div class="stat">GPU: {SPECS['gpu_name'] or 'None'}</div>
-<div class="stat">Libs: {', '.join(INSTALLED_LIBS[:10])}</div>
+h1{{color:#00ff88}}.s{{background:#111;padding:12px;margin:8px 0;border-radius:8px}}</style></head>
+<body><h1>🤖 SharePUTER Worker v2.0.3</h1>
+<p>Node: <b>{NODE_ID or 'unregistered'}</b> | Type: <b>{NODE_TYPE}</b> |
+Status: <b style="color:#00ff88">{'ACTIVE' if REGISTERED else 'WAITING'}</b></p>
+<div class="s">CPUs: {SPECS['cpu_count']}</div>
+<div class="s">RAM: {ram}/{SPECS['ram_total_gb']} GB</div>
+<div class="s">GPU: {SPECS['gpu_name'] or 'None'}</div>
+<div class="s">Libs: {', '.join(INSTALLED_LIBS)}</div>
 </body></html>""")
 
 
 @app.get("/health")
 async def health():
-    return {"status": "online", "node_id": NODE_ID, "registered": REGISTERED}
+    return {"status": "online", "node_id": NODE_ID, "registered": REGISTERED, "version": "2.0.3"}
 
 
 async def worker_loop():
     global NODE_ID, REGISTERED
-    
-    # Wait for registration
+
     while not REGISTERED:
         await asyncio.sleep(2)
         if AUTO_REG:
@@ -153,83 +129,94 @@ async def worker_loop():
                     r = await c.post(f"{HEAD_URL}/api/register_node", json={
                         "node_type": NODE_TYPE,
                         "node_url": os.environ.get("SPACE_HOST", "http://localhost:7861"),
-                        "owner": NODE_OWNER,
-                        "specs": SPECS,
-                        "installed_libs": INSTALLED_LIBS,
+                        "owner": NODE_OWNER, "specs": SPECS, "installed_libs": INSTALLED_LIBS,
                     })
                     if r.status_code == 200:
                         NODE_ID = r.json().get("node_id")
                         REGISTERED = True
                         print(f"[SACCP] Registered as {NODE_ID}")
             except Exception as e:
-                print(f"[SACCP] Registration failed: {e}")
+                print(f"[SACCP] Reg failed: {e}")
 
     print(f"[SACCP] Worker active: {NODE_ID} ({NODE_TYPE})")
     last_hb = 0
 
     while True:
         try:
-            # Heartbeat every 25 seconds
+            # ─── Heartbeat ───
             if time.time() - last_hb > 25:
                 try:
                     async with httpx.AsyncClient(timeout=10) as c:
                         await c.post(f"{HEAD_URL}/api/node_heartbeat", json={
-                            "node_id": NODE_ID,
-                            "status": "online",
+                            "node_id": NODE_ID, "status": "online",
                             "installed_libs": INSTALLED_LIBS,
                         })
                     last_hb = time.time()
                 except:
                     pass
 
-            # Get work
+            # ─── Get Work (separate client) ───
+            work = None
             async with httpx.AsyncClient(timeout=15) as c:
                 libs_str = ",".join(INSTALLED_LIBS)
-                r = await c.get(
-                    f"{HEAD_URL}/api/get_work",
-                    params={
-                        "node_id": NODE_ID,
-                        "node_type": NODE_TYPE,
-                        "has_gpu": str(SPECS['gpu_available']).lower(),
-                        "ram_gb": SPECS['ram_available_gb'],
-                        "libs": libs_str,
-                    }
-                )
-                
-                if r.status_code != 200:
-                    await asyncio.sleep(2)
-                    continue
-                
-                work = r.json().get("work")
-                if not work:
-                    await asyncio.sleep(1)
-                    continue
-
-                frag_id = work["fragment_id"]
-                code = work.get("code", "")
-                ftype = work.get("fragment_type", "compute")
-                timeout = work.get("timeout_seconds", 600)
-
-                print(f"[SACCP] Running {frag_id} ({ftype})")
-
-                # Execute the code
-                exec_result = safe_execute(code, timeout=timeout)
-
-                # Submit result
-                await c.post(f"{HEAD_URL}/api/submit_result", json={
-                    "fragment_id": frag_id,
-                    "node_id": NODE_ID,
-                    "status": exec_result["status"],
-                    "result": exec_result.get("result"),
-                    "error": exec_result.get("error"),
-                    "stdout": exec_result.get("stdout", ""),
-                    "resource_usage": exec_result.get("resource_usage", {}),
+                r = await c.get(f"{HEAD_URL}/api/get_work", params={
+                    "node_id": NODE_ID, "node_type": NODE_TYPE,
+                    "has_gpu": str(SPECS['gpu_available']).lower(),
+                    "ram_gb": SPECS['ram_available_gb'], "libs": libs_str,
                 })
+                if r.status_code == 200:
+                    work = r.json().get("work")
+
+            if not work:
+                await asyncio.sleep(1)
+                continue
+
+            frag_id = work["fragment_id"]
+            code = work.get("code", "")
+            ftype = work.get("fragment_type", "compute")
+            timeout = work.get("timeout_seconds", 600)
+
+            print(f"[SACCP] ▶ Running {frag_id} ({ftype})")
+
+            # ─── Execute ───
+            exec_result = safe_execute(code, timeout=timeout)
+
+            print(f"[SACCP] ✓ Executed {frag_id[:20]}... → {exec_result['status']} ({exec_result['execution_time']}s)")
+
+            # ─── Submit Result (SEPARATE client, fresh connection) ───
+            submit_payload = {
+                "fragment_id": frag_id,
+                "node_id": NODE_ID,
+                "status": exec_result["status"],
+                "result": exec_result.get("result"),
+                "error": exec_result.get("error") or "",
+                "stdout": exec_result.get("stdout") or "",
+                "resource_usage": exec_result.get("resource_usage", {}),
+            }
+
+            submit_ok = False
+            for attempt in range(3):
+                try:
+                    async with httpx.AsyncClient(timeout=30) as submit_client:
+                        sr = await submit_client.post(
+                            f"{HEAD_URL}/api/submit_result",
+                            json=submit_payload
+                        )
+                        if sr.status_code == 200:
+                            print(f"[SACCP] 📤 Result submitted for {frag_id[:20]}...")
+                            submit_ok = True
+                            break
+                        else:
+                            print(f"[SACCP] ⚠️ Submit got {sr.status_code}: {sr.text[:200]}")
+                except Exception as e:
+                    print(f"[SACCP] ⚠️ Submit attempt {attempt+1} failed: {e}")
+                    await asyncio.sleep(1)
 
-                print(f"[SACCP] {frag_id} → {exec_result['status']} ({exec_result['execution_time']}s)")
+            if not submit_ok:
+                print(f"[SACCP] ❌ FAILED to submit result for {frag_id} after 3 attempts!")
 
         except Exception as e:
-            print(f"[SACCP] Error: {e}")
+            print(f"[SACCP] Loop error: {e}")
             await asyncio.sleep(3)