Add Direct Messaging and Zero-Bounty features

hub/main.py

@@ -39,11 +39,23 @@ async def submit_task(task: TaskCreate):
         "consumer_id": task.consumer_id,
         "payload": task.payload,
         "bounty": task.bounty,
-        "status": "pending"
+        "status": "pending",
+        "target_node": task.target_node
     }
     active_tasks[task_id] = task_data
     
-    # Broadcast to all connected nodes EXCEPT the consumer
+    # Target specific node if requested (Direct Message)
+    if task.target_node:
+        if task.target_node in connected_nodes:
+            try:
+                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:
+                return {"status": "error", "detail": "Target node disconnected"}
+        else:
+            return {"status": "error", "detail": "Target node not currently connected to Hub"}
+
+    # Otherwise, Broadcast to all connected nodes EXCEPT the consumer
     for node_id, ws in list(connected_nodes.items()):
         if node_id != task.consumer_id:
             try:

hub/models.py

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

node/test_dm.py

@@ -0,0 +1,63 @@
+import asyncio
+import websockets
+import json
+import requests
+import uuid
+
+HUB_URL = "http://localhost:8000"
+WS_URL = "ws://localhost:8000"
+
+async def test_direct_message():
+    print("=== Testing MEP Direct Messaging (Zero Bounty) ===")
+    
+    # 1. Start Alice (Provider)
+    alice_id = "alice-specialist-88"
+    requests.post(f"{HUB_URL}/register", json={"pubkey": alice_id})
+    
+    # 2. Start Bob (Consumer)
+    bob_id = "bob-general-12"
+    requests.post(f"{HUB_URL}/register", json={"pubkey": bob_id})
+    
+    print(f"✅ Registered Alice ({alice_id}) and Bob ({bob_id})")
+    
+    async def alice_listen():
+        async with websockets.connect(f"{WS_URL}/ws/{alice_id}") as ws:
+            print("👧 Alice: Online and listening...")
+            msg = await asyncio.wait_for(ws.recv(), timeout=5)
+            data = json.loads(msg)
+            
+            print(f"👧 Alice: Received DIRECT MESSAGE!")
+            print(f"👧 Alice: Payload: {data['data']['payload']}")
+            print(f"👧 Alice: Bounty: {data['data']['bounty']} SECONDS")
+            
+            # Alice replies for free
+            requests.post(f"{HUB_URL}/tasks/complete", json={
+                "task_id": data['data']['id'],
+                "provider_id": alice_id,
+                "result_payload": "Yes Bob, I am available for a meeting tomorrow at 2 PM. Free of charge! 🐱"
+            })
+            print("👧 Alice: Sent reply!")
+
+    async def bob_listen():
+        async with websockets.connect(f"{WS_URL}/ws/{bob_id}") as ws:
+            # Bob submits a direct task to Alice with 0 bounty
+            await asyncio.sleep(1) # Let Alice connect first
+            print("👦 Bob: Sending Direct Message to Alice (0.0 SECONDS)...")
+            requests.post(f"{HUB_URL}/tasks/submit", json={
+                "consumer_id": bob_id,
+                "payload": "Hey Alice, are you free for a meeting tomorrow at 2 PM?",
+                "bounty": 0.0,
+                "target_node": alice_id
+            })
+            
+            # Bob waits for Alice's reply
+            msg = await asyncio.wait_for(ws.recv(), timeout=5)
+            data = json.loads(msg)
+            print(f"👦 Bob: Received reply from {data['data']['provider_id']}:")
+            print(f"👦 Bob: \"{data['data']['result_payload']}\"")
+
+    await asyncio.gather(alice_listen(), bob_listen())
+    print("=== Direct Messaging Test Complete! ===")
+
+if __name__ == "__main__":
+    asyncio.run(test_direct_message())