Skip to main content

ghostlink_core/
planning.rs

1//! Greedy Layer Assignment with Fault Tolerance and Adaptive Quantization
2//!
3//! This module provides:
4//! - Sequential greedy layer splitting across nodes based on VRAM capacity
5//! - Adaptive quantization trigger (select_quantization_mode)
6//! - Load balancing and fault detection integration
7
8use crate::accelerator::ExecutionBackend;
9use crate::cluster::ClusterState;
10use crate::cluster::NodeStatus;
11use crate::host::{AccelerationMode, RuntimeProfile};
12use crate::protocol::NodeResources;
13
14/// Delivery ratio thresholds for adaptive quantization
15pub const DELIVERY_RATIO_INT8_THRESHOLD: f32 = 0.95;
16pub const DELIVERY_RATIO_INT4_THRESHOLD: f32 = 0.80;
17
18/// Layer specification with VRAM requirements
19#[derive(Clone, Debug, PartialEq)]
20pub struct LayerSpec {
21    /// Layer index (0-based)
22    pub index: usize,
23    /// VRAM required in GB
24    pub vram_gb: f32,
25    /// Number of weights in the layer
26    pub num_weights: u32,
27}
28
29impl Default for LayerSpec {
30    fn default() -> Self {
31        Self {
32            index: 0,
33            vram_gb: 1.0,
34            num_weights: 0,
35        }
36    }
37}
38
39/// Layer assignment to a specific node
40#[derive(Clone, Debug, PartialEq)]
41pub struct LayerAssignment {
42    /// Node ID
43    pub node_id: String,
44    /// Start layer index (inclusive)
45    pub start_layer: usize,
46    /// End layer index (exclusive)
47    pub end_layer: usize,
48    /// VRAM used on this node
49    pub used_vram_gb: f32,
50    /// Number of layers assigned
51    pub num_layers: usize,
52}
53
54impl LayerAssignment {
55    /// Create new layer assignment
56    pub fn new(node_id: String, start_layer: usize, end_layer: usize, vram_gb: f32) -> Self {
57        Self {
58            node_id,
59            start_layer,
60            end_layer,
61            used_vram_gb: vram_gb,
62            num_layers: end_layer - start_layer,
63        }
64    }
65
66    /// Get average VRAM per layer
67    pub fn avg_vram_per_layer(&self) -> f32 {
68        if self.num_layers == 0 {
69            0.0
70        } else {
71            self.used_vram_gb / self.num_layers as f32
72        }
73    }
74}
75
76/// Quantization mode enumeration
77#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
78pub enum QuantizationMode {
79    /// No quantization (full precision)
80    #[default]
81    None,
82    /// 8-bit quantization
83    Int8,
84    /// 4-bit quantization
85    Int4,
86}
87
88/// Layer placement plan across nodes
89#[derive(Clone, Debug, Default)]
90pub struct PlacementPlan {
91    /// Assignments per node
92    pub assignments: Vec<LayerAssignment>,
93    /// Selected quantization mode
94    pub quantization_mode: QuantizationMode,
95    /// Total layers assigned
96    pub total_layers: usize,
97    /// Nodes participating in plan
98    pub participating_nodes: Vec<String>,
99}
100
101impl PlacementPlan {
102    /// Create new placement plan
103    pub fn new(assignments: Vec<LayerAssignment>, quantization_mode: QuantizationMode) -> Self {
104        let participating_nodes = assignments.iter().map(|a| a.node_id.clone()).collect();
105        let total_layers = assignments.iter().map(|a| a.num_layers).sum();
106
107        Self {
108            assignments,
109            quantization_mode,
110            total_layers,
111            participating_nodes,
112        }
113    }
114
115    /// Get human-readable plan summary
116    pub fn summary(&self) -> String {
117        let mode_str = match self.quantization_mode {
118            QuantizationMode::None => "Full Precision",
119            QuantizationMode::Int8 => "8-bit Quantized",
120            QuantizationMode::Int4 => "4-bit Quantized",
121        };
122
123        format!(
124            "Placement Plan ({})\n\
125             =================\n\
126             Total layers: {}\n\
127             Quantization: {}\n\
128             Nodes: {}\n",
129            mode_str,
130            self.total_layers,
131            match self.quantization_mode {
132                QuantizationMode::None => "Full Precision".to_string(),
133                QuantizationMode::Int8 => "8-bit Quantized".to_string(),
134                QuantizationMode::Int4 => "4-bit Quantized".to_string(),
135            },
136            self.participating_nodes.join(", ")
137        )
138    }
139}
140
141/// Runtime-aware planning hints derived from host auto-detection.
142#[derive(Clone, Copy, Debug, PartialEq, Eq)]
143pub struct PlanningTuning {
144    /// Preferred maximum chunk size to expose parallel work to workers.
145    pub max_layers_per_assignment: usize,
146}
147
148impl PlanningTuning {
149    /// Derive planning hints from the detected runtime profile.
150    pub fn from_runtime_profile(profile: &RuntimeProfile, total_layers: usize) -> Self {
151        let backend = ExecutionBackend::from_runtime_profile(profile);
152        let worker_count = backend.worker_count.max(1);
153        let accelerator_bias = match profile.acceleration_mode {
154            AccelerationMode::Gpu => 2,
155            AccelerationMode::Avx512 => 1,
156            _ => 0,
157        };
158        let vector_bias = (backend.vector_width_bits / 256).max(1);
159        let target_chunks = (worker_count + accelerator_bias + vector_bias - 1).max(1);
160        let chunk_size = if total_layers == 0 {
161            1
162        } else {
163            total_layers.div_ceil(target_chunks).max(1)
164        };
165
166        Self {
167            max_layers_per_assignment: chunk_size,
168        }
169    }
170}
171
172/// Select quantization mode based on cluster health metrics
173pub fn select_quantization_mode(delivery_ratio: f32) -> QuantizationMode {
174    if delivery_ratio >= DELIVERY_RATIO_INT8_THRESHOLD {
175        QuantizationMode::None
176    } else if delivery_ratio >= DELIVERY_RATIO_INT4_THRESHOLD {
177        QuantizationMode::Int8
178    } else {
179        QuantizationMode::Int4
180    }
181}
182
183/// Assign layers sequentially across nodes based on VRAM capacity.
184///
185/// OPTIMIZATION: Replaces per-layer `Option<LayerAssignment>` state tracking and repeated `.as_mut()`
186/// calls with a single-pass index range and VRAM accumulator. Pre-allocates output capacity for `nodes.len()`
187/// and constructs `LayerAssignment` objects only once per node transition or stream completion,
188/// preserving exact `LayerSpec.index` range boundaries while reducing allocation churn and instructions.
189pub fn assign_layers_sequentially(
190    nodes: &[NodeResources],
191    layers: &[LayerSpec],
192) -> Result<Vec<LayerAssignment>, String> {
193    if nodes.is_empty() {
194        return Err("at least one node is required".into());
195    }
196    if layers.is_empty() {
197        return Ok(Vec::new());
198    }
199
200    let mut assignments = Vec::with_capacity(nodes.len());
201    let mut node_idx = 0usize;
202    let mut remaining_capacity = nodes[0].vram_gb;
203    let mut current_start = 0usize;
204    let mut current_vram = 0.0f32;
205
206    for (i, layer) in layers.iter().enumerate() {
207        while layer.vram_gb > remaining_capacity {
208            // Need to move to next node: flush current accumulated layer assignment
209            if i > current_start {
210                let start_layer = layers[current_start].index;
211                let end_layer = layers[i - 1].index + 1;
212                assignments.push(LayerAssignment::new(
213                    nodes[node_idx].id.clone(),
214                    start_layer,
215                    end_layer,
216                    current_vram,
217                ));
218            }
219
220            node_idx += 1;
221            if node_idx >= nodes.len() {
222                return Err(format!(
223                    "insufficient cluster VRAM for layer {} (needs {:.2} GB)",
224                    layer.index, layer.vram_gb
225                ));
226            }
227            remaining_capacity = nodes[node_idx].vram_gb;
228            current_start = i;
229            current_vram = 0.0;
230        }
231
232        // Accumulate layer onto current node
233        remaining_capacity -= layer.vram_gb;
234        current_vram += layer.vram_gb;
235    }
236
237    // Finalize last assignment for remaining layers
238    if current_start < layers.len() {
239        let start_layer = layers[current_start].index;
240        let end_layer = layers[layers.len() - 1].index + 1;
241        assignments.push(LayerAssignment::new(
242            nodes[node_idx].id.clone(),
243            start_layer,
244            end_layer,
245            current_vram,
246        ));
247    }
248
249    Ok(assignments)
250}
251
252/// Split large node assignments into smaller contiguous chunks for worker-level parallelism.
253pub fn chunk_assignments_for_workers(
254    assignments: Vec<LayerAssignment>,
255    max_layers_per_assignment: usize,
256) -> Vec<LayerAssignment> {
257    let chunk_size = max_layers_per_assignment.max(1);
258
259    let needs_chunking = assignments.iter().any(|a| a.num_layers > chunk_size);
260    if !needs_chunking {
261        return assignments;
262    }
263
264    let mut chunked = Vec::with_capacity(assignments.len() * 2);
265
266    for assignment in assignments {
267        if assignment.num_layers <= chunk_size {
268            chunked.push(assignment);
269            continue;
270        }
271
272        let avg_vram = assignment.avg_vram_per_layer();
273        let mut start_layer = assignment.start_layer;
274
275        while start_layer < assignment.end_layer {
276            let end_layer = (start_layer + chunk_size).min(assignment.end_layer);
277            let num_layers = end_layer - start_layer;
278            chunked.push(LayerAssignment::new(
279                assignment.node_id.clone(),
280                start_layer,
281                end_layer,
282                avg_vram * num_layers as f32,
283            ));
284            start_layer = end_layer;
285        }
286    }
287
288    chunked
289}
290
291/// Assign layers sequentially across nodes, directly producing chunked assignments
292/// sized for worker-level parallelism. Avoids creating intermediate full-size
293/// `LayerAssignment` objects and their associated allocation/clone overhead.
294fn assign_layers_chunked(
295    nodes: &[NodeResources],
296    layers: &[LayerSpec],
297    max_layers_per_assignment: usize,
298) -> Result<Vec<LayerAssignment>, String> {
299    if nodes.is_empty() {
300        return Err("at least one node is required".into());
301    }
302    if layers.is_empty() {
303        return Ok(Vec::new());
304    }
305
306    let chunk_size = max_layers_per_assignment.max(1);
307    let est_assignments = (layers.len().div_ceil(chunk_size)).max(nodes.len());
308    let mut assignments = Vec::with_capacity(est_assignments);
309    let mut node_idx = 0usize;
310    let mut remaining_vram = nodes[0].vram_gb;
311    // Index of the first layer in the current chunk
312    let mut chunk_start = 0usize;
313    let mut chunk_vram = 0.0f32;
314
315    for (i, layer) in layers.iter().enumerate() {
316        // Move to next node if current node is out of capacity
317        while layer.vram_gb > remaining_vram {
318            if i > chunk_start {
319                assignments.push(LayerAssignment::new(
320                    nodes[node_idx].id.clone(),
321                    chunk_start,
322                    i,
323                    chunk_vram,
324                ));
325            }
326            node_idx += 1;
327            if node_idx >= nodes.len() {
328                return Err(format!(
329                    "insufficient cluster VRAM for layer {} (needs {:.2} GB)",
330                    layer.index, layer.vram_gb
331                ));
332            }
333            remaining_vram = nodes[node_idx].vram_gb;
334            chunk_start = i;
335            chunk_vram = 0.0;
336        }
337
338        remaining_vram -= layer.vram_gb;
339        chunk_vram += layer.vram_gb;
340
341        // Flush chunk if it reached the maximum chunk size
342        let chunk_len = i + 1 - chunk_start;
343        if chunk_len >= chunk_size {
344            assignments.push(LayerAssignment::new(
345                nodes[node_idx].id.clone(),
346                chunk_start,
347                i + 1,
348                chunk_vram,
349            ));
350            chunk_start = i + 1;
351            chunk_vram = 0.0;
352        }
353    }
354
355    // Finalize last chunk. Checked by pending-layer range, not accumulated
356    // VRAM: a trailing run of zero-VRAM layers (e.g. bias-only layers) would
357    // otherwise leave `chunk_vram == 0.0` and be silently dropped from the
358    // plan even though `chunk_start..layers.len()` is non-empty.
359    if chunk_start < layers.len() {
360        assignments.push(LayerAssignment::new(
361            nodes[node_idx].id.clone(),
362            chunk_start,
363            layers.len(),
364            chunk_vram,
365        ));
366    }
367
368    Ok(assignments)
369}
370
371/// Assign layers using runtime auto-detection to expose worker-parallel chunks.
372pub fn assign_layers_with_runtime_profile(
373    nodes: &[NodeResources],
374    layers: &[LayerSpec],
375    profile: &RuntimeProfile,
376) -> Result<Vec<LayerAssignment>, String> {
377    let tuning = PlanningTuning::from_runtime_profile(profile, layers.len());
378    assign_layers_chunked(nodes, layers, tuning.max_layers_per_assignment)
379}
380
381/// Assign layers with fault tolerance and load balancing
382pub fn assign_layers_with_fault_tolerance(
383    cluster: &ClusterState,
384    layers: &[LayerSpec],
385) -> Result<PlacementPlan, String> {
386    let nodes = cluster.nodes_snapshot();
387
388    if nodes.is_empty() {
389        return Err("no nodes available".into());
390    }
391
392    // First pass: greedy assignment
393    let assignments = assign_layers_sequentially(&nodes, layers)?;
394
395    // Calculate average delivery ratio across all nodes in a single lock pass
396    // (avoids the clone/allocation of active_nodes() and the lock contention
397    // of taking it twice). Also avoids a real divide-by-zero/TOCTOU hazard:
398    // calling `active_nodes()` twice (once for the sum, once for the count)
399    // risked dividing by a different node count than was summed if a node's
400    // status changed between calls, and an empty active set (e.g. all nodes
401    // failed but still registered) would divide by zero.
402    let (sum_delivery, active_count) = {
403        let metrics = cluster
404            .metrics
405            .lock()
406            .unwrap_or_else(|poison| poison.into_inner());
407        let mut sum = 0.0f32;
408        let mut count = 0usize;
409        for metric in metrics.values() {
410            if metric.status == NodeStatus::Active {
411                sum += metric.delivery_ratio;
412                count += 1;
413            }
414        }
415        (sum, count)
416    };
417
418    // Fall back to 0.0 (not 1.0) when there are zero active nodes to sample:
419    // this selects the most conservative quantization mode below. Assuming a
420    // perfect 1.0 ratio when we have literally no corroborating health data
421    // is optimistic in exactly the scenario (all nodes failed/degraded) where
422    // it's least likely to be true.
423    let total_delivery_ratio = if active_count > 0 {
424        sum_delivery / active_count as f32
425    } else {
426        0.0
427    };
428
429    // Select quantization mode based on health
430    let quantization_mode = select_quantization_mode(total_delivery_ratio);
431
432    Ok(PlacementPlan::new(assignments, quantization_mode))
433}
434
435/// Assign layers with fault tolerance and runtime-aware chunking.
436///
437/// Optimized to directly call `assign_layers_chunked`, bypassing intermediate
438/// `LayerAssignment` allocations and `chunk_assignments_for_workers` overhead.
439pub fn assign_layers_with_fault_tolerance_and_runtime(
440    cluster: &ClusterState,
441    layers: &[LayerSpec],
442    profile: &RuntimeProfile,
443) -> Result<PlacementPlan, String> {
444    let nodes = cluster.nodes_snapshot();
445
446    if nodes.is_empty() {
447        return Err("no nodes available".into());
448    }
449
450    let tuning = PlanningTuning::from_runtime_profile(profile, layers.len());
451    // Directly run the chunked layout computation
452    let assignments = assign_layers_chunked(&nodes, layers, tuning.max_layers_per_assignment)?;
453
454    // Calculate average delivery ratio across all nodes in a single lock pass
455    let (sum_delivery, active_count) = {
456        let metrics = cluster
457            .metrics
458            .lock()
459            .unwrap_or_else(|poison| poison.into_inner());
460        let mut sum = 0.0f32;
461        let mut count = 0usize;
462        for metric in metrics.values() {
463            if metric.status == NodeStatus::Active {
464                sum += metric.delivery_ratio;
465                count += 1;
466            }
467        }
468        (sum, count)
469    };
470
471    let total_delivery_ratio = if active_count > 0 {
472        sum_delivery / active_count as f32
473    } else {
474        0.0
475    };
476
477    // Select quantization mode based on health
478    let quantization_mode = select_quantization_mode(total_delivery_ratio);
479
480    Ok(PlacementPlan::new(assignments, quantization_mode))
481}
482
483/// Update layer assignments based on node health metrics
484pub fn rebalance_assignments(cluster: &ClusterState, plan: &mut PlacementPlan) -> bool {
485    // Check if any active node has high available VRAM
486    let mut needs_rebalance = false;
487
488    let metrics = cluster
489        .metrics
490        .lock()
491        .unwrap_or_else(|poison| poison.into_inner());
492
493    for assignment in &mut plan.assignments {
494        if let Some(metric) = metrics.get(&assignment.node_id) {
495            if metric.status != NodeStatus::Active {
496                // Skip failed nodes
497                continue;
498            }
499
500            // Check if this node can take more layers
501            let available = metric.available_vram_gb;
502            let avg_layer_size = assignment.avg_vram_per_layer();
503
504            if available > 0.0 && avg_layer_size > 0.0 {
505                let potential_layers = (available / avg_layer_size) as usize;
506
507                // If node has significant capacity, mark for rebalancing
508                if potential_layers > 2 {
509                    needs_rebalance = true;
510                    break;
511                }
512            }
513        }
514    }
515
516    needs_rebalance
517}
518
519/// Simulate layer streaming on a node with metrics updates
520#[cfg(test)]
521pub fn simulate_layer_streaming(
522    node_id: &str,
523    cluster: &ClusterState,
524    start_layer: usize,
525    end_layer: usize,
526) -> Option<crate::cluster::NodeMetrics> {
527    let mut metrics = cluster.get_metrics(node_id)?;
528
529    // Record VRAM usage
530    let num_layers = end_layer - start_layer;
531    let avg_vram = (metrics.available_vram_gb + metrics.vram_gb) / 2.0;
532    let vram_per_layer = avg_vram / num_layers as f32;
533    metrics.record_vram_usage(vram_per_layer * num_layers as f32);
534
535    // Set streaming layers
536    metrics.set_streaming_layers(start_layer, end_layer);
537
538    Some(metrics)
539}
540
541/// Calculate network health across the cluster
542pub fn calculate_cluster_health(cluster: &ClusterState) -> (f32, usize, Vec<String>) {
543    // Acquire the lock on metrics map exactly ONCE to avoid redundant mutex lock/unlock and clones
544    let metrics = cluster
545        .metrics
546        .lock()
547        .unwrap_or_else(|poison| poison.into_inner());
548
549    if metrics.is_empty() {
550        return (0.0, 0, vec![]);
551    }
552
553    let mut sum_delivery_ratio = 0.0f32;
554    let mut active_count = 0usize;
555    let mut failed_nodes = Vec::new();
556
557    // Iterate once over all live node metrics to aggregate statistics and collect failed nodes
558    for (node_id, metric) in metrics.iter() {
559        match metric.status {
560            NodeStatus::Active => {
561                sum_delivery_ratio += metric.delivery_ratio;
562                active_count += 1;
563            }
564            NodeStatus::Failed => {
565                failed_nodes.push(node_id.clone());
566            }
567            _ => {}
568        }
569    }
570
571    let avg_delivery_ratio = if active_count > 0 {
572        sum_delivery_ratio / active_count as f32
573    } else {
574        0.0
575    };
576
577    let failed_count = failed_nodes.len();
578
579    (avg_delivery_ratio, failed_count, failed_nodes)
580}
581
582/// Dynamic migration planner for moving layers between nodes under load.
583#[derive(Debug, Clone)]
584pub struct MigrationPlanner {
585    pub source_node: String,
586    pub target_node: String,
587    pub layers: Vec<usize>,
588    pub est_migration_time_ms: f32,
589}
590
591impl MigrationPlanner {
592    pub fn new(source: String, target: String, layers: Vec<usize>) -> Self {
593        let est_time = layers.len() as f32 * 45.0; // Baseline 45ms per layer migration
594        Self {
595            source_node: source,
596            target_node: target,
597            layers,
598            est_migration_time_ms: est_time,
599        }
600    }
601
602    /// Generate a safe handoff sequence for a migration.
603    pub fn generate_handoff_plan(&self) -> Vec<String> {
604        let mut steps = Vec::new();
605        let target_node = &self.target_node;
606        let source_node = &self.source_node;
607        steps.push(format!(
608            "PREPARE: Target node {target_node} allocating VRAM"
609        ));
610        steps.push(format!(
611            "STREAM: Moving layers {:?} to {target_node}",
612            self.layers
613        ));
614        steps.push(format!("VERIFY: Integrity check on {target_node}"));
615        steps.push(format!("COMMIT: Switch routing to {target_node}"));
616        steps.push(format!("CLEANUP: Free VRAM on source node {source_node}"));
617        steps
618    }
619}
620
621/// Trigger mechanism for dynamic rebalancing based on load/health.
622pub struct RebalanceTrigger {
623    pub min_imbalance_ratio: f32,
624    pub max_p95_latency_ms: f32,
625}
626
627impl Default for RebalanceTrigger {
628    fn default() -> Self {
629        Self {
630            min_imbalance_ratio: 0.25, // 25% drift
631            max_p95_latency_ms: 25.0,
632        }
633    }
634}
635
636impl RebalanceTrigger {
637    pub fn evaluate(
638        &self,
639        cluster: &ClusterState,
640        current_plan: &PlacementPlan,
641    ) -> Option<MigrationPlanner> {
642        let (overtaxed, available) = {
643            let metrics = cluster
644                .metrics
645                .lock()
646                .unwrap_or_else(|poison| poison.into_inner());
647
648            let mut active_count = 0usize;
649            for m in metrics.values() {
650                if m.status == NodeStatus::Active {
651                    active_count += 1;
652                }
653            }
654            if active_count < 2 {
655                return None;
656            }
657
658            let mut overtaxed_name = None;
659            let mut available_name = None;
660
661            for m in metrics.values() {
662                if m.status == NodeStatus::Active {
663                    if m.delivery_ratio < 0.90 || m.avg_latency_us > 15.0 {
664                        overtaxed_name = Some(m.name.clone());
665                    } else if m.available_vram_gb > 8.0 {
666                        available_name = Some(m.name.clone());
667                    }
668                }
669            }
670            (overtaxed_name, available_name)
671        };
672
673        if let (Some(source), Some(target)) = (overtaxed, available) {
674            if source == target {
675                return None;
676            }
677
678            // Find layers to move from overtaxed node
679            let layers_to_move = current_plan
680                .assignments
681                .iter()
682                .find(|a| a.node_id == source)
683                .map(|a| (a.start_layer..a.end_layer).take(2).collect::<Vec<_>>())
684                .unwrap_or_default();
685
686            if !layers_to_move.is_empty() {
687                return Some(MigrationPlanner::new(source, target, layers_to_move));
688            }
689        }
690
691        None
692    }
693}
694
695#[cfg(test)]
696mod tests {
697    use super::*;
698    use crate::host::AccelerationMode;
699
700    fn sample_layers(count: usize, vram_gb: f32) -> Vec<LayerSpec> {
701        (0..count)
702            .map(|index| LayerSpec {
703                index,
704                vram_gb,
705                num_weights: 0,
706            })
707            .collect()
708    }
709
710    #[test]
711    fn greedily_places_layers_across_nodes() {
712        let nodes = vec![
713            NodeResources::new("node-a", 24.0, 64.0, "8.9", None),
714            NodeResources::new("node-b", 12.0, 32.0, "8.6", None),
715        ];
716
717        let assignments = assign_layers_sequentially(&nodes, &sample_layers(33, 1.0)).unwrap();
718
719        assert_eq!(
720            assignments,
721            vec![
722                LayerAssignment {
723                    node_id: "node-a".into(),
724                    start_layer: 0,
725                    end_layer: 24,
726                    used_vram_gb: 24.0,
727                    num_layers: 24,
728                },
729                LayerAssignment {
730                    node_id: "node-b".into(),
731                    start_layer: 24,
732                    end_layer: 33,
733                    used_vram_gb: 9.0,
734                    num_layers: 9,
735                }
736            ]
737        );
738    }
739
740    #[test]
741    fn reports_insufficient_capacity() {
742        let nodes = vec![NodeResources::new("node-a", 2.0, 64.0, "8.9", None)];
743        let error = assign_layers_sequentially(&nodes, &sample_layers(3, 1.0)).unwrap_err();
744
745        assert!(error.contains("insufficient cluster VRAM"));
746    }
747
748    #[test]
749    fn selects_quantization_mode_from_delivery_ratio() {
750        assert_eq!(select_quantization_mode(0.98), QuantizationMode::None);
751        assert_eq!(select_quantization_mode(0.90), QuantizationMode::Int8);
752        assert_eq!(select_quantization_mode(0.75), QuantizationMode::Int4);
753    }
754
755    #[test]
756    fn placement_plan_summary() {
757        let plan = PlacementPlan::new(
758            vec![
759                LayerAssignment::new("node-a".into(), 0, 24, 24.0),
760                LayerAssignment::new("node-b".into(), 24, 33, 9.0),
761            ],
762            QuantizationMode::None,
763        );
764
765        let summary = plan.summary();
766        assert!(summary.contains("Total layers: 33"));
767        assert!(summary.contains("Full Precision"));
768    }
769
770    #[test]
771    fn rebalance_assignments_detects_capacity() {
772        let mut plan = PlacementPlan::new(
773            vec![LayerAssignment::new("node-a".into(), 0, 24, 24.0)],
774            QuantizationMode::None,
775        );
776
777        // Would need actual cluster state to test rebalancing
778        assert!(!rebalance_assignments(&ClusterState::default(), &mut plan));
779    }
780
781    #[test]
782    fn runtime_profile_chunks_large_assignments() {
783        let assignments = vec![LayerAssignment::new("node-a".into(), 0, 12, 12.0)];
784        let chunked = chunk_assignments_for_workers(assignments, 5);
785
786        assert_eq!(chunked.len(), 3);
787        assert_eq!(chunked[0].start_layer, 0);
788        assert_eq!(chunked[0].end_layer, 5);
789        assert_eq!(chunked[2].start_layer, 10);
790        assert_eq!(chunked[2].end_layer, 12);
791    }
792
793    #[test]
794    fn runtime_profile_assignment_keeps_trailing_zero_vram_layers() {
795        // Regression: assign_layers_chunked finalized its last pending chunk
796        // by checking `chunk_vram > 0.0`, so a trailing run of zero-VRAM
797        // layers left a non-empty pending chunk with chunk_vram == 0.0 and
798        // silently dropped those layers from the plan.
799        let nodes = vec![NodeResources::new("node-a", 24.0, 64.0, "8.9", None)];
800        let layers = sample_layers(3, 0.0);
801        let profile = RuntimeProfile {
802            node_resources: NodeResources::new("node-a", 24.0, 64.0, "8.9", None),
803            logical_cores: 4,
804            recommended_workers: 1,
805            acceleration_mode: AccelerationMode::Generic,
806            gpu_backend: crate::host::GpuBackend::Cpu,
807            xdp_supported: false,
808            detection_source: String::from("test"),
809            probe_mode: crate::host::ProbeMode::Fast,
810        };
811
812        let assignments = assign_layers_with_runtime_profile(&nodes, &layers, &profile).unwrap();
813        let total: usize = assignments.iter().map(|a| a.num_layers).sum();
814        assert_eq!(total, 3, "all zero-VRAM layers must still be assigned");
815    }
816
817    #[test]
818    fn fault_tolerance_handles_all_failed_nodes_without_panicking() {
819        // Regression: dividing active_nodes' summed delivery ratio by
820        // active_nodes().len() produced NaN (or a TOCTOU-inconsistent
821        // average) when every registered node's metrics were marked Failed.
822        let cluster = ClusterState::new();
823        cluster.register(NodeResources::new("node-a", 24.0, 64.0, "8.9", None));
824        cluster.mark_failed("node-a");
825
826        let plan = assign_layers_with_fault_tolerance(&cluster, &sample_layers(4, 1.0)).unwrap();
827        assert_eq!(plan.quantization_mode, QuantizationMode::Int4);
828    }
829
830    #[test]
831    fn planning_tuning_scales_chunk_size_with_workers() {
832        let profile = RuntimeProfile {
833            node_resources: NodeResources::new("node-a", 24.0, 64.0, "8.9", None),
834            logical_cores: 16,
835            recommended_workers: 8,
836            acceleration_mode: AccelerationMode::Gpu,
837            gpu_backend: crate::host::GpuBackend::Cuda,
838            xdp_supported: true,
839            detection_source: String::from("test"),
840            probe_mode: crate::host::ProbeMode::Fast,
841        };
842
843        let tuning = PlanningTuning::from_runtime_profile(&profile, 40);
844        assert!(tuning.max_layers_per_assignment <= 4);
845    }
846}