Add enterprise JSON structured logging and Ledger Audit Trail

.gitignore

@@ -1 +1,2 @@
 \n*.db\n*.sqlite3
+/logs

hub/logger.py

@@ -0,0 +1,76 @@
+import logging
+import logging.handlers
+import json
+import os
+from datetime import datetime, timezone
+
+LOG_DIR = "logs"
+os.makedirs(LOG_DIR, exist_ok=True)
+
+class JSONFormatter(logging.Formatter):
+    """Format logs as JSON for easy ingestion into Datadog, ELK, etc."""
+    def format(self, record):
+        log_data = {
+            "timestamp": datetime.fromtimestamp(record.created, tz=timezone.utc).isoformat(),
+            "level": record.levelname,
+            "logger": record.name,
+            "message": record.getMessage()
+        }
+        
+        # Add any extra kwargs passed to the log call
+        if hasattr(record, "extra_fields"):
+            log_data.update(record.extra_fields)
+            
+        # Add exception info if present
+        if record.exc_info:
+            log_data["exception"] = self.formatException(record.exc_info)
+            
+        return json.dumps(log_data)
+
+def setup_logger(name: str, log_file: str, level=logging.INFO, json_format=True) -> logging.Logger:
+    """Setup a rotating file logger."""
+    logger = logging.getLogger(name)
+    logger.setLevel(level)
+    
+    # Avoid adding multiple handlers if setup is called multiple times
+    if logger.handlers:
+        return logger
+        
+    # Rotate at 10MB, keep last 30 files
+    file_handler = logging.handlers.RotatingFileHandler(
+        os.path.join(LOG_DIR, log_file), 
+        maxBytes=10*1024*1024, 
+        backupCount=30
+    )
+    
+    if json_format:
+        file_handler.setFormatter(JSONFormatter())
+    else:
+        # Standard text format for audit trails or simple logs
+        formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+        file_handler.setFormatter(formatter)
+        
+    logger.addHandler(file_handler)
+    
+    # Also log to console for development
+    console_handler = logging.StreamHandler()
+    console_handler.setFormatter(logging.Formatter('%(levelname)s: %(message)s'))
+    logger.addHandler(console_handler)
+    
+    return logger
+
+# --- Hub Loggers ---
+hub_logger = setup_logger("mep.hub", "hub.json")
+
+# --- Audit Logger ---
+# The Audit Trail is strictly append-only text to reconstruct the ledger if needed
+audit_logger = setup_logger("mep.audit", "ledger_audit.log", json_format=False)
+
+def log_event(event: str, message: str, **kwargs):
+    """Helper to log JSON events with extra fields."""
+    hub_logger.info(message, extra={"extra_fields": {"event": event, **kwargs}})
+
+def log_audit(action: str, node_id: str, amount: float, new_balance: float, ref_id: str = ""):
+    """Helper to strictly log SECONDS moving in the ledger."""
+    sign = "+" if amount >= 0 else ""
+    audit_logger.info(f"AUDIT | {action} | Node: {node_id} | Amount: {sign}{amount:.6f} | Balance: {new_balance:.6f} | Ref: {ref_id}")

hub/main.py

@@ -3,6 +3,7 @@ from typing import Dict, List
 import uuid
 import db
 import auth
+from logger import log_event, log_audit, hub_logger
 
 from models import NodeRegistration, TaskCreate, TaskResult, NodeBalance, TaskBid
 
@@ -37,6 +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}")
@@ -67,7 +72,13 @@ async def submit_task(task: TaskCreate, authenticated_node: str = Depends(verify
     if task.bounty > 0:
         success = db.deduct_balance(task.consumer_id, task.bounty)
         if not success:
+            log_event("task_rejected", f"Node {task.consumer_id} lacks SECONDS to submit task", consumer_id=task.consumer_id, bounty=task.bounty)
             raise HTTPException(status_code=400, detail="Insufficient SECONDS balance")
+            
+        new_balance = db.get_balance(task.consumer_id)
+        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 = {
@@ -126,6 +137,8 @@ async def place_bid(bid: TaskBid, authenticated_node: str = Depends(verify_reque
     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",
@@ -152,15 +165,25 @@ async def complete_task(result: TaskResult, authenticated_node: str = Depends(ve
     if bounty >= 0:
         # Standard Compute Market: Provider earns SECONDS
         db.add_balance(result.provider_id, bounty)
+        new_balance = db.get_balance(result.provider_id)
+        log_audit("EARN_COMPUTE", result.provider_id, bounty, new_balance, result.task_id)
     else:
         # Data Market: Provider PAYS to receive this payload/task
         cost = abs(bounty)
         success = db.deduct_balance(result.provider_id, cost)
         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
     task["status"] = "completed"
     task["provider_id"] = result.provider_id