eryon_rt/runtime/

impl_inner.rs

/*
    Appellation: impl_inner <module>
    Contrib: @FL03
*/
use crate::runtime::RuntimeInner;

use crate::actors::prelude::{Driver, TriadDriver, VNode};
use crate::frag::FragTonnetz;
use crate::orcha::Orchestrator;
use crate::scheduler::Scheduler;
use crate::tasks::{Task, TaskStatus, TaskType, TaskTypeKey};
use crate::types::{PerformanceTracker, SystemInfo};
use rshyper::EdgeId;
use rstmt::nrt::Triad;
use scsys::Timestamp;
use std::collections::HashMap;
use std::time::{Duration, Instant};

impl<D> RuntimeInner<D>
where
    D: Driver<Triad>,
{
    /// Create a new runtime
    pub fn new() -> Self {
        let fragment = FragTonnetz::new();
        let orchestrator = Orchestrator::new();
        let system_info = SystemInfo::new();
        let perf_tracker = PerformanceTracker::new();

        Self {
            fragment,
            orchestrator,
            scheduler: Scheduler::new(),
            perf_tracker,
            system_info,
        }
    }
    /// Get a reference to the fragment
    pub fn fragment(&self) -> &FragTonnetz<D> {
        &self.fragment
    }
    /// Get a mutable reference to the fragment
    pub fn fragment_mut(&mut self) -> &mut FragTonnetz<D> {
        &mut self.fragment
    }
    /// Get a reference to the orchestrator
    pub fn orchestrator(&self) -> &Orchestrator {
        &self.orchestrator
    }
    /// returns a mutable reference to the orchestrator
    pub fn orchestrator_mut(&mut self) -> &mut Orchestrator {
        &mut self.orchestrator
    }
    /// returns an immutable reference to the performance tracker
    pub const fn perf_tracker(&self) -> &PerformanceTracker {
        &self.perf_tracker
    }
    /// returns a mutable reference to the performance tracker
    pub fn perf_tracker_mut(&mut self) -> &mut PerformanceTracker {
        &mut self.perf_tracker
    }
    /// Get a performance trend report
    pub fn performance_trend_report(&self) -> String {
        self.perf_tracker.generate_trend_report()
    }
    /// returns a reference to the scheduler
    pub const fn scheduler(&self) -> &Scheduler {
        &self.scheduler
    }
    /// returns a mutable reference to the scheduler
    pub fn scheduler_mut(&mut self) -> &mut Scheduler {
        &mut self.scheduler
    }
    /// returns an immutable reference to the system's info
    pub const fn system_info(&self) -> &SystemInfo {
        &self.system_info
    }
    /// returns a mutable reference to the system's info
    pub fn system_info_mut(&mut self) -> &mut SystemInfo {
        &mut self.system_info
    }
    /// returns a reference to the tasks stored in the scheduler
    pub fn task_queue(&self) -> &[Task] {
        &self.scheduler.task_queue
    }
    /// returns a mutable reference to the task queue within the scheduler
    pub fn task_queue_mut(&mut self) -> &mut Vec<Task> {
        &mut self.scheduler.task_queue
    }
    /// returns an immutable reference to the initialized virtual node's
    pub fn vnodes(&self) -> &HashMap<EdgeId, VNode<D>> {
        self.fragment.partitions()
    }
    /// returns a mutable reference to the initialized virtual nodes
    pub fn vnodes_mut(&mut self) -> &mut HashMap<EdgeId, VNode<D>> {
        self.fragment.partitions_mut()
    }
    /// set the runtimes fragment to the given value
    pub fn set_fragment(&mut self, fragment: FragTonnetz<D>) {
        self.fragment = fragment;
    }
    /// set the runtime's orchestrator
    pub fn set_orchestrator(&mut self, orchestrator: Orchestrator) {
        self.orchestrator = orchestrator;
    }
    /// set the runtime's scheduler
    pub fn set_scheduler(&mut self, scheduler: Scheduler) {
        self.scheduler = scheduler;
    }
    /// consumes the current instance to create another with the given fragment
    pub fn with_fragment(self, fragment: FragTonnetz<D>) -> Self {
        Self { fragment, ..self }
    }
    /// consumes the current instance to create another with the given orchestrator
    pub fn with_orchestrator(self, orchestrator: Orchestrator) -> Self {
        Self {
            orchestrator,
            ..self
        }
    }
    /// consumes the current instance to create another with the given scheduler
    pub fn with_scheduler(self, scheduler: Scheduler) -> Self {
        Self { scheduler, ..self }
    }
    /// Update performance tracking with current metrics
    pub fn update_performance_metrics(&mut self) {
        // Get current metrics
        let metrics = self.runtime_metrics();

        // Record in the tracker
        self.perf_tracker.record_metrics(metrics);
    }
}

impl<D> RuntimeInner<D>
where
    D: Driver<Triad>,
{
    /// Adaptive task scheduling based on system resources and task performance history
    pub fn adaptive_schedule(&mut self) -> crate::Result<()>
    where
        D: TriadDriver,
    {
        // Track historical execution times by task type discriminant (TaskTypeKey)
        let mut task_type_metrics = HashMap::new();

        // Get completed tasks from scheduler
        for completed_task in self.scheduler.completed_tasks() {
            let execution_time = completed_task
                .completed_at
                .unwrap_or_else(Instant::now)
                .duration_since(
                    completed_task
                        .started_at
                        .unwrap_or(completed_task.created_at),
                );

            // Use TaskTypeKey for the hashmap key instead of TaskType
            let key = TaskTypeKey::from(&completed_task.kind);

            let entry = task_type_metrics
                .entry(key)
                .or_insert_with(|| (Duration::from_secs(0), 0));

            entry.0 += execution_time;
            entry.1 += 1;
        }

        // Calculate average execution times
        let mut avg_execution_times = HashMap::new();
        for (key, (total_time, count)) in task_type_metrics {
            if count > 0 {
                let avg_time = total_time / count as u32;
                avg_execution_times.insert(key, avg_time);
            }
        }

        // Calculate current system load
        let system_load = self.system_info.get_current_load();
        let node_loads = self.fragment.get_all_edge_loads();

        // For each queued task, adjust priority
        let now = Instant::now();
        for task in self.task_queue_mut() {
            if matches!(task.status, TaskStatus::Queued) {
                let base_priority = task.priority as f32;

                // Factor 1: Expected execution time - use TaskTypeKey
                let key = TaskTypeKey::from(&task.kind);
                let time_factor = if let Some(avg_time) = avg_execution_times.get(&key) {
                    // Normalize: shorter tasks get higher priority
                    let max_time = Duration::from_secs(60); // Assume 60s max task time
                    1.0 - (avg_time.as_secs_f32() / max_time.as_secs_f32()).min(1.0)
                } else {
                    0.5 // Default for unknown task types
                };

                // Factor 2: System load
                let load_factor = 1.0 - system_load;

                // Factor 3: Task age
                let age = now.duration_since(task.created_at);
                let age_factor = (age.as_secs_f32() / 60.0).min(1.0); // Max boost after 60 seconds

                // Factor 4: Target node resource availability
                let node_factor = match &task.kind {
                    TaskType::Transform { node_id, .. } => {
                        1.0 - node_loads.get(node_id).copied().unwrap_or(0.5)
                    }
                    TaskType::LearnPattern { node_id, .. } => {
                        1.0 - node_loads.get(node_id).copied().unwrap_or(0.5)
                    }
                    _ => 0.5, // Default for tasks not targeting specific nodes
                };

                // Compute adjusted priority
                let adjusted_priority = base_priority
                    * (0.3 * time_factor
                        + 0.2 * load_factor
                        + 0.2 * age_factor
                        + 0.3 * node_factor);

                // Update task's effective priority (but keep original base priority)
                task.effective_priority = adjusted_priority.round() as usize;
            }
        }

        // Sort by effective priority (higher first)
        self.task_queue_mut()
            .sort_by(|a, b| b.effective_priority.cmp(&a.effective_priority));

        Ok(())
    }
    /// Add a triad to the fragment
    pub fn add_triad(&mut self, triad: Triad) -> crate::Result<rshyper::EdgeId> {
        self.fragment.add_triad(triad)
    }
    /// Auto-configure the runtime based on available system resources
    pub fn auto_configure(&mut self) -> crate::Result<()> {
        // Calculate optimal number of partitions based on CPU cores
        let optimal_partitions = self.system_info.get_cpu_cores().max(1);

        // If we already have enough partitions, we're good
        if self.fragment.total_partitions() >= optimal_partitions {
            return Ok(());
        }

        // Otherwise, initialize with all 24 major and minor triads
        self.init_complete_layer()
    }

    pub fn completed_tasks(&self) -> Vec<&Task> {
        self.scheduler.completed_tasks()
    }
    /// Coordinate distributed learning with gradient sharing
    pub fn coordinate_distributed_learning(&mut self) -> crate::Result<usize>
    where
        D: TriadDriver,
    {
        // Group nodes by chord class
        let mut nodes_by_class = std::collections::HashMap::new();

        for (&edge_id, node) in self.fragment.partitions() {
            if node.has_surface_network() {
                nodes_by_class
                    .entry(node.class())
                    .or_insert_with(Vec::new)
                    .push(edge_id);
            }
        }

        // For each class, coordinate learning
        for (_class, node_ids) in nodes_by_class {
            if node_ids.len() <= 1 {
                continue;
            }

            // 1. Collect gradients from all nodes
            let mut all_gradients = Vec::new();

            for node_id in &node_ids {
                if let Some(node) = self.fragment.get_vnode(node_id) {
                    let gradients = node.extract_learning_gradients()?;
                    all_gradients.push((node_id, gradients));
                }
            }

            // 2. Average the gradients
            let mut averaged_gradients = HashMap::new();

            for (_node_id, gradients) in all_gradients {
                for (pattern_id, gradient) in gradients {
                    averaged_gradients
                        .entry(pattern_id)
                        .or_insert_with(Vec::new)
                        .push(gradient);
                }
            }

            // 3. Apply averaged gradients back to nodes
            for node_id in &node_ids {
                if let Some(node) = self.fragment.get_vnode_mut(node_id) {
                    node.apply_federated_gradients(&averaged_gradients)?;
                }
            }

            // 4. Schedule evaluation task
            self.schedule_task(TaskType::EvaluatePerformance, 8);
        }

        // Record this coordination activity
        let task_id = self.schedule_task(TaskType::SharePatterns, 6);

        Ok(task_id)
    }
    /// Coordinate knowledge sharing between surface networks
    pub fn coordinate_knowledge_transfer(&mut self) -> crate::Result<()>
    where
        D: TriadDriver,
    {
        // 1. Identify knowledge-bearing nodes
        let nodes_with_knowledge: Vec<_> = self
            .fragment
            .partitions()
            .iter()
            .filter(|(_, node)| node.has_surface_network() && node.total_features() > 10)
            .map(|(&id, _)| id)
            .collect();

        if nodes_with_knowledge.len() <= 1 {
            return Ok(()); // Nothing to coordinate
        }

        // 2. Extract knowledge patterns
        let mut knowledge_patterns = HashMap::new();
        for node_id in &nodes_with_knowledge {
            if let Some(node) = self.fragment.get_vnode(node_id) {
                // Extract discrete pattern representations
                match node.extract_knowledge_patterns() {
                    Ok(patterns) => {
                        if !patterns.is_empty() {
                            knowledge_patterns.insert(*node_id, patterns);
                        }
                    }
                    Err(_e) => {
                        #[cfg(feature = "tracing")]
                        tracing::warn!(
                            "Failed to extract knowledge from node {}: {:?}",
                            node_id,
                            _e
                        );
                        continue;
                    }
                }
            }
        }

        if knowledge_patterns.is_empty() {
            return Ok(());
        }

        // 3. Generate a consensus hash for verification
        let consensus_hash = self.compute_knowledge_consensus_hash(&knowledge_patterns);

        // 4. Calculate all proximities first to avoid borrowing conflicts
        let mut transfer_data = Vec::new();

        for &target_id in &nodes_with_knowledge {
            let mut source_pattern_data = Vec::new();

            for (&source_id, patterns) in &knowledge_patterns {
                if source_id != target_id {
                    // Calculate proximity in the tonal space
                    let proximity = if let Some(distance) =
                        self.fragment.voice_leading_distance(&target_id, &source_id)
                    {
                        1.0 / (1.0 + distance as f32 / 6.0) // Normalize by octave divisions
                    } else {
                        0.5 // Default if we can't calculate distance
                    };

                    // Only integrate if there's significant proximity
                    if proximity > 0.3 {
                        // Select patterns to integrate based on proximity
                        let pattern_count = (patterns.len() as f32 * proximity).ceil() as usize;
                        if pattern_count > 0 {
                            let selected_patterns = if pattern_count >= patterns.len() {
                                patterns.clone()
                            } else {
                                // Take most important patterns first
                                patterns.iter().take(pattern_count).cloned().collect()
                            };

                            source_pattern_data.push((source_id, selected_patterns));
                        }
                    }
                }
            }

            // Store all transfer data for this target
            if !source_pattern_data.is_empty() {
                transfer_data.push((target_id, source_pattern_data));
            }
        }

        // 5. Now perform the actual knowledge transfer without concurrent borrowing
        for (target_id, source_patterns) in transfer_data {
            if let Some(target_node) = self.fragment.get_vnode_mut(&target_id) {
                // Apply knowledge from sources
                for (source_id, patterns) in source_patterns {
                    if let Err(e) = target_node.integrate_external_knowledge(&patterns, &source_id)
                    {
                        #[cfg(feature = "tracing")]
                        tracing::error!("Failed to integrate knowledge: {:?}", e);
                        #[cfg(not(feature = "tracing"))]
                        eprintln!("Failed to integrate knowledge: {:?}", e);
                    }
                }
            }
        }

        // 6. Register the knowledge transfer event
        self.register_knowledge_transfer(consensus_hash, nodes_with_knowledge.len());

        Ok(())
    }
    /// coordinate learning between nodes
    pub fn coordinate_learning(&mut self) -> crate::Result<()>
    where
        D: TriadDriver,
    {
        // 1. Identify knowledge-bearing nodes
        let nodes_with_knowledge: Vec<_> = self
            .fragment
            .partitions()
            .iter()
            .filter(|(_, node)| node.has_surface_network() && node.total_features() > 0)
            .map(|(&id, _)| id)
            .collect();

        if nodes_with_knowledge.len() <= 1 {
            return Ok(()); // Nothing to coordinate
        }

        // 2. Extract patterns once per node - this is our main optimization
        let mut extracted_patterns = Vec::with_capacity(nodes_with_knowledge.len());

        for &node_id in &nodes_with_knowledge {
            if let Some(node) = self.get_vnode(&node_id) {
                match node.extract_knowledge_patterns() {
                    Ok(patterns) if !patterns.is_empty() => {
                        extracted_patterns.push((node_id, patterns));
                    }
                    _ => continue,
                }
            }
        }

        // 3. Use a constant limit on targets per source to control complexity
        let max_targets_per_source = core::cmp::min(5, nodes_with_knowledge.len() - 1);

        // 4. Perform transfers using the extracted patterns
        for (source_id, patterns) in extracted_patterns {
            // Skip empty pattern sets
            if patterns.is_empty() {
                continue;
            }

            // Choose target nodes (limited number)
            let mut target_count = 0;

            for target_id in &nodes_with_knowledge {
                if target_id != &source_id && target_count < max_targets_per_source {
                    if let Some(target_node) = self.get_vnode_mut(target_id) {
                        if let Err(_e) =
                            target_node.integrate_external_knowledge(&patterns, &source_id)
                        {
                            #[cfg(feature = "tracing")]
                            tracing::error!("Failed to integrate knowledge: {_e:?}");
                            #[cfg(not(feature = "tracing"))]
                            eprintln!("Failed to integrate knowledge: {_e:?}");
                        }

                        target_count += 1;
                    }
                }
            }
        }

        // 5. Record coordination event
        for node in self.fragment_mut().partitions_mut().values_mut() {
            node.store_mut().record_event(
                "knowledge_coordination",
                Some(vec![nodes_with_knowledge.len()]),
            );
        }

        Ok(())
    }
    /// executes the task associated with the given identitifier
    pub fn execute_task(&mut self, task_id: usize) -> crate::Result<()>
    where
        D: TriadDriver,
    {
        let task = self
            .scheduler
            .get_task(task_id)
            .ok_or_else(|| crate::RuntimeError::TaskNotFound(task_id))?;
        self.orchestrator.execute_task(&mut self.fragment, task)
    }
    /// Execute tasks in the queue
    pub fn execute_tasks(&mut self) -> crate::Result<()>
    where
        D: TriadDriver,
    {
        // Use the scheduler to process tasks
        let _res = self
            .scheduler
            .process_batch(&mut self.fragment, &mut self.orchestrator)?;
        Ok(())
    }
    /// returns an immutable reference to the node with the given identifier; if any
    pub fn get_vnode(&self, id: &EdgeId) -> Option<&VNode<D>> {
        self.fragment().get_vnode(id)
    }
    /// returns a mutable reference to the partition with the given identifier; if any
    pub fn get_vnode_mut(&mut self, id: &EdgeId) -> Option<&mut VNode<D>> {
        self.fragment_mut().get_vnode_mut(id)
    }
    /// Initialize the runtime with major and minor triads
    pub fn init_complete_layer(&mut self) -> crate::Result<()> {
        // Initialize the fragment with all triads
        self.fragment.initialize_with_all_triads()?;

        // Initialize the orchestrator
        self.orchestrator.initialize(&self.fragment);

        Ok(())
    }
    /// Initialize surface networks for all nodes in the fragment
    pub fn init_node_surfaces(&mut self) -> crate::Result<()>
    where
        D: TriadDriver,
    {
        let mut initialized_count = 0;

        // Iterate through all partitions
        for node in self.fragment.partitions_mut().values_mut() {
            // Skip nodes that already have a surface network
            if node.has_surface_network() {
                continue;
            }

            // Initialize the surface network for this node
            node.init_surface()?;

            // Track how many we've initialized
            initialized_count += 1;
        }

        if initialized_count > 0 {
            #[cfg(feature = "tracing")]
            tracing::debug!("Initialized {} node surface networks", initialized_count);
        }

        Ok(())
    }
    /// Initialize surface networks with class-specific tuning
    pub fn init_specialized_surfaces(&mut self) -> crate::Result<()>
    where
        D: TriadDriver,
    {
        use crate::nrt::Triads::*;
        // Group nodes by chord class
        let mut nodes_by_class = std::collections::HashMap::new();

        for (&edge_id, node) in self.fragment.partitions() {
            nodes_by_class
                .entry(node.class())
                .or_insert_with(Vec::new)
                .push(edge_id);
        }

        // Parameters optimized for each chord class
        let class_params: HashMap<_, _> = [
            // Major triads: More emphasis on dominant transformations
            (Major, (0.02, 0.6, 0.4)),
            // Minor triads: More sensitivity to subdominant transformations
            (Minor, (0.01, 0.8, 0.5)),
            // Diminished triads: Higher resolution for chromatic movement
            (Diminished, (0.04, 0.7, 0.6)),
            // Augmented triads: Focus on symmetrical properties
            (Augmented, (0.05, 0.5, 0.8)),
        ]
        .into_iter()
        .collect();

        // Apply specialized parameters for each node
        for (class, node_ids) in &nodes_by_class {
            let params = class_params.get(class).unwrap_or(&(0.01, 0.7, 0.5));

            for edge_id in node_ids {
                if let Some(node) = self.fragment.get_vnode_mut(edge_id) {
                    if !node.has_surface_network() {
                        let config = crate::actors::surface::SurfaceModelConfig::new(
                            params.0, params.1, params.2,
                        );
                        // Initialize with class-specific parameters
                        node.surface_mut().map(|i| i.model_mut().set_config(config));
                    }
                }
            }
        }

        #[cfg(feature = "tracing")]
        tracing::debug!(
            "Initialized specialized surface networks for {} classes",
            nodes_by_class.len()
        );

        Ok(())
    }
    /// Optimize memory allocation across nodes
    pub fn optimize_memory_allocation(&mut self) -> crate::Result<()>
    where
        D: TriadDriver,
    {
        // Get current memory usage per node
        let mut node_memory_usage = HashMap::new();
        let mut total_features = 0;

        for (&edge_id, node) in self.fragment.partitions() {
            let feature_count = node.total_features();
            node_memory_usage.insert(edge_id, feature_count);
            total_features += feature_count;
        }

        // Calculate average memory usage
        if self.fragment.total_partitions() == 0 {
            return Ok(());
        }

        let avg_features = total_features / self.fragment.total_partitions();
        let target_max = (avg_features as f32 * 1.5) as usize; // Allow 50% above average

        // Find nodes with excessive features
        let mut nodes_to_optimize = Vec::new();
        for (&edge_id, &feature_count) in &node_memory_usage {
            if feature_count > target_max {
                nodes_to_optimize.push((edge_id, feature_count));
            }
        }

        // Sort by feature count (highest first)
        nodes_to_optimize.sort_by(|a, b| b.1.cmp(&a.1));

        // Optimize each node
        for (edge_id, _) in nodes_to_optimize {
            if let Some(node) = self.fragment.get_vnode_mut(&edge_id) {
                // Consolidate patterns with similar features
                node.consolidate_similar_patterns(0.8)?;

                // Prune low-importance features
                node.prune_low_importance_features(0.2)?;
            }
        }

        Ok(())
    }
    /// Optimize the surface networks across all nodes
    pub fn optimize_surface_networks(&mut self) -> crate::Result<()>
    where
        D: TriadDriver,
    {
        // Group nodes by chord class
        let mut nodes_by_class = std::collections::HashMap::new();

        // Analyze all nodes and group by chord class
        for (&edge_id, node) in self.fragment.partitions() {
            nodes_by_class
                .entry(node.class())
                .or_insert_with(Vec::new)
                .push(edge_id);
        }

        // For each class, find best-performing node and share patterns
        for (_class, node_ids) in nodes_by_class {
            if node_ids.len() <= 1 {
                continue;
            }

            // Find node with most learning
            let mut best_node_id = None;
            let mut most_features = 0;

            for id in &node_ids {
                if let Some(node) = self.fragment.get_vnode(id) {
                    let features = node.total_features();
                    if features > most_features && node.has_surface_network() {
                        best_node_id = Some(*id);
                        most_features = features;
                    }
                }
            }

            // Share patterns from best node to others
            if let Some(source_id) = best_node_id {
                for &target_id in &node_ids {
                    if target_id != source_id {
                        // Create message to share patterns
                        self.orchestrator
                            .send_message(crate::NodeMessage::PatternShare {
                                source: source_id,
                                pattern: vec![], // Will be populated by the node
                                importance: 0.8,
                            });
                    }
                }
            }
        }

        // Process the messages
        self.process_messages()?;

        Ok(())
    }
    /// Get predicted resource requirements for a task
    pub fn predict_task_resources(&self, task_type: &TaskType) -> (f32, f32) {
        // Use discriminant/TaskTypeKey to avoid floating-point issues in pattern matching
        match TaskTypeKey::from(task_type) {
            TaskTypeKey::Transform => (5.0, 50.0),
            TaskTypeKey::LearnPattern => (20.0, 100.0),
            TaskTypeKey::SharePatterns => (10.0, 80.0),
            TaskTypeKey::OptimizeNode => (15.0, 30.0),
            _ => (5.0, 20.0), // Default for unknown tasks
        }
    }
    /// Schedule tasks using predictive resource management
    pub fn predictive_schedule(&mut self, look_ahead_ms: u64) -> crate::Result<()>
    where
        D: TriadDriver,
    {
        // 1. Get current system state
        let current_cpu = self.system_info.get_cpu_usage();
        let current_memory = self.system_info.get_ram_memory_usage();
        let available_memory =
            self.system_info.get_free_memory_mb() as f32 * (1.0 - current_memory);

        // 2. Estimate resource availability window
        // TODO: find a way to use these values
        let _cpu_headroom = 1.0 - current_cpu;
        let _mem_headroom = available_memory;

        // 3. Build resource timeline for currently executing tasks
        let mut cpu_timeline = vec![current_cpu];
        let mut mem_timeline = vec![current_memory];
        let timeline_slots = 10; // Divide look-ahead into slots

        // Get currently running tasks
        let running_tasks = self.scheduler.get_running_tasks();

        // Estimate when each task will complete and free resources
        for slot in 1..timeline_slots {
            let slot_time = Instant::now()
                + Duration::from_millis(look_ahead_ms * slot as u64 / timeline_slots as u64);

            // Start with current usage
            let mut cpu_usage = current_cpu;
            let mut mem_usage = current_memory;

            // Adjust for tasks expected to complete by this time
            for task in &running_tasks {
                let task_started = task.started_at.unwrap_or(task.created_at);
                let (task_cpu, task_mem) = self.predict_task_resources(&task.kind);

                // Estimate task duration based on task type discriminant
                let est_duration = {
                    let task_key = TaskTypeKey::from(&task.kind);
                    self.scheduler.estimate_task_duration(task_key)
                };
                let expected_completion = task_started + est_duration;

                // If task still running at this slot, count its resources
                if expected_completion > slot_time {
                    cpu_usage += task_cpu / 100.0;
                    mem_usage += task_mem / self.system_info.get_free_memory_mb() as f32;
                }
            }

            // Cap at reasonable values
            cpu_usage = cpu_usage.min(1.0).max(0.1);
            mem_usage = mem_usage.min(1.0).max(0.1);

            cpu_timeline.push(cpu_usage);
            mem_timeline.push(mem_usage);
        }

        // 4. Find optimal scheduling slot for each queued task
        let mut scheduled_tasks = Vec::new();

        for task in self.task_queue() {
            if !matches!(task.status, TaskStatus::Queued) {
                continue;
            }

            let (task_cpu, task_mem) = self.predict_task_resources(&task.kind);
            let normalized_cpu = task_cpu / 100.0;
            let normalized_mem = task_mem / self.system_info.get_free_memory_mb() as f32;

            // Find the best slot to schedule this task
            let mut best_slot = 0;
            let mut best_fit = f32::MAX;

            for slot in 0..timeline_slots {
                // Check if resources available in this slot
                let slot_cpu = cpu_timeline[slot];
                let slot_mem = mem_timeline[slot];

                if slot_cpu + normalized_cpu <= 1.0 && slot_mem + normalized_mem <= 1.0 {
                    // Calculate how well the task fits this slot (smaller is better)
                    let fit = normalized_cpu / (1.0 - slot_cpu) + normalized_mem / (1.0 - slot_mem);

                    if fit < best_fit {
                        best_fit = fit;
                        best_slot = slot;
                    }
                }
            }

            // If we found a slot, schedule task with delay
            if best_fit < f32::MAX {
                let delay_ms = look_ahead_ms * best_slot as u64 / timeline_slots as u64;

                // Add task to the scheduled list with its delay
                scheduled_tasks.push((task.id, delay_ms));

                // Update resource timeline
                if best_slot < timeline_slots {
                    for slot in best_slot..timeline_slots {
                        cpu_timeline[slot] += normalized_cpu;
                        mem_timeline[slot] += normalized_mem;
                    }
                }
            }
        }

        // 5. Apply scheduling delays
        for (task_id, delay_ms) in scheduled_tasks {
            if let Some(task) = self.task_queue_mut().iter_mut().find(|t| t.id == task_id) {
                task.scheduled_delay = Some(Duration::from_millis(delay_ms));
            }
        }

        Ok(())
    }
    /// Prioritize tasks based on observer position in Tonnetz space
    pub fn prioritize_tonnetz_tasks(&mut self) -> crate::Result<()>
    where
        D: TriadDriver,
    {
        // 1. Find observer position (a node with Observer role)
        let mut observer_position = None;

        for (&edge_id, node) in self.fragment.partitions() {
            if matches!(node.operator(), crate::actors::Operator::Observer(_)) {
                observer_position = Some(edge_id);
                break;
            }
        }

        // If no observer found, use any node
        let observer_position = observer_position.unwrap_or_else(|| {
            self.fragment
                .partitions()
                .keys()
                .next()
                .copied()
                .unwrap_or_default()
        });

        // 2. Collect tasks and their affected nodes first
        let mut task_adjustments = Vec::new();

        for (idx, task) in self.task_queue().iter().enumerate() {
            if !matches!(task.status, TaskStatus::Queued) {
                continue;
            }

            // Get the node ID affected by this task
            let affected_node = match &task.kind {
                TaskType::Transform { node_id, .. } => Some(*node_id),
                TaskType::OptimizeNode { node_id, .. } => Some(*node_id),
                TaskType::ChangeNodeMode(node_id, _) => Some(*node_id),
                TaskType::LearnPattern { node_id, .. } => Some(*node_id),
                TaskType::EvaluateEnsemble { node_id, .. } => Some(*node_id),
                _ => None, // Tasks that don't target specific nodes
            };

            if let Some(node_id) = affected_node {
                // Calculate voice-leading distance between observer and affected node
                if let Some(distance) = self.voice_leading_distance(&observer_position, &node_id) {
                    // Convert to octave-based distance factor
                    let octave_layer = distance as f32 / 12.0;
                    let distance_factor = (1.0 + octave_layer).recip();

                    // Calculate adjusted priority
                    let base_priority = task.priority as f32;
                    let new_priority = (base_priority * distance_factor).round() as usize;

                    // Store index and new priority for later application
                    task_adjustments.push((idx, new_priority));
                }
            }
        }

        // 3. Apply the priority adjustments after computation is done
        for (idx, priority) in task_adjustments {
            if let Some(task) = self.task_queue_mut().get_mut(idx) {
                task.effective_priority = priority;
            }
        }

        // 4. Re-sort task queue
        self.task_queue_mut()
            .sort_by(|a, b| b.effective_priority.cmp(&a.effective_priority));

        Ok(())
    }
    /// Process messages between nodes
    pub fn process_messages(&mut self) -> crate::Result<()>
    where
        D: TriadDriver,
    {
        self.orchestrator.process_messages(&mut self.fragment)
    }
    /// Run maintenance including knowledge coordination
    pub fn run_maintenance(&mut self) -> crate::Result<Vec<usize>>
    where
        D: TriadDriver,
    {
        let mut task_ids = Vec::new();

        // 1. Memory optimization
        let memory_task_id = self.schedule_task(
            TaskType::OptimizeMemory { max_features: 5000 },
            3, // Low priority
        );
        task_ids.push(memory_task_id);

        // 2. Knowledge transfer (higher priority than memory optimization)
        let knowledge_task_id = self.schedule_knowledge_transfer(6);
        task_ids.push(knowledge_task_id);

        // 3. Resource balancing
        let balance_task_id = self.schedule_task(
            TaskType::BalanceResources,
            4, // Medium priority
        );
        task_ids.push(balance_task_id);

        Ok(task_ids)
    }
    /// Get detailed resource metrics
    pub fn runtime_metrics(&self) -> crate::ResourceMetrics {
        crate::ResourceMetrics::default()
            .with_cpu_usage(self.system_info.get_cpu_usage())
            .with_memory_usage(self.system_info.memory_usage())
            .with_available_memory(self.system_info.get_free_memory_mb())
            .with_partition_count(self.fragment.total_partitions())
            .with_edge_count(self.fragment.total_edges())
            .with_vertex_count(self.fragment.total_nodes())
            .with_queued_tasks(self.task_queue().len())
            .with_running_tasks(self.orchestrator.active_task_count())
            .with_learning_nodes(
                self.fragment
                    .partitions()
                    .values()
                    .filter(|node| node.has_surface_network())
                    .count(),
            )
            .with_feature_count(
                self.fragment
                    .partitions()
                    .values()
                    .map(|node| node.total_features())
                    .sum::<usize>(),
            )
    }
    /// Schedule a knowledge transfer task with the given priority
    pub fn schedule_knowledge_transfer(&mut self, priority: usize) -> usize {
        // Schedule a task for the next step
        self.schedule_task(TaskType::CoordinateLearning, priority)
    }
    /// Schedule memory optimization based on current memory usage
    pub fn schedule_memory_optimization(&mut self) -> crate::Result<usize> {
        // Calculate total features across the fragment
        let total_features: usize = self
            .fragment
            .partitions()
            .values()
            .map(|node| node.total_features())
            .sum();

        // Get node count for average calculation
        let node_count = self.fragment.partitions().len().max(1);

        // Calculate average features per node
        let avg_features_per_node = total_features / node_count;

        // Set target max features - higher for busy fragments
        let max_features = if avg_features_per_node > 1000 {
            // Heavy memory usage - be more aggressive
            avg_features_per_node * 2 / 3
        } else {
            // Normal usage - be conservative
            avg_features_per_node * 3 / 4
        }
        .max(500); // Ensure reasonable minimum

        // Schedule the optimization task with appropriate priority
        let priority = if avg_features_per_node > 2000 {
            // High memory pressure - higher priority
            7
        } else {
            // Normal memory pressure - standard maintenance priority
            3
        };

        // Schedule the task
        let task_id = self.schedule_task(TaskType::OptimizeMemory { max_features }, priority);
        Ok(task_id)
    }
    /// Schedule a task for execution
    pub fn schedule_task(&mut self, task_type: TaskType, priority: usize) -> usize {
        // Use the scheduler to create and queue the task
        self.scheduler.create_task(task_type, priority)
    }
    /// Generate a comprehensive system health report with performance insights
    pub fn system_health_report(&self) -> String {
        let metrics = self.runtime_metrics();
        let info = self.system_info();
        let timestamp = Timestamp::<u64>::now();

        let mut report = String::new();

        // Header with timestamp
        report.push_str(&format!("System Health Report - {}\n", timestamp));
        report.push_str("==================================\n\n");

        // System resources section
        report.push_str("SYSTEM RESOURCES\n");
        report.push_str("-----------------\n");
        report.push_str(&format!("CPU Cores:         {}\n", info.get_cpu_cores()));
        report.push_str(&format!(
            "Memory:            {} MB\n",
            info.get_free_memory_mb()
        ));
        report.push_str(&format!(
            "CPU Usage:         {:.1}%\n",
            metrics.cpu_usage() * 100.0
        ));
        report.push_str(&format!(
            "Memory Usage:      {:.1}%\n",
            metrics.memory_usage() * 100.0
        ));
        report.push_str(&format!(
            "Available Compute: {}%\n\n",
            metrics.available_compute()
        ));

        // Fragment topology section
        report.push_str("TONNETZ TOPOLOGY\n");
        report.push_str("----------------\n");
        report.push_str(&format!(
            "Partitions:        {}\n",
            metrics.partition_count()
        ));
        report.push_str(&format!("Vertices:          {}\n", metrics.vertex_count()));
        report.push_str(&format!("Edges:             {}\n\n", metrics.edge_count()));

        // Runtime activity section
        report.push_str("RUNTIME ACTIVITY\n");
        report.push_str("----------------\n");
        report.push_str(&format!("Queued Tasks:      {}\n", metrics.queued_tasks()));
        report.push_str(&format!("Running Tasks:     {}\n", metrics.running_tasks()));

        // Get task distribution
        let mut task_counts = std::collections::HashMap::new();
        for task in self.task_queue() {
            let key = TaskTypeKey::from(&task.kind);
            *task_counts.entry(key).or_insert(0) += 1;
        }

        // Show task distribution if there are any tasks
        if !task_counts.is_empty() {
            report.push_str("Task Distribution:\n");
            for (task_type, count) in task_counts {
                report.push_str(&format!("  - {}: {}\n", task_type, count));
            }
        }
        report.push('\n');

        // Learning status
        report.push_str("LEARNING STATUS\n");
        report.push_str("--------------\n");
        report.push_str(&format!(
            "Learning Nodes:     {}\n",
            metrics.learning_nodes()
        ));
        report.push_str(&format!(
            "Surface Networks:   {}\n",
            metrics.surface_network_count()
        ));
        report.push_str(&format!(
            "Features:          {}\n\n",
            metrics.feature_count()
        ));

        // Capabilities
        report.push_str("CAPABILITIES\n");
        report.push_str("-----------\n");
        for capability in metrics.capabilities() {
            report.push_str(&format!("- {}\n", capability));
        }
        report.push('\n');

        // Performance assessment
        report.push_str("PERFORMANCE ASSESSMENT\n");
        report.push_str("---------------------\n");

        // Memory pressure assessment
        let memory_ratio =
            (metrics.feature_count() as f32) / (1000.0 * metrics.partition_count().max(1) as f32);
        let memory_status = if memory_ratio > 2.0 {
            "HIGH - Consider memory optimization"
        } else if memory_ratio > 1.0 {
            "MODERATE - Monitor memory growth"
        } else {
            "GOOD - Memory usage is acceptable"
        };
        report.push_str(&format!("Memory Pressure:    {}\n", memory_status));

        // CPU pressure assessment
        let cpu_status = if metrics.cpu_usage() > &0.8 {
            "HIGH - Consider scaling out"
        } else if metrics.cpu_usage() > &0.5 {
            "MODERATE - Monitor CPU usage"
        } else {
            "GOOD - CPU usage is acceptable"
        };
        report.push_str(&format!("CPU Pressure:       {}\n", cpu_status));

        // Task backlog assessment
        let task_ratio = metrics.queued_tasks() as f32 / (info.get_cpu_cores() * 2) as f32;
        let task_status = if task_ratio > 3.0 {
            "HIGH - Large task backlog"
        } else if task_ratio > 1.0 {
            "MODERATE - Tasks may experience delays"
        } else {
            "GOOD - Task processing is timely"
        };
        report.push_str(&format!("Task Backlog:       {}\n", task_status));

        report
    }
    /// Tune surface network parameters based on node performance
    pub fn tune_surface_networks(&mut self) -> crate::Result<()>
    where
        D: TriadDriver,
    {
        // Analyze performance of existing surface networks
        let mut performance_metrics = HashMap::new();

        for (&edge_id, node) in self.fragment.partitions() {
            if node.has_surface_network() {
                // Get learning metrics for this node
                let accuracy = node.get_learning_accuracy();
                let convergence_rate = node.get_convergence_rate();
                let feature_quality = node.get_feature_quality();

                // Combined performance score
                let performance = accuracy * 0.5 + convergence_rate * 0.3 + feature_quality * 0.2;

                performance_metrics.insert(edge_id, performance);
            }
        }

        // Find best and worst performing nodes
        if performance_metrics.is_empty() {
            return Ok(());
        }

        let mut entries: Vec<_> = performance_metrics.iter().collect();
        entries.sort_by(|a, b| b.1.partial_cmp(a.1).unwrap_or(std::cmp::Ordering::Equal));

        let best_node_id = *entries.first().unwrap().0;

        // Get parameters from best node
        if let Some(best_node) = self.fragment().get_vnode(&best_node_id) {
            let optimal_params = best_node.get_surface_parameters();

            // Apply to lower performing nodes
            for (edge_id, performance) in performance_metrics {
                // Only adjust nodes with below-average performance
                if performance < 0.7 && edge_id != best_node_id {
                    if let Some(node) = self.fragment_mut().get_vnode_mut(&edge_id) {
                        // Adjust toward optimal parameters
                        node.adjust_surface_parameters(&optimal_params, 0.3)?;
                    }
                }
            }
        }

        Ok(())
    }
    /// compute the vocal leading distance between two facets; see
    /// [voice_leading_distance](crate::FragTonnetz::voice_leading_distance) for more information
    pub fn voice_leading_distance(&self, from: &EdgeId, to: &EdgeId) -> Option<usize> {
        self.fragment().voice_leading_distance(from, to)
    }
}

impl<D> RuntimeInner<D>
where
    D: Driver<Triad>,
{
    /// Compute a hash representing the consensus state of knowledge
    #[doc(hidden)]
    pub fn compute_knowledge_consensus_hash(
        &self,
        patterns: &HashMap<EdgeId, Vec<Vec<usize>>>,
    ) -> [u8; 32] {
        use std::collections::hash_map::DefaultHasher;
        use std::hash::Hasher;

        // Simple hash implementation - would be replaced by cryptographic consensus
        let mut hasher = DefaultHasher::new();

        // Order by node ID for determinism
        let mut ordered_nodes: Vec<_> = patterns.keys().collect();
        ordered_nodes.sort();

        for &node_id in &ordered_nodes {
            if let Some(node_patterns) = patterns.get(node_id) {
                // Hash node patterns
                hasher.write_usize(**node_id);
                hasher.write_usize(node_patterns.len());

                for pattern in node_patterns {
                    hasher.write_usize(pattern.len());
                    for &value in pattern {
                        hasher.write_usize(value);
                    }
                }
            }
        }

        // Convert to 32 byte array (simulating SHA-256)
        let hash_value = hasher.finish();
        let mut result = [0u8; 32];
        let bytes = hash_value.to_be_bytes();
        result[..bytes.len()].copy_from_slice(&bytes);

        result
    }
    // TODO: integrate with consensus mechanisms & hybrid-multichain models
    /// Register a knowledge transfer event
    #[allow(dead_code)]
    fn register_knowledge_transfer(&mut self, consensus_hash: [u8; 32], node_count: usize) {
        // Record the transfer in system metrics
        let timestamp = Timestamp::<u64>::now();

        // Store in runtime properties
        let transfer_key = format!("knowledge_transfer_{timestamp}");
        let transfer_value = format!("{:?}:{}", consensus_hash, node_count);

        // Store this in the fragment's properties
        for node in self.fragment.partitions_mut().values_mut() {
            node.store_mut()
                .set_property(&transfer_key, &transfer_value);
        }

        // Log the transfer
        #[cfg(feature = "tracing")]
        tracing::info!(
            "Knowledge transfer completed: {} nodes, consensus: {:?}",
            node_count,
            &consensus_hash[0..4]
        );
    }
}

impl<D> Default for RuntimeInner<D>
where
    D: Driver<Triad>,
{
    fn default() -> Self {
        Self::new()
    }
}

impl<D> eryon::Execute<usize> for RuntimeInner<D>
where
    D: TriadDriver,
{
    type Output = crate::Result;
    fn execute(mut self, rhs: usize) -> Self::Output {
        self.execute_task(rhs)
    }
}

impl<D> eryon::ExecuteMut<usize> for RuntimeInner<D>
where
    D: TriadDriver,
{
    type Output = crate::Result;
    fn execute_mut(&mut self, rhs: usize) -> Self::Output {
        self.execute_task(rhs)
    }
}