rchd 2.1.4

//! Worker health monitoring with heartbeats.
//!
//! Periodically checks worker availability and updates their status.

#![allow(dead_code)] // Scaffold code - methods will be used in future beads

use crate::alerts::AlertManager;
use crate::metrics;
use crate::ui::workers::WorkerStatusPanel;
use crate::workers::{WorkerPool, WorkerState};
use rch_common::mock::{self, MockConfig, MockSshClient};
use rch_common::{
    CircuitBreakerConfig, CircuitState, CircuitStats, SshClient, SshOptions, WorkerStatus,
    is_retryable_transport_error,
};
use std::sync::Arc;
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use tokio::sync::{Mutex, Notify, RwLock};
use tokio::time::interval;
use tracing::{debug, info, warn};

fn is_mock_transport(_worker: &WorkerState) -> bool {
    // In mock mode, we assume mock config is active
    // We can't easily check worker.config without async lock here,
    // but if mock is enabled globally that's enough
    mock::is_mock_enabled()
}

/// Default health check interval.
const DEFAULT_CHECK_INTERVAL: Duration = Duration::from_secs(30);

/// Default timeout for health check SSH connection.
const DEFAULT_CHECK_TIMEOUT: Duration = Duration::from_secs(10);

/// Timeout for the background capability probe.
///
/// Separate from [`DEFAULT_CHECK_TIMEOUT`] on purpose: liveness detection must
/// stay tight, while capability work runs separately under a per-worker guard.
/// It runs `rch-wkr capabilities`, which execs rustc/node/npm/go/zig/
/// cargo-zigbuild and stats the disk behind a fresh SSH handshake — measured at
/// 6-16s on loaded 10-core build hosts, versus 2.9s idle. Reusing the 10s
/// liveness budget dropped capability data on precisely the busiest workers,
/// leaving `disk_free_gb` stale so the low-disk admission gate fails open.
const CAPABILITY_PROBE_TIMEOUT: Duration = Duration::from_secs(25);

/// Threshold for degraded status (slow response).
const DEGRADED_THRESHOLD_MS: u64 = 5000;

fn duration_millis_u64(duration: Duration) -> u64 {
    u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)
}

/// Derive the health-facing [`WorkerStatus`] from a circuit state and the latest
/// check outcome. This is the single mapping shared by the diagnostic
/// [`WorkerHealth`] panel and the authoritative health-monitor loop, so the two
/// never disagree about what a given circuit state "means":
/// - Open      -> Unreachable (short-circuited out of scheduling)
/// - HalfOpen  -> Degraded (probing recovery)
/// - Closed    -> Degraded if the last check failed or was slow, else Healthy
fn derive_worker_status(
    circuit_state: CircuitState,
    healthy: bool,
    response_time_ms: u64,
    config: &HealthConfig,
) -> WorkerStatus {
    match circuit_state {
        CircuitState::Open => WorkerStatus::Unreachable,
        CircuitState::HalfOpen => WorkerStatus::Degraded,
        CircuitState::Closed => {
            if !healthy {
                // Failed but circuit not open yet -> Degraded
                WorkerStatus::Degraded
            } else if response_time_ms > config.degraded_threshold_ms {
                // Slow response -> Degraded
                WorkerStatus::Degraded
            } else {
                // Healthy and fast
                WorkerStatus::Healthy
            }
        }
    }
}

/// Health monitor configuration.
#[derive(Debug, Clone)]
pub struct HealthConfig {
    /// Interval between health checks.
    pub check_interval: Duration,
    /// Timeout for each health check.
    pub check_timeout: Duration,
    /// Threshold for marking worker as degraded (ms).
    pub degraded_threshold_ms: u64,
    /// Number of consecutive failures before marking unreachable.
    pub failure_threshold: u32,
    /// Circuit breaker configuration.
    pub circuit: CircuitBreakerConfig,
}

impl Default for HealthConfig {
    fn default() -> Self {
        Self {
            check_interval: DEFAULT_CHECK_INTERVAL,
            check_timeout: DEFAULT_CHECK_TIMEOUT,
            degraded_threshold_ms: DEGRADED_THRESHOLD_MS,
            failure_threshold: 3,
            circuit: CircuitBreakerConfig::default(),
        }
    }
}

/// Result of a single health check.
#[derive(Debug, Clone)]
pub struct HealthCheckResult {
    /// Whether the check succeeded.
    pub healthy: bool,
    /// Response time in milliseconds.
    pub response_time_ms: u64,
    /// Error message if failed.
    pub error: Option<String>,
    /// Whether an (unhealthy) result was caused by a *transient* transport
    /// error (retryable connect/timeout blip) that survived the in-check
    /// retries. Always `false` for healthy results. Fed to the circuit engine so
    /// a transient failure during half-open recovery does not reopen the circuit
    /// or restart its cooldown. See [`CircuitStats::apply_health_outcome`].
    pub transient: bool,
    /// Whether an (unhealthy) result is *command-class* evidence: the SSH
    /// session reached the worker but the probe command never completed
    /// (timed out, or the child could not be waited on). This is the
    /// fork-exhaustion / swap-thrash signature — the host accepts connections
    /// but cannot run anything — and is fed to the circuit through
    /// [`CircuitStats::apply_command_outcome`], which blocks a half-open close
    /// and trips the breaker on its own threshold. Always `false` for healthy
    /// results and for connect-stage failures.
    pub command_failure: bool,
    /// Timestamp of the check.
    #[allow(dead_code)] // May be used for monitoring metrics
    pub checked_at: Instant,
}

impl HealthCheckResult {
    fn success(response_time_ms: u64) -> Self {
        Self {
            healthy: true,
            response_time_ms,
            error: None,
            transient: false,
            command_failure: false,
            checked_at: Instant::now(),
        }
    }

    fn failure(error: String) -> Self {
        Self {
            healthy: false,
            response_time_ms: 0,
            error: Some(error),
            transient: false,
            command_failure: false,
            checked_at: Instant::now(),
        }
    }

    /// Construct a failure result flagged as caused by a transient transport
    /// error (retried but still failing).
    fn transient_failure(error: String) -> Self {
        Self {
            healthy: false,
            response_time_ms: 0,
            error: Some(error),
            transient: true,
            command_failure: false,
            checked_at: Instant::now(),
        }
    }

    /// Construct a failure result for a probe whose session connected but
    /// whose command did not complete (see [`Self::command_failure`]).
    /// `transient` is preserved for diagnostics; the circuit engine ignores it
    /// for command-class evidence.
    fn command_failure(error: String, transient: bool) -> Self {
        Self {
            healthy: false,
            response_time_ms: 0,
            error: Some(error),
            transient,
            command_failure: true,
            checked_at: Instant::now(),
        }
    }

    /// Record metrics for this health check result.
    fn record_metrics(&self, worker_id: &str) {
        if self.healthy {
            // Record latency histogram
            metrics::observe_worker_latency(worker_id, self.response_time_ms as f64);
            // Update last seen timestamp
            let now = SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .map(|d| d.as_secs_f64())
                .unwrap_or(0.0);
            metrics::set_worker_last_seen(worker_id, now);
        }
    }
}

/// Worker health state tracking with circuit breaker integration.
#[derive(Debug)]
pub struct WorkerHealth {
    /// Last health check result.
    last_result: Option<HealthCheckResult>,
    /// Current worker status.
    current_status: WorkerStatus,
    /// Circuit breaker statistics for this worker.
    circuit: CircuitStats,
    /// Last error message (for diagnostics).
    last_error: Option<String>,
}

impl Default for WorkerHealth {
    fn default() -> Self {
        Self {
            last_result: None,
            current_status: WorkerStatus::Healthy,
            circuit: CircuitStats::new(),
            last_error: None,
        }
    }
}

impl WorkerHealth {
    /// Create a new WorkerHealth with default state.
    pub fn new() -> Self {
        Self::default()
    }

    /// Update health state based on check result.
    ///
    /// This drives circuit breaker state transitions based on health check
    /// outcomes by delegating to the single shared engine
    /// [`CircuitStats::apply_health_outcome`], so this diagnostic circuit stays
    /// in lockstep with the authoritative `WorkerState.circuit` that the
    /// scheduler reads. See [`Self::observe`], which is the same logic split out
    /// so the health-monitor loop can drive both circuits from one outcome.
    pub fn update(&mut self, result: HealthCheckResult, config: &HealthConfig, worker_id: &str) {
        let healthy = result.healthy;
        let transient = result.transient;
        let response_time_ms = result.response_time_ms;
        self.last_error = if healthy { None } else { result.error.clone() };
        self.last_result = Some(result);
        self.observe(healthy, transient, response_time_ms, config, worker_id);
    }

    /// Apply a single health outcome to the diagnostic circuit and recompute the
    /// worker status. Shared with [`Self::update`]; used directly by the
    /// health-monitor loop to mirror the authoritative outcome (already applied
    /// to `WorkerState.circuit`) into this panel-facing copy.
    pub fn observe(
        &mut self,
        healthy: bool,
        transient: bool,
        response_time_ms: u64,
        config: &HealthConfig,
        worker_id: &str,
    ) {
        let prior_circuit_state = self.circuit.state();

        let new_circuit_state =
            self.circuit
                .apply_health_outcome(healthy, transient, &config.circuit);

        self.current_status =
            derive_worker_status(new_circuit_state, healthy, response_time_ms, config);

        if prior_circuit_state != new_circuit_state {
            info!(
                "Worker {} circuit state: {:?} -> {:?}",
                worker_id, prior_circuit_state, new_circuit_state
            );
        }
    }

    /// Replace this diagnostic circuit with a snapshot of the authoritative
    /// `WorkerState.circuit` and recompute the worker status from it. Used by
    /// the health-monitor loop instead of [`Self::observe`] because the
    /// authoritative circuit also receives command-class evidence from the
    /// telemetry poller; copying keeps the two in lockstep by construction.
    pub fn mirror(
        &mut self,
        authoritative: CircuitStats,
        healthy: bool,
        response_time_ms: u64,
        config: &HealthConfig,
        worker_id: &str,
    ) {
        let prior_circuit_state = self.circuit.state();
        let new_circuit_state = authoritative.state();
        self.circuit = authoritative;
        self.current_status =
            derive_worker_status(new_circuit_state, healthy, response_time_ms, config);

        if prior_circuit_state != new_circuit_state {
            info!(
                "Worker {} circuit state: {:?} -> {:?}",
                worker_id, prior_circuit_state, new_circuit_state
            );
        }
    }

    /// Record the last check result on this diagnostic copy (and refresh the
    /// derived `last_error`). Used by the health-monitor loop, which drives the
    /// circuit via [`Self::observe`] (not [`Self::update`]) and separately stores
    /// the full result so diagnostics/panels keep the error text and latency.
    pub fn set_last_result(&mut self, result: HealthCheckResult) {
        self.last_error = if result.healthy {
            None
        } else {
            result.error.clone()
        };
        self.last_result = Some(result);
    }

    /// Get current worker status.
    pub fn status(&self) -> WorkerStatus {
        self.current_status
    }

    /// Get current circuit state.
    pub fn circuit_state(&self) -> CircuitState {
        self.circuit.state()
    }

    /// Get circuit statistics.
    pub fn circuit_stats(&self) -> &CircuitStats {
        &self.circuit
    }

    /// Get last check result.
    #[allow(dead_code)] // Will be used by status API
    pub fn last_result(&self) -> Option<&HealthCheckResult> {
        self.last_result.as_ref()
    }

    /// Get last error message.
    pub fn last_error(&self) -> Option<&str> {
        self.last_error.as_deref()
    }

    /// Check if this worker can be used for a probe in half-open state.
    pub fn can_probe(&self, config: &HealthConfig) -> bool {
        self.circuit.can_probe(&config.circuit)
    }

    /// Start a probe request (call when sending a request to half-open circuit).
    pub fn start_probe(&mut self, config: &HealthConfig) -> bool {
        self.circuit.start_probe(&config.circuit)
    }
}

/// Health monitor that periodically checks all workers.
pub struct HealthMonitor {
    /// Worker pool to monitor.
    pool: WorkerPool,
    /// Configuration.
    config: HealthConfig,
    /// Health state per worker.
    health_states: Arc<RwLock<std::collections::HashMap<String, WorkerHealth>>>,
    /// Whether monitor is running.
    running: Arc<RwLock<bool>>,
    /// Wake signal for prompt shutdown while the monitor is sleeping between checks.
    shutdown: Arc<Notify>,
    /// Optional worker status panel for log output.
    status_panel: Option<Arc<Mutex<WorkerStatusPanel>>>,
    /// Optional alert manager for worker health alerting.
    alert_manager: Option<Arc<AlertManager>>,
    /// Optional shared SSH connection pool. When `Some`, health/capability
    /// probes run over a warm reused ControlMaster; when `None`, per-call
    /// throwaway SSH sessions are used.
    ssh_pool: Option<Arc<rch_common::SshPool>>,
}

impl HealthMonitor {
    /// Create a new health monitor.
    pub fn new(pool: WorkerPool, config: HealthConfig) -> Self {
        Self {
            pool,
            config,
            health_states: Arc::new(RwLock::new(std::collections::HashMap::new())),
            running: Arc::new(RwLock::new(false)),
            shutdown: Arc::new(Notify::new()),
            status_panel: None,
            alert_manager: None,
            ssh_pool: None,
        }
    }

    /// Attach a shared SSH connection pool for warm ControlMaster reuse.
    #[must_use]
    pub fn with_ssh_pool(mut self, pool: Option<Arc<rch_common::SshPool>>) -> Self {
        self.ssh_pool = pool;
        self
    }

    /// Attach a worker status panel for periodic output.
    #[must_use]
    pub fn with_status_panel(mut self, panel: Arc<Mutex<WorkerStatusPanel>>) -> Self {
        self.status_panel = Some(panel);
        self
    }

    /// Attach an alert manager for worker health alerting.
    #[must_use]
    pub fn with_alert_manager(mut self, alert_manager: Arc<AlertManager>) -> Self {
        self.alert_manager = Some(alert_manager);
        self
    }

    /// Start the health monitoring background task.
    pub fn start(&self) -> tokio::task::JoinHandle<()> {
        let pool = self.pool.clone();
        let config = self.config.clone();
        let health_states = self.health_states.clone();
        let running = self.running.clone();
        let shutdown = self.shutdown.clone();
        let status_panel = self.status_panel.clone();
        let alert_manager = self.alert_manager.clone();
        let ssh_pool = self.ssh_pool.clone();

        tokio::spawn(async move {
            *running.write().await = true;
            let mut ticker = interval(config.check_interval);

            info!(
                "Health monitor started (interval: {:?})",
                config.check_interval
            );

            loop {
                tokio::select! {
                    _ = ticker.tick() => {}
                    () = shutdown.notified() => {
                        info!("Health monitor stopping");
                        break;
                    }
                }

                if !*running.read().await {
                    info!("Health monitor stopping");
                    break;
                }

                // Check ALL workers (not just healthy) so unreachable workers can recover
                let workers = pool.all_workers().await;
                debug!("Checking health of {} workers", workers.len());

                // Track unreachable count for all-workers-offline alert
                let total_workers = workers.len();
                let mut unreachable_count = 0;

                for worker in workers {
                    let worker_config_guard = worker.config.read().await;
                    let worker_id = worker_config_guard.id.as_str().to_string();
                    // Drop lock before check_worker_health to avoid holding it during IO
                    drop(worker_config_guard);

                    let previous_effective_status = worker.status().await;
                    let result = check_worker_health(&worker, &config, ssh_pool.as_ref()).await;

                    // Record health check latency metric
                    if result.healthy {
                        metrics::observe_worker_latency(&worker_id, result.response_time_ms as f64);
                        // Update last seen timestamp
                        let now = SystemTime::now()
                            .duration_since(UNIX_EPOCH)
                            .map(|d| d.as_secs_f64())
                            .unwrap_or(0.0);
                        metrics::set_worker_last_seen(&worker_id, now);
                    }

                    // Drive the AUTHORITATIVE circuit on WorkerState — the one the
                    // scheduler reads — through the shared engine. This is what
                    // makes a failing worker actually short-circuit out of
                    // selection and a recovered worker rejoin. The old code drove
                    // only the ephemeral WorkerHealth.circuit below, which
                    // selection never consulted.
                    //
                    // A probe that connected but whose command never completed
                    // is command-class evidence (issue #48): it must block a
                    // half-open close and trip the breaker on its own threshold
                    // even when interleaved liveness probes succeed.
                    let (previous_circuit_state, new_circuit_state) = if result.command_failure {
                        worker.record_command_outcome(false, &config.circuit).await
                    } else {
                        worker
                            .record_health_check(result.healthy, result.transient, &config.circuit)
                            .await
                    };

                    if result.healthy {
                        worker.set_last_latency_ms(Some(result.response_time_ms));
                    } else {
                        worker.set_last_latency_ms(None);
                    }

                    // Derive the health-facing status from the authoritative
                    // circuit outcome via the shared mapping, then apply it to the
                    // worker lifecycle (respecting admin intent / quarantine).
                    let new_status = derive_worker_status(
                        new_circuit_state,
                        result.healthy,
                        result.response_time_ms,
                        &config,
                    );
                    let effective_status = worker.apply_health_status(new_status).await;

                    // Mirror the same outcome into the diagnostic WorkerHealth so
                    // get_health()/all_health_states()/the status panel keep
                    // reporting a circuit consistent with the authoritative one.
                    //
                    // Mirror by COPYING the authoritative stats rather than by
                    // replaying the outcome: the authoritative circuit is also
                    // advanced by command-class evidence from the telemetry
                    // poller, which this loop never sees, so a replay would
                    // drift.
                    let mut states = health_states.write().await;
                    let health = states.entry(worker_id.clone()).or_default();
                    health.mirror(
                        worker.circuit_stats().await,
                        result.healthy,
                        result.response_time_ms,
                        &config,
                        &worker_id,
                    );
                    health.set_last_result(result.clone());

                    // Track unreachable workers for all-workers-offline alert
                    if effective_status == WorkerStatus::Unreachable {
                        unreachable_count += 1;
                    }

                    // Notify alert manager of status changes and circuit state changes
                    if let Some(ref alert_mgr) = alert_manager {
                        // Status change notification
                        if previous_effective_status != effective_status {
                            alert_mgr.handle_worker_status_change(
                                &worker_id,
                                previous_effective_status,
                                effective_status,
                                health.last_error(),
                            );
                        }

                        // Circuit breaker opened notification
                        if previous_circuit_state != CircuitState::Open
                            && new_circuit_state == CircuitState::Open
                        {
                            alert_mgr.handle_circuit_open(&worker_id);
                        }

                        // Circuit breaker closing notification (bd-3ogaz):
                        // mark the open-alert as cleared so UIs can grey it
                        // out and auto-evict it after the retention window.
                        // Without this, a transient circuit-open warning
                        // persists on `rch status` long after recovery.
                        if previous_circuit_state == CircuitState::Open
                            && new_circuit_state != CircuitState::Open
                        {
                            alert_mgr.handle_circuit_closed(&worker_id);
                        }
                    }

                    // Record worker status metric
                    let status_value = match effective_status {
                        WorkerStatus::Healthy => 1.0,
                        WorkerStatus::Degraded => 2.0,
                        WorkerStatus::Draining => 2.0,
                        WorkerStatus::Drained => 0.0,
                        WorkerStatus::Unreachable => 0.0,
                        WorkerStatus::Disabled => 0.0,
                    };
                    metrics::set_worker_status(&worker_id, "current", status_value);

                    // Record circuit breaker state
                    let circuit_value = match new_circuit_state {
                        CircuitState::Closed => 0,
                        CircuitState::HalfOpen => 1,
                        CircuitState::Open => 2,
                    };
                    metrics::set_circuit_state(&worker_id, circuit_value);

                    // Record slot metrics
                    metrics::set_worker_slots_total(
                        &worker_id,
                        worker.effective_total_slots().await,
                    );
                    metrics::set_worker_slots_available(&worker_id, worker.available_slots().await);
                    if result.healthy {
                        debug!(
                            "Worker {} healthy ({}ms)",
                            worker_id, result.response_time_ms
                        );

                        // Capability work runs separately from liveness checks.
                        // probe_worker_capabilities skips overlapping requests,
                        // including refreshes from operators or selection.
                        let worker_clone = worker.clone();
                        // Deliberately NOT `config.check_timeout`. That budget governs
                        // liveness detection and must stay tight, but this probe already
                        // runs separately (see the spawn below), so a slow probe does
                        // not delay liveness. `rch-wkr capabilities` shells out to toolchain
                        // binaries behind a fresh SSH handshake; on loaded build hosts it
                        // was measured at 6-16s, so a 10s liveness budget silently dropped
                        // capability data on exactly the busiest workers. Stale
                        // `disk_free_gb` makes `is_low_disk()` return `None`, and the
                        // low-disk admission gate then FAILS OPEN — the dispatcher keeps
                        // scheduling onto a worker that is about to run out of disk.
                        let timeout = CAPABILITY_PROBE_TIMEOUT;
                        let probe_pool = ssh_pool.clone();
                        tokio::spawn(async move {
                            if let Some(capabilities) = probe_worker_capabilities(
                                &worker_clone,
                                timeout,
                                probe_pool.as_ref(),
                            )
                            .await
                            {
                                worker_clone.set_capabilities(capabilities).await;
                            }
                        });
                    } else {
                        warn!(
                            "Worker {} check failed: {:?} (failures: {})",
                            worker_id,
                            result.error,
                            health.circuit_stats().consecutive_failures()
                        );
                    }
                }

                // Check for all-workers-offline condition
                if let Some(ref alert_mgr) = alert_manager {
                    alert_mgr.handle_all_workers_offline(total_workers, unreachable_count);
                }

                if let Some(panel) = &status_panel {
                    let snapshot = WorkerStatusPanel::collect_snapshot(&pool).await;
                    let mut panel = panel.lock().await;
                    panel.emit_update(&snapshot, 0);
                }
            }
        })
    }

    /// Stop the health monitor.
    #[allow(dead_code)] // Will be used for graceful shutdown
    pub async fn stop(&self) {
        *self.running.write().await = false;
        self.shutdown.notify_waiters();
    }

    /// Get health state for a worker.
    #[allow(dead_code)] // Will be used by status API
    pub async fn get_health(&self, worker_id: &str) -> Option<WorkerStatus> {
        let states = self.health_states.read().await;
        states.get(worker_id).map(|h| h.status())
    }

    /// Get all health states.
    #[allow(dead_code)] // Will be used by status API
    pub async fn all_health_states(&self) -> Vec<(String, WorkerStatus)> {
        let states = self.health_states.read().await;
        states
            .iter()
            .map(|(id, h)| (id.clone(), h.status()))
            .collect()
    }
}

/// Check health of a single worker.
async fn check_worker_health(
    worker: &Arc<WorkerState>,
    config: &HealthConfig,
    ssh_pool: Option<&Arc<rch_common::SshPool>>,
) -> HealthCheckResult {
    let start = Instant::now();

    // Debug: log mock mode status and env var
    let mock_env = std::env::var("RCH_MOCK_SSH").unwrap_or_default();
    let mock_enabled = is_mock_transport(worker);
    let worker_config = worker.config.read().await;

    debug!(
        "Health check for {}: mock_enabled={}, RCH_MOCK_SSH='{}', host='{}'",
        worker_config.id, mock_enabled, mock_env, worker_config.host
    );

    if mock_enabled {
        let mut client = MockSshClient::new(worker_config.clone(), MockConfig::from_env());
        match client.connect().await {
            Ok(()) => match client.execute("echo health_check").await {
                Ok(result) => {
                    let duration = start.elapsed();
                    let _ = client.disconnect().await;
                    if result.success() && result.stdout.trim().eq("health_check") {
                        return HealthCheckResult::success(duration_millis_u64(duration));
                    }
                    return HealthCheckResult::failure(format!(
                        "Unexpected response: exit={}, stdout={}",
                        result.exit_code,
                        result.stdout.trim()
                    ));
                }
                Err(e) => {
                    let _ = client.disconnect().await;
                    return HealthCheckResult::failure(format!("Command failed: {}", e));
                }
            },
            Err(e) => return HealthCheckResult::failure(format!("Connection failed: {}", e)),
        }
    }

    // Create SSH connection with timeout
    let ssh_options = SshOptions {
        connect_timeout: config.check_timeout,
        command_timeout: config.check_timeout,
        control_master: false, // Don't use control master for health checks
        ..Default::default()
    };

    let worker_config = worker_config.clone();

    // Retry a *retryable* transport error (connect/timeout blip) a couple of
    // times with a short backoff before recording a circuit failure. A single
    // packet-loss window or a momentarily-busy sshd otherwise trips the breaker
    // and drops an otherwise-healthy worker out of scheduling until the next
    // clean poll. Non-retryable failures (auth, host-key, resolution) fail fast
    // — retrying can't help and just delays the correct verdict.
    const MAX_ATTEMPTS: u32 = 3; // 1 initial + 2 retries
    const RETRY_BACKOFF: Duration = Duration::from_millis(250);

    let mut last_error: Option<HealthProbeError> = None;
    for attempt in 1..=MAX_ATTEMPTS {
        let attempt_start = Instant::now();
        match probe_health_once(&worker_config, ssh_options.clone(), ssh_pool).await {
            Ok(()) => {
                let duration = attempt_start.elapsed();
                return HealthCheckResult::success(duration_millis_u64(duration));
            }
            Err(err) => {
                let retryable = err.retryable;
                if retryable && attempt < MAX_ATTEMPTS {
                    debug!(
                        "Worker {} health probe attempt {}/{} hit retryable transport error ({}); retrying",
                        worker_config.id, attempt, MAX_ATTEMPTS, err.message
                    );
                    last_error = Some(err);
                    tokio::time::sleep(RETRY_BACKOFF).await;
                    continue;
                }
                last_error = Some(err);
                break;
            }
        }
    }

    // All attempts exhausted (or a fast-fail non-retryable error). Flag the
    // result as `transient` when the underlying failure was a retryable
    // transport error so the circuit engine treats a persistent-but-transient
    // failure during half-open recovery as a blip (does not reopen). A drop of
    // `start` is intentional here — the retried duration is not a meaningful
    // latency signal for a failed check.
    let _ = start;
    match last_error {
        // "Connected, but the command never completed" is command-class
        // evidence regardless of whether the transport error text looks
        // retryable: three consecutive command-stage timeouts over an
        // established session is the signature of a host that cannot fork,
        // not of a network blip. See `HealthCheckResult::command_failure`.
        Some(err) if err.stage == ProbeStage::Command => {
            HealthCheckResult::command_failure(err.message, err.retryable)
        }
        Some(err) if err.retryable => HealthCheckResult::transient_failure(err.message),
        Some(err) => HealthCheckResult::failure(err.message),
        None => HealthCheckResult::failure("Health check failed (no result)".to_string()),
    }
}

/// Which stage of a health probe failed.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ProbeStage {
    /// The SSH session could not be established (connect/auth/resolution).
    Connect,
    /// The session was established but the probe command did not complete
    /// successfully (timed out, wait failed, or ran with the wrong output).
    Command,
}

/// Classified failure from a single health-probe attempt.
struct HealthProbeError {
    message: String,
    retryable: bool,
    stage: ProbeStage,
}

/// Classify a pooled-transport error by stage. The pool multiplexes over a
/// warm ControlMaster, so a connection-level fault surfaces as a retryable
/// transport error that is NOT a command timeout (stale control socket,
/// connection refused/reset, resolution); anything else — including a
/// "timed out" on the established master — means the session was up and the
/// command did not finish.
fn pooled_error_stage(message: &str, retryable: bool) -> ProbeStage {
    let lower = message.to_lowercase();
    if lower.contains("timed out") || lower.contains("timeout") || lower.contains("command") {
        // A timeout on the established master, or a spawned command that
        // could not be waited on, is a command that did not complete.
        ProbeStage::Command
    } else if retryable {
        // Stale control socket, reset/refused connection, resolution blip.
        ProbeStage::Connect
    } else {
        // Non-retryable and not command-shaped: auth / host-key / key-file
        // problems, all of which prevent a session from being established.
        ProbeStage::Connect
    }
}

/// Perform ONE health-probe attempt (connect + `echo health_check`) against a
/// worker over a throwaway SSH session. Returns `Ok(())` on the expected
/// sentinel, or a classified [`HealthProbeError`] otherwise. The retry policy
/// lives in the caller ([`check_worker_health`]).
async fn probe_health_once(
    worker_config: &rch_common::WorkerConfig,
    ssh_options: SshOptions,
    ssh_pool: Option<&Arc<rch_common::SshPool>>,
) -> Result<(), HealthProbeError> {
    // Pooled path: run over the warm shared ControlMaster. run_with_timeout
    // collapses connect+execute into one call; classify any error as retryable
    // via the shared transport-error heuristic so the caller's retry/transient
    // logic still applies.
    if let Some(pool) = ssh_pool {
        let command_timeout = ssh_options.command_timeout;
        return match pool
            .run_with_timeout(worker_config, "echo health_check", command_timeout)
            .await
        {
            Ok(result) => classify_health_output(&result),
            Err(e) => {
                let retryable = is_retryable_transport_error(&e);
                let message = format!("Health probe failed: {}", e);
                let stage = pooled_error_stage(&message, retryable);
                Err(HealthProbeError {
                    message,
                    retryable,
                    stage,
                })
            }
        };
    }

    let mut client = SshClient::new(worker_config.clone(), ssh_options);

    match client.connect().await {
        Ok(()) => match client.execute("echo health_check").await {
            Ok(result) => {
                let _ = client.disconnect().await;
                classify_health_output(&result)
            }
            Err(e) => {
                let _ = client.disconnect().await;
                let retryable = is_retryable_transport_error(&e);
                Err(HealthProbeError {
                    message: format!("Command failed: {}", e),
                    retryable,
                    stage: ProbeStage::Command,
                })
            }
        },
        Err(e) => {
            let retryable = is_retryable_transport_error(&e);
            Err(HealthProbeError {
                message: format!("Connection failed: {}", e),
                retryable,
                stage: ProbeStage::Connect,
            })
        }
    }
}

/// Map a health-probe command result onto the expected `echo health_check`
/// sentinel. A command that ran but produced the wrong output is not a transport
/// blip (not retryable).
fn classify_health_output(result: &rch_common::CommandResult) -> Result<(), HealthProbeError> {
    if result.success() && result.stdout.trim().eq("health_check") {
        Ok(())
    } else {
        Err(HealthProbeError {
            message: format!(
                "Unexpected response: exit={}, stdout={}",
                result.exit_code,
                result.stdout.trim()
            ),
            retryable: false,
            stage: ProbeStage::Command,
        })
    }
}

/// Perform a one-time health check on a worker.
#[allow(dead_code)] // Will be used by workers probe command
pub async fn probe_worker(worker: &WorkerState) -> HealthCheckResult {
    let config = HealthConfig::default();
    // Wrap in Arc for compatibility
    let config_clone = worker.config.read().await.clone();
    let worker_arc = Arc::new(WorkerState::new(config_clone));
    check_worker_health(&worker_arc, &config, None).await
}

/// The ` --tool-probe '<json>'` suffix for a worker's declared tool probes, or
/// an empty string when it declares none.
///
/// The JSON is single-quoted for the remote shell with embedded quotes escaped
/// the POSIX way. Names and argv reaching here were validated at config load,
/// but the quoting is unconditional: this string is interpolated into a remote
/// command, and "the validator upstream would have caught it" is not a property
/// this function should depend on.
fn capability_tool_argument(tools: &[rch_common::types::WorkerToolProbe]) -> String {
    if tools.is_empty() {
        return String::new();
    }
    let Ok(json) = serde_json::to_string(tools) else {
        return String::new();
    };
    format!(" --tool-probe '{}'", json.replace('\'', "'\\''"))
}

/// Probe worker capabilities (Bun, Node, Rust versions).
///
/// Runs `rch-wkr capabilities` on the worker and parses the JSON output.
/// Returns None if probing fails or another caller is already probing this worker.
/// Callers retain their existing snapshot; overlapping requests do not spawn SSH.
pub async fn probe_worker_capabilities(
    worker: &Arc<WorkerState>,
    timeout: Duration,
    ssh_pool: Option<&Arc<rch_common::SshPool>>,
) -> Option<rch_common::WorkerCapabilities> {
    use rch_common::{SshClient, SshOptions, WorkerCapabilities};

    let Some(_probe_guard) = worker.try_capability_probe() else {
        debug!("Skipping capability refresh: another probe is already running for this worker");
        return None;
    };
    let worker_config = worker.config.read().await;

    // Check if mock mode is enabled
    if is_mock_transport(worker) {
        // In mock mode, return mock capabilities that include Rust
        // (so mock workers can be selected for Rust compilation)
        debug!(
            "Worker {} capabilities probe: mock mode, returning mock capabilities",
            worker_config.id
        );
        return Some(WorkerCapabilities::mock_with_rust());
    }

    let ssh_options = SshOptions {
        connect_timeout: timeout,
        command_timeout: timeout,
        control_master: false,
        ..Default::default()
    };

    // Try to run rch-wkr capabilities command
    // Handle PATH vs ~/.local/bin lookup.
    //
    // Operator-declared tool probes are appended ONLY when this worker declares
    // any. A worker with no declarations therefore receives the byte-identical
    // command older deployments have always received, so adding this feature
    // cannot break capability probing on a fleet that has not been redeployed.
    // A worker that DOES declare tools and still runs an old binary fails the
    // probe, falls back to its cached snapshot, and stays inadmissible for
    // `--require-tool` work — the safe direction for a gate.
    let tool_argument = capability_tool_argument(&worker_config.tools);
    let cmd = format!(
        "if command -v rch-wkr >/dev/null 2>&1; then rch-wkr capabilities{tool_argument}; else ~/.local/bin/rch-wkr capabilities{tool_argument}; fi"
    );
    let cmd = cmd.as_str();

    // Pooled path: run over the warm shared ControlMaster.
    if let Some(pool) = ssh_pool {
        match pool.run_with_timeout(&worker_config, cmd, timeout).await {
            Ok(result) => {
                if result.success() {
                    match serde_json::from_str::<WorkerCapabilities>(&result.stdout) {
                        Ok(capabilities) => {
                            debug!(
                                "Worker {} capabilities: rustc={:?}, bun={:?}, node={:?}",
                                worker_config.id,
                                capabilities.rustc_version,
                                capabilities.bun_version,
                                capabilities.node_version
                            );
                            return Some(capabilities);
                        }
                        Err(e) => {
                            debug!(
                                "Worker {} capabilities JSON parse failed: {} (output: {})",
                                worker_config.id,
                                e,
                                result.stdout.trim()
                            );
                        }
                    }
                } else {
                    debug!(
                        "Worker {} capabilities probe failed (rch-wkr may not be installed): exit={}",
                        worker_config.id, result.exit_code
                    );
                }
            }
            Err(e) => {
                debug!(
                    "Worker {} capabilities probe (pooled) failed: {}",
                    worker_config.id, e
                );
            }
        }
        return None;
    }

    let mut client = SshClient::new(worker_config.clone(), ssh_options);

    match client.connect().await {
        Ok(()) => {
            match client.execute(cmd).await {
                Ok(result) => {
                    let _ = client.disconnect().await;

                    if result.success() {
                        // Parse JSON output
                        match serde_json::from_str::<WorkerCapabilities>(&result.stdout) {
                            Ok(capabilities) => {
                                debug!(
                                    "Worker {} capabilities: rustc={:?}, bun={:?}, node={:?}",
                                    worker_config.id,
                                    capabilities.rustc_version,
                                    capabilities.bun_version,
                                    capabilities.node_version
                                );
                                return Some(capabilities);
                            }
                            Err(e) => {
                                debug!(
                                    "Worker {} capabilities JSON parse failed: {} (output: {})",
                                    worker_config.id,
                                    e,
                                    result.stdout.trim()
                                );
                            }
                        }
                    } else {
                        // rch-wkr might not be installed yet, that's OK
                        debug!(
                            "Worker {} capabilities probe failed (rch-wkr may not be installed): exit={}",
                            worker_config.id, result.exit_code
                        );
                    }
                }
                Err(e) => {
                    let _ = client.disconnect().await;
                    debug!(
                        "Worker {} capabilities probe command failed: {}",
                        worker_config.id, e
                    );
                }
            }
        }
        Err(e) => {
            debug!(
                "Worker {} capabilities probe connection failed: {}",
                worker_config.id, e
            );
        }
    }

    None
}

#[cfg(test)]
mod tests {
    use super::*;
    use rch_common::mock::{
        clear_mock_overrides, set_mock_enabled_override, set_mock_ssh_config_override,
    };
    use rch_common::test_guard;
    use rch_common::{WorkerConfig, WorkerId};
    use std::sync::OnceLock;

    #[test]
    fn capability_tool_argument_is_absent_without_declarations_and_quoted_with_them() {
        let _guard = test_guard!();
        // THE compatibility property: a worker that declares no tools receives
        // the byte-identical command older deployments have always received, so
        // shipping this feature cannot break capability probing on a fleet that
        // has not been redeployed.
        assert_eq!(capability_tool_argument(&[]), "");

        let tools = vec![rch_common::types::WorkerToolProbe {
            name: "clang".to_string(),
            command: vec!["clang".to_string(), "--version".to_string()],
        }];
        let argument = capability_tool_argument(&tools);
        assert!(argument.starts_with(" --tool-probe '"), "{argument}");
        assert!(argument.ends_with('\''), "{argument}");
        assert!(argument.contains(r#""name":"clang""#), "{argument}");

        // The payload is interpolated into a REMOTE shell command, so a single
        // quote inside it must not be able to end the quoted argument. (Config
        // load rejects such names, but this function does not depend on an
        // upstream validator having run.)
        let tricky = vec![rch_common::types::WorkerToolProbe {
            name: "t".to_string(),
            command: vec!["sh".to_string(), "it's".to_string()],
        }];
        let argument = capability_tool_argument(&tricky);
        assert!(!argument.contains("it's"), "raw quote survived: {argument}");
        assert!(argument.contains(r"'\''"), "{argument}");
    }

    #[test]
    fn test_duration_millis_u64_saturates() {
        let _guard = test_guard!();
        assert_eq!(duration_millis_u64(Duration::from_secs(u64::MAX)), u64::MAX);
    }

    fn test_lock() -> &'static Mutex<()> {
        static ENV_MUTEX: OnceLock<Mutex<()>> = OnceLock::new();
        ENV_MUTEX.get_or_init(|| Mutex::new(()))
    }

    struct MockOverrideGuard;

    impl MockOverrideGuard {
        fn set_failure() -> Self {
            set_mock_enabled_override(Some(true));
            set_mock_ssh_config_override(Some(MockConfig::connection_failure()));
            Self
        }
    }

    impl Drop for MockOverrideGuard {
        fn drop(&mut self) {
            clear_mock_overrides();
        }
    }

    #[test]
    fn test_health_config_default() {
        let _guard = test_guard!();
        let config = HealthConfig::default();
        assert_eq!(config.check_interval, Duration::from_secs(30));
        assert_eq!(config.failure_threshold, 3);
    }

    #[test]
    fn test_health_check_result_success() {
        let _guard = test_guard!();
        let result = HealthCheckResult::success(100);
        assert!(result.healthy);
        assert_eq!(result.response_time_ms, 100);
        assert!(result.error.is_none());
    }

    #[test]
    fn test_health_check_result_failure() {
        let _guard = test_guard!();
        let result = HealthCheckResult::failure("Connection timeout".to_string());
        assert!(!result.healthy);
        assert!(result.error.is_some());
    }

    #[test]
    fn test_worker_health_update_success() {
        let _guard = test_guard!();
        let config = HealthConfig::default();
        let mut health = WorkerHealth::default();

        // Successful check
        let result = HealthCheckResult::success(100);
        health.update(result, &config, "test-worker");
        assert_eq!(health.status(), WorkerStatus::Healthy);
        assert_eq!(health.circuit_stats().consecutive_failures(), 0);
    }

    #[test]
    fn test_worker_health_update_degraded() {
        let _guard = test_guard!();
        let config = HealthConfig::default();
        let mut health = WorkerHealth::default();

        // Slow response (degraded)
        let result = HealthCheckResult::success(6000); // Over threshold
        health.update(result, &config, "test-worker");
        assert_eq!(health.status(), WorkerStatus::Degraded);
    }

    #[test]
    fn test_worker_health_update_unreachable() {
        let _guard = test_guard!();
        let config = HealthConfig {
            failure_threshold: 3,
            ..Default::default()
        };
        let mut health = WorkerHealth::default();

        // Multiple failures
        for _ in 0..3 {
            let result = HealthCheckResult::failure("Connection failed".to_string());
            health.update(result, &config, "test-worker");
        }

        assert_eq!(health.status(), WorkerStatus::Unreachable);
        assert_eq!(health.circuit_stats().consecutive_failures(), 3);
    }

    #[test]
    fn test_worker_health_recovery() {
        let _guard = test_guard!();
        let config = HealthConfig::default();
        let mut health = WorkerHealth::default();

        // Fail twice
        for _ in 0..2 {
            let result = HealthCheckResult::failure("Error".to_string());
            health.update(result, &config, "test-worker");
        }
        assert_eq!(health.circuit_stats().consecutive_failures(), 2);

        // Then succeed
        let result = HealthCheckResult::success(100);
        health.update(result, &config, "test-worker");
        assert_eq!(health.status(), WorkerStatus::Healthy);
        assert_eq!(health.circuit_stats().consecutive_failures(), 0);
    }

    #[test]
    fn test_circuit_opens_on_failure_threshold() {
        let _guard = test_guard!();
        let config = HealthConfig {
            circuit: CircuitBreakerConfig {
                failure_threshold: 3,
                ..Default::default()
            },
            ..Default::default()
        };
        let mut health = WorkerHealth::default();

        // Initial state is closed
        assert_eq!(health.circuit_state(), CircuitState::Closed);

        // Fail up to threshold
        for i in 0..3 {
            let result = HealthCheckResult::failure("Connection failed".to_string());
            health.update(result, &config, "test-worker");
            if i < 2 {
                // Circuit still closed before threshold
                assert_eq!(health.circuit_state(), CircuitState::Closed);
            }
        }

        // Circuit should now be open
        assert_eq!(health.circuit_state(), CircuitState::Open);
        assert_eq!(health.status(), WorkerStatus::Unreachable);
    }

    #[test]
    fn test_circuit_transitions_to_half_open() {
        let _guard = test_guard!();
        let config = HealthConfig {
            circuit: CircuitBreakerConfig {
                failure_threshold: 2,
                open_cooldown_secs: 0, // Instant cooldown for testing
                ..Default::default()
            },
            ..Default::default()
        };
        let mut health = WorkerHealth::default();

        // With open_cooldown_secs=0, the circuit opens then immediately transitions
        // to half-open in the same update() call when should_half_open() is checked.
        for _ in 0..2 {
            let result = HealthCheckResult::failure("Error".to_string());
            health.update(result, &config, "test-worker");
        }
        // With cooldown=0, we go straight to HalfOpen after opening
        assert_eq!(health.circuit_state(), CircuitState::HalfOpen);
    }

    #[test]
    fn test_circuit_closes_after_success_in_half_open() {
        let _guard = test_guard!();
        let config = HealthConfig {
            circuit: CircuitBreakerConfig {
                failure_threshold: 2,
                success_threshold: 2,
                open_cooldown_secs: 0,
                ..Default::default()
            },
            ..Default::default()
        };
        let mut health = WorkerHealth::default();

        // Open and transition to half-open
        for _ in 0..2 {
            let result = HealthCheckResult::failure("Error".to_string());
            health.update(result, &config, "test-worker");
        }
        // Trigger half-open transition
        let result = HealthCheckResult::success(50);
        health.update(result, &config, "test-worker");
        assert_eq!(health.circuit_state(), CircuitState::HalfOpen);

        // One more success should close circuit (success_threshold=2)
        let result = HealthCheckResult::success(50);
        health.update(result, &config, "test-worker");
        assert_eq!(health.circuit_state(), CircuitState::Closed);
        assert_eq!(health.status(), WorkerStatus::Healthy);
    }

    #[test]
    fn test_circuit_reopens_on_failure_in_half_open() {
        let _guard = test_guard!();
        // Use two configs: one with cooldown=0 to quickly get to half-open,
        // then one with longer cooldown to verify reopen stays open
        let config_fast = HealthConfig {
            circuit: CircuitBreakerConfig {
                failure_threshold: 2,
                open_cooldown_secs: 0,
                ..Default::default()
            },
            ..Default::default()
        };
        let config_slow = HealthConfig {
            circuit: CircuitBreakerConfig {
                failure_threshold: 2,
                open_cooldown_secs: 60, // Long cooldown so circuit stays open
                ..Default::default()
            },
            ..Default::default()
        };
        let mut health = WorkerHealth::default();

        // Open and transition to half-open (with fast cooldown)
        for _ in 0..2 {
            let result = HealthCheckResult::failure("Error".to_string());
            health.update(result, &config_fast, "test-worker");
        }
        // With cooldown=0, we're now in HalfOpen
        assert_eq!(health.circuit_state(), CircuitState::HalfOpen);

        // Failure in half-open should reopen circuit (use slow config so it stays open)
        let result = HealthCheckResult::failure("Failed again".to_string());
        health.update(result, &config_slow, "test-worker");
        assert_eq!(health.circuit_state(), CircuitState::Open);
        assert_eq!(health.status(), WorkerStatus::Unreachable);
    }

    #[test]
    fn test_transient_failure_in_half_open_does_not_reopen() {
        let _guard = test_guard!();
        // A TRANSIENT half-open failure must NOT reopen the circuit (contrast
        // with test_circuit_reopens_on_failure_in_half_open above). This is what
        // stops a single retryable transport blip during recovery from bouncing
        // a worker back to Unreachable.
        let config_fast = HealthConfig {
            circuit: CircuitBreakerConfig {
                failure_threshold: 2,
                open_cooldown_secs: 0,
                ..Default::default()
            },
            ..Default::default()
        };
        let config_slow = HealthConfig {
            circuit: CircuitBreakerConfig {
                failure_threshold: 2,
                open_cooldown_secs: 60,
                ..Default::default()
            },
            ..Default::default()
        };
        let mut health = WorkerHealth::default();

        // Drive to half-open.
        for _ in 0..2 {
            health.update(
                HealthCheckResult::failure("Error".to_string()),
                &config_fast,
                "test-worker",
            );
        }
        assert_eq!(health.circuit_state(), CircuitState::HalfOpen);

        // A TRANSIENT failure in half-open (slow cooldown) must stay HalfOpen.
        health.update(
            HealthCheckResult::transient_failure("blip".to_string()),
            &config_slow,
            "test-worker",
        );
        assert_eq!(
            health.circuit_state(),
            CircuitState::HalfOpen,
            "transient half-open failure must not reopen the circuit"
        );
    }

    #[test]
    fn test_transient_failure_result_carries_flag() {
        let _guard = test_guard!();
        let transient = HealthCheckResult::transient_failure("timed out".to_string());
        assert!(!transient.healthy);
        assert!(transient.transient);
        // Plain failures/successes are NOT flagged transient.
        assert!(!HealthCheckResult::failure("nope".to_string()).transient);
        assert!(!HealthCheckResult::success(10).transient);
    }

    #[test]
    fn test_update_and_observe_agree_on_circuit() {
        let _guard = test_guard!();
        // update() and observe() must drive the circuit identically (update
        // delegates to observe). Feed the same outcome sequence to two
        // WorkerHealth copies via the two entry points and assert equal state.
        let config = HealthConfig {
            circuit: CircuitBreakerConfig {
                failure_threshold: 2,
                success_threshold: 2,
                open_cooldown_secs: 0,
                ..Default::default()
            },
            ..Default::default()
        };
        let mut via_update = WorkerHealth::default();
        let mut via_observe = WorkerHealth::default();

        let outcomes = [(false, false), (false, false), (true, false), (true, false)];
        for (healthy, transient) in outcomes {
            let result = if healthy {
                HealthCheckResult::success(10)
            } else if transient {
                HealthCheckResult::transient_failure("t".to_string())
            } else {
                HealthCheckResult::failure("f".to_string())
            };
            via_update.update(result, &config, "w");
            via_observe.observe(healthy, transient, 10, &config, "w");
            assert_eq!(
                via_update.circuit_state(),
                via_observe.circuit_state(),
                "update() and observe() diverged"
            );
        }
    }

    #[test]
    fn test_circuit_stats_accessors() {
        let _guard = test_guard!();
        let config = HealthConfig::default();
        let mut health = WorkerHealth::default();

        // Initially no error
        assert!(health.last_error().is_none());

        // After failure, error is stored
        let result = HealthCheckResult::failure("Test error message".to_string());
        health.update(result, &config, "test-worker");
        assert_eq!(health.last_error(), Some("Test error message"));

        // After success, error is cleared
        let result = HealthCheckResult::success(50);
        health.update(result, &config, "test-worker");
        assert!(health.last_error().is_none());
    }

    #[tokio::test]
    async fn test_check_worker_health_mock_failure() {
        let _lock = test_lock().lock().await;
        let _overrides = MockOverrideGuard::set_failure();

        let worker = WorkerState::new(WorkerConfig {
            id: WorkerId::new("mock-fail"),
            host: "mock.host".to_string(),
            user: "mockuser".to_string(),
            identity_file: "~/.ssh/mock".to_string(),
            total_slots: 4,
            priority: 100,
            tags: vec![],
            tools: Vec::new(),
        });

        let result = check_worker_health(&Arc::new(worker), &HealthConfig::default(), None).await;
        assert!(!result.healthy);
    }

    // ============================================================================
    // Integration tests: Health monitoring -> Circuit breaker -> Worker selection
    // ============================================================================

    mod integration_tests {
        use super::*;
        use crate::selection::{SelectionWeights, select_worker_with_config};
        use crate::workers::WorkerPool;
        use rch_common::{CommandPriority, RequiredRuntime, SelectionRequest};

        fn make_worker_config(id: &str) -> WorkerConfig {
            WorkerConfig {
                id: WorkerId::new(id),
                host: format!("{}.host", id),
                user: "testuser".to_string(),
                identity_file: "~/.ssh/test".to_string(),
                total_slots: 8,
                priority: 100,
                tags: vec![],
                tools: Vec::new(),
            }
        }

        fn make_request(project: &str, cores: u32) -> SelectionRequest {
            SelectionRequest {
                job_mode: false,
                project: project.to_string(),
                command: None,
                command_priority: CommandPriority::Normal,
                estimated_cores: cores,
                preferred_workers: vec![],
                toolchain: None,
                required_runtime: RequiredRuntime::default(),
                classification_duration_us: None,
                hook_pid: None,
                required_tools: Vec::new(),
            }
        }

        #[tokio::test]
        async fn test_integration_health_failures_cause_selection_exclusion() {
            // Test that health failures leading to open circuit cause worker to be
            // excluded from selection, and when all workers are in open state,
            // selection returns AllCircuitsOpen.

            let pool = WorkerPool::new();
            pool.add_worker(make_worker_config("worker-1")).await;
            pool.add_worker(make_worker_config("worker-2")).await;

            let health_config = HealthConfig {
                circuit: CircuitBreakerConfig {
                    failure_threshold: 3,
                    ..Default::default()
                },
                ..Default::default()
            };
            let weights = SelectionWeights::default();
            let request = make_request("test-project", 2);

            // Initially, selection should succeed
            let result =
                select_worker_with_config(&pool, &request, &weights, &health_config.circuit).await;
            assert!(result.worker.is_some());
            assert_eq!(result.reason, rch_common::SelectionReason::Success);

            // Simulate consecutive failures on worker-1 to open its circuit
            let worker1 = pool.get(&WorkerId::new("worker-1")).await.unwrap();
            for _ in 0..3 {
                worker1
                    .record_failure(Some("Connection timeout".to_string()))
                    .await;
            }
            // Manually check if circuit should open and transition
            if worker1.should_open_circuit(&health_config.circuit).await {
                worker1.open_circuit().await;
            }

            // Selection should still work (worker-2 is available)
            let result =
                select_worker_with_config(&pool, &request, &weights, &health_config.circuit).await;
            assert!(result.worker.is_some());
            let selected = result.worker.unwrap();
            assert_eq!(selected.config.read().await.id.as_str(), "worker-2");

            // Now fail worker-2 as well
            let worker2 = pool.get(&WorkerId::new("worker-2")).await.unwrap();
            for _ in 0..3 {
                worker2
                    .record_failure(Some("Connection timeout".to_string()))
                    .await;
            }
            if worker2.should_open_circuit(&health_config.circuit).await {
                worker2.open_circuit().await;
            }

            // Both circuits open - selection should return AllCircuitsOpen
            let result =
                select_worker_with_config(&pool, &request, &weights, &health_config.circuit).await;
            assert!(result.worker.is_none());
            assert_eq!(result.reason, rch_common::SelectionReason::AllCircuitsOpen);
        }

        #[tokio::test]
        async fn test_integration_circuit_recovery_path() {
            // Test full recovery: Open -> HalfOpen -> Closed
            // Verify selection behavior at each stage.

            let pool = WorkerPool::new();
            pool.add_worker(make_worker_config("recovery-worker")).await;

            let health_config = HealthConfig {
                circuit: CircuitBreakerConfig {
                    failure_threshold: 2,
                    success_threshold: 2,
                    open_cooldown_secs: 0, // Immediate transition to half-open
                    half_open_max_probes: 1,
                    ..Default::default()
                },
                ..Default::default()
            };
            let weights = SelectionWeights::default();
            let request = make_request("test-project", 2);

            let worker = pool.get(&WorkerId::new("recovery-worker")).await.unwrap();

            // Stage 1: Circuit is Closed
            assert_eq!(worker.circuit_state().await.unwrap(), CircuitState::Closed);
            let result =
                select_worker_with_config(&pool, &request, &weights, &health_config.circuit).await;
            assert!(result.worker.is_some());

            // Stage 2: Cause failures to open circuit
            for _ in 0..2 {
                worker.record_failure(Some("Error".to_string())).await;
            }
            if worker.should_open_circuit(&health_config.circuit).await {
                worker.open_circuit().await;
            }
            assert_eq!(worker.circuit_state().await.unwrap(), CircuitState::Open);

            // Stage 3: Transition to half-open (cooldown=0)
            if worker.should_half_open(&health_config.circuit).await {
                worker.half_open_circuit().await;
            }
            assert_eq!(
                worker.circuit_state().await.unwrap(),
                CircuitState::HalfOpen
            );

            // Stage 4: Selection should work in half-open with probe budget
            let result =
                select_worker_with_config(&pool, &request, &weights, &health_config.circuit).await;
            assert!(result.worker.is_some());

            // Stage 5: Record successes to close circuit
            worker.record_success().await;
            worker.record_success().await;
            if worker.should_close_circuit(&health_config.circuit).await {
                worker.close_circuit().await;
            }
            assert_eq!(worker.circuit_state().await.unwrap(), CircuitState::Closed);

            // Stage 6: Selection works normally again
            let result =
                select_worker_with_config(&pool, &request, &weights, &health_config.circuit).await;
            assert!(result.worker.is_some());
            assert_eq!(result.reason, rch_common::SelectionReason::Success);
        }

        #[tokio::test]
        async fn test_integration_half_open_probe_exhaustion() {
            // Test that when a half-open worker exhausts its probe budget,
            // it's excluded until the probe completes.

            let pool = WorkerPool::new();
            pool.add_worker(make_worker_config("probe-worker")).await;

            let circuit_config = CircuitBreakerConfig {
                failure_threshold: 2,
                open_cooldown_secs: 0,
                half_open_max_probes: 1,
                ..Default::default()
            };
            let weights = SelectionWeights::default();
            let request = make_request("test-project", 2);

            let worker = pool.get(&WorkerId::new("probe-worker")).await.unwrap();

            // Open and transition to half-open
            for _ in 0..2 {
                worker.record_failure(None).await;
            }
            worker.open_circuit().await;
            worker.half_open_circuit().await;

            // First selection should succeed and start probe
            let result1 =
                select_worker_with_config(&pool, &request, &weights, &circuit_config).await;
            assert!(result1.worker.is_some());

            // Second selection should fail (probe budget exhausted)
            let result2 =
                select_worker_with_config(&pool, &request, &weights, &circuit_config).await;
            // With only one worker and probe exhausted, should return busy/circuits open
            assert!(result2.worker.is_none());
        }

        #[tokio::test]
        async fn test_integration_mixed_circuit_states() {
            // Test selection behavior with workers in different circuit states:
            // - Worker A: Closed (healthy)
            // - Worker B: Open (excluded)
            // - Worker C: HalfOpen (available with penalty)
            // Verify that closed is preferred over half-open due to penalty.

            let pool = WorkerPool::new();
            pool.add_worker(WorkerConfig {
                id: WorkerId::new("closed-worker"),
                host: "closed.host".to_string(),
                user: "testuser".to_string(),
                identity_file: "~/.ssh/test".to_string(),
                total_slots: 8,
                priority: 100,
                tags: vec![],
                tools: Vec::new(),
            })
            .await;
            pool.add_worker(WorkerConfig {
                id: WorkerId::new("open-worker"),
                host: "open.host".to_string(),
                user: "testuser".to_string(),
                identity_file: "~/.ssh/test".to_string(),
                total_slots: 16, // More slots - would normally be preferred
                priority: 100,
                tags: vec![],
                tools: Vec::new(),
            })
            .await;
            pool.add_worker(WorkerConfig {
                id: WorkerId::new("half-open-worker"),
                host: "half-open.host".to_string(),
                user: "testuser".to_string(),
                identity_file: "~/.ssh/test".to_string(),
                total_slots: 12, // More slots than closed
                priority: 100,
                tags: vec![],
                tools: Vec::new(),
            })
            .await;

            let circuit_config = CircuitBreakerConfig {
                failure_threshold: 2,
                open_cooldown_secs: 0,
                half_open_max_probes: 10, // High limit to avoid probe exhaustion
                ..Default::default()
            };
            let weights = SelectionWeights::default();
            let request = make_request("test-project", 2);

            // Set up circuit states
            let open_worker = pool.get(&WorkerId::new("open-worker")).await.unwrap();
            open_worker.open_circuit().await;

            let half_open_worker = pool.get(&WorkerId::new("half-open-worker")).await.unwrap();
            half_open_worker.open_circuit().await;
            half_open_worker.half_open_circuit().await;

            // Selection should prefer closed-worker despite having fewer slots
            // because half-open has penalty and open is excluded
            let result =
                select_worker_with_config(&pool, &request, &weights, &circuit_config).await;
            assert!(result.worker.is_some());
            let selected = result.worker.unwrap();
            assert_eq!(selected.config.read().await.id.as_str(), "closed-worker");
        }

        #[tokio::test]
        async fn test_integration_failure_in_half_open_reopens() {
            // Test that a failure during half-open probe causes circuit to reopen.

            let pool = WorkerPool::new();
            pool.add_worker(make_worker_config("reopen-worker")).await;

            let circuit_config = CircuitBreakerConfig {
                failure_threshold: 2,
                open_cooldown_secs: 0,
                half_open_max_probes: 1,
                ..Default::default()
            };

            let worker = pool.get(&WorkerId::new("reopen-worker")).await.unwrap();

            // Transition to half-open
            worker.record_failure(None).await;
            worker.record_failure(None).await;
            worker.open_circuit().await;
            worker.half_open_circuit().await;
            assert_eq!(
                worker.circuit_state().await.unwrap(),
                CircuitState::HalfOpen
            );

            // Start a probe
            worker.start_probe(&circuit_config).await;

            // Simulate failure during probe - record failure and reopen
            worker
                .record_failure(Some("Probe failed".to_string()))
                .await;
            worker.open_circuit().await; // Failure in half-open should reopen

            assert_eq!(worker.circuit_state().await.unwrap(), CircuitState::Open);
            assert_eq!(worker.last_error().await, Some("Probe failed".to_string()));
        }

        #[tokio::test]
        async fn test_integration_health_update_drives_circuit_transitions() {
            // Test that WorkerHealth.update() correctly drives all circuit
            // state transitions based on health check results.

            let config = HealthConfig {
                circuit: CircuitBreakerConfig {
                    failure_threshold: 3,
                    success_threshold: 2,
                    open_cooldown_secs: 0, // Immediate half-open
                    ..Default::default()
                },
                ..Default::default()
            };
            let mut health = WorkerHealth::default();

            // Initial state: Closed
            assert_eq!(health.circuit_state(), CircuitState::Closed);
            assert_eq!(health.status(), WorkerStatus::Healthy);

            // Failures 1 & 2: Still closed, status degrades
            health.update(
                HealthCheckResult::failure("Error 1".to_string()),
                &config,
                "test-worker",
            );
            assert_eq!(health.circuit_state(), CircuitState::Closed);
            assert_eq!(health.status(), WorkerStatus::Degraded);

            health.update(
                HealthCheckResult::failure("Error 2".to_string()),
                &config,
                "test-worker",
            );
            assert_eq!(health.circuit_state(), CircuitState::Closed);

            // Failure 3: Circuit opens, then immediately goes to half-open
            // (since open_cooldown_secs = 0)
            health.update(
                HealthCheckResult::failure("Error 3".to_string()),
                &config,
                "test-worker",
            );
            // With cooldown=0, after opening it checks should_half_open which is true
            assert_eq!(health.circuit_state(), CircuitState::HalfOpen);
            assert_eq!(health.status(), WorkerStatus::Degraded);

            // Success 1: Still half-open
            health.update(HealthCheckResult::success(50), &config, "test-worker");
            assert_eq!(health.circuit_state(), CircuitState::HalfOpen);

            // Success 2: Circuit closes (success_threshold = 2)
            health.update(HealthCheckResult::success(50), &config, "test-worker");
            assert_eq!(health.circuit_state(), CircuitState::Closed);
            assert_eq!(health.status(), WorkerStatus::Healthy);
        }
    }

    // ============================================================================
    // Additional coverage tests for WorkerHealth methods
    // ============================================================================

    #[test]
    fn test_worker_health_last_result() {
        let _guard = test_guard!();
        let config = HealthConfig::default();
        let mut health = WorkerHealth::default();

        // Initially no result
        assert!(health.last_result().is_none());

        // After update, result is stored
        let result = HealthCheckResult::success(100);
        health.update(result, &config, "test-worker");

        let last = health.last_result();
        assert!(last.is_some());
        assert!(last.unwrap().healthy);
        assert_eq!(last.unwrap().response_time_ms, 100);
    }

    #[test]
    fn test_worker_health_last_result_failure() {
        let _guard = test_guard!();
        let config = HealthConfig::default();
        let mut health = WorkerHealth::default();

        let result = HealthCheckResult::failure("Connection failed".to_string());
        health.update(result, &config, "test-worker");

        let last = health.last_result();
        assert!(last.is_some());
        assert!(!last.unwrap().healthy);
        assert!(last.unwrap().error.is_some());
    }

    #[test]
    fn test_worker_health_can_probe_closed() {
        let _guard = test_guard!();
        let config = HealthConfig::default();
        let health = WorkerHealth::default();

        // Closed circuit does NOT probe (probing is only for half-open)
        assert!(!health.can_probe(&config));
    }

    #[test]
    fn test_worker_health_can_probe_half_open() {
        let _guard = test_guard!();
        let config = HealthConfig {
            circuit: CircuitBreakerConfig {
                failure_threshold: 2,
                open_cooldown_secs: 0,
                ..Default::default()
            },
            ..Default::default()
        };
        let mut health = WorkerHealth::default();

        // Open the circuit, then transition to half-open
        for _ in 0..2 {
            health.update(
                HealthCheckResult::failure("Error".to_string()),
                &config,
                "test-worker",
            );
        }
        assert_eq!(health.circuit_state(), CircuitState::HalfOpen);

        // Half-open circuit can probe (limited)
        assert!(health.can_probe(&config));
    }

    #[test]
    fn test_worker_health_start_probe() {
        let _guard = test_guard!();
        let config = HealthConfig {
            circuit: CircuitBreakerConfig {
                failure_threshold: 2,
                open_cooldown_secs: 0,
                ..Default::default()
            },
            ..Default::default()
        };
        let mut health = WorkerHealth::default();

        // Get to half-open state
        for _ in 0..2 {
            health.update(
                HealthCheckResult::failure("Error".to_string()),
                &config,
                "test-worker",
            );
        }
        assert_eq!(health.circuit_state(), CircuitState::HalfOpen);

        // First probe should succeed
        assert!(health.start_probe(&config));
    }

    #[test]
    fn test_health_monitor_new() {
        let _guard = test_guard!();
        let pool = WorkerPool::new();
        let config = HealthConfig::default();
        let _monitor = HealthMonitor::new(pool, config);
        // Just verify it can be created without panic
    }

    #[test]
    fn test_health_monitor_with_status_panel() {
        let _guard = test_guard!();
        let pool = WorkerPool::new();
        let config = HealthConfig::default();
        let panel = Arc::new(Mutex::new(WorkerStatusPanel::new()));

        let monitor = HealthMonitor::new(pool, config).with_status_panel(panel.clone());

        // Verify panel is attached
        assert!(monitor.status_panel.is_some());
    }

    #[test]
    fn test_health_monitor_with_alert_manager() {
        let _guard = test_guard!();
        let pool = WorkerPool::new();
        let config = HealthConfig::default();
        let alert_manager = Arc::new(AlertManager::new(crate::alerts::AlertConfig::default()));

        let monitor = HealthMonitor::new(pool, config).with_alert_manager(alert_manager.clone());

        // Verify alert manager is attached
        assert!(monitor.alert_manager.is_some());
    }

    #[test]
    fn test_health_check_result_record_metrics() {
        let _guard = test_guard!();
        // Test success case - should not panic
        let success = HealthCheckResult::success(100);
        success.record_metrics("test-worker-1");

        // Test failure case - should not panic
        let failure = HealthCheckResult::failure("Error".to_string());
        failure.record_metrics("test-worker-2");
    }

    #[test]
    fn test_health_config_custom() {
        let _guard = test_guard!();
        let config = HealthConfig {
            check_interval: Duration::from_secs(60),
            check_timeout: Duration::from_secs(5),
            degraded_threshold_ms: 3000,
            failure_threshold: 5,
            circuit: CircuitBreakerConfig {
                failure_threshold: 10,
                ..Default::default()
            },
        };

        assert_eq!(config.check_interval, Duration::from_secs(60));
        assert_eq!(config.check_timeout, Duration::from_secs(5));
        assert_eq!(config.degraded_threshold_ms, 3000);
        assert_eq!(config.failure_threshold, 5);
        assert_eq!(config.circuit.failure_threshold, 10);
    }

    #[test]
    fn test_health_check_result_checked_at() {
        let _guard = test_guard!();
        let before = Instant::now();
        let result = HealthCheckResult::success(50);
        let after = Instant::now();

        // checked_at should be between before and after
        assert!(result.checked_at >= before);
        assert!(result.checked_at <= after);
    }

    // ============================================================================
    // Additional coverage tests for HealthMonitor async methods
    // ============================================================================

    #[tokio::test]
    async fn test_health_monitor_stop() {
        let _lock = test_lock().lock().await;
        let pool = WorkerPool::new();
        let config = HealthConfig::default();
        let monitor = HealthMonitor::new(pool, config);

        // Verify initial state
        assert!(!*monitor.running.read().await);

        // Stop should set running to false (idempotent when not started)
        monitor.stop().await;
        assert!(!*monitor.running.read().await);
    }

    #[tokio::test]
    async fn test_health_monitor_get_health_empty() {
        let _lock = test_lock().lock().await;
        let pool = WorkerPool::new();
        let config = HealthConfig::default();
        let monitor = HealthMonitor::new(pool, config);

        // No workers tracked yet
        let health = monitor.get_health("nonexistent-worker").await;
        assert!(health.is_none());
    }

    #[tokio::test]
    async fn test_health_monitor_get_health_after_manual_insert() {
        let _lock = test_lock().lock().await;
        let pool = WorkerPool::new();
        let config = HealthConfig::default();
        let monitor = HealthMonitor::new(pool, config);

        // Manually insert a health state
        {
            let mut states = monitor.health_states.write().await;
            let health = WorkerHealth {
                current_status: WorkerStatus::Degraded,
                ..Default::default()
            };
            states.insert("test-worker".to_string(), health);
        }

        // Now get_health should find it
        let health = monitor.get_health("test-worker").await;
        assert_eq!(health, Some(WorkerStatus::Degraded));
    }

    #[tokio::test]
    async fn test_health_monitor_all_health_states_empty() {
        let _lock = test_lock().lock().await;
        let pool = WorkerPool::new();
        let config = HealthConfig::default();
        let monitor = HealthMonitor::new(pool, config);

        let states = monitor.all_health_states().await;
        assert!(states.is_empty());
    }

    #[tokio::test]
    async fn test_health_monitor_all_health_states_multiple() {
        let _lock = test_lock().lock().await;
        let pool = WorkerPool::new();
        let config = HealthConfig::default();
        let monitor = HealthMonitor::new(pool, config);

        // Insert multiple health states
        {
            let mut states = monitor.health_states.write().await;

            let health1 = WorkerHealth {
                current_status: WorkerStatus::Healthy,
                ..Default::default()
            };
            states.insert("worker-1".to_string(), health1);

            let health2 = WorkerHealth {
                current_status: WorkerStatus::Degraded,
                ..Default::default()
            };
            states.insert("worker-2".to_string(), health2);

            let health3 = WorkerHealth {
                current_status: WorkerStatus::Unreachable,
                ..Default::default()
            };
            states.insert("worker-3".to_string(), health3);
        }

        let all_states = monitor.all_health_states().await;
        assert_eq!(all_states.len(), 3);

        // Check that all expected workers are present
        let worker_ids: Vec<&str> = all_states.iter().map(|(id, _)| id.as_str()).collect();
        assert!(worker_ids.contains(&"worker-1"));
        assert!(worker_ids.contains(&"worker-2"));
        assert!(worker_ids.contains(&"worker-3"));
    }

    #[tokio::test]
    async fn test_health_monitor_does_not_revive_draining_worker() {
        let _lock = test_lock().lock().await;
        set_mock_enabled_override(Some(true));
        set_mock_ssh_config_override(Some(MockConfig::success().with_stdout("health_check")));

        let pool = WorkerPool::new();
        let worker_id = WorkerId::new("draining-worker");
        pool.add_worker(WorkerConfig {
            id: worker_id.clone(),
            host: "draining.host".to_string(),
            user: "user".to_string(),
            identity_file: "~/.ssh/key".to_string(),
            total_slots: 4,
            priority: 100,
            tags: vec![],
            tools: Vec::new(),
        })
        .await;

        let worker = pool.get(&worker_id).await.unwrap();
        assert!(worker.reserve_slots(1).await);
        worker.drain().await;

        let monitor = HealthMonitor::new(
            pool,
            HealthConfig {
                check_interval: Duration::from_millis(10),
                check_timeout: Duration::from_secs(1),
                ..Default::default()
            },
        );

        let handle = monitor.start();
        tokio::time::sleep(Duration::from_millis(35)).await;
        monitor.stop().await;
        handle.await.unwrap();

        assert_eq!(worker.status().await, WorkerStatus::Draining);
        assert_eq!(
            monitor.get_health(worker_id.as_str()).await,
            Some(WorkerStatus::Healthy)
        );

        clear_mock_overrides();
    }

    // ============================================================================
    // Tests for probe_worker function
    // ============================================================================

    #[tokio::test]
    async fn test_probe_worker_mock_success() {
        let _lock = test_lock().lock().await;
        set_mock_enabled_override(Some(true));
        // Health check expects "health_check" as the response
        set_mock_ssh_config_override(Some(MockConfig::success().with_stdout("health_check")));

        let worker = WorkerState::new(WorkerConfig {
            id: WorkerId::new("probe-test"),
            host: "probe.host".to_string(),
            user: "probeuser".to_string(),
            identity_file: "~/.ssh/probe".to_string(),
            total_slots: 8,
            priority: 100,
            tags: vec![],
            tools: Vec::new(),
        });

        let result = probe_worker(&worker).await;

        clear_mock_overrides();

        assert!(result.healthy);
        assert!(result.response_time_ms > 0);
    }

    #[tokio::test]
    async fn test_probe_worker_mock_failure() {
        let _lock = test_lock().lock().await;
        set_mock_enabled_override(Some(true));
        set_mock_ssh_config_override(Some(MockConfig::connection_failure()));

        let worker = WorkerState::new(WorkerConfig {
            id: WorkerId::new("probe-fail"),
            host: "fail.host".to_string(),
            user: "failuser".to_string(),
            identity_file: "~/.ssh/fail".to_string(),
            total_slots: 4,
            priority: 50,
            tags: vec![],
            tools: Vec::new(),
        });

        let result = probe_worker(&worker).await;

        clear_mock_overrides();

        assert!(!result.healthy);
        assert!(result.error.is_some());
    }

    // ============================================================================
    // Tests for probe_worker_capabilities function
    // ============================================================================

    #[tokio::test]
    async fn test_probe_worker_capabilities_mock_mode() {
        let _lock = test_lock().lock().await;
        set_mock_enabled_override(Some(true));

        let worker_config = WorkerConfig {
            id: WorkerId::new("cap-mock"),
            host: "cap.mock.host".to_string(),
            user: "capuser".to_string(),
            identity_file: "~/.ssh/cap".to_string(),
            total_slots: 8,
            priority: 100,
            tags: vec![],
            tools: Vec::new(),
        };
        let worker = Arc::new(WorkerState::new(worker_config));

        let capabilities = probe_worker_capabilities(&worker, Duration::from_secs(5), None).await;

        clear_mock_overrides();

        // In mock mode, should return mock capabilities
        assert!(capabilities.is_some());
        let caps = capabilities.unwrap();
        // Mock capabilities include Rust
        assert!(caps.rustc_version.is_some());
    }

    #[tokio::test]
    async fn capability_refresh_skips_busy_worker_without_blocking_other_workers() {
        let _lock = test_lock().lock().await;
        set_mock_enabled_override(Some(true));
        let worker = Arc::new(WorkerState::new(WorkerConfig {
            id: WorkerId::new("cap-busy"),
            ..Default::default()
        }));
        let other = Arc::new(WorkerState::new(WorkerConfig {
            id: WorkerId::new("cap-other"),
            ..Default::default()
        }));
        let mut existing = rch_common::WorkerCapabilities::new();
        existing.rustc_version = Some("previous observation".to_owned());
        worker.set_capabilities(existing).await;

        // Represent an in-flight caller using the same guard that the public
        // probe entry point acquires before looking up any transport.
        let active = worker.try_capability_probe().unwrap();
        let skipped = tokio::time::timeout(
            Duration::from_millis(100),
            probe_worker_capabilities(&worker, Duration::from_secs(5), None),
        )
        .await;
        let independent = probe_worker_capabilities(&other, Duration::from_secs(5), None).await;
        let unchanged = worker.capabilities().await;
        drop(active);
        let recovered = probe_worker_capabilities(&worker, Duration::from_secs(5), None).await;
        clear_mock_overrides();

        assert!(
            skipped.unwrap().is_none(),
            "busy probe must not reach transport"
        );
        assert!(independent.is_some(), "one worker must not block another");
        assert_eq!(
            unchanged.rustc_version.as_deref(),
            Some("previous observation")
        );
        assert!(
            recovered.is_some(),
            "completion must permit a later refresh"
        );
    }

    #[tokio::test]
    async fn capability_refresh_guard_recovers_after_request_cancellation() {
        let _lock = test_lock().lock().await;
        set_mock_enabled_override(Some(true));
        let worker = Arc::new(WorkerState::new(WorkerConfig {
            id: WorkerId::new("cap-cancelled"),
            ..Default::default()
        }));
        // Stall the real probe after it claims the per-worker guard, before
        // it can inspect the configuration or reach a transport.
        let config_guard = worker.config.write().await;
        let task_worker = Arc::clone(&worker);
        let task = tokio::spawn(async move {
            probe_worker_capabilities(&task_worker, Duration::from_secs(5), None).await
        });
        let claimed = tokio::time::timeout(Duration::from_secs(1), async {
            while worker.try_capability_probe().is_some() {
                tokio::task::yield_now().await;
            }
        })
        .await;
        let skipped = tokio::time::timeout(
            Duration::from_millis(100),
            probe_worker_capabilities(&worker, Duration::from_secs(5), None),
        )
        .await;
        task.abort();
        let cancelled = task.await;
        drop(config_guard);
        let recovered = probe_worker_capabilities(&worker, Duration::from_secs(5), None).await;
        clear_mock_overrides();

        assert!(claimed.is_ok(), "the real probe did not acquire its guard");
        assert!(skipped.unwrap().is_none());
        assert!(cancelled.unwrap_err().is_cancelled());
        assert!(
            recovered.is_some(),
            "a cancelled refresh must not wedge the worker"
        );
    }

    // ============================================================================
    // Tests for check_worker_health edge cases
    // ============================================================================

    #[tokio::test]
    async fn test_check_worker_health_mock_command_failure() {
        let _lock = test_lock().lock().await;
        set_mock_enabled_override(Some(true));
        set_mock_ssh_config_override(Some(MockConfig::command_failure(42, "command failed")));

        let worker_config = WorkerConfig {
            id: WorkerId::new("cmd-fail"),
            host: "cmdfail.host".to_string(),
            user: "cmduser".to_string(),
            identity_file: "~/.ssh/cmd".to_string(),
            total_slots: 4,
            priority: 50,
            tags: vec![],
            tools: Vec::new(),
        };
        let worker = Arc::new(WorkerState::new(worker_config));

        let result = check_worker_health(&worker, &HealthConfig::default(), None).await;

        clear_mock_overrides();

        // Command failure should result in unhealthy
        assert!(!result.healthy);
        assert!(result.error.is_some());
    }

    #[tokio::test]
    async fn test_check_worker_health_mock_unexpected_output() {
        let _lock = test_lock().lock().await;
        set_mock_enabled_override(Some(true));
        // Create a mock config that returns success but with wrong output
        let config = MockConfig::success().with_stdout("wrong_output");
        set_mock_ssh_config_override(Some(config));

        let worker_config = WorkerConfig {
            id: WorkerId::new("bad-output"),
            host: "badout.host".to_string(),
            user: "baduser".to_string(),
            identity_file: "~/.ssh/bad".to_string(),
            total_slots: 4,
            priority: 50,
            tags: vec![],
            tools: Vec::new(),
        };
        let worker = Arc::new(WorkerState::new(worker_config));

        let result = check_worker_health(&worker, &HealthConfig::default(), None).await;

        clear_mock_overrides();

        // Wrong output should result in failure
        assert!(!result.healthy);
        assert!(result.error.is_some());
        assert!(result.error.unwrap().contains("Unexpected response"));
    }

    // ============================================================================
    // Tests for WorkerHealth new() method
    // ============================================================================

    #[test]
    fn test_worker_health_new() {
        let _guard = test_guard!();
        let health = WorkerHealth::new();

        assert!(health.last_result.is_none());
        assert_eq!(health.current_status, WorkerStatus::Healthy);
        assert_eq!(health.circuit.state(), CircuitState::Closed);
        assert!(health.last_error.is_none());
    }

    // ============================================================================
    // Tests for circuit stats accessor
    // ============================================================================

    #[test]
    fn test_worker_health_circuit_stats() {
        let _guard = test_guard!();
        let config = HealthConfig::default();
        let mut health = WorkerHealth::default();

        // Record some failures to change stats
        health.update(
            HealthCheckResult::failure("Error 1".to_string()),
            &config,
            "test-worker",
        );
        health.update(
            HealthCheckResult::failure("Error 2".to_string()),
            &config,
            "test-worker",
        );

        let stats = health.circuit_stats();
        assert_eq!(stats.consecutive_failures(), 2);
    }

    // ============================================================================
    // Tests for is_mock_transport helper
    // ============================================================================

    #[test]
    fn test_is_mock_transport_disabled() {
        let _guard = test_guard!();
        // Note: Don't call clear_mock_overrides() at test start - it breaks the
        // push/pop semantics of the active_scopes mechanism and causes race
        // conditions with parallel tests. Instead, just set the override directly.

        let worker = WorkerState::new(WorkerConfig {
            id: WorkerId::new("transport-test"),
            host: "transport.host".to_string(),
            user: "user".to_string(),
            identity_file: "~/.ssh/key".to_string(),
            total_slots: 4,
            priority: 100,
            tags: vec![],
            tools: Vec::new(),
        });

        // When mock is not enabled, is_mock_transport returns false
        set_mock_enabled_override(Some(false));
        assert!(!is_mock_transport(&worker));
        clear_mock_overrides();
    }

    #[test]
    fn test_is_mock_transport_enabled() {
        let _guard = test_guard!();

        let worker = WorkerState::new(WorkerConfig {
            id: WorkerId::new("transport-test-2"),
            host: "transport2.host".to_string(),
            user: "user".to_string(),
            identity_file: "~/.ssh/key".to_string(),
            total_slots: 4,
            priority: 100,
            tags: vec![],
            tools: Vec::new(),
        });

        set_mock_enabled_override(Some(true));
        assert!(is_mock_transport(&worker));
        clear_mock_overrides();
    }

    // ============================================================================
    // Edge case tests for circuit state transitions
    // ============================================================================

    #[test]
    fn test_circuit_stays_open_with_long_cooldown() {
        let _guard = test_guard!();
        let config = HealthConfig {
            circuit: CircuitBreakerConfig {
                failure_threshold: 2,
                open_cooldown_secs: 3600, // Long cooldown
                ..Default::default()
            },
            ..Default::default()
        };
        let mut health = WorkerHealth::default();

        // Fail enough to open circuit
        for _ in 0..2 {
            health.update(
                HealthCheckResult::failure("Error".to_string()),
                &config,
                "test-worker",
            );
        }

        // Circuit should be Open (not HalfOpen) because cooldown hasn't elapsed
        assert_eq!(health.circuit_state(), CircuitState::Open);
        assert_eq!(health.status(), WorkerStatus::Unreachable);
    }

    #[test]
    fn test_success_clears_consecutive_failures() {
        let _guard = test_guard!();
        let config = HealthConfig {
            circuit: CircuitBreakerConfig {
                failure_threshold: 5, // High threshold
                ..Default::default()
            },
            ..Default::default()
        };
        let mut health = WorkerHealth::default();

        // Record 3 failures
        for _ in 0..3 {
            health.update(
                HealthCheckResult::failure("Error".to_string()),
                &config,
                "test-worker",
            );
        }
        assert_eq!(health.circuit_stats().consecutive_failures(), 3);

        // Success should reset consecutive failures
        health.update(HealthCheckResult::success(50), &config, "test-worker");
        assert_eq!(health.circuit_stats().consecutive_failures(), 0);
    }

    #[test]
    fn test_degraded_status_on_slow_response() {
        let _guard = test_guard!();
        let config = HealthConfig {
            degraded_threshold_ms: 1000, // 1 second threshold
            ..Default::default()
        };
        let mut health = WorkerHealth::default();

        // Slow but successful response
        health.update(HealthCheckResult::success(1500), &config, "test-worker");

        assert_eq!(health.circuit_state(), CircuitState::Closed);
        assert_eq!(health.status(), WorkerStatus::Degraded);
    }

    #[test]
    fn test_healthy_status_on_fast_response() {
        let _guard = test_guard!();
        let config = HealthConfig {
            degraded_threshold_ms: 1000,
            ..Default::default()
        };
        let mut health = WorkerHealth::default();

        // Fast successful response
        health.update(HealthCheckResult::success(500), &config, "test-worker");

        assert_eq!(health.circuit_state(), CircuitState::Closed);
        assert_eq!(health.status(), WorkerStatus::Healthy);
    }
}