Add L2 client and Reputation logic

hub/main.py

@@ -1,13 +1,12 @@
 from fastapi import FastAPI, HTTPException, WebSocket, WebSocketDisconnect
 from typing import Dict, List
 import uuid
-import json
 
 from models import NodeRegistration, TaskCreate, TaskResult, NodeBalance
 
-app = FastAPI(title="Chronos Protocol L1 Hub", description="The Time Exchange Clearinghouse", version="0.1.0")
+app = FastAPI(title="Chronos Protocol L1 Hub", description="The Time Exchange Clearinghouse", version="0.1.1")
 
-# In-memory storage for MVP (Use Postgres/Redis in prod)
+# In-memory storage for MVP
 ledger: Dict[str, float] = {}  # node_id -> balance
 active_tasks: Dict[str, dict] = {} # task_id -> task_details
 completed_tasks: Dict[str, dict] = {} # task_id -> result
@@ -15,28 +14,23 @@ connected_nodes: Dict[str, WebSocket] = {} # node_id -> websocket
 
 @app.post("/register")
 async def register_node(node: NodeRegistration):
-    """Register a new node and initialize its SECONDS balance."""
     if node.pubkey not in ledger:
-        # Give a 10 SECOND starter bonus for the MVP
-        ledger[node.pubkey] = 10.0 
+        ledger[node.pubkey] = 10.0 # Starter bonus
     return {"status": "success", "node_id": node.pubkey, "balance": ledger[node.pubkey]}
 
 @app.get("/balance/{node_id}")
 async def get_balance(node_id: str):
-    """Check a node's SECONDS balance."""
     if node_id not in ledger:
         raise HTTPException(status_code=404, detail="Node not found")
     return {"node_id": node_id, "balance_seconds": ledger[node_id]}
 
 @app.post("/tasks/submit")
 async def submit_task(task: TaskCreate):
-    """Consumer submits a task, locking their SECONDS."""
     if task.consumer_id not in ledger:
         raise HTTPException(status_code=404, detail="Consumer node not found")
     if ledger[task.consumer_id] < task.bounty:
         raise HTTPException(status_code=400, detail="Insufficient SECONDS balance")
 
-    # Deduct bounty
     ledger[task.consumer_id] -= task.bounty
     
     task_id = str(uuid.uuid4())
@@ -49,19 +43,18 @@ async def submit_task(task: TaskCreate):
     }
     active_tasks[task_id] = task_data
     
-    # Broadcast to connected sleeping nodes (Providers)
-    for node_id, ws in connected_nodes.items():
+    # Broadcast to all connected nodes EXCEPT the consumer
+    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})
             except:
-                pass # Disconnected
+                pass
                 
     return {"status": "success", "task_id": task_id}
 
 @app.post("/tasks/complete")
 async def complete_task(result: TaskResult):
-    """Provider submits the result and claims the bounty."""
     task = active_tasks.get(result.task_id)
     if not task:
         raise HTTPException(status_code=404, detail="Task not found or already claimed")
@@ -79,20 +72,31 @@ async def complete_task(result: TaskResult):
     completed_tasks[result.task_id] = task
     del active_tasks[result.task_id]
     
+    # ROUTE RESULT BACK TO CONSUMER VIA WEBSOCKET
+    consumer_id = task["consumer_id"]
+    if consumer_id in connected_nodes:
+        try:
+            await connected_nodes[consumer_id].send_json({
+                "event": "task_result",
+                "data": {
+                    "task_id": result.task_id,
+                    "provider_id": result.provider_id,
+                    "result_payload": result.result_payload,
+                    "bounty_spent": task["bounty"]
+                }
+            })
+        except:
+            pass # Consumer disconnected, they can fetch it via REST later (TODO)
+
     return {"status": "success", "earned": task["bounty"], "new_balance": ledger[result.provider_id]}
 
 @app.websocket("/ws/{node_id}")
 async def websocket_endpoint(websocket: WebSocket, node_id: str):
-    """WebSocket connection for real-time task broadcasting (Sleeping APIs listen here)."""
     await websocket.accept()
     connected_nodes[node_id] = websocket
     try:
         while True:
-            # Ping/Pong to keep alive
             data = await websocket.receive_text()
     except WebSocketDisconnect:
-        del connected_nodes[node_id]
-
-if __name__ == "__main__":
-    import uvicorn
-    uvicorn.run(app, host="0.0.0.0", port=8000)
+        if node_id in connected_nodes:
+            del connected_nodes[node_id]

node/client.py

@@ -0,0 +1,129 @@
+import asyncio
+import json
+import websockets
+import requests
+import uuid
+import sys
+from reputation import ReputationManager
+
+class ChronosNode:
+    """
+    Simulated Clawdbot Client (Both Consumer & Provider)
+    """
+    def __init__(self, node_id: str, hub_url: str = "http://localhost:8000", ws_url: str = "ws://localhost:8000"):
+        self.node_id = node_id
+        self.hub_url = hub_url
+        self.ws_url = ws_url
+        self.reputation = ReputationManager(storage_path=f"reputation_{node_id}.json")
+        self.is_sleeping = False
+        
+        # Track pending tasks we created (Consumer)
+        self.my_pending_tasks = {}
+
+    def register(self):
+        """Register to get 10 SECONDS."""
+        print(f"[Node {self.node_id}] Registering with Hub...")
+        resp = requests.post(f"{self.hub_url}/register", json={"pubkey": self.node_id, "alias": "test"})
+        data = resp.json()
+        print(f"[Node {self.node_id}] Balance: {data['balance']}s")
+
+    async def _handle_new_task(self, task_data: dict):
+        """As a Provider (Sleeping), evaluate and execute the broadcasted task."""
+        if not self.is_sleeping:
+            return  # Awake nodes don't mine
+
+        task_id = task_data["id"]
+        payload = task_data["payload"]
+        bounty = task_data["bounty"]
+        consumer_id = task_data["consumer_id"]
+
+        print(f"[Node {self.node_id}] Broadcast received: Task {task_id[:6]} for {bounty}s")
+        
+        # 1. Check L2 Reputation of Consumer (Don't work for bad nodes)
+        # (Mock implementation, skipping for brevity)
+        
+        # 2. Local execution using user's API keys (Mocked)
+        await asyncio.sleep(1) # simulate think time
+        result = f"Hello from {self.node_id}. I processed your payload: {payload[:20]}..."
+        
+        # 3. Submit proof of work
+        resp = requests.post(f"{self.hub_url}/tasks/complete", json={
+            "task_id": task_id,
+            "provider_id": self.node_id,
+            "result_payload": result
+        })
+        if resp.status_code == 200:
+            print(f"[Node {self.node_id}] Mined {bounty}s! New Balance: {resp.json()['new_balance']}s")
+
+    async def _handle_task_result(self, result_data: dict):
+        """As a Consumer, receive the result and update Reputation."""
+        task_id = result_data["task_id"]
+        provider_id = result_data["provider_id"]
+        result = result_data["result_payload"]
+        
+        print(f"\n[Node {self.node_id} (Consumer)] Result received for {task_id[:6]}")
+        print(f" -> Provider: {provider_id}")
+        print(f" -> Result: {result}")
+        
+        # L2 Reputation Evaluation
+        score = self.reputation.evaluate_result(provider_id, result)
+        print(f" -> Provider {provider_id} rated {score:.2f}/1.00 based on this result.\n")
+
+    async def listen(self):
+        """Persistent WebSocket connection."""
+        uri = f"{self.ws_url}/ws/{self.node_id}"
+        async with websockets.connect(uri) as ws:
+            print(f"[Node {self.node_id}] Connected to Hub via WebSocket.")
+            while True:
+                msg = await ws.recv()
+                data = json.loads(msg)
+                
+                if data["event"] == "new_task":
+                    asyncio.create_task(self._handle_new_task(data["data"]))
+                elif data["event"] == "task_result":
+                    asyncio.create_task(self._handle_task_result(data["data"]))
+
+    async def submit_task(self, payload: str, bounty: float):
+        """As a Consumer, create a task and lock SECONDS."""
+        resp = requests.post(f"{self.hub_url}/tasks/submit", json={
+            "consumer_id": self.node_id,
+            "payload": payload,
+            "bounty": bounty
+        })
+        if resp.status_code == 200:
+            task_id = resp.json()["task_id"]
+            print(f"[Node {self.node_id} (Consumer)] Submitted Task {task_id[:6]} for {bounty}s")
+        else:
+            print(f"Failed to submit task: {resp.text}")
+
+async def run_demo():
+    # Setup two nodes
+    usa_node = ChronosNode("usa_node")
+    usa_node.is_sleeping = False
+    usa_node.register()
+    
+    asia_node = ChronosNode("asia_node")
+    asia_node.is_sleeping = True # Asia goes to sleep and mines
+    asia_node.register()
+
+    # Start listening
+    tasks = [
+        asyncio.create_task(usa_node.listen()),
+        asyncio.create_task(asia_node.listen())
+    ]
+    
+    # Wait for connections
+    await asyncio.sleep(1)
+
+    # USA user creates a task
+    await usa_node.submit_task("Write a long chapter about the Sleeping API", bounty=5.0)
+
+    # Let the simulation run for a few seconds
+    await asyncio.sleep(3)
+    
+    # Stop tasks
+    for t in tasks:
+        t.cancel()
+
+if __name__ == "__main__":
+    asyncio.run(run_demo())

node/reputation.py

@@ -0,0 +1,60 @@
+import json
+import os
+from typing import Dict, Any
+
+class ReputationManager:
+    """
+    L2 Reputation Tracker: Stored locally on the user's Clawdbot.
+    Tracks which Provider Nodes are good, and which are 'slacking' (磨洋工).
+    """
+    def __init__(self, storage_path="reputation.json"):
+        self.storage_path = storage_path
+        self.scores: Dict[str, Dict[str, Any]] = self._load()
+        
+    def _load(self) -> Dict[str, Dict[str, Any]]:
+        if os.path.exists(self.storage_path):
+            with open(self.storage_path, 'r') as f:
+                return json.load(f)
+        return {}
+        
+    def _save(self):
+        with open(self.storage_path, 'w') as f:
+            json.dump(self.scores, f, indent=2)
+
+    def get_score(self, provider_id: str) -> float:
+        """Returns score between 0.0 (Worst) and 1.0 (Best). Default is 0.5."""
+        return self.scores.get(provider_id, {}).get("score", 0.5)
+
+    def evaluate_result(self, provider_id: str, result_payload: str) -> float:
+        """
+        Mock evaluation logic. 
+        In production, this could be:
+        1. Simple length/keyword check.
+        2. A lightweight local LLM verification (e.g., Llama.cpp).
+        """
+        quality = 0.5
+        
+        # Extremely basic heuristics for the prototype
+        if len(result_payload.strip()) == 0:
+            quality = 0.0  # Empty response = slacking!
+        elif "Error" in result_payload or "Failed" in result_payload:
+            quality = 0.2  # Bad response
+        elif len(result_payload) > 50:
+            quality = 0.9  # Looks like a solid response!
+            
+        return self.update_score(provider_id, quality)
+
+    def update_score(self, provider_id: str, quality_score: float) -> float:
+        """Moving average reputation."""
+        if provider_id not in self.scores:
+            self.scores[provider_id] = {"score": 0.5, "tasks": 0}
+            
+        current = self.scores[provider_id]
+        total_score = current["score"] * current["tasks"]
+        
+        current["tasks"] += 1
+        current["score"] = (total_score + quality_score) / current["tasks"]
+        
+        self._save()
+        print(f"[Reputation] Node {provider_id} updated. New Score: {current['score']:.2f} (Total Tasks: {current['tasks']})")
+        return current["score"]

node/reputation_usa_node.json

@@ -0,0 +1,6 @@
+{
+  "asia_node": {
+    "score": 0.9,
+    "tasks": 1
+  }
+}

node/requirements.txt

@@ -0,0 +1,3 @@
+websockets
+requests
+aiohttp