Create app.py

app.py

@@ -0,0 +1,1013 @@
+"""
+SharePUTER Head Node - Runs on HF Spaces (Free CPU)
+Orchestrates task fragmentation, scheduling, and reassembly.
+FastAPI application.
+"""
+
+from fastapi import FastAPI, HTTPException, Request, BackgroundTasks
+from fastapi.responses import HTMLResponse, JSONResponse
+from fastapi.middleware.cors import CORSMiddleware
+from pydantic import BaseModel
+from typing import Optional, List, Dict, Any
+import httpx
+import asyncio
+import uuid
+import time
+import json
+import ast
+import textwrap
+import math
+import os
+from datetime import datetime, timedelta
+
+app = FastAPI(title="SharePUTER Head Node", version="1.0.0")
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# ─── Configuration ──────────────────────────────────────────────────────────
+DB_NODE_URL = os.environ.get("SACCP_DB_URL", "http://localhost:5000")
+DB_SECRET = os.environ.get("SACCP_DB_SECRET", "saccp-master-key-change-me")
+HEAD_NODE_ID = os.environ.get("HEAD_NODE_ID", f"head-{uuid.uuid4().hex[:8]}")
+HEAD_NODE_URL = os.environ.get("HEAD_NODE_URL", "http://localhost:7860")
+
+# Pricing tiers (SACCP per hour)
+PRICING = {
+    "basic_cpu": {"price_per_hour": 1.0, "vcpus": 1, "ram_gb": 4, "gpu": False, "label": "Basic CPU"},
+    "medium_cpu": {"price_per_hour": 4.0, "vcpus": 10, "ram_gb": 20, "gpu": False, "label": "Medium CPU"},
+    "large_cpu": {"price_per_hour": 10.0, "vcpus": 30, "ram_gb": 50, "gpu": False, "label": "Large CPU"},
+    "basic_gpu": {"price_per_hour": 8.0, "vcpus": 4, "ram_gb": 16, "gpu": True, "gpu_type": "T4", "label": "Basic GPU"},
+    "medium_gpu": {"price_per_hour": 20.0, "vcpus": 8, "ram_gb": 32, "gpu": True, "gpu_type": "A10G", "label": "Medium GPU"},
+    "large_gpu": {"price_per_hour": 50.0, "vcpus": 16, "ram_gb": 64, "gpu": True, "gpu_type": "A100", "label": "Large GPU"},
+    "ram_only": {"price_per_hour": 0.5, "vcpus": 0, "ram_gb": 8, "gpu": False, "label": "RAM Only"},
+}
+
+# Node payment rates (SACCP per hour for contributing)
+NODE_PAY_RATES = {
+    "RAM": 0.3,
+    "CPU": 1.5,
+    "GPU": 5.0,
+}
+
+# In-memory caches (will be lost on restart, that's fine — DB is source of truth)
+active_nodes: Dict[str, dict] = {}
+running_tasks: Dict[str, dict] = {}
+
+
+def db_headers():
+    return {"X-SACCP-Secret": DB_SECRET, "Content-Type": "application/json"}
+
+
+async def db_request(method: str, path: str, **kwargs):
+    """Make a request to the database node."""
+    async with httpx.AsyncClient(timeout=30.0) as client:
+        url = f"{DB_NODE_URL}{path}"
+        resp = await getattr(client, method)(url, headers=db_headers(), **kwargs)
+        return resp
+
+
+# ─── Models ─────────────────────────────────────────────────────────────────
+class UserCreate(BaseModel):
+    username: str
+    password: str
+
+class UserLogin(BaseModel):
+    username: str
+    password: str
+
+class TaskSubmit(BaseModel):
+    api_key: str
+    code: str
+    tier: str = "basic_cpu"
+
+class NodeRegister(BaseModel):
+    node_type: str  # RAM, CPU, GPU
+    node_url: str
+    owner: str = "anonymous"
+    specs: dict = {}
+
+class FragmentResult(BaseModel):
+    fragment_id: str
+    node_id: str
+    status: str  # completed, failed
+    result: Any = None
+    error: str = None
+    stdout: str = ""
+
+
+# ─── Homepage ───────────────────────────────────────────────────────────────
+@app.get("/", response_class=HTMLResponse)
+async def homepage():
+    try:
+        stats_resp = await db_request("get", "/stats")
+        stats = stats_resp.json() if stats_resp.status_code == 200 else {}
+    except Exception:
+        stats = {}
+
+    pricing_rows = ""
+    for tier_id, tier in PRICING.items():
+        gpu_badge = '<span class="badge gpu">GPU</span>' if tier.get("gpu") else '<span class="badge cpu">CPU</span>'
+        gpu_info = f' ({tier.get("gpu_type", "")})' if tier.get("gpu") else ""
+        pricing_rows += f"""
+        <tr>
+            <td>{gpu_badge} {tier['label']}</td>
+            <td>{tier['vcpus']} vCPU</td>
+            <td>{tier['ram_gb']} GB</td>
+            <td>{gpu_info if gpu_info.strip() else '—'}</td>
+            <td class="price">{tier['price_per_hour']} SACCP/hr</td>
+        </tr>"""
+
+    node_pay_rows = ""
+    for ntype, rate in NODE_PAY_RATES.items():
+        node_pay_rows += f"<tr><td>{ntype} Node</td><td class='price'>{rate} SACCP/hr</td></tr>"
+
+    html = f"""<!DOCTYPE html>
+<html lang="en">
+<head>
+    <meta charset="UTF-8">
+    <meta name="viewport" content="width=device-width, initial-scale=1.0">
+    <title>SharePUTER™ — SACCP Network</title>
+    <style>
+        @import url('https://fonts.googleapis.com/css2?family=JetBrains+Mono:wght@300;400;500;700&family=Inter:wght@300;400;600;700;900&display=swap');
+
+        * {{ margin: 0; padding: 0; box-sizing: border-box; }}
+
+        body {{
+            font-family: 'Inter', sans-serif;
+            background: #0a0a0f;
+            color: #e0e0e0;
+            min-height: 100vh;
+            overflow-x: hidden;
+        }}
+
+        .bg-grid {{
+            position: fixed; top: 0; left: 0; right: 0; bottom: 0;
+            background-image:
+                linear-gradient(rgba(0,255,136,0.03) 1px, transparent 1px),
+                linear-gradient(90deg, rgba(0,255,136,0.03) 1px, transparent 1px);
+            background-size: 50px 50px;
+            z-index: 0;
+        }}
+
+        .glow-orb {{
+            position: fixed;
+            border-radius: 50%;
+            filter: blur(120px);
+            z-index: 0;
+            animation: float 8s ease-in-out infinite;
+        }}
+        .glow-orb.green {{ width: 400px; height: 400px; background: rgba(0,255,136,0.08); top: 10%; left: 10%; }}
+        .glow-orb.blue {{ width: 350px; height: 350px; background: rgba(0,136,255,0.06); top: 60%; right: 10%; animation-delay: 3s; }}
+        .glow-orb.purple {{ width: 300px; height: 300px; background: rgba(136,0,255,0.05); bottom: 10%; left: 40%; animation-delay: 5s; }}
+
+        @keyframes float {{
+            0%, 100% {{ transform: translate(0, 0); }}
+            50% {{ transform: translate(30px, -30px); }}
+        }}
+
+        .container {{ max-width: 1100px; margin: 0 auto; padding: 40px 20px; position: relative; z-index: 1; }}
+
+        header {{
+            text-align: center;
+            margin-bottom: 60px;
+            padding: 40px 0;
+        }}
+
+        .logo {{
+            font-size: 4rem;
+            font-weight: 900;
+            letter-spacing: -2px;
+            background: linear-gradient(135deg, #00ff88, #00aaff, #8800ff);
+            -webkit-background-clip: text;
+            -webkit-text-fill-color: transparent;
+            text-shadow: none;
+            margin-bottom: 8px;
+        }}
+
+        .logo span {{
+            font-size: 1.2rem;
+            vertical-align: super;
+            -webkit-text-fill-color: #00ff88;
+        }}
+
+        .tagline {{
+            font-family: 'JetBrains Mono', monospace;
+            color: #00ff88;
+            font-size: 0.9rem;
+            letter-spacing: 4px;
+            text-transform: uppercase;
+        }}
+
+        .subtitle {{
+            color: #888;
+            margin-top: 16px;
+            font-size: 1.1rem;
+            font-weight: 300;
+        }}
+
+        .status-bar {{
+            display: flex;
+            justify-content: center;
+            gap: 40px;
+            flex-wrap: wrap;
+            margin: 40px 0;
+        }}
+
+        .stat-card {{
+            background: linear-gradient(135deg, rgba(255,255,255,0.03), rgba(255,255,255,0.01));
+            border: 1px solid rgba(255,255,255,0.06);
+            border-radius: 16px;
+            padding: 24px 32px;
+            text-align: center;
+            backdrop-filter: blur(10px);
+            min-width: 140px;
+            transition: all 0.3s;
+        }}
+        .stat-card:hover {{
+            border-color: rgba(0,255,136,0.3);
+            transform: translateY(-4px);
+            box-shadow: 0 8px 32px rgba(0,255,136,0.1);
+        }}
+        .stat-card .number {{
+            font-size: 2.5rem;
+            font-weight: 700;
+            color: #00ff88;
+            font-family: 'JetBrains Mono', monospace;
+        }}
+        .stat-card .label {{
+            color: #888;
+            font-size: 0.8rem;
+            text-transform: uppercase;
+            letter-spacing: 2px;
+            margin-top: 4px;
+        }}
+
+        .section {{
+            margin: 50px 0;
+        }}
+
+        .section h2 {{
+            font-size: 1.8rem;
+            margin-bottom: 24px;
+            color: #fff;
+            display: flex;
+            align-items: center;
+            gap: 12px;
+        }}
+        .section h2 .icon {{ font-size: 1.4rem; }}
+
+        table {{
+            width: 100%;
+            border-collapse: collapse;
+            background: rgba(255,255,255,0.02);
+            border-radius: 12px;
+            overflow: hidden;
+            border: 1px solid rgba(255,255,255,0.06);
+        }}
+        th {{
+            background: rgba(0,255,136,0.08);
+            color: #00ff88;
+            font-family: 'JetBrains Mono', monospace;
+            font-size: 0.8rem;
+            letter-spacing: 1px;
+            text-transform: uppercase;
+            padding: 14px 20px;
+            text-align: left;
+        }}
+        td {{
+            padding: 14px 20px;
+            border-bottom: 1px solid rgba(255,255,255,0.04);
+            font-size: 0.95rem;
+        }}
+        tr:hover td {{ background: rgba(0,255,136,0.02); }}
+        .price {{
+            color: #00ff88;
+            font-weight: 600;
+            font-family: 'JetBrains Mono', monospace;
+        }}
+
+        .badge {{
+            display: inline-block;
+            padding: 3px 10px;
+            border-radius: 20px;
+            font-size: 0.7rem;
+            font-weight: 600;
+            letter-spacing: 1px;
+            text-transform: uppercase;
+        }}
+        .badge.cpu {{ background: rgba(0,136,255,0.15); color: #00aaff; border: 1px solid rgba(0,136,255,0.3); }}
+        .badge.gpu {{ background: rgba(136,0,255,0.15); color: #aa66ff; border: 1px solid rgba(136,0,255,0.3); }}
+
+        .node-id {{
+            font-family: 'JetBrains Mono', monospace;
+            font-size: 0.85rem;
+            color: #888;
+            margin-top: 20px;
+            text-align: center;
+            padding: 12px;
+            background: rgba(255,255,255,0.02);
+            border-radius: 8px;
+            border: 1px solid rgba(255,255,255,0.04);
+        }}
+
+        .endpoint-list {{
+            display: grid;
+            grid-template-columns: repeat(auto-fill, minmax(300px, 1fr));
+            gap: 12px;
+            margin-top: 16px;
+        }}
+        .endpoint {{
+            background: rgba(255,255,255,0.02);
+            border: 1px solid rgba(255,255,255,0.06);
+            border-radius: 8px;
+            padding: 14px 18px;
+            font-family: 'JetBrains Mono', monospace;
+            font-size: 0.8rem;
+            transition: all 0.2s;
+        }}
+        .endpoint:hover {{
+            border-color: rgba(0,255,136,0.3);
+        }}
+        .endpoint .method {{
+            display: inline-block;
+            padding: 2px 8px;
+            border-radius: 4px;
+            font-size: 0.7rem;
+            font-weight: 700;
+            margin-right: 8px;
+        }}
+        .method.get {{ background: rgba(0,255,136,0.15); color: #00ff88; }}
+        .method.post {{ background: rgba(0,136,255,0.15); color: #00aaff; }}
+        .method.patch {{ background: rgba(255,170,0,0.15); color: #ffaa00; }}
+
+        .pulse {{
+            display: inline-block;
+            width: 10px; height: 10px;
+            border-radius: 50%;
+            background: #00ff88;
+            margin-right: 8px;
+            animation: pulse 2s ease-in-out infinite;
+        }}
+        @keyframes pulse {{
+            0%, 100% {{ opacity: 1; box-shadow: 0 0 0 0 rgba(0,255,136,0.4); }}
+            50% {{ opacity: 0.7; box-shadow: 0 0 0 8px rgba(0,255,136,0); }}
+        }}
+
+        footer {{
+            text-align: center;
+            color: #555;
+            padding: 40px 0;
+            font-size: 0.8rem;
+            border-top: 1px solid rgba(255,255,255,0.04);
+            margin-top: 60px;
+        }}
+    </style>
+</head>
+<body>
+    <div class="bg-grid"></div>
+    <div class="glow-orb green"></div>
+    <div class="glow-orb blue"></div>
+    <div class="glow-orb purple"></div>
+
+    <div class="container">
+        <header>
+            <div class="logo">SharePUTER<span>™</span></div>
+            <div class="tagline">━━━ SACCP Network ━━━</div>
+            <p class="subtitle">Shared Cloud Computing Protocol — Link computers. Share power. Earn SACCP.</p>
+            <p style="margin-top:12px;"><span class="pulse"></span> Head Node Online</p>
+        </header>
+
+        <div class="status-bar">
+            <div class="stat-card">
+                <div class="number">{stats.get('online_nodes', 0)}</div>
+                <div class="label">Nodes Online</div>
+            </div>
+            <div class="stat-card">
+                <div class="number">{stats.get('total_users', 0)}</div>
+                <div class="label">Users</div>
+            </div>
+            <div class="stat-card">
+                <div class="number">{stats.get('active_tasks', 0)}</div>
+                <div class="label">Active Tasks</div>
+            </div>
+            <div class="stat-card">
+                <div class="number">{stats.get('completed_fragments', 0)}</div>
+                <div class="label">Fragments Done</div>
+            </div>
+            <div class="stat-card">
+                <div class="number">{stats.get('total_tasks', 0)}</div>
+                <div class="label">Total Tasks</div>
+            </div>
+        </div>
+
+        <div class="section">
+            <h2><span class="icon">💰</span> Compute Pricing</h2>
+            <table>
+                <thead><tr><th>Tier</th><th>vCPUs</th><th>RAM</th><th>GPU</th><th>Price</th></tr></thead>
+                <tbody>{pricing_rows}</tbody>
+            </table>
+        </div>
+
+        <div class="section">
+            <h2><span class="icon">🖥️</span> Node Contribution Rewards</h2>
+            <table>
+                <thead><tr><th>Node Type</th><th>Earn Rate</th></tr></thead>
+                <tbody>{node_pay_rows}</tbody>
+            </table>
+        </div>
+
+        <div class="section">
+            <h2><span class="icon">🔌</span> API Endpoints</h2>
+            <div class="endpoint-list">
+                <div class="endpoint"><span class="method post">POST</span>/api/register</div>
+                <div class="endpoint"><span class="method post">POST</span>/api/login</div>
+                <div class="endpoint"><span class="method post">POST</span>/api/submit_task</div>
+                <div class="endpoint"><span class="method get">GET</span>/api/task/{{task_id}}</div>
+                <div class="endpoint"><span class="method get">GET</span>/api/my_tasks?api_key=...</div>
+                <div class="endpoint"><span class="method post">POST</span>/api/register_node</div>
+                <div class="endpoint"><span class="method post">POST</span>/api/node_heartbeat</div>
+                <div class="endpoint"><span class="method get">GET</span>/api/get_work?node_id=...</div>
+                <div class="endpoint"><span class="method post">POST</span>/api/submit_result</div>
+                <div class="endpoint"><span class="method get">GET</span>/api/stats</div>
+                <div class="endpoint"><span class="method get">GET</span>/api/pricing</div>
+                <div class="endpoint"><span class="method get">GET</span>/api/balance?api_key=...</div>
+            </div>
+        </div>
+
+        <div class="node-id">Head Node ID: {HEAD_NODE_ID} | DB: {DB_NODE_URL}</div>
+
+        <footer>
+            SharePUTER™ SACCP Network &copy; 2024 — Decentralized Shared Cloud Computing<br>
+            <span style="color:#00ff88;">Prototype v1.0</span>
+        </footer>
+    </div>
+</body>
+</html>"""
+    return HTMLResponse(content=html)
+
+
+# ─── User Endpoints ─────────────────────────────────────────────────────────
+@app.post("/api/register")
+async def register_user(user: UserCreate):
+    resp = await db_request("post", "/users", json={"username": user.username, "password": user.password})
+    if resp.status_code != 201:
+        raise HTTPException(status_code=resp.status_code, detail=resp.json().get("error", "Registration failed"))
+    return resp.json()
+
+
+@app.post("/api/login")
+async def login_user(user: UserLogin):
+    resp = await db_request("post", "/users/auth", json={"username": user.username, "password": user.password})
+    if resp.status_code != 200:
+        raise HTTPException(status_code=401, detail="Invalid credentials")
+    return resp.json()
+
+
+@app.get("/api/balance")
+async def get_balance(api_key: str):
+    resp = await db_request("post", "/users/by_api_key", json={"api_key": api_key})
+    if resp.status_code != 200:
+        raise HTTPException(status_code=401, detail="Invalid API key")
+    user = resp.json()
+    return {"username": user["username"], "balance": user["balance"]}
+
+
+# ─── Task Fragmentation Engine ──────────────────────────────────────────────
+def fragment_python_code(code: str, tier: str, task_id: str) -> List[dict]:
+    """
+    Split Python code into executable fragments.
+    Strategy:
+    1. Parse the code AST
+    2. Separate imports, functions/classes, and top-level statements
+    3. Create fragments: setup fragment + computation fragments + aggregation fragment
+    4. For loops/comprehensions, split iterations across fragments
+    """
+    fragments = []
+    tier_info = PRICING.get(tier, PRICING["basic_cpu"])
+
+    try:
+        tree = ast.parse(code)
+    except SyntaxError as e:
+        # Can't parse — send as single fragment
+        frag_id = f"{task_id}_frag_0"
+        fragments.append({
+            "fragment_id": frag_id,
+            "task_id": task_id,
+            "fragment_index": 0,
+            "fragment_type": "compute",
+            "code": code,
+            "input_data": json.dumps({"full_execution": True}),
+        })
+        return fragments
+
+    imports = []
+    functions = []
+    classes = []
+    statements = []
+
+    for node in ast.iter_child_nodes(tree):
+        segment = ast.get_source_segment(code, node)
+        if segment is None:
+            # Fallback: reconstruct from AST
+            try:
+                segment = ast.unparse(node)
+            except Exception:
+                continue
+
+        if isinstance(node, (ast.Import, ast.ImportFrom)):
+            imports.append(segment)
+        elif isinstance(node, ast.FunctionDef):
+            functions.append(segment)
+        elif isinstance(node, ast.ClassDef):
+            classes.append(segment)
+        else:
+            statements.append(segment)
+
+    preamble = "\n".join(imports + [""] + classes + [""] + functions)
+
+    # Setup fragment — defines all imports, functions, classes
+    setup_frag_id = f"{task_id}_setup"
+    fragments.append({
+        "fragment_id": setup_frag_id,
+        "task_id": task_id,
+        "fragment_index": 0,
+        "fragment_type": "setup",
+        "code": preamble,
+        "input_data": json.dumps({"type": "setup"}),
+    })
+
+    # Try to detect for-loops over ranges and split them
+    loop_fragments = []
+    other_statements = []
+
+    for stmt_str in statements:
+        try:
+            stmt_tree = ast.parse(stmt_str)
+            stmt_node = stmt_tree.body[0] if stmt_tree.body else None
+
+            if isinstance(stmt_node, ast.For):
+                # Try to detect range-based loops
+                if (isinstance(stmt_node.iter, ast.Call) and
+                    isinstance(stmt_node.iter.func, ast.Name) and
+                    stmt_node.iter.func.id == "range"):
+                    args = stmt_node.iter.args
+                    if len(args) == 1:
+                        try:
+                            total = ast.literal_eval(args[0])
+                            loop_fragments.append({
+                                "original": stmt_str,
+                                "var": stmt_node.target.id if isinstance(stmt_node.target, ast.Name) else "i",
+                                "start": 0,
+                                "end": total,
+                                "body": ast.get_source_segment(code, stmt_node) or stmt_str
+                            })
+                            continue
+                        except (ValueError, TypeError):
+                            pass
+                    elif len(args) >= 2:
+                        try:
+                            start = ast.literal_eval(args[0])
+                            end = ast.literal_eval(args[1])
+                            loop_fragments.append({
+                                "original": stmt_str,
+                                "var": stmt_node.target.id if isinstance(stmt_node.target, ast.Name) else "i",
+                                "start": start,
+                                "end": end,
+                                "body": stmt_str
+                            })
+                            continue
+                        except (ValueError, TypeError):
+                            pass
+        except Exception:
+            pass
+        other_statements.append(stmt_str)
+
+    frag_index = 1
+
+    # Split loops into sub-fragments
+    target_frags_per_loop = max(4, tier_info["vcpus"] * 2)  # More fragments for bigger tiers
+    for loop_info in loop_fragments:
+        total_iters = loop_info["end"] - loop_info["start"]
+        num_frags = min(target_frags_per_loop, max(1, total_iters))
+        chunk_size = math.ceil(total_iters / num_frags)
+
+        for i in range(num_frags):
+            chunk_start = loop_info["start"] + i * chunk_size
+            chunk_end = min(loop_info["start"] + (i + 1) * chunk_size, loop_info["end"])
+            if chunk_start >= loop_info["end"]:
+                break
+
+            # Rewrite the loop for this chunk
+            var = loop_info["var"]
+            # Extract loop body
+            try:
+                loop_tree = ast.parse(loop_info["original"])
+                loop_node = loop_tree.body[0]
+                body_source = ""
+                for child in loop_node.body:
+                    seg = ast.get_source_segment(loop_info["original"], child)
+                    if seg:
+                        body_source += f"    {seg}\n"
+                    else:
+                        body_source += f"    {ast.unparse(child)}\n"
+            except Exception:
+                body_source = "    pass\n"
+
+            chunk_code = f"""
+# Fragment {frag_index} — Loop chunk [{chunk_start}:{chunk_end})
+__fragment_results__ = []
+for {var} in range({chunk_start}, {chunk_end}):
+{body_source}
+"""
+            frag_id = f"{task_id}_frag_{frag_index}"
+            fragments.append({
+                "fragment_id": frag_id,
+                "task_id": task_id,
+                "fragment_index": frag_index,
+                "fragment_type": "compute",
+                "code": preamble + "\n" + chunk_code,
+                "input_data": json.dumps({
+                    "chunk_start": chunk_start,
+                    "chunk_end": chunk_end,
+                    "loop_var": var
+                }),
+            })
+            frag_index += 1
+
+    # Non-loop statements as individual fragments
+    if other_statements:
+        for stmt in other_statements:
+            frag_id = f"{task_id}_frag_{frag_index}"
+            fragments.append({
+                "fragment_id": frag_id,
+                "task_id": task_id,
+                "fragment_index": frag_index,
+                "fragment_type": "compute",
+                "code": preamble + "\n" + stmt,
+                "input_data": json.dumps({"type": "statement"}),
+            })
+            frag_index += 1
+
+    # If no compute fragments were created, make one with all statements
+    if frag_index == 1:
+        all_stmts = "\n".join(statements) if statements else "pass"
+        frag_id = f"{task_id}_frag_1"
+        fragments.append({
+            "fragment_id": frag_id,
+            "task_id": task_id,
+            "fragment_index": 1,
+            "fragment_type": "compute",
+            "code": preamble + "\n" + all_stmts,
+            "input_data": json.dumps({"type": "full"}),
+        })
+        frag_index = 2
+
+    # Aggregation fragment
+    agg_frag_id = f"{task_id}_agg"
+    fragments.append({
+        "fragment_id": agg_frag_id,
+        "task_id": task_id,
+        "fragment_index": 999999,
+        "fragment_type": "aggregate",
+        "code": "# Aggregation — collect all fragment results",
+        "input_data": json.dumps({"type": "aggregate", "total_compute_fragments": frag_index - 1}),
+    })
+
+    return fragments
+
+
+# ─── Task Submission ─────────────────────────────────────────────────────────
+@app.post("/api/submit_task")
+async def submit_task(task: TaskSubmit, background_tasks: BackgroundTasks):
+    # Verify API key
+    user_resp = await db_request("post", "/users/by_api_key", json={"api_key": task.api_key})
+    if user_resp.status_code != 200:
+        raise HTTPException(status_code=401, detail="Invalid API key")
+    user = user_resp.json()
+
+    # Check balance
+    tier_info = PRICING.get(task.tier, PRICING["basic_cpu"])
+    min_cost = tier_info["price_per_hour"] * 0.01  # Minimum charge: ~36 seconds
+    if user["balance"] < min_cost:
+        raise HTTPException(status_code=402, detail=f"Insufficient balance. Need at least {min_cost} SACCP")
+
+    # Create task in DB
+    task_resp = await db_request("post", "/tasks", json={
+        "owner": user["username"],
+        "code": task.code,
+        "tier": task.tier
+    })
+    if task_resp.status_code != 201:
+        raise HTTPException(status_code=500, detail="Failed to create task")
+
+    task_data = task_resp.json()
+    task_id = task_data["task_id"]
+
+    # Fragment the code
+    fragments = fragment_python_code(task.code, task.tier, task_id)
+
+    # Save fragments to DB
+    frag_resp = await db_request("post", "/fragments/batch", json={"fragments": fragments})
+    if frag_resp.status_code != 201:
+        raise HTTPException(status_code=500, detail="Failed to create fragments")
+
+    # Update task with fragment count
+    compute_frags = [f for f in fragments if f["fragment_type"] == "compute"]
+    await db_request("patch", f"/tasks/{task_id}", json={
+        "status": "fragmenting",
+        "total_fragments": len(compute_frags)
+    })
+
+    # Start processing in background
+    background_tasks.add_task(process_task, task_id)
+
+    return {
+        "task_id": task_id,
+        "status": "fragmenting",
+        "total_fragments": len(compute_frags),
+        "tier": task.tier,
+        "estimated_cost": f"{min_cost:.2f} - {tier_info['price_per_hour']:.2f} SACCP"
+    }
+
+
+async def process_task(task_id: str):
+    """Background task processor — marks task as running."""
+    await asyncio.sleep(1)  # Brief delay
+    await db_request("patch", f"/tasks/{task_id}", json={"status": "running"})
+
+    # Monitor loop: check if all fragments completed
+    max_wait = 3600  # 1 hour max
+    start = time.time()
+    while time.time() - start < max_wait:
+        await asyncio.sleep(5)
+
+        task_resp = await db_request("get", f"/tasks/{task_id}")
+        if task_resp.status_code != 200:
+            break
+        task = task_resp.json()
+        if task["status"] in ("completed", "failed"):
+            break
+
+        frags_resp = await db_request("get", f"/fragments/by_task/{task_id}")
+        if frags_resp.status_code != 200:
+            continue
+        frags = frags_resp.json()
+
+        compute_frags = [f for f in frags if f["fragment_type"] == "compute"]
+        completed = [f for f in compute_frags if f["status"] == "completed"]
+        failed = [f for f in compute_frags if f["status"] == "failed" and f.get("retries", 0) >= 3]
+
+        await db_request("patch", f"/tasks/{task_id}", json={
+            "completed_fragments": len(completed),
+            "failed_fragments": len(failed)
+        })
+
+        if len(completed) + len(failed) >= len(compute_frags):
+            # All done — assemble results
+            await assemble_results(task_id, frags)
+            break
+
+
+async def assemble_results(task_id: str, fragments: list):
+    """Piece together all fragment results into final output."""
+    compute_frags = [f for f in fragments if f["fragment_type"] == "compute"]
+    compute_frags.sort(key=lambda x: x["fragment_index"])
+
+    all_results = []
+    all_stdout = []
+    all_errors = []
+
+    for frag in compute_frags:
+        if frag["status"] == "completed":
+            if frag.get("result") is not None:
+                all_results.append(frag["result"])
+            if frag.get("stdout"):
+                all_stdout.append(f"[Fragment {frag['fragment_index']}]\n{frag['stdout']}")
+        elif frag["status"] == "failed":
+            all_errors.append(f"[Fragment {frag['fragment_index']}] {frag.get('error', 'Unknown error')}")
+
+    final_result = {
+        "fragments_total": len(compute_frags),
+        "fragments_succeeded": len([f for f in compute_frags if f["status"] == "completed"]),
+        "fragments_failed": len([f for f in compute_frags if f["status"] == "failed"]),
+        "results": all_results,
+        "stdout": "\n\n".join(all_stdout),
+        "errors": all_errors
+    }
+
+    status = "completed" if not all_errors else ("completed" if len(all_errors) < len(compute_frags) else "failed")
+
+    # Calculate cost
+    task_resp = await db_request("get", f"/tasks/{task_id}")
+    task = task_resp.json()
+    tier_info = PRICING.get(task.get("tier", "basic_cpu"), PRICING["basic_cpu"])
+
+    created = datetime.fromisoformat(task["created_at"])
+    elapsed_hours = max((datetime.utcnow() - created).total_seconds() / 3600, 0.01)
+    cost = round(tier_info["price_per_hour"] * elapsed_hours, 4)
+
+    # Charge user
+    await db_request("patch", f"/users/{task['owner']}/balance", json={
+        "amount": -cost,
+        "reason": f"task_{task_id}"
+    })
+
+    await db_request("patch", f"/tasks/{task_id}", json={
+        "status": status,
+        "result": json.dumps(final_result),
+        "cost": cost,
+        "completed_at": datetime.utcnow().isoformat(),
+        "error": "\n".join(all_errors) if all_errors else None
+    })
+
+
+# ─── Task Status ─────────────────────────────────────────────────────────────
+@app.get("/api/task/{task_id}")
+async def get_task_status(task_id: str):
+    resp = await db_request("get", f"/tasks/{task_id}")
+    if resp.status_code != 200:
+        raise HTTPException(status_code=404, detail="Task not found")
+    return resp.json()
+
+
+@app.get("/api/my_tasks")
+async def get_my_tasks(api_key: str):
+    user_resp = await db_request("post", "/users/by_api_key", json={"api_key": api_key})
+    if user_resp.status_code != 200:
+        raise HTTPException(status_code=401, detail="Invalid API key")
+    user = user_resp.json()
+
+    resp = await db_request("get", f"/tasks?owner={user['username']}")
+    return resp.json()
+
+
+@app.get("/api/task/{task_id}/fragments")
+async def get_task_fragments(task_id: str):
+    resp = await db_request("get", f"/fragments/by_task/{task_id}")
+    return resp.json()
+
+
+# ─── Node Management ────────────────────────────────────────────────────────
+@app.post("/api/register_node")
+async def register_node(node: NodeRegister):
+    node_id = f"node-{uuid.uuid4().hex[:12]}"
+    resp = await db_request("post", "/nodes", json={
+        "node_id": node_id,
+        "node_type": node.node_type,
+        "node_url": node.node_url,
+        "owner": node.owner,
+        "specs": node.specs
+    })
+    if resp.status_code != 201:
+        raise HTTPException(status_code=500, detail="Failed to register node")
+
+    active_nodes[node_id] = {
+        "node_id": node_id,
+        "node_type": node.node_type,
+        "node_url": node.node_url,
+        "last_heartbeat": time.time()
+    }
+
+    pay_rate = NODE_PAY_RATES.get(node.node_type, 0)
+    return {
+        "node_id": node_id,
+        "message": f"Node registered as {node.node_type}",
+        "pay_rate": f"{pay_rate} SACCP/hr"
+    }
+
+
+@app.post("/api/node_heartbeat")
+async def node_heartbeat(data: dict):
+    node_id = data.get("node_id", "")
+    if node_id in active_nodes:
+        active_nodes[node_id]["last_heartbeat"] = time.time()
+
+    await db_request("post", f"/nodes/{node_id}/heartbeat", json=data)
+    return {"message": "OK"}
+
+
+@app.get("/api/get_work")
+async def get_work(node_id: str):
+    """Worker nodes call this to get fragments to execute."""
+    # Get pending fragments
+    resp = await db_request("get", "/fragments/pending?limit=1")
+    if resp.status_code != 200:
+        return {"work": None}
+
+    pending = resp.json()
+    if not pending:
+        return {"work": None}
+
+    fragment = pending[0]
+
+    # Assign to this node
+    await db_request("patch", f"/fragments/{fragment['fragment_id']}", json={
+        "status": "assigned",
+        "assigned_node": node_id,
+        "started_at": datetime.utcnow().isoformat()
+    })
+
+    return {"work": fragment}
+
+
+@app.post("/api/submit_result")
+async def submit_result(result: FragmentResult):
+    """Worker nodes submit fragment results here."""
+    await db_request("patch", f"/fragments/{result.fragment_id}", json={
+        "status": result.status,
+        "result": result.result,
+        "error": result.error,
+        "stdout": result.stdout,
+        "completed_at": datetime.utcnow().isoformat()
+    })
+
+    # Pay the node
+    if result.status == "completed" and result.node_id:
+        node_info = active_nodes.get(result.node_id, {})
+        node_type = node_info.get("node_type", "CPU")
+        # Pay a small amount per fragment
+        pay_rate = NODE_PAY_RATES.get(node_type, 0)
+        fragment_pay = pay_rate / 360  # ~10 seconds of work
+        if fragment_pay > 0:
+            await db_request("post", f"/nodes/{result.node_id}/pay", json={"amount": fragment_pay})
+
+    return {"message": "Result recorded"}
+
+
+# ─── Stats & Pricing ──��─────────────────────────────────────────────────────
+@app.get("/api/stats")
+async def network_stats():
+    try:
+        resp = await db_request("get", "/stats")
+        return resp.json()
+    except Exception:
+        return {"status": "database unavailable"}
+
+
+@app.get("/api/pricing")
+async def get_pricing():
+    return PRICING
+
+
+@app.get("/api/node_pay_rates")
+async def get_node_pay_rates():
+    return NODE_PAY_RATES
+
+
+@app.get("/health")
+async def health():
+    return {
+        "status": "healthy",
+        "head_node_id": HEAD_NODE_ID,
+        "active_nodes_cached": len(active_nodes),
+        "timestamp": datetime.utcnow().isoformat()
+    }
+
+
+# ─── Background: Pay nodes periodically ────────────────────────────────────
+@app.on_event("startup")
+async def startup():
+    asyncio.create_task(node_payment_loop())
+    asyncio.create_task(node_cleanup_loop())
+
+
+async def node_payment_loop():
+    """Every 5 minutes, pay online nodes."""
+    while True:
+        await asyncio.sleep(300)  # 5 minutes
+        try:
+            resp = await db_request("get", "/nodes")
+            if resp.status_code == 200:
+                nodes = resp.json()
+                for node in nodes:
+                    if node["status"] == "online":
+                        rate = NODE_PAY_RATES.get(node["node_type"], 0)
+                        payment = rate * (5 / 60)  # 5 minutes worth
+                        if payment > 0:
+                            await db_request("post", f"/nodes/{node['node_id']}/pay",
+                                             json={"amount": round(payment, 4)})
+        except Exception:
+            pass
+
+
+async def node_cleanup_loop():
+    """Mark nodes as offline if no heartbeat in 2 minutes."""
+    while True:
+        await asyncio.sleep(60)
+        cutoff = time.time() - 120
+        dead = [nid for nid, n in active_nodes.items() if n["last_heartbeat"] < cutoff]
+        for nid in dead:
+            active_nodes.pop(nid, None)
+            try:
+                await db_request("post", f"/nodes/{nid}/heartbeat", json={"status": "offline"})
+            except Exception:
+                pass
+
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=7860)