## Coverage Report

#### Created: 2026-01-25 15:05

next uncovered line (L), next uncovered region (R), next uncovered branch (B)

    /home/noah/src/realizar/src/moe.rs

    Line
    Count
    Source
    1
    //! Mixture-of-Experts (MOE) routing with Capacity Factor load balancing
    2
    //!
    3
    //! Implements inference-time load balancing per Fedus et al. (2022) Switch Transformers.
    4
    //!
    5
    //! ## Features
    6
    //!
    7
    //! - **Power of Two Choices**: Mitzenmacher (2001) load balancing algorithm
    8
    //! - **Capacity Factor Routing**: Fedus et al. (2022) expert capacity limits
    9
    //! - **Circuit Breaker**: Nygard (2018) failure isolation pattern
    10
    //! - **Heijunka Controller**: Toyota Production System load leveling via Little's Law
    11
    //! - **Andon Triggers**: Jidoka (built-in quality) automated quality control
    12
    13
    use std::{
    14
        sync::{
    15
            atomic::{AtomicUsize, Ordering},
    16
            Mutex,
    17
        },
    18
        time::{Duration, Instant},
    19
    };
    20
    21
    use crate::error::{RealizarError, Result};
    22
    23
    /// Configuration for capacity factor routing
    24
    #[derive(Debug, Clone)]
    25
    pub struct CapacityConfig {
    26
        /// Maximum queue depth per expert
    27
        pub capacity: usize,
    28
        /// Number of experts
    29
        pub num_experts: usize,
    30
    }
    31
    32
    /// Capacity Factor Router for inference-time load balancing
    33
    pub struct CapacityFactorRouter {
    34
        config: CapacityConfig,
    35
        queue_depths: Vec<AtomicUsize>,
    36
    }
    37
    38
    impl CapacityFactorRouter {
    39
        /// Create new router
    40
        #[must_use]
    41
    4
        pub fn new(config: CapacityConfig) -> Self {
    42
    4
            let queue_depths = (0..config.num_experts)
    43
    14
                .map4(|_| AtomicUsize::new(0))
    44
    4
                .collect();
    45
    4
            Self {
    46
    4
                config,
    47
    4
                queue_depths,
    48
    4
            }
    49
    4
        }
    50
    51
        /// Route to best expert, falling back if at capacity
    52
        ///
    53
        /// # Errors
    54
        ///
    55
        /// Returns `MoeError` if score count doesn't match expert count.
    56
        /// Returns `ExpertCapacityExceeded` if all top experts are at capacity.
    57
    3
        pub fn route(&self, scores: &[f32]) -> Result<usize> {
    58
    3
            if scores.len() != self.config.num_experts {
    59
    1
                return Err(RealizarError::MoeError(format!(
    60
    1
                    "Expected {} scores, got {}",
    61
    1
                    self.config.num_experts,
    62
    1
                    scores.len()
    63
    1
                )));
    64
    2
            }
    65
    66
    2
            let top2 = Self::top_k_indices(scores, 2);
    67
    2
            let primary = top2[0];
    68
    69
    2
            if self.queue_depths[primary].load(Ordering::Relaxed) < self.config.capacity {
    70
    1
                Ok(primary)
    71
    1
            } else if top2.len() > 1 {
    72
    1
                Ok(top2[1])
    73
            } else {
    74
    0
                Err(RealizarError::ExpertCapacityExceeded {
    75
    0
                    expert_id: primary,
    76
    0
                    queue_depth: self.queue_depths[primary].load(Ordering::Relaxed),
    77
    0
                    capacity: self.config.capacity,
    78
    0
                })
    79
            }
    80
    3
        }
    81
    82
        /// Record expert usage
    83
    2
        pub fn record_start(&self, expert_id: usize) {
    84
    2
            self.queue_depths[expert_id].fetch_add(1, Ordering::Relaxed);
    85
    2
        }
    86
    87
        /// Record expert completion
    88
    1
        pub fn record_end(&self, expert_id: usize) {
    89
    1
            self.queue_depths[expert_id].fetch_sub(1, Ordering::Relaxed);
    90
    1
        }
    91
    92
        /// Get queue depth for expert
    93
        #[must_use]
    94
    3
        pub fn queue_depth(&self, expert_id: usize) -> usize {
    95
    3
            self.queue_depths[expert_id].load(Ordering::Relaxed)
    96
    3
        }
    97
    98
    2
        fn top_k_indices(scores: &[f32], k: usize) -> Vec<usize> {
    99
    2
            let mut indexed: Vec<(usize, f32)> = scores.iter().copied().enumerate().collect();
    100
    8
            indexed2.sort_by2(|a, b| b.1.partial_cmp(&a.1).unwrap_or(std::cmp::Ordering::Equal));
    101
    2
            indexed.into_iter().take(k).map(|(i, _)| i).collect()
    102
    2
        }
    103
    }
    104
    105
    // ============================================================================
    106
    // Power of Two Choices Router (Mitzenmacher 2001)
    107
    // ============================================================================
    108
    109
    /// Configuration for Power of Two Choices routing
    110
    #[derive(Debug, Clone)]
    111
    pub struct PowerOfTwoConfig {
    112
        /// Number of experts available
    113
        pub num_experts: usize,
    114
        /// Maximum queue depth per expert
    115
        pub capacity: usize,
    116
    }
    117
    118
    /// Power of Two Choices Router per Mitzenmacher (2001)
    119
    ///
    120
    /// Instead of always routing to the highest-scoring expert, this router
    121
    /// picks the top 2 experts by score and routes to the *least loaded* one.
    122
    /// This dramatically improves load balancing compared to simple top-k routing.
    123
    ///
    124
    /// ## Algorithm
    125
    ///
    126
    /// 1. Select top-2 experts by score
    127
    /// 2. Compare their current queue depths
    128
    /// 3. Route to the one with lower load (breaking ties by score)
    129
    ///
    130
    /// ## Citation
    131
    ///
    132
    /// Mitzenmacher, M. (2001). "The Power of Two Choices in Randomized Load Balancing."
    133
    /// IEEE Transactions on Parallel and Distributed Systems.
    134
    pub struct PowerOfTwoChoicesRouter {
    135
        config: PowerOfTwoConfig,
    136
        queue_depths: Vec<AtomicUsize>,
    137
    }
    138
    139
    impl PowerOfTwoChoicesRouter {
    140
        /// Create a new Power of Two Choices router
    141
        #[must_use]
    142
    3
        pub fn new(config: PowerOfTwoConfig) -> Self {
    143
    3
            let queue_depths = (0..config.num_experts)
    144
    10
                .map3(|_| AtomicUsize::new(0))
    145
    3
                .collect();
    146
    3
            Self {
    147
    3
                config,
    148
    3
                queue_depths,
    149
    3
            }
    150
    3
        }
    151
    152
        /// Route request using Power of Two Choices algorithm
    153
        ///
    154
        /// # Errors
    155
        ///
    156
        /// Returns error if score count doesn't match expert count or all top experts at capacity.
    157
    3
        pub fn route(&self, scores: &[f32]) -> Result<usize> {
    158
    3
            if scores.len() != self.config.num_experts {
    159
    0
                return Err(RealizarError::MoeError(format!(
    160
    0
                    "Expected {} scores, got {}",
    161
    0
                    self.config.num_experts,
    162
    0
                    scores.len()
    163
    0
                )));
    164
    3
            }
    165
    166
            // Get top 2 experts by score
    167
    3
            let top2 = Self::top_k_indices(scores, 2);
    168
    169
            // Check both for capacity and pick least loaded
    170
    3
            let mut best_choice = None;
    171
    3
            let mut best_load = usize::MAX;
    172
    173
    9
            for &expert_id6 in &top2 {
    174
    6
                let load = self.queue_depths[expert_id].load(Ordering::Relaxed);
    175
    6
                if load < self.config.capacity && load < best_load4 {
    176
    3
                    best_load = load;
    177
    3
                    best_choice = Some(expert_id);
    178
    3
                }
    179
            }
    180
    181
    3
            best_choice.ok_or_else(|| RealizarError::ExpertCapacityExceeded {
    182
    1
                expert_id: top2[0],
    183
    1
                queue_depth: self.queue_depths[top2[0]].load(Ordering::Relaxed),
    184
    1
                capacity: self.config.capacity,
    185
    1
            })
    186
    3
        }
    187
    188
        /// Record that an expert started processing a request
    189
    60
        pub fn record_start(&self, expert_id: usize) {
    190
    60
            self.queue_depths[expert_id].fetch_add(1, Ordering::Relaxed);
    191
    60
        }
    192
    193
        /// Record that an expert finished processing a request
    194
    0
        pub fn record_end(&self, expert_id: usize) {
    195
    0
            self.queue_depths[expert_id].fetch_sub(1, Ordering::Relaxed);
    196
    0
        }
    197
    198
        /// Get current queue depth for an expert
    199
        #[must_use]
    200
    0
        pub fn queue_depth(&self, expert_id: usize) -> usize {
    201
    0
            self.queue_depths[expert_id].load(Ordering::Relaxed)
    202
    0
        }
    203
    204
    3
        fn top_k_indices(scores: &[f32], k: usize) -> Vec<usize> {
    205
    3
            let mut indexed: Vec<(usize, f32)> = scores.iter().copied().enumerate().collect();
    206
    9
            indexed3.sort_by3(|a, b| b.1.partial_cmp(&a.1).unwrap_or(std::cmp::Ordering::Equal));
    207
    3
            indexed.into_iter().take(k).map(|(i, _)| i).collect()
    208
    3
        }
    209
    }
    210
    211
    // ============================================================================
    212
    // Circuit Breaker (Nygard 2018)
    213
    // ============================================================================
    214
    215
    /// Circuit breaker states
    216
    #[derive(Debug, Clone, Copy, PartialEq, Eq)]
    217
    pub enum CircuitState {
    218
        /// Normal operation - requests flow through
    219
        Closed,
    220
        /// Failure threshold exceeded - requests blocked
    221
        Open,
    222
        /// Testing if service recovered - limited requests allowed
    223
        HalfOpen,
    224
    }
    225
    226
    /// Configuration for circuit breaker
    227
    #[derive(Debug, Clone)]
    228
    pub struct CircuitBreakerConfig {
    229
        /// Number of consecutive failures before opening
    230
        pub failure_threshold: usize,
    231
        /// Number of successes needed to close from half-open
    232
        pub success_threshold: usize,
    233
        /// Time in milliseconds before transitioning from open to half-open
    234
        pub timeout_ms: u64,
    235
    }
    236
    237
    /// Circuit Breaker per Nygard (2018) "Release It!"
    238
    ///
    239
    /// Prevents cascading failures by isolating failing components.
    240
    ///
    241
    /// ## State Machine
    242
    ///
    243
    /// ```text
    244
    /// CLOSED --[failures >= threshold]--> OPEN
    245
    ///    ^                                  |
    246
    ///    |                                  v
    247
    ///    +--[successes >= threshold]-- HALF_OPEN <--[timeout]--+
    248
    /// ```
    249
    ///
    250
    /// ## Citation
    251
    ///
    252
    /// Nygard, M. (2018). "Release It! Design and Deploy Production-Ready Software."
    253
    /// Pragmatic Bookshelf, 2nd Edition.
    254
    pub struct CircuitBreaker {
    255
        config: CircuitBreakerConfig,
    256
        /// Protected mutable state
    257
        state: Mutex<CircuitBreakerState>,
    258
    }
    259
    260
    struct CircuitBreakerState {
    261
        current: CircuitState,
    262
        failure_count: usize,
    263
        success_count: usize,
    264
        last_failure_time: Option<Instant>,
    265
    }
    266
    267
    impl CircuitBreaker {
    268
        /// Create a new circuit breaker
    269
        #[must_use]
    270
    5
        pub fn new(config: CircuitBreakerConfig) -> Self {
    271
    5
            Self {
    272
    5
                config,
    273
    5
                state: Mutex::new(CircuitBreakerState {
    274
    5
                    current: CircuitState::Closed,
    275
    5
                    failure_count: 0,
    276
    5
                    success_count: 0,
    277
    5
                    last_failure_time: None,
    278
    5
                }),
    279
    5
            }
    280
    5
        }
    281
    282
        /// Get current circuit state
    283
        ///
    284
        /// # Panics
    285
        ///
    286
        /// Panics if the internal mutex is poisoned.
    287
        #[must_use]
    288
    5
        pub fn state(&self) -> CircuitState {
    289
    5
            let mut state = self.state.lock().expect("CircuitBreaker mutex poisoned");
    290
    5
            self.maybe_transition_to_half_open(&mut state);
    291
    5
            state.current
    292
    5
        }
    293
    294
        /// Check if request should be allowed
    295
        ///
    296
        /// # Panics
    297
        ///
    298
        /// Panics if the internal mutex is poisoned.
    299
        #[must_use]
    300
    2
        pub fn allow_request(&self) -> bool {
    301
    2
            let mut state = self.state.lock().expect("CircuitBreaker mutex poisoned");
    302
    2
            self.maybe_transition_to_half_open(&mut state);
    303
    304
    2
            match state.current {
    305
    1
                CircuitState::Open => false,
    306
    1
                CircuitState::Closed | CircuitState::HalfOpen => true,
    307
            }
    308
    2
        }
    309
    310
        /// Record a successful request
    311
        ///
    312
        /// # Panics
    313
        ///
    314
        /// Panics if the internal mutex is poisoned.
    315
    2
        pub fn record_success(&self) {
    316
    2
            let mut state = self.state.lock().expect("CircuitBreaker mutex poisoned");
    317
    2
            self.maybe_transition_to_half_open(&mut state);
    318
    319
    2
            match state.current {
    320
    0
                CircuitState::Closed => {
    321
    0
                    state.failure_count = 0; // Reset on success
    322
    0
                },
    323
                CircuitState::HalfOpen => {
    324
    2
                    state.success_count += 1;
    325
    2
                    if state.success_count >= self.config.success_threshold {
    326
    1
                        state.current = CircuitState::Closed;
    327
    1
                        state.failure_count = 0;
    328
    1
                        state.success_count = 0;
    329
    1
                    }
    330
                },
    331
    0
                CircuitState::Open => {}, // Shouldn't happen, but ignore
    332
            }
    333
    2
        }
    334
    335
        /// Record a failed request
    336
        ///
    337
        /// # Panics
    338
        ///
    339
        /// Panics if the internal mutex is poisoned.
    340
    6
        pub fn record_failure(&self) {
    341
    6
            let mut state = self.state.lock().expect("CircuitBreaker mutex poisoned");
    342
    343
    6
            state.failure_count += 1;
    344
    6
            state.last_failure_time = Some(Instant::now());
    345
    346
    6
            if state.failure_count >= self.config.failure_threshold {
    347
    4
                state.current = CircuitState::Open;
    348
    4
                state.success_count = 0;
    349
    4
            }2
    350
    6
        }
    351
    352
    9
        fn maybe_transition_to_half_open(&self, state: &mut CircuitBreakerState) {
    353
    9
            if state.current == CircuitState::Open {
    354
    4
                if let Some(last_failure) = state.last_failure_time {
    355
    4
                    let timeout = Duration::from_millis(self.config.timeout_ms);
    356
    4
                    if last_failure.elapsed() >= timeout {
    357
    2
                        state.current = CircuitState::HalfOpen;
    358
    2
                        state.success_count = 0;
    359
    2
                    }
    360
    0
                }
    361
    5
            }
    362
    9
        }
    363
    }
    364
    365
    // ============================================================================
    366
    // Heijunka Controller (Toyota Production System)
    367
    // ============================================================================
    368
    369
    /// Configuration for Heijunka (load leveling) controller
    370
    #[derive(Debug, Clone)]
    371
    pub struct HeijunkaConfig {
    372
        /// Target latency in milliseconds
    373
        pub target_latency_ms: f64,
    374
        /// Maximum allowed concurrency
    375
        pub max_concurrency: usize,
    376
    }
    377
    378
    /// Load shedding decision
    379
    #[derive(Debug, Clone)]
    380
    pub struct LoadSheddingDecision {
    381
        /// Whether to shed load (reject requests)
    382
        pub shed_load: bool,
    383
        /// Recommended concurrency level
    384
        pub recommended_concurrency: usize,
    385
    }
    386
    387
    /// Heijunka Controller for load leveling via Little's Law
    388
    ///
    389
    /// Little's Law: L = λW
    390
    /// - L = average number of items in system (concurrency)
    391
    /// - λ = arrival rate (requests per second)
    392
    /// - W = average wait time (latency)
    393
    ///
    394
    /// Rearranging: `optimal_concurrency = arrival_rate × (latency_ms / 1000)`
    395
    ///
    396
    /// ## Toyota Production System
    397
    ///
    398
    /// Heijunka (平準化) means "leveling" - smoothing production to avoid overburden.
    399
    /// In ML inference, this means maintaining steady throughput without latency spikes.
    400
    pub struct HeijunkaController {
    401
        config: HeijunkaConfig,
    402
    }
    403
    404
    impl HeijunkaController {
    405
        /// Create a new Heijunka controller
    406
        #[must_use]
    407
    3
        pub fn new(config: HeijunkaConfig) -> Self {
    408
    3
            Self { config }
    409
    3
        }
    410
    411
        /// Calculate optimal concurrency using Little's Law
    412
        ///
    413
        /// # Arguments
    414
        ///
    415
        /// * `arrival_rate` - Requests per second
    416
        /// * `latency_ms` - Average latency in milliseconds
    417
        #[must_use]
    418
        #[allow(clippy::cast_possible_truncation)]
    419
        #[allow(clippy::cast_sign_loss)]
    420
    2
        pub fn optimal_concurrency(&self, arrival_rate: f64, latency_ms: f64) -> usize {
    421
            // Little's Law: L = λW
    422
    2
            let optimal = (arrival_rate * latency_ms / 1000.0).ceil() as usize;
    423
    2
            optimal.clamp(1, self.config.max_concurrency)
    424
    2
        }
    425
    426
        /// Determine if load should be shed based on current state
    427
        ///
    428
        /// # Arguments
    429
        ///
    430
        /// * `current_latency_ms` - Current observed latency
    431
        /// * `current_concurrency` - Current number of concurrent requests
    432
        #[must_use]
    433
        #[allow(clippy::cast_possible_truncation)]
    434
        #[allow(clippy::cast_sign_loss)]
    435
        #[allow(clippy::cast_precision_loss)]
    436
    2
        pub fn should_shed_load(
    437
    2
            &self,
    438
    2
            current_latency_ms: f64,
    439
    2
            current_concurrency: usize,
    440
    2
        ) -> LoadSheddingDecision {
    441
    2
            let should_shed = current_latency_ms > self.config.target_latency_ms
    442
    1
                && current_concurrency >= self.config.max_concurrency;
    443
    444
            // Calculate recommended concurrency to meet target latency
    445
            // If latency is 2x target, we need to halve concurrency
    446
    2
            let ratio = self.config.target_latency_ms / current_latency_ms;
    447
    2
            let concurrency_f64: f64 = current_concurrency as f64;
    448
    2
            let recommended = (concurrency_f64 * ratio).ceil() as usize;
    449
    450
    2
            LoadSheddingDecision {
    451
    2
                shed_load: should_shed,
    452
    2
                recommended_concurrency: recommended.clamp(1, self.config.max_concurrency),
    453
    2
            }
    454
    2
        }
    455
    456
        /// Get the target latency
    457
        #[must_use]
    458
    0
        pub fn target_latency_ms(&self) -> f64 {
    459
    0
            self.config.target_latency_ms
    460
    0
        }
    461
    }
    462
    463
    /// Andon trigger types per Toyota Production System (Jidoka)
    464
    #[derive(Debug, Clone, PartialEq)]
    465
    pub enum AndonTrigger {
    466
        /// Model checksum mismatch - corrupted model
    467
        ModelChecksumMismatch {
    468
            /// ID of the corrupted model
    469
            model_id: String,
    470
        },
    471
        /// Latency P99 exceeded threshold
    472
        LatencyExceeded {
    473
            /// Observed P99 latency in milliseconds
    474
            p99_ms: f64,
    475
            /// Threshold that was exceeded
    476
            threshold_ms: f64,
    477
        },
    478
        /// Error rate above threshold
    479
        ErrorRateThreshold {
    480
            /// Observed error rate (0.0 - 1.0)
    481
            rate: f64,
    482
            /// Threshold that was exceeded
    483
            threshold: f64,
    484
        },
    485
        /// Expert load imbalance detected
    486
        ExpertImbalance {
    487
            /// Ratio of max/min expert utilization
    488
            imbalance_ratio: f64,
    489
        },
    490
    }
    491
    492
    /// Response action for Andon triggers
    493
    #[derive(Debug, Clone, PartialEq, Eq)]
    494
    pub enum AndonResponse {
    495
        /// Automatically rollback to previous known-good state
    496
        Rollback,
    497
        /// Notify operators but continue serving
    498
        Notify,
    499
        /// Quarantine the failing expert (stop routing to it)
    500
        Quarantine,
    501
    }
    502
    503
    impl AndonTrigger {
    504
        /// Determine appropriate response for this trigger
    505
        #[must_use]
    506
    7
        pub fn response(&self) -> AndonResponse {
    507
    7
            match self {
    508
    2
                Self::ModelChecksumMismatch { .. } => AndonResponse::Rollback,
    509
    3
                Self::ErrorRateThreshold { rate, threshold } => {
    510
    3
                    if *rate > threshold * 2.0 {
    511
    2
                        AndonResponse::Quarantine
    512
                    } else {
    513
    1
                        AndonResponse::Notify
    514
                    }
    515
                },
    516
    2
                Self::LatencyExceeded { .. } | Self::ExpertImbalance { .. } => AndonResponse::Notify,
    517
            }
    518
    7
        }
    519
    520
        /// Check if this trigger is critical (requires immediate action)
    521
        #[must_use]
    522
    3
        pub fn is_critical(&self) -> bool {
    523
    1
            matches!(
    524
    3
                self.response(),
    525
                AndonResponse::Rollback | AndonResponse::Quarantine
    526
            )
    527
    3
        }
    528
    }
    529
    530
    #[cfg(test)]
    531
    mod tests {
    532
        use super::*;
    533
    534
        // ========================================================================
    535
        // Power of Two Choices Router Tests (Mitzenmacher 2001)
    536
        // ========================================================================
    537
    538
        #[test]
    539
    1
        fn test_power_of_two_choices_selects_least_loaded() {
    540
    1
            let router = PowerOfTwoChoicesRouter::new(PowerOfTwoConfig {
    541
    1
                num_experts: 4,
    542
    1
                capacity: 100,
    543
    1
            });
    544
    545
            // Load expert 1 heavily
    546
    51
            for _ in 0..50 {
    547
    50
                router.record_start(1);
    548
    50
            }
    549
    550
            // With scores favoring experts 1 and 2, should pick least loaded (2)
    551
    1
            let scores = vec![0.1, 0.9, 0.8, 0.1];
    552
    1
            let choice = router.route(&scores).expect("test");
    553
    554
            // Should pick expert 2 (second best) since expert 1 is heavily loaded
    555
    1
            assert_eq!(choice, 2);
    556
    1
        }
    557
    558
        #[test]
    559
    1
        fn test_power_of_two_choices_equal_load_picks_best_score() {
    560
    1
            let router = PowerOfTwoChoicesRouter::new(PowerOfTwoConfig {
    561
    1
                num_experts: 4,
    562
    1
                capacity: 100,
    563
    1
            });
    564
    565
            // No load on any expert
    566
    1
            let scores = vec![0.1, 0.9, 0.8, 0.1];
    567
    1
            let choice = router.route(&scores).expect("test");
    568
    569
            // Should pick expert 1 (best score) since all equally loaded
    570
    1
            assert_eq!(choice, 1);
    571
    1
        }
    572
    573
        #[test]
    574
    1
        fn test_power_of_two_choices_respects_capacity() {
    575
    1
            let router = PowerOfTwoChoicesRouter::new(PowerOfTwoConfig {
    576
    1
                num_experts: 2,
    577
    1
                capacity: 5,
    578
    1
            });
    579
    580
            // Fill both experts to capacity
    581
    6
            for _ in 0..5 {
    582
    5
                router.record_start(0);
    583
    5
                router.record_start(1);
    584
    5
            }
    585
    586
    1
            let scores = vec![0.9, 0.8];
    587
    1
            let result = router.route(&scores);
    588
    589
            // Should error - both at capacity
    590
    1
            assert!(result.is_err());
    591
    1
        }
    592
    593
        // ========================================================================
    594
        // Circuit Breaker Tests (Nygard 2018)
    595
        // ========================================================================
    596
    597
        #[test]
    598
    1
        fn test_circuit_breaker_starts_closed() {
    599
    1
            let cb = CircuitBreaker::new(CircuitBreakerConfig {
    600
    1
                failure_threshold: 5,
    601
    1
                success_threshold: 3,
    602
    1
                timeout_ms: 1000,
    603
    1
            });
    604
    1
            assert_eq!(cb.state(), CircuitState::Closed);
    605
    1
        }
    606
    607
        #[test]
    608
    1
        fn test_circuit_breaker_opens_on_failures() {
    609
    1
            let cb = CircuitBreaker::new(CircuitBreakerConfig {
    610
    1
                failure_threshold: 3,
    611
    1
                success_threshold: 2,
    612
    1
                timeout_ms: 1000,
    613
    1
            });
    614
    615
    1
            cb.record_failure();
    616
    1
            cb.record_failure();
    617
    1
            assert_eq!(cb.state(), CircuitState::Closed);
    618
    619
    1
            cb.record_failure();
    620
    1
            assert_eq!(cb.state(), CircuitState::Open);
    621
    1
        }
    622
    623
        #[test]
    624
    1
        fn test_circuit_breaker_blocks_when_open() {
    625
    1
            let cb = CircuitBreaker::new(CircuitBreakerConfig {
    626
    1
                failure_threshold: 1,
    627
    1
                success_threshold: 1,
    628
    1
                timeout_ms: 100_000, // Long timeout
    629
    1
            });
    630
    631
    1
            cb.record_failure();
    632
    1
            assert!(!cb.allow_request());
    633
    1
        }
    634
    635
        #[test]
    636
    1
        fn test_circuit_breaker_half_open_after_timeout() {
    637
    1
            let cb = CircuitBreaker::new(CircuitBreakerConfig {
    638
    1
                failure_threshold: 1,
    639
    1
                success_threshold: 1,
    640
    1
                timeout_ms: 1, // Very short timeout
    641
    1
            });
    642
    643
    1
            cb.record_failure();
    644
    1
            std::thread::sleep(std::time::Duration::from_millis(5));
    645
    646
    1
            assert_eq!(cb.state(), CircuitState::HalfOpen);
    647
    1
            assert!(cb.allow_request()); // Should allow probe request
    648
    1
        }
    649
    650
        #[test]
    651
    1
        fn test_circuit_breaker_closes_on_success_in_half_open() {
    652
    1
            let cb = CircuitBreaker::new(CircuitBreakerConfig {
    653
    1
                failure_threshold: 1,
    654
    1
                success_threshold: 2,
    655
    1
                timeout_ms: 1,
    656
    1
            });
    657
    658
    1
            cb.record_failure();
    659
    1
            std::thread::sleep(std::time::Duration::from_millis(5));
    660
    661
            // In half-open state
    662
    1
            cb.record_success();
    663
    1
            cb.record_success();
    664
    665
    1
            assert_eq!(cb.state(), CircuitState::Closed);
    666
    1
        }
    667
    668
        // ========================================================================
    669
        // Heijunka Controller Tests (Toyota Production System)
    670
        // ========================================================================
    671
    672
        #[test]
    673
    1
        fn test_heijunka_calculates_optimal_concurrency() {
    674
    1
            let controller = HeijunkaController::new(HeijunkaConfig {
    675
    1
                target_latency_ms: 100.0,
    676
    1
                max_concurrency: 100,
    677
    1
            });
    678
    679
            // Little's Law: L = λW
    680
            // If arrival_rate = 10 req/s and latency = 100ms = 0.1s
    681
            // Optimal concurrency = 10 * 0.1 = 1
    682
    1
            let concurrency = controller.optimal_concurrency(10.0, 100.0);
    683
    1
            assert_eq!(concurrency, 1);
    684
    1
        }
    685
    686
        #[test]
    687
    1
        fn test_heijunka_caps_at_max_concurrency() {
    688
    1
            let controller = HeijunkaController::new(HeijunkaConfig {
    689
    1
                target_latency_ms: 100.0,
    690
    1
                max_concurrency: 10,
    691
    1
            });
    692
    693
            // High arrival rate would want 100 concurrent, but capped at 10
    694
    1
            let concurrency = controller.optimal_concurrency(1000.0, 100.0);
    695
    1
            assert_eq!(concurrency, 10);
    696
    1
        }
    697
    698
        #[test]
    699
    1
        fn test_heijunka_load_leveling_decision() {
    700
    1
            let controller = HeijunkaController::new(HeijunkaConfig {
    701
    1
                target_latency_ms: 100.0,
    702
    1
                max_concurrency: 50,
    703
    1
            });
    704
    705
            // Current load exceeds target
    706
    1
            let decision = controller.should_shed_load(150.0, 50);
    707
    1
            assert!(decision.shed_load);
    708
    709
            // Current load under target
    710
    1
            let decision = controller.should_shed_load(50.0, 10);
    711
    1
            assert!(!decision.shed_load);
    712
    1
        }
    713
    714
        // ========================================================================
    715
        // Original Capacity Factor Router Tests
    716
        // ========================================================================
    717
    718
        #[test]
    719
    1
        fn test_route_to_best_expert() {
    720
    1
            let router = CapacityFactorRouter::new(CapacityConfig {
    721
    1
                capacity: 10,
    722
    1
                num_experts: 4,
    723
    1
            });
    724
    1
            let scores = vec![0.1, 0.5, 0.3, 0.1];
    725
    1
            assert_eq!(router.route(&scores).expect("test"), 1);
    726
    1
        }
    727
    728
        #[test]
    729
    1
        fn test_fallback_when_primary_full() {
    730
    1
            let router = CapacityFactorRouter::new(CapacityConfig {
    731
    1
                capacity: 1,
    732
    1
                num_experts: 4,
    733
    1
            });
    734
    1
            router.record_start(1); // Fill expert 1
    735
    1
            let scores = vec![0.1, 0.5, 0.3, 0.1];
    736
    1
            assert_eq!(router.route(&scores).expect("test"), 2); // Falls back to #2
    737
    1
        }
    738
    739
        #[test]
    740
    1
        fn test_queue_depth_tracking() {
    741
    1
            let router = CapacityFactorRouter::new(CapacityConfig {
    742
    1
                capacity: 10,
    743
    1
                num_experts: 2,
    744
    1
            });
    745
    1
            assert_eq!(router.queue_depth(0), 0);
    746
    1
            router.record_start(0);
    747
    1
            assert_eq!(router.queue_depth(0), 1);
    748
    1
            router.record_end(0);
    749
    1
            assert_eq!(router.queue_depth(0), 0);
    750
    1
        }
    751
    752
        #[test]
    753
    1
        fn test_wrong_score_count_error() {
    754
    1
            let router = CapacityFactorRouter::new(CapacityConfig {
    755
    1
                capacity: 10,
    756
    1
                num_experts: 4,
    757
    1
            });
    758
    1
            let scores = vec![0.5, 0.5]; // Wrong count
    759
    1
            assert!(router.route(&scores).is_err());
    760
    1
        }
    761
    762
        #[test]
    763
    1
        fn test_andon_checksum_triggers_rollback() {
    764
    1
            let trigger = AndonTrigger::ModelChecksumMismatch {
    765
    1
                model_id: "model-1".to_string(),
    766
    1
            };
    767
    1
            assert_eq!(trigger.response(), AndonResponse::Rollback);
    768
    1
            assert!(trigger.is_critical());
    769
    1
        }
    770
    771
        #[test]
    772
    1
        fn test_andon_latency_triggers_notify() {
    773
    1
            let trigger = AndonTrigger::LatencyExceeded {
    774
    1
                p99_ms: 150.0,
    775
    1
                threshold_ms: 100.0,
    776
    1
            };
    777
    1
            assert_eq!(trigger.response(), AndonResponse::Notify);
    778
    1
            assert!(!trigger.is_critical());
    779
    1
        }
    780
    781
        #[test]
    782
    1
        fn test_andon_high_error_rate_quarantines() {
    783
    1
            let trigger = AndonTrigger::ErrorRateThreshold {
    784
    1
                rate: 0.25,
    785
    1
                threshold: 0.1,
    786
    1
            };
    787
    1
            assert_eq!(trigger.response(), AndonResponse::Quarantine);
    788
    1
            assert!(trigger.is_critical());
    789
    1
        }
    790
    791
        #[test]
    792
    1
        fn test_andon_moderate_error_rate_notifies() {
    793
    1
            let trigger = AndonTrigger::ErrorRateThreshold {
    794
    1
                rate: 0.15,
    795
    1
                threshold: 0.1,
    796
    1
            };
    797
    1
            assert_eq!(trigger.response(), AndonResponse::Notify);
    798
    1
        }
    799
    }