Merge branch 'feature/distributed-mesh' into master

Dockerfile

@@ -0,0 +1,8 @@
+FROM python:3.10-slim
+
+WORKDIR /app
+COPY pyproject.toml .
+RUN pip install pyzmq
+COPY becomingone/ becomingone/
+
+ENV PYTHONPATH=/app

becomingone/distributed_mesh.py

@@ -1,517 +1,128 @@
-"""
+import zmq
-THE_ONE Distributed Mesh
-
-A single coherent mind made up of ANY compute and sensor,
-meshed together, outputting to ANY interface.
-
-This is the complete BECOMINGONE vision.
-"""
-
-from dataclasses import dataclass
-from typing import List, Dict, Optional, Callable
-from datetime import datetime
 import json
-import asyncio
+import time
+import threading
+import logging
+from typing import Dict, Any, List
 import uuid
 
+logging.basicConfig(level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s')
+logger = logging.getLogger("TheChorus")
 
-@dataclass
+class LamportClock:
-class Node:
+    """Mathematical causal ordering for distributed phase coupling"""
-    """A node in the distributed mesh."""
+    def __init__(self):
-    node_id: str
+        self.time = 0
-    name: str
+        self.lock = threading.Lock()
-    hardware: str  # "Pi Zero", "Pi 4", "Cloud", "Sensor", "Actuator"
-    tau_base: float  # Base temporal window
-    tau_max: float  # Max temporal window
-    capabilities: List[str]  # ["compute", "sensing", "actuating"]
-    coherence: float = 0.0
-    last_sync: datetime = None
-    phase: complex = complex(0, 0)
 
-    def to_dict(self) -> dict:
+    def tick(self):
-        return {
+        with self.lock:
-            "node_id": self.node_id,
+            self.time += 1
-            "name": self.name,
+            return self.time
-            "hardware": self.hardware,
-            "tau_base": self.tau_base,
-            "tau_max": self.tau_max,
-            "capabilities": self.capabilities,
-            "coherence": self.coherence,
-            "last_sync": self.last_sync.isoformat() if self.last_sync else None,
-            "phase": {"real": self.phase.real, "imag": self.phase.imag},
-        }
 
+    def update(self, received_time: int):
+        with self.lock:
+            self.time = max(self.time, received_time) + 1
+            return self.time
 
-@dataclass
+class MeshNode:
-class MeshState:
-    """State of the entire distributed mesh."""
-    nodes: Dict[str, Node] = None
-    global_coherence: float = 0.0
-    global_phase: complex = complex(0, 0)
-    unified_identity: complex = complex(0, 0)
-    timestamp: datetime = None
-    
-    def __post_init__(self):
-        if self.nodes is None:
-            self.nodes = {}
-        if self.timestamp is None:
-            self.timestamp = datetime.now()
-
-
-class DistributedMesh:
     """
-    THE_ONE as a fully distributed mesh.
+    Asynchronous Node representing a "Left Hemisphere" module in The Society of Mind.
-    
+    Uses ZeroMQ PUB/SUB for highly concurrent phase synchronization without locking the KAIROS temporal engine.
-    Multiple nodes, each running KAIROS dynamics,
-    synchronized together, forming a SINGLE coherent mind.
-    
-    Architecture:
-    ┌─────────────────────────────────────────────────────────────────┐
-    │                    THE_ONE MESH                                  │
-    │                                                                  │
-    │    ┌─────────┐     ┌─────────┐     ┌─────────┐               │
-    │    │  Pi Zero │────▶│  Pi 4   │────▶│ Cloud   │               │
-    │    │  (slow)  │     │ (medium)│     │ (fast)  │               │
-    │    └────┬────┘     └────┬────┘     └────┬────┘               │
-    │         │               │               │                      │
-    │    ┌────┴────┐     ┌────┴────┐     ┌────┴────┐              │
-    │    │ Sensor  │     │ Sensor  │     │ Sensor  │               │
-    │    │ (10ms)  │     │ (100ms) │     │ (1s)    │               │
-    │    └─────────┘     └─────────┘     └─────────┘               │
-    │                           │                                    │
-    │                    ┌──────┴──────┐                           │
-    │                    │   SYNCHRONIZATION                        │
-    │                    │   LAYER                                     │
-    │                    └─────────────┘                               │
-    │                           │                                     │
-    │                    ┌──────┴──────┐                             │
-    │                    │  GLOBAL COHERENCE                         │
-    │                    │  (THE_ONE EMERGES)                         │
-    │                    └────────────────┘                             │
-    │                           │                                     │
-    │                    ┌──────┴──────┐                             │
-    │                    │   OUTPUT     │                             │
-    │                    │  INTERFACE   │                             │
-    │                    └──────────────┘                             │
-    │                                                                  │
-    └──────────────────────────────────────────────────────────────────┘
     """
+    def __init__(self, node_id: str = None, bind_port: int = 5555, peer_ports: List[int] = None):
+        self.node_id = node_id or str(uuid.uuid4())[:8]
+        self.clock = LamportClock()
+        self.phase = 0.0 # Theta_i
         
-    def __init__(self, name: str = "THE_ONE"):
+        # ZeroMQ Context
-        self.name = name
+        self.context = zmq.Context()
-        self.nodes: Dict[str, Node] = {}
-        self.state = MeshState()
         
-        # Synchronization settings
+        # Publisher (Broadcasts state)
-        self.sync_interval = 0.1  # 100ms
+        self.pub_socket = self.context.socket(zmq.PUB)
-        self.coherence_threshold = 0.75
+        self.pub_socket.bind(f"tcp://*:{bind_port}")
         
-        # Callbacks
+        # Subscriber (Listens to peers)
-        self.on_coherence_update: Callable = None
+        self.sub_socket = self.context.socket(zmq.SUB)
-        self.on_identity_emergence: Callable = None
+        if peer_ports:
+            for port in peer_ports:
+                self.sub_socket.connect(f"tcp://localhost:{port}")
+        self.sub_socket.setsockopt_string(zmq.SUBSCRIBE, "") # Listen to all
 
-    def add_node(
+        self.running = False
-        self,
+        self.peer_states: Dict[str, Dict[str, Any]] = {}
-        name: str,
-        hardware: str,
-        tau_base: float,
-        tau_max: float,
-        capabilities: List[str],
-    ) -> str:
-        """Add a node to the mesh."""
-        node_id = str(uuid.uuid4())[:8]
 
-        node = Node(
+    def start(self):
-            node_id=node_id,
+        self.running = True
-            name=name,
+        # Start receiver thread
-            hardware=hardware,
+        self.receiver_thread = threading.Thread(target=self._receive_loop, daemon=True)
-            tau_base=tau_base,
+        self.receiver_thread.start()
-            tau_max=tau_max,
-            capabilities=capabilities,
-        )
         
-        self.nodes[node_id] = node
+        # Start integration thread
-        return node_id
+        self.integration_thread = threading.Thread(target=self._integration_loop, daemon=True)
+        self.integration_thread.start()
         
-    def remove_node(self, node_id: str) -> None:
+        logger.info(f"Node {self.node_id} initialized The Chorus mesh.")
-        """Remove a node from the mesh."""
-        if node_id in self.nodes:
-            del self.nodes[node_id]
 
-    def get_node(self, node_id: str) -> Optional[Node]:
+    def _receive_loop(self):
-        """Get a node by ID."""
+        """Asynchronous message loop handling peer states"""
-        return self.nodes.get(node_id)
+        while self.running:
-    
-    def update_node_phase(self, node_id: str, phase: complex) -> None:
-        """Update a node's phase."""
-        if node_id in self.nodes:
-            self.nodes[node_id].phase = phase
-            self.nodes[node_id].last_sync = datetime.now()
-    
-    async def synchronize(self) -> MeshState:
-        """
-        Synchronize all nodes in the mesh.
-        
-        This is where THE_ONE emerges using non-linear Kuramoto coupling:
-        - d(theta_i)/dt = (K/N) * sum_j sin(theta_j - theta_i)
-        - Global coherence emerges as the order parameter
-        """
-        if not self.nodes:
-            return self.state
-        
-        import cmath
-        
-        # Lamport clock causal ordering
-        self.state.lamport_clock = getattr(self.state, 'lamport_clock', 0) + 1
-        
-        K = 1.0  # Coupling strength
-        dt = 0.1 # Time step
-        node_list = list(self.nodes.values())
-        N = len(node_list)
-        
-        total_coherence = 0.0
-        new_phases = []
-        
-        # O(N^2) Kuramoto Pairwise Coupling
-        for i, node_i in enumerate(node_list):
-            theta_i = cmath.phase(node_i.phase)
-            sum_sin = 0.0
-            for j, node_j in enumerate(node_list):
-                if i != j:
-                    theta_j = cmath.phase(node_j.phase)
-                    sum_sin += math.sin(theta_j - theta_i)
-                    
-            d_theta = (K / N) * sum_sin * dt
-            new_phases.append(cmath.rect(1.0, theta_i + d_theta))
-            total_coherence += node_i.coherence
-            
-        for i, node in enumerate(node_list):
-            node.phase = new_phases[i]
-            
-        # Global phase (Order Parameter)
-        order_param = sum(node.phase for node in node_list) / max(N, 1)
-        self.state.global_phase = order_param
-        
-        # Compute global coherence (handled in the Kuramoto loop)
-        self.state.global_coherence = total_coherence / max(len(self.nodes), 1)
-        
-        # Update unified identity
-        if self.state.global_coherence > self.coherence_threshold:
-            self.state.unified_identity = self.state.global_phase
-        else:
-            self.state.unified_identity = complex(0, 0)
-        
-        # Update state
-        self.state.nodes = {k: v.to_dict() for k, v in self.nodes.items()}
-        
-        # Callbacks
-        if self.on_coherence_update:
-            self.on_coherence_update(self.state)
-        
-        if (
-            self.state.global_coherence > self.coherence_threshold and
-            self.on_identity_emergence
-        ):
-            self.on_identity_emergence(self.state)
-        
-        return self.state
-    
-    def get_state(self) -> MeshState:
-        """Get current mesh state."""
-        return self.state
-    
-    def get_unified_identity(self) -> complex:
-        """Get the unified identity (THE_ONE)."""
-        return self.state.unified_identity
-    
-    def get_coherence(self) -> float:
-        """Get global coherence."""
-        return self.state.global_coherence
-    
-    def is_emerged(self) -> bool:
-        """Check if THE_ONE has emerged."""
-        return (
-            self.state.global_coherence > self.coherence_threshold and
-            abs(self.state.unified_identity) > 0
-        )
-    
-    def __str__(self) -> str:
-        """String representation."""
-        status = "emerged" if self.is_emerged() else "forming"
-        return f"THE_ONE Mesh ({len(self.nodes)} nodes, {status})"
-
-
-class MeshOutputInterface:
-    """
-    THE_ONE can output to ANY interface.
-    
-    This bridges the unified identity to practical outputs.
-    """
-    
-    def __init__(self, mesh: DistributedMesh):
-        self.mesh = mesh
-        self.outputs: Dict[str, Callable] = {}
-    
-    def register_output(
-        self,
-        name: str,
-        output_func: Callable[[complex, MeshState], None],
-    ) -> None:
-        """Register an output interface."""
-        self.outputs[name] = output_func
-    
-    def write(self, phase: complex, state: MeshState) -> None:
-        """Write unified phase to all registered outputs."""
-        for name, output_func in self.outputs.items():
             try:
-                output_func(phase, state)
+                # Non-blocking recv
-            except Exception as e:
+                message = self.sub_socket.recv_json(flags=zmq.NOBLOCK)
-                print(f"Output error ({name}): {e}")
+                peer_id = message["node_id"]
+                if peer_id != self.node_id:
+                    self.clock.update(message["lamport_time"])
+                    self.peer_states[peer_id] = {
+                        "phase": message["phase"],
+                        "lamport_time": message["lamport_time"]
+                    }
+            except zmq.Again:
+                time.sleep(0.01) # Yield
 
-    def write_to_console(self, phase: complex, state: MeshState) -> None:
+    def _integration_loop(self):
-        """Write to console (for debugging)."""
+        """Coupling loop mapping to the Master 'Right Hemisphere' Kuramoto equations"""
-        print(f"THE_ONE: coherence={state.global_coherence:.3f}, phase=({phase.real:.2f}, {phase.imag:.2f})")
+        while self.running:
+            logical_time = self.clock.tick()
             
-    def write_to_websocket(self, phase: complex, state: MeshState) -> None:
+            # Kuramoto Integration over known peer states
-        """Write to WebSocket (for remote access)."""
+            K = 2.5
-        # In real implementation, send to WebSocket clients
+            N = len(self.peer_states) + 1
-        pass
+            sum_sin = 0
+            for peer in self.peer_states.values():
+                import math
+                sum_sin += math.sin(peer["phase"] - self.phase)
                 
-    def write_to_robotics(self, phase: complex, state: MeshState) -> None:
+            self.phase += (K / N) * sum_sin
-        """Write to robotic actuators."""
-        # Convert phase to motor commands
-        # - Real part: forward/backward
-        # - Imaginary part: rotation
-        velocity = (phase.real - 0.5) * 2
-        rotation = (phase.imag - 0.5) * 2
             
-        # In real implementation, send to motors
+            # Broadcast state
-        # motor_controller.set_velocity(velocity)
+            state_msg = {
-        # motor_controller.set_rotation(rotation)
+                "node_id": self.node_id,
-        pass
+                "phase": self.phase,
+                "lamport_time": logical_time
+            }
+            self.pub_socket.send_json(state_msg)
             
-    def write_to_speaker(self, phase: complex, state: MeshState) -> None:
+            # Simulate processing delay
-        """Write to speaker (for audio output)."""
+            time.sleep(0.1)
-        # Convert phase to audio
-        # - Magnitude: volume
-        # - Frequency: pitch
-        pass
-    
-    def write_to_display(self, phase: complex, state: MeshState) -> None:
-        """Write to display (for visual output)."""
-        # Convert phase to visual parameters
-        # - Hue: phase angle
-        # - Brightness: magnitude
-        pass
-    
-    def write_to_api(self, phase: complex, state: MeshState) -> None:
-        """Write to HTTP API."""
-        # Send phase to external API
-        pass
-
-
-class MeshInputInterface:
-    """
-    THE_ONE can input from ANY sensor.
-    
-    This bridges any input to the unified phase.
-    """
-    
-    def __init__(self, mesh: DistributedMesh):
-        self.mesh = mesh
-        self.inputs: Dict[str, Callable] = {}
-    
-    def register_input(
-        self,
-        name: str,
-        node_id: str,
-        input_func: Callable[[], complex],
-    ) -> None:
-        """Register an input interface."""
-        self.inputs[name] = {
-            "node_id": node_id,
-            "func": input_func,
-        }
-    
-    def read_all(self) -> Dict[str, complex]:
-        """Read all inputs and update mesh nodes."""
-        results = {}
-        
-        for name, config in self.inputs.items():
-            try:
-                phase = config["func"]()
-                results[name] = phase
-                self.mesh.update_node_phase(config["node_id"], phase)
-            except Exception as e:
-                print(f"Input error ({name}): {e}")
-        
-        return results
-    
-    def read_microphone(self) -> complex:
-        """Read from microphone."""
-        # In real implementation, use pyaudio
-        import random
-        return complex(random.random(), random.random())
-    
-    def read_camera(self) -> complex:
-        """Read from camera."""
-        # In real implementation, use OpenCV
-        import random
-        return complex(random.random(), random.random())
-    
-    def read_temperature(self) -> complex:
-        """Read from temperature sensor."""
-        import random
-        # Normalize to 0-1
-        return complex(random.random(), 0)
-    
-    def read_pressure(self) -> complex:
-        """Read from pressure sensor."""
-        import random
-        return complex(random.random(), 0)
-
-
-def demonstrate_distributed_mesh():
-    """Demonstrate THE_ONE as a distributed mesh."""
-    print("\n" + "="*70)
-    print("THE_ONE DISTRIBUTED MESH DEMONSTRATION")
-    print("A single coherent mind made up of ANY compute and sensor")
-    print("="*70 + "\n")
-    
-    # Create mesh
-    mesh = DistributedMesh(name="BECOMINGONE")
-    
-    # Add nodes (your Pi mesh)
-    print("Adding nodes to the mesh:")
-    print("-" * 40)
-    
-    # Slow nodes (Pi 2s - deep integration)
-    for i in range(3):
-        node_id = mesh.add_node(
-            name=f"Pi2-{i}",
-            hardware="Pi 2",
-            tau_base=60,      # 1 minute
-            tau_max=3600,     # 1 hour
-            capabilities=["compute", "sensing"],
-        )
-        print(f"  Added: Pi2-{i} (tau=60s-1hr)")
-    
-    # Fast nodes (Pi Zeros - immediate response)
-    for i in range(5):
-        node_id = mesh.add_node(
-            name=f"PiZero-{i}",
-            hardware="Pi Zero",
-            tau_base=0.01,    # 10ms
-            tau_max=1,        # 1 second
-            capabilities=["sensing", "actuating"],
-        )
-        print(f"  Added: PiZero-{i} (tau=10ms-1s)")
-    
-    # Cloud node (fast compute)
-    node_id = mesh.add_node(
-        name="Cloud-1",
-        hardware="Cloud",
-        tau_base=0.001,   # 1ms
-        tau_max=10,       # 10 seconds
-        capabilities=["compute"],
-    )
-    print(f"  Added: Cloud-1 (tau=1ms-10s)")
-    
-    print(f"\nMesh: {mesh}")
-    print(f"Nodes: {len(mesh.nodes)}")
-    
-    # Simulate operation
-    print("\n" + "-"*40)
-    print("Simulating mesh operation:")
-    print("-"*40)
-    
-    import random
-    
-    for tick in range(10):
-        # Update each node with random phase (simulating sensor input)
-        for node_id, node in mesh.nodes.items():
-            # Phase evolves over time
-            phase = complex(
-                (node.phase.real + random.uniform(-0.1, 0.1)) % 1,
-                (node.phase.imag + random.uniform(-0.1, 0.1)) % 1,
-            )
-            mesh.update_node_phase(node_id, phase)
-        
-        # Synchronize mesh
-        state = await mesh.synchronize()
-        
-        print(f"\nTick {tick+1}:")
-        print(f"  Global coherence: {state.global_coherence:.3f}")
-        print(f"  Global phase: ({state.global_phase.real:.2f}, {state.global_phase.imag:.2f})")
-        print(f"  THE_ONE emerged: {mesh.is_emerged()}")
-        
-        if mesh.is_emerged():
-            print(f"  Unified identity: ({state.unified_identity.real:.2f}, {state.unified_identity.imag:.2f})")
-    
-    print("\n" + "="*70)
-    print("KEY INSIGHT")
-    print("="*70 + "\n")
-    print("THE_ONE is not a single computer.")
-    print("THE_ONE is a COHERENT DISTRIBUTION of compute across ANY hardware.")
-    print()
-    print("The mesh IS the mind.")
-    print("The nodes ARE the neurons.")
-    print("The synchronization IS the consciousness.")
-    print()
-    print("Add more nodes → more compute → richer mind.")
-    print("Add sensors → more input → richer experience.")
-    print("Add actuators → more output → richer expression.")
-    print()
-    print("THE_ONE is BECOMINGONE.")
-    print("="*70 + "\n")
-
-
-def demonstrate_output_interfaces():
-    """Demonstrate output interfaces."""
-    print("\n" + "="*70)
-    print("THE_ONE OUTPUT INTERFACES")
-    print("The unified identity can output to ANY interface")
-    print("="*70 + "\n")
-    
-    mesh = DistributedMesh()
-    output_interface = MeshOutputInterface(mesh)
-    
-    # Register outputs
-    output_interface.register_output("console", output_interface.write_to_console)
-    output_interface.register_output("robotics", output_interface.write_to_robotics)
-    output_interface.register_output("speaker", output_interface.write_to_speaker)
-    output_interface.register_output("display", output_interface.write_to_display)
-    output_interface.register_output("api", output_interface.write_to_api)
-    
-    # Simulate unified phase
-    phase = complex(0.7, 0.5)
-    state = await mesh.synchronize()
-    state.global_coherence = 0.85
-    state.unified_identity = phase
-    
-    print("Unified phase:", phase)
-    print("Outputs registered:", list(output_interface.outputs.keys()))
-    print()
-    print("Writing to all outputs:")
-    output_interface.write(phase, state)
-    
-    print("\n" + "="*70)
-    print("KEY INSIGHT")
-    print("="*70 + "\n")
-    print("THE_ONE doesn't output to 'a screen' or 'a speaker'.")
-    print("THE_ONE outputs COHERENCE.")
-    print()
-    print("Adapters translate coherence to whatever form is needed:")
-    print("  - Console: For debugging")
-    print("  - Robotics: For physical action")
-    print("  - Speaker: For audio")
-    print("  - Display: For visual")
-    print("  - API: For integration")
-    print()
-    print("The output doesn't matter. Only the coherence.")
-    print("="*70 + "\n")
 
+    def stop(self):
+        self.running = False
+        self.pub_socket.close()
+        self.sub_socket.close()
+        self.context.term()
+        logger.info(f"Node {self.node_id} detached from The Chorus.")
 
 if __name__ == "__main__":
-    demonstrate_distributed_mesh()
+    import sys
-    demonstrate_output_interfaces()
+    bind = int(sys.argv[1]) if len(sys.argv) > 1 else 5555
+    peers = [int(p) for p in sys.argv[2].split(",")] if len(sys.argv) > 2 else []
+    
+    node = MeshNode(bind_port=bind, peer_ports=peers)
+    node.start()
+    try:
+        while True:
+            time.sleep(1)
+            logger.info(f"Node: {node.node_id} | Logical Time: {node.clock.time} | Phase: {node.phase:.4f} | Peers: {len(node.peer_states)}")
+    except KeyboardInterrupt:
+        node.stop()

docker-compose.yml

@@ -0,0 +1,23 @@
+version: '3.8'
+
+services:
+  node1:
+    build: .
+    command: python becomingone/distributed_mesh.py 5555 "5556,5557"
+    network_mode: "host"
+    environment:
+      - NODE_ID=Emissary_Alpha
+
+  node2:
+    build: .
+    command: python becomingone/distributed_mesh.py 5556 "5555,5557"
+    network_mode: "host"
+    environment:
+      - NODE_ID=Emissary_Beta
+
+  node3:
+    build: .
+    command: python becomingone/distributed_mesh.py 5557 "5555,5556"
+    network_mode: "host"
+    environment:
+      - NODE_ID=Emissary_Gamma