strange_loop/

swarm_real.rs

// Real Agent Swarm with Message Passing
// This replaces the fake swarm with actual distributed computation

use crossbeam::channel::{bounded, Sender, Receiver};
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, AtomicBool, Ordering};
use std::thread;
use std::time::{Duration, Instant};

/// Message types for inter-agent communication
#[derive(Debug, Clone)]
pub enum AgentMessage {
    Compute { task_id: u64, data: Vec<f64> },
    Result { task_id: u64, result: Vec<f64> },
    Coordinate { from: usize, to: usize, payload: String },
    Heartbeat { agent_id: usize, timestamp: u64 },
}

/// Real computational agent that performs actual work
pub struct ComputeAgent {
    pub id: usize,
    pub agent_type: AgentType,
    pub inbox: Receiver<AgentMessage>,
    pub outbox: Vec<Sender<AgentMessage>>,
    pub metrics: Arc<AgentMetrics>,
}

#[derive(Debug, Clone)]
pub enum AgentType {
    LinearAlgebra,  // Matrix operations
    Optimization,   // Gradient descent, etc
    Quantum,        // Quantum simulation
    Analysis,       // Data analysis
    Coordinator,    // Task distribution
}

/// Shared metrics for monitoring
pub struct AgentMetrics {
    pub messages_processed: AtomicU64,
    pub compute_time_us: AtomicU64,
    pub is_active: AtomicBool,
}

impl ComputeAgent {
    /// Run the agent's main processing loop
    pub fn run(self) {
        self.metrics.is_active.store(true, Ordering::SeqCst);

        while self.metrics.is_active.load(Ordering::SeqCst) {
            // Process messages with 25μs budget
            let deadline = Instant::now() + Duration::from_micros(25);

            while let Ok(msg) = self.inbox.recv_timeout(Duration::from_micros(1)) {
                let start = Instant::now();

                match msg {
                    AgentMessage::Compute { task_id, data } => {
                        let result = self.process_computation(&data);

                        // Send result to coordinator
                        if let Some(coordinator) = self.outbox.get(0) {
                            let _ = coordinator.send(AgentMessage::Result {
                                task_id,
                                result,
                            });
                        }
                    }
                    AgentMessage::Coordinate { from, to, payload } => {
                        // Route message to destination
                        if let Some(dest) = self.outbox.get(to) {
                            let _ = dest.send(AgentMessage::Coordinate {
                                from,
                                to,
                                payload,
                            });
                        }
                    }
                    _ => {}
                }

                let elapsed = start.elapsed().as_micros() as u64;
                self.metrics.compute_time_us.fetch_add(elapsed, Ordering::Relaxed);
                self.metrics.messages_processed.fetch_add(1, Ordering::Relaxed);

                if Instant::now() >= deadline {
                    break; // Respect tick budget
                }
            }
        }
    }

    /// Perform actual computation based on agent type
    fn process_computation(&self, data: &[f64]) -> Vec<f64> {
        match self.agent_type {
            AgentType::LinearAlgebra => {
                // Real matrix multiplication (simplified)
                let n = (data.len() as f64).sqrt() as usize;
                let mut result = vec![0.0; n];

                for i in 0..n {
                    for j in 0..n {
                        result[i] += data[i * n + j];
                    }
                }
                result
            }
            AgentType::Optimization => {
                // Gradient descent step
                data.iter()
                    .map(|&x| x - 0.01 * (2.0 * x - 1.0)) // Simple gradient
                    .collect()
            }
            AgentType::Quantum => {
                // Quantum-inspired optimization
                data.iter()
                    .map(|&x| (x * std::f64::consts::PI).sin())
                    .collect()
            }
            AgentType::Analysis => {
                // Statistical analysis
                let mean = data.iter().sum::<f64>() / data.len() as f64;
                let variance = data.iter()
                    .map(|&x| (x - mean).powi(2))
                    .sum::<f64>() / data.len() as f64;
                vec![mean, variance, variance.sqrt()]
            }
            AgentType::Coordinator => {
                // Task distribution logic
                data.to_vec()
            }
        }
    }
}

/// Real swarm coordinator that manages agents
pub struct SwarmCoordinator {
    pub agents: Vec<thread::JoinHandle<()>>,
    pub channels: Vec<Sender<AgentMessage>>,
    pub metrics: Vec<Arc<AgentMetrics>>,
    pub topology: SwarmTopology,
}

#[derive(Debug, Clone)]
pub enum SwarmTopology {
    Mesh,         // Fully connected
    Hierarchical, // Tree structure
    Ring,         // Circular connections
    Star,         // Central hub
}

impl SwarmCoordinator {
    /// Create a new swarm with real agents
    pub fn new(agent_count: usize, topology: SwarmTopology) -> Self {
        let mut channels = Vec::new();
        let mut receivers = Vec::new();
        let mut metrics = Vec::new();

        // Create communication channels
        for _ in 0..agent_count {
            let (tx, rx) = bounded(1000);
            channels.push(tx);
            receivers.push(rx);
            metrics.push(Arc::new(AgentMetrics {
                messages_processed: AtomicU64::new(0),
                compute_time_us: AtomicU64::new(0),
                is_active: AtomicBool::new(false),
            }));
        }

        // Create agents based on topology
        let mut agents = Vec::new();

        for i in 0..agent_count {
            let agent_type = match i % 5 {
                0 => AgentType::Coordinator,
                1 => AgentType::LinearAlgebra,
                2 => AgentType::Optimization,
                3 => AgentType::Quantum,
                _ => AgentType::Analysis,
            };

            // Set up connections based on topology
            let connections = match topology {
                SwarmTopology::Mesh => {
                    // Connect to all other agents
                    channels.clone()
                }
                SwarmTopology::Star => {
                    // Connect only to coordinator (agent 0)
                    if i == 0 {
                        channels.clone()
                    } else {
                        vec![channels[0].clone()]
                    }
                }
                SwarmTopology::Ring => {
                    // Connect to next agent in ring
                    vec![channels[(i + 1) % agent_count].clone()]
                }
                SwarmTopology::Hierarchical => {
                    // Binary tree connections
                    let mut conns = vec![];
                    if i > 0 {
                        conns.push(channels[(i - 1) / 2].clone()); // Parent
                    }
                    if 2 * i + 1 < agent_count {
                        conns.push(channels[2 * i + 1].clone()); // Left child
                    }
                    if 2 * i + 2 < agent_count {
                        conns.push(channels[2 * i + 2].clone()); // Right child
                    }
                    conns
                }
            };

            let agent = ComputeAgent {
                id: i,
                agent_type,
                inbox: receivers.pop().unwrap(),
                outbox: connections,
                metrics: metrics[i].clone(),
            };

            agents.push(thread::spawn(move || agent.run()));
        }

        SwarmCoordinator {
            agents,
            channels,
            metrics,
            topology,
        }
    }

    /// Submit a task to the swarm
    pub fn submit_task(&self, task_id: u64, data: Vec<f64>) -> Result<(), String> {
        // Send to coordinator agents
        for (i, channel) in self.channels.iter().enumerate() {
            if i % 5 == 0 { // Coordinators
                channel.send(AgentMessage::Compute { task_id, data: data.clone() })
                    .map_err(|e| format!("Failed to submit task: {}", e))?;
            }
        }
        Ok(())
    }

    /// Get swarm performance metrics
    pub fn get_metrics(&self) -> SwarmMetrics {
        let mut total_messages = 0;
        let mut total_compute_us = 0;
        let mut active_agents = 0;

        for metric in &self.metrics {
            total_messages += metric.messages_processed.load(Ordering::Relaxed);
            total_compute_us += metric.compute_time_us.load(Ordering::Relaxed);
            if metric.is_active.load(Ordering::Relaxed) {
                active_agents += 1;
            }
        }

        SwarmMetrics {
            total_messages,
            total_compute_us,
            active_agents,
            agent_count: self.metrics.len(),
            throughput: if total_compute_us > 0 {
                (total_messages as f64 * 1_000_000.0) / total_compute_us as f64
            } else {
                0.0
            },
        }
    }

    /// Shutdown the swarm gracefully
    pub fn shutdown(&self) {
        for metric in &self.metrics {
            metric.is_active.store(false, Ordering::SeqCst);
        }
    }
}

#[derive(Debug)]
pub struct SwarmMetrics {
    pub total_messages: u64,
    pub total_compute_us: u64,
    pub active_agents: usize,
    pub agent_count: usize,
    pub throughput: f64, // messages per second
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_real_swarm_creation() {
        let swarm = SwarmCoordinator::new(10, SwarmTopology::Mesh);

        // Give agents time to start
        thread::sleep(Duration::from_millis(10));

        let metrics = swarm.get_metrics();
        assert_eq!(metrics.agent_count, 10);
        assert!(metrics.active_agents > 0);

        swarm.shutdown();
    }

    #[test]
    fn test_task_submission() {
        let swarm = SwarmCoordinator::new(5, SwarmTopology::Star);

        let data = vec![1.0, 2.0, 3.0, 4.0];
        swarm.submit_task(1, data).unwrap();

        // Give agents time to process
        thread::sleep(Duration::from_millis(50));

        let metrics = swarm.get_metrics();
        assert!(metrics.total_messages > 0);

        swarm.shutdown();
    }
}