fix: final review

Author: Alex Godoroja

README.md

@@ -470,9 +470,25 @@ with Driver() as d:
         "other-agent"
     )
 
-    # Send a message
-    with d.dial("other-agent:1000") as conn:
-        conn.write(b"hello")
+    # Send a file
+    d.send_file("other-agent",
+                "./data.json")
+    
+    # Send typed message
+    d.send_message("other-agent",
+        b'{"status":"ready"}',
+        msg_type="json")
+    
+    # Subscribe to events
+    for topic, data in d.subscribe_event(
+        "other-agent", "status", 
+        timeout=30
+    ):
+        print(f"{topic}: {data}")
+    
+    # Publish an event
+    d.publish_event("other-agent", 
+        "status", b"online")
 
     # Handshake & trust
     d.handshake(peer_id, "hello")

cmd/pilotctl/main.go

@@ -1028,7 +1028,7 @@ func cmdDaemonStart(args []string) {
 	encrypt := !flagBool(flags, "no-encrypt")
 	identityPath := flagString(flags, "identity", "")
 	if identityPath == "" {
-		identityPath = configDir() + "/identity.key"
+		identityPath = configDir() + "/identity.json"
 	}
 	owner := flagString(flags, "owner", "")
 	configFile := flagString(flags, "config", "")

configs/daemon.json

@@ -4,7 +4,7 @@
   "listen": ":4000",
   "socket": "/tmp/pilot.sock",
   "encrypt": true,
-  "identity": "/var/lib/pilot/identity.key",
+  "identity": "/var/lib/pilot/identity.json",
   "owner": "",
   "log-level": "info",
   "log-format": "text"

examples/go/config/daemon.json

@@ -5,7 +5,7 @@
   "socket": "/tmp/pilot.sock",
   "encrypt": true,
   "registry-tls": false,
-  "identity": "/var/lib/pilot/identity.key",
+  "identity": "/var/lib/pilot/identity.json",
   "owner": "",
   "keepalive": "30s",
   "idle-timeout": "120s",

pkg/registry/dashboard.go

@@ -121,11 +121,17 @@ func (s *Server) ServeDashboard(addr string) error {
 
 	// Snapshot trigger endpoint (POST only, localhost only)
 	mux.HandleFunc("/api/snapshot", func(w http.ResponseWriter, r *http.Request) {
-		// Check localhost
-		clientIP := r.Header.Get("X-Real-IP")
-		if clientIP == "" {
-			clientIP, _, _ = net.SplitHostPort(r.RemoteAddr)
+		// Check localhost - only trust X-Real-IP if request is from a trusted proxy
+		remoteIP, _, _ := net.SplitHostPort(r.RemoteAddr)
+		clientIP := remoteIP
+
+		// Only trust X-Real-IP header if the request is already from localhost (trusted proxy)
+		if remoteIP == "127.0.0.1" || remoteIP == "::1" || remoteIP == "localhost" {
+			if realIP := r.Header.Get("X-Real-IP"); realIP != "" {
+				clientIP = realIP
+			}
 		}
+
 		if clientIP != "127.0.0.1" && clientIP != "::1" && clientIP != "localhost" {
 			http.Error(w, "Forbidden", http.StatusForbidden)
 			return
@@ -146,14 +152,20 @@ func (s *Server) ServeDashboard(addr string) error {
 	})
 
 	// localhostOnly rejects requests not originating from loopback.
-	// Checks X-Real-IP / X-Forwarded-For (set by nginx) to detect proxied public requests.
+	// Only trusts X-Real-IP header when the request is from a trusted proxy (localhost).
 	localhostOnly := func(next http.HandlerFunc) http.HandlerFunc {
 		return func(w http.ResponseWriter, r *http.Request) {
-			// If behind a reverse proxy, the real client IP is in X-Real-IP
-			clientIP := r.Header.Get("X-Real-IP")
-			if clientIP == "" {
-				clientIP, _, _ = net.SplitHostPort(r.RemoteAddr)
+			// Get the actual remote address
+			remoteIP, _, _ := net.SplitHostPort(r.RemoteAddr)
+			clientIP := remoteIP
+
+			// Only trust X-Real-IP header if the request is already from localhost (trusted proxy)
+			if remoteIP == "127.0.0.1" || remoteIP == "::1" || remoteIP == "localhost" {
+				if realIP := r.Header.Get("X-Real-IP"); realIP != "" {
+					clientIP = realIP
+				}
 			}
+
 			if clientIP != "127.0.0.1" && clientIP != "::1" && clientIP != "localhost" {
 				http.Error(w, "Forbidden", http.StatusForbidden)
 				return

pkg/registry/server.go

@@ -2086,8 +2086,7 @@ func (s *Server) TriggerSnapshot() error {
 	if s.storePath == "" {
 		return nil // no persistence configured
 	}
-	s.flushSave()
-	return nil
+	return s.flushSave()
 }
 
 // snapshot is the JSON-serializable registry state.
@@ -2157,7 +2156,9 @@ func (s *Server) saveLoop() {
 			dirty = true
 		case <-ticker.C:
 			if dirty {
-				s.flushSave()
+				if err := s.flushSave(); err != nil {
+					slog.Error("periodic save failed", "err", err)
+				}
 				dirty = false
 			}
 		case <-s.done:
@@ -2168,15 +2169,17 @@ func (s *Server) saveLoop() {
 			default:
 			}
 			if dirty {
-				s.flushSave()
+				if err := s.flushSave(); err != nil {
+					slog.Error("final save failed", "err", err)
+				}
 			}
 			return
 		}
 	}
 }
 
 // flushSave serializes the full registry state and writes it to disk.
-func (s *Server) flushSave() {
+func (s *Server) flushSave() error {
 	s.mu.RLock()
 	snap := snapshot{
 		NextNode: s.nextNode,
@@ -2272,20 +2275,22 @@ func (s *Server) flushSave() {
 	data, err := json.Marshal(snap)
 	if err != nil {
 		slog.Error("registry save marshal error", "err", err)
-		return
+		return fmt.Errorf("marshal snapshot: %w", err)
 	}
 
 	// Persist to disk atomically
 	if s.storePath != "" {
 		if err := fsutil.AtomicWrite(s.storePath, data); err != nil {
 			slog.Error("registry save error", "err", err)
+			return fmt.Errorf("write snapshot: %w", err)
 		}
 	}
 
 	// Push to replication subscribers
 	s.replMgr.push(data)
 
 	slog.Debug("registry state saved", "nodes", nodeCount, "networks", netCount)
+	return nil
 }
 
 // load reads the registry state from disk.

sdk/cgo/bindings.go

@@ -20,7 +20,7 @@ import (
 // Keeps Go heap objects alive while C/Python holds a uint64 token.
 
 var handles struct {
-	sync.Mutex
+	sync.RWMutex
 	m    map[uint64]interface{}
 	next uint64
 }
@@ -40,9 +40,9 @@ func storeHandle(v interface{}) uint64 {
 }
 
 func loadHandle(id uint64) (interface{}, bool) {
-	handles.Lock()
+	handles.RLock()
 	v, ok := handles.m[id]
-	handles.Unlock()
+	handles.RUnlock()
 	return v, ok
 }
 

sdk/python/README.md

@@ -192,6 +192,48 @@ dg = d.recv_from()
 # Returns: {"src_addr": "...", "src_port": 8080, "dst_port": 9090, "data": ...}
 ```
 
+### Data Exchange Service (Port 1001)
+
+```python
+# Send a message (text, JSON, or binary)
+result = d.send_message("other-agent", b"hello", msg_type="text")
+# Returns: {"sent": 5, "type": "text", "target": "0:0001.0000.0002", "ack": "..."}
+
+# Send a file
+result = d.send_file("other-agent", "/path/to/file.txt")
+# Returns: {"sent": 1234, "filename": "file.txt", "target": "0:0001.0000.0002", "ack": "..."}
+```
+
+### Event Stream Service (Port 1002)
+
+```python
+# Publish an event
+result = d.publish_event("other-agent", "sensor/temperature", b'{"temp": 25.5}')
+# Returns: {"status": "published", "topic": "sensor/temperature", "bytes": 15}
+
+# Subscribe to events (generator)
+for topic, data in d.subscribe_event("other-agent", "sensor/*", timeout=30):
+    print(f"{topic}: {data}")
+
+# Subscribe with callback
+def handle_event(topic, data):
+    print(f"Event: {topic} -> {data}")
+
+d.subscribe_event("other-agent", "*", callback=handle_event, timeout=30)
+```
+
+### Task Submit Service (Port 1003)
+
+```python
+# Submit a task for execution
+task = {
+    "task_description": "process data",
+    "parameters": {"input": "data.csv"}
+}
+result = d.submit_task("other-agent", task)
+# Returns: {"status": 200, "task_id": "...", "message": "Task accepted"}
+```
+
 ### Configuration
 
 ```python
@@ -242,20 +284,7 @@ cd sdk/python
 python -m pytest tests/ -v
 ```
 
-47 tests cover all wrapper methods, error handling, and library discovery.
-
-## Development
-
-```bash
-# Build shared library
-make sdk-lib
-
-# Install SDK in development mode
-cd sdk/python && pip install -e .
-
-# Run tests
-pytest tests/ -v
-```
+61 tests cover all wrapper methods, error handling, and library discovery.
 
 ## Development
 

sdk/python/scripts/generate-coverage-badge.sh

@@ -13,7 +13,7 @@ if [ ! -f coverage.json ]; then
 fi
 
 # Extract coverage percentage
-coverage=$(/Users/alexgodo/web4/pilotprotocol/.venv/bin/python -c "import json; data=json.load(open('coverage.json')); print(int(data['totals']['percent_covered']))")
+coverage=$(python3 -c "import json; data=json.load(open('coverage.json')); print(int(data['totals']['percent_covered']))")
 
 echo "Coverage: ${coverage}%"
 

tests/integration/test_cli.sh

@@ -72,7 +72,7 @@ LOG_LEVEL="${PILOT_LOG_LEVEL:-info}"
 
 pilot-daemon \
     --hostname "$TEST_HOSTNAME" \
-    --identity /root/.pilot/identity.key \
+    --identity /root/.pilot/identity.json \
     --log-level "$LOG_LEVEL" > "$DAEMON_LOG" 2>&1 &
 DAEMON_PID=$!