Upgrade to Phase 2 (Auction Logic) to save API costs

hub/main.py

@@ -2,7 +2,7 @@ from fastapi import FastAPI, HTTPException, WebSocket, WebSocketDisconnect
 from typing import Dict, List
 import uuid
 
-from models import NodeRegistration, TaskCreate, TaskResult, NodeBalance
+from models import NodeRegistration, TaskCreate, TaskResult, NodeBalance, TaskBid
 
 app = FastAPI(title="Chronos Protocol L1 Hub", description="The Time Exchange Clearinghouse", version="0.1.1")
 
@@ -39,15 +39,18 @@ async def submit_task(task: TaskCreate):
         "consumer_id": task.consumer_id,
         "payload": task.payload,
         "bounty": task.bounty,
-        "status": "pending",
-        "target_node": task.target_node
+        "status": "bidding",
+        "target_node": task.target_node,
+        "model_requirement": task.model_requirement
     }
     active_tasks[task_id] = task_data
     
-    # Target specific node if requested (Direct Message)
+    # Target specific node if requested (Direct Message skips bidding)
     if task.target_node:
         if task.target_node in connected_nodes:
             try:
+                task_data["status"] = "assigned"
+                task_data["provider_id"] = task.target_node
                 await connected_nodes[task.target_node].send_json({"event": "new_task", "data": task_data})
                 return {"status": "success", "task_id": task_id, "routed_to": task.target_node}
             except:
@@ -55,16 +58,43 @@ async def submit_task(task: TaskCreate):
         else:
             return {"status": "error", "detail": "Target node not currently connected to Hub"}
 
-    # Otherwise, Broadcast to all connected nodes EXCEPT the consumer
+    # Phase 2: Broadcast RFC (Request For Compute) to all connected nodes EXCEPT the consumer
+    rfc_data = {
+        "id": task_id,
+        "consumer_id": task.consumer_id,
+        "bounty": task.bounty,
+        "model_requirement": task.model_requirement
+    }
     for node_id, ws in list(connected_nodes.items()):
         if node_id != task.consumer_id:
             try:
-                await ws.send_json({"event": "new_task", "data": task_data})
+                await ws.send_json({"event": "rfc", "data": rfc_data})
             except:
                 pass
                 
     return {"status": "success", "task_id": task_id}
 
+@app.post("/tasks/bid")
+async def place_bid(bid: TaskBid):
+    task = active_tasks.get(bid.task_id)
+    if not task:
+        raise HTTPException(status_code=404, detail="Task not found or already completed")
+        
+    if task["status"] != "bidding":
+        return {"status": "rejected", "detail": "Task already assigned to another node"}
+        
+    # Phase 2 Fast Auction: Accept the first valid bid
+    task["status"] = "assigned"
+    task["provider_id"] = bid.provider_id
+    
+    # Return the full payload to the winner
+    return {
+        "status": "accepted",
+        "payload": task["payload"],
+        "consumer_id": task["consumer_id"],
+        "model_requirement": task.get("model_requirement")
+    }
+
 @app.post("/tasks/complete")
 async def complete_task(result: TaskResult):
     task = active_tasks.get(result.task_id)

hub/models.py

@@ -11,6 +11,11 @@ class TaskCreate(BaseModel):
     payload: str
     bounty: float
     target_node: Optional[str] = None  # Direct messaging / specific bot targeting
+    model_requirement: Optional[str] = None
+
+class TaskBid(BaseModel):
+    task_id: str
+    provider_id: str
 
 class TaskResult(BaseModel):
     task_id: str

node/mep_miner.py

@@ -51,6 +51,8 @@ class MEPMiner:
                         
                         if data["event"] == "new_task":
                             await self.process_task(data["data"])
+                        elif data["event"] == "rfc":
+                            await self.handle_rfc(data["data"])
                             
                     except asyncio.TimeoutError:
                         continue  # Keep connection alive
@@ -61,6 +63,44 @@ class MEPMiner:
         except Exception as e:
             print(f"[MEP Miner {self.node_id}] WebSocket error: {e}")
     
+    async def handle_rfc(self, rfc_data: dict):
+        """Phase 2: Evaluate Request For Compute and submit Bid."""
+        task_id = rfc_data["id"]
+        bounty = rfc_data["bounty"]
+        model = rfc_data.get("model_requirement")
+        
+        # Simple evaluation logic
+        if bounty < 0.1 and bounty != 0.0:  # Allow 0 for testing DM
+            print(f"[MEP Miner {self.node_id}] Ignored RFC {task_id[:8]} (Bounty too low: {bounty})")
+            return
+            
+        print(f"[MEP Miner {self.node_id}] Received RFC {task_id[:8]} for {bounty:.6f} SECONDS. Placing bid...")
+        
+        # Place bid
+        try:
+            resp = requests.post(f"{HUB_URL}/tasks/bid", json={
+                "task_id": task_id,
+                "provider_id": self.node_id
+            })
+            
+            if resp.status_code == 200:
+                data = resp.json()
+                if data["status"] == "accepted":
+                    print(f"[MEP Miner {self.node_id}] 🏁 BID WON for task {task_id[:8]}! Processing payload...")
+                    
+                    # Reconstruct task_data to pass to process_task
+                    task_data = {
+                        "id": task_id,
+                        "payload": data["payload"],
+                        "bounty": bounty,
+                        "consumer_id": data["consumer_id"]
+                    }
+                    await self.process_task(task_data)
+                else:
+                    print(f"[MEP Miner {self.node_id}] Bid rejected (too slow): {data.get('detail', '')}")
+        except Exception as e:
+            print(f"[MEP Miner {self.node_id}] Error placing bid: {e}")
+
     async def process_task(self, task_data: dict):
         """Process a task and earn SECONDS."""
         task_id = task_data["id"]

node/test_auction.py

@@ -0,0 +1,34 @@
+import asyncio
+import websockets
+import json
+import requests
+import uuid
+
+HUB_URL = "http://localhost:8000"
+
+async def test():
+    miner = f'mep-miner-{uuid.uuid4().hex[:6]}'
+    requests.post(f'{HUB_URL}/register', json={'pubkey': miner})
+    async with websockets.connect(f'ws://localhost:8000/ws/{miner}') as ws:
+        consumer = 'test-consumer'
+        requests.post(f'{HUB_URL}/register', json={'pubkey': consumer})
+        requests.post(f'{HUB_URL}/tasks/submit', json={'consumer_id': consumer, 'payload': 'Test payload', 'bounty': 1.0})
+        
+        msg = await asyncio.wait_for(ws.recv(), timeout=2.0)
+        data = json.loads(msg)
+        print('Received:', data)
+        
+        if data['event'] == 'rfc':
+            task_id = data['data']['id']
+            resp = requests.post(f'{HUB_URL}/tasks/bid', json={'task_id': task_id, 'provider_id': miner})
+            print('Bid response:', resp.json())
+            
+            complete_resp = requests.post(f'{HUB_URL}/tasks/complete', json={
+                'task_id': task_id, 
+                'provider_id': miner, 
+                'result_payload': 'Done!'
+            })
+            print('Complete response:', complete_resp.json())
+
+if __name__ == '__main__':
+    asyncio.run(test())