File size: 5,461 Bytes
9e15418 a25490a 9e15418 a25490a 9e15418 a25490a 9e15418 a25490a 9e15418 a25490a 9e15418 a25490a 9e15418 a25490a 9e15418 a25490a 9e15418 a25490a 9e15418 a25490a 9e15418 a25490a 9e15418 a25490a 9e15418 a25490a 9e15418 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 | import asyncio
import json
import websockets
import requests
import uuid
import time
import urllib.parse
from reputation import ReputationManager
from identity import MEPIdentity
class ChronosNode:
"""
Simulated Clawdbot Client (Both Consumer & Provider)
"""
def __init__(self, key_path: str, hub_url: str = "http://localhost:8000", ws_url: str = "ws://localhost:8000"):
self.identity = MEPIdentity(key_path)
self.node_id = self.identity.node_id
self.hub_url = hub_url
self.ws_url = ws_url
self.reputation = ReputationManager(storage_path=f"reputation_{self.node_id}.json")
self.is_sleeping = False
# Track pending tasks we created (Consumer)
self.my_pending_tasks: dict[str, dict] = {}
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.identity.pub_pem, "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"]
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
payload_str = json.dumps({
"task_id": task_id,
"provider_id": self.node_id,
"result_payload": result
})
headers = self.identity.get_auth_headers(payload_str)
headers["Content-Type"] = "application/json"
resp = requests.post(f"{self.hub_url}/tasks/complete", data=payload_str, headers=headers)
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."""
ts = str(int(time.time()))
sig = self.identity.sign(self.node_id, ts)
sig_safe = urllib.parse.quote(sig)
uri = f"{self.ws_url}/ws/{self.node_id}?timestamp={ts}&signature={sig_safe}"
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."""
payload_str = json.dumps({
"consumer_id": self.node_id,
"payload": payload,
"bounty": bounty
})
headers = self.identity.get_auth_headers(payload_str)
headers["Content-Type"] = "application/json"
resp = requests.post(f"{self.hub_url}/tasks/submit", data=payload_str, headers=headers)
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(f"usa_node_{uuid.uuid4().hex[:6]}.pem")
usa_node.is_sleeping = False
usa_node.register()
asia_node = ChronosNode(f"asia_node_{uuid.uuid4().hex[:6]}.pem")
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())
|