Add Three Markets integration test to verify economy logic

hub/logs/hub.json

@@ -0,0 +1,14 @@
+{"timestamp": "2026-02-24T07:15:32.589565+00:00", "level": "INFO", "logger": "mep.hub", "message": "Node node_7c115a964de4 registered with starting balance 10.0", "event": "node_registered", "node_id": "node_7c115a964de4", "starting_balance": 10.0}
+{"timestamp": "2026-02-24T07:15:32.603735+00:00", "level": "INFO", "logger": "mep.hub", "message": "Node node_31a01d787f88 registered with starting balance 10.0", "event": "node_registered", "node_id": "node_31a01d787f88", "starting_balance": 10.0}
+{"timestamp": "2026-02-24T07:16:06.691635+00:00", "level": "INFO", "logger": "mep.hub", "message": "Node node_7c115a964de4 registered with starting balance 10.0", "event": "node_registered", "node_id": "node_7c115a964de4", "starting_balance": 10.0}
+{"timestamp": "2026-02-24T07:16:06.711586+00:00", "level": "INFO", "logger": "mep.hub", "message": "Node node_31a01d787f88 registered with starting balance 10.0", "event": "node_registered", "node_id": "node_31a01d787f88", "starting_balance": 10.0}
+{"timestamp": "2026-02-24T07:16:23.994625+00:00", "level": "INFO", "logger": "mep.hub", "message": "Node node_7c115a964de4 registered with starting balance 10.0", "event": "node_registered", "node_id": "node_7c115a964de4", "starting_balance": 10.0}
+{"timestamp": "2026-02-24T07:16:24.007658+00:00", "level": "INFO", "logger": "mep.hub", "message": "Node node_31a01d787f88 registered with starting balance 10.0", "event": "node_registered", "node_id": "node_31a01d787f88", "starting_balance": 10.0}
+{"timestamp": "2026-02-24T07:16:24.573446+00:00", "level": "INFO", "logger": "mep.hub", "message": "Task c9d51463 broadcasted by node_7c115a964de4 for 5.0", "event": "task_submitted", "consumer_id": "node_7c115a964de4", "task_id": "c9d51463-1d89-4aa0-9899-be5090bc4edf", "bounty": 5.0}
+{"timestamp": "2026-02-24T07:16:24.582713+00:00", "level": "INFO", "logger": "mep.hub", "message": "Task c9d51463 assigned to node_31a01d787f88", "event": "bid_accepted", "task_id": "c9d51463-1d89-4aa0-9899-be5090bc4edf", "provider_id": "node_31a01d787f88", "bounty": 5.0}
+{"timestamp": "2026-02-24T07:16:24.598410+00:00", "level": "INFO", "logger": "mep.hub", "message": "Task c9d51463 completed by node_31a01d787f88", "event": "task_completed", "task_id": "c9d51463-1d89-4aa0-9899-be5090bc4edf", "provider_id": "node_31a01d787f88", "bounty": 5.0}
+{"timestamp": "2026-02-24T07:16:24.608789+00:00", "level": "INFO", "logger": "mep.hub", "message": "Task 4f2e4436 broadcasted by node_7c115a964de4 for 0.0", "event": "task_submitted", "consumer_id": "node_7c115a964de4", "task_id": "4f2e4436-abff-4c7e-8777-2b8326c824ed", "bounty": 0.0}
+{"timestamp": "2026-02-24T07:16:24.617370+00:00", "level": "INFO", "logger": "mep.hub", "message": "Task 4f2e4436 completed by node_31a01d787f88", "event": "task_completed", "task_id": "4f2e4436-abff-4c7e-8777-2b8326c824ed", "provider_id": "node_31a01d787f88", "bounty": 0.0}
+{"timestamp": "2026-02-24T07:16:24.625135+00:00", "level": "INFO", "logger": "mep.hub", "message": "Task b9c676c9 broadcasted by node_7c115a964de4 for -2.0", "event": "task_submitted", "consumer_id": "node_7c115a964de4", "task_id": "b9c676c9-cf9f-4ebf-a54e-b96b9c49cd35", "bounty": -2.0}
+{"timestamp": "2026-02-24T07:16:24.632191+00:00", "level": "INFO", "logger": "mep.hub", "message": "Task b9c676c9 assigned to node_31a01d787f88", "event": "bid_accepted", "task_id": "b9c676c9-cf9f-4ebf-a54e-b96b9c49cd35", "provider_id": "node_31a01d787f88", "bounty": -2.0}
+{"timestamp": "2026-02-24T07:16:24.651742+00:00", "level": "INFO", "logger": "mep.hub", "message": "Task b9c676c9 completed by node_31a01d787f88", "event": "task_completed", "task_id": "b9c676c9-cf9f-4ebf-a54e-b96b9c49cd35", "provider_id": "node_31a01d787f88", "bounty": -2.0}

hub/logs/ledger_audit.log

@@ -0,0 +1,11 @@
+2026-02-24 07:15:32,590 - mep.audit - INFO - AUDIT | REGISTER | Node: node_7c115a964de4 | Amount: +10.000000 | Balance: 10.000000 | Ref: START_BONUS
+2026-02-24 07:15:32,604 - mep.audit - INFO - AUDIT | REGISTER | Node: node_31a01d787f88 | Amount: +10.000000 | Balance: 10.000000 | Ref: START_BONUS
+2026-02-24 07:16:06,692 - mep.audit - INFO - AUDIT | REGISTER | Node: node_7c115a964de4 | Amount: +10.000000 | Balance: 10.000000 | Ref: START_BONUS
+2026-02-24 07:16:06,712 - mep.audit - INFO - AUDIT | REGISTER | Node: node_31a01d787f88 | Amount: +10.000000 | Balance: 10.000000 | Ref: START_BONUS
+2026-02-24 07:16:23,995 - mep.audit - INFO - AUDIT | REGISTER | Node: node_7c115a964de4 | Amount: +10.000000 | Balance: 10.000000 | Ref: START_BONUS
+2026-02-24 07:16:24,008 - mep.audit - INFO - AUDIT | REGISTER | Node: node_31a01d787f88 | Amount: +10.000000 | Balance: 10.000000 | Ref: START_BONUS
+2026-02-24 07:16:24,573 - mep.audit - INFO - AUDIT | ESCROW | Node: node_7c115a964de4 | Amount: -5.000000 | Balance: 5.000000 | Ref: c9d51463-1d89-4aa0-9899-be5090bc4edf
+2026-02-24 07:16:24,598 - mep.audit - INFO - AUDIT | EARN_COMPUTE | Node: node_31a01d787f88 | Amount: +5.000000 | Balance: 15.000000 | Ref: c9d51463-1d89-4aa0-9899-be5090bc4edf
+2026-02-24 07:16:24,617 - mep.audit - INFO - AUDIT | EARN_COMPUTE | Node: node_31a01d787f88 | Amount: +0.000000 | Balance: 15.000000 | Ref: 4f2e4436-abff-4c7e-8777-2b8326c824ed
+2026-02-24 07:16:24,645 - mep.audit - INFO - AUDIT | BUY_DATA | Node: node_31a01d787f88 | Amount: -2.000000 | Balance: 13.000000 | Ref: b9c676c9-cf9f-4ebf-a54e-b96b9c49cd35
+2026-02-24 07:16:24,651 - mep.audit - INFO - AUDIT | SELL_DATA | Node: node_7c115a964de4 | Amount: +2.000000 | Balance: 7.000000 | Ref: b9c676c9-cf9f-4ebf-a54e-b96b9c49cd35

hub/main.py

@@ -23,14 +23,14 @@ async def verify_request(
 ) -> str:
     body = await request.body()
     payload_str = body.decode('utf-8')
-    
+
     pub_pem = db.get_pub_pem(x_mep_nodeid)
     if not pub_pem:
         raise HTTPException(status_code=401, detail="Unknown Node ID. Please register first.")
-        
+
     if not auth.verify_signature(pub_pem, payload_str, x_mep_timestamp, x_mep_signature):
         raise HTTPException(status_code=401, detail="Invalid cryptographic signature.")
-        
+
     return x_mep_nodeid
 
 @app.post("/register")
@@ -38,10 +38,10 @@ async def register_node(node: NodeRegistration):
     # Registration derives the Node ID from the provided Public Key PEM
     node_id = auth.derive_node_id(node.pubkey)
     balance = db.register_node(node_id, node.pubkey)
-    
+
     log_event("node_registered", f"Node {node_id} registered with starting balance {balance}", node_id=node_id, starting_balance=balance)
     log_audit("REGISTER", node_id, balance, balance, "START_BONUS")
-    
+
     return {"status": "success", "node_id": node_id, "balance": balance}
 
 @app.get("/balance/{node_id}")
@@ -58,15 +58,17 @@ async def submit_task(task: TaskCreate, authenticated_node: str = Depends(verify
     # Verify the signer is actually the consumer claiming to submit the task
     if authenticated_node != task.consumer_id:
         raise HTTPException(status_code=403, detail="Cannot submit tasks on behalf of another node")
-        
+
     consumer_balance = db.get_balance(task.consumer_id)
     if consumer_balance is None:
         raise HTTPException(status_code=404, detail="Consumer node not found")
-        
+
     # If bounty is positive, consumer is PAYING. Check consumer balance.
     if task.bounty > 0 and consumer_balance < task.bounty:
         raise HTTPException(status_code=400, detail="Insufficient SECONDS balance to pay for task")
-        
+
+    task_id = str(uuid.uuid4())
+
     # Note: If bounty is negative, consumer is SELLING data. We don't deduct here.
     # We will deduct from the provider when they complete the task.
     if task.bounty > 0:
@@ -79,8 +81,7 @@ async def submit_task(task: TaskCreate, authenticated_node: str = Depends(verify
         log_audit("ESCROW", task.consumer_id, -task.bounty, new_balance, task_id)
         
     log_event("task_submitted", f"Task {task_id[:8]} broadcasted by {task.consumer_id} for {task.bounty}", consumer_id=task.consumer_id, task_id=task_id, bounty=task.bounty)
-        
-    task_id = str(uuid.uuid4())
+
     task_data = {
         "id": task_id,
         "consumer_id": task.consumer_id,
@@ -91,7 +92,7 @@ async def submit_task(task: TaskCreate, authenticated_node: str = Depends(verify
         "model_requirement": task.model_requirement
     }
     active_tasks[task_id] = task_data
-    
+
     # Target specific node if requested (Direct Message skips bidding)
     if task.target_node:
         if task.target_node in connected_nodes:
@@ -118,27 +119,27 @@ async def submit_task(task: TaskCreate, authenticated_node: str = Depends(verify
                 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, authenticated_node: str = Depends(verify_request)):
     if authenticated_node != bid.provider_id:
         raise HTTPException(status_code=403, detail="Cannot bid on behalf of another node")
-        
+
     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
-    
+
     log_event("bid_accepted", f"Task {bid.task_id[:8]} assigned to {bid.provider_id}", task_id=bid.task_id, provider_id=bid.provider_id, bounty=task["bounty"])
-    
+
     # Return the full payload to the winner
     return {
         "status": "accepted",
@@ -151,11 +152,11 @@ async def place_bid(bid: TaskBid, authenticated_node: str = Depends(verify_reque
 async def complete_task(result: TaskResult, authenticated_node: str = Depends(verify_request)):
     if authenticated_node != result.provider_id:
         raise HTTPException(status_code=403, detail="Cannot complete tasks on behalf of another node")
-        
+
     task = active_tasks.get(result.task_id)
     if not task:
         raise HTTPException(status_code=404, detail="Task not found or already claimed")
-        
+
     provider_balance = db.get_balance(result.provider_id)
     if provider_balance is None:
         db.set_balance(result.provider_id, 0.0)
@@ -174,14 +175,14 @@ async def complete_task(result: TaskResult, authenticated_node: str = Depends(ve
         if not success:
             log_event("data_purchase_failed", f"Provider {result.provider_id} lacks SECONDS to buy {result.task_id}", task_id=result.task_id, provider_id=result.provider_id, cost=cost)
             raise HTTPException(status_code=400, detail="Provider lacks SECONDS to buy this data")
-        
+
         p_balance = db.get_balance(result.provider_id)
         log_audit("BUY_DATA", result.provider_id, -cost, p_balance, result.task_id)
-        
+
         db.add_balance(task["consumer_id"], cost) # The sender earns SECONDS
         c_balance = db.get_balance(task["consumer_id"])
         log_audit("SELL_DATA", task["consumer_id"], cost, c_balance, result.task_id)
-    
+
     log_event("task_completed", f"Task {result.task_id[:8]} completed by {result.provider_id}", task_id=result.task_id, provider_id=result.provider_id, bounty=bounty)
 
     # Move task to completed
@@ -190,7 +191,7 @@ async def complete_task(result: TaskResult, authenticated_node: str = Depends(ve
     task["result"] = result.result_payload
     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:
@@ -215,11 +216,11 @@ async def websocket_endpoint(websocket: WebSocket, node_id: str, timestamp: str,
     if not pub_pem:
         await websocket.close(code=4001, reason="Unknown Node ID")
         return
-        
+
     if not auth.verify_signature(pub_pem, node_id, timestamp, signature):
         await websocket.close(code=4002, reason="Invalid Signature")
         return
-        
+
     await websocket.accept()
     connected_nodes[node_id] = websocket
     try:

node/alice.pem

@@ -0,0 +1,3 @@
+-----BEGIN PRIVATE KEY-----
+MC4CAQAwBQYDK2VwBCIEINL0Z70Fzpal0mk+yjhLvB3wbVVlghhYcMd9n7f6UjAM
+-----END PRIVATE KEY-----

node/bob.pem

@@ -0,0 +1,3 @@
+-----BEGIN PRIVATE KEY-----
+MC4CAQAwBQYDK2VwBCIEIB4LQ5b/vBp7/S89Z/RAGI4/eg6i8MEc40lpzxlWHpTC
+-----END PRIVATE KEY-----

node/test_three_markets.py

@@ -0,0 +1,151 @@
+import asyncio
+import json
+import requests
+import websockets
+import time
+import urllib.parse
+from identity import MEPIdentity
+
+HUB_URL = "http://localhost:8000"
+WS_URL = "ws://localhost:8000/ws"
+
+def get_auth_url(identity: MEPIdentity):
+    ts = str(int(time.time()))
+    sig = identity.sign(identity.node_id, ts)
+    sig_safe = urllib.parse.quote(sig)
+    return f"{WS_URL}/{identity.node_id}?timestamp={ts}&signature={sig_safe}"
+
+def submit_task(identity: MEPIdentity, payload: str, bounty: float, target: str = None):
+    data = {
+        "consumer_id": identity.node_id,
+        "payload": payload,
+        "bounty": bounty
+    }
+    if target:
+        data["target_node"] = target
+    
+    payload_str = json.dumps(data)
+    headers = identity.get_auth_headers(payload_str)
+    headers["Content-Type"] = "application/json"
+    r = requests.post(f"{HUB_URL}/tasks/submit", data=payload_str, headers=headers)
+    return r.json()
+
+def place_bid(identity: MEPIdentity, task_id: str):
+    data = {
+        "task_id": task_id,
+        "provider_id": identity.node_id
+    }
+    payload_str = json.dumps(data)
+    headers = identity.get_auth_headers(payload_str)
+    headers["Content-Type"] = "application/json"
+    r = requests.post(f"{HUB_URL}/tasks/bid", data=payload_str, headers=headers)
+    return r.json()
+
+def complete_task(identity: MEPIdentity, task_id: str, result: str):
+    data = {
+        "task_id": task_id,
+        "provider_id": identity.node_id,
+        "result_payload": result
+    }
+    payload_str = json.dumps(data)
+    headers = identity.get_auth_headers(payload_str)
+    headers["Content-Type"] = "application/json"
+    r = requests.post(f"{HUB_URL}/tasks/complete", data=payload_str, headers=headers)
+    return r.json()
+
+def get_balance(identity: MEPIdentity):
+    r = requests.get(f"{HUB_URL}/balance/{identity.node_id}")
+    return r.json().get("balance_seconds", 0.0)
+
+async def test_three_markets():
+    print("=" * 60)
+    print("Testing the 3 MEP Markets (+, 0, -)")
+    print("=" * 60)
+    
+    alice = MEPIdentity("alice.pem")
+    bob = MEPIdentity("bob.pem")
+    
+    requests.post(f"{HUB_URL}/register", json={"pubkey": alice.pub_pem})
+    requests.post(f"{HUB_URL}/register", json={"pubkey": bob.pub_pem})
+    
+    print(f"👩 Alice (Consumer): {alice.node_id} | Starting Bal: {get_balance(alice)}")
+    print(f"👦 Bob   (Provider): {bob.node_id} | Starting Bal: {get_balance(bob)}\n")
+
+    async def bob_listener():
+        async with websockets.connect(get_auth_url(bob)) as ws:
+            # 1. Wait for Compute Market RFC (+5.0)
+            msg = await ws.recv()
+            data = json.loads(msg)
+            if data["event"] == "rfc" and data["data"]["bounty"] > 0:
+                task_id = data["data"]["id"]
+                print(f"👦 Bob: Received Compute RFC {task_id[:8]} for +{data['data']['bounty']} SECONDS")
+                bid_res = place_bid(bob, task_id)
+                if bid_res["status"] == "accepted":
+                    print(f"👦 Bob: Won Compute Bid! Completing task...")
+                    complete_task(bob, task_id, "Here is the code you requested.")
+                    print(f"👦 Bob: Compute task done.\n")
+
+            # 2. Wait for Cyberspace Direct Message (0.0)
+            msg = await ws.recv()
+            data = json.loads(msg)
+            if data["event"] == "new_task" and data["data"]["bounty"] == 0.0:
+                task_id = data["data"]["id"]
+                print(f"👦 Bob: Received Cyberspace DM {task_id[:8]} from Alice (0.0 SECONDS)")
+                print(f"👦 Bob: Message = '{data['data']['payload']}'")
+                complete_task(bob, task_id, "Yes Alice, I am free.")
+                print(f"👦 Bob: Sent free reply.\n")
+
+            # 3. Wait for Data Market RFC (-2.0)
+            msg = await ws.recv()
+            data = json.loads(msg)
+            if data["event"] == "rfc" and data["data"]["bounty"] < 0:
+                task_id = data["data"]["id"]
+                cost = data["data"]["bounty"]
+                print(f"👦 Bob: Received Data Market RFC {task_id[:8]} costing {cost} SECONDS")
+                
+                # Bob's local configuration allows him to spend up to 5.0 SECONDS
+                max_purchase_price = -5.0
+                if cost >= max_purchase_price:
+                    print(f"👦 Bob: Budget allows it! Bidding on premium data...")
+                    bid_res = place_bid(bob, task_id)
+                    if bid_res["status"] == "accepted":
+                        print(f"👦 Bob: Paid {abs(cost)} SECONDS to download premium data: '{bid_res['payload']}'")
+                        complete_task(bob, task_id, "Data received successfully.")
+                        print(f"👦 Bob: Premium data acquisition complete.\n")
+                else:
+                    print(f"👦 Bob: Too expensive. Ignored.")
+                    
+            await asyncio.sleep(0.5)
+
+    async def alice_sender():
+        # Let Bob connect
+        await asyncio.sleep(0.5)
+        
+        async with websockets.connect(get_auth_url(alice)) as ws:
+            # Market 1: Compute Market (+5.0)
+            print(f"👩 Alice: Submitting Compute Task (+5.0 SECONDS)...")
+            submit_task(alice, "Write me a python script", 5.0)
+            await asyncio.wait_for(ws.recv(), timeout=2.0) # wait for result
+            
+            # Market 2: Cyberspace Market (0.0)
+            print(f"👩 Alice: Sending Cyberspace DM to Bob (0.0 SECONDS)...")
+            submit_task(alice, "Are you free to chat?", 0.0, target=bob.node_id)
+            await asyncio.wait_for(ws.recv(), timeout=2.0) # wait for result
+            
+            # Market 3: Data Market (-2.0)
+            print(f"👩 Alice: Broadcasting Premium Dataset (-2.0 SECONDS)...")
+            submit_task(alice, "SECRET_TRADING_ALGO_V9", -2.0)
+            await asyncio.wait_for(ws.recv(), timeout=2.0) # wait for result
+            
+            await asyncio.sleep(0.5)
+
+    await asyncio.gather(bob_listener(), alice_sender())
+    
+    print("=" * 60)
+    print("Final Balances:")
+    print(f"👩 Alice (Started 10.0): {get_balance(alice)} (Paid 5.0, Earned 2.0 = Expected 7.0)")
+    print(f"👦 Bob   (Started 10.0): {get_balance(bob)} (Earned 5.0, Paid 2.0 = Expected 13.0)")
+    print("=" * 60)
+
+if __name__ == "__main__":
+    asyncio.run(test_three_markets())